diff --git a/services/syncbase/common/key_util.go b/services/syncbase/common/key_util.go
index 45038f3..976052e 100644
--- a/services/syncbase/common/key_util.go
+++ b/services/syncbase/common/key_util.go
@@ -73,11 +73,29 @@
 func ParseRowKeyOrDie(key string) (collection wire.Id, row string) {
 	collection, row, err := ParseRowKey(key)
 	if err != nil {
+		// TODO(ivanpi): Get rid of *OrDie() functions. Always log internal errors.
 		vlog.Fatal(err)
 	}
 	return collection, row
 }
 
+// ParseCollectionPermsKey extracts the collection id from the given storage
+// engine key for a collection perms entry. Returns an error if the given key
+// is not a storage engine key for a collection perms entry.
+func ParseCollectionPermsKey(key string) (collection wire.Id, err error) {
+	// TODO(rdaoud,ivanpi): See hack in collection.go.
+	parts := SplitNKeyParts(key, 3)
+	pfx := parts[0]
+	if len(parts) < 3 || pfx != CollectionPermsPrefix || parts[2] != "" {
+		return wire.Id{}, fmt.Errorf("ParseCollectionPermsKey: invalid key %q", key)
+	}
+	cxId, err := util.DecodeId(parts[1])
+	if err != nil {
+		return wire.Id{}, fmt.Errorf("ParseCollectionPermsKey: invalid collection %q: %v", parts[1], err)
+	}
+	return cxId, nil
+}
+
 // ScanPrefixArgs returns args for sn.Scan() for the specified prefix.
 func ScanPrefixArgs(stKeyPrefix, prefix string) ([]byte, []byte) {
 	return ScanRangeArgs(stKeyPrefix, util.PrefixRangeStart(prefix), util.PrefixRangeLimit(prefix))
diff --git a/services/syncbase/common/key_util_test.go b/services/syncbase/common/key_util_test.go
index 3b2e45c..4c61d7e 100644
--- a/services/syncbase/common/key_util_test.go
+++ b/services/syncbase/common/key_util_test.go
@@ -162,6 +162,36 @@
 	}
 }
 
