atom

package module
v0.0.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 16, 2026 License: Apache-2.0 Imports: 8 Imported by: 0

README

Atom

Atom implements the Transaction/Unit of Work pattern with compensating actions for ensuring atomicity in non-database operations like network requests, file I/O, and other side effects.

Features

  • Automatic Rollback: Failed operations trigger automatic rollback of all previous operations in reverse order (LIFO)
  • Panic Recovery: Panics in operations or compensating functions are caught and handled gracefully
  • Hooks System: Add callbacks at key points (BeforeCommit, AfterCommit, BeforeRollback, AfterRollback, BeforeOperation, AfterOperation)
  • Critical Hooks: Hooks can block operations (e.g., validation before commit)
  • Operation Metadata: Track execution time, errors, and operation history
  • Thread-Safe: Concurrent access is protected with RWMutex
  • Context Support: Full context propagation for cancellation and timeouts
  • Savepoints: Create savepoints for partial rollback
  • Operation Timeouts: Configure per-operation timeouts
  • Compensation Retry: Automatic retry with exponential backoff for failed compensations
  • Observability: Observer interface for metrics and tracing integration

Installation

go get go.alis.build/atom

Basic Usage

ctx := context.Background()
tx := atom.NewTransaction()

// Always ensure cleanup
defer func() {
    if !tx.IsCommitted() {
        _ = tx.Rollback(ctx)
    }
}()

// Execute operations with compensating functions
err := tx.Do(ctx, "create-file", 
    func(ctx context.Context) error {
        return os.WriteFile("/tmp/file.txt", []byte("data"), 0644)
    },
    func(ctx context.Context) error {
        return os.Remove("/tmp/file.txt")
    },
)
if err != nil {
    return err // Rollback happens automatically via defer
}

// More operations...

// Commit to finalize
return tx.Commit(ctx)

Operations with Options

Use functional options for advanced operation configuration:

err := tx.Do(ctx, "api-call",
    func(ctx context.Context) error {
        return callExternalAPI(ctx)
    },
    func(ctx context.Context) error {
        return rollbackAPICall(ctx)
    },
    atom.WithTimeout(5*time.Second),
    atom.WithCompensationRetry(&atom.RetryOptions{
        MaxRetries:        3,
        InitialDelay:      100 * time.Millisecond,
        BackoffMultiplier: 2.0,
        MaxDelay:          5 * time.Second,
    }),
)

Savepoints

Create savepoints for partial rollback:

tx := atom.NewTransaction()

// First batch of operations
_ = tx.Do(ctx, "op1", op1Func, comp1Func)
_ = tx.Do(ctx, "op2", op2Func, comp2Func)

// Create a savepoint
sp := tx.CreateSavepoint("after-batch-1")

// Second batch of operations
_ = tx.Do(ctx, "op3", op3Func, comp3Func)
_ = tx.Do(ctx, "op4", op4Func, comp4Func)

// Rollback only the second batch
if someCondition {
    err := tx.RollbackToSavepoint(ctx, sp)
    // op3 and op4 are rolled back, op1 and op2 remain
}

// Continue with more operations or commit
_ = tx.Commit(ctx)

Hooks

Hook Types
  • BeforeCommit: Called before commit (critical by default - failures block commit)
  • AfterCommit: Called after successful commit (non-critical)
  • BeforeRollback: Called before rollback (non-critical)
  • AfterRollback: Called after rollback (non-critical)
  • BeforeOperation: Called before each operation (critical by default)
  • AfterOperation: Called after each operation (non-critical)
Adding Hooks
// Non-critical hook (errors logged but don't block)
tx.AddHook(atom.AfterCommit, func(ctx context.Context, t *atom.Transaction) error {
    fmt.Println("Transaction committed!")
    return nil
})

// Critical hook (errors block the operation)
tx.AddCriticalHook(atom.BeforeCommit, func(ctx context.Context, t *atom.Transaction) error {
    // Validation logic
    if someCondition {
        return errors.New("validation failed")
    }
    return nil
})

// Hook with default criticality based on type
tx.AddDefaultHook(atom.BeforeCommit, validationFunc)

// Operation-specific hooks
tx.AddOperationHook(atom.BeforeOperation, func(ctx context.Context, tx *atom.Transaction, opName string, opIndex int) error {
    log.Printf("Starting operation %s (index %d)", opName, opIndex)
    return nil
})
Hook Management
// Remove all hooks of a specific type
tx.ClearHooks(atom.BeforeCommit)

