fixes pg replica error: "canceling statement due to conflict with recovery

go.mod

@@ -99,7 +99,7 @@ require (
 	go.opentelemetry.io/otel v1.33.0
 	go.opentelemetry.io/otel/sdk v1.33.0
 	go.opentelemetry.io/otel/trace v1.33.0
-	go.uber.org/atomic v1.11.0
+	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/goleak v1.3.0
 	golang.org/x/exp v0.0.0-20240909161429-701f63a606c0
 	golang.org/x/mod v0.22.0

internal/datastore/postgres/strictreader.go

@@ -43,15 +43,22 @@ func (srqf strictReaderQueryFuncs) rewriteError(err error) error {
 		return nil
 	}
 
+	if srqf.isReplicationLagError(err) {
+		return common.NewRevisionUnavailableError(fmt.Errorf("revision %s is not available on the replica", srqf.revision.String()))
+	}
+
+	return err
+}
+
+func (srqf strictReaderQueryFuncs) isReplicationLagError(err error) bool {
 	var pgerr *pgconn.PgError
 	if errors.As(err, &pgerr) {
-		if (pgerr.Code == pgInvalidArgument && strings.Contains(pgerr.Message, "is in the future")) ||
-			strings.Contains(pgerr.Message, "replica missing revision") {
-			return common.NewRevisionUnavailableError(fmt.Errorf("revision %s is not available on the replica", srqf.revision.String()))
-		}
+		return (pgerr.Code == pgInvalidArgument && strings.Contains(pgerr.Message, "is in the future")) ||
+			strings.Contains(pgerr.Message, "replica missing revision") ||
+			pgxcommon.IsSerializationError(err)
 	}
 
-	return err
+	return false
 }
 
 func (srqf strictReaderQueryFuncs) addAssertToSelectSQL(sql string) string {

internal/datastore/postgres/strictreader_test.go

@@ -0,0 +1,97 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/stretchr/testify/require"
+
+	"github.com/authzed/spicedb/internal/datastore/common"
+)
+
+type fakeQuerier struct {
+	err error
+}
+
+func (mc fakeQuerier) QueryFunc(ctx context.Context, rowsFunc func(ctx context.Context, rows pgx.Rows) error, sql string, optionsAndArgs ...any) error {
+	return mc.err
+}
+
+func (mc fakeQuerier) QueryRowFunc(ctx context.Context, rowFunc func(ctx context.Context, row pgx.Row) error, sql string, optionsAndArgs ...any) error {
+	return mc.err
+}
+
+func (mc fakeQuerier) ExecFunc(_ context.Context, _ func(ctx context.Context, tag pgconn.CommandTag, err error) error, _ string, _ ...interface{}) error {
+	return mc.err
+}
+
+func TestStrictReaderDetectsLagErrors(t *testing.T) {
+	mc := fakeQuerier{}
+	reader := strictReaderQueryFuncs{
+		wrapped: mc,
+	}
+
+	cases := []struct {
+		name string
+		in   error
+		want error
+	}{
+		{
+			name: "no error is bubbledtest",
+		},
+		{
+			name: "missing revision on replica - invalid argument",
+			in:   &pgconn.PgError{Code: pgInvalidArgument, Message: "is in the future"},
+			want: common.RevisionUnavailableError{},
+		},
+		{
+			name: "missing revision on replica",
+			in:   &pgconn.PgError{Message: "replica missing revision"},
+			want: common.RevisionUnavailableError{},
+		},
+		{
+			name: "serialization error due to conflicting WAL changes on replica",
+			in:   &pgconn.PgError{Code: "40001"},
+			want: common.RevisionUnavailableError{},
+		},
+		{
+			name: "bubbles up unrelated error",
+			in:   fmt.Errorf("unrelated error"),
+			want: fmt.Errorf("unrelated error"),
+		},
+	}
+
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			received := tt.in
+			expected := tt.want
+
+			mc.err = received
+			reader.wrapped = mc
+			err := reader.ExecFunc(context.Background(), nil, "SELECT 1")
+			if expected != nil {
+				require.ErrorAs(t, err, &expected)
+			} else {
+				require.NoError(t, err)
+			}
+
+			err = reader.QueryFunc(context.Background(), nil, "SELECT 1")
+			if expected != nil {
+				require.ErrorAs(t, err, &expected)
+			} else {
+				require.NoError(t, err)
+			}
+
+			err = reader.QueryRowFunc(context.Background(), nil, "SELECT 1")
+			if expected != nil {
+				require.ErrorAs(t, err, &expected)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}