diff --git a/services/syncbase/bridge/cgo/impl.go b/services/syncbase/bridge/cgo/impl.go
index b7c232c..aa78bed 100644
--- a/services/syncbase/bridge/cgo/impl.go
+++ b/services/syncbase/bridge/cgo/impl.go
@@ -550,10 +550,18 @@
 	if !ok {
 		return verror.NewErrInternal(s.ctx)
 	}
+	if wireWC.State == watch.InitialStateSkipped {
+		return nil
+	}
+	wc := syncbase.ToWatchChange(wireWC)
+	if wc.EntityType != syncbase.EntityRow {
+		// TODO(ivanpi): Pipe collection and initial changes through.
+		return nil
+	}
 	// C.CallDbWatchPatternsCallbacksOnChange() blocks until the client acks the
 	// previous invocation, thus providing flow control.
 	cWatchChange := C.v23_syncbase_WatchChange{}
-	cWatchChange.init(syncbase.ToWatchChange(wireWC))
+	cWatchChange.init(*wc)
 	C.CallDbWatchPatternsCallbacksOnChange(s.cbs, cWatchChange)
 	return nil
 }
diff --git a/services/syncbase/longevity_tests/client/watcher_test.go b/services/syncbase/longevity_tests/client/watcher_test.go
index c6bbb96..bc73216 100644
--- a/services/syncbase/longevity_tests/client/watcher_test.go
+++ b/services/syncbase/longevity_tests/client/watcher_test.go
@@ -33,8 +33,10 @@
 		OnChange: func(watchChange syncbase.WatchChange) {
 			mu.Lock()
 			defer mu.Unlock()
-			gotRows = append(gotRows, watchChange.Row)
-			wg.Done()
+			if watchChange.EntityType == syncbase.EntityRow {
+				gotRows = append(gotRows, watchChange.Row)
+				wg.Done()
+			}
 		},
 	}
 