+func TestParseCollectionPermsKey(t *testing.T) {
+	tests := []struct {
+		key        string
+		collection wire.Id
+		err        bool
+	}{
+		{common.CollectionPermsPrefix + "\xfeu,c\xfe", wire.Id{"u", "c"}, false},
+		{common.CollectionPermsPrefix + "\xfeu,\xfe", wire.Id{"u", ""}, false},
+		{common.CollectionPermsPrefix + "\xfe,c\xfe", wire.Id{"", "c"}, false},
+		{common.CollectionPermsPrefix + "\xfe,\xfe", wire.Id{"", ""}, false},
+		{"\xfeu,c\xfe", wire.Id{}, true},
+		{"pfx\xfeu,c\xfe", wire.Id{}, true},
+		{common.RowPrefix + "\xfeu,c\xfe", wire.Id{}, true},
+		{common.CollectionPermsPrefix + "\xfe", wire.Id{}, true},
+		{common.CollectionPermsPrefix + "\xfe\xfe", wire.Id{}, true},
+		{common.CollectionPermsPrefix + "\xfeu,c\xferow", wire.Id{}, true},
+		{common.CollectionPermsPrefix + "\xfeu,c", wire.Id{}, true},
+		{common.CollectionPermsPrefix + "\xfe", wire.Id{}, true},
+	}
+	for _, test := range tests {
+		collection, err := common.ParseCollectionPermsKey(test.key)
+		if !reflect.DeepEqual(collection, test.collection) {
+			t.Errorf("%q: got %v, want %v", test.key, collection, test.collection)
+		}
+		if !reflect.DeepEqual(err != nil, test.err) {
+			t.Errorf("%q: got %v, want %v", test.key, err != nil, test.err)
+		}
+	}
+}
+
 func TestScanPrefixArgs(t *testing.T) {
 	tests := []struct {
 		stKeyPrefix, prefix, wantStart, wantLimit string
diff --git a/services/syncbase/server/database.go b/services/syncbase/server/database.go
index 8211194..eb7de69 100644
--- a/services/syncbase/server/database.go
+++ b/services/syncbase/server/database.go
@@ -460,8 +460,7 @@
 		keyBytes := []byte{}
 		for it.Advance() {
 			keyBytes = it.Key(keyBytes)
-			parts := common.SplitNKeyParts(string(keyBytes), 3)
-			id, err := pubutil.DecodeId(parts[1])
+			id, err := common.ParseCollectionPermsKey(string(keyBytes))
 			if err != nil {
 				it.Cancel()
 				return verror.New(verror.ErrInternal, ctx, err)
diff --git a/services/syncbase/server/database_watch.go b/services/syncbase/server/database_watch.go
index 991bfd1..0d8a6cd 100644
--- a/services/syncbase/server/database_watch.go
+++ b/services/syncbase/server/database_watch.go
@@ -10,6 +10,7 @@
 	"v.io/v23/context"
 	"v.io/v23/naming"
 	"v.io/v23/rpc"
+	"v.io/v23/security/access"
 	wire "v.io/v23/services/syncbase"
 	"v.io/v23/services/watch"
 	pubutil "v.io/v23/syncbase/util"
@@ -17,6 +18,7 @@
 	"v.io/v23/vom"
 	"v.io/x/ref/services/syncbase/common"
 	"v.io/x/ref/services/syncbase/server/filter"
+	"v.io/x/ref/services/syncbase/server/interfaces"
 	"v.io/x/ref/services/syncbase/store"
 	"v.io/x/ref/services/syncbase/store/watchable"
 )
@@ -110,6 +112,8 @@
 // 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): Assumes Read perms on database. Careful if supporting RPCs
+// requiring only Resolve (e.g. WatchGlob).
 // 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 {
@@ -117,13 +121,12 @@
 	// TODO(ivanpi): Collection scan order not alphabetical.
 	cxIt := sntx.Scan(common.ScanPrefixArgs(common.CollectionPermsPrefix, ""))
 	defer cxIt.Cancel()
-	cxKey := []byte{}
+	cxKey, cxPermsValue := []byte{}, []byte{}
 	for cxIt.Advance() {
-		cxKey = cxIt.Key(cxKey)
-		// See comment in util/constants.go for why we use SplitNKeyParts.
-		// TODO(rdaoud,ivanpi): See hack in collection.go.
-		cxParts := common.SplitNKeyParts(string(cxKey), 3)
-		cxId, err := pubutil.DecodeId(cxParts[1])
+		cxKey, cxPermsValue = cxIt.Key(cxKey), cxIt.Value(cxPermsValue)
+		// Database permissions for Watch ensure that the user is always allowed
+		// to know that a collection exists.
+		cxId, err := common.ParseCollectionPermsKey(string(cxKey))
 		if err != nil {
 			return verror.New(verror.ErrInternal, ctx, err)
 		}
@@ -131,21 +134,43 @@
 		if !watchFilter.CollectionMatches(cxId) {
 			continue
 		}
+		// Send collection info.
+		var cxPerms interfaces.CollectionPerms
+		if err := vom.Decode(cxPermsValue, &cxPerms); err != nil {
+			return verror.NewErrInternal(ctx) // no detailed error for cxPerms before filtering cxInfo
+		}
+		cxInfo := collectionInfoFromPerms(ctx, call, cxPerms.GetPerms())
+		cxInfoAsRawBytes, err := vom.RawBytesFromValue(cxInfo)
+		if err != nil {
+			return verror.New(verror.ErrInternal, ctx, err)
+		}
+		if err := sender.addChange(
+			pubutil.EncodeId(cxId),
+			watch.Exists,
+			&wire.StoreChange{
+				Value: cxInfoAsRawBytes,
+				// Note: FromSync cannot be reconstructed from scan.
+				FromSync: false,
+			}); err != nil {
+			return err
+		}
+		// Check permissions for row access.
+		// TODO(ivanpi): Collection scan already gets perms, optimize?
 		c := &collectionReq{
 			id: cxId,
 			d:  d,
 		}
-		// 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 {
-				// TODO(ivanpi): Inaccessible rows are skipped. Figure out how to signal
-				// this to caller.
+				// Skip sending rows if the collection is inaccessible. Caller can see
+				// from collection info that they have no read access and may therefore
+				// have missing rows.
+				// TODO(ivanpi): If read access is regained, should skipped rows be sent
+				// retroactively?
 				continue
 			}
 			return err
 		}
-		// TODO(ivanpi): Send collection info.
 		// Send matching rows.
 		if err := c.scanInitialState(ctx, call, sender, sntx, watchFilter); err != nil {
 			return err
@@ -246,6 +271,8 @@
 // Note: Since the governing ACL for each change is no longer tracked, the
 // permissions check uses the ACLs in effect at the time processLogBatch is
 // called.
+// TODO(ivanpi): Assumes Read perms on database. Careful if supporting RPCs
+// requiring only Resolve (e.g. WatchGlob).
 func (d *database) processLogBatch(ctx *context.T, call rpc.ServerCall, sender *watchBatchSender, watchFilter filter.CollectionRowFilter, logs []*watchable.LogEntry) error {
 	sn := d.st.NewSnapshot()
 	defer sn.Abort()
@@ -266,73 +293,145 @@
 		default:
 			continue
 		}
-		// TODO(rogulenko): Currently we only process rows, i.e. keys of the form
-		// <RowPrefix>:xxx:yyy. Consider processing other keys.
-		if !common.IsRowKey(opKey) {
-			continue
-		}
-		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
-		}
-		c := &collectionReq{
-			id: cxId,
-			d:  d,
-		}
-		// Filter out rows that we can't access.
-		// TODO(ivanpi): Check only once per collection per batch.
-		if _, err := c.checkAccess(ctx, call, sn); err != nil {
-			if verror.ErrorID(err) == verror.ErrNoAccess.ID || verror.ErrorID(err) == verror.ErrNoExist.ID {
-				// 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.
+		// TODO(rogulenko,ivanpi): Currently we only process rows and collection
+		// perms. Consider making watchable and processing other keys.
+		switch common.FirstKeyPart(opKey) {
+		case common.RowPrefix:
+			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
 			}
-			return err
-		}
-		switch op := op.(type) {
-		case *watchable.PutOp:
-			// 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)
+			// Filter out rows that we can't access.
+			// TODO(ivanpi): Check only once per collection per batch.
+			c := &collectionReq{
+				id: cxId,
+				d:  d,
 			}
-			var rowValueAsRawBytes *vom.RawBytes
-			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),
-				watch.Exists,
-				&wire.StoreChange{
-					Value:    rowValueAsRawBytes,
-					FromSync: logEntry.FromSync,
-				}); err != nil {
+			if _, err := c.checkAccess(ctx, call, sn); err != nil {
+				if verror.ErrorID(err) == verror.ErrNoAccess.ID || verror.ErrorID(err) == verror.ErrNoExist.ID {
+					// Skip sending rows if the collection is inaccessible. Caller can see
+					// from collection info that they have no read access and may therefore
+					// have missing rows.
+					// 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): If read access is regained, should skipped rows be sent
+					// retroactively?
+					continue
+				}
 				return err
 			}
-		case *watchable.DeleteOp:
-			if err := sender.addChange(
-				naming.Join(pubutil.EncodeId(cxId), row),
-				watch.DoesNotExist,
-				&wire.StoreChange{
-					FromSync: logEntry.FromSync,
-				}); err != nil {
-				return err
+			switch op := op.(type) {
+			case *watchable.PutOp:
+				// 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(valueBytes, &rowValueAsRawBytes); err != nil {
+					return verror.New(verror.ErrInternal, ctx, err)
+				}
+				if err := sender.addChange(
+					naming.Join(pubutil.EncodeId(cxId), row),
+					watch.Exists,
+					&wire.StoreChange{
+						Value:    rowValueAsRawBytes,
+						FromSync: logEntry.FromSync,
+					}); err != nil {
+					return err
+				}
+			case *watchable.DeleteOp:
+				if err := sender.addChange(
+					naming.Join(pubutil.EncodeId(cxId), row),
+					watch.DoesNotExist,
+					&wire.StoreChange{
+						FromSync: logEntry.FromSync,
+					}); err != nil {
+					return err
+				}
 			}
+
+		case common.CollectionPermsPrefix:
+			// Database permissions for Watch ensure that the user is always allowed
+			// to know that a collection exists.
+			cxId, err := common.ParseCollectionPermsKey(opKey)
+			if err != nil {
+				return verror.New(verror.ErrInternal, ctx, err)
+			}
+			// Filter out unnecessary collections.
+			if !watchFilter.CollectionMatches(cxId) {
+				continue
+			}
+			switch op := op.(type) {
+			case *watchable.PutOp:
+				if valueBytes, err = watchable.GetAtVersion(ctx, sn, op.Key, valueBytes, op.Version); err != nil {
+					return verror.NewErrInternal(ctx) // no detailed error for cxPerms before filtering cxInfo
+				}
+				var cxPerms interfaces.CollectionPerms
+				if err := vom.Decode(valueBytes, &cxPerms); err != nil {
+					return verror.NewErrInternal(ctx) // no detailed error for cxPerms before filtering cxInfo
+				}
+				cxInfo := collectionInfoFromPerms(ctx, call, cxPerms.GetPerms())
+				cxInfoAsRawBytes, err := vom.RawBytesFromValue(cxInfo)
+				if err != nil {
+					return verror.New(verror.ErrInternal, ctx, err)
+				}
+				if err := sender.addChange(
+					pubutil.EncodeId(cxId),
+					watch.Exists,
+					&wire.StoreChange{
+						Value:    cxInfoAsRawBytes,
+						FromSync: logEntry.FromSync,
+					}); err != nil {
+					return err
+				}
+			case *watchable.DeleteOp:
+				if err := sender.addChange(
+					pubutil.EncodeId(cxId),
+					watch.DoesNotExist,
+					&wire.StoreChange{
+						FromSync: logEntry.FromSync,
+					}); err != nil {
+					return err
+				}
+			}
+
+		default:
+			continue
 		}
 	}
 	return nil
 }
 
