diff --git a/x/ref/services/syncbase/server/watchable/transaction.go b/x/ref/services/syncbase/server/watchable/transaction.go
index 3cbff62..ccb291d 100644
--- a/x/ref/services/syncbase/server/watchable/transaction.go
+++ b/x/ref/services/syncbase/server/watchable/transaction.go
@@ -72,14 +72,16 @@
 	if tx.err != nil {
 		return convertError(tx.err)
 	}
-	var err error
 	if !tx.st.managesKey(key) {
-		err = tx.itx.Put(key, value)
-	} else {
-		err = putVersioned(tx.itx, key, value)
-		tx.ops = append(tx.ops, &OpPut{PutOp{Key: key, Value: value}})
+		return tx.itx.Put(key, value)
 	}
-	return err
+
+	version, err := putVersioned(tx.itx, key, value)
+	if err != nil {
+		return err
+	}
+	tx.ops = append(tx.ops, &OpPut{PutOp{Key: key, Version: version}})
+	return nil
 }
 
 // Delete implements the store.StoreWriter interface.
@@ -118,15 +120,15 @@
 	}
 	// Write LogEntry records.
 	timestamp := tx.st.clock.Now().UnixNano()
+	// TODO(rdaoud): switch to using a single counter for log entries
+	// instead of a (sequence, index) combo.
 	keyPrefix := getLogEntryKeyPrefix(tx.st.seq)
 	for txSeq, op := range tx.ops {
 		key := join(keyPrefix, fmt.Sprintf("%04x", txSeq))
 		value := &LogEntry{
 			Op:              op,
 			CommitTimestamp: timestamp,
-			// TODO(sadovsky): This information is also captured in LogEntry keys.
-			// Optimize to avoid redundancy.
-			Continued: txSeq < len(tx.ops)-1,
+			Continued:       txSeq < len(tx.ops)-1,
 		}
 		if err := util.PutObject(tx.itx, key, value); err != nil {
 			return err
@@ -150,6 +152,22 @@
 	return tx.itx.Abort()
 }
 
+// AddSyncGroupOp injects a SyncGroup operation notification in the log entries
+// that the transaction writes when it is committed.  It allows the SyncGroup
+// operations (create, join, leave, destroy) to notify the sync watcher of the
+// change at its proper position in the timeline (the transaction commit).
+// Note: this is an internal function used by sync, not part of the interface.
+func AddSyncGroupOp(tx store.Transaction, prefixes []string, remove bool) error {
+	wtx := tx.(*transaction)
+	wtx.mu.Lock()
+	defer wtx.mu.Unlock()
+	if wtx.err != nil {
+		return convertError(wtx.err)
+	}
+	wtx.ops = append(wtx.ops, &OpSyncGroup{SyncGroupOp{Prefixes: prefixes, Remove: remove}})
+	return nil
+}
+
 // Exported as a helper function for testing purposes
 func getLogEntryKeyPrefix(seq uint64) string {
 	// Note, MaxUint16 is 0xffff and MaxUint64 is 0xffffffffffffffff.
diff --git a/x/ref/services/syncbase/server/watchable/types.vdl b/x/ref/services/syncbase/server/watchable/types.vdl
index 07d1ce1..6f07a1c 100644
--- a/x/ref/services/syncbase/server/watchable/types.vdl
+++ b/x/ref/services/syncbase/server/watchable/types.vdl
@@ -15,10 +15,12 @@
 	Limit []byte
 }
 
-// PutOp represents a store put operation.
+// PutOp represents a store put operation.  The new version is written instead
+// of the value to avoid duplicating the user data in the store.  The version
+// is used to access the user data of that specific mutation.
 type PutOp struct {
-	Key   []byte
-	Value []byte
+	Key     []byte
+	Version []byte
 }
 
 // DeleteOp represents a store delete operation.
@@ -26,12 +28,21 @@
 	Key []byte
 }
 
+// SyncGroupOp represents a change in SyncGroup tracking, adding or removing
+// key prefixes to sync.  SyncGroup prefixes cannot be changed, this is used
+// to track changes due to SyncGroup create/join/leave/destroy.
+type SyncGroupOp struct {
+	Prefixes []string
+	Remove   bool
+}
+
 // Op represents a store operation.
 type Op union {
-	Get    GetOp
-	Scan   ScanOp
-	Put    PutOp
-	Delete DeleteOp
+	Get       GetOp
+	Scan      ScanOp
+	Put       PutOp
+	Delete    DeleteOp
+	SyncGroup SyncGroupOp
 }
 
 // LogEntry represents a single store operation. This operation may have been
diff --git a/x/ref/services/syncbase/server/watchable/types.vdl.go b/x/ref/services/syncbase/server/watchable/types.vdl.go
index e94587a..a4f0619 100644
--- a/x/ref/services/syncbase/server/watchable/types.vdl.go
+++ b/x/ref/services/syncbase/server/watchable/types.vdl.go
@@ -33,10 +33,12 @@
 }) {
 }
 
-// PutOp represents a store put operation.
+// PutOp represents a store put operation.  The new version is written instead
+// of the value to avoid duplicating the user data in the store.  The version
+// is used to access the user data of that specific mutation.
 type PutOp struct {
-	Key   []byte
-	Value []byte
+	Key     []byte
+	Version []byte
 }
 
 func (PutOp) __VDLReflect(struct {
@@ -54,6 +56,19 @@
 }) {
 }
 
