diff --git a/services/syncbase/server/database_watch.go b/services/syncbase/server/database_watch.go
index 0d8a6cd..8df5782 100644
--- a/services/syncbase/server/database_watch.go
+++ b/services/syncbase/server/database_watch.go
@@ -228,7 +228,7 @@
 // - wait for one of two signals: new updates available or the call is canceled
 // The 'new updates' signal is sent by the watcher via a Go channel.
 func (d *database) watchUpdates(ctx *context.T, call rpc.ServerCall, sender *watchBatchSender, resumeMarker watch.ResumeMarker, watchFilter filter.CollectionRowFilter) error {
-	hasUpdates, cancelWatch := watchable.WatchUpdates(d.st)
+	watcher, cancelWatch := d.st.WatchUpdates(resumeMarker)
 	defer cancelWatch()
 	for {
 		// Drain the log queue.
@@ -237,7 +237,7 @@
 			// conflict resolution merge batches, very large batches may not be
 			// unrealistic. However, sync currently also processes an entire batch at
 			// a time, and would need to be updated as well.
-			logs, nextResumeMarker, err := watchable.ReadBatchFromLog(d.st, resumeMarker)
+			logs, nextResumeMarker, err := watcher.NextBatchFromLog(d.st)
 			if err != nil {
 				// TODO(ivanpi): Log all internal errors, especially ones not returned.
 				return verror.NewErrInternal(ctx) // no detailed error before access check
@@ -256,9 +256,9 @@
 		}
 		// Wait for new updates or cancel.
 		select {
-		case _, ok := <-hasUpdates:
+		case _, ok := <-watcher.Wait():
 			if !ok {
-				return verror.NewErrAborted(ctx)
+				return watcher.Err()
 			}
 		case <-ctx.Done():
 			return ctx.Err()
diff --git a/services/syncbase/server/watchlog_test.go b/services/syncbase/server/watchlog_test.go
index ed5f76c..3f46db0 100644
--- a/services/syncbase/server/watchlog_test.go
+++ b/services/syncbase/server/watchlog_test.go
@@ -75,6 +75,8 @@
 	call := &mockCall{p: v23.GetPrincipal(ctx), b: blessings}
 	var expected []interface{}
 	resumeMarker, _ := watchable.GetResumeMarker(st)
+	watcher, cancel := st.WatchUpdates(resumeMarker)
+	defer cancel()
 	// Generate Put/Delete events.
 	for i := 0; i < 5; i++ {
 		// Set initial collection permissions.
@@ -119,7 +121,7 @@
 	expectedIndex := 0
 	for {
 		var logs []*watchable.LogEntry
-		if logs, resumeMarker, _ = watchable.ReadBatchFromLog(st, resumeMarker); logs == nil {
+		if logs, resumeMarker, _ = watcher.NextBatchFromLog(st); logs == nil {
 			break
 		}
 		for _, logRecord := range logs {
diff --git a/services/syncbase/store/watchable/watcher.go b/services/syncbase/store/watchable/watcher.go
index e00cc24..c4ca18a 100644
--- a/services/syncbase/store/watchable/watcher.go
+++ b/services/syncbase/store/watchable/watcher.go
@@ -25,17 +25,19 @@
 	// Channel used by broadcastUpdates() to notify watcherLoop. When watcher is
 	// closed, updater is closed.
 	updater chan struct{}
+
 	// Protects the clients map.
 	mu sync.RWMutex
-	// Channels used by watcherLoop to notify currently registered clients. When
-	// watcher is closed, all client channels are closed and clients is set to nil.
-	clients map[chan struct{}]struct{}
+	// Currently registered clients, notified by watcherLoop via their channels.
+	// When watcher is closed, all clients are stopped (and their channels closed)
+	// with ErrAborted and clients is set to nil.
+	clients map[*Client]struct{}
 }
 
 func newWatcher() *watcher {
 	ret := &watcher{
 		updater: make(chan struct{}, 1),
-		clients: make(map[chan struct{}]struct{}),
+		clients: make(map[*Client]struct{}),
 	}
 	go ret.watcherLoop()
 	return ret
@@ -45,9 +47,9 @@
 func (w *watcher) close() {
 	w.mu.Lock()
 	if w.clients != nil {
-		// Close all client channels.
+		// Stop all clients and close their channels.
 		for c := range w.clients {
-			closeAndDrain(c)
+			c.stop(verror.NewErrAborted(nil))
 		}
 		// Set clients to nil to mark watcher as closed.
 		w.clients = nil
@@ -79,7 +81,7 @@
 		}
 		w.mu.RLock()
 		for c := range w.clients { // safe for w.clients == nil
-			ping(c)
+			ping(c.update)
 		}
 		w.mu.RUnlock()
 	}
@@ -104,40 +106,126 @@
 }
 
 // watchUpdates - see WatchUpdates.
-func (w *watcher) watchUpdates() (update <-chan struct{}, cancel func()) {
-	updateRW := make(chan struct{}, 1)
+func (w *watcher) watchUpdates(seq uint64) (_ *Client, cancel func()) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	if w.clients == nil {
-		// watcher is closed, return a closed update channel and no-op cancel.
-		close(updateRW)
-		cancel = func() {}
-		return updateRW, cancel
+		// watcher is closed. Return stopped Client.
+		return newStoppedClient(verror.NewErrAborted(nil)), func() {}
 	}
-	// Register update channel.
-	w.clients[updateRW] = struct{}{}
-	// Cancel is idempotent. It unregisters and closes the update channel.
+	// Register and return client.
+	c := newClient(seq)
+	w.clients[c] = struct{}{}
 	cancel = func() {
 		w.mu.Lock()
-		if _, ok := w.clients[updateRW]; ok { // safe for w.clients == nil
-			delete(w.clients, updateRW)
-			closeAndDrain(updateRW)
+		if _, ok := w.clients[c]; ok { // safe for w.clients == nil
+			c.stop(verror.NewErrCanceled(nil))
+			delete(w.clients, c)
 		}
 		w.mu.Unlock()
 	}
-	return updateRW, cancel
+	return c, cancel
 }
 
-// WatchUpdates returns a channel that can be used to wait for changes of the
-// database, as well as a cancel function which MUST be called to release the
-// watch resources. If the update channel is closed, the store is closed and
-// no more updates will happen. Otherwise, the channel will have a value
-// available whenever the store has changed since the last receive on the
-// channel.
-func WatchUpdates(st store.Store) (update <-chan struct{}, cancel func()) {
-	// TODO(rogulenko): Remove dynamic type assertion here and in other places.
-	watcher := st.(*Store).watcher
-	return watcher.watchUpdates()
+// WatchUpdates returns a Client which supports waiting for changes and
+// iterating over the watch log starting from resumeMarker, as well as a
+// cancel function which MUST be called to release watch resources.
+func (st *Store) WatchUpdates(resumeMarker watch.ResumeMarker) (_ *Client, cancel func()) {
+	seq, err := parseResumeMarker(string(resumeMarker))
+	if err != nil {
+		// resumeMarker is invalid. Return stopped Client.
+		return newStoppedClient(err), func() {}
+	}
+	return st.watcher.watchUpdates(seq)
+}
+
+// Client encapsulates a channel used to notify watch clients of store updates
+// and an iterator over the watch log.
+type Client struct {
+	// Channel used by watcherLoop to notify the client. When the client is
+	// stopped, update is closed.
+	update chan struct{}
+
+	// Protects the fields below.
+	mu sync.Mutex
+	// Sequence number pointing to the start of the previously retrieved log
+	// batch. Equal to nextSeq if the retrieved batch was empty.
+	prevSeq uint64
+	// Sequence number pointing to the start of the next log batch to retrieve.
+	nextSeq uint64
+	// When the client is stopped, err is set to the reason for stopping.
+	err error
+}
+
+func newClient(seq uint64) *Client {
+	return &Client{
+		update:  make(chan struct{}, 1),
+		prevSeq: seq,
+		nextSeq: seq,
+	}
+}
+
+func newStoppedClient(err error) *Client {
+	c := newClient(0)
+	c.stop(err)
+	return c
+}
+
+// Wait returns the update channel that can be used to wait for new changes in
+// the store. If the update channel is closed, the client is stopped and no more
+// updates will happen. Otherwise, the channel will have a value available
+// whenever the store has changed since the last receive on the channel.
+func (c *Client) Wait() <-chan struct{} {
+	return c.update
+}
+
+// NextBatchFromLog returns the next batch of watch log records (transaction)
+// from the given database and the resume marker at the end of the batch. If
+// there is no batch available, it returns a nil slice and the same resume
+// marker as the previous NextBatchFromLog call. The returned log entries are
+// guaranteed to point to existing data versions until either the client is
+// stopped or NextBatchFromLog is called again. If the client is stopped,
+// NextBatchFromLog returns the same error as Err.
+func (c *Client) NextBatchFromLog(st store.Store) ([]*LogEntry, watch.ResumeMarker, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.err != nil {
+		return nil, nil, c.err
+	}
+	batch, batchEndSeq, err := readBatchFromLog(st, c.nextSeq)
+	if err != nil {
+		// We cannot call stop() here since c.mu is locked. However, we checked
+		// above that c.err is nil, so it is safe to set c.err anc close c.update.
+		c.err = err
+		closeAndDrain(c.update)
+		return nil, nil, err
+	}
+	c.prevSeq = c.nextSeq
+	c.nextSeq = batchEndSeq
+	return batch, MakeResumeMarker(batchEndSeq), nil
+}
+
+// Err returns the error that caused the client to stop watching. If the error
+// is nil, the client is active. Otherwise:
+// * ErrCanceled - watch was canceled by the client.
+// * ErrAborted - watcher was closed (store was closed, possibly destroyed).
+// * ErrUnknownResumeMarker - watch was started with an invalid resume marker.
+// * other errors - NextBatchFromLog encountered an error.
+func (c *Client) Err() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.err
+}
+
+// stop closes the client update channel and sets the error returned by Err.
+// Idempotent (only the error from the first call to stop is kept).
+func (c *Client) stop(err error) {
+	c.mu.Lock()
+	if c.err == nil {
+		c.err = err
+		closeAndDrain(c.update)
+	}
+	c.mu.Unlock()
 }
 
 // GetResumeMarker returns the ResumeMarker that points to the current end
@@ -158,15 +246,11 @@
 	return join(common.LogPrefix, fmt.Sprintf("%016x", seq))
 }
 
-// ReadBatchFromLog returns a batch of watch log records (a transaction) from
-// the given database and the new resume marker at the end of the batch.
-func ReadBatchFromLog(st store.Store, resumeMarker watch.ResumeMarker) ([]*LogEntry, watch.ResumeMarker, error) {
-	seq, err := parseResumeMarker(string(resumeMarker))
-	if err != nil {
-		return nil, resumeMarker, err
-	}
+// readBatchFromLog returns a batch of watch log records (a transaction) from
+// the given database and the next sequence number at the end of the batch.
+func readBatchFromLog(st store.Store, seq uint64) ([]*LogEntry, uint64, error) {
 	_, scanLimit := common.ScanPrefixArgs(common.LogPrefix, "")
-	scanStart := resumeMarker
+	scanStart := MakeResumeMarker(seq)
 	endOfBatch := false
 
 	// Use the store directly to scan these read-only log entries, no need
@@ -179,7 +263,7 @@
 		var logEnt LogEntry
 		if err := vom.Decode(stream.Value(nil), &logEnt); err != nil {
 			stream.Cancel()
-			return nil, resumeMarker, err
+			return nil, seq, err
 		}
 
 		logs = append(logs, &logEnt)
@@ -193,15 +277,15 @@
 	}
 
 	if !endOfBatch {
-		if err = stream.Err(); err != nil {
-			return nil, resumeMarker, err
+		if err := stream.Err(); err != nil {
+			return nil, seq, err
 		}
 		if len(logs) > 0 {
 			vlog.Fatalf("end of batch not found after %d entries", len(logs))
 		}
-		return nil, resumeMarker, nil
+		return nil, seq, nil
 	}
-	return logs, watch.ResumeMarker(logEntryKey(seq)), nil
+	return logs, seq, nil
 }
 
 func parseResumeMarker(resumeMarker string) (uint64, error) {
diff --git a/services/syncbase/store/watchable/watcher_test.go b/services/syncbase/store/watchable/watcher_test.go
index 86985ae..f7005fb 100644
--- a/services/syncbase/store/watchable/watcher_test.go
+++ b/services/syncbase/store/watchable/watcher_test.go
@@ -10,6 +10,7 @@
 	"testing"
 	"time"
 
+	"v.io/v23/verror"
 	"v.io/x/ref/services/syncbase/common"
 	"v.io/x/ref/services/syncbase/store"
 )
@@ -42,11 +43,11 @@
 	}
 
 	// Fetch the batches and a few more empty fetches and verify them.
-	resmark := MakeResumeMarker(0)
-	var seq uint64
+	var seq uint64 = 0
+	var wantSeq uint64 = 0
 
 	for i := 0; i < (numTx + 3); i++ {
-		logs, newResmark, err := ReadBatchFromLog(st, resmark)
+		logs, newSeq, err := readBatchFromLog(st, seq)
 		if err != nil {
 			t.Fatalf("can't get watch log batch: %v", err)
 		}
@@ -56,11 +57,10 @@
 					i, len(logs), numPut)
 			}
 
-			seq += uint64(len(logs))
-			expResmark := MakeResumeMarker(seq)
-			if !bytes.Equal(newResmark, expResmark) {
-				t.Errorf("log fetch (i=%d) wrong resmark: %s instead of %s",
-					i, newResmark, expResmark)
+			wantSeq += uint64(len(logs))
+			if newSeq != wantSeq {
+				t.Errorf("log fetch (i=%d) wrong seq: %d instead of %d",
+					i, newSeq, wantSeq)
 			}
 
 			for j, log := range logs {
@@ -87,12 +87,12 @@
 				tx.Abort()
 			}
 		} else {
-			if logs != nil || !bytes.Equal(newResmark, resmark) {
-				t.Errorf("NOP log fetch (i=%d) had changes: %d logs, resmask %s",
-					i, len(logs), newResmark)
+			if logs != nil || newSeq != seq {
+				t.Errorf("NOP log fetch (i=%d) had changes: %d logs, seq %d",
+					i, len(logs), newSeq)
 			}
 		}
-		resmark = newResmark
+		seq = newSeq
 	}
 }
 
@@ -106,23 +106,26 @@
 	w.broadcastUpdates()
 
 	// Never-receiving client should not block watcher.
-	_, cancel1 := w.watchUpdates()
+	_, cancel1 := w.watchUpdates(0)
 	defer cancel1()
 
 	// Cancelled client should not affect watcher.
-	chan2, cancel2 := w.watchUpdates()
+	c2, cancel2 := w.watchUpdates(0)
 	cancel2()
 	// Cancel should be idempotent.
 	cancel2()
 
 	// Channel should be closed when client is cancelled.
 	select {
-	case _, ok := <-chan2:
+	case _, ok := <-c2.Wait():
 		if ok {
-			t.Fatalf("cancel2 was called, chan2 should be drained and closed")
+			t.Fatalf("cancel2 was called, c2 channel should be drained and closed")
 		}
 	default:
-		t.Fatalf("cancel2 was called, chan2 should be closed")
+		t.Fatalf("cancel2 was called, c2 channel should be closed")
+	}
+	if verror.ErrorID(c2.Err()) != verror.ErrCanceled.ID {
+		t.Fatalf("expected c2.Err() to return ErrCanceled, got: %v", c2.Err())
 	}
 
 	// Update broadcast should not block client registration or vice versa.
@@ -130,13 +133,13 @@
 	registerLoop1 := make(chan bool)
 	go func() {
 		for i := 0; i < 5000; i++ {
-			_, canceli := w.watchUpdates()
+			_, canceli := w.watchUpdates(0)
 			defer canceli()
 		}
 		registerLoop1 <- true
 	}()
 
-	chan3, cancel3 := w.watchUpdates()
+	c3, cancel3 := w.watchUpdates(0)
 
 	for i := 0; i < 5000; i++ {
 		w.broadcastUpdates()
@@ -154,19 +157,22 @@
 
 	// chan3 should have a single pending notification.
 	select {
-	case _, ok := <-chan3:
+	case _, ok := <-c3.Wait():
 		if !ok {
-			t.Fatalf("chan3 should not be closed")
+			t.Fatalf("c3 channel should not be closed")
 		}
 	default:
-		t.Fatalf("chan3 should have a notification")
+		t.Fatalf("c3 channel should have a notification")
 	}
 	select {
-	case <-chan3:
-		t.Fatalf("chan3 should not have another notification")
+	case <-c3.Wait():
+		t.Fatalf("c3 channel should not have another notification")
 	default:
 		// ok
 	}
+	if c3.Err() != nil {
+		t.Fatalf("expected c3.Err() to return nil, got: %v", c3.Err())
+	}
 
 	// After notification was read, chan3 still receives updates.
 	go func() {
@@ -175,12 +181,15 @@
 	}()
 
 	select {
-	case _, ok := <-chan3:
+	case _, ok := <-c3.Wait():
 		if !ok {
-			t.Fatalf("chan3 should not be closed")
+			t.Fatalf("c3 channel should not be closed")
 		}
 	case <-time.After(5 * time.Second):
-		t.Fatalf("chan3 didn't receive after 5s")
+		t.Fatalf("c3 channel didn't receive after 5s")
+	}
+	if c3.Err() != nil {
+		t.Fatalf("expected c3.Err() to return nil, got: %v", c3.Err())
 	}
 
 	// Closing the watcher.
@@ -190,28 +199,38 @@
 
 	// Client channels should be closed when watcher is closed.
 	select {
-	case _, ok := <-chan3:
+	case _, ok := <-c3.Wait():
 		if ok {
-			t.Fatalf("watcher was closed, chan3 should be drained and closed")
+			t.Fatalf("watcher was closed, c3 channel should be drained and closed")
 		}
 	default:
-		t.Fatalf("watcher was closed, chan3 should be closed")
+		t.Fatalf("watcher was closed, c3 channel should be closed")
+	}
+	if verror.ErrorID(c3.Err()) != verror.ErrAborted.ID {
+		t.Fatalf("expected c3.Err() to return ErrAborted, got: %v", c3.Err())
 	}
 
 	// Cancel is safe to call after the store is closed.
 	cancel3()
+	// ErrAborted should be preserved instead of being overridden by ErrCanceled.
+	if verror.ErrorID(c3.Err()) != verror.ErrAborted.ID {
+		t.Fatalf("expected c3.Err() to return ErrAborted, got: %v", c3.Err())
+	}
 
 	// watchUpdates is safe to call after the store is closed, returning closed
 	// channel.
-	chan4, cancel4 := w.watchUpdates()
+	c4, cancel4 := w.watchUpdates(0)
 
 	select {
-	case _, ok := <-chan4:
+	case _, ok := <-c4.Wait():
 		if ok {
-			t.Fatalf("watcher was closed, chan4 should be drained and closed")
+			t.Fatalf("watcher was closed, c4 channel should be drained and closed")
 		}
 	default:
-		t.Fatalf("watcher was closed, chan4 should be closed")
+		t.Fatalf("watcher was closed, c4 channel should be closed")
+	}
+	if verror.ErrorID(c4.Err()) != verror.ErrAborted.ID {
+		t.Fatalf("expected c4.Err() to return ErrAborted, got: %v", c4.Err())
 	}
 
 	cancel4()
diff --git a/services/syncbase/vsync/watcher.go b/services/syncbase/vsync/watcher.go
index b14fcbc..e512f94 100644
--- a/services/syncbase/vsync/watcher.go
+++ b/services/syncbase/vsync/watcher.go
@@ -71,62 +71,70 @@
 func (s *syncService) watchStore(ctx *context.T, dbId wire.Id, st *watchable.Store) {
 	vlog.VI(1).Infof("sync: watchStore: DB %v: start watching updates", dbId)
 
-	updatesChan, cancel := watchable.WatchUpdates(st)
+	resMark, err := getResMark(ctx, st)
+	if err != nil {
+		if verror.ErrorID(err) != verror.ErrNoExist.ID {
+			// TODO(rdaoud): Quarantine this database.
+			vlog.Errorf("sync: watchStore: %v: cannot get resMark, stop watching and exit: %v", dbId, err)
+			return
+		}
+		resMark = watchable.MakeResumeMarker(0)
+	}
+
+	watcher, cancel := st.WatchUpdates(resMark)
 	defer cancel()
 
 	moreWork := true
 	for moreWork && !s.isClosed() {
-		if s.processDatabase(ctx, dbId, st) {
+		if hadUpdate, err := s.processDatabase(ctx, dbId, st, watcher); err != nil {
+			// TODO(rdaoud): Quarantine this database.
+			vlog.Errorf("sync: watchStore: DB %v: processDatabase failed, stop watching and exit: %v", dbId, err)
+			return
+		} else if hadUpdate {
 			vlog.VI(2).Infof("sync: watchStore: DB %v: had updates", dbId)
 		} else {
 			vlog.VI(2).Infof("sync: watchStore: DB %v: idle, wait for updates", dbId)
 			select {
-			case _, moreWork = <-updatesChan:
+			case _, moreWork = <-watcher.Wait():
 
-			case <-s.closed:
-				moreWork = false
+			case _, moreWork = <-s.closed:
+
 			}
 		}
 	}
 
-	vlog.VI(1).Infof("sync: watchStore: DB %v: channel closed, stop watching and exit", dbId)
+	vlog.VI(1).Infof("sync: watchStore: DB %v: channel closed, stop watching and exit, watch status %v", dbId, watcher.Err())
 }
 
 // processDatabase fetches from the given database at most one new batch update
 // (transaction) and processes it.  A batch is stored as a contiguous set of log
 // records ending with one record having the "continued" flag set to false.  The
 // call returns true if a new batch update was processed.
-func (s *syncService) processDatabase(ctx *context.T, dbId wire.Id, st store.Store) bool {
+func (s *syncService) processDatabase(ctx *context.T, dbId wire.Id, st store.Store, watcher *watchable.Client) (bool, error) {
 	vlog.VI(2).Infof("sync: processDatabase: begin: %v", dbId)
 	defer vlog.VI(2).Infof("sync: processDatabase: end: %v", dbId)
 
-	resMark, err := getResMark(ctx, st)
-	if err != nil {
-		if verror.ErrorID(err) != verror.ErrNoExist.ID {
-			vlog.Errorf("sync: processDatabase: %v: cannot get resMark: %v", dbId, err)
-			return false
-		}
-		resMark = watchable.MakeResumeMarker(0)
-	}
-
 	// Initialize Database sync state if needed.
 	s.initSyncStateInMem(ctx, dbId, "")
 
-	// Get a batch of watch log entries, if any, after this resume marker.
-	logs, nextResmark, err := watchable.ReadBatchFromLog(st, resMark)
+	// Get the next batch of watch log entries, if any.
+	logs, nextResmark, err := watcher.NextBatchFromLog(st)
 	if err != nil {
-		// An error here (scan stream cancelled) is possible when the watcher is in
-		// the middle of processing a database when it is destroyed. Hence, we just
-		// ignore this database and proceed.
 		vlog.Errorf("sync: processDatabase: %v: cannot get watch log batch: %v", dbId, verror.DebugString(err))
-		return false
+		if verror.ErrorID(err) == verror.ErrAborted.ID {
+			// Watch was aborted because the database was closed, probably destroyed.
+			// We suppress this error and let the watcher exit gracefully.
+			vlog.VI(1).Infof("sync: processDatabase: %v: watch aborted because database was closed", dbId)
+			return false, nil
+		}
+		return false, err
 	} else if logs == nil {
-		return false
+		return false, nil
 	}
 
-	if err = s.processWatchLogBatch(ctx, dbId, st, logs, nextResmark); err != nil {
-		// TODO(rdaoud): quarantine this database.
-		return false
+	if err = s.processWatchLogBatch(ctx, dbId, st, watcher.Err, logs, nextResmark); err != nil {
+		vlog.Errorf("sync: processDatabase: %v: cannot process watch log batch: %v", dbId, verror.DebugString(err))
+		return false, err
 	}
 
 	// The requirement for pause is that any write after the pause must not
@@ -143,23 +151,25 @@
 		// The database is online. Cut a gen.
 		if err := s.checkptLocalGen(ctx, dbId, nil); err != nil {
 			vlog.Errorf("sync: processDatabase: %v: cannot cut a generation: %v", dbId, verror.DebugString(err))
-			return false
+			// TODO(ivanpi): This may not be a fatal error, continue watching?
+			return false, err
 		}
 	} else {
 		vlog.VI(1).Infof("sync: processDatabase: %v database not allowed to sync, skipping cutting a gen", dbId)
 	}
-	return true
+	return true, nil
 }
 
 // processWatchLogBatch parses the given batch of watch log records, updates the
 // watchable syncgroup prefixes, uses the prefixes to filter the batch to the
 // subset of syncable records, and transactionally applies these updates to the
 // sync metadata (DAG & log records) and updates the watch resume marker.
-func (s *syncService) processWatchLogBatch(ctx *context.T, dbId wire.Id, st store.Store, logs []*watchable.LogEntry, resMark watch.ResumeMarker) error {
+func (s *syncService) processWatchLogBatch(ctx *context.T, dbId wire.Id, st store.Store, watchStatus func() error, logs []*watchable.LogEntry, resMark watch.ResumeMarker) error {
 	if len(logs) == 0 {
 		return nil
 	}
 
+	// TODO(ivanpi): Pipe through errors from here instead of calling Fatalf.
 	if processDbStateChangeLogRecord(ctx, s, st, dbId, logs[0], resMark) {
 		// A batch containing DbStateChange will not have any more records.
 		// This batch is done processing.
@@ -213,10 +223,17 @@
 	vlog.VI(3).Infof("sync: processWatchLogBatch: %v: sg snap %t, syncable %d, total %d", dbId, !appBatch, len(batch), totalCount)
 
 	if err := s.processWatchBlobRefs(ctx, dbId, st, batch); err != nil {
-		// There may be an error here if the database is recently
-		// destroyed.  Ignore the error and continue to another database.
 		vlog.Errorf("sync: processWatchLogBatch: %v: watcher cannot process blob refs: %v", dbId, err)
-		return nil
+		if verror.ErrorID(watchStatus()) == verror.ErrAborted.ID {
+			// Watch was aborted because the database was closed, probably destroyed.
+			// We suppress this error and let the watcher exit gracefully.
+			// Note, we use the watcher error for this because it is guaranteed to be
+			// set to ErrAborted when the database is closed gracefully, unlike the
+			// error returned from processWatchBlobRefs.
+			vlog.VI(1).Infof("sync: processWatchLogBatch: %v: watch aborted because database was closed", dbId)
+			return nil
+		}
+		return err
 	}
 
 	// Transactional processing of the batch: Fixup syncable log records to
@@ -241,10 +258,17 @@
 	})
 
 	if err != nil {
-		// There may be an error here if the database is recently
-		// destroyed. Ignore the error and continue to another database.
-		// TODO(rdaoud): quarantine this database for other errors.
 		vlog.Errorf("sync: processWatchLogBatch: %v: watcher cannot process batch: %v", dbId, err)
+		if verror.ErrorID(watchStatus()) == verror.ErrAborted.ID {
+			// Watch was aborted because the database was closed, probably destroyed.
+			// We suppress this error and let the watcher exit gracefully.
+			// Note, we use the watcher error for this because it is guaranteed to be
+			// set to ErrAborted when the database is closed gracefully, unlike the
+			// error returned from RunInTransaction.
+			vlog.VI(1).Infof("sync: processWatchLogBatch: %v: watch aborted because database was closed", dbId)
+			return nil
+		}
+		return err
 	}
 	return nil
 }
diff --git a/services/syncbase/vsync/watcher_test.go b/services/syncbase/vsync/watcher_test.go
index 82b396e..ac6c04b 100644
--- a/services/syncbase/vsync/watcher_test.go
+++ b/services/syncbase/vsync/watcher_test.go
@@ -207,8 +207,10 @@
 	barKey := makeRowKey("bar")
 	fooxyzKey := makeRowKey("fooxyz")
 
+	noError := func() error { return nil }
+
 	// Empty logs does not fail.
-	s.processWatchLogBatch(nil, mockDbId, st, nil, nil)
+	s.processWatchLogBatch(nil, mockDbId, st, noError, nil, nil)
 
 	// Non-syncable logs.
 	batch := []*watchable.LogEntry{
@@ -217,7 +219,7 @@
 	}
 
 	resmark := watchable.MakeResumeMarker(1234)
-	s.processWatchLogBatch(nil, mockDbId, st, batch, resmark)
+	s.processWatchLogBatch(nil, mockDbId, st, noError, batch, resmark)
 
 	if res, err := getResMark(nil, st); err != nil && !bytes.Equal(res, resmark) {
 		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
@@ -250,7 +252,7 @@
 	}
 
 	resmark = watchable.MakeResumeMarker(3456)
-	s.processWatchLogBatch(nil, mockDbId, st, batch, resmark)
+	s.processWatchLogBatch(nil, mockDbId, st, noError, batch, resmark)
 
 	if res, err := getResMark(nil, st); err != nil && !bytes.Equal(res, resmark) {
 		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
@@ -290,7 +292,7 @@
 	}
 
 	resmark = watchable.MakeResumeMarker(7890)
-	s.processWatchLogBatch(nil, mockDbId, st, batch, resmark)
+	s.processWatchLogBatch(nil, mockDbId, st, noError, batch, resmark)
 
 	if res, err := getResMark(nil, st); err != nil && !bytes.Equal(res, resmark) {
 		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
@@ -333,7 +335,7 @@
 	}
 
 	resmark = watchable.MakeResumeMarker(20212223)
-	s.processWatchLogBatch(nil, mockDbId, st, batch, resmark)
+	s.processWatchLogBatch(nil, mockDbId, st, noError, batch, resmark)
 
 	if res, err := getResMark(nil, st); err != nil && !bytes.Equal(res, resmark) {
 		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)