diff --git a/services/syncbase/longevity_tests/control/control_test.go b/services/syncbase/longevity_tests/control/control_test.go
index 9a10020..f5d2523 100644
--- a/services/syncbase/longevity_tests/control/control_test.go
+++ b/services/syncbase/longevity_tests/control/control_test.go
@@ -328,9 +328,11 @@
 	control.RegisterClient("test-watcher", func() client.Client {
 		return &client.Watcher{
 			OnChange: func(wc syncbase.WatchChange) {
-				changesReceived++
-				if changesReceived == 5 {
-					mu.Unlock()
+				if wc.EntityType == syncbase.EntityRow {
+					changesReceived++
+					if changesReceived == 5 {
+						mu.Unlock()
+					}
 				}
 			},
 		}
@@ -381,9 +383,11 @@
 	control.RegisterClient("test-watcher", func() client.Client {
 		return &client.Watcher{
 			OnChange: func(wc syncbase.WatchChange) {
-				changesReceived++
-				if changesReceived == 5 {
-					mu.Unlock()
+				if wc.EntityType == syncbase.EntityRow {
+					changesReceived++
+					if changesReceived == 5 {
+						mu.Unlock()
+					}
 				}
 			},
 		}
diff --git a/services/syncbase/server/database_watch.go b/services/syncbase/server/database_watch.go
index fd5870d..991bfd1 100644
--- a/services/syncbase/server/database_watch.go
+++ b/services/syncbase/server/database_watch.go
@@ -70,24 +70,35 @@
 	}
 	initImpl := func(sntx store.SnapshotOrTransaction) error {
 		// TODO(ivanpi): Check permissions here.
-		// Get the resume marker and fetch the initial state if necessary.
-		if len(resumeMarker) == 0 {
+		needInitialState := len(resumeMarker) == 0
+		needResumeMarker := needInitialState || bytes.Equal(resumeMarker, []byte("now"))
+		// Get the resume marker if necessary.
+		if needResumeMarker {
 			var err error
 			if resumeMarker, err = watchable.GetResumeMarker(sntx); err != nil {
 				return err
 			}
-			// Send initial state.
-			if err = d.scanInitialState(ctx, call, sender, sntx, watchFilter); err != nil {
-				return err
-			}
-		} else if bytes.Equal(resumeMarker, []byte("now")) {
-			var err error
-			// TODO(ivanpi): Add initial_state_skipped change.
-			if resumeMarker, err = watchable.GetResumeMarker(sntx); err != nil {
-				return err
-			}
 		}
-		// Finalize initial state batch if necessary.
+		// Send the root update to notify the client that watch has started.
+		rootChangeState := watch.InitialStateSkipped
+		if needInitialState {
+			rootChangeState = watch.Exists
+		}
+		if err := sender.addChange(
+			"",
+			rootChangeState,
+			&wire.StoreChange{
+				FromSync: false,
+			}); err != nil {
+			return err
+		}
+		// Send initial state if necessary.
+		if needInitialState {
+			if err := d.scanInitialState(ctx, call, sender, sntx, watchFilter); err != nil {
+				return err
+			}
+		}
+		// Finalize initial state or root update batch.
 		return sender.finishBatch(resumeMarker)
 	}
 	if err := store.RunWithSnapshot(d.st, initImpl); err != nil {
@@ -99,8 +110,6 @@
 // scanInitialState sends the initial state of all matching and accessible
 // collections and rows in the database. Checks access on collections, but
 // not on database.
-// TODO(ivanpi): Send dummy update for empty prefix to be compatible with
-// v.io/v23/services/watch.
 // TODO(ivanpi): Abstract out multi-scan for scan and possibly query support.
 // TODO(ivanpi): Use watch pattern prefixes to optimize scan ranges.
 func (d *database) scanInitialState(ctx *context.T, call rpc.ServerCall, sender *watchBatchSender, sntx store.SnapshotOrTransaction, watchFilter filter.CollectionRowFilter) error {
@@ -126,7 +135,7 @@
 			id: cxId,
 			d:  d,
 		}
-		// Check permissions.
+		// Check permissions for row access.
 		// TODO(ivanpi): Collection scan already gets perms, optimize?
 		if _, err := c.checkAccess(ctx, call, sntx); err != nil {
 			if verror.ErrorID(err) == verror.ErrNoAccess.ID {
@@ -169,7 +178,7 @@
 		// Send row.
 		var valueAsRawBytes *vom.RawBytes
 		if err := vom.Decode(value, &valueAsRawBytes); err != nil {
-			return err
+			return verror.New(verror.ErrInternal, ctx, err)
 		}
 		if err := sender.addChange(
 			naming.Join(pubutil.EncodeId(c.id), externalKey),
@@ -205,7 +214,8 @@
 			// a time, and would need to be updated as well.
 			logs, nextResumeMarker, err := watchable.ReadBatchFromLog(d.st, resumeMarker)
 			if err != nil {
-				return err
+				// TODO(ivanpi): Log all internal errors, especially ones not returned.
+				return verror.NewErrInternal(ctx) // no detailed error before access check
 			}
 			if logs == nil {
 				// No new log records available at this time.
@@ -241,11 +251,12 @@
 	defer sn.Abort()
 	// TODO(ivanpi): Recheck database perms here and fail, or cache for collection
 	// access checks.
+	valueBytes := []byte{}
 	for _, logEntry := range logs {
 		var opKey string
 		var op interface{}
 		if err := logEntry.Op.ToValue(&op); err != nil {
-			return err
+			return verror.NewErrInternal(ctx) // no detailed error before access check
 		}
 		switch op := op.(type) {
 		case *watchable.PutOp:
@@ -260,7 +271,10 @@
 		if !common.IsRowKey(opKey) {
 			continue
 		}
-		cxId, row := common.ParseRowKeyOrDie(opKey)
+		cxId, row, err := common.ParseRowKey(opKey)
+		if err != nil {
+			return verror.NewErrInternal(ctx) // no detailed error before access check
+		}
 		// Filter out unnecessary rows.
 		if !watchFilter.RowMatches(cxId, row) {
 			continue
@@ -276,6 +290,8 @@
 				// Note, the collection may not exist anymore, in which case permissions
 				// cannot be retrieved. This case is treated the same as ErrNoAccess, by
 				// skipping the row.
+				// TODO(ivanpi): Consider using the implicit ACL instead for nonexistent
+				// collections.
 				// TODO(ivanpi): Inaccessible rows are skipped. Figure out how to signal
 				// this to caller.
 				continue
@@ -284,15 +300,18 @@
 		}
 		switch op := op.(type) {
 		case *watchable.PutOp:
-			rowValue, err := watchable.GetAtVersion(ctx, sn, op.Key, nil, op.Version)
-			if err != nil {
-				return err
+			// Note, valueBytes is reused on each iteration, so the reference must not
+			// be used beyond this case block. The code below is safe since only the
+			// VOM-decoded copy is used after the call to vom.Decode.
+			if valueBytes, err = watchable.GetAtVersion(ctx, sn, op.Key, valueBytes, op.Version); err != nil {
+				return verror.New(verror.ErrInternal, ctx, err)
 			}
 			var rowValueAsRawBytes *vom.RawBytes
-			if err := vom.Decode(rowValue, &rowValueAsRawBytes); err != nil {
-				return err
+			if err := vom.Decode(valueBytes, &rowValueAsRawBytes); err != nil {
+				return verror.New(verror.ErrInternal, ctx, err)
 			}
-			if err := sender.addChange(naming.Join(pubutil.EncodeId(cxId), row),
+			if err := sender.addChange(
+				naming.Join(pubutil.EncodeId(cxId), row),
 				watch.Exists,
 				&wire.StoreChange{
 					Value:    rowValueAsRawBytes,
@@ -301,7 +320,8 @@
 				return err
 			}
 		case *watchable.DeleteOp:
-			if err := sender.addChange(naming.Join(pubutil.EncodeId(cxId), row),
+			if err := sender.addChange(
+				naming.Join(pubutil.EncodeId(cxId), row),
 				watch.DoesNotExist,
 				&wire.StoreChange{
 					FromSync: logEntry.FromSync,
diff --git a/services/syncbase/testutil/util.go b/services/syncbase/testutil/util.go
index 06b52fd..875f97b 100644
--- a/services/syncbase/testutil/util.go
+++ b/services/syncbase/testutil/util.go
@@ -203,9 +203,21 @@
 	ValueBytes *vom.RawBytes
 }
 
+func WatchChangeTestRootPut(resumeMarker watch.ResumeMarker) WatchChangeTest {
+	return WatchChangeTest{
+		WatchChange: syncbase.WatchChange{
+			EntityType:   syncbase.EntityRoot,
+			ChangeType:   syncbase.PutChange,
+			ResumeMarker: resumeMarker,
+			Continued:    (resumeMarker == nil),
+		},
+	}
+}
+
 func WatchChangeTestRowPut(cxId wire.Id, rowKey string, value interface{}, resumeMarker watch.ResumeMarker) WatchChangeTest {
 	return WatchChangeTest{
 		WatchChange: syncbase.WatchChange{
+			EntityType:   syncbase.EntityRow,
 			Collection:   cxId,
 			Row:          rowKey,
 			ChangeType:   syncbase.PutChange,
@@ -219,6 +231,7 @@
 func WatchChangeTestRowDelete(cxId wire.Id, rowKey string, resumeMarker watch.ResumeMarker) WatchChangeTest {
 	return WatchChangeTest{
 		WatchChange: syncbase.WatchChange{
+			EntityType:   syncbase.EntityRow,
 			Collection:   cxId,
 			Row:          rowKey,
 			ChangeType:   syncbase.DeleteChange,
@@ -230,7 +243,8 @@
 
 // WatchChangeEq returns whether *want and *got represent the same value.
 func WatchChangeEq(got *syncbase.WatchChange, want *WatchChangeTest) (eq bool) {
-	if want.Collection == got.Collection &&
+	if want.EntityType == got.EntityType &&
+		want.Collection == got.Collection &&
 		want.Row == got.Row &&
 		want.ChangeType == got.ChangeType &&
 		bytes.Equal(want.ResumeMarker, got.ResumeMarker) &&
@@ -240,12 +254,17 @@
 		if want.ChangeType == syncbase.DeleteChange {
 			eq = true
 		} else {
-			var wantValue interface{}
-			var gotValue interface{}
-			gotErr := got.Value(&gotValue)
-			wantErr := want.ValueBytes.ToValue(&wantValue)
-			eq = ((gotErr == nil) == (wantErr == nil)) &&
-				reflect.DeepEqual(gotValue, wantValue)
+			switch want.EntityType {
+			case syncbase.EntityRow:
+				var wantValue interface{}
+				var gotValue interface{}
+				gotErr := got.Value(&gotValue)
+				wantErr := want.ValueBytes.ToValue(&wantValue)
+				eq = ((gotErr == nil) == (wantErr == nil)) &&
+					reflect.DeepEqual(gotValue, wantValue)
+			default:
+				eq = true
+			}
 		}
 	}
 	return eq