store/transaction.go

Rama main · réplica de lectura del repositorio original.

package store

import (
	"context"
	"database/sql"
	stderrors "errors"
	"math/rand/v2"
	"time"

	"github.com/pkg/errors"
)

// serializableMaxAttempts bounds how many times runSerializable runs a
// transaction that conflicted with a concurrent one.
const serializableMaxAttempts = 5

// errTransactionConflict is returned by a transaction body that found a
// concurrent writer had changed what it read, so the body runs again.
var errTransactionConflict = stderrors.New("transaction conflicted with a concurrent write")

// runSerializable runs fn in one serializable transaction and commits it when
// fn returns nil. When the database reports a serialization or locking
// conflict, or fn returns errTransactionConflict, the transaction is rolled
// back and fn runs again, up to serializableMaxAttempts times, so fn must
// derive everything it writes from what it reads inside tx.
func (s *Store) runSerializable(ctx context.Context, fn func(tx *sql.Tx) error) error {
	var err error
	for attempt := range serializableMaxAttempts {
		err = s.runSerializableOnce(ctx, fn)
		retryable := errors.Is(err, errTransactionConflict) || s.driver.IsRetryableTransactionError(err)
		if err == nil || !retryable {
			return err
		}
		// Jitter keeps transactions that conflicted once from retrying in lockstep
		// and conflicting again.
		backoff := time.Duration(attempt+1)*10*time.Millisecond + rand.N(10*time.Millisecond)
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(backoff):
		}
	}
	return errors.Wrap(err, "serializable transaction failed after retries")
}

func (s *Store) runSerializableOnce(ctx context.Context, fn func(tx *sql.Tx) error) error {
	tx, err := s.driver.GetDB().BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return errors.Wrap(err, "failed to begin serializable transaction")
	}
	defer func() {
		_ = tx.Rollback()
	}()
	if err := fn(tx); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return errors.Wrap(err, "failed to commit serializable transaction")
	}
	return nil
}