diff --git a/v23/syncbase/nosql/syncgroup_v23_test.go b/v23/syncbase/nosql/syncgroup_v23_test.go
index ff4b5cc..849f352 100644
--- a/v23/syncbase/nosql/syncgroup_v23_test.go
+++ b/v23/syncbase/nosql/syncgroup_v23_test.go
@@ -6,18 +6,24 @@
 
 import (
 	"fmt"
+	"time"
 
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
 	"v.io/syncbase/v23/syncbase"
+	"v.io/syncbase/v23/syncbase/nosql"
 	tu "v.io/syncbase/v23/syncbase/testutil"
 	constants "v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/v23"
 	"v.io/v23/naming"
+	"v.io/x/ref"
 	_ "v.io/x/ref/runtime/factories/generic"
 	"v.io/x/ref/test/modules"
 	"v.io/x/ref/test/v23tests"
 )
 
+// TODO(rdaoud): change the credentials of servers s0 and s1 to independent
+// principals derived from the same root.
+
 //go:generate v23 test generate
 
 func V23TestSyncbasedJoinSyncGroup(t *v23tests.T) {
@@ -47,10 +53,13 @@
 	d := a.NoSQLDatabase("d")
 	d.Create(ctx, nil)
 
+	mtName := env.Vars[ref.EnvNamespacePrefix]
+
 	spec := wire.SyncGroupSpec{
 		Description: "test syncgroup sg",
 		Perms:       perms("root/s0", "root/s0/s1"),
-		Prefixes:    []string{"t1/foo"},
+		Prefixes:    []string{"t1:foo"},
+		MountTables: []string{mtName},
 	}
 	sgName := naming.Join("sync0", constants.SyncbaseSuffix, "foo")
 	sg := d.SyncGroup(sgName)
@@ -74,7 +83,133 @@
 	sg := d.SyncGroup(sgName)
 	info := wire.SyncGroupMemberInfo{10}
 	if _, err := sg.Join(ctx, info); err != nil {
-		return fmt.Errorf("Join SG %v failed: %v", sgName, err)
+		return fmt.Errorf("Join SG %q failed: %v", sgName, err)
 	}
 	return nil
 }, "runJoinSyncGroup")
