diff --git a/internal/datastore/crdb/crdb.go b/internal/datastore/crdb/crdb.go index 2175cc1055..9882420bb0 100644 --- a/internal/datastore/crdb/crdb.go +++ b/internal/datastore/crdb/crdb.go @@ -164,23 +164,16 @@ func newCRDBDatastore(ctx context.Context, url string, options ...Option) (datas keyer = noOverlapKeyer } - maxRevisionStaleness := time.Duration(float64(config.revisionQuantization.Nanoseconds())* - config.maxRevisionStalenessPercent) * time.Nanosecond - headMigration, err := migrations.CRDBMigrations.HeadRevision() if err != nil { return nil, fmt.Errorf("invalid head migration found for cockroach: %w", err) } ds := &crdbDatastore{ - RemoteClockRevisions: revisions.NewRemoteClockRevisions( - config.gcWindow, - maxRevisionStaleness, - config.followerReadDelay, - config.revisionQuantization, - ), CommonDecoder: revisions.CommonDecoder{Kind: revisions.HybridLogicalClock}, MigrationValidator: common.NewMigrationValidator(headMigration, config.allowedMigrations), + followerReadDelay: config.followerReadDelay, + revisionQuantization: config.revisionQuantization, dburl: url, acquireTimeout: config.acquireTimeout, watchBufferLength: config.watchBufferLength, @@ -198,8 +191,6 @@ func newCRDBDatastore(ctx context.Context, url string, options ...Option) (datas watchEnabled: !config.watchDisabled, schema: *schema.Schema(config.columnOptimizationOption, config.withIntegrity, false), } - ds.SetNowFunc(ds.headRevisionInternal) - ds.SetNowOnlyFunc(ds.headRevisionInternalNoHash) // this ctx and cancel is tied to the lifetime of the datastore ds.ctx, ds.cancel = context.WithCancel(context.Background()) @@ -257,11 +248,68 @@ func NewCRDBDatastore(ctx context.Context, url string, options ...Option) (datas return datastore.NewSeparatingContextDatastoreProxy(ds), nil } +// OptimizedRevision computes the uncached quantized revision for CockroachDB. It +// reads the current HLC time and quantizes it in Go (CRDB's only revision +// primitive is cluster_logical_timestamp()). Caching, deduplication, and jitter +// are layered on by proxy.NewOptimizedRevisionProxy. +func (cds *crdbDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { + ctx, span := tracer.Start(ctx, "OptimizedRevision") + defer span.End() + + nowRev, schemaHash, err := cds.headRevisionInternal(ctx) + if err != nil { + return datastore.NoRevision, 0, "", err + } + + if nowRev == datastore.NoRevision { + return datastore.NoRevision, 0, "", datastore.NewInvalidRevisionErr(nowRev, datastore.CouldNotDetermineRevision) + } + + nowTS, ok := nowRev.(revisions.WithTimestampRevision) + if !ok { + return datastore.NoRevision, 0, "", spiceerrors.MustBugf("expected with-timestamp revision, got %T", nowRev) + } + + rev, validFor := revisions.QuantizeHLC(nowTS, cds.followerReadDelay, cds.revisionQuantization) + return rev, validFor, schemaHash, nil +} + +// CheckRevision verifies the given revision is within the software GC window. It +// lives on the datastore (not the caching layer) because it is not a caching +// concern. It uses the hash-free head revision query since the schema hash is +// not needed here. +func (cds *crdbDatastore) CheckRevision(ctx context.Context, dsRevision datastore.Revision) error { + if dsRevision == datastore.NoRevision { + return datastore.NewInvalidRevisionErr(dsRevision, datastore.CouldNotDetermineRevision) + } + + revision, ok := dsRevision.(revisions.WithTimestampRevision) + if !ok { + return spiceerrors.MustBugf("expected HLC revision, got %T", dsRevision) + } + + ctx, span := tracer.Start(ctx, "CheckRevision") + defer span.End() + + now, err := cds.headRevisionInternalNoHash(ctx) + if err != nil { + return err + } + + nowTS, ok := now.(revisions.WithTimestampRevision) + if !ok { + return spiceerrors.MustBugf("expected HLC revision, got %T", now) + } + + return revisions.CheckHLCGCWindow(nowTS, revision, cds.gcWindow) +} + type crdbDatastore struct { - *revisions.RemoteClockRevisions revisions.CommonDecoder *common.MigrationValidator + followerReadDelay time.Duration + revisionQuantization time.Duration dburl string readPool, writePool *pool.RetryPool collectors []prometheus.Collector diff --git a/internal/datastore/crdb/crdb_test.go b/internal/datastore/crdb/crdb_test.go index 69a2f9f4c7..7500786927 100644 --- a/internal/datastore/crdb/crdb_test.go +++ b/internal/datastore/crdb/crdb_test.go @@ -167,13 +167,13 @@ func TestCRDBDatastoreWithFollowerReads(t *testing.T) { // Revisions should be at least the follower read delay amount in the past for start := time.Now(); time.Since(start) < 50*time.Millisecond; { - testRevisionResult, err := ds.OptimizedRevision(ctx) + testRevision, _, _, err := ds.OptimizedRevision(ctx) require.NoError(t, err) nowRevisionResult, err := ds.HeadRevision(ctx) require.NoError(t, err) - diff := nowRevisionResult.Revision.(revisions.HLCRevision).TimestampNanoSec() - testRevisionResult.Revision.(revisions.HLCRevision).TimestampNanoSec() + diff := nowRevisionResult.Revision.(revisions.HLCRevision).TimestampNanoSec() - testRevision.(revisions.HLCRevision).TimestampNanoSec() require.Greater(t, diff, followerReadDelay.Nanoseconds()) } }) diff --git a/internal/datastore/memdb/memdb_test.go b/internal/datastore/memdb/memdb_test.go index b16298dc55..75cc868b66 100644 --- a/internal/datastore/memdb/memdb_test.go +++ b/internal/datastore/memdb/memdb_test.go @@ -140,7 +140,7 @@ func TestAnythingAfterCloseDoesNotPanic(t *testing.T) { err = ds.CheckRevision(t.Context(), lowestRevision.Revision) require.ErrorIs(err, ErrMemDBIsClosed) - _, err = ds.OptimizedRevision(t.Context()) + _, _, _, err = ds.OptimizedRevision(t.Context()) require.ErrorIs(err, ErrMemDBIsClosed) reader := ds.SnapshotReader(datastore.NoRevision) diff --git a/internal/datastore/memdb/revisions.go b/internal/datastore/memdb/revisions.go index 39aae92414..4a5ade409e 100644 --- a/internal/datastore/memdb/revisions.go +++ b/internal/datastore/memdb/revisions.go @@ -68,17 +68,20 @@ func (mdb *memdbDatastore) headRevisionNoLock() revisions.TimestampRevision { return mdb.revisions[len(mdb.revisions)-1].revision } -func (mdb *memdbDatastore) OptimizedRevision(_ context.Context) (datastore.RevisionWithSchemaHash, error) { +func (mdb *memdbDatastore) OptimizedRevision(_ context.Context) (datastore.Revision, time.Duration, string, error) { mdb.RLock() defer mdb.RUnlock() if err := mdb.checkNotClosed(); err != nil { - return datastore.RevisionWithSchemaHash{}, err + return datastore.NoRevision, 0, "", err } now := nowRevision() var optimized revisions.TimestampRevision + var validFor time.Duration if mdb.quantizationPeriod > 0 { - optimized = revisions.NewForTimestamp(now.TimestampNanoSec() - now.TimestampNanoSec()%mdb.quantizationPeriod) + afterLastQuantization := now.TimestampNanoSec() % mdb.quantizationPeriod + optimized = revisions.NewForTimestamp(now.TimestampNanoSec() - afterLastQuantization) + validFor = time.Duration(mdb.quantizationPeriod-afterLastQuantization) * time.Nanosecond } else { optimized = now } @@ -94,7 +97,7 @@ func (mdb *memdbDatastore) OptimizedRevision(_ context.Context) (datastore.Revis } } - return datastore.RevisionWithSchemaHash{Revision: optimized, SchemaHash: hash}, nil + return optimized, validFor, hash, nil } func (mdb *memdbDatastore) CheckRevision(_ context.Context, dr datastore.Revision) error { diff --git a/internal/datastore/mysql/datastore.go b/internal/datastore/mysql/datastore.go index 85e2b52a53..f390b5ec9a 100644 --- a/internal/datastore/mysql/datastore.go +++ b/internal/datastore/mysql/datastore.go @@ -201,9 +201,6 @@ func newMySQLDatastore(ctx context.Context, uri string, replicaIndex int, option gcCtx, cancelGc := context.WithCancel(context.Background()) - maxRevisionStaleness := time.Duration(float64(config.revisionQuantization.Nanoseconds())* - config.maxRevisionStalenessPercent) * time.Nanosecond - quantizationPeriodNanos := max(config.revisionQuantization.Nanoseconds(), 1) followerReadDelayNanos := max(config.followerReadDelay.Nanoseconds(), 0) @@ -269,17 +266,12 @@ func newMySQLDatastore(ctx context.Context, uri string, replicaIndex int, option maxRetries: config.maxRetries, analyzeBeforeStats: config.analyzeBeforeStats, schema: *schema, - CachedOptimizedRevisions: revisions.NewCachedOptimizedRevisions( - maxRevisionStaleness, - ), CommonDecoder: revisions.CommonDecoder{ Kind: revisions.TransactionID, }, filterMaximumIDCount: config.filterMaximumIDCount, } - store.SetOptimizedRevisionFunc(store.optimizedRevisionFunc) - ctx, cancel := context.WithTimeout(context.Background(), seedingTimeout) defer cancel() err = store.seedDatabase(ctx) @@ -475,7 +467,6 @@ func newMySQLExecutor(tx querier, explainable datastore.Explainable) common.Exec // Datastore is a MySQL-based implementation of the datastore.Datastore interface type mysqlDatastore struct { - *revisions.CachedOptimizedRevisions *common.MigrationValidator db *sql.DB diff --git a/internal/datastore/mysql/datastore_test.go b/internal/datastore/mysql/datastore_test.go index 97b0d41b46..aadba4b133 100644 --- a/internal/datastore/mysql/datastore_test.go +++ b/internal/datastore/mysql/datastore_test.go @@ -764,9 +764,9 @@ func TransactionTimestampsTest(t *testing.T, ds datastore.Datastore) { // Let's make sure both Now() and transactionCreated() have timezones aligned req.Less(ts.Sub(startTimeUTC), 5*time.Minute) - revisionResult, err := ds.OptimizedRevision(ctx) + rev, _, _, err := ds.OptimizedRevision(ctx) req.NoError(err) - req.Equal(revisions.NewForTransactionID(txID), revisionResult.Revision) + req.Equal(revisions.NewForTransactionID(txID), rev) } func TestMySQLMigrations(t *testing.T) { diff --git a/internal/datastore/mysql/revisions.go b/internal/datastore/mysql/revisions.go index 5c8e9f7f28..5495891224 100644 --- a/internal/datastore/mysql/revisions.go +++ b/internal/datastore/mysql/revisions.go @@ -76,7 +76,7 @@ const ( ) as unknown;` ) -func (mds *mysqlDatastore) optimizedRevisionFunc(ctx context.Context) (datastore.Revision, time.Duration, string, error) { +func (mds *mysqlDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { var rev uint64 var validForNanos time.Duration var schemaHash []byte diff --git a/internal/datastore/postgres/postgres.go b/internal/datastore/postgres/postgres.go index 8ac1fbf88e..85ef7463d4 100644 --- a/internal/datastore/postgres/postgres.go +++ b/internal/datastore/postgres/postgres.go @@ -29,7 +29,6 @@ import ( pgxcommon "github.com/authzed/spicedb/internal/datastore/postgres/common" "github.com/authzed/spicedb/internal/datastore/postgres/migrations" "github.com/authzed/spicedb/internal/datastore/postgres/schema" - "github.com/authzed/spicedb/internal/datastore/revisions" log "github.com/authzed/spicedb/internal/logging" "github.com/authzed/spicedb/internal/sharederrors" "github.com/authzed/spicedb/pkg/datastore" @@ -287,9 +286,6 @@ func newPostgresDatastore( schema.ColSnapshot, ) - maxRevisionStaleness := time.Duration(float64(config.revisionQuantization.Nanoseconds())* - config.maxRevisionStalenessPercent) * time.Nanosecond - isolationLevel := pgx.Serializable if config.relaxedIsolationLevel { isolationLevel = pgx.RepeatableRead @@ -298,9 +294,6 @@ func newPostgresDatastore( startGarbageCollector := datastore.StartGarbageCollector datastore := &pgDatastore{ - CachedOptimizedRevisions: revisions.NewCachedOptimizedRevisions( - maxRevisionStaleness, - ), MigrationValidator: common.NewMigrationValidator(headMigration, config.allowedMigrations), dburl: pgURL, readPool: pgxcommon.MustNewInterceptorPooler(readPool, config.queryInterceptor), @@ -338,8 +331,6 @@ func newPostgresDatastore( datastore.writePool = pgxcommon.MustNewInterceptorPooler(writePool, config.queryInterceptor) } - datastore.SetOptimizedRevisionFunc(datastore.optimizedRevisionFunc) - // Start a goroutine for garbage collection and the revision heartbeat. if isPrimary { datastore.workerGroup, datastore.workerCtx = errgroup.WithContext(datastore.workerCtx) @@ -368,7 +359,6 @@ func newPostgresDatastore( } type pgDatastore struct { - *revisions.CachedOptimizedRevisions *common.MigrationValidator dburl string diff --git a/internal/datastore/postgres/postgres_shared_test.go b/internal/datastore/postgres/postgres_shared_test.go index 8e82ab22af..f2d5350dfa 100644 --- a/internal/datastore/postgres/postgres_shared_test.go +++ b/internal/datastore/postgres/postgres_shared_test.go @@ -1045,7 +1045,7 @@ func assertRevisionLowerAndHigher(ctx context.Context, t *testing.T, ds datastor var snapshot pgSnapshot pgDS, ok := ds.(*pgDatastore) require.True(t, ok) - rev, _, _, err := pgDS.optimizedRevisionFunc(ctx) + rev, _, _, err := pgDS.OptimizedRevision(ctx) require.NoError(t, err) pgRev, ok := rev.(postgresRevision) diff --git a/internal/datastore/postgres/revisions.go b/internal/datastore/postgres/revisions.go index b800346a4a..5ba7fcffa3 100644 --- a/internal/datastore/postgres/revisions.go +++ b/internal/datastore/postgres/revisions.go @@ -128,7 +128,7 @@ const ( queryLatestXID = `SELECT max(xid)::text::integer FROM relation_tuple_transaction;` ) -func (pgd *pgDatastore) optimizedRevisionFunc(ctx context.Context) (datastore.Revision, time.Duration, string, error) { +func (pgd *pgDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { var revision xid8 var snapshot pgSnapshot var validForNanos time.Duration diff --git a/internal/datastore/proxy/checkingreplicated_test.go b/internal/datastore/proxy/checkingreplicated_test.go index 4c4c5fd047..f879eb742e 100644 --- a/internal/datastore/proxy/checkingreplicated_test.go +++ b/internal/datastore/proxy/checkingreplicated_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "testing" + "time" "github.com/stretchr/testify/require" @@ -179,8 +180,8 @@ func (f fakeDatastore) ReadWriteTx(_ context.Context, _ datastore.TxUserFunc, _ return nil, nil } -func (f fakeDatastore) OptimizedRevision(_ context.Context) (datastore.RevisionWithSchemaHash, error) { - return datastore.RevisionWithSchemaHash{}, nil +func (f fakeDatastore) OptimizedRevision(_ context.Context) (datastore.Revision, time.Duration, string, error) { + return datastore.NoRevision, 0, "", nil } func (f fakeDatastore) HeadRevision(_ context.Context) (datastore.RevisionWithSchemaHash, error) { diff --git a/internal/datastore/proxy/indexcheck/fakedatastore_test.go b/internal/datastore/proxy/indexcheck/fakedatastore_test.go index 770a332924..353a39301c 100644 --- a/internal/datastore/proxy/indexcheck/fakedatastore_test.go +++ b/internal/datastore/proxy/indexcheck/fakedatastore_test.go @@ -3,6 +3,7 @@ package indexcheck import ( "context" "fmt" + "time" v1 "github.com/authzed/authzed-go/proto/authzed/api/v1" @@ -39,8 +40,8 @@ func (f fakeDatastore) ReadWriteTx(ctx context.Context, fn datastore.TxUserFunc, }) } -func (f fakeDatastore) OptimizedRevision(_ context.Context) (datastore.RevisionWithSchemaHash, error) { - return datastore.RevisionWithSchemaHash{}, nil +func (f fakeDatastore) OptimizedRevision(_ context.Context) (datastore.Revision, time.Duration, string, error) { + return nil, 0, "", nil } func (f fakeDatastore) HeadRevision(_ context.Context) (datastore.RevisionWithSchemaHash, error) { diff --git a/internal/datastore/proxy/indexcheck/indexcheck.go b/internal/datastore/proxy/indexcheck/indexcheck.go index 8b501a4e5f..08989ed9fc 100644 --- a/internal/datastore/proxy/indexcheck/indexcheck.go +++ b/internal/datastore/proxy/indexcheck/indexcheck.go @@ -3,6 +3,7 @@ package indexcheck import ( "context" "fmt" + "time" v1 "github.com/authzed/authzed-go/proto/authzed/api/v1" @@ -56,7 +57,7 @@ func (p *indexcheckingProxy) UniqueID(ctx context.Context) (string, error) { return p.delegate.UniqueID(ctx) } -func (p *indexcheckingProxy) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (p *indexcheckingProxy) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { return p.delegate.OptimizedRevision(ctx) } diff --git a/internal/datastore/proxy/indexcheck/indexcheck_test.go b/internal/datastore/proxy/indexcheck/indexcheck_test.go index d4e0ef354b..9ff129a75b 100644 --- a/internal/datastore/proxy/indexcheck/indexcheck_test.go +++ b/internal/datastore/proxy/indexcheck/indexcheck_test.go @@ -97,9 +97,9 @@ func TestIndexCheckingProxyMethods(t *testing.T) { }) t.Run("OptimizedRevision", func(t *testing.T) { - result, err := proxy.OptimizedRevision(t.Context()) + rev, _, _, err := proxy.OptimizedRevision(t.Context()) require.NoError(t, err) - require.Nil(t, result.Revision) + require.Nil(t, rev) }) t.Run("CheckRevision", func(t *testing.T) { diff --git a/internal/datastore/proxy/observable.go b/internal/datastore/proxy/observable.go index e819f7d45a..424d00179e 100644 --- a/internal/datastore/proxy/observable.go +++ b/internal/datastore/proxy/observable.go @@ -2,6 +2,7 @@ package proxy import ( "context" + "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" @@ -98,7 +99,7 @@ func (p *observableProxy) ReadWriteTx( }, opts...) } -func (p *observableProxy) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (p *observableProxy) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { ctx, closer := observe(ctx, "OptimizedRevision", "") defer closer() diff --git a/internal/datastore/proxy/observable_test.go b/internal/datastore/proxy/observable_test.go index 60f641a503..350156cc05 100644 --- a/internal/datastore/proxy/observable_test.go +++ b/internal/datastore/proxy/observable_test.go @@ -3,6 +3,7 @@ package proxy import ( "context" "testing" + "time" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" @@ -74,10 +75,10 @@ func TestObservableProxy_DatastoreMethodsWithMetrics(t *testing.T) { name: "OptimizedRevision", metricOp: "OptimizedRevision", setupMock: func(ds *proxy_test.MockDatastore) { - ds.On("OptimizedRevision").Return(datastore.RevisionWithSchemaHash{Revision: testRev}, nil).Once() + ds.On("OptimizedRevision").Return(testRev, time.Duration(0), "", nil).Once() }, call: func(t *testing.T, ds datastore.Datastore) { - _, err := ds.OptimizedRevision(t.Context()) + _, _, _, err := ds.OptimizedRevision(t.Context()) require.NoError(t, err) }, }, diff --git a/internal/datastore/proxy/optimized_revision.go b/internal/datastore/proxy/optimized_revision.go new file mode 100644 index 0000000000..3a8b0a994e --- /dev/null +++ b/internal/datastore/proxy/optimized_revision.go @@ -0,0 +1,207 @@ +package proxy + +import ( + "context" + "fmt" + "math/rand" + "sync" + "time" + + "github.com/benbjohnson/clock" + "go.opentelemetry.io/otel/trace" + "resenje.org/singleflight" + + log "github.com/authzed/spicedb/internal/logging" + "github.com/authzed/spicedb/internal/telemetry/otelconv" + "github.com/authzed/spicedb/pkg/datastore" +) + +// defaultOptimizedRevisionSharedTimeout aggressively bounds the *shared*, +// singleflighted computation of the optimized revision. +// +// The computation runs under singleflight, which replaces the caller's context +// with one that has *no* deadline (see resenje.org/singleflight: the work context +// is `context.WithoutCancel(ctx)`, cancelled only once every awaiting caller has +// gone away). The per-request gRPC deadline that would normally abort a stuck +// datastore query is therefore stripped. Worse, because every API request needs +// an optimized revision and they all de-duplicate onto this one in-flight call, +// every concurrent caller observes its latency directly: a wedged computation +// (e.g. a half-open connection silently dropped by a load balancer) inflates the +// P99 of the entire system, not just one request. +// +// Keeping this bound low limits that blast radius. The healthy computation takes +// single-digit milliseconds, so this is extremely generous for the happy path. On +// failure, OptimizedRevision retries the computation directly (bypassing +// singleflight) on the caller's own context, so the aggressive bound does not turn +// a transient slow/wedged shared attempt into a failed request. +const defaultOptimizedRevisionSharedTimeout = 2 * time.Second + +// defaultOptimizedRevisionFallbackTimeout bounds the *direct* retry performed +// after a failed shared attempt. The retry runs on the caller's own context, so +// its deadline applies when present; this fallback only takes effect for callers +// with no deadline of their own, preserving the guarantee that a revision +// computation can never block forever. +const defaultOptimizedRevisionFallbackTimeout = 10 * time.Second + +// NewOptimizedRevisionProxy wraps a datastore with an in-process cache for its +// optimized revision. Concurrent misses are deduplicated via singleflight, the +// expiry decision is jittered by up to maxStaleness to avoid a thundering herd at +// quantization boundaries, and the shared computation is bounded so a wedged +// datastore call cannot pin the latency of every waiting caller. +// +// The wrapped datastore's OptimizedRevision is expected to be uncached: one +// database round-trip per call, returning the revision, how long it remains +// valid, and the schema hash visible at it. Caching is entirely the proxy's job. +func NewOptimizedRevisionProxy(d datastore.Datastore, maxStaleness time.Duration) datastore.Datastore { + return &optimizedRevisionProxy{ + Datastore: d, + maxStaleness: maxStaleness, + sharedTimeout: defaultOptimizedRevisionSharedTimeout, + fallbackTimeout: defaultOptimizedRevisionFallbackTimeout, + clock: clock.New(), + } +} + +// optimizedRevisionProxy is both the proxy in the datastore chain and the cache +// itself: the candidate list, its mutex, the singleflight group, and the jitter +// all live directly on it. +type optimizedRevisionProxy struct { + datastore.Datastore + + maxStaleness time.Duration + sharedTimeout time.Duration + fallbackTimeout time.Duration + clock clock.Clock + + mu sync.Mutex + candidates []validRevision // GUARDED_BY(mu) + + // flight consolidates concurrent misses into a single datastore round-trip. + flight singleflight.Group[string, cachedRevision] +} + +// cachedRevision is the value carried through the singleflight group. +type cachedRevision struct { + revision datastore.Revision + validFor time.Duration + schemaHash string +} + +// validRevision is a cached candidate revision and the wall-clock time through +// which it remains acceptable. +type validRevision struct { + revision datastore.Revision + validThrough time.Time + schemaHash string +} + +func (p *optimizedRevisionProxy) Unwrap() datastore.Datastore { + return p.Datastore +} + +func (p *optimizedRevisionProxy) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { + ctx, span := tracer.Start(ctx, "optimizedRevisionProxy.OptimizedRevision") + defer span.End() + + localNow := p.clock.Now() + + // Subtract a random amount of time from now, to let barely expired candidates get selected. + adjustedNow := localNow + if p.maxStaleness > 0 { + // nolint:gosec + // G404 use of non cryptographically secure random number generator is not a security concern here, + // as we are using it to introduce randomness to the accepted staleness of a revision and reduce the odds of + // a thundering herd to the datastore + adjustedNow = localNow.Add(-1 * time.Duration(rand.Int63n(p.maxStaleness.Nanoseconds())) * time.Nanosecond) + } + + p.mu.Lock() + for _, candidate := range p.candidates { + if candidate.validThrough.After(adjustedNow) { + p.mu.Unlock() + log.Ctx(ctx).Debug().Time("now", localNow).Time("valid", candidate.validThrough).Msg("returning cached revision") + span.AddEvent(otelconv.EventDatastoreRevisionsCacheReturned) + return candidate.revision, candidate.validThrough.Sub(localNow), candidate.schemaHash, nil + } + } + p.mu.Unlock() + + // Compute the revision under singleflight so concurrent callers share a single + // datastore round-trip. The shared call is aggressively bounded (see + // defaultOptimizedRevisionSharedTimeout) so a wedged computation cannot pin the + // latency of every waiting caller. + result, _, err := p.flight.Do(ctx, "", func(sfCtx context.Context) (cachedRevision, error) { + // NOTE: singleflight hands this function a context with the caller's + // deadline stripped. Re-impose a (low) deadline so a hung datastore call + // cannot block this in-flight call, and therefore every caller waiting on + // it, beyond the bound. + if p.sharedTimeout > 0 { + var cancel context.CancelFunc + sfCtx, cancel = context.WithTimeout(sfCtx, p.sharedTimeout) + defer cancel() + } + + return p.compute(sfCtx, localNow, span) + }) + if err == nil { + return result.revision, result.validFor, result.schemaHash, nil + } + + // If this caller's own context is already done, surface its cancellation / + // deadline error directly; there is nothing to retry. + if ctx.Err() != nil { + return datastore.NoRevision, 0, "", ctx.Err() + } + + // The shared, aggressively-bounded attempt failed (e.g. a wedged connection + // tripped the timeout, or the datastore returned a transient error). Give this + // request one more chance, directly and outside of singleflight, on its own + // context: this is not capped by the low shared bound and cannot re-attach to a + // poisoned in-flight call, so a brief wedge does not fail the request. For the + // pooled SQL datastores the retry is naturally throttled by the pool's maximum + // size; Spanner's client similarly bounds concurrent calls via its own session + // pool. Even when many waiters retry at once, the datastore is not overwhelmed. + span.AddEvent(otelconv.EventDatastoreRevisionsSharedFailedRetrying) + log.Ctx(ctx).Warn().Err(err).Msg("shared optimized revision computation failed; retrying directly") + + retryCtx, cancel := context.WithTimeout(ctx, p.fallbackTimeout) + defer cancel() + result, err = p.compute(retryCtx, p.clock.Now(), span) + if err != nil { + return datastore.NoRevision, 0, "", err + } + return result.revision, result.validFor, result.schemaHash, nil +} + +// compute fetches an uncached optimized revision from the wrapped datastore and +// records it as a cache candidate. It is invoked both from the shared singleflight +// path and from the direct retry that follows a failed shared attempt. +func (p *optimizedRevisionProxy) compute(ctx context.Context, localNow time.Time, span trace.Span) (cachedRevision, error) { + log.Ctx(ctx).Debug().Time("now", localNow).Msg("computing new revision") + + optimized, validFor, schemaHash, err := p.Datastore.OptimizedRevision(ctx) + if err != nil { + return cachedRevision{}, fmt.Errorf("unable to compute optimized revision: %w", err) + } + + rvt := localNow.Add(validFor) + + // Prune the candidates that have definitely expired. + p.mu.Lock() + var numToDrop uint + for _, candidate := range p.candidates { + if candidate.validThrough.Add(p.maxStaleness).Before(localNow) { + numToDrop++ + } else { + break + } + } + + p.candidates = p.candidates[numToDrop:] + p.candidates = append(p.candidates, validRevision{optimized, rvt, schemaHash}) + p.mu.Unlock() + + span.AddEvent(otelconv.EventDatastoreRevisionsComputed) + log.Ctx(ctx).Debug().Time("now", localNow).Time("valid", rvt).Stringer("validFor", validFor).Msg("setting valid through") + return cachedRevision{revision: optimized, validFor: validFor, schemaHash: schemaHash}, nil +} diff --git a/internal/datastore/revisions/optimized_test.go b/internal/datastore/proxy/optimized_revision_test.go similarity index 59% rename from internal/datastore/revisions/optimized_test.go rename to internal/datastore/proxy/optimized_revision_test.go index 6f7ae57b5f..e8bb63dff5 100644 --- a/internal/datastore/revisions/optimized_test.go +++ b/internal/datastore/proxy/optimized_revision_test.go @@ -1,4 +1,4 @@ -package revisions +package proxy import ( "context" @@ -10,27 +10,36 @@ import ( "github.com/benbjohnson/clock" "github.com/ccoveille/go-safecast/v2" - "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "golang.org/x/sync/errgroup" + "github.com/authzed/spicedb/internal/datastore/proxy/proxy_test" + "github.com/authzed/spicedb/internal/datastore/revisions" "github.com/authzed/spicedb/pkg/datastore" "github.com/authzed/spicedb/pkg/genutil/slicez" ) -type trackingRevisionFunction struct { - mock.Mock +// fakeOptimizedRevisionDatastore is a minimal datastore whose OptimizedRevision +// is driven by a swappable function. Only OptimizedRevision is ever invoked by +// the proxy under test; all other Datastore methods are inherited from the nil +// embedded interface and will panic if called. +type fakeOptimizedRevisionDatastore struct { + datastore.Datastore + fn func(ctx context.Context) (datastore.Revision, time.Duration, string, error) } -func (m *trackingRevisionFunction) optimizedRevisionFunc(_ context.Context) (datastore.Revision, time.Duration, string, error) { - args := m.Called() - return args.Get(0).(datastore.Revision), args.Get(1).(time.Duration), args.String(2), args.Error(3) +func (f *fakeOptimizedRevisionDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { + return f.fn(ctx) +} + +func newOptimizedRevisionProxyForTest(d datastore.Datastore, maxStaleness time.Duration) *optimizedRevisionProxy { + return NewOptimizedRevisionProxy(d, maxStaleness).(*optimizedRevisionProxy) } var ( - one = NewForTransactionID(1) - two = NewForTransactionID(2) - three = NewForTransactionID(3) + one = revisions.NewForTransactionID(1) + two = revisions.NewForTransactionID(2) + three = revisions.NewForTransactionID(3) ) func cand(revs ...datastore.Revision) []datastore.Revision { @@ -109,14 +118,13 @@ func TestOptimizedRevisionCache(t *testing.T) { t.Run(tc.name, func(t *testing.T) { require := require.New(t) - or := NewCachedOptimizedRevisions(tc.maxStaleness) + mockDS := &proxy_test.MockDatastore{} + or := newOptimizedRevisionProxyForTest(mockDS, tc.maxStaleness) mockTime := clock.NewMock() - or.clockFn = mockTime - mock := trackingRevisionFunction{} - or.SetOptimizedRevisionFunc(mock.optimizedRevisionFunc) + or.clock = mockTime for _, callSpec := range tc.expectedCallResponses { - mock.On("optimizedRevisionFunc").Return(callSpec.rev, callSpec.validFor, "", nil).Once() + mockDS.On("OptimizedRevision").Return(callSpec.rev, callSpec.validFor, "", nil).Once() } ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) @@ -129,9 +137,8 @@ func TestOptimizedRevisionCache(t *testing.T) { } require.Eventually(func() bool { - revisionResult, err := or.OptimizedRevision(ctx) + revision, _, _, err := or.OptimizedRevision(ctx) require.NoError(err) - revision := revisionResult.Revision printableRevSet := slicez.Map(expectedRevSet, func(val datastore.Revision) string { return val.String() }) @@ -144,7 +151,7 @@ func TestOptimizedRevisionCache(t *testing.T) { mockTime.Add(5 * time.Millisecond) } - mock.AssertExpectations(t) + mockDS.AssertExpectations(t) }) } } @@ -152,12 +159,11 @@ func TestOptimizedRevisionCache(t *testing.T) { func TestOptimizedRevisionCacheSingleFlight(t *testing.T) { require := require.New(t) - or := NewCachedOptimizedRevisions(0) - mock := trackingRevisionFunction{} - or.SetOptimizedRevisionFunc(mock.optimizedRevisionFunc) + mockDS := &proxy_test.MockDatastore{} + or := newOptimizedRevisionProxyForTest(mockDS, 0) - mock. - On("optimizedRevisionFunc"). + mockDS. + On("OptimizedRevision"). Return(one, time.Duration(0), "", nil). After(50 * time.Millisecond). Once() @@ -168,11 +174,11 @@ func TestOptimizedRevisionCacheSingleFlight(t *testing.T) { g := errgroup.Group{} for range 10 { g.Go(func() error { - revisionResult, err := or.OptimizedRevision(ctx) + revision, _, _, err := or.OptimizedRevision(ctx) if err != nil { return err } - require.True(one.Equal(revisionResult.Revision), "must return the proper revision %s != %s", one, revisionResult.Revision) + require.True(one.Equal(revision), "must return the proper revision %s != %s", one, revision) return nil }) time.Sleep(1 * time.Millisecond) @@ -181,29 +187,30 @@ func TestOptimizedRevisionCacheSingleFlight(t *testing.T) { err := g.Wait() require.NoError(err) - mock.AssertExpectations(t) + mockDS.AssertExpectations(t) } func BenchmarkOptimizedRevisions(b *testing.B) { b.SetParallelism(1024) quantization := 1 * time.Millisecond - or := NewCachedOptimizedRevisions(quantization) - - or.SetOptimizedRevisionFunc(func(ctx context.Context) (datastore.Revision, time.Duration, string, error) { - nowNS := time.Now().UnixNano() - validForNS := nowNS % quantization.Nanoseconds() - roundedNS := nowNS - validForNS - // This should be non-negative. - uintRoundedNs := safecast.RequireConvert[uint64](b, roundedNS) - rev := NewForTransactionID(uintRoundedNs) - return rev, time.Duration(validForNS) * time.Nanosecond, "", nil - }) + fake := &fakeOptimizedRevisionDatastore{ + fn: func(_ context.Context) (datastore.Revision, time.Duration, string, error) { + nowNS := time.Now().UnixNano() + validForNS := nowNS % quantization.Nanoseconds() + roundedNS := nowNS - validForNS + // This should be non-negative. + uintRoundedNs := safecast.RequireConvert[uint64](b, roundedNS) + rev := revisions.NewForTransactionID(uintRoundedNs) + return rev, time.Duration(validForNS) * time.Nanosecond, "", nil + }, + } + or := newOptimizedRevisionProxyForTest(fake, quantization) ctx := b.Context() b.RunParallel(func(p *testing.PB) { for p.Next() { - if _, err := or.OptimizedRevision(ctx); err != nil { + if _, _, _, err := or.OptimizedRevision(ctx); err != nil { b.FailNow() } } @@ -213,23 +220,22 @@ func BenchmarkOptimizedRevisions(b *testing.B) { func TestSingleFlightError(t *testing.T) { req := require.New(t) - or := NewCachedOptimizedRevisions(0) - mock := trackingRevisionFunction{} - or.SetOptimizedRevisionFunc(mock.optimizedRevisionFunc) + mockDS := &proxy_test.MockDatastore{} + or := newOptimizedRevisionProxyForTest(mockDS, 0) // The shared attempt fails, and the direct retry (on the caller's context) // fails too, so the call returns an error. Both attempts invoke the function. - mock. - On("optimizedRevisionFunc"). + mockDS. + On("OptimizedRevision"). Return(one, time.Duration(0), "", errors.New("fail")). Twice() ctx, cancel := context.WithTimeout(t.Context(), 1*time.Second) defer cancel() - _, err := or.OptimizedRevision(ctx) + _, _, _, err := or.OptimizedRevision(ctx) req.Error(err) - mock.AssertExpectations(t) + mockDS.AssertExpectations(t) } // TestOptimizedRevisionRetriesAfterSharedFailure ensures that when the shared, @@ -238,23 +244,24 @@ func TestSingleFlightError(t *testing.T) { func TestOptimizedRevisionRetriesAfterSharedFailure(t *testing.T) { req := require.New(t) - or := NewCachedOptimizedRevisions(0) - var calls atomic.Int32 - or.SetOptimizedRevisionFunc(func(_ context.Context) (datastore.Revision, time.Duration, string, error) { - // Fail the first (shared) attempt; succeed on the direct retry. - if calls.Add(1) == 1 { - return datastore.NoRevision, 0, "", errors.New("transient failure") - } - return one, 0, "", nil - }) + fake := &fakeOptimizedRevisionDatastore{ + fn: func(_ context.Context) (datastore.Revision, time.Duration, string, error) { + // Fail the first (shared) attempt; succeed on the direct retry. + if calls.Add(1) == 1 { + return datastore.NoRevision, 0, "", errors.New("transient failure") + } + return one, 0, "", nil + }, + } + or := newOptimizedRevisionProxyForTest(fake, 0) ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) defer cancel() - res, err := or.OptimizedRevision(ctx) + res, _, _, err := or.OptimizedRevision(ctx) req.NoError(err) - req.True(one.Equal(res.Revision), "expected the direct retry to succeed") + req.True(one.Equal(res), "expected the direct retry to succeed") req.Equal(int32(2), calls.Load(), "expected one shared attempt and one direct retry") } @@ -267,23 +274,29 @@ func TestOptimizedRevisionTimeout(t *testing.T) { synctest.Test(t, func(t *testing.T) { req := require.New(t) - or := NewCachedOptimizedRevisions(0) - or.SetOptimizedRevisionSharedTimeout(10 * time.Millisecond) - or.SetOptimizedRevisionFallbackTimeout(50 * time.Millisecond) + fake := &fakeOptimizedRevisionDatastore{ + fn: func(ctx context.Context) (datastore.Revision, time.Duration, string, error) { + // Simulate a hung datastore call that only unblocks when its context is + // cancelled (as pgx does once a deadline is present on the context). + <-ctx.Done() + return datastore.NoRevision, 0, "", ctx.Err() + }, + } + or := newOptimizedRevisionProxyForTest(fake, 0) + or.sharedTimeout = 10 * time.Millisecond + or.fallbackTimeout = 50 * time.Millisecond var calls atomic.Int32 - or.SetOptimizedRevisionFunc(func(ctx context.Context) (datastore.Revision, time.Duration, string, error) { + baseFn := fake.fn + fake.fn = func(ctx context.Context) (datastore.Revision, time.Duration, string, error) { calls.Add(1) - // Simulate a hung datastore call that only unblocks when its context is - // cancelled (as pgx does once a deadline is present on the context). - <-ctx.Done() - return datastore.NoRevision, 0, "", ctx.Err() - }) + return baseFn(ctx) + } // The caller intentionally has no deadline of its own; the shared timeout and // the fallback timeout must together bound the call. If they fail to, every // goroutine in the bubble is durably blocked and synctest fails the test. - _, err := or.OptimizedRevision(t.Context()) + _, _, _, err := or.OptimizedRevision(t.Context()) req.Error(err, "hung revision call must return an error rather than block forever") // Both the shared attempt and the direct retry must have been attempted. @@ -291,12 +304,12 @@ func TestOptimizedRevisionTimeout(t *testing.T) { // The singleflight key must have been released so a subsequent call computes a // fresh result rather than re-attaching to the dead one. - or.SetOptimizedRevisionFunc(func(_ context.Context) (datastore.Revision, time.Duration, string, error) { + fake.fn = func(_ context.Context) (datastore.Revision, time.Duration, string, error) { return one, 0, "", nil - }) + } - res, err := or.OptimizedRevision(t.Context()) + res, _, _, err := or.OptimizedRevision(t.Context()) req.NoError(err) - req.True(one.Equal(res.Revision), "expected a fresh successful call after the hung call timed out") + req.True(one.Equal(res), "expected a fresh successful call after the hung call timed out") }) } diff --git a/internal/datastore/proxy/proxy_test/mock.go b/internal/datastore/proxy/proxy_test/mock.go index 36b1bd2178..5bc15d4620 100644 --- a/internal/datastore/proxy/proxy_test/mock.go +++ b/internal/datastore/proxy/proxy_test/mock.go @@ -2,6 +2,7 @@ package proxy_test import ( "context" + "time" "github.com/ccoveille/go-safecast/v2" "github.com/stretchr/testify/mock" @@ -52,9 +53,9 @@ func (dm *MockDatastore) ReadWriteTx( return args.Get(1).(datastore.Revision), args.Error(2) } -func (dm *MockDatastore) OptimizedRevision(_ context.Context) (datastore.RevisionWithSchemaHash, error) { +func (dm *MockDatastore) OptimizedRevision(_ context.Context) (datastore.Revision, time.Duration, string, error) { args := dm.Called() - return args.Get(0).(datastore.RevisionWithSchemaHash), args.Error(1) + return args.Get(0).(datastore.Revision), args.Get(1).(time.Duration), args.Get(2).(string), args.Error(3) } func (dm *MockDatastore) HeadRevision(_ context.Context) (datastore.RevisionWithSchemaHash, error) { diff --git a/internal/datastore/proxy/readonly_test.go b/internal/datastore/proxy/readonly_test.go index e4c6ce6dc8..f11dfa4c93 100644 --- a/internal/datastore/proxy/readonly_test.go +++ b/internal/datastore/proxy/readonly_test.go @@ -3,6 +3,7 @@ package proxy import ( "context" "testing" + "time" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -82,11 +83,11 @@ func TestOptimizedRevisionPassthrough(t *testing.T) { ds := NewReadonlyDatastore(delegate) ctx := t.Context() - delegate.On("OptimizedRevision").Return(datastore.RevisionWithSchemaHash{Revision: expectedRevision}, nil).Times(1) + delegate.On("OptimizedRevision").Return(expectedRevision, time.Duration(0), "", nil).Times(1) - result, err := ds.OptimizedRevision(ctx) + result, _, _, err := ds.OptimizedRevision(ctx) require.NoError(err) - require.Equal(expectedRevision, result.Revision) + require.Equal(expectedRevision, result) delegate.AssertExpectations(t) } diff --git a/internal/datastore/proxy/relationshipintegrity.go b/internal/datastore/proxy/relationshipintegrity.go index f7f4b64029..b8e28b2c07 100644 --- a/internal/datastore/proxy/relationshipintegrity.go +++ b/internal/datastore/proxy/relationshipintegrity.go @@ -202,7 +202,7 @@ func (r *relationshipIntegrityProxy) HeadRevision(ctx context.Context) (datastor return r.ds.HeadRevision(ctx) } -func (r *relationshipIntegrityProxy) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (r *relationshipIntegrityProxy) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { return r.ds.OptimizedRevision(ctx) } diff --git a/internal/datastore/proxy/relationshipintegrity_test.go b/internal/datastore/proxy/relationshipintegrity_test.go index f09527e25a..b0df184fb1 100644 --- a/internal/datastore/proxy/relationshipintegrity_test.go +++ b/internal/datastore/proxy/relationshipintegrity_test.go @@ -400,7 +400,7 @@ func TestRelationshipIntegrityProxyPassThroughs(t *testing.T) { require.NoError(t, pds.CheckRevision(ctx, headRev.Revision)) - optRev, err := pds.OptimizedRevision(ctx) + optRev, _, _, err := pds.OptimizedRevision(ctx) require.NoError(t, err) require.NotNil(t, optRev) diff --git a/internal/datastore/proxy/schemacaching/watchingcache_test.go b/internal/datastore/proxy/schemacaching/watchingcache_test.go index 770fe9944a..4f2fa5a0c6 100644 --- a/internal/datastore/proxy/schemacaching/watchingcache_test.go +++ b/internal/datastore/proxy/schemacaching/watchingcache_test.go @@ -1030,8 +1030,8 @@ func (*fakeDatastore) OfflineFeatures() (*datastore.Features, error) { return nil, fmt.Errorf("not implemented") } -func (*fakeDatastore) OptimizedRevision(context.Context) (datastore.RevisionWithSchemaHash, error) { - return datastore.RevisionWithSchemaHash{}, fmt.Errorf("not implemented") +func (*fakeDatastore) OptimizedRevision(context.Context) (datastore.Revision, time.Duration, string, error) { + return datastore.NoRevision, 0, "", fmt.Errorf("not implemented") } func (*fakeDatastore) ReadyState(context.Context) (datastore.ReadyState, error) { diff --git a/internal/datastore/proxy/singleflight.go b/internal/datastore/proxy/singleflight.go index a4a9c1261a..233287b797 100644 --- a/internal/datastore/proxy/singleflight.go +++ b/internal/datastore/proxy/singleflight.go @@ -2,6 +2,7 @@ package proxy import ( "context" + "time" "resenje.org/singleflight" @@ -17,11 +18,20 @@ func NewSingleflightDatastoreProxy(d datastore.Datastore) datastore.Datastore { type singleflightProxy struct { headRevGroup singleflight.Group[string, datastore.RevisionWithSchemaHash] + optRevGroup singleflight.Group[string, optimizedRevision] checkRevGroup singleflight.Group[string, string] statsGroup singleflight.Group[string, datastore.Stats] delegate datastore.Datastore } +// optimizedRevision carries the multi-valued result of OptimizedRevision through +// the singleflight group. +type optimizedRevision struct { + revision datastore.Revision + validFor time.Duration + schemaHash string +} + var _ datastore.Datastore = (*singleflightProxy)(nil) func (p *singleflightProxy) MetricsID() (string, error) { @@ -40,10 +50,15 @@ func (p *singleflightProxy) ReadWriteTx(ctx context.Context, f datastore.TxUserF return p.delegate.ReadWriteTx(ctx, f, opts...) } -func (p *singleflightProxy) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { - // NOTE: Optimized revisions are singleflighted by the underlying datastore via the - // CachedOptimizedRevisions struct. - return p.delegate.OptimizedRevision(ctx) +func (p *singleflightProxy) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { + ctx, span := tracer.Start(ctx, "singleflightProxy.OptimizedRevision") + defer span.End() + + res, _, err := p.optRevGroup.Do(ctx, "", func(ctx context.Context) (optimizedRevision, error) { + rev, validFor, schemaHash, err := p.delegate.OptimizedRevision(ctx) + return optimizedRevision{revision: rev, validFor: validFor, schemaHash: schemaHash}, err + }) + return res.revision, res.validFor, res.schemaHash, err } func (p *singleflightProxy) CheckRevision(ctx context.Context, revision datastore.Revision) error { diff --git a/internal/datastore/revisions/hlc.go b/internal/datastore/revisions/hlc.go new file mode 100644 index 0000000000..f5663b92c7 --- /dev/null +++ b/internal/datastore/revisions/hlc.go @@ -0,0 +1,48 @@ +package revisions + +import ( + "time" + + "github.com/authzed/spicedb/pkg/datastore" +) + +// QuantizeHLC rounds an HLC "now" revision down to a quantization boundary, +// after subtracting the follower-read delay, and reports how long the resulting +// quantized revision remains valid (i.e. until the next quantization boundary). +// +// This is the in-Go quantization used by datastores whose only revision +// primitive is a raw HLC clock (CockroachDB, Spanner). Datastores that quantize +// in SQL (Postgres, MySQL) compute validFor directly in their query and do not +// use this helper. +func QuantizeHLC(now WithTimestampRevision, followerReadDelay, quantization time.Duration) (datastore.Revision, time.Duration) { + delayedNow := now.TimestampNanoSec() - followerReadDelay.Nanoseconds() + quantized := delayedNow + validForNanos := int64(0) + if quantization.Nanoseconds() > 0 { + afterLastQuantization := delayedNow % quantization.Nanoseconds() + quantized -= afterLastQuantization + validForNanos = quantization.Nanoseconds() - afterLastQuantization + } + + return now.ConstructForTimestamp(quantized), time.Duration(validForNanos) * time.Nanosecond +} + +// CheckHLCGCWindow verifies that the given revision is within the datastore's +// software GC window relative to the current HLC time: not so old that it has +// (likely) been garbage collected, and not from the future. +func CheckHLCGCWindow(now, rev WithTimestampRevision, gcWindow time.Duration) error { + nowNanos := now.TimestampNanoSec() + revisionNanos := rev.TimestampNanoSec() + + isStale := revisionNanos < (nowNanos - gcWindow.Nanoseconds()) + if isStale { + return datastore.NewInvalidRevisionErr(rev, datastore.RevisionStale) + } + + isUnknown := revisionNanos > nowNanos + if isUnknown { + return datastore.NewInvalidRevisionErr(rev, datastore.CouldNotDetermineRevision) + } + + return nil +} diff --git a/internal/datastore/revisions/optimized.go b/internal/datastore/revisions/optimized.go deleted file mode 100644 index 0f1e2fdcc0..0000000000 --- a/internal/datastore/revisions/optimized.go +++ /dev/null @@ -1,210 +0,0 @@ -package revisions - -import ( - "context" - "fmt" - "math/rand" - "sync" - "time" - - "github.com/benbjohnson/clock" - "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/trace" - "resenje.org/singleflight" - - log "github.com/authzed/spicedb/internal/logging" - "github.com/authzed/spicedb/internal/telemetry/otelconv" - "github.com/authzed/spicedb/pkg/datastore" -) - -var tracer = otel.Tracer("spicedb/internal/datastore/common/revisions") - -// defaultOptimizedRevisionSharedTimeout aggressively bounds the *shared*, -// singleflighted computation of the optimized revision. -// -// The computation runs under singleflight, which replaces the caller's context -// with one that has *no* deadline (see resenje.org/singleflight: the work context -// is `context.WithoutCancel(ctx)`, cancelled only once every awaiting caller has -// gone away). The per-request gRPC deadline that would normally abort a stuck -// datastore query is therefore stripped. Worse, because every API request needs -// an optimized revision and they all de-duplicate onto this one in-flight call, -// every concurrent caller observes its latency directly: a wedged computation -// (e.g. a half-open connection silently dropped by a load balancer) inflates the -// P99 of the entire system, not just one request. -// -// Keeping this bound low limits that blast radius. The healthy computation takes -// single-digit milliseconds, so this is extremely generous for the happy path. On -// failure, OptimizedRevision retries the computation directly (bypassing -// singleflight) on the caller's own context, so the aggressive bound does not turn -// a transient slow/wedged shared attempt into a failed request. -const defaultOptimizedRevisionSharedTimeout = 2 * time.Second - -// defaultOptimizedRevisionFallbackTimeout bounds the *direct* retry performed -// after a failed shared attempt. The retry runs on the caller's own context, so -// its deadline applies when present; this fallback only takes effect for callers -// with no deadline of their own, preserving the guarantee that a revision -// computation can never block forever. -const defaultOptimizedRevisionFallbackTimeout = 10 * time.Second - -// OptimizedRevisionFunction instructs the datastore to compute its own current -// optimized revision given the specific quantization, and return for how long -// it will remain valid, along with the schema hash at that revision (or "" if -// the datastore does not provide one on this code path). -type OptimizedRevisionFunction func(context.Context) (rev datastore.Revision, validFor time.Duration, schemaHash string, err error) - -// NewCachedOptimizedRevisions returns a CachedOptimizedRevisions for the given configuration -func NewCachedOptimizedRevisions(maxRevisionStaleness time.Duration) *CachedOptimizedRevisions { - return &CachedOptimizedRevisions{ - maxRevisionStaleness: maxRevisionStaleness, - optimizedRevisionSharedTimeout: defaultOptimizedRevisionSharedTimeout, - optimizedRevisionFallbackTimeout: defaultOptimizedRevisionFallbackTimeout, - clockFn: clock.New(), - } -} - -// SetOptimizedRevisionFunc must be called after construction, and is the method -// by which one specializes this helper for a specific datastore. -func (cor *CachedOptimizedRevisions) SetOptimizedRevisionFunc(revisionFunc OptimizedRevisionFunction) { - cor.optimizedFunc = revisionFunc -} - -// SetOptimizedRevisionSharedTimeout overrides the maximum duration the shared, -// singleflighted call to the optimized revision function is allowed to run before -// it is cancelled. See defaultOptimizedRevisionSharedTimeout for why this bound is -// required. -func (cor *CachedOptimizedRevisions) SetOptimizedRevisionSharedTimeout(timeout time.Duration) { - cor.optimizedRevisionSharedTimeout = timeout -} - -// SetOptimizedRevisionFallbackTimeout overrides the bound applied to the direct -// retry performed after a failed shared attempt for callers that have no deadline -// of their own. See defaultOptimizedRevisionFallbackTimeout. -func (cor *CachedOptimizedRevisions) SetOptimizedRevisionFallbackTimeout(timeout time.Duration) { - cor.optimizedRevisionFallbackTimeout = timeout -} - -func (cor *CachedOptimizedRevisions) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { - ctx, span := tracer.Start(ctx, "CachedOptimizedRevisions.OptimizedRevision") - defer span.End() - - localNow := cor.clockFn.Now() - - // Subtract a random amount of time from now, to let barely expired candidates get selected - adjustedNow := localNow - if cor.maxRevisionStaleness > 0 { - // nolint:gosec - // G404 use of non cryptographically secure random number generator is not a security concern here, - // as we are using it to introduce randomness to the accepted staleness of a revision and reduce the odds of - // a thundering herd to the datastore - adjustedNow = localNow.Add(-1 * time.Duration(rand.Int63n(cor.maxRevisionStaleness.Nanoseconds())) * time.Nanosecond) - } - - cor.RLock() - for _, candidate := range cor.candidates { - if candidate.validThrough.After(adjustedNow) { - cor.RUnlock() - log.Ctx(ctx).Debug().Time("now", localNow).Time("valid", candidate.validThrough).Msg("returning cached revision") - span.AddEvent(otelconv.EventDatastoreRevisionsCacheReturned) - return datastore.RevisionWithSchemaHash{Revision: candidate.revision, SchemaHash: candidate.schemaHash}, nil - } - } - cor.RUnlock() - - // Compute the revision under singleflight so concurrent callers share a single - // datastore round-trip. The shared call is aggressively bounded (see - // defaultOptimizedRevisionSharedTimeout) so a wedged computation cannot pin the - // latency of every waiting caller. - result, _, err := cor.updateGroup.Do(ctx, "", func(sfCtx context.Context) (datastore.RevisionWithSchemaHash, error) { - // NOTE: singleflight hands this function a context with the caller's - // deadline stripped. Re-impose a (low) deadline so a hung datastore call - // cannot block this in-flight call, and therefore every caller waiting on - // it, beyond the bound. - if cor.optimizedRevisionSharedTimeout > 0 { - var cancel context.CancelFunc - sfCtx, cancel = context.WithTimeout(sfCtx, cor.optimizedRevisionSharedTimeout) - defer cancel() - } - - return cor.computeOptimizedRevision(sfCtx, localNow, span) - }) - if err == nil { - return result, nil - } - - // If this caller's own context is already done, surface its cancellation / - // deadline error directly; there is nothing to retry. - if ctx.Err() != nil { - return datastore.RevisionWithSchemaHash{}, ctx.Err() - } - - // The shared, aggressively-bounded attempt failed (e.g. a wedged connection - // tripped the timeout, or the datastore returned a transient error). Give this - // request one more chance, directly and outside of singleflight, on its own - // context: this is not capped by the low shared bound and cannot re-attach to a - // poisoned in-flight call, so a brief wedge does not fail the request. For the - // pooled SQL datastores the retry is naturally throttled by the pool's maximum - // size; Spanner's client similarly bounds concurrent calls via its own session - // pool. Even when many waiters retry at once, the datastore is not overwhelmed. - span.AddEvent(otelconv.EventDatastoreRevisionsSharedFailedRetrying) - log.Ctx(ctx).Warn().Err(err).Msg("shared optimized revision computation failed; retrying directly") - - retryCtx, cancel := context.WithTimeout(ctx, cor.optimizedRevisionFallbackTimeout) - defer cancel() - return cor.computeOptimizedRevision(retryCtx, cor.clockFn.Now(), span) -} - -// computeOptimizedRevision invokes the datastore's optimized revision function and -// records the result as a cache candidate. It is invoked both from the shared -// singleflight path and from the direct retry that follows a failed shared attempt. -func (cor *CachedOptimizedRevisions) computeOptimizedRevision(ctx context.Context, localNow time.Time, span trace.Span) (datastore.RevisionWithSchemaHash, error) { - log.Ctx(ctx).Debug().Time("now", localNow).Msg("computing new revision") - - optimized, validFor, schemaHash, err := cor.optimizedFunc(ctx) - if err != nil { - return datastore.RevisionWithSchemaHash{}, fmt.Errorf("unable to compute optimized revision: %w", err) - } - - rvt := localNow.Add(validFor) - - // Prune the candidates that have definitely expired - cor.Lock() - var numToDrop uint - for _, candidate := range cor.candidates { - if candidate.validThrough.Add(cor.maxRevisionStaleness).Before(localNow) { - numToDrop++ - } else { - break - } - } - - cor.candidates = cor.candidates[numToDrop:] - cor.candidates = append(cor.candidates, validRevision{optimized, rvt, schemaHash}) - cor.Unlock() - - span.AddEvent(otelconv.EventDatastoreRevisionsComputed) - log.Ctx(ctx).Debug().Time("now", localNow).Time("valid", rvt).Stringer("validFor", validFor).Msg("setting valid through") - return datastore.RevisionWithSchemaHash{Revision: optimized, SchemaHash: schemaHash}, nil -} - -// CachedOptimizedRevisions does caching and deduplication for requests for optimized revisions. -type CachedOptimizedRevisions struct { - sync.RWMutex - - maxRevisionStaleness time.Duration - optimizedRevisionSharedTimeout time.Duration - optimizedRevisionFallbackTimeout time.Duration - optimizedFunc OptimizedRevisionFunction - clockFn clock.Clock - - // these values are read and set by multiple consumers - candidates []validRevision // GUARDED_BY(RWMutex) - - // the updategroup consolidates concurrent requests to the database into 1 - updateGroup singleflight.Group[string, datastore.RevisionWithSchemaHash] -} - -type validRevision struct { - revision datastore.Revision - validThrough time.Time - schemaHash string -} diff --git a/internal/datastore/revisions/remoteclock.go b/internal/datastore/revisions/remoteclock.go deleted file mode 100644 index b2f40f7db7..0000000000 --- a/internal/datastore/revisions/remoteclock.go +++ /dev/null @@ -1,148 +0,0 @@ -package revisions - -import ( - "context" - "time" - - log "github.com/authzed/spicedb/internal/logging" - "github.com/authzed/spicedb/pkg/datastore" - "github.com/authzed/spicedb/pkg/spiceerrors" -) - -// RemoteNowFunction queries the datastore to get a current revision and the -// schema hash visible at that revision. Implementations that do not have a -// schema hash on this code path may return "". -type RemoteNowFunction func(context.Context) (datastore.Revision, string, error) - -// RemoteNowOnlyFunction queries the datastore to get a current revision only, -// without reading the schema hash. Used by CheckRevision where the hash is -// not needed; allows backends to skip a schema_revision subquery. -type RemoteNowOnlyFunction func(context.Context) (datastore.Revision, error) - -// RemoteClockRevisions handles revision calculation for datastores that provide -// their own clocks. -type RemoteClockRevisions struct { - *CachedOptimizedRevisions - - gcWindowNanos int64 - nowFunc RemoteNowFunction - nowOnlyFunc RemoteNowOnlyFunction - followerReadDelayNanos int64 - quantizationNanos int64 -} - -// NewRemoteClockRevisions returns a RemoteClockRevisions for the given configuration -func NewRemoteClockRevisions(gcWindow, maxRevisionStaleness, followerReadDelay, quantization time.Duration) *RemoteClockRevisions { - // Ensure the max revision staleness never exceeds the GC window. - if maxRevisionStaleness > gcWindow { - log.Warn(). - Dur("maxRevisionStaleness", maxRevisionStaleness). - Dur("gcWindow", gcWindow). - Msg("the configured maximum revision staleness exceeds the configured gc window, so capping to gcWindow") - maxRevisionStaleness = gcWindow - 1 - } - - revisions := &RemoteClockRevisions{ - CachedOptimizedRevisions: NewCachedOptimizedRevisions( - maxRevisionStaleness, - ), - gcWindowNanos: gcWindow.Nanoseconds(), - followerReadDelayNanos: followerReadDelay.Nanoseconds(), - quantizationNanos: quantization.Nanoseconds(), - } - - revisions.SetOptimizedRevisionFunc(revisions.optimizedRevisionFunc) - - return revisions -} - -func (rcr *RemoteClockRevisions) optimizedRevisionFunc(ctx context.Context) (datastore.Revision, time.Duration, string, error) { - nowRev, schemaHash, err := rcr.nowFunc(ctx) - if err != nil { - return datastore.NoRevision, 0, "", err - } - - if nowRev == datastore.NoRevision { - return datastore.NoRevision, 0, "", datastore.NewInvalidRevisionErr(nowRev, datastore.CouldNotDetermineRevision) - } - - nowTS, ok := nowRev.(WithTimestampRevision) - if !ok { - return datastore.NoRevision, 0, "", spiceerrors.MustBugf("expected with-timestamp revision, got %T", nowRev) - } - - delayedNow := nowTS.TimestampNanoSec() - rcr.followerReadDelayNanos - quantized := delayedNow - validForNanos := int64(0) - if rcr.quantizationNanos > 0 { - afterLastQuantization := delayedNow % rcr.quantizationNanos - quantized -= afterLastQuantization - validForNanos = rcr.quantizationNanos - afterLastQuantization - } - log.Ctx(ctx).Debug(). - Time("quantized", time.Unix(0, quantized)). - Int64("readSkew", rcr.followerReadDelayNanos). - Int64("totalSkew", nowTS.TimestampNanoSec()-quantized). - Msg("revision skews") - - return nowTS.ConstructForTimestamp(quantized), time.Duration(validForNanos) * time.Nanosecond, schemaHash, nil -} - -// SetNowFunc sets the function used to determine the head revision -func (rcr *RemoteClockRevisions) SetNowFunc(nowFunc RemoteNowFunction) { - rcr.nowFunc = nowFunc -} - -// SetNowOnlyFunc sets a timestamp-only revision function used by CheckRevision. -// Datastores that implement this can elide a schema-hash subquery on -// revision-validation paths. If unset, CheckRevision falls back to nowFunc and -// discards the hash. -func (rcr *RemoteClockRevisions) SetNowOnlyFunc(nowOnlyFunc RemoteNowOnlyFunction) { - rcr.nowOnlyFunc = nowOnlyFunc -} - -func (rcr *RemoteClockRevisions) CheckRevision(ctx context.Context, dsRevision datastore.Revision) error { - if dsRevision == datastore.NoRevision { - return datastore.NewInvalidRevisionErr(dsRevision, datastore.CouldNotDetermineRevision) - } - - revision := dsRevision.(WithTimestampRevision) - - ctx, span := tracer.Start(ctx, "CheckRevision") - defer span.End() - - // Make sure the system time indicated is within the software GC window. - // Use nowOnlyFunc when available to avoid reading the schema hash unnecessarily. - var now datastore.Revision - var err error - if rcr.nowOnlyFunc != nil { - now, err = rcr.nowOnlyFunc(ctx) - } else { - now, _, err = rcr.nowFunc(ctx) - } - if err != nil { - return err - } - - nowTS, ok := now.(WithTimestampRevision) - if !ok { - return spiceerrors.MustBugf("expected HLC revision, got %T", now) - } - - nowNanos := nowTS.TimestampNanoSec() - revisionNanos := revision.TimestampNanoSec() - - isStale := revisionNanos < (nowNanos - rcr.gcWindowNanos) - if isStale { - log.Ctx(ctx).Debug().Stringer("now", now).Stringer("revision", revision).Msg("stale revision") - return datastore.NewInvalidRevisionErr(revision, datastore.RevisionStale) - } - - isUnknown := revisionNanos > nowNanos - if isUnknown { - log.Ctx(ctx).Debug().Stringer("now", now).Stringer("revision", revision).Msg("unknown revision") - return datastore.NewInvalidRevisionErr(revision, datastore.CouldNotDetermineRevision) - } - - return nil -} diff --git a/internal/datastore/revisions/remoteclock_test.go b/internal/datastore/revisions/remoteclock_test.go deleted file mode 100644 index e900e02718..0000000000 --- a/internal/datastore/revisions/remoteclock_test.go +++ /dev/null @@ -1,217 +0,0 @@ -package revisions - -import ( - "context" - "testing" - "time" - - "github.com/benbjohnson/clock" - "github.com/stretchr/testify/require" - - log "github.com/authzed/spicedb/internal/logging" - "github.com/authzed/spicedb/pkg/datastore" -) - -func TestRemoteClockOptimizedRevisions(t *testing.T) { - type timeAndExpectedRevision struct { - unixTime int64 - expected int64 - } - - testCases := []struct { - name string - followerReadDelay time.Duration - quantization time.Duration - times []timeAndExpectedRevision - }{ - { - "direct", 0, 0, - []timeAndExpectedRevision{ - {1230, 1230}, - {1231, 1231}, - {1232, 1232}, - {1233, 1233}, - {1234, 1234}, - {1235, 1235}, - }, - }, - { - "simple quantized", 0, 5 * time.Second, - []timeAndExpectedRevision{ - {1230, 1230}, - {1231, 1230}, - {1232, 1230}, - {1233, 1230}, - {1234, 1230}, - {1235, 1235}, - }, - }, - { - "simple with skew", 5 * time.Second, 5 * time.Second, - []timeAndExpectedRevision{ - {1230, 1225}, - {1231, 1225}, - {1232, 1225}, - {1233, 1225}, - {1234, 1225}, - {1235, 1230}, - }, - }, - { - "skew no quantization", 5 * time.Second, 0, - []timeAndExpectedRevision{ - {1230, 1225}, - {1231, 1226}, - {1232, 1227}, - {1233, 1228}, - {1234, 1229}, - {1235, 1230}, - }, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - require := require.New(t) - - rcr := NewRemoteClockRevisions(1*time.Hour, 0, tc.followerReadDelay, tc.quantization) - - remoteClock := clock.NewMock() - rcr.clockFn = remoteClock - rcr.SetNowFunc(func(ctx context.Context) (datastore.Revision, string, error) { - log.Debug().Stringer("now", remoteClock.Now()).Msg("current remote time") - return NewForTime(remoteClock.Now()), "", nil - }) - - for _, timeAndExpected := range tc.times { - remoteClock.Set(time.Unix(timeAndExpected.unixTime, 0)) - - expected := NewForTimestamp(timeAndExpected.expected * 1_000_000_000) - optimizedResult, err := rcr.OptimizedRevision(t.Context()) - require.NoError(err) - optimized := optimizedResult.Revision - require.True( - expected.Equal(optimized), - "optimized revision does not match expected: %s != %s", - expected, - optimized, - ) - } - }) - } -} - -func TestRemoteClockCheckRevisions(t *testing.T) { - testCases := []struct { - name string - gcWindow time.Duration - currentTime int64 - testRevisionSeconds int64 - expectError bool - }{ - {"now is valid", 1 * time.Hour, 12345, 12345, false}, - {"near future", 1 * time.Hour, 12345, 12346, true}, - {"far future", 1 * time.Hour, 12345, 1650599916, true}, - {"recent past", 1 * time.Hour, 12345, 12344, false}, - {"expired", 1 * time.Second, 12345, 12343, true}, - {"very old", 1 * time.Hour, 12345, 8744, true}, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - require := require.New(t) - - rcr := NewRemoteClockRevisions(tc.gcWindow, 0, 0, 0) - - remoteClock := clock.NewMock() - rcr.clockFn = remoteClock - rcr.SetNowFunc(func(ctx context.Context) (datastore.Revision, string, error) { - log.Debug().Stringer("now", remoteClock.Now()).Msg("current remote time") - return NewForTime(remoteClock.Now()), "", nil - }) - - remoteClock.Set(time.Unix(tc.currentTime, 0)) - - testRevision := NewForTimestamp(tc.testRevisionSeconds * 1_000_000_000) - err := rcr.CheckRevision(t.Context(), testRevision) - if tc.expectError { - require.Error(err) - } else { - require.NoError(err) - } - }) - } -} - -func TestCheckRevisionPrefersNowOnlyFunc(t *testing.T) { - require := require.New(t) - rcr := NewRemoteClockRevisions(time.Hour, time.Minute, 0, 0) - - nowFuncCalls := 0 - nowOnlyCalls := 0 - rcr.SetNowFunc(func(ctx context.Context) (datastore.Revision, string, error) { - nowFuncCalls++ - return NewForTime(time.Now()), "hash", nil - }) - rcr.SetNowOnlyFunc(func(ctx context.Context) (datastore.Revision, error) { - nowOnlyCalls++ - return NewForTime(time.Now()), nil - }) - - require.NoError(rcr.CheckRevision(t.Context(), NewForTime(time.Now()))) - require.Equal(0, nowFuncCalls, "CheckRevision must not call nowFunc when nowOnlyFunc is set") - require.Equal(1, nowOnlyCalls) -} - -func TestCheckRevisionFallsBackToNowFuncWhenNowOnlyUnset(t *testing.T) { - require := require.New(t) - rcr := NewRemoteClockRevisions(time.Hour, time.Minute, 0, 0) - - nowFuncCalls := 0 - rcr.SetNowFunc(func(ctx context.Context) (datastore.Revision, string, error) { - nowFuncCalls++ - return NewForTime(time.Now()), "hash", nil - }) - - require.NoError(rcr.CheckRevision(t.Context(), NewForTime(time.Now()))) - require.Equal(1, nowFuncCalls) -} - -func TestRemoteClockStalenessBeyondGC(t *testing.T) { - // Set a GC window of 1 hour. - gcWindow := 1 * time.Hour - - // Set a max revision staleness of 100 hours, well in excess of the GC window. - maxRevisionStaleness := 100 * time.Hour - - rcr := NewRemoteClockRevisions(gcWindow, maxRevisionStaleness, 0, 0) - - remoteClock := clock.NewMock() - rcr.clockFn = remoteClock - rcr.SetNowFunc(func(ctx context.Context) (datastore.Revision, string, error) { - log.Debug().Stringer("now", remoteClock.Now()).Msg("current remote time") - return NewForTime(remoteClock.Now()), "", nil - }) - - // Set the current time to 1. - currentTime := int64(1) - remoteClock.Set(time.Unix(currentTime, 0)) - - // Call optimized revision. - optimizedResult, err := rcr.OptimizedRevision(t.Context()) - require.NoError(t, err) - - // Ensure the optimized revision is not past the GC window. - err = rcr.CheckRevision(t.Context(), optimizedResult.Revision) - require.NoError(t, err) - - // Set the current time to 100001 to ensure the optimized revision is past the GC window. - remoteClock.Set(time.Unix(100001, 0)) - - newOptimizedResult, err := rcr.OptimizedRevision(t.Context()) - require.NoError(t, err) - - // Ensure the new optimized revision is not past the GC window. - err = rcr.CheckRevision(t.Context(), newOptimizedResult.Revision) - require.NoError(t, err) -} diff --git a/internal/datastore/spanner/spanner.go b/internal/datastore/spanner/spanner.go index ca2594f513..b530864fcd 100644 --- a/internal/datastore/spanner/spanner.go +++ b/internal/datastore/spanner/spanner.go @@ -33,6 +33,7 @@ import ( "github.com/authzed/spicedb/internal/telemetry/otelconv" "github.com/authzed/spicedb/pkg/datastore" "github.com/authzed/spicedb/pkg/datastore/options" + "github.com/authzed/spicedb/pkg/spiceerrors" "github.com/authzed/spicedb/pkg/tuple" ) @@ -81,10 +82,12 @@ var ( ) type spannerDatastore struct { - *revisions.RemoteClockRevisions revisions.CommonDecoder *common.MigrationValidator + followerReadDelay time.Duration + revisionQuantization time.Duration + gcWindow time.Duration watchBufferLength uint16 watchChangeBufferMaximumSize uint64 watchBufferWriteTimeout time.Duration @@ -170,9 +173,6 @@ func NewSpannerDatastore(ctx context.Context, database string, opts ...Option) ( return nil, common.RedactAndLogSensitiveConnString(ctx, errUnableToInstantiate, err, database) } - maxRevisionStaleness := time.Duration(float64(config.revisionQuantization.Nanoseconds())* - config.maxRevisionStalenessPercent) * time.Nanosecond - headMigration, err := migrations.SpannerMigrations.HeadRevision() if err != nil { return nil, fmt.Errorf("invalid head migration found for spanner: %w", err) @@ -208,16 +208,13 @@ func NewSpannerDatastore(ctx context.Context, database string, opts ...Option) ( ) ds := &spannerDatastore{ - RemoteClockRevisions: revisions.NewRemoteClockRevisions( - defaultChangeStreamRetention, - maxRevisionStaleness, - config.followerReadDelay, - config.revisionQuantization, - ), CommonDecoder: revisions.CommonDecoder{ Kind: revisions.Timestamp, }, MigrationValidator: common.NewMigrationValidator(headMigration, config.allowedMigrations), + followerReadDelay: config.followerReadDelay, + revisionQuantization: config.revisionQuantization, + gcWindow: defaultChangeStreamRetention, client: client, config: config, database: database, @@ -231,16 +228,69 @@ func NewSpannerDatastore(ctx context.Context, database string, opts ...Option) ( schema: *schema, meterProvider: meterProvider, } - // Optimized revision and revision checking use a stale read for the - // current timestamp. - // TODO: Still investigating whether a stale read can be used for - // HeadRevision for FullConsistency queries. - ds.SetNowFunc(ds.staleHeadRevision) - ds.SetNowOnlyFunc(ds.nowOnly) - return ds, nil } +// OptimizedRevision computes the uncached quantized revision for Spanner. It +// reads the current timestamp via a stale read and quantizes it in Go. Caching, +// deduplication, and jitter are layered on by proxy.NewOptimizedRevisionProxy. +// +// Optimized revision and revision checking use a stale read for the current +// timestamp. +// TODO: Still investigating whether a stale read can be used for HeadRevision +// +// for FullConsistency queries. +func (sd *spannerDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { + ctx, span := tracer.Start(ctx, "OptimizedRevision") + defer span.End() + + nowRev, schemaHash, err := sd.staleHeadRevision(ctx) + if err != nil { + return datastore.NoRevision, 0, "", err + } + + if nowRev == datastore.NoRevision { + return datastore.NoRevision, 0, "", datastore.NewInvalidRevisionErr(nowRev, datastore.CouldNotDetermineRevision) + } + + nowTS, ok := nowRev.(revisions.WithTimestampRevision) + if !ok { + return datastore.NoRevision, 0, "", spiceerrors.MustBugf("expected with-timestamp revision, got %T", nowRev) + } + + rev, validFor := revisions.QuantizeHLC(nowTS, sd.followerReadDelay, sd.revisionQuantization) + return rev, validFor, schemaHash, nil +} + +// CheckRevision verifies the given revision is within the software GC window. It +// lives on the datastore (not the caching layer) because it is not a caching +// concern, and uses the hash-free stale read since the schema hash is not needed. +func (sd *spannerDatastore) CheckRevision(ctx context.Context, dsRevision datastore.Revision) error { + if dsRevision == datastore.NoRevision { + return datastore.NewInvalidRevisionErr(dsRevision, datastore.CouldNotDetermineRevision) + } + + revision, ok := dsRevision.(revisions.WithTimestampRevision) + if !ok { + return spiceerrors.MustBugf("expected HLC revision, got %T", dsRevision) + } + + ctx, span := tracer.Start(ctx, "CheckRevision") + defer span.End() + + now, err := sd.nowOnly(ctx) + if err != nil { + return err + } + + nowTS, ok := now.(revisions.WithTimestampRevision) + if !ok { + return spiceerrors.MustBugf("expected HLC revision, got %T", now) + } + + return revisions.CheckHLCGCWindow(nowTS, revision, sd.gcWindow) +} + func getMeterProviderWithPromExporter(res *otelres.Resource) (*metric.MeterProvider, error) { exporter, err := otelprom.New() if err != nil { diff --git a/internal/datastore/spanner/spanner_test.go b/internal/datastore/spanner/spanner_test.go index f52657d61f..52633ed227 100644 --- a/internal/datastore/spanner/spanner_test.go +++ b/internal/datastore/spanner/spanner_test.go @@ -161,8 +161,8 @@ func FakeStatsTest(t *testing.T, ds datastore.Datastore) { // This reproduces the case hit by mage testcons:spanner, where the default // 4.8s FollowerReadDelay lands before the freshly-run migration. func OptimizedRevisionAfterFreshMigrationTest(t *testing.T, ds datastore.Datastore) { - result, err := ds.OptimizedRevision(t.Context()) + rev, _, schemaHash, err := ds.OptimizedRevision(t.Context()) require.NoError(t, err) - require.NotEqual(t, datastore.NoRevision, result.Revision) - require.Empty(t, result.SchemaHash, "schema hash should be empty when no schema has been written yet") + require.NotEqual(t, datastore.NoRevision, rev) + require.Empty(t, schemaHash, "schema hash should be empty when no schema has been written yet") } diff --git a/internal/services/v1/permissions_test.go b/internal/services/v1/permissions_test.go index e0ab0c7579..e2c46fccbc 100644 --- a/internal/services/v1/permissions_test.go +++ b/internal/services/v1/permissions_test.go @@ -525,7 +525,7 @@ func (c *readStoredSchemaCounter) ReadWriteTx(ctx context.Context, fn datastore. return c.ds.ReadWriteTx(ctx, fn, opts...) } -func (c *readStoredSchemaCounter) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (c *readStoredSchemaCounter) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { return c.ds.OptimizedRevision(ctx) } diff --git a/pkg/cmd/datastore/datastore.go b/pkg/cmd/datastore/datastore.go index 6ba677efc9..f2ed932e6a 100644 --- a/pkg/cmd/datastore/datastore.go +++ b/pkg/cmd/datastore/datastore.go @@ -488,6 +488,21 @@ func NewDatastore(ctx context.Context, options ...ConfigOption) (datastore.Datas return nil, err } + // Layer the optimized-revision cache onto the datastore. Concrete datastores + // return the uncached optimized revision (one round-trip per call); caching, + // singleflighting, and jitter are provided here by the proxy stack. + maxRevisionStaleness := time.Duration(float64(opts.RevisionQuantization.Nanoseconds()) * + opts.MaxRevisionStalenessPercent) + // Ensure the max revision staleness never exceeds the GC window. + if opts.GCWindow > 0 && maxRevisionStaleness > opts.GCWindow { + log.Ctx(ctx).Warn(). + Dur("maxRevisionStaleness", maxRevisionStaleness). + Dur("gcWindow", opts.GCWindow). + Msg("the configured maximum revision staleness exceeds the configured gc window, so capping to gcWindow") + maxRevisionStaleness = opts.GCWindow - 1 + } + ds = proxy.NewOptimizedRevisionProxy(ds, maxRevisionStaleness) + if len(opts.BootstrapFiles) > 0 || len(opts.BootstrapFileContents) > 0 { ctx, cancel := context.WithTimeout(ctx, opts.BootstrapTimeout) defer cancel() diff --git a/pkg/datalayer/datalayer_test.go b/pkg/datalayer/datalayer_test.go index c86edcc619..c4d28ef48e 100644 --- a/pkg/datalayer/datalayer_test.go +++ b/pkg/datalayer/datalayer_test.go @@ -2114,8 +2114,8 @@ func (d *erroringRevisionDS) HeadRevision(_ context.Context) (datastore.Revision return datastore.RevisionWithSchemaHash{}, d.headErr } -func (d *erroringRevisionDS) OptimizedRevision(_ context.Context) (datastore.RevisionWithSchemaHash, error) { - return datastore.RevisionWithSchemaHash{}, d.optimizedErr +func (d *erroringRevisionDS) OptimizedRevision(_ context.Context) (datastore.Revision, time.Duration, string, error) { + return datastore.NoRevision, 0, "", d.optimizedErr } // schemaHashInjectingDS wraps a Datastore and injects a fixed SchemaHash into @@ -2126,12 +2126,12 @@ type schemaHashInjectingDS struct { injectHash string } -func (d *schemaHashInjectingDS) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { - r, err := d.Datastore.OptimizedRevision(ctx) - if err == nil { - r.SchemaHash = d.injectHash +func (d *schemaHashInjectingDS) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { + rev, validFor, _, err := d.Datastore.OptimizedRevision(ctx) + if err != nil { + return datastore.NoRevision, 0, "", err } - return r, err + return rev, validFor, d.injectHash, nil } // TestOptimizedRevisionSeedsLastSchemaHash verifies that OptimizedRevision in a diff --git a/pkg/datalayer/impl.go b/pkg/datalayer/impl.go index bfac344a77..6d26d7bf2e 100644 --- a/pkg/datalayer/impl.go +++ b/pkg/datalayer/impl.go @@ -180,18 +180,18 @@ func (d *defaultDataLayer) ReadWriteTx(ctx context.Context, fn TxUserFunc, opts } func (d *defaultDataLayer) OptimizedRevision(ctx context.Context) (datastore.Revision, SchemaHash, error) { - result, err := d.ds.OptimizedRevision(ctx) + rev, _, schemaHash, err := d.ds.OptimizedRevision(ctx) if err != nil { return datastore.NoRevision, NoSchemaHashInLegacyMode, err } - if d.schemaMode.ReadsFromNew() && result.SchemaHash != "" { - hash := SchemaHash(result.SchemaHash) + if d.schemaMode.ReadsFromNew() && schemaHash != "" { + hash := SchemaHash(schemaHash) d.observeSchemaHash(hash) - return result.Revision, hash, nil + return rev, hash, nil } - return result.Revision, NoSchemaHashInLegacyMode, nil + return rev, NoSchemaHashInLegacyMode, nil } func (d *defaultDataLayer) HeadRevision(ctx context.Context) (datastore.Revision, SchemaHash, error) { @@ -438,11 +438,11 @@ func (r *readOnlyDatastoreAdapter) ReadWriteTx(_ context.Context, _ TxUserFunc, } func (r *readOnlyDatastoreAdapter) OptimizedRevision(ctx context.Context) (datastore.Revision, SchemaHash, error) { - result, err := r.ds.OptimizedRevision(ctx) + rev, _, _, err := r.ds.OptimizedRevision(ctx) if err != nil { return datastore.NoRevision, NoSchemaHashInLegacyMode, err } - return result.Revision, NoSchemaHashInLegacyMode, nil + return rev, NoSchemaHashInLegacyMode, nil } func (r *readOnlyDatastoreAdapter) HeadRevision(ctx context.Context) (datastore.Revision, SchemaHash, error) { diff --git a/pkg/datastore/context.go b/pkg/datastore/context.go index 434fdea646..024147c45f 100644 --- a/pkg/datastore/context.go +++ b/pkg/datastore/context.go @@ -2,6 +2,7 @@ package datastore import ( "context" + "time" "github.com/authzed/spicedb/pkg/datastore/options" core "github.com/authzed/spicedb/pkg/proto/core/v1" @@ -48,7 +49,7 @@ func (p *ctxProxy) IsStrictReadModeEnabled() bool { return false } -func (p *ctxProxy) OptimizedRevision(ctx context.Context) (RevisionWithSchemaHash, error) { +func (p *ctxProxy) OptimizedRevision(ctx context.Context) (Revision, time.Duration, string, error) { return p.delegate.OptimizedRevision(context.WithoutCancel(ctx)) } diff --git a/pkg/datastore/datastore.go b/pkg/datastore/datastore.go index 5e6096e4d7..bfaa2264e8 100644 --- a/pkg/datastore/datastore.go +++ b/pkg/datastore/datastore.go @@ -708,7 +708,13 @@ type ReadOnlyDatastore interface { // OptimizedRevision gets a revision that will likely already be replicated // and will likely be shared amongst many queries. - OptimizedRevision(ctx context.Context) (RevisionWithSchemaHash, error) + // + // Implementations return the *uncached* answer: one datastore round-trip per + // call, yielding the revision, how long it remains valid, and the schema hash + // visible at that revision (or "" if not provided on this path). Caching, + // deduplication, and jitter are layered on by proxy.NewOptimizedRevisionProxy. + // Callers that do not need validFor may discard it with _. + OptimizedRevision(ctx context.Context) (rev Revision, validFor time.Duration, schemaHash string, err error) // HeadRevision gets a revision that is guaranteed to be at least as fresh as // right now. diff --git a/pkg/datastore/datastore_test.go b/pkg/datastore/datastore_test.go index 08e4252a3a..11164bd381 100644 --- a/pkg/datastore/datastore_test.go +++ b/pkg/datastore/datastore_test.go @@ -3,6 +3,7 @@ package datastore import ( "context" "testing" + "time" "github.com/stretchr/testify/require" @@ -622,8 +623,8 @@ func (f fakeDatastore) ReadWriteTx(_ context.Context, _ TxUserFunc, _ ...options return nil, nil } -func (f fakeDatastore) OptimizedRevision(_ context.Context) (RevisionWithSchemaHash, error) { - return RevisionWithSchemaHash{}, nil +func (f fakeDatastore) OptimizedRevision(_ context.Context) (Revision, time.Duration, string, error) { + return NoRevision, 0, "", nil } func (f fakeDatastore) HeadRevision(_ context.Context) (RevisionWithSchemaHash, error) { diff --git a/pkg/datastore/mocks/mock_datastore.go b/pkg/datastore/mocks/mock_datastore.go index 5f91bc62bc..dbe882ed17 100644 --- a/pkg/datastore/mocks/mock_datastore.go +++ b/pkg/datastore/mocks/mock_datastore.go @@ -12,6 +12,7 @@ package mocks import ( context "context" reflect "reflect" + time "time" v1 "github.com/authzed/authzed-go/proto/authzed/api/v1" datastore "github.com/authzed/spicedb/pkg/datastore" @@ -794,12 +795,14 @@ func (mr *MockReadOnlyDatastoreMockRecorder) OfflineFeatures() *gomock.Call { } // OptimizedRevision mocks base method. -func (m *MockReadOnlyDatastore) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (m *MockReadOnlyDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "OptimizedRevision", ctx) - ret0, _ := ret[0].(datastore.RevisionWithSchemaHash) - ret1, _ := ret[1].(error) - return ret0, ret1 + ret0, _ := ret[0].(datastore.Revision) + ret1, _ := ret[1].(time.Duration) + ret2, _ := ret[2].(string) + ret3, _ := ret[3].(error) + return ret0, ret1, ret2, ret3 } // OptimizedRevision indicates an expected call of OptimizedRevision. @@ -1010,12 +1013,14 @@ func (mr *MockDatastoreMockRecorder) OfflineFeatures() *gomock.Call { } // OptimizedRevision mocks base method. -func (m *MockDatastore) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (m *MockDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "OptimizedRevision", ctx) - ret0, _ := ret[0].(datastore.RevisionWithSchemaHash) - ret1, _ := ret[1].(error) - return ret0, ret1 + ret0, _ := ret[0].(datastore.Revision) + ret1, _ := ret[1].(time.Duration) + ret2, _ := ret[2].(string) + ret3, _ := ret[3].(error) + return ret0, ret1, ret2, ret3 } // OptimizedRevision indicates an expected call of OptimizedRevision. @@ -1370,12 +1375,14 @@ func (mr *MockSQLDatastoreMockRecorder) OfflineFeatures() *gomock.Call { } // OptimizedRevision mocks base method. -func (m *MockSQLDatastore) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (m *MockSQLDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "OptimizedRevision", ctx) - ret0, _ := ret[0].(datastore.RevisionWithSchemaHash) - ret1, _ := ret[1].(error) - return ret0, ret1 + ret0, _ := ret[0].(datastore.Revision) + ret1, _ := ret[1].(time.Duration) + ret2, _ := ret[2].(string) + ret3, _ := ret[3].(error) + return ret0, ret1, ret2, ret3 } // OptimizedRevision indicates an expected call of OptimizedRevision. @@ -1649,12 +1656,14 @@ func (mr *MockStrictReadDatastoreMockRecorder) OfflineFeatures() *gomock.Call { } // OptimizedRevision mocks base method. -func (m *MockStrictReadDatastore) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (m *MockStrictReadDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "OptimizedRevision", ctx) - ret0, _ := ret[0].(datastore.RevisionWithSchemaHash) - ret1, _ := ret[1].(error) - return ret0, ret1 + ret0, _ := ret[0].(datastore.Revision) + ret1, _ := ret[1].(time.Duration) + ret2, _ := ret[2].(string) + ret3, _ := ret[3].(error) + return ret0, ret1, ret2, ret3 } // OptimizedRevision indicates an expected call of OptimizedRevision. @@ -1885,12 +1894,14 @@ func (mr *MockStartableDatastoreMockRecorder) OfflineFeatures() *gomock.Call { } // OptimizedRevision mocks base method. -func (m *MockStartableDatastore) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (m *MockStartableDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "OptimizedRevision", ctx) - ret0, _ := ret[0].(datastore.RevisionWithSchemaHash) - ret1, _ := ret[1].(error) - return ret0, ret1 + ret0, _ := ret[0].(datastore.Revision) + ret1, _ := ret[1].(time.Duration) + ret2, _ := ret[2].(string) + ret3, _ := ret[3].(error) + return ret0, ret1, ret2, ret3 } // OptimizedRevision indicates an expected call of OptimizedRevision. @@ -2135,12 +2146,14 @@ func (mr *MockRepairableDatastoreMockRecorder) OfflineFeatures() *gomock.Call { } // OptimizedRevision mocks base method. -func (m *MockRepairableDatastore) OptimizedRevision(ctx context.Context) (datastore.RevisionWithSchemaHash, error) { +func (m *MockRepairableDatastore) OptimizedRevision(ctx context.Context) (datastore.Revision, time.Duration, string, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "OptimizedRevision", ctx) - ret0, _ := ret[0].(datastore.RevisionWithSchemaHash) - ret1, _ := ret[1].(error) - return ret0, ret1 + ret0, _ := ret[0].(datastore.Revision) + ret1, _ := ret[1].(time.Duration) + ret2, _ := ret[2].(string) + ret3, _ := ret[3].(error) + return ret0, ret1, ret2, ret3 } // OptimizedRevision indicates an expected call of OptimizedRevision. diff --git a/pkg/datastore/test/revisions.go b/pkg/datastore/test/revisions.go index ac16409e2d..0eadc1c750 100644 --- a/pkg/datastore/test/revisions.go +++ b/pkg/datastore/test/revisions.go @@ -39,9 +39,8 @@ func RevisionQuantizationTest(t *testing.T, tester DatastoreTester) { require.NoError(err) ctx := t.Context() - veryFirstRevisionResult, err := ds.OptimizedRevision(ctx) + veryFirstRevision, _, _, err := ds.OptimizedRevision(ctx) require.NoError(err) - veryFirstRevision := veryFirstRevisionResult.Revision postSetupRevision := setupDatastore(t, ds) require.True(postSetupRevision.GreaterThan(veryFirstRevision), "post-setup revision should be greater than the first revision") @@ -65,9 +64,8 @@ func RevisionQuantizationTest(t *testing.T, tester DatastoreTester) { // Now we should ONLY get revisions later than the now revision for start := time.Now(); time.Since(start) < 10*time.Millisecond; { - testRevisionResult, err := ds.OptimizedRevision(ctx) + testRevision, _, _, err := ds.OptimizedRevision(ctx) require.NoError(err) - testRevision := testRevisionResult.Revision require.True(nowRevision.LessThan(testRevision) || nowRevision.Equal(testRevision)) } }) diff --git a/pkg/datastore/test/storedschema.go b/pkg/datastore/test/storedschema.go index 1ef2451e86..49359014e1 100644 --- a/pkg/datastore/test/storedschema.go +++ b/pkg/datastore/test/storedschema.go @@ -757,9 +757,9 @@ func OptimizedRevisionSchemaHashTest(t *testing.T, tester DatastoreTester) { expectedFirstHash, err := generator.ComputeSchemaHash(firstDefs) require.NoError(err) - resultAfterFirst, err := ds.OptimizedRevision(ctx) + _, _, firstHash, err := ds.OptimizedRevision(ctx) require.NoError(err) - require.Equal(expectedFirstHash, resultAfterFirst.SchemaHash, + require.Equal(expectedFirstHash, firstHash, "OptimizedRevision should return the schema hash matching the written schema") // Write a different schema. @@ -774,11 +774,11 @@ func OptimizedRevisionSchemaHashTest(t *testing.T, tester DatastoreTester) { expectedSecondHash, err := generator.ComputeSchemaHash(secondDefs) require.NoError(err) - resultAfterSecond, err := ds.OptimizedRevision(ctx) + _, _, secondHash, err := ds.OptimizedRevision(ctx) require.NoError(err) - require.Equal(expectedSecondHash, resultAfterSecond.SchemaHash, + require.Equal(expectedSecondHash, secondHash, "OptimizedRevision should return the updated schema hash after a schema change") - require.NotEqual(resultAfterFirst.SchemaHash, resultAfterSecond.SchemaHash, + require.NotEqual(firstHash, secondHash, "OptimizedRevision schema hashes should differ after writing a different schema") } diff --git a/pkg/middleware/consistency/consistency_test.go b/pkg/middleware/consistency/consistency_test.go index 45b8c0d052..89917cc384 100644 --- a/pkg/middleware/consistency/consistency_test.go +++ b/pkg/middleware/consistency/consistency_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "testing" + "time" "github.com/stretchr/testify/require" "google.golang.org/grpc/codes" @@ -35,7 +36,7 @@ func TestAddRevisionToContextNoneSupplied(t *testing.T) { require := require.New(t) ds := &proxy_test.MockDatastore{} - ds.On("OptimizedRevision").Return(optimizedWithHash, nil).Once() + ds.On("OptimizedRevision").Return(optimizedWithHash.Revision, time.Duration(0), optimizedWithHash.SchemaHash, nil).Once() dl := datalayer.NewDataLayer(ds) updated := ContextWithHandle(t.Context()) @@ -55,7 +56,7 @@ func TestAddRevisionToContextMinimizeLatency(t *testing.T) { require := require.New(t) ds := &proxy_test.MockDatastore{} - ds.On("OptimizedRevision").Return(optimizedWithHash, nil).Once() + ds.On("OptimizedRevision").Return(optimizedWithHash.Revision, time.Duration(0), optimizedWithHash.SchemaHash, nil).Once() dl := datalayer.NewDataLayer(ds) updated := ContextWithHandle(t.Context()) @@ -107,7 +108,7 @@ func TestAddRevisionToContextAtLeastAsFresh(t *testing.T) { require := require.New(t) ds := &proxy_test.MockDatastore{} - ds.On("OptimizedRevision").Return(optimizedWithHash, nil).Once() + ds.On("OptimizedRevision").Return(optimizedWithHash.Revision, time.Duration(0), optimizedWithHash.SchemaHash, nil).Once() ds.On("RevisionFromString", exact.String()).Return(exact, nil).Once() dl := datalayer.NewDataLayer(ds) @@ -285,7 +286,7 @@ func TestAddRevisionToContextAtMalformedExactSnapshot(t *testing.T) { func TestAddRevisionToContextMalformedAtLeastAsFreshSnapshot(t *testing.T) { ds := &proxy_test.MockDatastore{} - ds.On("OptimizedRevision").Return(optimizedWithHash, nil).Once() + ds.On("OptimizedRevision").Return(optimizedWithHash.Revision, time.Duration(0), optimizedWithHash.SchemaHash, nil).Once() dl := datalayer.NewDataLayer(ds) err := AddRevisionToContext(ContextWithHandle(t.Context()), &v1.LookupResourcesRequest{ @@ -383,7 +384,7 @@ func TestAtLeastAsFreshWithMismatchedTokenExpectError(t *testing.T) { require := require.New(t) ds := &proxy_test.MockDatastore{} - ds.On("OptimizedRevision").Return(optimizedWithHash, nil).Once() + ds.On("OptimizedRevision").Return(optimizedWithHash.Revision, time.Duration(0), optimizedWithHash.SchemaHash, nil).Once() ds.On("RevisionFromString", optimized.String()).Return(optimized, nil).Once() dl := datalayer.NewDataLayer(ds) @@ -412,7 +413,7 @@ func TestAtLeastAsFreshWithMismatchedTokenExpectMinLatency(t *testing.T) { require := require.New(t) ds := &proxy_test.MockDatastore{} - ds.On("OptimizedRevision").Return(optimizedWithHash, nil).Once() + ds.On("OptimizedRevision").Return(optimizedWithHash.Revision, time.Duration(0), optimizedWithHash.SchemaHash, nil).Once() ds.On("RevisionFromString", optimized.String()).Return(optimized, nil).Once() dl := datalayer.NewDataLayer(ds) @@ -447,7 +448,7 @@ func TestAtLeastAsFreshWithMismatchedTokenExpectFullConsistency(t *testing.T) { ds := &proxy_test.MockDatastore{} ds.On("HeadRevision").Return(headWithHash, nil).Once() - ds.On("OptimizedRevision").Return(optimizedWithHash, nil).Once() + ds.On("OptimizedRevision").Return(optimizedWithHash.Revision, time.Duration(0), optimizedWithHash.SchemaHash, nil).Once() ds.On("RevisionFromString", optimized.String()).Return(optimized, nil).Once() dl := datalayer.NewDataLayer(ds) @@ -513,7 +514,7 @@ func TestAddRevisionToContextAtLeastAsFreshUsesTokenSchemaHashWhenTokenWins(t *t // Make OptimizedRevision return `optimized` (100) so the token at `exact` (123) wins. ds := &proxy_test.MockDatastore{} - ds.On("OptimizedRevision").Return(optimizedWithHash, nil).Once() + ds.On("OptimizedRevision").Return(optimizedWithHash.Revision, time.Duration(0), optimizedWithHash.SchemaHash, nil).Once() ds.On("RevisionFromString", exact.String()).Return(exact, nil).Once() dl := datalayer.NewDataLayer(ds) @@ -567,7 +568,7 @@ func TestAddRevisionToContextAtLeastAsFreshMatchingIDs(t *testing.T) { require := require.New(t) ds := &proxy_test.MockDatastore{} - ds.On("OptimizedRevision").Return(optimizedWithHash, nil).Once() + ds.On("OptimizedRevision").Return(optimizedWithHash.Revision, time.Duration(0), optimizedWithHash.SchemaHash, nil).Once() ds.On("RevisionFromString", exact.String()).Return(exact, nil).Once() ds.CurrentUniqueID = "foo"