Experimental: changelog table for crdb

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)

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
 			}

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

internal/datastore/crdb/caveat.go

@@ -155,6 +155,20 @@
 	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 @@
 	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
 }