+// SyncGroupOp represents a change in SyncGroup tracking, adding or removing
+// key prefixes to sync.  SyncGroup prefixes cannot be changed, this is used
+// to track changes due to SyncGroup create/join/leave/destroy.
+type SyncGroupOp struct {
+	Prefixes []string
+	Remove   bool
+}
+
+func (SyncGroupOp) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/x/ref/services/syncbase/server/watchable.SyncGroupOp"`
+}) {
+}
+
 type (
 	// Op represents any single field of the Op union type.
 	//
@@ -76,15 +91,18 @@
 	OpPut struct{ Value PutOp }
 	// OpDelete represents field Delete of the Op union type.
 	OpDelete struct{ Value DeleteOp }
+	// OpSyncGroup represents field SyncGroup of the Op union type.
+	OpSyncGroup struct{ Value SyncGroupOp }
 	// __OpReflect describes the Op union type.
 	__OpReflect struct {
 		Name  string `vdl:"v.io/syncbase/x/ref/services/syncbase/server/watchable.Op"`
 		Type  Op
 		Union struct {
-			Get    OpGet
-			Scan   OpScan
-			Put    OpPut
-			Delete OpDelete
+			Get       OpGet
+			Scan      OpScan
+			Put       OpPut
+			Delete    OpDelete
+			SyncGroup OpSyncGroup
 		}
 	}
 )
@@ -109,6 +127,11 @@
 func (x OpDelete) Name() string             { return "Delete" }
 func (x OpDelete) __VDLReflect(__OpReflect) {}
 
+func (x OpSyncGroup) Index() int               { return 4 }
+func (x OpSyncGroup) Interface() interface{}   { return x.Value }
+func (x OpSyncGroup) Name() string             { return "SyncGroup" }
+func (x OpSyncGroup) __VDLReflect(__OpReflect) {}
+
 // LogEntry represents a single store operation. This operation may have been
 // part of a transaction, as signified by the Continued boolean. Read-only
 // operations (and read-only transactions) are not logged.
@@ -132,6 +155,7 @@
 	vdl.Register((*ScanOp)(nil))
 	vdl.Register((*PutOp)(nil))
 	vdl.Register((*DeleteOp)(nil))
+	vdl.Register((*SyncGroupOp)(nil))
 	vdl.Register((*Op)(nil))
 	vdl.Register((*LogEntry)(nil))
 }
diff --git a/x/ref/services/syncbase/server/watchable/util.go b/x/ref/services/syncbase/server/watchable/util.go
index 3dc3e78..036d326 100644
--- a/x/ref/services/syncbase/server/watchable/util.go
+++ b/x/ref/services/syncbase/server/watchable/util.go
@@ -7,6 +7,10 @@
 // TODO(sadovsky): Avoid copying back and forth between []byte's and strings.
 // We should probably convert incoming strings to []byte's as early as possible,
 // and deal exclusively in []byte's internally.
+// TODO(rdaoud): I propose we standardize on key and version being strings and
+// the value being []byte within Syncbase.  We define invalid characters in the
+// key space (and reserve "$" and ":").  The lower storage engine layers are
+// free to map that to what they need internally ([]byte or string).
 
 import (
 	"fmt"
@@ -25,6 +29,19 @@
 	rngLock sync.Mutex
 )
 
+// NewVersion returns a new version for a store entry mutation.
+func NewVersion() []byte {
+	// TODO(rdaoud): revisit the number of bits: should we use 128 bits?
+	// Note: the version has to be unique per object key, not on its own.
+	// TODO(rdaoud): move sync's rand64() to a general Syncbase spot and
+	// reuse it here.
+	rngLock.Lock()
+	num := rng.Int63()
+	rngLock.Unlock()
+
+	return []byte(fmt.Sprintf("%x", num))
+}
+
 func makeVersionKey(key []byte) []byte {
 	return []byte(join(util.VersionPrefix, string(key)))
 }
@@ -109,20 +126,19 @@
 	if err := wtx.itx.Put(makeVersionKey(key), version); err != nil {
 		return err
 	}
-	wtx.ops = append(wtx.ops, &OpPut{PutOp{Key: key, Value: version}})
+	wtx.ops = append(wtx.ops, &OpPut{PutOp{Key: key, Version: version}})
 	return nil
 }
 
-func putVersioned(tx store.Transaction, key, value []byte) error {
-	rngLock.Lock()
-	num := rng.Int63()
-	rngLock.Unlock()
-
-	version := []byte(fmt.Sprintf("%x", num))
+func putVersioned(tx store.Transaction, key, value []byte) ([]byte, error) {
+	version := NewVersion()
 	if err := tx.Put(makeVersionKey(key), version); err != nil {
-		return err
+		return nil, err
 	}
-	return tx.Put(makeAtVersionKey(key, version), value)
+	if err := tx.Put(makeAtVersionKey(key, version), value); err != nil {
+		return nil, err
+	}
+	return version, nil
 }
 
 func deleteVersioned(tx store.Transaction, key []byte) error {
diff --git a/x/ref/services/syncbase/vsync/dag.go b/x/ref/services/syncbase/vsync/dag.go
index 5a432c7..72e8d91 100644
--- a/x/ref/services/syncbase/vsync/dag.go
+++ b/x/ref/services/syncbase/vsync/dag.go
@@ -726,8 +726,9 @@
 	}
 
 	var node dagNode
-	if err := util.GetObject(st, nodeKey(oid, version), &node); err != nil {
-		return nil, verror.New(verror.ErrInternal, ctx, err)
+	key := nodeKey(oid, version)
+	if err := util.GetObject(st, key, &node); err != nil {
+		return nil, translateError(ctx, err, key)
 	}
 	return &node, nil
 }
@@ -777,8 +778,9 @@
 // getHead retrieves the DAG object head.
 func getHead(ctx *context.T, st store.StoreReader, oid string) (string, error) {
 	var version string
-	if err := util.GetObject(st, headKey(oid), &version); err != nil {
-		return NoVersion, verror.New(verror.ErrInternal, ctx, err)
+	key := headKey(oid)
+	if err := util.GetObject(st, key, &version); err != nil {
+		return NoVersion, translateError(ctx, err, key)
 	}
 	return version, nil
 }
@@ -819,8 +821,9 @@
 	}
 
 	var info batchInfo
-	if err := util.GetObject(st, batchKey(btid), &info); err != nil {
-		return nil, verror.New(verror.ErrInternal, ctx, err)
+	key := batchKey(btid)
+	if err := util.GetObject(st, key, &info); err != nil {
+		return nil, translateError(ctx, err, key)
 	}
 	return &info, nil
 }
diff --git a/x/ref/services/syncbase/vsync/sync.go b/x/ref/services/syncbase/vsync/sync.go
index ca2e3fc..eac07f7 100644
--- a/x/ref/services/syncbase/vsync/sync.go
+++ b/x/ref/services/syncbase/vsync/sync.go
@@ -129,7 +129,7 @@
 	s.pending.Add(2)
 
 	// Start watcher thread to watch for updates to local store.
-	go s.watchStore()
+	go s.watchStore(ctx)
 
 	// Start initiator thread to periodically get deltas from peers.
 	go s.syncer(ctx)
diff --git a/x/ref/services/syncbase/vsync/sync_state.go b/x/ref/services/syncbase/vsync/sync_state.go
index f025327..101d08c 100644
--- a/x/ref/services/syncbase/vsync/sync_state.go
+++ b/x/ref/services/syncbase/vsync/sync_state.go
@@ -55,10 +55,10 @@
 }
 
 // initSync initializes the sync module during startup. It scans all the