+// collectionInfoFromPerms converts a collection permissions object into a
+// StoreChangeCollectionInfo tailored to the caller. The returned collection
+// info is safe to send to the caller, assuming the caller is allowed to know
+// the collection exists. It includes a set listing all access tags that the
+// caller has on the collection. The collection permissions object itself is
+// included only if the caller is allowed to see it (has Admin permissions).
+func collectionInfoFromPerms(ctx *context.T, call rpc.ServerCall, cxPerms access.Permissions) *wire.StoreChangeCollectionInfo {
+	ci := &wire.StoreChangeCollectionInfo{
+		Allowed: make(map[access.Tag]struct{}),
+	}
+	for tag, acl := range cxPerms {
+		if acl.Authorize(ctx, call.Security()) == nil {
+			ci.Allowed[access.Tag(tag)] = struct{}{}
+		}
+	}
+	if _, isAdmin := ci.Allowed[access.Admin]; isAdmin {
+		ci.Perms = cxPerms
+	}
+	return ci
+}
+
 // watchBatchSender sends a sequence of watch changes forming a batch, delaying
 // sends to allow setting the Continued flag on the last change.
 type watchBatchSender struct {
diff --git a/services/syncbase/testutil/util.go b/services/syncbase/testutil/util.go
index 875f97b..22f9a6f 100644
--- a/services/syncbase/testutil/util.go
+++ b/services/syncbase/testutil/util.go
@@ -197,10 +197,11 @@
 }
 
 // A WatchChangeTest is a syncbase.WatchChange that has a public ValueBytes