// Remove all hooks
tx.ClearAllHooks()

Observability

Integrate with metrics and tracing systems using the Observer interface:

// Define a custom observer
type MyObserver struct {
    metrics *prometheus.Registry
}

func (o *MyObserver) OnOperationStart(ctx context.Context, name string) {
    // Record operation start
}

func (o *MyObserver) OnOperationEnd(ctx context.Context, name string, duration time.Duration, err error) {
    // Record operation metrics
}

func (o *MyObserver) OnCommit(ctx context.Context) {
    // Record commit
}

func (o *MyObserver) OnRollback(ctx context.Context, errors []error) {
    // Record rollback
}

// Use the observer
tx := atom.NewTransaction(atom.WithObserver(&MyObserver{}))

Built-in observers:

  • NoOpObserver: Does nothing (useful for testing)
  • MetricsObserver: Collects basic metrics (operation count, success/failure, duration)

Logging

By default, the transaction does not log anything. You can optionally provide a *slog.Logger to enable structured logging:

import "log/slog"

// Enable logging with the default slog logger
tx := atom.NewTransaction(atom.WithLogger(slog.Default()))

// Or use a custom logger
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
    Level: slog.LevelDebug,
}))
tx := atom.NewTransaction(atom.WithLogger(logger))

When a logger is configured, the transaction will log:

  • Warnings for failed compensations during rollback
  • Warnings for non-critical hook failures
  • Info messages for compensation retries

Error Handling

The package provides specific error types:

import "go.alis.build/atom/errors"

// Check error types
var opErr *errors.OperationError
if errors.As(err, &opErr) {
    fmt.Printf("Operation '%s' failed\n", opErr.Operation)
}

var panicErr *errors.PanicError
if errors.As(err, &panicErr) {
    fmt.Printf("Panic occurred: %v\n", panicErr.Value)
}

// Check for critical hook failures
if errors.Is(err, errors.ErrHookFailed) {
    fmt.Println("A critical hook failed")
}

// RollbackError supports multi-error unwrapping (Go 1.20+)
var rollbackErr *errors.RollbackError
if errors.As(err, &rollbackErr) {
    for _, e := range rollbackErr.Unwrap() {
        fmt.Printf("Compensation error: %v\n", e)
    }
}

Utility Methods

// Check transaction state
tx.IsCommitted()   // true if committed
tx.IsRolledBack()  // true if rolled back
tx.IsPending()     // true if neither committed nor rolled back

// Get operation count
count := tx.OperationCount()

// Get operation history
history := tx.GetHistory()
for _, op := range history {
    fmt.Printf("Operation: %s, Duration: %v, Error: %v\n", op.Name, op.Duration, op.Error)
}

Best Practices

  1. Always use defer for cleanup

    defer func() {
    if !tx.IsCommitted() {
        _ = tx.Rollback(ctx)
    }
}()

  2. Make compensating functions idempotent - They might be called multiple times (especially with retry enabled)

  3. Handle partial failures gracefully - Rollback does best-effort cleanup

  4. Use operation names - Helpful for debugging and logging

  5. Validate before commit - Use BeforeCommit hooks for final validation

  6. Don't ignore rollback errors - Log them for debugging

  7. Use timeouts for external calls - Prevent operations from hanging indefinitely

  8. Configure retry for flaky compensations - Network operations may need retries

Thread Safety

The transaction is thread-safe for concurrent access to its methods. The implementation uses RWMutex and atomic operations to ensure safe concurrent access. However, the operations and compensating functions you provide should handle their own thread safety if needed.

Limitations

  • Compensating functions are best-effort - if they fail after all retries, the error is logged but rollback continues
  • Not suitable for distributed transactions across multiple services
  • Operations should be relatively quick - long-running operations should use context cancellation
  • Savepoints cannot be used after commit or rollback

Documentation

Overview

Package atom implements the Transaction/Unit of Work pattern with compensating actions for ensuring atomicity in non-database operations like network requests, file I/O, and other side effects.

Overview

Atom provides a way to group multiple operations into a single unit of work that can be rolled back if any operation fails. Each operation is paired with a compensating function that undoes its effects.

Basic Usage 

