Serializable failure with recursion for DoTransaction on this error, and transaction aborted error handled

connectors/storage/v2/api.go

@@ -5,6 +5,7 @@ package storage
 import (
 	"context"
 	"strings"
+	stdlibtime "time"
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/georgysavva/scany/v2/pgxscan"
@@ -38,6 +39,11 @@ func DoInTransaction(ctx context.Context, db *DB, fn func(conn QueryExecer) erro
 			return nil, backoff.Permanent(err) //nolint:wrapcheck // Not needed.
 		}
 	})
+	if err != nil && (errors.Is(err, ErrSerializationFailure) || errors.Is(err, ErrTxAborted)) {
+		stdlibtime.Sleep(10 * stdlibtime.Millisecond)
+
+		return DoInTransaction(ctx, db, fn)
+	}
 
 	return err
 }
@@ -172,7 +178,10 @@ func IsUnexpected(err error) bool {
 		!IsErr(err, ErrRelationNotFound) &&
 		!IsErr(err, ErrNotFound) &&
 		!IsErr(err, ErrCheckFailed) &&
-		!IsErr(err, ErrRelationInUse)
+		!IsErr(err, ErrRelationInUse) &&
+		!IsErr(err, ErrSerializationFailure) &&
+		!IsErr(err, ErrTxAborted) &&
+		!IsErr(err, ErrExclusionViolation)
 }
 
 func parseDBError(err error) error { //nolint:funlen // .
@@ -206,6 +215,15 @@ func parseDBError(err error) error { //nolint:funlen // .
 
 			return terror.New(ErrCheckFailed, map[string]any{"column": column})
 		}
+		if dbErr.SQLState() == "40001" {
+			return ErrSerializationFailure
+		}
+		if dbErr.SQLState() == "25P02" {
+			return ErrTxAborted
+		}
+		if dbErr.SQLState() == "23P01" {
+			return ErrExclusionViolation
+		}
 
 		return err
 	}

connectors/storage/v2/contract.go

@@ -10,11 +10,14 @@ import (
 // Public API.
 
 var (
-	ErrNotFound         = errors.New("not found")
-	ErrRelationNotFound = errors.New("relation not found")
-	ErrRelationInUse    = errors.New("relation in use")
-	ErrDuplicate        = errors.New("duplicate")
-	ErrCheckFailed      = errors.New("check failed")
+	ErrNotFound             = errors.New("not found")
+	ErrRelationNotFound     = errors.New("relation not found")
+	ErrRelationInUse        = errors.New("relation in use")
+	ErrDuplicate            = errors.New("duplicate")
+	ErrCheckFailed          = errors.New("check failed")
+	ErrSerializationFailure = errors.New("serialization failure")
+	ErrTxAborted            = errors.New("transaction aborted")
+	ErrExclusionViolation   = errors.New("exclusion violation")
 )
 
 type (

connectors/storage/v2/storage_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/ice-blockchain/wintr/terror"
 )
@@ -140,3 +141,17 @@ $$ LANGUAGE plpgsql;`
 		return nil
 	}))
 }
+
+func TestStopWhenTxAborted(t *testing.T) {
+	t.Parallel()
+	db := MustConnect(context.Background(), "", "self")
+	require.NotNil(t, db)
+
+	err := DoInTransaction(context.Background(), db, func(tx QueryExecer) error {
+		_, gErr := Get[bool](context.Background(), tx, `SELECT $1 + $2`, 1, "2")
+
+		return gErr
+	})
+	require.ErrorIs(t, err, ErrTxAborted)
+	require.NoError(t, db.Close())
+}