-// field, to allow tests to set it.
+// and CollectionInfo field, to allow tests to set them.
 type WatchChangeTest struct {
 	syncbase.WatchChange
-	ValueBytes *vom.RawBytes
+	ValueBytes     *vom.RawBytes
+	CollectionInfo *wire.StoreChangeCollectionInfo
 }
 
 func WatchChangeTestRootPut(resumeMarker watch.ResumeMarker) WatchChangeTest {
@@ -214,6 +215,42 @@
 	}
 }
 
+func WatchChangeTestCollectionPut(cxId wire.Id, allowedTags []access.Tag, perms access.Permissions, resumeMarker watch.ResumeMarker) WatchChangeTest {
+	allowedTagsSet := make(map[access.Tag]struct{})
+	if len(allowedTags) == 0 {
+		allowedTagsSet = nil
+	} else {
+		for _, tag := range allowedTags {
+			allowedTagsSet[tag] = struct{}{}
+		}
+	}
+	return WatchChangeTest{
+		WatchChange: syncbase.WatchChange{
+			EntityType:   syncbase.EntityCollection,
+			Collection:   cxId,
+			ChangeType:   syncbase.PutChange,
+			ResumeMarker: resumeMarker,
+			Continued:    (resumeMarker == nil),
+		},
+		CollectionInfo: &wire.StoreChangeCollectionInfo{
+			Allowed: allowedTagsSet,
+			Perms:   perms,
+		},
+	}
+}
+
+func WatchChangeTestCollectionDelete(cxId wire.Id, resumeMarker watch.ResumeMarker) WatchChangeTest {
+	return WatchChangeTest{
+		WatchChange: syncbase.WatchChange{
+			EntityType:   syncbase.EntityCollection,
+			Collection:   cxId,
+			ChangeType:   syncbase.DeleteChange,
+			ResumeMarker: resumeMarker,
+			Continued:    (resumeMarker == nil),
+		},
+	}
+}
+
 func WatchChangeTestRowPut(cxId wire.Id, rowKey string, value interface{}, resumeMarker watch.ResumeMarker) WatchChangeTest {
 	return WatchChangeTest{
 		WatchChange: syncbase.WatchChange{
@@ -262,6 +299,8 @@
 				wantErr := want.ValueBytes.ToValue(&wantValue)
 				eq = ((gotErr == nil) == (wantErr == nil)) &&
 					reflect.DeepEqual(gotValue, wantValue)
+			case syncbase.EntityCollection:
+				eq = reflect.DeepEqual(got.CollectionInfo(), want.CollectionInfo)
 			default:
 				eq = true
 			}
@@ -278,7 +317,11 @@
 			Fatalf(t, "wstream.Advance() reached the end: %v", wstream.Err())
 		}
 		if got := wstream.Change(); !WatchChangeEq(&got, &want) {
-			Fatalf(t, "unexpected watch change: got %+v, want %+v", got, want)
+			if got.ChangeType == syncbase.PutChange && got.EntityType == syncbase.EntityCollection {
+				Fatalf(t, "unexpected watch change: got %+v, want %+v (cx info: got %+v, want %+v)", got, want, got.CollectionInfo(), want.CollectionInfo)
+			} else {
+				Fatalf(t, "unexpected watch change: got %+v, want %+v", got, want)
+			}
 		}
 	}
 }