+
+// V23TestSyncbasedGetDeltas tests the exchange of deltas between two Syncbase
+// instances and their clients.  The 1st client creates a SyncGroup and puts
+// some database entries in it.  The 2nd client joins that SyncGroup and reads
+// the database entries.  This verifies the end-to-end synchronization of data
+// along the path: client1--Syncbase1--Syncbase2--client2.
+func V23TestSyncbasedGetDeltas(t *v23tests.T) {
+	v23tests.RunRootMT(t, "--v23.tcp.address=127.0.0.1:0")
+	server0Creds, _ := t.Shell().NewChildCredentials("s0")
+	client0Creds, _ := t.Shell().NewChildCredentials("s0/c0")
+	cleanSync0 := tu.StartSyncbased(t, server0Creds, "sync0", "",
+		`{"Read": {"In":["root/s0/c0"]}, "Write": {"In":["root/s0/c0"]}}`)
+	defer cleanSync0()
+
+	server1Creds, _ := t.Shell().NewChildCredentials("s0/s1")
+	client1Creds, _ := t.Shell().NewChildCredentials("s0/s1/c1")
+	cleanSync1 := tu.StartSyncbased(t, server1Creds, "sync1", "",
+		`{"Read": {"In":["root/s0/s1/c1"]}, "Write": {"In":["root/s0/s1/c1"]}}`)
+	defer cleanSync1()
+
+	tu.RunClient(t, client0Creds, runCreateAndPopulateSyncGroup)
+	tu.RunClient(t, client1Creds, runJoinSyncGroupAndFetchData)
+}
+
+var runCreateAndPopulateSyncGroup = modules.Register(func(env *modules.Env, args ...string) error {
+	ctx, shutdown := v23.Init()
+	defer shutdown()
+
+	a := syncbase.NewService("sync0").App("a")
+	a.Create(ctx, nil)
+	d := a.NoSQLDatabase("d")
+	d.Create(ctx, nil)
+	d.CreateTable(ctx, "tb", nil)
+
+	mtName := env.Vars[ref.EnvNamespacePrefix]
+
+	spec := wire.SyncGroupSpec{
+		Description: "test syncgroup sg",
+		Perms:       perms("root/s0", "root/s0/s1"),
+		Prefixes:    []string{"tb:foo"},
+		MountTables: []string{mtName},
+	}
+	sgName := naming.Join("sync0", constants.SyncbaseSuffix, "foo")
+	sg := d.SyncGroup(sgName)
+	info := wire.SyncGroupMemberInfo{8}
+	if err := sg.Create(ctx, spec, info); err != nil {
+		return fmt.Errorf("Create SG %q failed: %v", sgName, err)
+	}
+
+	// Do Puts.
+	tb := d.Table("tb")
+	for i := 0; i < 10; i++ {
+		key := fmt.Sprintf("foo%d", i)
+		r := tb.Row(key)
+		if err := r.Put(ctx, "testkey"+key); err != nil {
+			return fmt.Errorf("r.Put() failed: %v", err)
+		}
+	}
+
+	return nil
+}, "runCreateAndPopulateSyncGroup")
+
+var runJoinSyncGroupAndFetchData = modules.Register(func(env *modules.Env, args ...string) error {
+	ctx, shutdown := v23.Init()
+	defer shutdown()
+
+	a := syncbase.NewService("sync1").App("a")
+	a.Create(ctx, nil)
+	d := a.NoSQLDatabase("d")
+	d.Create(ctx, nil)
+	d.CreateTable(ctx, "tb", nil)
+
+	sgName := naming.Join("sync0", constants.SyncbaseSuffix, "foo")
+	sg := d.SyncGroup(sgName)
+	info := wire.SyncGroupMemberInfo{10}
+	if _, err := sg.Join(ctx, info); err != nil {
+		return fmt.Errorf("Join SG %q failed: %v\n", sgName, err)
+	}
+
+	// Wait for a bit (up to 4 sec) until the last key appears.
+	tb := d.Table("tb")
+	r := tb.Row("foo9")
+	for i := 0; i < 8; i++ {
+		time.Sleep(500 * time.Millisecond)
+		var value string
+		if err := r.Get(ctx, &value); err == nil {
+			break
+		}
+	}
+
+	// Verify that all keys and values made it correctly.
+	for i := 0; i < 10; i++ {
+		key := fmt.Sprintf("foo%d", i)
+		r := tb.Row(key)
+		var got string
+		if err := r.Get(ctx, &got); err != nil {
+			return fmt.Errorf("r.Get() failed: %v\n", err)
+		}
+		want := "testkey" + key
+		if got != want {
+			return fmt.Errorf("unexpected value: got %q, want %q\n", got, want)
+		}
+	}
+
+	// Re-verify using a scan operation.
+	stream := tb.Scan(ctx, nosql.Prefix("foo"))
+	for i := 0; stream.Advance(); i++ {
+		want := fmt.Sprintf("foo%d", i)
+		got := stream.Key()
+		if got != want {
+			return fmt.Errorf("unexpected key in scan: got %q, want %q\n", got, want)
+		}
+		want = "testkey" + want
+		if err := stream.Value(&got); err != nil {
+			return fmt.Errorf("cannot fetch value in scan: %v\n", err)
+		}
+		if got != want {
+			return fmt.Errorf("unexpected value in scan: got %q, want %q\n", got, want)
+		}
+	}
+	if err := stream.Err(); err != nil {
+		return fmt.Errorf("scan stream error: %v\n", err)
+	}
+
+	return nil
+}, "runJoinSyncGroupAndFetchData")
diff --git a/v23/syncbase/nosql/v23_test.go b/v23/syncbase/nosql/v23_test.go
index efbdd39..a14dbe0 100644
--- a/v23/syncbase/nosql/v23_test.go
+++ b/v23/syncbase/nosql/v23_test.go
@@ -28,3 +28,7 @@
 func TestV23SyncbasedJoinSyncGroup(t *testing.T) {
 	v23tests.RunTest(t, V23TestSyncbasedJoinSyncGroup)
 }
