diff --git a/CHANGELOG.md b/CHANGELOG.md index 85b5239355..16e28c0b93 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,9 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ## [Unreleased] +### Added +- Added an experimental CockroachDB `--datastore-experimental-crdb-changelog-watch` flag that serves the Watch API from a dual-written changelog table polled at a closed timestamp, keeping Watch advancing during bulk loads (at the cost of write latency). + ### Changed - Schema: reads inside write transactions now use a cheap hash-only lookup (`schema_revision`) to check the cache before loading the full schema blob, reducing DB round-trips on cache hits (https://github.com/authzed/spicedb/pull/3160) - Updated the Prometheus buckets for `grpc_server_handling_seconds` and `spicedb_datastore_query_latency` to be able to correlate them (https://github.com/authzed/spicedb/pull/3188) diff --git a/docs/spicedb.md b/docs/spicedb.md index 3efc6bd7c5..815eb07963 100644 --- a/docs/spicedb.md +++ b/docs/spicedb.md @@ -91,6 +91,8 @@ spicedb datastore gc [flags] --datastore-disable-watch-support disable watch support (only enable if you absolutely do not need watch) --datastore-engine string type of datastore to initialize ("cockroachdb", "mysql", "postgres", "spanner") (default "memory") --datastore-experimental-column-optimization enable experimental column optimization (default true) + --datastore-experimental-crdb-changelog-watch EXPERIMENTAL: use a dual-written changelog table polled at a closed timestamp for Watch instead of a changefeed (CockroachDB driver only). Keeps Watch advancing during bulk loads at the cost of write latency. + --datastore-experimental-crdb-changelog-watch-max-offset duration EXPERIMENTAL: safety margin subtracted from the cluster clock to derive the changelog Watch completeness cursor/checkpoint; MUST be >= the cluster's --max-offset (CockroachDB default 500ms) or Watch can miss commits (CockroachDB driver only) (default 250ms) --datastore-follower-read-delay-duration duration amount of time to subtract from non-sync revision timestamps to ensure they are sufficiently in the past to enable follower reads (CockroachDB and Spanner drivers only) or read replicas (Postgres and MySQL drivers only) (default 4.8s) --datastore-gc-interval duration amount of time between passes of garbage collection (Postgres driver only) (default 3m0s) --datastore-gc-max-operation-time duration maximum amount of time a garbage collection pass can operate before timing out (Postgres driver only) (default 1m0s) @@ -236,6 +238,8 @@ spicedb datastore repair [flags] --datastore-disable-watch-support disable watch support (only enable if you absolutely do not need watch) --datastore-engine string type of datastore to initialize ("cockroachdb", "mysql", "postgres", "spanner") (default "memory") --datastore-experimental-column-optimization enable experimental column optimization (default true) + --datastore-experimental-crdb-changelog-watch EXPERIMENTAL: use a dual-written changelog table polled at a closed timestamp for Watch instead of a changefeed (CockroachDB driver only). Keeps Watch advancing during bulk loads at the cost of write latency. + --datastore-experimental-crdb-changelog-watch-max-offset duration EXPERIMENTAL: safety margin subtracted from the cluster clock to derive the changelog Watch completeness cursor/checkpoint; MUST be >= the cluster's --max-offset (CockroachDB default 500ms) or Watch can miss commits (CockroachDB driver only) (default 250ms) --datastore-follower-read-delay-duration duration amount of time to subtract from non-sync revision timestamps to ensure they are sufficiently in the past to enable follower reads (CockroachDB and Spanner drivers only) or read replicas (Postgres and MySQL drivers only) (default 4.8s) --datastore-gc-interval duration amount of time between passes of garbage collection (Postgres driver only) (default 3m0s) --datastore-gc-max-operation-time duration maximum amount of time a garbage collection pass can operate before timing out (Postgres driver only) (default 1m0s) @@ -414,6 +418,8 @@ spicedb serve [flags] --datastore-disable-watch-support disable watch support (only enable if you absolutely do not need watch) --datastore-engine string type of datastore to initialize ("cockroachdb", "mysql", "postgres", "spanner") (default "memory") --datastore-experimental-column-optimization enable experimental column optimization (default true) + --datastore-experimental-crdb-changelog-watch EXPERIMENTAL: use a dual-written changelog table polled at a closed timestamp for Watch instead of a changefeed (CockroachDB driver only). Keeps Watch advancing during bulk loads at the cost of write latency. + --datastore-experimental-crdb-changelog-watch-max-offset duration EXPERIMENTAL: safety margin subtracted from the cluster clock to derive the changelog Watch completeness cursor/checkpoint; MUST be >= the cluster's --max-offset (CockroachDB default 500ms) or Watch can miss commits (CockroachDB driver only) (default 250ms) --datastore-follower-read-delay-duration duration amount of time to subtract from non-sync revision timestamps to ensure they are sufficiently in the past to enable follower reads (CockroachDB and Spanner drivers only) or read replicas (Postgres and MySQL drivers only) (default 4.8s) --datastore-gc-interval duration amount of time between passes of garbage collection (Postgres driver only) (default 3m0s) --datastore-gc-max-operation-time duration maximum amount of time a garbage collection pass can operate before timing out (Postgres driver only) (default 1m0s) diff --git a/internal/datastore/common/changes.go b/internal/datastore/common/changes.go index 8ae338ec90..d7881278a4 100644 --- a/internal/datastore/common/changes.go +++ b/internal/datastore/common/changes.go @@ -270,7 +270,7 @@ func (ch *Changes[R, K]) AddChangedDefinition( case *core.CaveatDefinition: delete(record.caveatsDeleted, t.Name) - if existing, ok := record.definitionsChanged[nsPrefix+t.Name]; ok { + if existing, ok := record.definitionsChanged[caveatPrefix+t.Name]; ok { if err := ch.adjustByteSize(existing, -1); err != nil { return err } diff --git a/internal/datastore/common/changes_test.go b/internal/datastore/common/changes_test.go index 9f5c872147..a0eb45515e 100644 --- a/internal/datastore/common/changes_test.go +++ b/internal/datastore/common/changes_test.go @@ -10,6 +10,7 @@ import ( "strings" "testing" + "github.com/ccoveille/go-safecast/v2" "github.com/stretchr/testify/require" "golang.org/x/sync/errgroup" @@ -828,6 +829,39 @@ func TestMaximumSizeReplacement(t *testing.T) { require.Equal(t, int64(243), ch.currentByteSize) } +// TestMaximumSizeCaveatReplacement is a regression test for a byte-accounting +// bug where re-changing the same caveat within one window looked up the +// prior entry under the namespace-prefixed key (nsPrefix+name) instead of the +// caveat-prefixed key it was actually stored under (caveatPrefix+name). That +// mismatch meant the prior entry's byte size was never decremented before the +// new one was added, so currentByteSize was over-counted on every repeated +// change to the same caveat. With the fix, replacing a caveat definition +// leaves currentByteSize unchanged (matching the analogous namespace +// behavior verified by TestMaximumSizeReplacement). +func TestMaximumSizeCaveatReplacement(t *testing.T) { + ctx := t.Context() + + caveatDef := &core.CaveatDefinition{Name: "somecaveat"} + size := caveatDef.SizeVT() + + ch := NewChanges(revisions.HLCKeyFunc, datastore.WatchRelationships|datastore.WatchSchema, safecast.MustConvert[uint64](size)) + require.True(t, ch.IsEmpty()) + + rev0, err := revisions.HLCRevisionFromString("1") + require.NoError(t, err) + + err = ch.AddChangedDefinition(ctx, rev0, caveatDef) + require.NoError(t, err) + require.EqualValues(t, size, ch.currentByteSize) + + // Changing the same caveat again must not accumulate byte size: the + // prior entry should be found (under caveatPrefix, not nsPrefix) and its + // size decremented before the new size is added back in. + err = ch.AddChangedDefinition(ctx, rev0, caveatDef) + require.NoError(t, err) + require.EqualValues(t, size, ch.currentByteSize) +} + func TestCanonicalize(t *testing.T) { testCases := []struct { name string diff --git a/internal/datastore/crdb/caveat.go b/internal/datastore/crdb/caveat.go index 5857a57028..1ba65ff21d 100644 --- a/internal/datastore/crdb/caveat.go +++ b/internal/datastore/crdb/caveat.go @@ -155,6 +155,20 @@ func (rwt *crdbReadWriteTXN) LegacyWriteCaveats(ctx context.Context, caveats []* if _, err := rwt.tx.Exec(ctx, sql, args...); err != nil { return fmt.Errorf(errWriteCaveat, err) } + + if rwt.changelogWatchEnabled { + ttl := rwt.changelogTTL() + for _, caveat := range caveats { + serialized, err := caveat.MarshalVT() + if err != nil { + return fmt.Errorf(errWriteCaveat, err) + } + if err := rwt.appendSchemaChangelog(ctx, "caveat", caveat.Name, serialized, rwt.nextChangelogOrdinal(), ttl); err != nil { + return err + } + } + } + return nil } @@ -170,5 +184,15 @@ func (rwt *crdbReadWriteTXN) LegacyDeleteCaveats(ctx context.Context, names []st if _, err := rwt.tx.Exec(ctx, sql, args...); err != nil { return fmt.Errorf(errDeleteCaveats, err) } + + if rwt.changelogWatchEnabled { + ttl := rwt.changelogTTL() + for _, name := range names { + if err := rwt.appendSchemaChangelog(ctx, "caveat", name, nil, rwt.nextChangelogOrdinal(), ttl); err != nil { + return err + } + } + } + return nil } diff --git a/internal/datastore/crdb/changelog.go b/internal/datastore/crdb/changelog.go new file mode 100644 index 0000000000..5838d448b7 --- /dev/null +++ b/internal/datastore/crdb/changelog.go @@ -0,0 +1,309 @@ +package crdb + +import ( + "context" + "fmt" + "time" + + sq "github.com/Masterminds/squirrel" + "github.com/jackc/pgx/v5/pgconn" + "google.golang.org/protobuf/proto" + + "github.com/authzed/spicedb/internal/datastore/crdb/pool" + "github.com/authzed/spicedb/internal/datastore/crdb/schema" + "github.com/authzed/spicedb/pkg/datastore" + core "github.com/authzed/spicedb/pkg/proto/core/v1" + "github.com/authzed/spicedb/pkg/tuple" +) + +// createChangelogTableSQL returns the DDL for the experimental changelog table. +// The table is intentionally created outside the linear migrate chain so it is +// only present when the experimental flag is set (prototype decision). +// +// A single table with a `kind` discriminator holds both relationship and schema +// changes. Each row is self-contained so Watch never joins back to the +// relationship tables. The primary key is hash-sharded over the monotonic +// change_ts to avoid a write hotspot. Row-level TTL matches the GC window; +// ttl_disable_changefeed_replication keeps TTL deletes from generating nudge +// noise. +// +// gcWindow is accepted but not yet threaded into the DDL: TTL is expressed +// via ttl_expiration_expression against the stored ttl_expiration column +// rather than a fixed interval. The parameter is kept so callers won't need +// to change once the GC window is wired into the expiration calculation. +// +//nolint:unparam // see above; parameter reserved for future use +func createChangelogTableSQL(gcWindow time.Duration) string { + return fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s ( + %s DECIMAL NOT NULL, + %s INT NOT NULL, + %s STRING NOT NULL, + %s STRING, %s STRING, %s STRING, + %s STRING, %s STRING, %s STRING, + %s STRING, %s JSONB, %s TIMESTAMPTZ, + %s STRING, + %s STRING, %s STRING, %s BYTES, + %s JSONB, + %s TIMESTAMPTZ NOT NULL, + PRIMARY KEY (%s, %s) USING HASH +) WITH ( + ttl_expiration_expression = '%s', + ttl_job_cron = '@hourly', + ttl_disable_changefeed_replication = 'true' +)`, + schema.TableRelationshipChangelog, + schema.ColChangeTS, + schema.ColChangeOrdinal, + schema.ColChangeKind, + schema.ColNamespace, schema.ColObjectID, schema.ColRelation, + schema.ColUsersetNamespace, schema.ColUsersetObjectID, schema.ColUsersetRelation, + schema.ColCaveatContextName, schema.ColCaveatContext, schema.ColChangeRelExpiration, + schema.ColChangeOperation, + schema.ColChangeSchemaKind, schema.ColChangeDefinitionName, schema.ColChangeSerializedDefinition, + schema.ColChangeMetadata, + schema.ColChangeTTLExpiration, + schema.ColChangeTS, schema.ColChangeOrdinal, + schema.ColChangeTTLExpiration, + ) +} + +// ensureChangelogTable creates the changelog table if it does not already exist. +func ensureChangelogTable(ctx context.Context, initPool *pool.RetryPool, gcWindow time.Duration) error { + return initPool.ExecFunc(ctx, func(ctx context.Context, tag pgconn.CommandTag, err error) error { + return err + }, createChangelogTableSQL(gcWindow)) +} + +// operationToChangelog maps a tuple update operation to its stored string. +func operationToChangelog(op tuple.UpdateOperation) (string, error) { + switch op { + case tuple.UpdateOperationCreate: + return "create", nil + case tuple.UpdateOperationTouch: + return "touch", nil + case tuple.UpdateOperationDelete: + return "delete", nil + default: + return "", fmt.Errorf("unknown changelog operation: %v", op) + } +} + +// appendRelationshipChangelog inserts a single relationship change row into the +// changelog in the current transaction. change_ts is set to the transaction's +// HLC via cluster_logical_timestamp() so it matches the relationship write's +// commit revision exactly. +func (rwt *crdbReadWriteTXN) appendRelationshipChangelog(ctx context.Context, rel tuple.Relationship, op tuple.UpdateOperation, ordinal int, ttlExpiration time.Time) error { + opString, err := operationToChangelog(op) + if err != nil { + return err + } + + var caveatName string + var caveatContext map[string]any + if rel.OptionalCaveat != nil { + caveatName = rel.OptionalCaveat.CaveatName + caveatContext = rel.OptionalCaveat.Context.AsMap() + } + + insert := psql.Insert(schema.TableRelationshipChangelog).Columns( + schema.ColChangeTS, + schema.ColChangeOrdinal, + schema.ColChangeKind, + schema.ColNamespace, schema.ColObjectID, schema.ColRelation, + schema.ColUsersetNamespace, schema.ColUsersetObjectID, schema.ColUsersetRelation, + schema.ColCaveatContextName, schema.ColCaveatContext, schema.ColChangeRelExpiration, + schema.ColChangeOperation, + schema.ColChangeTTLExpiration, + ).Values( + sq.Expr("cluster_logical_timestamp()"), + ordinal, + "rel", + rel.Resource.ObjectType, rel.Resource.ObjectID, rel.Resource.Relation, + rel.Subject.ObjectType, rel.Subject.ObjectID, rel.Subject.Relation, + caveatName, caveatContext, rel.OptionalExpiration, + opString, + ttlExpiration, + ) + + sql, args, err := insert.ToSql() + if err != nil { + return fmt.Errorf("unable to build changelog insert: %w", err) + } + if _, err := rwt.tx.Exec(ctx, sql, args...); err != nil { + return fmt.Errorf("unable to write changelog row: %w", err) + } + return nil +} + +// appendSchemaChangelog inserts a single schema change row. A nil serialized +// payload records a deletion of definitionName. +func (rwt *crdbReadWriteTXN) appendSchemaChangelog(ctx context.Context, schemaKind, definitionName string, serialized []byte, ordinal int, ttlExpiration time.Time) error { + insert := psql.Insert(schema.TableRelationshipChangelog).Columns( + schema.ColChangeTS, + schema.ColChangeOrdinal, + schema.ColChangeKind, + schema.ColChangeSchemaKind, schema.ColChangeDefinitionName, schema.ColChangeSerializedDefinition, + schema.ColChangeTTLExpiration, + ).Values( + sq.Expr("cluster_logical_timestamp()"), + ordinal, + "schema", + schemaKind, definitionName, serialized, + ttlExpiration, + ) + + sql, args, err := insert.ToSql() + if err != nil { + return fmt.Errorf("unable to build schema changelog insert: %w", err) + } + if _, err := rwt.tx.Exec(ctx, sql, args...); err != nil { + return fmt.Errorf("unable to write schema changelog row: %w", err) + } + return nil +} + +// appendMetadataChangelog inserts a single kind='metadata' changelog row +// carrying the raw user-supplied transaction metadata. All other typed +// columns are left null; only change_ts (via cluster_logical_timestamp()), +// ordinal, the metadata JSONB payload, and ttl_expiration are set. +// +// This is the changelog-mode replacement for the legacy transaction_metadata +// side-table: since the changelog only ever contains rows SpiceDB explicitly +// wrote, there is no ambiguity for a changefeed consumer to disambiguate +// (unlike the changefeed-on-relation_tuple path, where a TTL-driven deletion +// of expired relationships could otherwise be mistaken for a SpiceDB write), +// so no $spicedb_transaction_key marker is needed here. +func (rwt *crdbReadWriteTXN) appendMetadataChangelog(ctx context.Context, metadata map[string]any, ordinal int, ttlExpiration time.Time) error { + insert := psql.Insert(schema.TableRelationshipChangelog).Columns( + schema.ColChangeTS, + schema.ColChangeOrdinal, + schema.ColChangeKind, + schema.ColChangeMetadata, + schema.ColChangeTTLExpiration, + ).Values( + sq.Expr("cluster_logical_timestamp()"), + ordinal, + "metadata", + metadata, + ttlExpiration, + ) + + sql, args, err := insert.ToSql() + if err != nil { + return fmt.Errorf("unable to build metadata changelog insert: %w", err) + } + if _, err := rwt.tx.Exec(ctx, sql, args...); err != nil { + return fmt.Errorf("unable to write metadata changelog row: %w", err) + } + return nil +} + +// changelogTTL returns the TTL expiration to stamp on changelog rows. +func (rwt *crdbReadWriteTXN) changelogTTL() time.Time { + return time.Now().Add(rwt.gcWindow).Add(1 * time.Minute) +} + +// capturingBulkSource wraps a datastore.BulkWriteRelationshipSource, recording +// a copy of each relationship it yields so the changelog can be populated +// after the underlying COPY completes. +// +// datastore.BulkWriteRelationshipSource documents that sources "may re-use +// the same memory address for every tuple" across calls to Next, so this +// wrapper cannot simply retain the returned pointer: it copies the +// Relationship value and, defensively, deep-copies the caveat proto (the one +// nested pointer a source implementation might otherwise mutate/reuse +// in-place) before storing it. +type capturingBulkSource struct { + inner datastore.BulkWriteRelationshipSource + seen []tuple.Relationship +} + +func (c *capturingBulkSource) Next(ctx context.Context) (*tuple.Relationship, error) { + rel, err := c.inner.Next(ctx) + if err != nil { + return nil, err + } + if rel != nil { + captured := *rel + if rel.OptionalCaveat != nil { + captured.OptionalCaveat = proto.Clone(rel.OptionalCaveat).(*core.ContextualizedCaveat) + } + c.seen = append(c.seen, captured) + } + return rel, nil +} + +// changelogInsertChunkSize caps how many relationship rows are combined into +// a single multi-row changelog INSERT. Postgres (and pgx, which CRDB's wire +// protocol emulates) caps the number of bound parameters on a single +// statement at 65535. Each changelog row binds 14 columns, so this table's +// bulk-load path -- which can be handed many thousands of relationships in +// one transaction -- must split the INSERT into chunks well under that +// limit. It is a package-level var (rather than a const) so tests can lower +// it to exercise the multi-chunk path without loading a huge number of rows. +var changelogInsertChunkSize = 4000 + +// appendRelationshipChangelogBatch inserts one 'create' changelog row per +// relationship, split across one or more multi-row INSERTs of at most +// changelogInsertChunkSize rows each, all executed within rwt.tx and all +// stamped with the transaction's cluster_logical_timestamp(). Chunking +// exists solely to stay under the ~65535 bound-parameter limit enforced by +// the Postgres wire protocol (14 columns per row); it does not introduce +// separate transactions, so it has no effect on atomicity. +// +// Every row -- across every chunk -- gets a unique ordinal from +// rwt.nextChangelogOrdinal(), called exactly once per relationship in +// original order. The ordinal counter is NOT reset between chunks: since +// change_ts is derived from cluster_logical_timestamp() and is therefore +// constant for the whole transaction, uniqueness of the (change_ts, +// ordinal) primary key depends entirely on the ordinal being monotonically +// increasing across the entire batch, not just within one chunk. +func (rwt *crdbReadWriteTXN) appendRelationshipChangelogBatch(ctx context.Context, rels []tuple.Relationship, ttlExpiration time.Time) error { + if len(rels) == 0 { + return nil + } + + for start := 0; start < len(rels); start += changelogInsertChunkSize { + end := start + changelogInsertChunkSize + if end > len(rels) { + end = len(rels) + } + + insert := psql.Insert(schema.TableRelationshipChangelog).Columns( + schema.ColChangeTS, + schema.ColChangeOrdinal, + schema.ColChangeKind, + schema.ColNamespace, schema.ColObjectID, schema.ColRelation, + schema.ColUsersetNamespace, schema.ColUsersetObjectID, schema.ColUsersetRelation, + schema.ColCaveatContextName, schema.ColCaveatContext, schema.ColChangeRelExpiration, + schema.ColChangeOperation, + schema.ColChangeTTLExpiration, + ) + for _, rel := range rels[start:end] { + var caveatName string + var caveatContext map[string]any + if rel.OptionalCaveat != nil { + caveatName = rel.OptionalCaveat.CaveatName + caveatContext = rel.OptionalCaveat.Context.AsMap() + } + insert = insert.Values( + sq.Expr("cluster_logical_timestamp()"), + rwt.nextChangelogOrdinal(), + "rel", + rel.Resource.ObjectType, rel.Resource.ObjectID, rel.Resource.Relation, + rel.Subject.ObjectType, rel.Subject.ObjectID, rel.Subject.Relation, + caveatName, caveatContext, rel.OptionalExpiration, + "create", + ttlExpiration, + ) + } + sql, args, err := insert.ToSql() + if err != nil { + return fmt.Errorf("unable to build bulk changelog insert: %w", err) + } + if _, err := rwt.tx.Exec(ctx, sql, args...); err != nil { + return fmt.Errorf("unable to write bulk changelog rows: %w", err) + } + } + return nil +} diff --git a/internal/datastore/crdb/changelog_integration_test.go b/internal/datastore/crdb/changelog_integration_test.go new file mode 100644 index 0000000000..6aad6d9060 --- /dev/null +++ b/internal/datastore/crdb/changelog_integration_test.go @@ -0,0 +1,676 @@ +//go:build datastore + +package crdb + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/shopspring/decimal" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/structpb" + + "github.com/authzed/spicedb/internal/datastore/crdb/schema" + "github.com/authzed/spicedb/internal/datastore/revisions" + testdatastore "github.com/authzed/spicedb/internal/testserver/datastore" + "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/datastore/options" + ns "github.com/authzed/spicedb/pkg/namespace" + "github.com/authzed/spicedb/pkg/tuple" +) + +// testBulkSource is a slice-backed datastore.BulkWriteRelationshipSource for +// exercising BulkLoad in tests. It yields a fresh *tuple.Relationship on each +// call (never reusing the same pointer), mirroring the shape of +// testfixtures.BulkRelationshipGenerator. +type testBulkSource struct { + rels []tuple.Relationship + next int +} + +func (s *testBulkSource) Next(_ context.Context) (*tuple.Relationship, error) { + if s.next >= len(s.rels) { + return nil, nil + } + rel := s.rels[s.next] + s.next++ + return &rel, nil +} + +var _ datastore.BulkWriteRelationshipSource = &testBulkSource{} + +// TestChangelogTableCreatedWhenEnabled verifies that newCRDBDatastore creates +// the experimental relationship_changelog table when the changelog-watch flag +// is enabled. It constructs the datastore directly via the unexported +// newCRDBDatastore (rather than the exported NewCRDBDatastore, which wraps the +// result in datastore.NewSeparatingContextDatastoreProxy) so the test can +// reach the concrete *crdbDatastore and inspect its readPool. +func TestChangelogTableCreatedWhenEnabled(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + + ctx := t.Context() + ds := engine.NewDatastore(t, func(engine, uri string) datastore.Datastore { + ds, err := newCRDBDatastore(ctx, uri, ExperimentalChangelogWatch(true)) + require.NoError(t, err) + return ds + }) + t.Cleanup(func() { + _ = ds.Close() + }) + + cds, ok := ds.(*crdbDatastore) + require.True(t, ok, "expected *crdbDatastore, got %T", ds) + + var exists bool + require.NoError(t, cds.readPool.QueryRowFunc(ctx, func(_ context.Context, row pgx.Row) error { + return row.Scan(&exists) + }, "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = $1)", schema.TableRelationshipChangelog)) + require.True(t, exists, "changelog table should exist when flag is on") +} + +// extractCRDBDatastore unwraps a (possibly proxy-wrapped) datastore.Datastore +// into the concrete *crdbDatastore, so tests can reach unexported fields such +// as readPool. Mirrors the unwrapping approach used elsewhere in this package +// (see datastore.UnwrapAs and partitioner_test.go). +func extractCRDBDatastore(t *testing.T, ds datastore.Datastore) *crdbDatastore { + t.Helper() + cds := datastore.UnwrapAs[*crdbDatastore](ds) + require.NotNil(t, cds, "expected to unwrap into *crdbDatastore, got %T", ds) + return cds +} + +// TestRelationshipDualWrite verifies that WriteRelationships also inserts a +// self-contained changelog row per mutation, in the same transaction, when +// the changelog-watch flag is enabled. +func TestRelationshipDualWrite(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx := t.Context() + rel := tuple.MustParse("document:doc1#viewer@user:alice") + _, err := ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error { + return rwt.WriteRelationships(ctx, []tuple.RelationshipUpdate{tuple.Touch(rel)}) + }) + require.NoError(t, err) + + cds := extractCRDBDatastore(t, ds) + var count int + require.NoError(t, cds.readPool.QueryRowFunc(ctx, func(_ context.Context, row pgx.Row) error { + return row.Scan(&count) + }, "SELECT count(*) FROM "+schema.TableRelationshipChangelog+" WHERE kind = 'rel' AND object_id = 'doc1'")) + require.Equal(t, 1, count, "expected exactly one changelog row for the write") + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestSchemaDualWrite verifies that LegacyWriteNamespaces also inserts a +// self-contained schema changelog row in the same transaction, when the +// changelog-watch flag is enabled. +func TestSchemaDualWrite(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx := t.Context() + nsDef := ns.Namespace("document", ns.MustRelation("viewer", nil)) + _, err := ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error { + return rwt.LegacyWriteNamespaces(ctx, nsDef) + }) + require.NoError(t, err) + + cds := extractCRDBDatastore(t, ds) + var count int + require.NoError(t, cds.readPool.QueryRowFunc(ctx, func(_ context.Context, row pgx.Row) error { + return row.Scan(&count) + }, "SELECT count(*) FROM "+schema.TableRelationshipChangelog+" WHERE kind = 'schema' AND schema_kind = 'namespace' AND definition_name = 'document'")) + require.Equal(t, 1, count) + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogWatchReceivesWrite verifies that, when the changelog-watch flag +// is enabled, Watch serves relationship changes by polling the changelog table +// at a closed timestamp rather than by consuming a CRDB changefeed. It opens a +// watch from HeadRevision, writes a relationship, and asserts the change is +// delivered over the watch channel. +func TestChangelogWatchReceivesWrite(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + head, err := ds.HeadRevision(ctx) + require.NoError(t, err) + + changes, errchan := ds.Watch(ctx, head.Revision, datastore.WatchOptions{ + Content: datastore.WatchRelationships | datastore.WatchCheckpoints, + CheckpointInterval: 100 * time.Millisecond, + }) + + rel := tuple.MustParse("document:doc1#viewer@user:alice") + _, err = ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error { + return rwt.WriteRelationships(ctx, []tuple.RelationshipUpdate{tuple.Touch(rel)}) + }) + require.NoError(t, err) + + found := false + deadline := time.After(30 * time.Second) + for !found { + select { + case change, ok := <-changes: + require.True(t, ok) + for _, rc := range change.RelationshipChanges { + if rc.Relationship.Resource.ObjectID == "doc1" { + found = true + } + } + case err := <-errchan: + require.NoError(t, err) + case <-deadline: + t.Fatal("did not receive the write over changelog watch") + } + } + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogWatchEmitsMetadata verifies that, in changelog-watch mode, +// per-transaction metadata passed via options.WithMetadata is carried through +// the changelog itself (as a kind='metadata' row) and surfaced on the +// corresponding RevisionChanges.Metadatas over Watch. It also asserts that +// the legacy transaction_metadata side-table is never written to in this +// mode -- the changelog fully replaces its purpose here, including the +// TTL-delete disambiguation marker, since the changelog only ever contains +// rows SpiceDB explicitly wrote. +func TestChangelogWatchEmitsMetadata(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + head, err := ds.HeadRevision(ctx) + require.NoError(t, err) + + changes, errchan := ds.Watch(ctx, head.Revision, datastore.WatchOptions{ + Content: datastore.WatchRelationships | datastore.WatchCheckpoints, + CheckpointInterval: 100 * time.Millisecond, + }) + + metadata, err := structpb.NewStruct(map[string]any{ + "reason": "integration-test", + "count": float64(7), + }) + require.NoError(t, err) + + rel := tuple.MustParse("document:metadoc#viewer@user:alice") + _, err = ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error { + return rwt.WriteRelationships(ctx, []tuple.RelationshipUpdate{tuple.Touch(rel)}) + }, options.WithMetadata(metadata)) + require.NoError(t, err) + + found := false + deadline := time.After(30 * time.Second) + for !found { + select { + case change, ok := <-changes: + require.True(t, ok) + for _, rc := range change.RelationshipChanges { + if rc.Relationship.Resource.ObjectID == "metadoc" { + require.Len(t, change.Metadatas, 1) + require.Equal(t, metadata.AsMap(), change.Metadatas[0].AsMap()) + found = true + } + } + case err := <-errchan: + require.NoError(t, err) + case <-deadline: + t.Fatal("did not receive the metadata-bearing write over changelog watch") + } + } + + // The legacy transaction_metadata table is unused entirely in + // changelog mode -- the changelog row above is the sole carrier for + // this transaction's metadata. + cds := extractCRDBDatastore(t, ds) + var count int + require.NoError(t, cds.readPool.QueryRowFunc(ctx, func(_ context.Context, row pgx.Row) error { + return row.Scan(&count) + }, "SELECT count(*) FROM "+schema.TableTransactionMetadata)) + require.Equal(t, 0, count, "transaction_metadata must not be written in changelog mode") + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogWatchEmitsImmediately verifies that, with +// EmitImmediatelyStrategy, the changelog poll emits a change as soon as it is +// read (rather than buffering until a checkpoint) and preserves the Create +// operation — the buffered path normalizes Create to Touch, so observing a +// Create on the wire is a mode-distinguishing signal that the immediate path +// is actually taken. It also confirms checkpoints still flow. +func TestChangelogWatchEmitsImmediately(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + head, err := ds.HeadRevision(ctx) + require.NoError(t, err) + + changes, errchan := ds.Watch(ctx, head.Revision, datastore.WatchOptions{ + Content: datastore.WatchRelationships | datastore.WatchCheckpoints, + CheckpointInterval: 100 * time.Millisecond, + EmissionStrategy: datastore.EmitImmediatelyStrategy, + }) + + rel := tuple.MustParse("document:immdoc#viewer@user:alice") + _, err = ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error { + return rwt.WriteRelationships(ctx, []tuple.RelationshipUpdate{tuple.Create(rel)}) + }) + require.NoError(t, err) + + gotChange, gotCheckpoint := false, false + deadline := time.After(30 * time.Second) + for !(gotChange && gotCheckpoint) { + select { + case change, ok := <-changes: + require.True(t, ok) + if change.IsCheckpoint { + gotCheckpoint = true + } + for _, rc := range change.RelationshipChanges { + if rc.Relationship.Resource.ObjectID == "immdoc" { + require.Equal(t, tuple.UpdateOperationCreate, rc.Operation, + "immediate mode must preserve Create (buffered would normalize to Touch)") + gotChange = true + } + } + case err := <-errchan: + require.NoError(t, err) + case <-deadline: + t.Fatal("did not receive immediate change + checkpoint over changelog watch") + } + } + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogWatchAdvancesCheckpointsWhenIdle verifies that checkpoints keep +// advancing on an idle watch with no writes: each poll re-reads +// cluster_logical_timestamp(), so safeTS (and thus the emitted checkpoint +// revision) moves forward on the ticker even when nothing is being written. +func TestChangelogWatchAdvancesCheckpointsWhenIdle(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + head, err := ds.HeadRevision(ctx) + require.NoError(t, err) + + changes, errchan := ds.Watch(ctx, head.Revision, datastore.WatchOptions{ + Content: datastore.WatchRelationships | datastore.WatchCheckpoints, + CheckpointInterval: 100 * time.Millisecond, + }) + + // No writes. Capture the first checkpoint, then wait for one strictly + // greater (there may be an initial plateau while safeTS climbs past the + // starting cursor). + var first datastore.Revision + haveFirst := false + deadline := time.After(30 * time.Second) + for { + select { + case change, ok := <-changes: + require.True(t, ok) + if !change.IsCheckpoint { + continue + } + if !haveFirst { + first = change.Revision + haveFirst = true + continue + } + if change.Revision.GreaterThan(first) { + return + } + case err := <-errchan: + require.NoError(t, err) + case <-deadline: + t.Fatal("idle changelog watch did not emit an advancing checkpoint") + } + } + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogWatchNudgeDeliversQuickly verifies that the changefeed "nudge" +// wakes the poll loop well before the (deliberately long) ticker interval +// fires. It opens a watch with a 30s CheckpointInterval, waits long enough +// for the nudge's underlying changefeed job to finish standing up (CRDB +// changefeed job creation is a several-second, one-time cost, distinct from +// the per-event latency under test), then writes a relationship and asserts +// the change arrives well under the interval -- which is only possible if +// the nudge, and not the 30s ticker, triggered the poll. +func TestChangelogWatchNudgeDeliversQuickly(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + head, err := ds.HeadRevision(ctx) + require.NoError(t, err) + changes, errchan := ds.Watch(ctx, head.Revision, datastore.WatchOptions{ + Content: datastore.WatchRelationships | datastore.WatchCheckpoints, + CheckpointInterval: 30 * time.Second, + }) + + // Let the nudge's changefeed job finish standing up. Job creation in + // CockroachDB is a several-second, one-time fixed cost, orthogonal to + // how quickly a nudge wakes the poll loop once the changefeed is + // live; draining it here isolates the per-event latency we actually + // want to measure below. + time.Sleep(10 * time.Second) + + start := time.Now() + rel := tuple.MustParse("document:doc1#viewer@user:alice") + _, err = ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error { + return rwt.WriteRelationships(ctx, []tuple.RelationshipUpdate{tuple.Touch(rel)}) + }) + require.NoError(t, err) + for { + select { + case change, ok := <-changes: + require.True(t, ok) + for _, rc := range change.RelationshipChanges { + if rc.Relationship.Resource.ObjectID == "doc1" { + require.Less(t, time.Since(start), 10*time.Second, "nudge should deliver well before the 30s interval") + return + } + } + case err := <-errchan: + require.NoError(t, err) + case <-time.After(15 * time.Second): + t.Fatal("nudge did not deliver before near the interval") + } + } + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogPollEmitsRowAtExactlyTarget is a boundary regression test for the +// off-by-one where a changelog row whose change_ts exactly equals a poll's +// target revision was read into the tracker but never emitted (the strict +// "< target" filter dropped it) while the cursor still advanced to target, +// permanently losing that row. +// +// It inserts a changelog row with a KNOWN explicit change_ts value T (bypassing +// dual-write so change_ts is under test control), then invokes the poll range +// helper with cursor < T and target == T, and asserts the row is emitted exactly +// once. Against the pre-fix strict filter this fails (the row is dropped); with +// the inclusive emission it passes. +func TestChangelogPollEmitsRowAtExactlyTarget(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx := t.Context() + cds := extractCRDBDatastore(t, ds) + + conn, err := pgx.Connect(ctx, cds.dburl) + require.NoError(t, err) + defer func() { _ = conn.Close(ctx) }() + + // Choose T from the cluster's own logical clock so it is a valid HLC value, + // then insert a relationship changelog row stamped at exactly T. + var target decimal.Decimal + require.NoError(t, conn.QueryRow(ctx, "SELECT cluster_logical_timestamp()").Scan(&target)) + + insert := "INSERT INTO " + schema.TableRelationshipChangelog + " (" + + schema.ColChangeTS + ", " + schema.ColChangeOrdinal + ", " + schema.ColChangeKind + ", " + + schema.ColNamespace + ", " + schema.ColObjectID + ", " + schema.ColRelation + ", " + + schema.ColUsersetNamespace + ", " + schema.ColUsersetObjectID + ", " + schema.ColUsersetRelation + ", " + + schema.ColChangeOperation + ", " + schema.ColChangeTTLExpiration + + ") VALUES ($1, 0, 'rel', 'document', 'boundary', 'viewer', 'user', 'alice', '...', 'touch', $2)" + _, err = conn.Exec(ctx, insert, target, time.Now().Add(1*time.Hour)) + require.NoError(t, err) + + // cursor strictly less than T; target exactly equal to T. + cursor := target.Sub(decimal.NewFromInt(1)) + + tx, err := conn.BeginTx(ctx, pgx.TxOptions{AccessMode: pgx.ReadOnly}) + require.NoError(t, err) + defer func() { _ = tx.Rollback(ctx) }() + + var emitted []datastore.RevisionChanges + sendChange := func(change datastore.RevisionChanges) error { + emitted = append(emitted, change) + return nil + } + + require.NoError(t, cds.pollChangelogRange(ctx, tx, datastore.WatchOptions{ + Content: datastore.WatchRelationships, + }, cursor, target, cds.watchChangeBufferMaximumSize, sendChange)) + + count := 0 + for _, change := range emitted { + for _, rc := range change.RelationshipChanges { + if rc.Relationship.Resource.ObjectID == "boundary" { + count++ + } + } + } + require.Equal(t, 1, count, "row at exactly the poll target must be emitted exactly once") + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogImmediateCursorAndDedup deterministically exercises the two +// load-bearing properties of the immediate-emission poll without relying on +// real clock skew, by calling pollChangelogImmediate directly with pinned +// clusterNow/safeTS across three polls: +// - INV1: the completeness cursor advances only to safeTS, never clusterNow. +// - INV6: a provisional-window row (safeTS < change_ts <= clusterNow) is +// emitted exactly once even though later polls re-read it, and its dedup +// key is pruned once safeTS passes it. +func TestChangelogImmediateCursorAndDedup(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx := t.Context() + cds := extractCRDBDatastore(t, ds) + + conn, err := pgx.Connect(ctx, cds.dburl) + require.NoError(t, err) + defer func() { _ = conn.Close(ctx) }() + + // Base the row's change_ts (TR) on the cluster clock so it is a valid + // HLC, then insert a single relationship changelog row stamped at TR. + var tr decimal.Decimal + require.NoError(t, conn.QueryRow(ctx, "SELECT cluster_logical_timestamp()").Scan(&tr)) + + insert := "INSERT INTO " + schema.TableRelationshipChangelog + " (" + + schema.ColChangeTS + ", " + schema.ColChangeOrdinal + ", " + schema.ColChangeKind + ", " + + schema.ColNamespace + ", " + schema.ColObjectID + ", " + schema.ColRelation + ", " + + schema.ColUsersetNamespace + ", " + schema.ColUsersetObjectID + ", " + schema.ColUsersetRelation + ", " + + schema.ColChangeOperation + ", " + schema.ColChangeTTLExpiration + + ") VALUES ($1, 0, 'rel', 'document', 'provrow', 'viewer', 'user', 'alice', '...', 'create', $2)" + _, err = conn.Exec(ctx, insert, tr, time.Now().Add(1*time.Hour)) + require.NoError(t, err) + + emittedWindow := make(map[changelogRowKey]struct{}) + provCount := 0 + sendChange := func(change datastore.RevisionChanges) error { + for _, rc := range change.RelationshipChanges { + if rc.Relationship.Resource.ObjectID == "provrow" { + provCount++ + } + } + return nil + } + opts := datastore.WatchOptions{Content: datastore.WatchRelationships | datastore.WatchCheckpoints} + + poll := func(cursor, clusterNow, safeTS decimal.Decimal) decimal.Decimal { + tx, err := conn.BeginTx(ctx, pgx.TxOptions{AccessMode: pgx.ReadOnly}) + require.NoError(t, err) + defer func() { _ = tx.Rollback(ctx) }() + newCursor, err := cds.pollChangelogImmediate(ctx, tx, opts, cursor, clusterNow, safeTS, emittedWindow, sendChange) + require.NoError(t, err) + return newCursor + } + + one := decimal.NewFromInt(1) + ten := decimal.NewFromInt(10) + nine := decimal.NewFromInt(9) + + // Poll 1: TR is provisional (safeTS = TR-1 < TR <= clusterNow = TR). The + // cursor must advance only to safeTS (TR-1), NOT clusterNow (TR). + c1 := poll(tr.Sub(ten), tr, tr.Sub(one)) + require.True(t, c1.Equal(tr.Sub(one)), "cursor must advance to safeTS (TR-1), got %s", c1) + require.Equal(t, 1, provCount, "provisional row emitted once on the first poll") + require.Len(t, emittedWindow, 1, "provisional row key retained for dedup") + + // Poll 2: still provisional; the row is re-read but must be deduped, and + // the cursor is unchanged (safeTS == cursor, so no advance). + c2 := poll(c1, tr, tr.Sub(one)) + require.True(t, c2.Equal(tr.Sub(one)), "cursor unchanged while safeTS does not advance") + require.Equal(t, 1, provCount, "provisional row must not be re-emitted (dedup)") + require.Len(t, emittedWindow, 1, "dedup key retained while safeTS has not passed TR") + + // Poll 3: safeTS advances past TR (clusterNow = TR+10, safeTS = TR+9). The + // row is re-read but deduped; the cursor advances to safeTS (TR+9), not + // clusterNow (TR+10); the key is pruned since TR <= newCursor. + c3 := poll(c2, tr.Add(ten), tr.Add(nine)) + require.True(t, c3.Equal(tr.Add(nine)), "cursor must advance to safeTS (TR+9), not clusterNow, got %s", c3) + require.Equal(t, 1, provCount, "provisional row still emitted exactly once in total") + require.Empty(t, emittedWindow, "dedup key pruned once safeTS passed TR") + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogWatchRejectsStaleCursor verifies that opening a changelog +// Watch with an afterRevision far older than the datastore's GC window fails +// fast with the standard stale-revision error, rather than silently polling +// forward from a cursor whose changelog history may have already been +// garbage collected. +func TestChangelogWatchRejectsStaleCursor(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + // A revision far older than the GC window. + staleDecimal := decimal.NewFromInt(1) // ~1970, well outside gc window + stale, err := revisions.NewForHLC(staleDecimal) + require.NoError(t, err) + + _, errchan := ds.Watch(ctx, stale, datastore.WatchOptions{ + Content: datastore.WatchRelationships | datastore.WatchCheckpoints, + CheckpointInterval: 100 * time.Millisecond, + }) + select { + case err := <-errchan: + require.Error(t, err) + require.ErrorAs(t, err, &datastore.InvalidRevisionError{}) + case <-time.After(10 * time.Second): + t.Fatal("expected a stale-revision error") + } + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogWatchSeesBulkLoad is the motivating scenario for the +// changelog-table Watch feature: CRDB changefeeds stall resolved timestamps +// during bulk loads, so the poll-based Watch must be able to observe +// bulk-loaded relationships via the changelog table instead. It bulk-loads a +// batch of relationships in a single transaction and asserts every one of +// them arrives over the poll-based watch. +func TestChangelogWatchSeesBulkLoad(t *testing.T) { + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + head, err := ds.HeadRevision(ctx) + require.NoError(t, err) + changes, errchan := ds.Watch(ctx, head.Revision, datastore.WatchOptions{ + Content: datastore.WatchRelationships | datastore.WatchCheckpoints, + CheckpointInterval: 100 * time.Millisecond, + }) + + const total = 500 + rels := make([]tuple.Relationship, 0, total) + for i := 0; i < total; i++ { + rels = append(rels, tuple.MustParse(fmt.Sprintf("document:doc%d#viewer@user:alice", i))) + } + _, err = ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error { + _, berr := rwt.BulkLoad(ctx, &testBulkSource{rels: rels}) + return berr + }) + require.NoError(t, err) + + seen := map[string]struct{}{} + deadline := time.After(60 * time.Second) + for len(seen) < total { + select { + case change, ok := <-changes: + require.True(t, ok) + for _, rc := range change.RelationshipChanges { + seen[rc.Relationship.Resource.ObjectID] = struct{}{} + } + case err := <-errchan: + require.NoError(t, err) + case <-deadline: + t.Fatalf("only saw %d/%d bulk-loaded relationships over changelog watch", len(seen), total) + } + } + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} + +// TestChangelogWatchSeesBulkLoadAcrossChunks is a regression test for +// chunking the bulk changelog INSERT in appendRelationshipChangelogBatch. +// Postgres/pgx cap bound parameters at 65535 per statement; at 14 columns +// per changelog row, a single multi-row INSERT can only hold a few thousand +// rows before it would overflow that limit. appendRelationshipChangelogBatch +// splits large batches into chunks, but must keep every row's ordinal +// unique and monotonically increasing across chunks (change_ts is constant +// for the whole transaction via cluster_logical_timestamp(), so uniqueness +// of the (change_ts, ordinal) primary key depends entirely on the ordinal). +// +// This test lowers changelogInsertChunkSize to a tiny value so a small +// bulk load spans multiple chunks, then asserts every relationship is +// still observed over the changelog Watch -- proving all chunks landed and +// no ordinals collided across chunk boundaries. +func TestChangelogWatchSeesBulkLoadAcrossChunks(t *testing.T) { + const testChunkSize = 3 + original := changelogInsertChunkSize + changelogInsertChunkSize = testChunkSize + t.Cleanup(func() { changelogInsertChunkSize = original }) + + engine := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion()) + createDatastoreTest(engine, func(t *testing.T, ds datastore.Datastore) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + head, err := ds.HeadRevision(ctx) + require.NoError(t, err) + changes, errchan := ds.Watch(ctx, head.Revision, datastore.WatchOptions{ + Content: datastore.WatchRelationships | datastore.WatchCheckpoints, + CheckpointInterval: 100 * time.Millisecond, + }) + + // 7 rows over a chunk size of 3 forces 3 chunks (3 + 3 + 1), so the + // ordinal counter must span multiple INSERT statements correctly. + const total = 7 + rels := make([]tuple.Relationship, 0, total) + for i := 0; i < total; i++ { + rels = append(rels, tuple.MustParse(fmt.Sprintf("document:chunkdoc%d#viewer@user:alice", i))) + } + _, err = ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error { + _, berr := rwt.BulkLoad(ctx, &testBulkSource{rels: rels}) + return berr + }) + require.NoError(t, err) + + seen := map[string]struct{}{} + deadline := time.After(30 * time.Second) + for len(seen) < total { + select { + case change, ok := <-changes: + require.True(t, ok) + for _, rc := range change.RelationshipChanges { + seen[rc.Relationship.Resource.ObjectID] = struct{}{} + } + case err := <-errchan: + require.NoError(t, err) + case <-deadline: + t.Fatalf("only saw %d/%d bulk-loaded relationships over changelog watch across chunks", len(seen), total) + } + } + }, ExperimentalChangelogWatch(true), WithAcquireTimeout(5*time.Second))(t) +} diff --git a/internal/datastore/crdb/changelog_test.go b/internal/datastore/crdb/changelog_test.go new file mode 100644 index 0000000000..b088dea9f0 --- /dev/null +++ b/internal/datastore/crdb/changelog_test.go @@ -0,0 +1,20 @@ +package crdb + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/authzed/spicedb/internal/datastore/crdb/schema" +) + +func TestCreateChangelogTableSQL(t *testing.T) { + sql := createChangelogTableSQL(24 * time.Hour) + require.Contains(t, sql, "CREATE TABLE IF NOT EXISTS "+schema.TableRelationshipChangelog) + require.Contains(t, sql, "PRIMARY KEY ("+schema.ColChangeTS+", "+schema.ColChangeOrdinal+") USING HASH") + require.Contains(t, sql, "ttl_expiration_expression") + require.Contains(t, sql, schema.ColChangeTTLExpiration) + require.Contains(t, sql, "ttl_disable_changefeed_replication") + require.Contains(t, sql, schema.ColChangeMetadata+" JSONB") +} diff --git a/internal/datastore/crdb/crdb.go b/internal/datastore/crdb/crdb.go index 2175cc1055..5ea634c270 100644 --- a/internal/datastore/crdb/crdb.go +++ b/internal/datastore/crdb/crdb.go @@ -144,6 +144,13 @@ func newCRDBDatastore(ctx context.Context, url string, options ...Option) (datas config.gcWindow = time.Duration(clusterTTLNanos) * time.Nanosecond } + if config.changelogWatchEnabled { + if err := ensureChangelogTable(initCtx, initPool, config.gcWindow); err != nil { + return nil, common.RedactAndLogSensitiveConnString(ctx, errUnableToInstantiate, err, url) + } + log.Ctx(initCtx).Info().Msg("experimental CRDB changelog-watch table ensured") + } + keySetInit := newKeySet var keyer overlapKeyer switch config.overlapStrategy { @@ -196,6 +203,8 @@ func newCRDBDatastore(ctx context.Context, url string, options ...Option) (datas supportsIntegrity: config.withIntegrity, gcWindow: config.gcWindow, watchEnabled: !config.watchDisabled, + changelogWatchEnabled: config.changelogWatchEnabled, + changelogWatchMaxOffset: config.changelogWatchMaxOffset, schema: *schema.Schema(config.columnOptimizationOption, config.withIntegrity, false), } ds.SetNowFunc(ds.headRevisionInternal) @@ -283,12 +292,14 @@ type crdbDatastore struct { cachedFeatures *datastore.Features // GUARDED_BY(featuresLock) featuresLock sync.Mutex - pruneGroup *errgroup.Group - ctx context.Context - cancel context.CancelFunc - filterMaximumIDCount uint16 - supportsIntegrity bool - watchEnabled bool + pruneGroup *errgroup.Group + ctx context.Context + cancel context.CancelFunc + filterMaximumIDCount uint16 + supportsIntegrity bool + watchEnabled bool + changelogWatchEnabled bool + changelogWatchMaxOffset time.Duration uniqueID atomic.Pointer[string] } @@ -343,10 +354,13 @@ func (cds *crdbDatastore) ReadWriteTx( } rwt := &crdbReadWriteTXN{ - reader, - tx, - 0, - false, + crdbReader: reader, + tx: tx, + relCountChange: 0, + hasNonExpiredDeletionChange: false, + changelogWatchEnabled: cds.changelogWatchEnabled, + gcWindow: cds.gcWindow, + changelogOrdinal: 0, } if config.SchemaHashPrecondition != "" { @@ -359,34 +373,48 @@ func (cds *crdbDatastore) ReadWriteTx( return err } - // A transaction metadata entry is required if either: - // 1) len(metadata) > 0, which requires writing the metadata provided - // 2) metadata is required to mark the transaction as not matching - // a deletion of expired relationships. - // - // A transaction is marked as such IF and only IF the operations in the transaction - // consist solely of deletions, as in that scenario, we cannot be certain in the Watch - // changefeed that the transaction is not a deletion of expired relationships performed - // by CRDB itself. This is also only necessary if both expiration and watch are enabled. metadata := config.Metadata.AsMap() - requiresMetadata := len(metadata) > 0 || (cds.watchEnabled && (config.IncludesExpiredAt || !rwt.hasNonExpiredDeletionChange)) - if requiresMetadata { - // Mark the transaction as coming from SpiceDB. See the comment in watch.go - // for why this is necessary. - metadata[spicedbTransactionKey] = true - - expiresAt := time.Now().Add(cds.gcWindow).Add(1 * time.Minute) - insertTransactionMetadata := psql.Insert(schema.TableTransactionMetadata). - Columns(schema.ColExpiresAt, schema.ColMetadata). - Values(expiresAt, metadata) - - sql, args, err := insertTransactionMetadata.ToSql() - if err != nil { - return fmt.Errorf("error building metadata insert: %w", err) + if cds.changelogWatchEnabled { + // In changelog mode, the changelog table itself replaces + // transaction_metadata entirely: it only ever contains rows + // SpiceDB explicitly wrote, so the changefeed path's TTL-delete + // disambiguation marker ($spicedbTransactionKey) is unnecessary + // here. If the caller supplied metadata, carry it as a single + // kind='metadata' changelog row; otherwise write nothing. + if len(metadata) > 0 { + if err := rwt.appendMetadataChangelog(ctx, metadata, rwt.nextChangelogOrdinal(), rwt.changelogTTL()); err != nil { + return err + } } + } else { + // A transaction metadata entry is required if either: + // 1) len(metadata) > 0, which requires writing the metadata provided + // 2) metadata is required to mark the transaction as not matching + // a deletion of expired relationships. + // + // A transaction is marked as such IF and only IF the operations in the transaction + // consist solely of deletions, as in that scenario, we cannot be certain in the Watch + // changefeed that the transaction is not a deletion of expired relationships performed + // by CRDB itself. This is also only necessary if both expiration and watch are enabled. + requiresMetadata := len(metadata) > 0 || (cds.watchEnabled && (config.IncludesExpiredAt || !rwt.hasNonExpiredDeletionChange)) + if requiresMetadata { + // Mark the transaction as coming from SpiceDB. See the comment in watch.go + // for why this is necessary. + metadata[spicedbTransactionKey] = true + + expiresAt := time.Now().Add(cds.gcWindow).Add(1 * time.Minute) + insertTransactionMetadata := psql.Insert(schema.TableTransactionMetadata). + Columns(schema.ColExpiresAt, schema.ColMetadata). + Values(expiresAt, metadata) + + sql, args, err := insertTransactionMetadata.ToSql() + if err != nil { + return fmt.Errorf("error building metadata insert: %w", err) + } - if _, err := tx.Exec(ctx, sql, args...); err != nil { - return fmt.Errorf("error writing metadata: %w", err) + if _, err := tx.Exec(ctx, sql, args...); err != nil { + return fmt.Errorf("error writing metadata: %w", err) + } } } diff --git a/internal/datastore/crdb/options.go b/internal/datastore/crdb/options.go index 7b7c66efb0..cdb6158e55 100644 --- a/internal/datastore/crdb/options.go +++ b/internal/datastore/crdb/options.go @@ -35,6 +35,8 @@ type crdbOptions struct { columnOptimizationOption common.ColumnOptimizationOption includeQueryParametersInTraces bool watchDisabled bool + changelogWatchEnabled bool + changelogWatchMaxOffset time.Duration acquireTimeout time.Duration } @@ -67,6 +69,12 @@ const ( defaultColumnOptimizationOption = common.ColumnOptimizationOptionStaticValues defaultIncludeQueryParametersInTraces = false defaultWatchDisabled = false + + // defaultChangelogWatchMaxOffset is the safety margin subtracted from the + // cluster logical clock to derive the changelog-watch completeness cursor + // (see ChangelogWatchMaxOffset). It MUST be >= the cluster's --max-offset or + // Watch can miss commits; CRDB's default max-offset is 500ms. + defaultChangelogWatchMaxOffset = 250 * time.Millisecond ) // Option provides the facility to configure how clients within the CRDB @@ -93,6 +101,7 @@ func generateConfig(options []Option) (crdbOptions, error) { columnOptimizationOption: defaultColumnOptimizationOption, includeQueryParametersInTraces: defaultIncludeQueryParametersInTraces, watchDisabled: defaultWatchDisabled, + changelogWatchMaxOffset: defaultChangelogWatchMaxOffset, acquireTimeout: defaultAcquireTimeout, } @@ -424,3 +433,34 @@ func WithWatchDisabled(isDisabled bool) Option { func WithAcquireTimeout(timeout time.Duration) Option { return func(po *crdbOptions) { po.acquireTimeout = timeout } } + +// ExperimentalChangelogWatch enables the experimental changelog-table Watch +// path for CockroachDB: writes dual-write into a relationship_changelog table +// and Watch polls that table at present time instead of using a changefeed. +// Defaults to off. +// +// This path honors the caller's datastore.EmissionStrategy: with +// datastore.EmitImmediatelyStrategy each change is emitted as soon as it is +// read (latency ≈ the poll/nudge interval), while the default +// datastore.EmitWhenCheckpointedStrategy releases changes grouped-by-revision +// with the checkpoint. Checkpoint latency is bounded by +// ChangelogWatchMaxOffset regardless of strategy. +func ExperimentalChangelogWatch(enabled bool) Option { + return func(po *crdbOptions) { po.changelogWatchEnabled = enabled } +} + +// ChangelogWatchMaxOffset sets the safety margin the experimental changelog +// Watch path subtracts from the cluster logical clock to derive its +// completeness cursor. Each poll reads the changelog at present time, then +// advances the "you have seen everything up to here" cursor (and emits its +// checkpoint) only to clusterNow - ChangelogWatchMaxOffset. Any transaction +// that has not yet become visible cluster-wide commits with a timestamp above +// this floor and is therefore caught by a later poll rather than skipped. +// +// This value MUST be >= the cluster's --max-offset setting, or Watch can miss +// commits (a straggler could commit below the cursor and never be re-read). +// CockroachDB's default --max-offset is 500ms. This defaults to 250ms; raise +// it to match or exceed your cluster's configured --max-offset. +func ChangelogWatchMaxOffset(maxOffset time.Duration) Option { + return func(po *crdbOptions) { po.changelogWatchMaxOffset = maxOffset } +} diff --git a/internal/datastore/crdb/options_test.go b/internal/datastore/crdb/options_test.go index 5fbb2066d6..f540ed01cb 100644 --- a/internal/datastore/crdb/options_test.go +++ b/internal/datastore/crdb/options_test.go @@ -65,3 +65,13 @@ func TestConfiguration(t *testing.T) { }) } } + +func TestExperimentalChangelogWatchOption(t *testing.T) { + cfg, err := generateConfig([]Option{ExperimentalChangelogWatch(true)}) + require.NoError(t, err) + require.True(t, cfg.changelogWatchEnabled) + + defaultCfg, err := generateConfig(nil) + require.NoError(t, err) + require.False(t, defaultCfg.changelogWatchEnabled, "must default to off") +} diff --git a/internal/datastore/crdb/readwrite.go b/internal/datastore/crdb/readwrite.go index eb05e6899c..b3d39cbe62 100644 --- a/internal/datastore/crdb/readwrite.go +++ b/internal/datastore/crdb/readwrite.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "time" sq "github.com/Masterminds/squirrel" "github.com/ccoveille/go-safecast/v2" @@ -52,6 +53,25 @@ type crdbReadWriteTXN struct { tx pgx.Tx relCountChange int64 hasNonExpiredDeletionChange bool + changelogWatchEnabled bool + gcWindow time.Duration + changelogOrdinal int +} + +// nextChangelogOrdinal returns the next ordinal to use for a changelog row +// appended within this transaction, and increments the counter. +// +// change_ts for changelog rows is derived from cluster_logical_timestamp(), +// which is constant for the lifetime of a CockroachDB transaction. Since the +// changelog primary key is (change_ts, ordinal), the ordinal must be unique +// across the entire transaction rather than reset per call-site; otherwise +// concurrent changelog-appending operations within the same transaction +// (e.g. relationship writes, schema writes, bulk loads) would collide on the +// same (change_ts, ordinal) pair. +func (rwt *crdbReadWriteTXN) nextChangelogOrdinal() int { + n := rwt.changelogOrdinal + rwt.changelogOrdinal++ + return n } var ( @@ -287,6 +307,11 @@ func (rwt *crdbReadWriteTXN) WriteRelationships(ctx context.Context, mutations [ bulkDeleteOr := sq.Or{} var bulkDeleteCount int64 + var changelogTTL time.Time + if rwt.changelogWatchEnabled { + changelogTTL = rwt.changelogTTL() + } + // Process the actual updates for _, mutation := range mutations { rel := mutation.Relationship @@ -355,6 +380,12 @@ func (rwt *crdbReadWriteTXN) WriteRelationships(ctx context.Context, mutations [ log.Ctx(ctx).Error().Msg("unknown operation type") return fmt.Errorf("unknown mutation operation: %v", mutation.Operation) } + + if rwt.changelogWatchEnabled { + if err := rwt.appendRelationshipChangelog(ctx, rel, mutation.Operation, rwt.nextChangelogOrdinal(), changelogTTL); err != nil { + return err + } + } } if bulkDeleteCount > 0 { @@ -505,6 +536,19 @@ func (rwt *crdbReadWriteTXN) LegacyWriteNamespaces(ctx context.Context, newConfi return fmt.Errorf(errUnableToWriteConfig, err) } + if rwt.changelogWatchEnabled { + ttl := rwt.changelogTTL() + for _, newConfig := range newConfigs { + serialized, err := newConfig.MarshalVT() + if err != nil { + return fmt.Errorf(errUnableToWriteConfig, err) + } + if err := rwt.appendSchemaChangelog(ctx, "namespace", newConfig.Name, serialized, rwt.nextChangelogOrdinal(), ttl); err != nil { + return err + } + } + } + return nil } @@ -545,6 +589,15 @@ func (rwt *crdbReadWriteTXN) LegacyDeleteNamespaces(ctx context.Context, nsNames return fmt.Errorf(errUnableToDeleteConfig, err) } + if rwt.changelogWatchEnabled { + ttl := rwt.changelogTTL() + for _, nsName := range nsNames { + if err := rwt.appendSchemaChangelog(ctx, "namespace", nsName, nil, rwt.nextChangelogOrdinal(), ttl); err != nil { + return err + } + } + } + if delOption == datastore.DeleteNamespacesAndRelationships { deleteTupleSQL, deleteTupleArgs, err := rwt.queryDeleteTuples(nil).Where(sq.Or(tplClauses)).ToSql() if err != nil { @@ -593,11 +646,29 @@ var copyColsWithIntegrity = []string{ func (rwt *crdbReadWriteTXN) BulkLoad(ctx context.Context, iter datastore.BulkWriteRelationshipSource) (uint64, error) { rwt.hasNonExpiredDeletionChange = true + cols := copyCols if rwt.withIntegrity { - return pgxcommon.BulkLoad(ctx, rwt.tx, rwt.schema.RelationshipTableName, copyColsWithIntegrity, iter) + cols = copyColsWithIntegrity + } + + if !rwt.changelogWatchEnabled { + return pgxcommon.BulkLoad(ctx, rwt.tx, rwt.schema.RelationshipTableName, cols, iter) } - return pgxcommon.BulkLoad(ctx, rwt.tx, rwt.schema.RelationshipTableName, copyCols, iter) + // CRDB changefeeds stall resolved timestamps during bulk loads, which is + // the motivating reason the changelog-table Watch exists. When it is + // enabled, tee every bulk-loaded relationship into the changelog (in the + // same transaction as the COPY) so the poll-based Watch can observe + // bulk-loaded data. + captured := &capturingBulkSource{inner: iter} + loaded, err := pgxcommon.BulkLoad(ctx, rwt.tx, rwt.schema.RelationshipTableName, cols, captured) + if err != nil { + return loaded, err + } + if err := rwt.appendRelationshipChangelogBatch(ctx, captured.seen, rwt.changelogTTL()); err != nil { + return loaded, err + } + return loaded, nil } var _ datastore.ReadWriteTransaction = &crdbReadWriteTXN{} diff --git a/internal/datastore/crdb/schema/schema.go b/internal/datastore/crdb/schema/schema.go index e1878e8e93..82a2c65eb4 100644 --- a/internal/datastore/crdb/schema/schema.go +++ b/internal/datastore/crdb/schema/schema.go @@ -7,13 +7,14 @@ import ( ) const ( - TableNamespace = "namespace_config" - TableTuple = "relation_tuple" - TableTupleWithIntegrity = "relation_tuple_with_integrity" - TableTransactions = "transactions" - TableCaveat = "caveat" - TableRelationshipCounter = "relationship_counter" - TableTransactionMetadata = "transaction_metadata" + TableNamespace = "namespace_config" + TableTuple = "relation_tuple" + TableTupleWithIntegrity = "relation_tuple_with_integrity" + TableTransactions = "transactions" + TableCaveat = "caveat" + TableRelationshipCounter = "relationship_counter" + TableTransactionMetadata = "transaction_metadata" + TableRelationshipChangelog = "relationship_changelog" ColNamespace = "namespace" ColConfig = "serialized_config" @@ -42,6 +43,17 @@ const ( ColCounterUpdatedAt = "updated_at_timestamp" ColExpiresAt = "expires_at" ColMetadata = "metadata" + + ColChangeTS = "change_ts" + ColChangeOrdinal = "ordinal" + ColChangeKind = "kind" + ColChangeRelExpiration = "rel_expiration" + ColChangeOperation = "operation" + ColChangeSchemaKind = "schema_kind" + ColChangeDefinitionName = "definition_name" + ColChangeSerializedDefinition = "serialized_definition" + ColChangeTTLExpiration = "ttl_expiration" + ColChangeMetadata = "metadata" ) func Schema(colOptimizationOpt common.ColumnOptimizationOption, withIntegrity bool, expirationDisabled bool) *common.SchemaInformation { diff --git a/internal/datastore/crdb/watch.go b/internal/datastore/crdb/watch.go index 4fbaddce89..4d89228dfb 100644 --- a/internal/datastore/crdb/watch.go +++ b/internal/datastore/crdb/watch.go @@ -98,7 +98,11 @@ func (cds *crdbDatastore) Watch(ctx context.Context, afterRevision datastore.Rev return updates, errs } - go cds.watch(ctx, afterRevision, options, updates, errs) + if cds.changelogWatchEnabled { + go cds.watchViaChangelog(ctx, afterRevision, options, updates, errs) + } else { + go cds.watch(ctx, afterRevision, options, updates, errs) + } return updates, errs } diff --git a/internal/datastore/crdb/watch_changelog.go b/internal/datastore/crdb/watch_changelog.go new file mode 100644 index 0000000000..2f3957290f --- /dev/null +++ b/internal/datastore/crdb/watch_changelog.go @@ -0,0 +1,671 @@ +package crdb + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/shopspring/decimal" + + "github.com/authzed/spicedb/internal/datastore/common" + "github.com/authzed/spicedb/internal/datastore/crdb/pool" + "github.com/authzed/spicedb/internal/datastore/crdb/schema" + pgxcommon "github.com/authzed/spicedb/internal/datastore/postgres/common" + "github.com/authzed/spicedb/internal/datastore/revisions" + log "github.com/authzed/spicedb/internal/logging" + "github.com/authzed/spicedb/pkg/datastore" + core "github.com/authzed/spicedb/pkg/proto/core/v1" + "github.com/authzed/spicedb/pkg/spiceerrors" + "github.com/authzed/spicedb/pkg/tuple" +) + +// changelogSelectColumns is the ordered column list read from the changelog on +// each poll. It must match the scan order in accumulateChangelogRows. +const changelogSelectColumns = schema.ColChangeTS + ", " + schema.ColChangeKind + ", " + + schema.ColNamespace + ", " + schema.ColObjectID + ", " + schema.ColRelation + ", " + + schema.ColUsersetNamespace + ", " + schema.ColUsersetObjectID + ", " + schema.ColUsersetRelation + ", " + + schema.ColCaveatContextName + ", " + schema.ColCaveatContext + ", " + + schema.ColChangeRelExpiration + ", " + schema.ColChangeOperation + ", " + + schema.ColChangeSchemaKind + ", " + schema.ColChangeDefinitionName + ", " + schema.ColChangeSerializedDefinition + ", " + + schema.ColChangeMetadata + +// changelogImmediateSelectColumns is changelogSelectColumns prefixed with the +// row ordinal, used only by the immediate-emission path so it can dedup on the +// (change_ts, ordinal) primary key. It must match the scan order in +// accumulateChangelogRowsImmediate. +const changelogImmediateSelectColumns = schema.ColChangeOrdinal + ", " + changelogSelectColumns + +// changelogRowKey identifies a single changelog row by its primary key +// (change_ts, ordinal). It is used as the dedup-set key for the provisional +// window in immediate-emission mode. +type changelogRowKey struct { + changeTS string + ordinal int64 +} + +// watchViaChangelog serves Watch by repeatedly polling the append-only +// changelog table at present time, instead of consuming a CRDB changefeed. +// +// Each poll runs a read-only transaction and reads cluster_logical_timestamp() +// (clusterNow) once. Because the changelog is append-only, a present read +// already observes every row committed with change_ts <= clusterNow regardless +// of changefeed health -- no AS OF SYSTEM TIME is needed -- which is what lets +// this path survive bulk loads. +// +// Correctness (no lost updates) lives in the COMPLETENESS CURSOR, not in the +// individual change rows. A checkpoint at revision C promises "you have seen +// every change <= C", and that promise is only safe once no straggler can +// still commit at change_ts <= C. Under the cluster's --max-offset clock skew +// bound that safe point is safeTS = clusterNow - maxOffset. The cursor (the +// lower bound carried to the next poll, and the value returned as the new +// cursor) therefore advances ONLY to safeTS, never to clusterNow: a +// transaction not yet visible cluster-wide commits above safeTS and is caught +// by a later poll rather than skipped. +// +// Emission strategy is honored: +// - datastore.EmitImmediatelyStrategy: each row in (cursor, clusterNow] is +// emitted the instant it is read (change latency ≈ the poll/nudge +// interval), deduped within the provisional window (safeTS, clusterNow] via +// the (change_ts, ordinal) primary key. Duplicates/reordering across polls +// are permitted by the Watch contract and minimized by that dedup set. +// - datastore.EmitWhenCheckpointedStrategy (default): only (cursor, safeTS] +// is read, buffered/deduped via common.NewChanges and released +// grouped-by-revision with the checkpoint -- never early, never reordered. +// +// Either way a checkpoint is emitted at safeTS and the cursor advances to +// safeTS. Checkpoints advance on the ticker even when idle, since clusterNow +// (and thus safeTS) is re-read every poll. +func (cds *crdbDatastore) watchViaChangelog( + ctx context.Context, + afterRevision datastore.Revision, + opts datastore.WatchOptions, + updates chan datastore.RevisionChanges, + errs chan error, +) { + defer close(updates) + defer close(errs) + + watchConnectTimeout := opts.WatchConnectTimeout + if watchConnectTimeout <= 0 { + watchConnectTimeout = cds.watchConnectTimeout + } + + // Use a dedicated, non-pooled connection for watch, mirroring the changefeed + // path. + conn, err := pgxcommon.ConnectWithInstrumentationAndTimeout(ctx, cds.dburl, watchConnectTimeout) + if err != nil { + errs <- err + return + } + defer func() { go func() { _ = conn.Close(ctx) }() }() + + interval := 1 * time.Second + if opts.CheckpointInterval > 0 { + interval = opts.CheckpointInterval + } + + watchBufferWriteTimeout := opts.WatchBufferWriteTimeout + if watchBufferWriteTimeout <= 0 { + watchBufferWriteTimeout = cds.watchBufferWriteTimeout + } + + sendChange := func(change datastore.RevisionChanges) error { + select { + case updates <- change: + return nil + default: + // If we cannot immediately write, set up the timer and try again. + } + + timer := time.NewTimer(watchBufferWriteTimeout) + defer timer.Stop() + + select { + case updates <- change: + return nil + case <-timer.C: + return datastore.NewWatchDisconnectedErr() + } + } + + sendError := func(err error) { + if errors.Is(ctx.Err(), context.Canceled) { + errs <- datastore.NewWatchCanceledErr() + return + } + + if strings.Contains(err.Error(), "must be after replica GC threshold") { + errs <- datastore.NewInvalidRevisionErr(afterRevision, datastore.RevisionStale) + return + } + + if pool.IsResettableError(ctx, err) || pool.IsRetryableError(ctx, err) { + errs <- datastore.NewWatchTemporaryErr(err) + return + } + + errs <- err + } + + // The cursor is the last revision we have fully emitted; the next poll reads + // strictly after it. + hlcAfter, ok := afterRevision.(revisions.HLCRevision) + if !ok { + sendError(spiceerrors.MustBugf("expected HLCRevision for changelog watch, got %T", afterRevision)) + return + } + cursor, err := hlcAfter.AsDecimal() + if err != nil { + sendError(fmt.Errorf("invalid afterRevision for changelog watch: %w", err)) + return + } + + // Guard against a cursor older than the changelog's GC/TTL window: rows + // covering that span may already have been reaped, so polling forward + // from it could silently skip history rather than replay it. Fail fast + // with the same stale-revision error the changefeed watch path surfaces + // (via the "must be after replica GC threshold" mapping in sendError + // above) instead of relying solely on the present read to notice. + cursorTime := time.Unix(0, hlcAfter.TimestampNanoSec()) + if time.Since(cursorTime) > cds.gcWindow { + sendError(datastore.NewInvalidRevisionErr(afterRevision, datastore.RevisionStale)) + return + } + + watchBufferSize := opts.MaximumBufferedChangesByteSize + if watchBufferSize == 0 { + watchBufferSize = cds.watchChangeBufferMaximumSize + } + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + // The nudge is a purely optional low-latency wake for the poll loop: a + // lightweight changefeed on the changelog table whose payload we never + // inspect. Any row event is a signal to poll sooner than the next tick. + // The ticker above remains the correctness backstop -- if the nudge + // changefeed stalls or fails to start (e.g. during the exact bulk-load + // scenario this feature routes around), polling still advances on the + // timer, so nothing here may affect correctness, only latency. + nudge := make(chan struct{}, 1) + go cds.runChangelogNudge(ctx, nudge) + + immediate := opts.EmissionStrategy == datastore.EmitImmediatelyStrategy + + // emittedWindow is the provisional-window dedup set for immediate mode: + // (change_ts, ordinal) keys already emitted in a poll whose clusterNow ran + // ahead of the cursor. Entries are pruned once the cursor advances past + // their change_ts (see pollChangelogImmediate). It is unused in buffered + // mode. + emittedWindow := make(map[changelogRowKey]struct{}) + + maxOffsetNanos := cds.changelogWatchMaxOffset.Nanoseconds() + + for { + newCursor, err := cds.pollChangelogOnce(ctx, conn, opts, cursor, maxOffsetNanos, watchBufferSize, immediate, emittedWindow, sendChange) + if err != nil { + sendError(err) + return + } + cursor = newCursor + + select { + case <-ctx.Done(): + sendError(ctx.Err()) + return + case <-ticker.C: + case <-nudge: + } + } +} + +// runChangelogNudge opens a best-effort CRDB changefeed on the changelog +// table purely to wake watchViaChangelog's poll loop early on new activity. +// The changefeed payload is never parsed -- any row event is a signal to +// poll sooner. This is strictly a latency optimization: the poll loop's +// ticker is the correctness backstop, so any failure here is logged and the +// goroutine simply exits without affecting the watch. +func (cds *crdbDatastore) runChangelogNudge(ctx context.Context, nudge chan<- struct{}) { + conn, err := pgxcommon.ConnectWithInstrumentationAndTimeout(ctx, cds.dburl, cds.watchConnectTimeout) + if err != nil { + log.Ctx(ctx).Warn().Err(err).Msg("changelog nudge changefeed unavailable; falling back to interval polling") + return + } + defer func() { _ = conn.Close(ctx) }() + + head, err := cds.HeadRevision(ctx) + if err != nil { + log.Ctx(ctx).Warn().Err(err).Msg("changelog nudge changefeed could not determine head revision; falling back to interval polling") + return + } + + query := fmt.Sprintf(cds.beginChangefeedQuery, schema.TableRelationshipChangelog, head.Revision, "1s") + rows, err := conn.Query(ctx, query) + if err != nil { + log.Ctx(ctx).Warn().Err(err).Msg("changelog nudge changefeed failed to start; falling back to interval polling") + return + } + defer rows.Close() + + for rows.Next() { + // Non-blocking: the poll loop only needs to know that *something* + // happened, not how many times. If a nudge is already pending, drop + // this one rather than blocking the changefeed read loop. + select { + case nudge <- struct{}{}: + default: + } + } + if err := rows.Err(); err != nil && !errors.Is(ctx.Err(), context.Canceled) { + log.Ctx(ctx).Warn().Err(err).Msg("changelog nudge changefeed ended with an error; falling back to interval polling") + } +} + +// safeTSFromClusterNow returns clusterNow - maxOffsetNanos. clusterNow is an +// HLC decimal whose integer part is nanoseconds since the Unix epoch (see +// HLCRevision.TimestampNanoSec and the HLCRevision decimal layout), so +// subtracting maxOffsetNanos (also in nanoseconds) yields the completeness +// floor safeTS at the same logical position. Subtracting from the decimal +// directly preserves the fractional logical-clock component of clusterNow. +func safeTSFromClusterNow(clusterNow decimal.Decimal, maxOffsetNanos int64) decimal.Decimal { + return clusterNow.Sub(decimal.NewFromInt(maxOffsetNanos)) +} + +// pollChangelogOnce runs one poll against the append-only changelog. It reads +// cluster_logical_timestamp() (clusterNow) in a single read-only transaction, +// derives safeTS = clusterNow - maxOffset, emits changes and (when requested) a +// checkpoint at safeTS, and returns the new cursor. +// +// INV1 (no lost updates): the returned cursor advances only to safeTS, never to +// clusterNow. Immediate mode reads up to clusterNow for low change latency, but +// anything in (safeTS, clusterNow] stays > cursor and is re-read next poll (so +// stragglers committing late with a low change_ts are never skipped); the +// provisional-window dedup set keeps that re-read from re-delivering a row the +// client already saw. +func (cds *crdbDatastore) pollChangelogOnce( + ctx context.Context, + conn *pgx.Conn, + opts datastore.WatchOptions, + cursor decimal.Decimal, + maxOffsetNanos int64, + watchBufferSize uint64, + immediate bool, + emittedWindow map[changelogRowKey]struct{}, + sendChange sendChangeFunc, +) (decimal.Decimal, error) { + tx, err := conn.BeginTx(ctx, pgx.TxOptions{AccessMode: pgx.ReadOnly}) + if err != nil { + return cursor, err + } + defer func() { _ = tx.Rollback(ctx) }() + + // Read the cluster clock and the changelog rows from the same read-only + // transaction so clusterNow and the rows come from one consistent point. + // No AS OF SYSTEM TIME: the changelog is append-only, so a present read + // already sees every row with change_ts <= clusterNow regardless of + // changefeed health. + var clusterNow decimal.Decimal + if err := tx.QueryRow(ctx, "SELECT cluster_logical_timestamp()").Scan(&clusterNow); err != nil { + return cursor, err + } + + safeTS := safeTSFromClusterNow(clusterNow, maxOffsetNanos) + + var newCursor decimal.Decimal + if immediate { + newCursor, err = cds.pollChangelogImmediate(ctx, tx, opts, cursor, clusterNow, safeTS, emittedWindow, sendChange) + } else { + newCursor, err = cds.pollChangelogBuffered(ctx, tx, opts, cursor, safeTS, watchBufferSize, sendChange) + } + if err != nil { + return cursor, err + } + + if err := tx.Commit(ctx); err != nil { + return cursor, err + } + + return newCursor, nil +} + +// pollChangelogImmediate implements EmitImmediatelyStrategy: every changelog +// row in (cursor, clusterNow] is emitted the instant it is read, so change +// latency is ≈ the poll/nudge interval rather than the checkpoint interval. +// +// Because clusterNow > safeTS, this poll may read rows in the provisional +// window (safeTS, clusterNow] that a later poll will re-read (the cursor only +// advances to safeTS). emittedWindow tracks the (change_ts, ordinal) keys +// already delivered so those re-reads do not re-emit them; it is pruned of +// keys <= the new cursor once the cursor advances. Duplicates/reordering are +// permitted by the Watch contract for immediate mode, but this keeps them +// minimal. +// +// The cursor advances only to max(cursor, safeTS) (INV1). A checkpoint is +// emitted at safeTS once safeTS > cursor. +func (cds *crdbDatastore) pollChangelogImmediate( + ctx context.Context, + tx pgx.Tx, + opts datastore.WatchOptions, + cursor decimal.Decimal, + clusterNow decimal.Decimal, + safeTS decimal.Decimal, + emittedWindow map[changelogRowKey]struct{}, + sendChange sendChangeFunc, +) (decimal.Decimal, error) { + // Emit changes as they are read, preserving Create (no touch normalization) + // and per-row ordering. streamingChangeProvider emits on each Add* call. + streamer := &streamingChangeProvider{ + sendChange: sendChange, + content: opts.Content, + } + + if clusterNow.GreaterThan(cursor) { + query := fmt.Sprintf( + "SELECT %s FROM %s WHERE %s > $1 AND %s <= $2 ORDER BY %s, %s", + changelogImmediateSelectColumns, schema.TableRelationshipChangelog, + schema.ColChangeTS, schema.ColChangeTS, schema.ColChangeTS, schema.ColChangeOrdinal, + ) + rows, err := tx.Query(ctx, query, cursor, clusterNow) + if err != nil { + return cursor, err + } + if err := cds.accumulateChangelogRowsImmediate(ctx, rows, streamer, emittedWindow); err != nil { + return cursor, err + } + } + + // Advance the completeness cursor only to safeTS, then checkpoint there. + newCursor := cursor + if safeTS.GreaterThan(cursor) { + newCursor = safeTS + if err := cds.emitChangelogCheckpoint(opts, safeTS, sendChange); err != nil { + return cursor, err + } + } + + // Prune the dedup set of any key whose change_ts is now <= the cursor: those + // rows will never be re-read (the next poll's lower bound is > newCursor), so + // they can no longer be duplicated. + for key := range emittedWindow { + keyTS, err := decimal.NewFromString(key.changeTS) + if err != nil { + return cursor, err + } + if keyTS.LessThanOrEqual(newCursor) { + delete(emittedWindow, key) + } + } + + return newCursor, nil +} + +// pollChangelogBuffered implements the default EmitWhenCheckpointedStrategy: it +// reads only the settled window (cursor, safeTS], buffers/dedups it via +// common.NewChanges, and releases it grouped-by-revision, in revision order, +// with the checkpoint at safeTS. Nothing is emitted early or out of order +// (INV4). +func (cds *crdbDatastore) pollChangelogBuffered( + ctx context.Context, + tx pgx.Tx, + opts datastore.WatchOptions, + cursor decimal.Decimal, + safeTS decimal.Decimal, + watchBufferSize uint64, + sendChange sendChangeFunc, +) (decimal.Decimal, error) { + // safeTS has not yet advanced past the cursor (e.g. just started at HEAD, or + // idle within one maxOffset window). Emit nothing but keep the idle-checkpoint + // behavior: a checkpoint at the (unadvanced) cursor lets consumers know the + // stream is live without regressing completeness. + if safeTS.LessThanOrEqual(cursor) { + return cursor, cds.emitChangelogCheckpoint(opts, cursor, sendChange) + } + + if err := cds.pollChangelogRange(ctx, tx, opts, cursor, safeTS, watchBufferSize, sendChange); err != nil { + return cursor, err + } + return safeTS, nil +} + +// pollChangelogRange reads all changelog rows in (cursor, target] using the +// provided transaction, emits them grouped by revision, and emits a checkpoint +// at target when checkpoints are requested. target is passed in explicitly so +// tests can pin it to a known value; production passes safeTS. +// +// Emission is INCLUSIVE of target: the SQL bound is change_ts <= target and the +// tracker is created fresh for this range, so every accumulated row is +// guaranteed to satisfy cursor < change_ts <= target. We therefore emit +// everything accumulated (AsRevisionChanges) rather than the strict "< target" +// filter, which would drop any row whose change_ts exactly equals target. +// +// Callers must ensure target > cursor; the idle/just-started case is handled by +// pollChangelogBuffered. +func (cds *crdbDatastore) pollChangelogRange( + ctx context.Context, + tx pgx.Tx, + opts datastore.WatchOptions, + cursor decimal.Decimal, + target decimal.Decimal, + watchBufferSize uint64, + sendChange sendChangeFunc, +) error { + tracked := common.NewChanges(revisions.HLCKeyFunc, opts.Content, watchBufferSize) + + query := fmt.Sprintf( + "SELECT %s FROM %s WHERE %s > $1 AND %s <= $2 ORDER BY %s, %s", + changelogSelectColumns, schema.TableRelationshipChangelog, + schema.ColChangeTS, schema.ColChangeTS, schema.ColChangeTS, schema.ColChangeOrdinal, + ) + rows, err := tx.Query(ctx, query, cursor, target) + if err != nil { + return err + } + if err := cds.accumulateChangelogRows(ctx, rows, tracked); err != nil { + return err + } + + // Emit every accumulated change in ascending revision order. Every row read + // is by construction in (cursor, target], so emitting all of them is exactly + // the (cursor, target] window, inclusive of target. + for revChange, err := range tracked.AsRevisionChanges(revisions.HLCKeyLessThanFunc) { + if err != nil { + return err + } + if err := sendChange(revChange); err != nil { + return err + } + } + + return cds.emitChangelogCheckpoint(opts, target, sendChange) +} + +// emitChangelogCheckpoint emits a checkpoint at target if WatchCheckpoints is set. +func (cds *crdbDatastore) emitChangelogCheckpoint(opts datastore.WatchOptions, target decimal.Decimal, sendChange sendChangeFunc) error { + if opts.Content&datastore.WatchCheckpoints != datastore.WatchCheckpoints { + return nil + } + rev, err := revisions.NewForHLC(target) + if err != nil { + return err + } + return sendChange(datastore.RevisionChanges{Revision: rev, IsCheckpoint: true}) +} + +// changelogRowFields holds the reconstructed non-key columns of a changelog +// row, shared by the buffered and immediate scan paths. +type changelogRowFields struct { + kind string + nsName, objectID, relation *string + usNs, usObjectID, usRelation *string + caveatName *string + caveatContext map[string]any + relExpiration *time.Time + operation *string + schemaKind, definitionName *string + serializedDefinition []byte + metadata map[string]any +} + +// feedChangelogRow reconstructs one changelog row and feeds it to the tracker. +// +// When normalizeCreate is true, a "create" operation is reported as Touch. The +// shared change tracker (internal/datastore/common.Changes, used by the +// buffered path and by the Postgres/MySQL watch implementations) only +// distinguishes Touch vs Delete — at the row level a physical insert is +// indistinguishable from a touch of a previously nonexistent row — so those +// backends report inserts as Touch. The immediate path emits through +// streamingChangeProvider, which does represent Create distinctly, so it passes +// normalizeCreate=false to preserve it. +func feedChangelogRow(ctx context.Context, tracked changeTracker[revisions.HLCRevision, revisions.HLCRevision], rev revisions.HLCRevision, f changelogRowFields, normalizeCreate bool) error { + switch f.kind { + case "rel": + ctxCaveat, err := common.ContextualizedCaveatFrom(deref(f.caveatName), f.caveatContext) + if err != nil { + return err + } + rel := tuple.Relationship{ + RelationshipReference: tuple.RelationshipReference{ + Resource: tuple.ObjectAndRelation{ObjectType: deref(f.nsName), ObjectID: deref(f.objectID), Relation: deref(f.relation)}, + Subject: tuple.ObjectAndRelation{ObjectType: deref(f.usNs), ObjectID: deref(f.usObjectID), Relation: deref(f.usRelation)}, + }, + OptionalCaveat: ctxCaveat, + OptionalExpiration: f.relExpiration, + } + op, err := changelogOperation(deref(f.operation)) + if err != nil { + return err + } + if normalizeCreate && op == tuple.UpdateOperationCreate { + op = tuple.UpdateOperationTouch + } + return tracked.AddRelationshipChange(ctx, rev, rel, op) + case "schema": + if f.serializedDefinition != nil { + switch deref(f.schemaKind) { + case "namespace": + def := &core.NamespaceDefinition{} + if err := def.UnmarshalVT(f.serializedDefinition); err != nil { + return err + } + return tracked.AddChangedDefinition(ctx, rev, def) + case "caveat": + def := &core.CaveatDefinition{} + if err := def.UnmarshalVT(f.serializedDefinition); err != nil { + return err + } + return tracked.AddChangedDefinition(ctx, rev, def) + default: + return spiceerrors.MustBugf("unknown schema_kind in changelog: %s", deref(f.schemaKind)) + } + } + switch deref(f.schemaKind) { + case "namespace": + return tracked.AddDeletedNamespace(ctx, rev, deref(f.definitionName)) + case "caveat": + return tracked.AddDeletedCaveat(ctx, rev, deref(f.definitionName)) + default: + return spiceerrors.MustBugf("unknown schema_kind in changelog: %s", deref(f.schemaKind)) + } + case "metadata": + // Metadata emission is independent of opts.Content, matching the + // changefeed path (AddRevisionMetadata is not content-gated). We never + // wrote $spicedbTransactionKey in changelog mode, so unlike the + // changefeed path there is no key to strip here. + if len(f.metadata) > 0 { + return tracked.AddRevisionMetadata(ctx, rev, f.metadata) + } + return nil + default: + return spiceerrors.MustBugf("unknown changelog kind: %s", f.kind) + } +} + +// accumulateChangelogRows scans changelog rows into the change tracker. The +// tracker itself applies the WatchContent filtering, so both relationship and +// schema rows are always fed in. Used by the buffered path, so create is +// normalized to touch. +func (cds *crdbDatastore) accumulateChangelogRows(ctx context.Context, rows pgx.Rows, tracked changeTracker[revisions.HLCRevision, revisions.HLCRevision]) error { + defer rows.Close() + for rows.Next() { + var changeTS decimal.Decimal + var f changelogRowFields + if err := rows.Scan(&changeTS, &f.kind, &f.nsName, &f.objectID, &f.relation, + &f.usNs, &f.usObjectID, &f.usRelation, &f.caveatName, &f.caveatContext, + &f.relExpiration, &f.operation, &f.schemaKind, &f.definitionName, &f.serializedDefinition, + &f.metadata); err != nil { + return err + } + + rev, err := revisions.NewForHLC(changeTS) + if err != nil { + return err + } + if err := feedChangelogRow(ctx, tracked, rev, f, true); err != nil { + return err + } + } + return rows.Err() +} + +// accumulateChangelogRowsImmediate scans changelog rows (prefixed with the +// ordinal, per changelogImmediateSelectColumns) and emits each through the +// streaming provider the instant it is read, skipping any (change_ts, ordinal) +// already delivered in the provisional window. Newly-emitted keys are recorded +// in emittedWindow; the caller prunes it as the cursor advances. Create is +// preserved (normalizeCreate=false). +func (cds *crdbDatastore) accumulateChangelogRowsImmediate(ctx context.Context, rows pgx.Rows, streamer changeTracker[revisions.HLCRevision, revisions.HLCRevision], emittedWindow map[changelogRowKey]struct{}) error { + defer rows.Close() + for rows.Next() { + var ordinal int64 + var changeTS decimal.Decimal + var f changelogRowFields + if err := rows.Scan(&ordinal, &changeTS, &f.kind, &f.nsName, &f.objectID, &f.relation, + &f.usNs, &f.usObjectID, &f.usRelation, &f.caveatName, &f.caveatContext, + &f.relExpiration, &f.operation, &f.schemaKind, &f.definitionName, &f.serializedDefinition, + &f.metadata); err != nil { + return err + } + + key := changelogRowKey{changeTS: changeTS.String(), ordinal: ordinal} + if _, ok := emittedWindow[key]; ok { + // Already delivered in a previous poll's provisional window. + continue + } + + rev, err := revisions.NewForHLC(changeTS) + if err != nil { + return err + } + if err := feedChangelogRow(ctx, streamer, rev, f, false); err != nil { + return err + } + emittedWindow[key] = struct{}{} + } + return rows.Err() +} + +func deref(s *string) string { + if s == nil { + return "" + } + return *s +} + +// changelogOperation maps the stored operation string back to a +// tuple.UpdateOperation. Any unrecognized value is fatal; there is no valid +// zero/unknown operation, so callers must treat a non-nil error as fatal. +func changelogOperation(s string) (tuple.UpdateOperation, error) { + switch s { + case "create": + return tuple.UpdateOperationCreate, nil + case "touch": + return tuple.UpdateOperationTouch, nil + case "delete": + return tuple.UpdateOperationDelete, nil + default: + return tuple.UpdateOperationTouch, fmt.Errorf("unknown changelog operation: %s", s) + } +} diff --git a/pkg/cmd/datastore/datastore.go b/pkg/cmd/datastore/datastore.go index 6ba677efc9..0f2c060512 100644 --- a/pkg/cmd/datastore/datastore.go +++ b/pkg/cmd/datastore/datastore.go @@ -146,13 +146,15 @@ type Config struct { RequestHedgingQuantile float64 `debugmap:"visible"` // CRDB - FollowerReadDelay time.Duration `debugmap:"visible" default:"4800ms"` - MaxRetries int `debugmap:"visible" default:"10"` - OverlapKey string `debugmap:"visible" default:"key"` - OverlapStrategy string `debugmap:"visible" default:"static"` - EnableConnectionBalancing bool `debugmap:"visible" default:"true"` - ConnectRate time.Duration `debugmap:"visible" default:"100ms"` - WriteAcquisitionTimeout time.Duration `debugmap:"visible" default:"30ms"` + FollowerReadDelay time.Duration `debugmap:"visible" default:"4800ms"` + MaxRetries int `debugmap:"visible" default:"10"` + OverlapKey string `debugmap:"visible" default:"key"` + OverlapStrategy string `debugmap:"visible" default:"static"` + EnableConnectionBalancing bool `debugmap:"visible" default:"true"` + ConnectRate time.Duration `debugmap:"visible" default:"100ms"` + WriteAcquisitionTimeout time.Duration `debugmap:"visible" default:"30ms"` + ExperimentalCRDBChangelogWatch bool `debugmap:"visible" default:"false"` + ChangelogWatchMaxOffset time.Duration `debugmap:"visible" default:"250ms"` // Postgres GCInterval time.Duration `debugmap:"visible" default:"3m"` @@ -347,6 +349,8 @@ func RegisterDatastoreFlagsWithPrefix(flagSet *pflag.FlagSet, prefix string, opt flagSet.DurationVar(&opts.FollowerReadDelay, flagName("datastore-follower-read-delay-duration"), DefaultFollowerReadDelay, "amount of time to subtract from non-sync revision timestamps to ensure they are sufficiently in the past to enable follower reads (CockroachDB and Spanner drivers only) or read replicas (Postgres and MySQL drivers only)") flagSet.IntVar(&opts.MaxRetries, flagName("datastore-max-tx-retries"), 10, "number of times a retriable transaction should be retried") flagSet.StringVar(&opts.OverlapStrategy, flagName("datastore-tx-overlap-strategy"), "static", "strategy to generate transaction overlap keys (\"request\", \"prefix\", \"static\", \"insecure\") (CockroachDB driver only - see "+sharederrors.CrdbOverlapErrorLink+" for details)") + flagSet.BoolVar(&opts.ExperimentalCRDBChangelogWatch, flagName("datastore-experimental-crdb-changelog-watch"), false, "EXPERIMENTAL: use a dual-written changelog table polled at a closed timestamp for Watch instead of a changefeed (CockroachDB driver only). Keeps Watch advancing during bulk loads at the cost of write latency.") + flagSet.DurationVar(&opts.ChangelogWatchMaxOffset, flagName("datastore-experimental-crdb-changelog-watch-max-offset"), 250*time.Millisecond, "EXPERIMENTAL: safety margin subtracted from the cluster clock to derive the changelog Watch completeness cursor/checkpoint; MUST be >= the cluster's --max-offset (CockroachDB default 500ms) or Watch can miss commits (CockroachDB driver only)") flagSet.StringVar(&opts.OverlapKey, flagName("datastore-tx-overlap-key"), "key", "static key to touch when writing to ensure transactions overlap (only used if --datastore-tx-overlap-strategy=static is set; CockroachDB driver only)") flagSet.BoolVar(&opts.EnableConnectionBalancing, flagName("datastore-connection-balancing"), defaults.EnableConnectionBalancing, "enable connection balancing between database nodes (CockroachDB driver only)") flagSet.DurationVar(&opts.ConnectRate, flagName("datastore-connect-rate"), 100*time.Millisecond, "rate at which new connections are allowed to the datastore (at a rate of 1/duration) (CockroachDB driver only)") @@ -425,6 +429,8 @@ func DefaultDatastoreConfig() *Config { MaxRetries: 10, OverlapKey: "key", OverlapStrategy: "static", + ExperimentalCRDBChangelogWatch: false, + ChangelogWatchMaxOffset: 250 * time.Millisecond, ConnectRate: 100 * time.Millisecond, EnableConnectionBalancing: true, GCInterval: 3 * time.Minute, @@ -657,6 +663,8 @@ func newCRDBDatastore(ctx context.Context, opts Config) (datastore.Datastore, er crdb.WithColumnOptimization(opts.ExperimentalColumnOptimization), crdb.IncludeQueryParametersInTraces(opts.IncludeQueryParametersInTraces), crdb.WithWatchDisabled(opts.DisableWatchSupport), + crdb.ExperimentalChangelogWatch(opts.ExperimentalCRDBChangelogWatch), + crdb.ChangelogWatchMaxOffset(opts.ChangelogWatchMaxOffset), ) } diff --git a/pkg/cmd/datastore/zz_generated.options.go b/pkg/cmd/datastore/zz_generated.options.go index 9d814ebf72..1743d2727e 100644 --- a/pkg/cmd/datastore/zz_generated.options.go +++ b/pkg/cmd/datastore/zz_generated.options.go @@ -66,6 +66,7 @@ func (c *Config) ToOption() ConfigOption { to.EnableConnectionBalancing = c.EnableConnectionBalancing to.ConnectRate = c.ConnectRate to.WriteAcquisitionTimeout = c.WriteAcquisitionTimeout + to.ExperimentalCRDBChangelogWatch = c.ExperimentalCRDBChangelogWatch to.GCInterval = c.GCInterval to.GCMaxOperationTime = c.GCMaxOperationTime to.RelaxedIsolationLevel = c.RelaxedIsolationLevel @@ -231,6 +232,7 @@ func (c *Config) DebugMap() map[string]any { } else { debugMap["WriteAcquisitionTimeout"] = c.WriteAcquisitionTimeout } + debugMap["ExperimentalCRDBChangelogWatch"] = c.ExperimentalCRDBChangelogWatch if dm, ok := any(&c.GCInterval).(interface { DebugMap() map[string]any }); ok { @@ -613,6 +615,13 @@ func WithWriteAcquisitionTimeout(writeAcquisitionTimeout time.Duration) ConfigOp } } +// WithExperimentalCRDBChangelogWatch returns an option that can set ExperimentalCRDBChangelogWatch on a Config +func WithExperimentalCRDBChangelogWatch(experimentalCRDBChangelogWatch bool) ConfigOption { + return func(c *Config) { + c.ExperimentalCRDBChangelogWatch = experimentalCRDBChangelogWatch + } +} + // WithGCInterval returns an option that can set GCInterval on a Config func WithGCInterval(gCInterval time.Duration) ConfigOption { return func(c *Config) {