Merge pull request #2082 from authzed/memdb-emit-checkpoints-after-changes

emit memdb checkpoints after changes

Author: vroldanbet

Diff for: 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
 	}
 

Diff for: internal/testserver/datastore/postgres.go

@@ -129,12 +129,12 @@ func (b *postgresTester) NewDatabase(t testing.TB) string {
 }
 
 const (
-	retryCount         = 3
-	timeBetweenRetries = 100 * time.Millisecond
+	retryCount         = 4
+	timeBetweenRetries = 1 * time.Second
 )
 
 func (b *postgresTester) NewDatastore(t testing.TB, initFunc InitFunc) datastore.Datastore {
-	for i := 0; i < retryCount; i++ {
+	for i := 1; i <= retryCount; i++ {
 		connectStr := b.NewDatabase(t)
 
 		migrationDriver, err := pgmigrations.NewAlembicPostgresDriver(context.Background(), connectStr, datastore.NoCredentialsProvider)
@@ -144,11 +144,11 @@ func (b *postgresTester) NewDatastore(t testing.TB, initFunc InitFunc) datastore
 			return initFunc("postgres", connectStr)
 		}
 
-		if i == retryCount-1 {
+		if i == retryCount {
 			require.NoError(t, err, "got error when trying to create migration driver")
 		}
 
-		time.Sleep(timeBetweenRetries)
+		time.Sleep(time.Duration(i) * timeBetweenRetries)
 	}
 
 	require.Fail(t, "failed to create datastore for testing")

Diff for: 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,14 +751,25 @@ 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(change.Revision.Equal(change.Revision) || change.Revision.GreaterThan(expectedRevision))
-				return
+				if change.Revision.Equal(expectedRevision) || change.Revision.GreaterThan(expectedRevision) {
+					require.True(relChangeEmitted || schemaChangeEmitted, "expected relationship/schema changes before checkpoint")
+					return
+				}
+
+				// we received a past revision checkpoint, ignore
 			}
 		case <-changeWait.C:
 			require.Fail("Timed out", "waited for checkpoint")