chore: add cockroachdb 26.1 to the matrix

Author: tstirrat15

.github/workflows/build-test.yaml

@@ -262,7 +262,11 @@ jobs:
       fail-fast: false
       matrix:
         datastore: ["crdb"]
-        crdbversion: ["25.2.0", "25.3.0"]
+        crdbversion:
+          # the version many folks are on
+          - "25.2.0"
+          # the most recent version
+          - "26.1.0"
     steps:
       - uses: "actions/checkout@v6"
         if: |
@@ -297,7 +301,11 @@ jobs:
       fail-fast: false
       matrix:
         datastore: ["crdb"]
-        crdbversion: ["25.2.0", "25.3.0"]
+        crdbversion:
+          # the version many folks are on
+          - "25.2.0"
+          # the most recent version
+          - "26.1.0"
     steps:
       - uses: "actions/checkout@v6"
         if: |

CHANGELOG.md

@@ -14,6 +14,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
   NOTE: in a future release, MySQL metrics will change.
 - Add support for imports and partials to the schemadsl package that drives the LSP and development server (https://github.com/authzed/spicedb/pull/2919).
 - `DatastoreTester.New` now takes a `testing.TB` as its first argument, allowing per-test cleanup in datastore test suites (https://github.com/authzed/spicedb/pull/2925).
+- Added support for CRDB 26.1 by fixing how version information is read from the cluster
 
 ### Fixed
 - enforce graceful shutdown on serve and serve-testing (https://github.com/authzed/spicedb/pull/2888)

internal/datastore/crdb/crdb_test.go

@@ -17,6 +17,8 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -118,7 +120,7 @@ type datastoreTestFunc func(t *testing.T, ds datastore.Datastore)
 
 func createDatastoreTest(b testdatastore.RunningEngineForTest, tf datastoreTestFunc, options ...Option) func(*testing.T) {
 	return func(t *testing.T) {
-		ctx := context.Background()
+		ctx := t.Context()
 		ds := b.NewDatastore(t, func(engine, uri string) datastore.Datastore {
 			ds, err := NewCRDBDatastore(ctx, uri, options...)
 			require.NoError(t, err)
@@ -960,3 +962,39 @@ func TestRegisterPrometheusCollectors(t *testing.T) {
 	require.NotNil(t, poolReadMetric)
 	require.Equal(t, float64(readMaxConns), poolReadMetric.GetGauge().GetValue()) //nolint:testifylint // we expect exact values
 }
+
+func TestVersionReading(t *testing.T) {
+	require := require.New(t)
+
+	expectedVersionList := strings.Split(crdbTestVersion(), ".")
+	expectedMajor, err := strconv.Atoi(expectedVersionList[0])
+	require.NoError(err)
+	expectedMinor, err := strconv.Atoi(expectedVersionList[1])
+	require.NoError(err)
+	expectedPatch, err := strconv.Atoi(expectedVersionList[2])
+	require.NoError(err)
+
+	var version crdbVersion
+
+	b := testdatastore.RunCRDBForTesting(t, "", crdbTestVersion())
+	uri := b.NewDatabase(t)
+
+	// Set up a raw connection to the DB
+	initPoolConfig, err := pgxpool.ParseConfig(uri)
+	require.NoError(err)
+	checker, err := pool.NewNodeHealthChecker(uri)
+	require.NoError(err)
+	initPool, err := pool.NewRetryPool(t.Context(), "pool", initPoolConfig, checker, 18, 20)
+	require.NoError(err)
+	t.Cleanup(func() {
+		initPool.Close()
+	})
+
+	// Make the query for the server version
+	err = queryServerVersion(t.Context(), initPool, &version)
+	require.NoError(err)
+
+	require.Equal(expectedMajor, version.Major)
+	require.Equal(expectedMinor, version.Minor)
+	require.Equal(expectedPatch, version.Patch)
+}

internal/datastore/crdb/version.go

@@ -2,48 +2,33 @@ package crdb
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"regexp"
 	"strconv"
 
 	"github.com/jackc/pgx/v5"
-	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/rs/zerolog"
 
 	pgxcommon "github.com/authzed/spicedb/internal/datastore/postgres/common"
 )
 
 const (
-	queryVersionJSON = "SELECT crdb_internal.active_version()::jsonb;"
-	queryVersion     = "SELECT version();"
-
-	errFunctionDoesNotExist = "42883"
+	queryVersion = "SELECT version();"
 )
 
 var versionRegex = regexp.MustCompile(`v([0-9]+)\.([0-9]+)\.([0-9]+)(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+[0-9A-Za-z-]+)?`)
 
+// queryServerVersion reads the version() string out of CRDB
+// and then parses it using a regex to determine the semver.
 func queryServerVersion(ctx context.Context, db pgxcommon.DBFuncQuerier, version *crdbVersion) error {
+	var versionStr string
 	if err := db.QueryRowFunc(ctx, func(ctx context.Context, row pgx.Row) error {
-		return row.Scan(version)
-	}, queryVersionJSON); err != nil {
-		var pgerr *pgconn.PgError
-		if !errors.As(err, &pgerr) || pgerr.Code != errFunctionDoesNotExist {
-			return err
-		}
-
-		// The crdb_internal.active_version() wasn't added until v22.1.X, try to parse the version
-		var versionStr string
-		if err := db.QueryRowFunc(ctx, func(ctx context.Context, row pgx.Row) error {
-			return row.Scan(&versionStr)
-		}, queryVersion); err != nil {
-			return err
-		}
-
-		return parseVersionStringInto(versionStr, version)
+		return row.Scan(&versionStr)
+	}, queryVersion); err != nil {
+		return err
 	}
 
-	return nil
+	return parseVersionStringInto(versionStr, version)
 }
 
 func parseVersionStringInto(versionStr string, version *crdbVersion) error {

internal/datastore/crdb/version/version.go

@@ -8,4 +8,4 @@ const MinimumSupportedCockroachDBVersion = "23.1.30"
 // LatestTestedCockroachDBVersion is the latest version of CockroachDB that has been tested with this driver.
 //
 // NOTE: must match a tag on DockerHub for the `cockroachdb/cockroach` image, without the `v` prefix.
-const LatestTestedCockroachDBVersion = "25.3.2"
+const LatestTestedCockroachDBVersion = "26.1.0"

internal/datastore/crdb/watch.go

@@ -28,7 +28,11 @@ import (
 )
 
 const (
-	queryChangefeed       = "CREATE CHANGEFEED FOR %s WITH updated, cursor = '%s', resolved = '%s', min_checkpoint_frequency = '0';"
+	// cursor: the revision before which we want to start the changefeed
+	// resolved: the minimum duration between "resolved" checkpoint messages
+	// min_checkpoint_frequency: how frequently CRDB will attempt to produce a checkpoint. the docs say that this can be 0,
+	// but that doesn't seem to be true for 26.1, so we set it to 1ms, which is still short but seems to work.
+	queryChangefeed       = "CREATE CHANGEFEED FOR %s WITH updated, cursor = '%s', resolved = '%s', min_checkpoint_frequency = '1ms';"
 	queryChangefeedPreV25 = "EXPERIMENTAL CHANGEFEED FOR %s WITH updated, cursor = '%s', resolved = '%s', min_checkpoint_frequency = '0';"
 	queryChangefeedPreV22 = "EXPERIMENTAL CHANGEFEED FOR %s WITH updated, cursor = '%s', resolved = '%s';"
 )

internal/datastore/postgres/postgres_shared_test.go

@@ -1937,7 +1937,7 @@ func NullCaveatWatchTest(t *testing.T, ds datastore.Datastore) {
 	require.NoError(err)
 
 	// Verify the relationship create was tracked by the watch.
-	test.VerifyUpdates(require, [][]tuple.RelationshipUpdate{
+	test.VerifyUpdates(t, require, [][]tuple.RelationshipUpdate{
 		{
 			tuple.Touch(tuple.MustParse("resource:someresourceid#somerelation@subject:somesubject")),
 		},
@@ -1953,7 +1953,7 @@ func NullCaveatWatchTest(t *testing.T, ds datastore.Datastore) {
 	require.NoError(err)
 
 	// Verify the delete.
-	test.VerifyUpdates(require, [][]tuple.RelationshipUpdate{
+	test.VerifyUpdates(t, require, [][]tuple.RelationshipUpdate{
 		{
 			tuple.Delete(tuple.MustParse("resource:someresourceid#somerelation@subject:somesubject")),
 		},

pkg/datastore/test/revisions.go

@@ -20,7 +20,6 @@ import (
 
 // RevisionQuantizationTest tests whether or not the requirements for revisions hold
 // for a particular datastore.
-// TODO: rewrite using synctest
 func RevisionQuantizationTest(t *testing.T, tester DatastoreTester) {
 	testCases := []struct {
 		quantizationRange        time.Duration
@@ -178,6 +177,7 @@ func RevisionGCTest(t *testing.T, tester DatastoreTester) {
 	time.Sleep(gcWindow)
 
 	gcable, ok := ds.(common.GarbageCollectableDatastore)
+	// NOTE: CRDB and Spanner both do garbage collection with row-level TTLs
 	if ok {
 		// Run garbage collection.
 		gcable.ResetGCCompleted()
@@ -201,6 +201,12 @@ func RevisionGCTest(t *testing.T, tester DatastoreTester) {
 	head, err = ds.HeadRevision(ctx)
 	require.NoError(err)
 
+	// assert that recent call to head revision is also valid, even after a GC window cycle without writes elapsed
+	require.NoError(ds.CheckRevision(ctx, head), "expected freshly obtained head revision to be valid")
+
+	// TODO: these reads are taking a significant amount of time on CRDB, on the order
+	// of 100ms for a row read. We need to ascertain whether this is a test artifact
+	// or a performance regression.
 	// check that we can read a caveat whose revision has been garbage collectged
 	_, _, err = ds.SnapshotReader(head).LegacyReadCaveatByName(ctx, testCaveat.Name)
 	require.NoError(err, "expected previously written caveat should exist at head")
@@ -213,9 +219,6 @@ func RevisionGCTest(t *testing.T, tester DatastoreTester) {
 	_, _, err = ds.SnapshotReader(head).LegacyReadNamespaceByName(ctx, "foo/bar")
 	require.NoError(err, "expected previously written schema to exist at head")
 
-	// and that recent call to head revision is also valid, even after a GC window cycle without writes elapsed
-	require.NoError(ds.CheckRevision(ctx, head), "expected freshly obtained head revision to be valid")
-
 	// write happens, we get a new head revision
 	newerRev, err := ds.ReadWriteTx(ctx, func(ctx context.Context, rwt datastore.ReadWriteTransaction) error {
 		return rwt.LegacyWriteNamespaces(ctx, testNamespace)

pkg/datastore/test/watch.go

@@ -69,8 +69,11 @@ func WatchTest(t *testing.T, tester DatastoreTester) {
 			require.NoError(err)
 
 			opts := datastore.WatchOptions{
-				Content:                 datastore.WatchRelationships,
-				WatchBufferLength:       50,
+				Content:           datastore.WatchRelationships,
+				WatchBufferLength: 50,
+				// This should mean that we can wait until the checkpoint is emitted and then
+				// make our writes.
+				// CheckpointInterval: 1*time.Second,
 				WatchBufferWriteTimeout: tc.bufferTimeout,
 			}
 			changes, errchan := ds.Watch(t.Context(), lowestRevision, opts)
@@ -122,22 +125,24 @@ func WatchTest(t *testing.T, tester DatastoreTester) {
 				testUpdates = append(testUpdates, bulkDeletes)
 			}
 
-			VerifyUpdates(require, testUpdates, changes, errchan, tc.expectFallBehind)
+			VerifyUpdates(t, require, testUpdates, changes, errchan, tc.expectFallBehind)
 
 			// Test the catch-up case
 			changes, errchan = ds.Watch(t.Context(), lowestRevision, opts)
-			VerifyUpdates(require, testUpdates, changes, errchan, tc.expectFallBehind)
+			VerifyUpdates(t, require, testUpdates, changes, errchan, tc.expectFallBehind)
 		})
 	}
 }
 
 func VerifyUpdates(
+	t *testing.T,
 	require *require.Assertions,
 	testUpdates [][]tuple.RelationshipUpdate,
 	changes <-chan datastore.RevisionChanges,
 	errchan <-chan error,
 	expectDisconnect bool,
 ) {
+	t.Helper()
 	for _, expected := range testUpdates {
 		changeWait := time.NewTimer(waitForChangesTimeout)
 		select {
@@ -174,12 +179,14 @@ func VerifyUpdates(
 }
 
 func VerifyUpdatesWithMetadata(
+	t *testing.T,
 	require *require.Assertions,
 	testUpdates []updateWithMetadata,
 	changes <-chan datastore.RevisionChanges,
 	errchan <-chan error,
 	expectDisconnect bool,
 ) {
+	t.Helper()
 	for _, expected := range testUpdates {
 		changeWait := time.NewTimer(waitForChangesTimeout)
 		select {
@@ -309,7 +316,7 @@ func WatchWithTouchTest(t *testing.T, tester DatastoreTester) {
 		tuple.MustParse("document:firstdoc#viewer@user:fred[thirdcaveat]"),
 	)
 
-	VerifyUpdates(require, [][]tuple.RelationshipUpdate{
+	VerifyUpdates(t, require, [][]tuple.RelationshipUpdate{
 		{
 			tuple.Touch(tuple.MustParse("document:firstdoc#viewer@user:tom")),
 			tuple.Touch(tuple.MustParse("document:firstdoc#viewer@user:sarah")),
@@ -353,7 +360,7 @@ func WatchWithTouchTest(t *testing.T, tester DatastoreTester) {
 		tuple.MustParse("document:firstdoc#viewer@user:fred[thirdcaveat]"),
 	)
 
-	VerifyUpdates(require, [][]tuple.RelationshipUpdate{
+	VerifyUpdates(t, require, [][]tuple.RelationshipUpdate{
 		{tuple.Touch(tuple.MustParse("document:firstdoc#viewer@user:tom[somecaveat]"))},
 	},
 		changes,
@@ -374,7 +381,7 @@ func WatchWithTouchTest(t *testing.T, tester DatastoreTester) {
 		tuple.MustParse("document:firstdoc#viewer@user:fred[thirdcaveat]"),
 	)
 
-	VerifyUpdates(require, [][]tuple.RelationshipUpdate{
+	VerifyUpdates(t, require, [][]tuple.RelationshipUpdate{
 		{tuple.Touch(tuple.MustParse("document:firstdoc#viewer@user:tom[somecaveat:{\"somecondition\": 42}]"))},
 	},
 		changes,
@@ -410,7 +417,7 @@ func WatchWithExpirationTest(t *testing.T, tester DatastoreTester) {
 	}, options.WithMetadata(metadata))
 	require.NoError(err)
 
-	VerifyUpdates(require, [][]tuple.RelationshipUpdate{
+	VerifyUpdates(t, require, [][]tuple.RelationshipUpdate{
 		{tuple.Touch(tuple.MustParse("document:firstdoc#viewer@user:tom[expiration:2321-01-01T00:00:00Z]"))},
 	},
 		changes,
@@ -455,7 +462,7 @@ func WatchWithMetadataTest(t *testing.T, tester DatastoreTester) {
 	}, options.WithMetadata(metadata))
 	require.NoError(err)
 
-	VerifyUpdatesWithMetadata(require, []updateWithMetadata{
+	VerifyUpdatesWithMetadata(t, require, []updateWithMetadata{
 		{
 			updates:  []tuple.RelationshipUpdate{tuple.Touch(tuple.MustParse("document:firstdoc#viewer@user:tom"))},
 			metadata: map[string]any{"somekey": "somevalue"},
@@ -498,7 +505,7 @@ func WatchWithDeleteTest(t *testing.T, tester DatastoreTester) {
 		tuple.MustParse("document:firstdoc#viewer@user:fred[thirdcaveat]"),
 	)
 
-	VerifyUpdates(require, [][]tuple.RelationshipUpdate{
+	VerifyUpdates(t, require, [][]tuple.RelationshipUpdate{
 		{
 			tuple.Touch(tuple.MustParse("document:firstdoc#viewer@user:tom")),
 			tuple.Touch(tuple.MustParse("document:firstdoc#viewer@user:sarah")),
@@ -522,7 +529,7 @@ func WatchWithDeleteTest(t *testing.T, tester DatastoreTester) {
 		tuple.MustParse("document:firstdoc#viewer@user:fred[thirdcaveat]"),
 	)
 
-	VerifyUpdates(require, [][]tuple.RelationshipUpdate{
+	VerifyUpdates(t, require, [][]tuple.RelationshipUpdate{
 		{tuple.Delete(tuple.MustParse("document:firstdoc#viewer@user:tom"))},
 	},
 		changes,