emit memdb checkpoints after changes

Clients may consider the revision has moved
forward before relationship / the changes were
emitted.

internal/datastore/memdb/watch.go

@@ -114,18 +114,18 @@ func (mdb *memdbDatastore) loadChanges(_ context.Context, currentTxn int64, opti
 			changes = append(changes, &change.changes)
 		}
 
+		if options.Content&datastore.WatchSchema == datastore.WatchSchema &&
+			len(change.changes.ChangedDefinitions) > 0 || len(change.changes.DeletedCaveats) > 0 || len(change.changes.DeletedNamespaces) > 0 {
+			changes = append(changes, &change.changes)
+		}
+
 		if options.Content&datastore.WatchCheckpoints == datastore.WatchCheckpoints && change.revisionNanos > lastRevision {
 			changes = append(changes, &datastore.RevisionChanges{
 				Revision:     revisions.NewForTimestamp(change.revisionNanos),
 				IsCheckpoint: true,
 			})
 		}
 
-		if options.Content&datastore.WatchSchema == datastore.WatchSchema &&
-			len(change.changes.ChangedDefinitions) > 0 || len(change.changes.DeletedCaveats) > 0 || len(change.changes.DeletedNamespaces) > 0 {
-			changes = append(changes, &change.changes)
-		}
-
 		lastRevision = change.revisionNanos
 	}
 

pkg/datastore/test/watch.go

@@ -726,7 +726,7 @@ func WatchCheckpointsTest(t *testing.T, tester DatastoreTester) {
 	require.NoError(err)
 
 	changes, errchan := ds.Watch(ctx, lowestRevision, datastore.WatchOptions{
-		Content:            datastore.WatchCheckpoints | datastore.WatchRelationships,
+		Content:            datastore.WatchCheckpoints | datastore.WatchRelationships | datastore.WatchSchema,
 		CheckpointInterval: 100 * time.Millisecond,
 	})
 	require.Zero(len(errchan))
@@ -735,6 +735,14 @@ func WatchCheckpointsTest(t *testing.T, tester DatastoreTester) {
 		tuple.Parse("document:firstdoc#viewer@user:tom"),
 	)
 	require.NoError(err)
+
+	verifyCheckpointUpdate(require, afterTouchRevision, changes)
+
+	afterTouchRevision, err = ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error {
+		return rwt.WriteNamespaces(ctx, &core.NamespaceDefinition{Name: "doesnotexist"})
+	})
+	require.NoError(err)
+
 	verifyCheckpointUpdate(require, afterTouchRevision, changes)
 }
 
@@ -743,12 +751,20 @@ func verifyCheckpointUpdate(
 	expectedRevision datastore.Revision,
 	changes <-chan *datastore.RevisionChanges,
 ) {
+	var relChangeEmitted, schemaChangeEmitted bool
 	changeWait := time.NewTimer(waitForChangesTimeout)
 	for {
 		select {
 		case change, ok := <-changes:
 			require.True(ok)
+			if len(change.ChangedDefinitions) > 0 {
+				schemaChangeEmitted = true
+			}
+			if len(change.RelationshipChanges) > 0 {
+				relChangeEmitted = true
+			}
 			if change.IsCheckpoint {
+				require.True(relChangeEmitted || schemaChangeEmitted, "expected relationship/schema changes before checkpoint")
 				require.True(change.Revision.Equal(change.Revision) || change.Revision.GreaterThan(expectedRevision))
 				return
 			}