-// databases across all apps to a) initialize the in-memory sync state of a
-// Database consisting of the current generation number, log position and
-// generation vector; b) initialize the prefixes that are currently being
-// synced.
+// databases across all apps to initialize the following:
+// a) in-memory sync state of a Database consisting of the current generation
+//    number, log position and generation vector.
+// b) watcher map of prefixes currently being synced.
 //
 // TODO(hpucha): This is incomplete. Flesh this out further.
 func (s *syncService) initSync(ctx *context.T) error {
@@ -68,40 +68,49 @@
 	var errFinal error
 	s.syncState = make(map[string]*dbSyncStateInMem)
 
-	// TODO(hpucha): Temporary hack.
-	return errFinal
-
-	/*s.forEachDatabaseStore(ctx, func(st store.Store) bool {
-		sn := st.NewSnapshot()
-		defer sn.Close()
-
-		ds, err := getDbSyncState(ctx, sn)
-		if err != nil && verror.ErrorID(err) != verror.ErrNoExist.ID {
-			errFinal = err
+	s.forEachDatabaseStore(ctx, func(appName, dbName string, st store.Store) bool {
+		// Scan the SyncGroups, skipping those not yet being watched.
+		forEachSyncGroup(st, func(sg *interfaces.SyncGroup) bool {
+			// TODO(rdaoud): only use SyncGroups that have been
+			// marked as "watchable" by the sync watcher thread.
+			// This is to handle the case of a SyncGroup being
+			// created but Syncbase restarting before the watcher
+			// processed the SyncGroupOp entry in the watch queue.
+			// It should not be syncing that SyncGroup's data after
+			// restart, but wait until the watcher processes the
+			// entry as would have happened without a restart.
+			for _, prefix := range sg.Spec.Prefixes {
+				incrWatchPrefix(appName, dbName, prefix)
+			}
 			return false
+		})
+
+		if false {
+			// Fetch the sync state.
+			ds, err := getDbSyncState(ctx, st)
+			if err != nil && verror.ErrorID(err) != verror.ErrNoExist.ID {
+				errFinal = err
+				return false
+			}
+			var scanStart, scanLimit []byte
+			// Figure out what to scan among local log records.
+			if verror.ErrorID(err) == verror.ErrNoExist.ID {
+				scanStart, scanLimit = util.ScanPrefixArgs(logRecsPerDeviceScanPrefix(s.id), "")
+			} else {
+				scanStart, scanLimit = util.ScanPrefixArgs(logRecKey(s.id, ds.Gen), "")
+			}
+			var maxpos uint64
+			var dbName string
+			// Scan local log records to find the most recent one.
+			st.Scan(scanStart, scanLimit)
+			// Scan remote log records using the persisted GenVector.
+			s.syncState[dbName] = &dbSyncStateInMem{pos: maxpos + 1}
 		}
 
-		var scanStart, scanLimit []byte
-		// Figure out what to scan among local log records.
-		if verror.ErrorID(err) == verror.ErrNoExist.ID {
-			scanStart, scanLimit = util.ScanPrefixArgs(logRecsPerDeviceScanPrefix(s.id), "")
-		} else {
-			scanStart, scanLimit = util.ScanPrefixArgs(logRecKey(s.id, ds.Gen), "")
-		}
-
-		var maxpos uint64
-		var dbName string
-		// Scan local log records to find the most recent one.
-		sn.Scan(scanStart, scanLimit)
-
-		// Scan remote log records using the persisted GenVector.
-
-		s.syncState[dbName] = &dbSyncStateInMem{pos: maxpos + 1}
-
 		return false
 	})
 
-	return errFinal*/
+	return errFinal
 }
 
 // reserveGenAndPosInDbLog reserves a chunk of generation numbers and log
diff --git a/x/ref/services/syncbase/vsync/syncgroup.go b/x/ref/services/syncbase/vsync/syncgroup.go
index 823aeb1..79cc9bf 100644
--- a/x/ref/services/syncbase/vsync/syncgroup.go
+++ b/x/ref/services/syncbase/vsync/syncgroup.go
@@ -209,7 +209,6 @@
 
 	// Create a new aggregate view of SyncGroup members across all app databases.
 	newMembers := make(map[string]*memberInfo)
