Fix missing/broken ds.Stop handling in raw DS.

filter/dscache/ds.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package dscache
 
 import (
         "time"
 
         ds "github.com/luci/gae/service/datastore"
@@ skipping 101 matching lines @@
                         if err := d.mc.CompareAndSwapMulti(toCas); err != nil {
                                 (log.Fields{log.ErrorKey: err}).Warningf(
                                         d.c, "dscache: GetMulti: memcache.CompareAndSwapMulti")
                         }
                 }
         }
 
         // finally, run the callback for all of the decoded items and the errors,
         // if any.
         for i, dec := range p.decoded {
-                cb(dec, p.lme.GetOne(i))
+                if err := cb(dec, p.lme.GetOne(i)); err != nil {
+                        return err
+                }
         }
 
         return nil
 }
 
 func (d *dsCache) RunInTransaction(f func(context.Context) error, opts *ds.TransactionOptions) error {
         txnState := dsTxnState{}
         err := d.RawInterface.RunInTransaction(func(ctx context.Context) error {
                 txnState.reset()
                 err := f(context.WithValue(ctx, dsTxnCacheKey, &txnState))
                 if err == nil {
                         err = txnState.apply(d.supportContext)
                 }
                 return err
         }, opts)
         if err == nil {
                 txnState.release(d.supportContext)
         }
         return err
 }