Merge pull request authzed#1720 from authzed/memdb-schemawatch-support

implements schema watch support for MemDB

Author: vroldanbet

internal/datastore/memdb/memdb.go

@@ -9,6 +9,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/authzed/spicedb/internal/datastore/common"
+	"github.com/authzed/spicedb/pkg/spiceerrors"
+
 	"github.com/google/uuid"
 	"github.com/hashicorp/go-memdb"
 
@@ -195,40 +198,72 @@ func (mdb *memdbDatastore) ReadWriteTx(
 		mdb.Lock()
 		defer mdb.Unlock()
 
-		// Record the changes that were made
-		newChanges := datastore.RevisionChanges{
-			Revision:            newRevision,
-			RelationshipChanges: nil,
-		}
+		tracked := common.NewChanges(revisions.TimestampIDKeyFunc, datastore.WatchRelationships|datastore.WatchSchema)
 		if tx != nil {
 			for _, change := range tx.Changes() {
-				if change.Table == tableRelationship {
+				switch change.Table {
+				case tableRelationship:
 					if change.After != nil {
 						rt, err := change.After.(*relationship).RelationTuple()
 						if err != nil {
 							return datastore.NoRevision, err
 						}
-						newChanges.RelationshipChanges = append(newChanges.RelationshipChanges, &corev1.RelationTupleUpdate{
-							Operation: corev1.RelationTupleUpdate_TOUCH,
-							Tuple:     rt,
-						})
-					}
-					if change.After == nil && change.Before != nil {
+
+						if err := tracked.AddRelationshipChange(ctx, newRevision, rt, corev1.RelationTupleUpdate_TOUCH); err != nil {
+							return datastore.NoRevision, err
+						}
+					} else if change.After == nil && change.Before != nil {
 						rt, err := change.Before.(*relationship).RelationTuple()
 						if err != nil {
 							return datastore.NoRevision, err
 						}
-						newChanges.RelationshipChanges = append(newChanges.RelationshipChanges, &corev1.RelationTupleUpdate{
-							Operation: corev1.RelationTupleUpdate_DELETE,
-							Tuple:     rt,
-						})
+
+						if err := tracked.AddRelationshipChange(ctx, newRevision, rt, corev1.RelationTupleUpdate_DELETE); err != nil {
+							return datastore.NoRevision, err
+						}
+					} else {
+						return datastore.NoRevision, spiceerrors.MustBugf("unexpected relationship change")
+					}
+				case tableNamespace:
+					if change.After != nil {
+						loaded := &corev1.NamespaceDefinition{}
+						if err := loaded.UnmarshalVT(change.After.(*namespace).configBytes); err != nil {
+							return datastore.NoRevision, err
+						}
+
+						tracked.AddChangedDefinition(ctx, newRevision, loaded)
+					} else if change.After == nil && change.Before != nil {
+						tracked.AddDeletedNamespace(ctx, newRevision, change.Before.(*namespace).name)
+					} else {
+						return datastore.NoRevision, spiceerrors.MustBugf("unexpected namespace change")
+					}
+				case tableCaveats:
+					if change.After != nil {
+						loaded := &corev1.CaveatDefinition{}
+						if err := loaded.UnmarshalVT(change.After.(*caveat).definition); err != nil {
+							return datastore.NoRevision, err
+						}
+
+						tracked.AddChangedDefinition(ctx, newRevision, loaded)
+					} else if change.After == nil && change.Before != nil {
+						tracked.AddDeletedCaveat(ctx, newRevision, change.Before.(*caveat).name)
+					} else {
+						return datastore.NoRevision, spiceerrors.MustBugf("unexpected namespace change")
 					}
 				}
 			}
 
+			var rc datastore.RevisionChanges
+			changes := tracked.AsRevisionChanges(revisions.TimestampIDKeyLessThanFunc)
+			if len(changes) > 1 {
+				return datastore.NoRevision, spiceerrors.MustBugf("unexpected MemDB transaction with multiple revision changes")
+			} else if len(changes) == 1 {
+				rc = changes[0]
+			}
+
 			change := &changelog{
 				revisionNanos: newRevision.TimestampNanoSec(),
-				changes:       newChanges,
+				changes:       rc,
 			}
 			if err := tx.Insert(tableChangelog, change); err != nil {
 				return datastore.NoRevision, fmt.Errorf("error writing changelog: %w", err)

internal/datastore/memdb/memdb_test.go

@@ -26,7 +26,7 @@ func (mdbt memDBTest) New(revisionQuantization, _, gcWindow time.Duration, watch
 }
 
 func TestMemdbDatastore(t *testing.T) {
-	test.AllWithExceptions(t, memDBTest{}, test.WithCategories(test.WatchSchemaCategory))
+	test.All(t, memDBTest{})
 }
 
 func TestConcurrentWritePanic(t *testing.T) {

internal/datastore/memdb/watch.go

@@ -23,11 +23,6 @@ func (mdb *memdbDatastore) Watch(ctx context.Context, ar datastore.Revision, opt
 	updates := make(chan *datastore.RevisionChanges, watchBufferLength)
 	errs := make(chan error, 1)
 
-	if options.Content&datastore.WatchSchema == datastore.WatchSchema {
-		errs <- errors.New("schema watch unsupported in MemDB")
-		return updates, errs
-	}
-
 	watchBufferWriteTimeout := options.WatchBufferWriteTimeout
 	if watchBufferWriteTimeout == 0 {
 		watchBufferWriteTimeout = mdb.watchBufferWriteTimeout

internal/datastore/revisions/commonrevision.go

@@ -65,7 +65,7 @@ func (cd CommonDecoder) RevisionFromString(s string) (datastore.Revision, error)
 // WithInexactFloat64 is an interface that can be implemented by a revision to
 // provide an inexact float64 representation of the revision.
 type WithInexactFloat64 interface {
-	// WithInexactFloat64 returns a float64 that is an inexact representation of the
+	// InexactFloat64 returns a float64 that is an inexact representation of the
 	// revision.
 	InexactFloat64() float64
 }