+
+func TestV23SyncbasedGetDeltas(t *testing.T) {
+	v23tests.RunTest(t, V23TestSyncbasedGetDeltas)
+}
diff --git a/x/ref/services/syncbase/vsync/initiator.go b/x/ref/services/syncbase/vsync/initiator.go
index 1624301..1d447c5 100644
--- a/x/ref/services/syncbase/vsync/initiator.go
+++ b/x/ref/services/syncbase/vsync/initiator.go
@@ -10,6 +10,7 @@
 	"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"
@@ -124,7 +125,13 @@
 		}
 
 		iSt.stream = stream
-		req := interfaces.DeltaReq{SgIds: iSt.sgIds, InitVec: iSt.local}
+		req := interfaces.DeltaReq{
+			AppName: iSt.appName,
+			DbName:  iSt.dbName,
+			SgIds:   iSt.sgIds,
+			InitVec: iSt.local,
+		}
+
 		sender := iSt.stream.SendStream()
 		sender.Send(req)
 
@@ -264,7 +271,8 @@
 		return nil, false
 	}
 	for mt := range iSt.mtTables {
-		c := interfaces.SyncClient(naming.Join(mt, iSt.peer))
+		absName := naming.Join(mt, iSt.peer, util.SyncbaseSuffix)
+		c := interfaces.SyncClient(absName)
 		stream, err := c.GetDeltas(ctx)
 		if err == nil {
 			return stream, true
@@ -382,7 +390,6 @@
 				return verror.New(verror.ErrInternal, ctx, "received finish followed by finish in delta response stream")
 			}
 			finish = true
-			break
 
 		case interfaces.DeltaRespRespVec:
 			iSt.remote = v.Value
@@ -411,6 +418,11 @@
 			// Mark object dirty.
 			iSt.updObjects[rec.Metadata.ObjId] = &objConflictState{}
 		}
