Setup defined (and configurable) behavior if a ZedToken from

an older datastore is used

All ZedTokens are now minted with the datastore's unique ID included
in the ZedToken and that ID is checked when the ZedToken is decoded.

In scenarios where the datastore ID does not match, either an error is
raised (watch, at_exact_snapshot) or configurable behavior is used
(at_least_as_fresh)

Fixes authzed#1541

e2e/newenemy/newenemy_test.go

@@ -376,8 +376,8 @@ func checkDataNoNewEnemy(ctx context.Context, t testing.TB, slowNodeID int, crdb
 		require.NoError(t, err)
 		t.Log("r2 token: ", r2.WrittenAt.Token)
 
-		z1, _ := zedtoken.DecodeRevision(r1.WrittenAt, revisions.CommonDecoder{Kind: revisions.HybridLogicalClock})
-		z2, _ := zedtoken.DecodeRevision(r2.WrittenAt, revisions.CommonDecoder{Kind: revisions.HybridLogicalClock})
+		z1, _, _ := zedtoken.DecodeRevision(r1.WrittenAt, revisions.CommonDecoder{Kind: revisions.HybridLogicalClock})
+		z2, _, _ := zedtoken.DecodeRevision(r2.WrittenAt, revisions.CommonDecoder{Kind: revisions.HybridLogicalClock})
 
 		t.Log("z1 revision: ", z1)
 		t.Log("z2 revision: ", z2)

internal/datastore/proxy/proxy_test/mock.go

@@ -15,10 +15,16 @@ import (
 
 type MockDatastore struct {
 	mock.Mock
+
+	CurrentUniqueID string
 }
 
 func (dm *MockDatastore) UniqueID(_ context.Context) (string, error) {
-	return "mockds", nil
+	if dm.CurrentUniqueID == "" {
+		return "mockds", nil
+	}
+
+	return dm.CurrentUniqueID, nil
 }
 
 func (dm *MockDatastore) SnapshotReader(rev datastore.Revision) datastore.Reader {

internal/datastore/proxy/relationshipintegrity.go

@@ -197,6 +197,10 @@ func (r *relationshipIntegrityProxy) OptimizedRevision(ctx context.Context) (dat
 	return r.ds.OptimizedRevision(ctx)
 }
 
+func (r *relationshipIntegrityProxy) UniqueID(ctx context.Context) (string, error) {
+	return r.ds.UniqueID(ctx)
+}
+
 func (r *relationshipIntegrityProxy) ReadyState(ctx context.Context) (datastore.ReadyState, error) {
 	return r.ds.ReadyState(ctx)
 }

internal/datastore/revisions/commonrevision.go

@@ -1,6 +1,8 @@
 package revisions
 
 import (
+	"context"
+
 	"github.com/authzed/spicedb/pkg/datastore"
 	"github.com/authzed/spicedb/pkg/spiceerrors"
 )
@@ -43,7 +45,12 @@ func RevisionParser(kind RevisionKind) ParsingFunc {
 
 // CommonDecoder is a revision decoder that can decode revisions of a given kind.
 type CommonDecoder struct {
-	Kind RevisionKind
+	Kind              RevisionKind
+	DatastoreUniqueID string
+}
+
+func (cd CommonDecoder) UniqueID(_ context.Context) (string, error) {
+	return cd.DatastoreUniqueID, nil
 }
 
 func (cd CommonDecoder) RevisionFromString(s string) (datastore.Revision, error) {

internal/services/integrationtesting/cert_test.go

@@ -148,7 +148,7 @@ func TestCertRotation(t *testing.T) {
 					},
 					{
 						Name:       "consistency",
-						Middleware: consistency.UnaryServerInterceptor("testing"),
+						Middleware: consistency.UnaryServerInterceptor("testing", consistency.TreatMismatchingTokensAsError),
 					},
 					{
 						Name:       "servicespecific",
@@ -167,7 +167,7 @@ func TestCertRotation(t *testing.T) {
 					},
 					{
 						Name:       "consistency",
-						Middleware: consistency.StreamServerInterceptor("testing"),
+						Middleware: consistency.StreamServerInterceptor("testing", consistency.TreatMismatchingTokensAsError),
 					},
 					{
 						Name:       "servicespecific",
@@ -211,7 +211,7 @@ func TestCertRotation(t *testing.T) {
 	_, err = client.CheckPermission(ctx, &v1.CheckPermissionRequest{
 		Consistency: &v1.Consistency{
 			Requirement: &v1.Consistency_AtLeastAsFresh{
-				AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+				AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 			},
 		},
 		Resource:   rel.Resource,
@@ -264,7 +264,7 @@ func TestCertRotation(t *testing.T) {
 		_, err = client.CheckPermission(ctx, &v1.CheckPermissionRequest{
 			Consistency: &v1.Consistency{
 				Requirement: &v1.Consistency_AtLeastAsFresh{
-					AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+					AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 				},
 			},
 			Resource:   rel.Resource,

internal/services/integrationtesting/consistencytestutil/servicetester.go

@@ -78,7 +78,7 @@ func (v1st v1ServiceTester) Check(ctx context.Context, resource tuple.ObjectAndR
 		},
 		Consistency: &v1.Consistency{
 			Requirement: &v1.Consistency_AtLeastAsFresh{
-				AtLeastAsFresh: zedtoken.MustNewFromRevision(atRevision),
+				AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(atRevision),
 			},
 		},
 		Context: context,
@@ -98,7 +98,7 @@ func (v1st v1ServiceTester) Expand(ctx context.Context, resource tuple.ObjectAnd
 		Permission: resource.Relation,
 		Consistency: &v1.Consistency{
 			Requirement: &v1.Consistency_AtLeastAsFresh{
-				AtLeastAsFresh: zedtoken.MustNewFromRevision(atRevision),
+				AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(atRevision),
 			},
 		},
 	})
@@ -134,7 +134,7 @@ func (v1st v1ServiceTester) Read(_ context.Context, namespaceName string, atRevi
 		},
 		Consistency: &v1.Consistency{
 			Requirement: &v1.Consistency_AtLeastAsFresh{
-				AtLeastAsFresh: zedtoken.MustNewFromRevision(atRevision),
+				AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(atRevision),
 			},
 		},
 	})
@@ -181,7 +181,7 @@ func (v1st v1ServiceTester) LookupResources(_ context.Context, resourceRelation
 		},
 		Consistency: &v1.Consistency{
 			Requirement: &v1.Consistency_AtLeastAsFresh{
-				AtLeastAsFresh: zedtoken.MustNewFromRevision(atRevision),
+				AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(atRevision),
 			},
 		},
 		OptionalLimit:  limit,
@@ -230,7 +230,7 @@ func (v1st v1ServiceTester) LookupSubjects(_ context.Context, resource tuple.Obj
 		OptionalSubjectRelation: optionalizeRelation(subjectRelation.Relation),
 		Consistency: &v1.Consistency{
 			Requirement: &v1.Consistency_AtLeastAsFresh{
-				AtLeastAsFresh: zedtoken.MustNewFromRevision(atRevision),
+				AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(atRevision),
 			},
 		},
 		Context: builtContext,
@@ -260,7 +260,7 @@ func (v1st v1ServiceTester) BulkCheck(ctx context.Context, items []*v1.BulkCheck
 		Items: items,
 		Consistency: &v1.Consistency{
 			Requirement: &v1.Consistency_AtLeastAsFresh{
-				AtLeastAsFresh: zedtoken.MustNewFromRevision(atRevision),
+				AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(atRevision),
 			},
 		},
 	})
@@ -276,7 +276,7 @@ func (v1st v1ServiceTester) CheckBulk(ctx context.Context, items []*v1.CheckBulk
 		Items: items,
 		Consistency: &v1.Consistency{
 			Requirement: &v1.Consistency_AtLeastAsFresh{
-				AtLeastAsFresh: zedtoken.MustNewFromRevision(atRevision),
+				AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(atRevision),
 			},
 		},
 	})

internal/services/integrationtesting/perf_test.go

@@ -58,7 +58,7 @@ func TestBurst(t *testing.T) {
 					_, err := client.CheckPermission(context.Background(), &v1.CheckPermissionRequest{
 						Consistency: &v1.Consistency{
 							Requirement: &v1.Consistency_AtLeastAsFresh{
-								AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+								AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 							},
 						},
 						Resource:   rel.Resource,

internal/services/v1/debug_test.go

@@ -507,7 +507,7 @@ func TestCheckPermissionWithDebug(t *testing.T) {
 					checkResp, err := client.CheckPermission(ctx, &v1.CheckPermissionRequest{
 						Consistency: &v1.Consistency{
 							Requirement: &v1.Consistency_AtLeastAsFresh{
-								AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+								AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 							},
 						},
 						Resource:   stc.checkRequest.resource,

internal/services/v1/experimental.go

@@ -525,10 +525,16 @@ func (es *experimentalServer) ExperimentalReflectSchema(ctx context.Context, req
 		}
 	}
 
+	ds := datastoremw.MustFromContext(ctx)
+	readAt, err := zedtoken.NewFromRevision(ctx, atRevision, ds)
+	if err != nil {
+		return nil, shared.RewriteErrorWithoutConfig(ctx, err)
+	}
+
 	return &v1.ExperimentalReflectSchemaResponse{
 		Definitions: definitions,
 		Caveats:     caveats,
-		ReadAt:      zedtoken.MustNewFromRevision(atRevision),
+		ReadAt:      readAt,
 	}, nil
 }
 
@@ -543,12 +549,21 @@ func (es *experimentalServer) ExperimentalDiffSchema(ctx context.Context, req *v
 		return nil, shared.RewriteErrorWithoutConfig(ctx, err)
 	}
 
-	resp, err := convertDiff(diff, existingSchema, comparisonSchema, atRevision)
+	ds := datastoremw.MustFromContext(ctx)
+	diffs, err := convertDiff(diff, existingSchema, comparisonSchema)
 	if err != nil {
 		return nil, shared.RewriteErrorWithoutConfig(ctx, err)
 	}
 
-	return resp, nil
+	readAt, err := zedtoken.NewFromRevision(ctx, atRevision, ds)
+	if err != nil {
+		return nil, shared.RewriteErrorWithoutConfig(ctx, err)
+	}
+
+	return &v1.ExperimentalDiffSchemaResponse{
+		Diffs:  diffs,
+		ReadAt: readAt,
+	}, nil
 }
 
 func (es *experimentalServer) ExperimentalComputablePermissions(ctx context.Context, req *v1.ExperimentalComputablePermissionsRequest) (*v1.ExperimentalComputablePermissionsResponse, error) {
@@ -749,11 +764,16 @@ func (es *experimentalServer) ExperimentalCountRelationships(ctx context.Context
 		return nil, spiceerrors.MustBugf("count should not be negative")
 	}
 
+	readAt, err := zedtoken.NewFromRevision(ctx, headRev, ds)
+	if err != nil {
+		return nil, shared.RewriteErrorWithoutConfig(ctx, err)
+	}
+
 	return &v1.ExperimentalCountRelationshipsResponse{
 		CounterResult: &v1.ExperimentalCountRelationshipsResponse_ReadCounterValue{
 			ReadCounterValue: &v1.ReadCounterValue{
 				RelationshipCount: uintCount,
-				ReadAt:            zedtoken.MustNewFromRevision(headRev),
+				ReadAt:            readAt,
 			},
 		},
 	}, nil

internal/services/v1/expreflection.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/authzed/spicedb/pkg/caveats"
 	caveattypes "github.com/authzed/spicedb/pkg/caveats/types"
-	"github.com/authzed/spicedb/pkg/datastore"
 	"github.com/authzed/spicedb/pkg/diff"
 	caveatdiff "github.com/authzed/spicedb/pkg/diff/caveats"
 	nsdiff "github.com/authzed/spicedb/pkg/diff/namespace"
@@ -18,7 +17,6 @@ import (
 	iv1 "github.com/authzed/spicedb/pkg/proto/impl/v1"
 	"github.com/authzed/spicedb/pkg/spiceerrors"
 	"github.com/authzed/spicedb/pkg/tuple"
-	"github.com/authzed/spicedb/pkg/zedtoken"
 )
 
 type schemaFilters struct {
@@ -180,8 +178,7 @@ func convertDiff(
 	diff *diff.SchemaDiff,
 	existingSchema *diff.DiffableSchema,
 	comparisonSchema *diff.DiffableSchema,
-	atRevision datastore.Revision,
-) (*v1.ExperimentalDiffSchemaResponse, error) {
+) ([]*v1.ExpSchemaDiff, error) {
 	size := len(diff.AddedNamespaces) + len(diff.RemovedNamespaces) + len(diff.AddedCaveats) + len(diff.RemovedCaveats) + len(diff.ChangedNamespaces) + len(diff.ChangedCaveats)
 	diffs := make([]*v1.ExpSchemaDiff, 0, size)
 
@@ -513,10 +510,7 @@ func convertDiff(
 		}
 	}
 
-	return &v1.ExperimentalDiffSchemaResponse{
-		Diffs:  diffs,
-		ReadAt: zedtoken.MustNewFromRevision(atRevision),
-	}, nil
+	return diffs, nil
 }
 
 // namespaceAPIReprForName builds an API representation of a namespace.

internal/services/v1/expreflection_test.go

@@ -9,7 +9,6 @@ import (
 	"github.com/ettle/strcase"
 	"github.com/stretchr/testify/require"
 
-	"github.com/authzed/spicedb/pkg/datastore/revisionparsing"
 	"github.com/authzed/spicedb/pkg/diff"
 	"github.com/authzed/spicedb/pkg/genutil/mapz"
 	"github.com/authzed/spicedb/pkg/schemadsl/compiler"
@@ -544,21 +543,18 @@ func TestConvertDiff(t *testing.T) {
 			diff, err := diff.DiffSchemas(es, cs)
 			require.NoError(t, err)
 
-			resp, err := convertDiff(
+			diffs, err := convertDiff(
 				diff,
 				&es,
 				&cs,
-				revisionparsing.MustParseRevisionForTest("1"),
 			)
 			if err != nil {
 				t.Fatalf("unexpected error: %v", err)
 			}
-			require.NotNil(t, resp.ReadAt)
-			resp.ReadAt = nil
 
-			testutil.RequireProtoEqual(t, tc.expectedResponse, resp, "got mismatch")
+			testutil.RequireProtoEqual(t, tc.expectedResponse, &v1.ExperimentalDiffSchemaResponse{Diffs: diffs}, "got mismatch")
 
-			for _, diff := range resp.Diffs {
+			for _, diff := range diffs {
 				name := reflect.TypeOf(diff.GetDiff()).String()
 				encounteredDiffTypes.Add(strings.ToLower(strings.Split(name, "_")[1]))
 			}

internal/services/v1/metadata_test.go

@@ -38,7 +38,7 @@ func TestAllMethodsReturnMetadata(t *testing.T) {
 				_, err := client.CheckPermission(ctx, &v1.CheckPermissionRequest{
 					Consistency: &v1.Consistency{
 						Requirement: &v1.Consistency_AtLeastAsFresh{
-							AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+							AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 						},
 					},
 					Resource:   obj("document", "masterplan"),
@@ -53,7 +53,7 @@ func TestAllMethodsReturnMetadata(t *testing.T) {
 				_, err := client.CheckBulkPermissions(ctx, &v1.CheckBulkPermissionsRequest{
 					Consistency: &v1.Consistency{
 						Requirement: &v1.Consistency_AtLeastAsFresh{
-							AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+							AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 						},
 					},
 					Items: []*v1.CheckBulkPermissionsRequestItem{
@@ -96,7 +96,7 @@ func TestAllMethodsReturnMetadata(t *testing.T) {
 				_, err := client.ExpandPermissionTree(ctx, &v1.ExpandPermissionTreeRequest{
 					Consistency: &v1.Consistency{
 						Requirement: &v1.Consistency_AtLeastAsFresh{
-							AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+							AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 						},
 					},
 					Resource:   obj("document", "masterplan"),
@@ -130,7 +130,7 @@ func TestAllMethodsReturnMetadata(t *testing.T) {
 				stream, err := client.LookupResources(ctx, &v1.LookupResourcesRequest{
 					Consistency: &v1.Consistency{
 						Requirement: &v1.Consistency_AtLeastAsFresh{
-							AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+							AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 						},
 					},
 					ResourceObjectType: "document",
@@ -155,7 +155,7 @@ func TestAllMethodsReturnMetadata(t *testing.T) {
 				stream, err := client.LookupSubjects(ctx, &v1.LookupSubjectsRequest{
 					Consistency: &v1.Consistency{
 						Requirement: &v1.Consistency_AtLeastAsFresh{
-							AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+							AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 						},
 					},
 					Resource:          obj("document", "masterplan"),
@@ -188,7 +188,7 @@ func TestAllMethodsReturnMetadata(t *testing.T) {
 				stream, err := client.ExportBulkRelationships(ctx, &v1.ExportBulkRelationshipsRequest{
 					Consistency: &v1.Consistency{
 						Requirement: &v1.Consistency_AtLeastAsFresh{
-							AtLeastAsFresh: zedtoken.MustNewFromRevision(revision),
+							AtLeastAsFresh: zedtoken.MustNewFromRevisionForTesting(revision),
 						},
 					},
 				}, grpc.Trailer(&trailer))