Merge pull request #1688 from authzed/fix-error-codes-conflict-and-re…

…tryable-crdb

fixes WriteRelationships/BulkImport GRPC error codes on conflicts and retryable errors

internal/datastore/crdb/crdb.go

@@ -336,12 +336,21 @@ func (cds *crdbDatastore) ReadWriteTx(
 		return nil
 	})
 	if err != nil {
-		return datastore.NoRevision, err
+		return datastore.NoRevision, wrapError(err)
 	}
 
 	return commitTimestamp, nil
 }
 
+func wrapError(err error) error {
+	// If a unique constraint violation is returned, then its likely that the cause
+	// was an existing relationship.
+	if cerr := pgxcommon.ConvertToWriteConstraintError(livingTupleConstraint, err); cerr != nil {
+		return cerr
+	}
+	return err
+}
+
 func (cds *crdbDatastore) ReadyState(ctx context.Context) (datastore.ReadyState, error) {
 	headMigration, err := migrations.CRDBMigrations.HeadRevision()
 	if err != nil {

internal/datastore/crdb/pool/slqerrors.go

@@ -5,6 +5,10 @@ import (
 	"fmt"
 
 	"github.com/jackc/pgx/v5/pgconn"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/authzed/spicedb/pkg/spiceerrors"
 )
 
 const (
@@ -35,6 +39,15 @@ func (e *MaxRetryError) Error() string {
 
 func (e *MaxRetryError) Unwrap() error { return e.LastErr }
 
+func (e *MaxRetryError) GRPCStatus() *status.Status {
+	s, ok := status.FromError(e.Unwrap())
+	if !ok {
+		return nil
+	}
+
+	return s
+}
+
 // ResettableError is an error that we think may succeed if retried against a new connection.
 type ResettableError struct {
 	Err error
@@ -43,6 +56,13 @@ type ResettableError struct {
 func (e *ResettableError) Error() string { return "resettable error" + ": " + e.Err.Error() }
 func (e *ResettableError) Unwrap() error { return e.Err }
 
+func (e *ResettableError) GRPCStatus() *status.Status {
+	return spiceerrors.WithCodeAndDetails(
+		e.Unwrap(),
+		codes.Unavailable, // return unavailable so clients are know it's ok to retry
+	)
+}
+
 // RetryableError is an error that can be retried against the existing connection.
 type RetryableError struct {
 	Err error
@@ -51,6 +71,13 @@ type RetryableError struct {
 func (e *RetryableError) Error() string { return "retryable error" + ": " + e.Err.Error() }
 func (e *RetryableError) Unwrap() error { return e.Err }
 
+func (e *RetryableError) GRPCStatus() *status.Status {
+	return spiceerrors.WithCodeAndDetails(
+		e.Unwrap(),
+		codes.Unavailable, // return unavailable so clients are know it's ok to retry
+	)
+}
+
 // sqlErrorCode attempts to extract the crdb error code from the error state.
 func sqlErrorCode(err error) string {
 	var pgerr *pgconn.PgError

internal/datastore/crdb/readwrite.go

@@ -168,12 +168,6 @@ func (rwt *crdbReadWriteTXN) WriteRelationships(ctx context.Context, mutations [
 		}
 
 		if _, err := rwt.tx.Exec(ctx, sql, args...); err != nil {
-			// If a unique constraint violation is returned, then its likely that the cause
-			// was an existing relationship given as a CREATE.
-			if cerr := pgxcommon.ConvertToWriteConstraintError(livingTupleConstraint, err); cerr != nil {
-				return cerr
-			}
-
 			return fmt.Errorf(errUnableToWriteRelationships, err)
 		}
 	}

internal/datastore/memdb/readwrite.go

@@ -213,7 +213,7 @@ func (rwt *memdbReadWriteTx) DeleteNamespaces(_ context.Context, nsNames ...stri
 
 func (rwt *memdbReadWriteTx) BulkLoad(ctx context.Context, iter datastore.BulkWriteRelationshipSource) (uint64, error) {
 	updates := []*core.RelationTupleUpdate{{
-		Operation: core.RelationTupleUpdate_TOUCH,
+		Operation: core.RelationTupleUpdate_CREATE,
 	}}
 
 	var numCopied uint64

internal/datastore/mysql/datastore.go

@@ -321,15 +321,26 @@ func (mds *Datastore) ReadWriteTx(
 				continue
 			}
 
-			return datastore.NoRevision, err
+			return datastore.NoRevision, wrapError(err)
 		}
 
 		return revisionFromTransaction(newTxnID), nil
 	}
 	if !config.DisableRetries {
 		err = fmt.Errorf("max retries exceeded: %w", err)
 	}
-	return datastore.NoRevision, err
+
+	return datastore.NoRevision, wrapError(err)
+}
+
+// wrapError maps any mysql internal error into a SpiceDB typed error or an error
+// that implements GRPCStatus().
+func wrapError(err error) error {
+	if cerr := convertToWriteConstraintError(err); cerr != nil {
+		return cerr
+	}
+
+	return err
 }
 
 func isErrorRetryable(err error) bool {

internal/datastore/mysql/readwrite.go

@@ -210,10 +210,6 @@ func (rwt *mysqlReadWriteTXN) WriteRelationships(ctx context.Context, mutations
 
 		_, err = rwt.tx.ExecContext(ctx, query, args...)
 		if err != nil {
-			if cerr := convertToWriteConstraintError(err); cerr != nil {
-				return cerr
-			}
-
 			return fmt.Errorf(errUnableToWriteRelationships, err)
 		}
 	}

internal/datastore/postgres/postgres.go

@@ -473,6 +473,12 @@ func (pgd *pgDatastore) RepairOperations() []datastore.RepairOperation {
 }
 
 func wrapError(err error) error {
+	// If a unique constraint violation is returned, then its likely that the cause
+	// was an existing relationship given as a CREATE.
+	if cerr := pgxcommon.ConvertToWriteConstraintError(livingTupleConstraint, err); cerr != nil {
+		return cerr
+	}
+
 	if pgxcommon.IsSerializationError(err) {
 		return common.NewSerializationError(err)
 	}

internal/datastore/postgres/readwrite.go

@@ -117,12 +117,6 @@ func (rwt *pgReadWriteTXN) WriteRelationships(ctx context.Context, mutations []*
 
 		_, err = rwt.tx.Exec(ctx, sql, args...)
 		if err != nil {
-			// If a unique constraint violation is returned, then its likely that the cause
-			// was an existing relationship given as a CREATE.
-			if cerr := pgxcommon.ConvertToWriteConstraintError(livingTupleConstraint, err); cerr != nil {
-				return cerr
-			}
-
 			return fmt.Errorf(errUnableToWriteRelationships, err)
 		}
 	}

pkg/datastore/test/tuples.go

@@ -566,6 +566,14 @@ func CreateAlreadyExistingTest(t *testing.T, tester DatastoreTester) {
 	require.ErrorAs(err, &common.CreateRelationshipExistsError{})
 	require.Contains(err.Error(), "could not CREATE relationship ")
 	grpcutil.RequireStatus(t, codes.AlreadyExists, err)
+
+	f := func(ctx context.Context, rwt datastore.ReadWriteTransaction) error {
+		_, err := rwt.BulkLoad(ctx, testfixtures.NewBulkTupleGenerator(testResourceNamespace, testReaderRelation, testUserNamespace, 1, t))
+		return err
+	}
+	_, _ = ds.ReadWriteTx(ctx, f)
+	_, err = ds.ReadWriteTx(ctx, f)
+	grpcutil.RequireStatus(t, codes.AlreadyExists, err)
 }
 
 // TouchAlreadyExistingTest tests touching a relationship twice.