+
+		// Break out of the stream.
+		if finish {
+			break
+		}
 	}
 
 	if !(start && finish) {
@@ -514,6 +526,7 @@
 
 	for {
 		iSt.tx = iSt.st.NewTransaction()
+		watchable.SetTransactionFromSync(iSt.tx) // for echo-suppression
 
 		if err := iSt.detectConflicts(ctx); err != nil {
 			return err
diff --git a/x/ref/services/syncbase/vsync/initiator_test.go b/x/ref/services/syncbase/vsync/initiator_test.go
index bfcb250..d3c55d4 100644
--- a/x/ref/services/syncbase/vsync/initiator_test.go
+++ b/x/ref/services/syncbase/vsync/initiator_test.go
@@ -256,7 +256,7 @@
 	testIfMapArrEqual(t, iSt.sgPfxs, sg1.Spec.Prefixes)
 	testIfMapArrEqual(t, iSt.mtTables, sg1.Spec.MountTables)
 
-	s.syncState[gdb] = &dbSyncStateInMem{}
+	s.initDbSyncStateInMem(nil, "mockapp", "mockdb")
 
 	// Create local genvec so that it contains knowledge only about common prefixes.
 	if err := iSt.createLocalGenVec(nil); err != nil {
diff --git a/x/ref/services/syncbase/vsync/responder.go b/x/ref/services/syncbase/vsync/responder.go
index 19b1366..d096aa8 100644
--- a/x/ref/services/syncbase/vsync/responder.go
+++ b/x/ref/services/syncbase/vsync/responder.go
@@ -10,15 +10,17 @@
 	"strings"
 
 	"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/x/lib/vlog"
 )
 
 // GetDeltas implements the responder side of the GetDeltas RPC.
 func (s *syncService) GetDeltas(ctx *context.T, call interfaces.SyncGetDeltasServerCall) error {
 	recvr := call.RecvStream()
-
 	for recvr.Advance() {
 		req := recvr.Value()
 		// Ignoring errors since if one Database fails for any reason,
@@ -281,10 +283,11 @@
 
 		if !filterLogRec(rec, rSt.req.InitVec, initPfxs) {
 			// Send on the wire.
-			wireRec := interfaces.LogRec{Metadata: rec.Metadata}
-			// TODO(hpucha): Hash out this fake stream stuff when
-			// defining the RPC and the rest of the responder.
-			sender.Send(interfaces.DeltaRespRec{wireRec})
+			wireRec, err := makeWireLogRec(ctx, rSt.st, rec)
+			if err != nil {
+				return err
+			}
+			sender.Send(interfaces.DeltaRespRec{*wireRec})
 		}
 
 		// Add a new record from the same device if not done.
@@ -387,11 +390,19 @@
 
 // Note: initPfxs is sorted.
 func filterLogRec(rec *localLogRec, initVec interfaces.GenVector, initPfxs []string) bool {
-	filter := true
+	// The key starts with one of the store's reserved prefixes for managed
+	// namespaces (e.g. $row, $perms).  Remove that prefix before comparing
+	// it with the SyncGroup prefixes which are defined by the application.
+	parts := util.SplitKeyParts(rec.Metadata.ObjId)
+	if len(parts) < 2 {
+		vlog.Fatalf("filterLogRec: invalid entry key %s", rec.Metadata.ObjId)
+	}
+	key := util.JoinKeyParts(parts[1:]...)
 
+	filter := true
 	var maxGen uint64
 	for _, p := range initPfxs {
-		if strings.HasPrefix(rec.Metadata.ObjId, p) {
+		if strings.HasPrefix(key, p) {
 			// Do not filter. Initiator is interested in this
 			// prefix.
 			filter = false
@@ -406,12 +417,31 @@
 
 	// Filter this record if the initiator already has it.
 	if maxGen >= rec.Metadata.Gen {
-		return true
+		filter = true
 	}
 
 	return filter
 }
 
+// makeWireLogRec creates a sync log record to send on the wire from a given
+// local sync record.
+func makeWireLogRec(ctx *context.T, st store.Store, rec *localLogRec) (*interfaces.LogRec, error) {
+	// Get the object value at the required version.  Note: GetAtVersion()
+	// requires a transaction to read the data, so create and abort one.
+	// TODO(hpucha): remove the fake Tx after the change in GetAtVersion().
+	tx := st.NewTransaction()
+	defer tx.Abort()
+
+	key, version := rec.Metadata.ObjId, rec.Metadata.CurVers
+	value, err := watchable.GetAtVersion(ctx, tx, []byte(key), nil, []byte(version))
+	if err != nil {
+		return nil, err
+	}
+
+	wireRec := &interfaces.LogRec{Metadata: rec.Metadata, Value: value}
+	return wireRec, nil
+}
+
 // A minHeap implements heap.Interface and holds local log records.
 type minHeap []*localLogRec
 
diff --git a/x/ref/services/syncbase/vsync/responder_test.go b/x/ref/services/syncbase/vsync/responder_test.go
index 82c0d94..b452c07 100644
--- a/x/ref/services/syncbase/vsync/responder_test.go
+++ b/x/ref/services/syncbase/vsync/responder_test.go
@@ -12,6 +12,7 @@
 	"time"
 
 	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
+	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
 	"v.io/v23/naming"
 	"v.io/v23/rpc"
 	"v.io/v23/security"
@@ -378,7 +379,7 @@
 				if opfx == "" {
 					continue
 				}
-				okey := fmt.Sprintf("%s~%x", opfx, tRng.Int())
+				okey := makeRowKey(fmt.Sprintf("%s~%x", opfx, tRng.Int()))
 				vers := fmt.Sprintf("%x", tRng.Int())
 				rec := &localLogRec{
 					Metadata: interfaces.LogRecMetadata{Id: id, Gen: k, ObjId: okey, CurVers: vers, UpdTime: time.Now().UTC()},
@@ -387,6 +388,10 @@
 				if err := putLogRec(nil, tx, rec); err != nil {
 					t.Fatalf("putLogRec(%d:%d) failed rec %v err %v", id, k, rec, err)
 				}
+				value := fmt.Sprintf("value_%s", okey)
+				if err := watchable.PutAtVersion(nil, tx, []byte(okey), []byte(value), []byte(vers)); err != nil {
+					t.Fatalf("PutAtVersion(%d:%d) failed rec %v value %s: err %v", id, k, rec, value, err)
+				}
 
 				initPfxs := extractAndSortPrefixes(test.initVec)
 				if !filterLogRec(rec, test.initVec, initPfxs) {
diff --git a/x/ref/services/syncbase/vsync/sync_state.go b/x/ref/services/syncbase/vsync/sync_state.go
index 49d9ff5..8db0bc6 100644
--- a/x/ref/services/syncbase/vsync/sync_state.go
+++ b/x/ref/services/syncbase/vsync/sync_state.go
@@ -155,10 +155,23 @@
 		return verror.New(verror.ErrInternal, ctx, "db state not found", name)
 	}
 
-	ds.ckPtGen = ds.gen
+	// The frozen generation is the last generation number used, i.e. one
+	// below the next available one to use.
+	ds.ckPtGen = ds.gen - 1
 	return nil
 }
 
+// initDbSyncStateInMem initializes the in memory sync state of the Database if needed.
+func (s *syncService) initDbSyncStateInMem(ctx *context.T, appName, dbName string) {
+	s.syncStateLock.Lock()
+	defer s.syncStateLock.Unlock()
+
+	name := appDbName(appName, dbName)
+	if s.syncState[name] == nil {
+		s.syncState[name] = &dbSyncStateInMem{gen: 1}
+	}
+}
+
 // getDbSyncStateInMem returns a copy of the current in memory sync state of the Database.
 func (s *syncService) getDbSyncStateInMem(ctx *context.T, appName, dbName string) (*dbSyncStateInMem, error) {
 	s.syncStateLock.Lock()
diff --git a/x/ref/services/syncbase/vsync/syncgroup.go b/x/ref/services/syncbase/vsync/syncgroup.go
index 31eaff0..4045416 100644
--- a/x/ref/services/syncbase/vsync/syncgroup.go
+++ b/x/ref/services/syncbase/vsync/syncgroup.go
@@ -21,6 +21,7 @@
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
 	"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/naming"
@@ -420,9 +421,8 @@
 			return err
 		}
 
-		// TODO(hpucha): Add watch notification to signal SG creation.
-
-		return nil
+		tx1 := tx.(store.Transaction)
+		return watchable.AddSyncGroupOp(ctx, tx1, spec.Prefixes, false)
 	})
 
 	if err != nil {
@@ -516,9 +516,8 @@
 			return err
 		}
 
-		// TODO(hpucha): Add a watch notification to signal new SG.
-
-		return nil
+		tx1 := tx.(store.Transaction)
+		return watchable.AddSyncGroupOp(ctx, tx1, sg.Spec.Prefixes, false)
 	})
 
 	if err != nil {
diff --git a/x/ref/services/syncbase/vsync/watcher.go b/x/ref/services/syncbase/vsync/watcher.go
index 804907c..4865d82 100644
--- a/x/ref/services/syncbase/vsync/watcher.go
+++ b/x/ref/services/syncbase/vsync/watcher.go
@@ -93,6 +93,9 @@
 		resMark = ""
 	}
 
+	// Initialize Database sync state if needed.
+	s.initDbSyncStateInMem(ctx, appName, dbName)
+
 	// 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)
@@ -327,22 +330,22 @@
 	timestamp := logEnt.CommitTimestamp
 
 	switch op := logEnt.Op.(type) {
-	case *watchable.OpGet:
+	case watchable.OpGet:
 		// TODO(rdaoud): save read-set in sync.
 
-	case *watchable.OpScan:
+	case watchable.OpScan:
 		// TODO(rdaoud): save scan-set in sync.
 
-	case *watchable.OpPut:
+	case watchable.OpPut:
 		rec = newLocalLogRec(ctx, tx, op.Value.Key, op.Value.Version, false, timestamp)
 
-	case *watchable.OpSyncSnapshot:
+	case watchable.OpSyncSnapshot:
 		rec = newLocalLogRec(ctx, tx, op.Value.Key, op.Value.Version, false, timestamp)
 
-	case *watchable.OpDelete:
+	case watchable.OpDelete:
 		rec = newLocalLogRec(ctx, tx, op.Value.Key, watchable.NewVersion(), true, timestamp)
 
-	case *watchable.OpSyncGroup:
+	case watchable.OpSyncGroup:
 		vlog.Errorf("watch LogEntry for SyncGroup should not be converted: %v", logEnt)
 
 	default:
@@ -377,7 +380,7 @@
 // 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:
+	case watchable.OpSyncGroup:
 		remove := op.Value.Remove
 		for _, prefix := range op.Value.Prefixes {
 			if remove {
@@ -399,11 +402,11 @@
 func syncable(appdb string, logEnt *watchable.LogEntry) bool {
 	var key string
 	switch op := logEnt.Op.(type) {
-	case *watchable.OpPut:
+	case watchable.OpPut:
 		key = string(op.Value.Key)
-	case *watchable.OpDelete:
+	case watchable.OpDelete:
 		key = string(op.Value.Key)
-	case *watchable.OpSyncSnapshot:
+	case watchable.OpSyncSnapshot:
 		key = string(op.Value.Key)
 	default:
 		return false
diff --git a/x/ref/services/syncbase/vsync/watcher_test.go b/x/ref/services/syncbase/vsync/watcher_test.go
index aa7a72f..bf4fb95 100644
--- a/x/ref/services/syncbase/vsync/watcher_test.go
+++ b/x/ref/services/syncbase/vsync/watcher_test.go
@@ -116,7 +116,7 @@
 
 	for _, test := range checkSyncableTests {
 		log := &watchable.LogEntry{
-			Op: &watchable.OpPut{
+			Op: watchable.OpPut{
 				watchable.PutOp{Key: []byte(makeRowKey(test.key))},
 			},
 		}
@@ -133,9 +133,9 @@
 	k, v := []byte(key), []byte(version)
 	log := &watchable.LogEntry{}
 	if delete {
-		log.Op = &watchable.OpDelete{watchable.DeleteOp{Key: k}}
+		log.Op = watchable.OpDelete{watchable.DeleteOp{Key: k}}
 	} else {
-		log.Op = &watchable.OpPut{watchable.PutOp{Key: k, Version: v}}
+		log.Op = watchable.OpPut{watchable.PutOp{Key: k, Version: v}}
 	}
 	return log
 }
@@ -143,7 +143,7 @@
 // newSGLog creates a SyncGroup watch log entry.
 func newSGLog(prefixes []string, remove bool) *watchable.LogEntry {
 	return &watchable.LogEntry{
-		Op: &watchable.OpSyncGroup{
+		Op: watchable.OpSyncGroup{
 			watchable.SyncGroupOp{Prefixes: prefixes, Remove: remove},
 		},
 	}