tx := atom.NewTransaction()
defer func() {
	if !tx.IsCommitted() {
		_ = tx.Rollback(ctx)
	}
}()

err := tx.Do(ctx, "create-file",
	func(ctx context.Context) error {
		return os.WriteFile("/tmp/file.txt", []byte("data"), 0644)
	},
	func(ctx context.Context) error {
		return os.Remove("/tmp/file.txt")
	},
)
if err != nil {
	return err
}

return tx.Commit(ctx)

Operations with Options

Use functional options for advanced configuration including timeouts and retry: 

err := tx.Do(ctx, "api-call", operationFunc, compensateFunc,
	atom.WithTimeout(5*time.Second),
	atom.WithCompensationRetry(atom.DefaultRetryOptions()))

Savepoints

Create savepoints for partial rollback: 

sp := tx.CreateSavepoint("checkpoint")
// ... more operations ...
tx.RollbackToSavepoint(ctx, sp) // rolls back only operations after savepoint

Hooks

Register callbacks at various points in the transaction lifecycle: 

tx.AddHook(atom.BeforeCommit, func(ctx context.Context, tx *atom.Transaction) error {
	// validation logic
	return nil
})

Hook types: BeforeCommit, AfterCommit, BeforeRollback, AfterRollback, BeforeOperation, AfterOperation.

Observability

Implement the Observer interface for metrics and tracing: 

tx := atom.NewTransaction(atom.WithObserver(&MyObserver{}))

Logging

Optionally configure a *slog.Logger for structured logging: 

tx := atom.NewTransaction(atom.WithLogger(slog.Default()))

By default, no logging is performed. When a logger is set, warnings and info messages are logged for rollback failures, hook failures, and retry attempts.

Error Types

The errors subpackage provides typed errors:

  • OperationError: wraps errors from operation execution
  • RollbackError: wraps errors from compensation (supports multi-error unwrap)
  • HookError: wraps errors from hook execution
  • PanicError: wraps recovered panics

Thread Safety

All Transaction methods are safe for concurrent use. The implementation uses RWMutex and atomic operations internally.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type CompensatingFunc 

type CompensatingFunc func(context.Context) error

CompensatingFunc is a function that undoes/compensates for an operation

type Hook 

type Hook struct {
	Fn       HookFunc
	Critical bool // If true, error blocks the operation
}

Hook represents a registered hook with its criticality setting

type HookFunc 

type HookFunc func(context.Context, *Transaction) error

HookFunc is a function that gets called at specific points in the transaction lifecycle

type HookType 

type HookType int

HookType represents the type of hook to execute 

const (
	// BeforeCommit is called before committing the transaction
	// Critical by default - failures block commit
	BeforeCommit HookType = iota

	// AfterCommit is called after successful commit
	// Non-critical by default - already committed, can't undo
	AfterCommit

	// BeforeRollback is called before rolling back operations
	// Non-critical by default - we're already in failure mode
	BeforeRollback

	// AfterRollback is called after rollback completes
	// Non-critical by default - rollback already done
	AfterRollback

	// BeforeOperation is called before each operation executes
	// Critical by default - failures block the operation
	BeforeOperation

	// AfterOperation is called after each operation completes (success or failure)
	// Non-critical by default - operation already executed
	AfterOperation
)

func (HookType) String 

func (h HookType) String() string

String returns the string representation of HookType

type LoggingObserver 

type LoggingObserver struct{}

LoggingObserver is an Observer implementation that logs events using alog

func NewLoggingObserver 

func NewLoggingObserver() *LoggingObserver

NewLoggingObserver creates a new LoggingObserver

func (*LoggingObserver) OnCommit 

func (l *LoggingObserver) OnCommit(ctx context.Context)

func (*LoggingObserver) OnOperationEnd 

func (l *LoggingObserver) OnOperationEnd(ctx context.Context, name string, duration time.Duration, err error)

func (*LoggingObserver) OnOperationStart 

func (l *LoggingObserver) OnOperationStart(ctx context.Context, name string)

func (*LoggingObserver) OnRollback 

func (l *LoggingObserver) OnRollback(ctx context.Context, errors []error)

type MetricsObserver 

type MetricsObserver struct {
	OperationCount     int64
	SuccessCount       int64
	FailureCount       int64
	TotalDuration      time.Duration
	CommitCount        int64
	RollbackCount      int64
	RollbackErrorCount int64
}