-	scanStart, scanLimit := util.ScanPrefixArgs(sgDataKeyScanPrefix(), "")
 
 	s.forEachDatabaseStore(ctx, func(appName, dbName string, st store.Store) bool {
 		// For each database, fetch its SyncGroup data entries by scanning their
@@ -218,14 +217,7 @@
 		defer sn.Close()
 		name := appDbName(appName, dbName)
 
-		stream := sn.Scan(scanStart, scanLimit)
-		for stream.Advance() {
-			var sg interfaces.SyncGroup
-			if vom.Decode(stream.Value(nil), &sg) != nil {
-				vlog.Errorf("invalid SyncGroup value for key %s", string(stream.Key(nil)))
-				continue
-			}
-
+		forEachSyncGroup(sn, func(sg *interfaces.SyncGroup) bool {
 			// Add all members of this SyncGroup to the membership view.
 			// A member's info is different across SyncGroups, so gather all of them.
 			for member, info := range sg.Joiners {
@@ -237,7 +229,8 @@
 				}
 				newMembers[member].db2sg[name][sg.Id] = info
 			}
-		}
+			return false
+		})
 		return false
 	})
 
@@ -245,6 +238,30 @@
 	view.expiration = time.Now().Add(memberViewTTL)
 }
 
+// forEachSyncGroup iterates over all SyncGroups in the Database and invokes
+// the callback function on each one.  The callback returns a "done" flag to
+// make forEachSyncGroup() stop the iteration earlier; otherwise the function
+// loops across all SyncGroups in the Database.
+func forEachSyncGroup(st store.StoreReader, callback func(*interfaces.SyncGroup) bool) {
+	scanStart, scanLimit := util.ScanPrefixArgs(sgDataKeyScanPrefix(), "")
+	stream := st.Scan(scanStart, scanLimit)
+	for stream.Advance() {
+		var sg interfaces.SyncGroup
+		if vom.Decode(stream.Value(nil), &sg) != nil {
+			vlog.Errorf("invalid SyncGroup value for key %s", string(stream.Key(nil)))
+			continue
+		}
+
+		if callback(&sg) {
+			break // done, early exit
+		}
+	}
+
+	if err := stream.Err(); err != nil {
+		vlog.Errorf("forEachSyncGroup: scan stream error: %v", err)
+	}
+}
+
 // getMembers returns all SyncGroup members and the count of SyncGroups each one
 // joined.
 func (s *syncService) getMembers(ctx *context.T) map[string]uint32 {
diff --git a/x/ref/services/syncbase/vsync/util.go b/x/ref/services/syncbase/vsync/util.go
index 62edcd2..65d47eb 100644
--- a/x/ref/services/syncbase/vsync/util.go
+++ b/x/ref/services/syncbase/vsync/util.go
@@ -7,6 +7,8 @@
 // Sync utility functions
 
 import (
+	"time"
+
 	"v.io/syncbase/x/ref/services/syncbase/store"
 	"v.io/v23/context"
 	"v.io/v23/rpc"
@@ -14,6 +16,10 @@
 	"v.io/x/lib/vlog"
 )
 
+const (
+	nanoPerSec = int64(1000000000)
+)
+
 // forEachDatabaseStore iterates over all Databases in all Apps within the
 // service and invokes the callback function on each database. The callback
 // returns a "done" flag to make forEachDatabaseStore() stop the iteration
@@ -76,3 +82,11 @@
 	}
 	return verror.New(verror.ErrInternal, ctx, key, err)
 }
+
+// unixNanoToTime converts a Unix timestamp in nanoseconds to a Time object.
+func unixNanoToTime(timestamp int64) time.Time {
+	if timestamp < 0 {
+		vlog.Fatalf("unixNanoToTime: invalid timestamp %d", timestamp)
+	}
+	return time.Unix(timestamp/nanoPerSec, timestamp%nanoPerSec)
+}
diff --git a/x/ref/services/syncbase/vsync/watcher.go b/x/ref/services/syncbase/vsync/watcher.go
index 5ac1b52..bbd1ac8 100644
--- a/x/ref/services/syncbase/vsync/watcher.go
+++ b/x/ref/services/syncbase/vsync/watcher.go
@@ -4,30 +4,159 @@
 
 package vsync
 
