chore: move off experimental changefeed query for CRDB

Author: miparnisari

internal/datastore/crdb/crdb.go

@@ -119,6 +119,10 @@ func newCRDBDatastore(ctx context.Context, url string, options ...Option) (datas
 		log.Info().Object("version", version).Msg("using changefeed query for CRDB version < 22")
 		changefeedQuery = queryChangefeedPreV22
 	}
+	if version.Major < 25 {
+		log.Info().Object("version", version).Msg("using changefeed query for CRDB version < 25")
+		changefeedQuery = queryChangefeedPreV25
+	}
 
 	transactionNowQuery := queryTransactionNow
 	if version.Major < 23 {

internal/datastore/crdb/crdb_test.go

@@ -251,7 +251,7 @@ func TestWatchFeatureDetection(t *testing.T) {
 		{
 			name: "rangefeeds disabled",
 			postInit: func(ctx context.Context, adminConn *pgx.Conn) {
-				_, err = adminConn.Exec(ctx, `SET CLUSTER SETTING kv.rangefeed.enabled = false;`)
+				_, err := adminConn.Exec(ctx, `SET CLUSTER SETTING kv.rangefeed.enabled = false;`)
 				require.NoError(t, err)
 			},
 			expectEnabled: false,
@@ -260,7 +260,7 @@ func TestWatchFeatureDetection(t *testing.T) {
 		{
 			name: "rangefeeds enabled, user doesn't have permission",
 			postInit: func(ctx context.Context, adminConn *pgx.Conn) {
-				_, err = adminConn.Exec(ctx, `SET CLUSTER SETTING kv.rangefeed.enabled = true;`)
+				_, err := adminConn.Exec(ctx, `SET CLUSTER SETTING kv.rangefeed.enabled = true;`)
 				require.NoError(t, err)
 			},
 			expectEnabled: false,
@@ -269,7 +269,7 @@ func TestWatchFeatureDetection(t *testing.T) {
 		{
 			name: "rangefeeds enabled, user has permission",
 			postInit: func(ctx context.Context, adminConn *pgx.Conn) {
-				_, err = adminConn.Exec(ctx, `SET CLUSTER SETTING kv.rangefeed.enabled = true;`)
+				_, err := adminConn.Exec(ctx, `SET CLUSTER SETTING kv.rangefeed.enabled = true;`)
 				require.NoError(t, err)
 
 				_, err = adminConn.Exec(ctx, fmt.Sprintf(`GRANT CHANGEFEED ON TABLE testspicedb.%s TO unprivileged;`, schema.TableTuple))

internal/datastore/crdb/pool_test.go

@@ -117,7 +117,7 @@ func TestTxReset(t *testing.T) {
 
 			r, err := ds.ReadyState(ctx)
 			require.NoError(err)
-			require.True(r.IsReady)
+			require.True(r.IsReady, "datastore not ready: %s", r.Message)
 
 			// WriteNamespace utilizes execute so we'll use it
 			i := 0

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 = "24.2.9"
+const LatestTestedCockroachDBVersion = "25.3.2"

internal/datastore/crdb/watch.go

@@ -27,7 +27,8 @@ import (
 )
 
 const (
-	queryChangefeed       = "EXPERIMENTAL CHANGEFEED FOR %s WITH updated, cursor = '%s', resolved = '%s', min_checkpoint_frequency = '0';"
+	queryChangefeed       = "CREATE CHANGEFEED FOR %s WITH updated, cursor = '%s', resolved = '%s', min_checkpoint_frequency = '0';"
+	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/testserver/datastore/crdb.go

@@ -6,6 +6,7 @@ package datastore
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 
 	"github.com/google/uuid"
@@ -24,15 +25,19 @@ const (
 	enableRangefeeds = `SET CLUSTER SETTING kv.rangefeed.enabled = true;`
 )
 
+// crdbTester is safe for concurrent use by tests.
 type crdbTester struct {
-	conn     *pgx.Conn
-	hostname string
-	creds    string
-	port     string
+	conn      *pgx.Conn // GUARDED_BY(connMutex)
+	connMutex sync.Mutex
+	hostname  string
+	creds     string
+	port      string
 }
 
+var _ RunningEngineForTest = (*crdbTester)(nil)
+
 // RunCRDBForTesting returns a RunningEngineForTest for CRDB
-func RunCRDBForTesting(t testing.TB, bridgeNetworkName string, crdbVersion string) RunningEngineForTest {
+func RunCRDBForTesting(t testing.TB, bridgeNetworkName string, crdbVersion string) *crdbTester {
 	pool, err := dockertest.NewPool("")
 	require.NoError(t, err)
 
@@ -55,6 +60,8 @@ func RunCRDBForTesting(t testing.TB, bridgeNetworkName string, crdbVersion strin
 		creds:    "root:fake",
 	}
 	t.Cleanup(func() {
+		builder.connMutex.Lock()
+		defer builder.connMutex.Unlock()
 		if builder.conn != nil {
 			require.NoError(t, builder.conn.Close(context.Background()))
 		}
@@ -74,26 +81,31 @@ func RunCRDBForTesting(t testing.TB, bridgeNetworkName string, crdbVersion strin
 		var err error
 		ctx, cancelConnect := context.WithTimeout(context.Background(), dockerBootTimeout)
 		defer cancelConnect()
-		builder.conn, err = pgx.Connect(ctx, uri)
+		conn, err := pgx.Connect(ctx, uri)
 		if err != nil {
 			return err
 		}
+		builder.connMutex.Lock()
+		builder.conn = conn
 		ctx, cancelRangeFeeds := context.WithTimeout(context.Background(), dockerBootTimeout)
 		defer cancelRangeFeeds()
 		_, err = builder.conn.Exec(ctx, enableRangefeeds)
+		builder.connMutex.Unlock()
 		return err
 	}))
 
 	return builder
 }
 
-// NewDatabase creates a database. It is NOT safe for concurrent use.
+// NewDatabase creates a database.
 func (r *crdbTester) NewDatabase(t testing.TB) string {
 	uniquePortion, err := secrets.TokenHex(4)
 	require.NoError(t, err)
 
 	newDBName := "db" + uniquePortion
 
+	r.connMutex.Lock()
+	defer r.connMutex.Unlock()
 	_, err = r.conn.Exec(context.Background(), "CREATE DATABASE "+newDBName)
 	require.NoError(t, err)
 
@@ -107,7 +119,7 @@ func (r *crdbTester) NewDatabase(t testing.TB) string {
 	return connectStr
 }
 
-// NewDatastore creates a database and runs migrations on it. It is NOT safe for concurrent use.
+// NewDatastore creates a database and runs migrations on it.
 func (r *crdbTester) NewDatastore(t testing.TB, initFunc InitFunc) datastore.Datastore {
 	connectStr := r.NewDatabase(t)
 

magefiles/util.go

@@ -44,6 +44,7 @@ func testWithArgs(ctx context.Context, args ...string) ([]string, error) {
 		"test",
 		"-failfast",
 		"-count=1",
+		"-race",
 	}, args...)
 
 	return testArgs, nil

pkg/datastore/test/relationships.go

@@ -54,7 +54,7 @@ func SimpleTest(t *testing.T, tester DatastoreTester) {
 
 			ok, err := ds.ReadyState(ctx)
 			require.NoError(err)
-			require.True(ok.IsReady)
+			require.True(ok.IsReady, "datastore not ready: %s", ok.Message)
 
 			setupDatastore(ds, require)