MetricsObserver is a sample Observer that collects basic metrics This is a reference implementation; production use would integrate with actual metrics systems like Prometheus, OpenTelemetry, etc.

func NewMetricsObserver 

func NewMetricsObserver() *MetricsObserver

NewMetricsObserver creates a new MetricsObserver

func (*MetricsObserver) OnCommit 

func (m *MetricsObserver) OnCommit(ctx context.Context)

func (*MetricsObserver) OnOperationEnd 

func (m *MetricsObserver) OnOperationEnd(ctx context.Context, name string, duration time.Duration, err error)

func (*MetricsObserver) OnOperationStart 

func (m *MetricsObserver) OnOperationStart(ctx context.Context, name string)

func (*MetricsObserver) OnRollback 

func (m *MetricsObserver) OnRollback(ctx context.Context, errors []error)

type NoOpObserver 

type NoOpObserver struct{}

NoOpObserver is an Observer implementation that does nothing Useful as a default or for testing

func (NoOpObserver) OnCommit 

func (NoOpObserver) OnCommit(ctx context.Context)

func (NoOpObserver) OnOperationEnd 

func (NoOpObserver) OnOperationEnd(ctx context.Context, name string, duration time.Duration, err error)

func (NoOpObserver) OnOperationStart 

func (NoOpObserver) OnOperationStart(ctx context.Context, name string)

func (NoOpObserver) OnRollback 

func (NoOpObserver) OnRollback(ctx context.Context, errors []error)

type Observer 

type Observer interface {
	// OnOperationStart is called before an operation begins execution
	OnOperationStart(ctx context.Context, name string)

	// OnOperationEnd is called after an operation completes (success or failure)
	OnOperationEnd(ctx context.Context, name string, duration time.Duration, err error)

	// OnCommit is called after a successful commit
	OnCommit(ctx context.Context)

	// OnRollback is called after a rollback completes
	// errors contains any errors that occurred during compensation
	OnRollback(ctx context.Context, errors []error)
}

Observer is an interface for observing transaction lifecycle events Implementations can be used for metrics, tracing, logging, or other observability needs

type OperationFunc 

type OperationFunc func(context.Context) error

OperationFunc is the main operation to execute

type OperationHook 

type OperationHook struct {
	Fn       OperationHookFunc
	Critical bool
}

OperationHook represents a registered operation hook with its criticality setting

type OperationHookFunc 

type OperationHookFunc func(ctx context.Context, tx *Transaction, opName string, opIndex int) error

OperationHookFunc is a function that gets called before/after each operation It receives the operation name and index in addition to the standard hook parameters

type OperationOption added in v0.0.2 

type OperationOption func(*operationOptions)

OperationOption is a functional option for the Do method.

func WithCompensationRetry added in v0.0.2 

func WithCompensationRetry(r *RetryOptions) OperationOption

WithCompensationRetry configures retry behavior for the compensating function.

func WithTimeout added in v0.0.2 

func WithTimeout(d time.Duration) OperationOption

WithTimeout sets a timeout for the operation (0 means no timeout).

type OperationRecord 

type OperationRecord struct {
	Name       string
	ExecutedAt time.Time
	Duration   time.Duration
	Error      error
}

OperationRecord is the public view of operation metadata

type RetryOptions 

type RetryOptions struct {
	// MaxRetries is the maximum number of retry attempts (0 means no retries)
	MaxRetries int
	// InitialDelay is the delay before the first retry
	InitialDelay time.Duration
	// BackoffMultiplier is the multiplier for exponential backoff (e.g., 2.0 doubles delay each retry)
	BackoffMultiplier float64
	// MaxDelay is the maximum delay between retries
	MaxDelay time.Duration
}

RetryOptions configures retry behavior for compensating functions

func DefaultRetryOptions 

func DefaultRetryOptions() *RetryOptions

DefaultRetryOptions returns sensible default retry options

type Savepoint 

type Savepoint struct {
	// contains filtered or unexported fields
}

Savepoint represents a point in the transaction that can be rolled back to

func (*Savepoint) Index 

func (sp *Savepoint) Index() int

Index returns the operation index at the time of savepoint creation

func (*Savepoint) Name 

func (sp *Savepoint) Name() string

Name returns the savepoint name

type Transaction 