-import (
-	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
-	"v.io/syncbase/x/ref/services/syncbase/store"
+// Syncbase Watcher is a goroutine that listens to local Database updates from
+// applications and modifies sync metadata (e.g. DAG and local log records).
+// The coupling between Syncbase storage and sync is loose, via asynchronous
+// listening by the Watcher, to unblock the application operations as soon as
+// possible, and offload the sync metadata update to the Watcher.  When the
+// application mutates objects in a Database, additional entries are written
+// to a log queue, persisted in the same Database.  This queue is read by the
+// sync Watcher to learn of the changes.
 
+import (
+	"strings"
+	"time"
+
+	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
+	"v.io/syncbase/x/ref/services/syncbase/server/util"
+	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
+	"v.io/syncbase/x/ref/services/syncbase/store"
 	"v.io/v23/context"
 	"v.io/v23/verror"
+	"v.io/v23/vom"
+	"v.io/x/lib/vlog"
 )
 
-// When applications update objects in the local Store, the sync
-// watcher thread learns about them asynchronously via the "watch"
-// stream of object mutations. In turn, this sync watcher thread
-// updates the DAG and log records to track the object change
-// histories.
+var (
+	// watchPollInterval is the duration between consecutive watch polling
+	// events across all app databases.  Every watch event loops across all
+	// app databases and fetches from each one at most one batch update
+	// (transaction) to process.
+	// TODO(rdaoud): add a channel between store and watch to get change
+	// notifications instead of using a polling solution.
+	watchPollInterval = 100 * time.Millisecond
 
-// watchStore processes updates obtained by watching the store.
-func (s *syncService) watchStore() {
+	// watchPrefixes is an in-memory cache of SyncGroup prefixes for each
+	// app database.  It is filled at startup from persisted SyncGroup data
+	// and updated at runtime when SyncGroups are joined or left.  It is
+	// not guarded by a mutex because only the watcher goroutine uses it
+	// beyond the startup phase (before any sync goroutines are started).
+	// The map keys are the appdb names (globally unique).
+	watchPrefixes = make(map[string]sgPrefixes)
+)
+
+// sgPrefixes tracks SyncGroup prefixes being synced in a database and their
+// counts.
+type sgPrefixes map[string]uint32
+
+// watchStore processes updates obtained by watching the store.  This is the
+// sync watcher goroutine that learns about store updates asynchronously by
+// reading log records that track object mutation histories in each database.
+// For each batch mutation, the watcher updates the sync DAG and log records.
+// When an application makes a single non-transactional put, it is represented
+// as a batch of one log record. Thus the watcher only deals with batches.
+func (s *syncService) watchStore(ctx *context.T) {
 	defer s.pending.Done()
+
+	ticker := time.NewTicker(watchPollInterval)
+	defer ticker.Stop()
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	for {
+		select {
+		case <-s.closed:
+			vlog.VI(1).Info("watchStore: sync channel closed, stop watching and exit")
+			return
+
+		case <-ticker.C:
+			s.forEachDatabaseStore(ctx, func(appName, dbName string, st store.Store) bool {
+				s.processDatabase(ctx, appName, dbName, st)
+				return false
+			})
+		}
+	}
 }
 
-// TODO(hpucha): This is a skeleton only to drive the change for log.
+// processDatabase fetches from the given database at most one new batch update
+// (transaction) and processes it.  The one-batch limit prevents one database
+// from starving others.  A batch is stored as a contiguous set of log records
+// ending with one record having the "continued" flag set to false.
+func (s *syncService) processDatabase(ctx *context.T, appName, dbName string, st store.Store) {
+	resMark, err := getResMark(ctx, st)
+	if err != nil {
+		if verror.ErrorID(err) != verror.ErrNoExist.ID {
+			vlog.Errorf("processDatabase: %s, %s: cannot get resMark: %v", appName, dbName, err)
+			return
+		}
+		resMark = ""
+	}
+
+	// Get a batch of watch log entries, if any, after this resume marker.
+	if logs, nextResmark := getWatchLogBatch(ctx, appName, dbName, st, resMark); logs != nil {
+		s.processWatchLogBatch(ctx, appName, dbName, st, logs, nextResmark)
+	}
+}
+
+// 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, appName, dbName string, st store.Store, logs []*watchable.LogEntry, resMark string) {
+	if len(logs) == 0 {
+		return
+	}
+
+	// First handle SyncGroup prefix changes within the batch by updating
+	// the watch prefixes.  It is as if these log entries were at the start
+	// of the batch.  Exclude them from the actual data batch.
+	dataLogs := make([]*watchable.LogEntry, 0, len(logs))
+	for _, entry := range logs {
+		if processSyncGroupLogRecord(appName, dbName, entry) == false {
+			dataLogs = append(dataLogs, entry)
+		}
+	}
+
+	// Filter out the log entries for keys not part of any SyncGroup.
+	// TODO(rdaoud): filter out entries made by sync (echo suppression).
+	totalCount := uint64(len(dataLogs))
+	appdb := appDbName(appName, dbName)
+	logs = make([]*watchable.LogEntry, 0, len(dataLogs))
+
+	for _, entry := range dataLogs {
+		if syncable(appdb, entry) {
+			logs = append(logs, entry)
+		}
+	}
+	dataLogs = nil
+
+	// Transactional processing of the batch: convert these syncable log
+	// records to a batch of sync log records, filling their parent versions
+	// from the DAG head nodes.
+	err := store.RunInTransaction(st, func(tx store.StoreReadWriter) error {
+		batch := make([]*localLogRec, 0, len(logs))
+		for _, entry := range logs {
+			if rec := convertLogRecord(ctx, tx, entry); rec != nil {
+				batch = append(batch, rec)
+			}
+		}
+
+		if err := s.processBatch(ctx, appName, dbName, batch, totalCount, tx); err != nil {
+			return err
+		}
+		return setResMark(ctx, tx, resMark)
+	})
+
+	if err != nil {
+		// TODO(rdaoud): don't crash, quarantine this app database.
+		vlog.Fatalf("processDatabase: %s, %s: watcher cannot process batch: %v", appName, dbName, err)
+	}
+}
 
 // processBatch applies a single batch of changes (object mutations) received
 // from watching a particular Database.
-func (s *syncService) processBatch(ctx *context.T, appName, dbName string, batch []*localLogRec, tx store.StoreReadWriter) error {
+func (s *syncService) processBatch(ctx *context.T, appName, dbName string, batch []*localLogRec, totalCount uint64, tx store.StoreReadWriter) error {
 	_ = tx.(store.Transaction)
 
 	count := uint64(len(batch))
@@ -37,7 +166,7 @@
 
 	// If the batch has more than one mutation, start a batch for it.
 	batchId := NoBatchId
-	if count > 1 {
+	if totalCount > 1 {
 		batchId = s.startBatch(ctx, tx, batchId)
 		if batchId == NoBatchId {
 			return verror.New(verror.ErrInternal, ctx, "failed to generate batch ID")
@@ -47,7 +176,6 @@
 	gen, pos := s.reserveGenAndPosInDbLog(ctx, appName, dbName, count)
 
 	for _, rec := range batch {
-
 		// Update the log record. Portions of the record Metadata must
 		// already be filled.
 		rec.Metadata.Id = s.id
@@ -55,7 +183,7 @@
 		rec.Metadata.RecType = interfaces.NodeRec
 
 		rec.Metadata.BatchId = batchId
-		rec.Metadata.BatchCount = count
+		rec.Metadata.BatchCount = totalCount
 
 		rec.Pos = pos
 
@@ -77,7 +205,7 @@
 	return nil
 }
 
-// processLogRec processes a local log record by adding to the Database and
+// processLocalLogRec processes a local log record by adding to the Database and
 // suitably updating the DAG metadata.
 func (s *syncService) processLocalLogRec(ctx *context.T, tx store.StoreReadWriter, rec *localLogRec) error {
 	// Insert the new log record into the log.
@@ -96,3 +224,216 @@
 	// Move the head.
 	return moveHead(ctx, tx, m.ObjId, m.CurVers)
 }
+
+// incrWatchPrefix increments (or sets) a SyncGroup prefix for an app database
+// in the watch prefix cache.
+func incrWatchPrefix(appName, dbName, prefix string) {
+	name := appDbName(appName, dbName)
+	if pfxs := watchPrefixes[name]; pfxs != nil {
+		pfxs[prefix]++ // it auto-initializes a non-existent prefix
+	} else {
+		watchPrefixes[name] = sgPrefixes{prefix: 1}
+	}
+}
+
+// decrWatchPrefix decrements (or unsets) a SyncGroup prefix for an app database
+// in the watch prefix cache.
+func decrWatchPrefix(appName, dbName, prefix string) {
+	name := appDbName(appName, dbName)
+	if pfxs := watchPrefixes[name]; pfxs != nil {
+		if pfxs[prefix] > 1 {
+			pfxs[prefix]--
+		} else if len(pfxs) > 1 {
+			delete(pfxs, prefix)
+		} else {
+			delete(watchPrefixes, name)
+		}
+	}
+}
+
+// dbLogScanArgs determines the arguments needed to start a new scan from a
+// given resume marker (last log entry read).  An empty resume marker is used
+// to begin the scan from the start of the log.
+func dbLogScanArgs(resMark string) ([]byte, []byte) {
+	start, limit := util.ScanPrefixArgs(util.LogPrefix, "")
+	if resMark != "" {
+		// To start just after the current resume marker, augment it by
+		// appending an extra byte at the end.  Use byte value zero to
+		// use the lowest value possible.  This works because resume
+		// markers have a fixed length and are sorted lexicographically.
+		// By creationg a fake longer resume marker that falls between
+		// real resume markers, the next scan will start right after
+		// where the previous one stopped without missing data.
+		start = append([]byte(resMark), 0)
+	}
+	return start, limit
+}
+
+// getWatchLogBatch 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 getWatchLogBatch(ctx *context.T, appName, dbName string, st store.Store, resMark string) ([]*watchable.LogEntry, string) {
+	scanStart, scanLimit := dbLogScanArgs(resMark)
+	endOfBatch := false
+	var newResmark string
+
+	// Use the store directly to scan these read-only log entries, no need
+	// to create a snapshot since they are never overwritten.  Read and
+	// buffer a batch before processing it.
+	var logs []*watchable.LogEntry
+	stream := st.Scan(scanStart, scanLimit)
+	for stream.Advance() {
+		logKey := string(stream.Key(nil))
+		var logEnt watchable.LogEntry
+		if vom.Decode(stream.Value(nil), &logEnt) != nil {
+			vlog.Fatalf("getWatchLogBatch: %s, %s: invalid watch LogEntry %s: %v",
+				appName, dbName, logKey, stream.Value(nil))
+		}
+
+		logs = append(logs, &logEnt)
+
+		// Stop if this is the end of the batch.
+		if logEnt.Continued == false {
+			newResmark = logKey
+			endOfBatch = true
+			break
+		}
+	}
+
+	if err := stream.Err(); err != nil {
+		vlog.Errorf("getWatchLogBatch: %s, %s: scan stream error: %v", appName, dbName, err)
+		return nil, resMark
+	}
+	if !endOfBatch {
+		if len(logs) > 0 {
+			vlog.Fatalf("processDatabase: %s, %s: end of batch not found after %d entries",
+				appName, dbName, len(logs))
+		}
+		return nil, resMark
+	}
+	return logs, newResmark
+}
+
+// convertLogRecord converts a store log entry to a sync log record.  It fills
+// the previous object version (parent) by fetching its current DAG head if it
+// has one.  For a delete, it generates a new object version because the store
+// does not version a deletion.
+// TODO(rdaoud): change Syncbase to store and version a deleted object to
+// simplify the store-to-sync interaction.  A deleted key would still have a
+// version and its value entry would encode the "deleted" flag, either in the
+// key or probably in a value wrapper that would contain other metadata.
+func convertLogRecord(ctx *context.T, tx store.StoreReadWriter, logEnt *watchable.LogEntry) *localLogRec {
+	_ = tx.(store.Transaction)
+
+	switch op := logEnt.Op.(type) {
+	case *watchable.OpGet:
+		// TODO(rdaoud): save read-set in sync.
+		return nil
+
+	case *watchable.OpScan:
+		// TODO(rdaoud): save scan-set in sync.
+		return nil
+
+	case *watchable.OpPut:
+		rec := localLogRec{}
+		oid := string(op.Value.Key)
+		rec.Metadata.ObjId = oid
+		rec.Metadata.CurVers = string(op.Value.Version)
+		if head, err := getHead(ctx, tx, oid); err == nil {
+			rec.Metadata.Parents = []string{head}
+		} else if verror.ErrorID(err) != verror.ErrNoExist.ID {
+			vlog.Fatalf("cannot getHead to convert Put log record for %s: %v", oid, err)
+		}
+		rec.Metadata.UpdTime = unixNanoToTime(logEnt.CommitTimestamp)
+		return &rec
+
+	case *watchable.OpDelete:
+		rec := localLogRec{}
+		oid := string(op.Value.Key)
+		rec.Metadata.ObjId = oid
+		rec.Metadata.CurVers = string(watchable.NewVersion())
+		rec.Metadata.Delete = true
+		if head, err := getHead(ctx, tx, oid); err == nil {
+			rec.Metadata.Parents = []string{head}
+		} else {
+			vlog.Fatalf("cannot getHead to convert Delete log record for %s: %v", oid, err)
+		}
+		rec.Metadata.UpdTime = unixNanoToTime(logEnt.CommitTimestamp)
+		return &rec
+
+	case *watchable.OpSyncGroup:
+		vlog.Errorf("watch LogEntry for SyncGroup should not be converted: %v", logEnt)
+		return nil
+
+	default:
+		vlog.Errorf("invalid watch LogEntry: %v", logEnt)
+		return nil
+	}
+}
+
+// processSyncGroupLogRecord checks if the log entry is a SyncGroup update and,
+// if it is, updates the watch prefixes for the app database and returns true.
+// Otherwise it returns false with no other changes.
+func processSyncGroupLogRecord(appName, dbName string, logEnt *watchable.LogEntry) bool {
+	switch op := logEnt.Op.(type) {
+	case *watchable.OpSyncGroup:
+		remove := op.Value.Remove
+		for _, prefix := range op.Value.Prefixes {
+			if remove {
+				decrWatchPrefix(appName, dbName, prefix)
+			} else {
+				incrWatchPrefix(appName, dbName, prefix)
+			}
+		}
+		return true
+
+	default:
+		return false
+	}
+}
+
+// syncable returns true if the given log entry falls within the scope of a
+// SyncGroup prefix for the given app database, and thus should be synced.
+// It is used to pre-filter the batch of log entries before sync processing.
+func syncable(appdb string, logEnt *watchable.LogEntry) bool {
+	var key string
+	switch op := logEnt.Op.(type) {
+	case *watchable.OpPut:
+		key = string(op.Value.Key)
+	case *watchable.OpDelete:
+		key = string(op.Value.Key)
+	default:
+		return false
+	}
+
+	for prefix := range watchPrefixes[appdb] {
+		if strings.HasPrefix(key, prefix) {
+			return true
+		}
+	}
+	return false
+}
+
+// resMarkKey returns the key used to access the watcher resume marker.
+func resMarkKey() string {
+	return util.JoinKeyParts(util.SyncPrefix, "w", "rm")
+}
+
+// setResMark stores the watcher resume marker for a database.
+func setResMark(ctx *context.T, tx store.StoreReadWriter, resMark string) error {
+	_ = tx.(store.Transaction)
+
+	if err := util.PutObject(tx, resMarkKey(), resMark); err != nil {
+		return verror.New(verror.ErrInternal, ctx, err)
+	}
+	return nil
+}
+
+// getResMark retrieves the watcher resume marker for a database.
+func getResMark(ctx *context.T, st store.StoreReader) (string, error) {
+	var resMark string
+	key := resMarkKey()
+	if err := util.GetObject(st, key, &resMark); err != nil {
+		return NoVersion, translateError(ctx, err, key)
+	}
+	return resMark, nil
+}
diff --git a/x/ref/services/syncbase/vsync/watcher_test.go b/x/ref/services/syncbase/vsync/watcher_test.go
new file mode 100644
index 0000000..ee5e600
--- /dev/null
+++ b/x/ref/services/syncbase/vsync/watcher_test.go
@@ -0,0 +1,236 @@
+// Copyright 2015 The Vanadium 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 vsync
+
+// Tests for the sync watcher in Syncbase.
+
+import (
+	"reflect"
+	"testing"
+
+	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
+)
+
+// TestSetResmark tests setting and getting a resume marker.
+func TestSetResmark(t *testing.T) {
+	svc := createService(t)
+	defer destroyService(t, svc)
+	st := svc.St()
+
+	resmark, err := getResMark(nil, st)
+	if err == nil || resmark != "" {
+		t.Errorf("found non-existent resume marker: %s, %v", resmark, err)
+	}
+
+	wantResmark := "1234567890"
+	tx := st.NewTransaction()
+	if err := setResMark(nil, tx, wantResmark); err != nil {
+		t.Errorf("cannot set resume marker: %v", err)
+	}
+	tx.Commit()
+
+	resmark, err = getResMark(nil, st)
+	if err != nil {
+		t.Errorf("cannot get new resume marker: %v", err)
+	}
+	if resmark != wantResmark {
+		t.Errorf("invalid new resume: got %s instead of %s", resmark, wantResmark)
+	}
+}
+
+// TestWatchPrefixes tests setting and updating the watch prefixes map.
+func TestWatchPrefixes(t *testing.T) {
+	svc := createService(t)
+	defer destroyService(t, svc)
+
+	if len(watchPrefixes) != 0 {
+		t.Errorf("watch prefixes not empty: %v", watchPrefixes)
+	}
+
+	watchPrefixOps := []struct {
+		appName, dbName, key string
+		incr                 bool
+	}{
+		{"app1", "db1", "foo", true},
+		{"app1", "db1", "bar", true},
+		{"app2", "db1", "xyz", true},
+		{"app3", "db1", "haha", true},
+		{"app1", "db1", "foo", true},
+		{"app1", "db1", "foo", true},
+		{"app1", "db1", "foo", false},
+		{"app2", "db1", "ttt", true},
+		{"app2", "db1", "ttt", true},
+		{"app2", "db1", "ttt", false},
+		{"app2", "db1", "ttt", false},
+		{"app2", "db2", "qwerty", true},
+		{"app3", "db1", "haha", true},
+		{"app2", "db2", "qwerty", false},
+		{"app3", "db1", "haha", false},
+	}
+
+	for _, op := range watchPrefixOps {
+		if op.incr {
+			incrWatchPrefix(op.appName, op.dbName, op.key)
+		} else {
+			decrWatchPrefix(op.appName, op.dbName, op.key)
+		}
+	}
+
+	expPrefixes := map[string]sgPrefixes{
+		"app1:db1": sgPrefixes{"foo": 2, "bar": 1},
+		"app2:db1": sgPrefixes{"xyz": 1},
+		"app3:db1": sgPrefixes{"haha": 1},
+	}
+	if !reflect.DeepEqual(watchPrefixes, expPrefixes) {
+		t.Errorf("invalid watch prefixes: got %v instead of %v", watchPrefixes, expPrefixes)
+	}
+
+	checkSyncableTests := []struct {
+		appName, dbName, key string
+		result               bool
+	}{
+		{"app1", "db1", "foo", true},
+		{"app1", "db1", "foobar", true},
+		{"app1", "db1", "bar", true},
+		{"app1", "db1", "bar123", true},
+		{"app1", "db1", "f", false},
+		{"app1", "db1", "ba", false},
+		{"app1", "db1", "xyz", false},
+		{"app1", "db555", "foo", false},
+		{"app555", "db1", "foo", false},
+		{"app2", "db1", "xyz123", true},
+		{"app2", "db1", "ttt123", false},
+		{"app2", "db2", "qwerty", false},
+		{"app3", "db1", "hahahoho", true},
+		{"app3", "db1", "hoho", false},
+		{"app3", "db1", "h", false},
+	}
+
+	for _, test := range checkSyncableTests {
+		log := &watchable.LogEntry{
+			Op: &watchable.OpPut{
+				watchable.PutOp{Key: []byte(test.key)},
+			},
+		}
+		res := syncable(appDbName(test.appName, test.dbName), log)
+		if res != test.result {
+			t.Errorf("checkSyncable: invalid output: %s, %s, %s: got %t instead of %t",
+				test.appName, test.dbName, test.key, res, test.result)
+		}
+	}
+}
+
+// newLog creates a Put or Delete watch log entry.
+func newLog(key, version string, delete bool) *watchable.LogEntry {
+	k, v := []byte(key), []byte(version)
+	log := &watchable.LogEntry{}
+	if delete {
+		log.Op = &watchable.OpDelete{watchable.DeleteOp{Key: k}}
+	} else {
+		log.Op = &watchable.OpPut{watchable.PutOp{Key: k, Version: v}}
+	}
+	return log
+}
+
+// newSGLog creates a SyncGroup watch log entry.
+func newSGLog(prefixes []string, remove bool) *watchable.LogEntry {
+	return &watchable.LogEntry{
+		Op: &watchable.OpSyncGroup{
+			watchable.SyncGroupOp{Prefixes: prefixes, Remove: remove},
+		},
+	}
+}
+
+// TestProcessWatchLogBatch tests the processing of a batch of log records.
+func TestProcessWatchLogBatch(t *testing.T) {
+	svc := createService(t)
+	defer destroyService(t, svc)
+	st := svc.St()
+	s := svc.sync
+
+	app, db := "mockapp", "mockdb"
+
+	// Empty logs does not fail.
+	s.processWatchLogBatch(nil, app, db, st, nil, "")
+
+	// Non-syncable logs.
+	batch := []*watchable.LogEntry{
+		newLog("foo", "123", false),
+		newLog("bar", "555", false),
+	}
+
+	resmark := "abcd"
+	s.processWatchLogBatch(nil, app, db, st, batch, resmark)
+
+	if res, err := getResMark(nil, st); err != nil && res != resmark {
+		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
+	}
+	if hasNode(nil, st, "foo", "123") || hasNode(nil, st, "bar", "555") {
+		t.Error("hasNode() found DAG entries for non-syncable logs")
+	}
+
+	// Partially syncable logs.
+	batch = []*watchable.LogEntry{
+		newSGLog([]string{"f", "x"}, false),
+		newLog("foo", "333", false),
+		newLog("fooxyz", "444", false),
+		newLog("bar", "222", false),
+	}
+
+	resmark = "cdef"
+	s.processWatchLogBatch(nil, app, db, st, batch, resmark)
+
+	if res, err := getResMark(nil, st); err != nil && res != resmark {
+		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
+	}
+	if head, err := getHead(nil, st, "foo"); err != nil && head != "333" {
+		t.Errorf("getHead() did not find foo: %s, %v", head, err)
+	}
+	node, err := getNode(nil, st, "foo", "333")
+	if err != nil {
+		t.Errorf("getNode() did not find foo: %v", err)
+	}
+	if node.Level != 0 || node.Parents != nil || node.Logrec == "" || node.BatchId == NoBatchId {
+		t.Errorf("invalid DAG node for foo: %v", node)
+	}
+	node2, err := getNode(nil, st, "fooxyz", "444")
+	if err != nil {
+		t.Errorf("getNode() did not find fooxyz: %v", err)
+	}
+	if node2.Level != 0 || node2.Parents != nil || node2.Logrec == "" || node2.BatchId != node.BatchId {
+		t.Errorf("invalid DAG node for fooxyz: %v", node2)
+	}
+	if hasNode(nil, st, "bar", "222") {
+		t.Error("hasNode() found DAG entries for non-syncable logs")
+	}
+
+	// Back to non-syncable logs (remove "f" prefix).
+	batch = []*watchable.LogEntry{
+		newSGLog([]string{"f"}, true),
+		newLog("foo", "99", false),
+		newLog("fooxyz", "888", true),
+		newLog("bar", "007", false),
+	}
+
+	resmark = "tuvw"
+	s.processWatchLogBatch(nil, app, db, st, batch, resmark)
+
+	if res, err := getResMark(nil, st); err != nil && res != resmark {
+		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
+	}
+	// No changes to "foo".
+	if head, err := getHead(nil, st, "foo"); err != nil && head != "333" {
+		t.Errorf("getHead() did not find foo: %s, %v", head, err)
+	}
+	if node, err := getNode(nil, st, "foo", "99"); err == nil {
+		t.Errorf("getNode() should not have found foo @ 99: %v", node)
+	}
+	if node, err := getNode(nil, st, "fooxyz", "888"); err == nil {
+		t.Errorf("getNode() should not have found fooxyz @ 888: %v", node)
+	}
+	if hasNode(nil, st, "bar", "007") {
+		t.Error("hasNode() found DAG entries for non-syncable logs")
+	}
+}