type Transaction struct {
	// contains filtered or unexported fields
}

Transaction manages a sequence of operations with rollback capability

func NewTransaction 

func NewTransaction(opts ...TransactionOption) *Transaction

NewTransaction creates a new transaction with the given options.

func (*Transaction) AddCriticalHook 

func (tx *Transaction) AddCriticalHook(hookType HookType, fn HookFunc)

AddCriticalHook adds a critical hook that will block the operation if it fails

func (*Transaction) AddCriticalOperationHook 

func (tx *Transaction) AddCriticalOperationHook(hookType HookType, fn OperationHookFunc)

AddCriticalOperationHook adds a critical operation hook that blocks the operation if it fails

func (*Transaction) AddDefaultHook 

func (tx *Transaction) AddDefaultHook(hookType HookType, fn HookFunc)

AddDefaultHook adds a hook with default criticality based on hook type

func (*Transaction) AddHook 

func (tx *Transaction) AddHook(hookType HookType, fn HookFunc)

AddHook adds a non-critical hook to the transaction

func (*Transaction) AddOperationHook 

func (tx *Transaction) AddOperationHook(hookType HookType, fn OperationHookFunc)

AddOperationHook adds a non-critical operation hook Operation hooks are called before/after each Do() call

func (*Transaction) ClearAllHooks 

func (tx *Transaction) ClearAllHooks()

ClearAllHooks removes all registered hooks

func (*Transaction) ClearHooks 

func (tx *Transaction) ClearHooks(hookType HookType)

ClearHooks removes all hooks of a specific type

func (*Transaction) Commit 

func (tx *Transaction) Commit(ctx context.Context) error

Commit finalizes the transaction, making rollback no longer possible It executes BeforeCommit and AfterCommit hooks

func (*Transaction) CreateSavepoint 

func (tx *Transaction) CreateSavepoint(name string) *Savepoint

CreateSavepoint creates a savepoint at the current position in the transaction This allows partial rollback to this point using RollbackToSavepoint

func (*Transaction) Do 

func (tx *Transaction) Do(ctx context.Context, name string, operation OperationFunc, compensate CompensatingFunc, opts ...OperationOption) error

Do executes an operation and registers its compensating function. If the operation fails, it automatically triggers a rollback. The name parameter is optional (can be empty string). Options like timeout and compensation retry can be passed as functional options.

func (*Transaction) GetHistory 

func (tx *Transaction) GetHistory() []OperationRecord

GetHistory returns a copy of the operation history for debugging

func (*Transaction) GetLogger 

func (tx *Transaction) GetLogger() *slog.Logger

GetLogger returns the current logger, if any

func (*Transaction) GetObserver 

func (tx *Transaction) GetObserver() Observer

GetObserver returns the current observer, if any

func (*Transaction) IsCommitted 

func (tx *Transaction) IsCommitted() bool

IsCommitted returns whether the transaction has been committed

func (*Transaction) IsPending 

func (tx *Transaction) IsPending() bool

IsPending returns true if the transaction is neither committed nor rolled back

func (*Transaction) IsRolledBack 

func (tx *Transaction) IsRolledBack() bool

IsRolledBack returns whether the transaction has been rolled back

func (*Transaction) OperationCount 

func (tx *Transaction) OperationCount() int

OperationCount returns the number of recorded operations

func (*Transaction) Rollback 

func (tx *Transaction) Rollback(ctx context.Context) error

Rollback executes all compensating functions in reverse order (LIFO) It executes BeforeRollback and AfterRollback hooks

func (*Transaction) RollbackToSavepoint 

func (tx *Transaction) RollbackToSavepoint(ctx context.Context, sp *Savepoint) error

RollbackToSavepoint rolls back operations to the specified savepoint Operations after the savepoint are compensated in reverse order (LIFO) The transaction remains pending and can continue with new operations

type TransactionOption added in v0.0.2 

type TransactionOption func(*transactionOptions)

TransactionOption is a functional option for NewTransaction.

func WithLogger added in v0.0.2 

func WithLogger(logger *slog.Logger) TransactionOption

WithLogger sets the logger for the transaction. If set, the transaction will log warnings and info messages using this logger.

func WithObserver added in v0.0.2 

func WithObserver(obs Observer) TransactionOption

WithObserver sets the observer for the transaction. Only one observer can be set; setting a new one replaces the previous.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.