idempotency

package
v0.1.0 Latest Latest
Warning

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

Go to latest
Published: Oct 27, 2026 License: MIT Imports: 31 Imported by: 0

Documentation

Overview

Package idempotency records one committed local outcome per retained scoped key. Callbacks use the supplied transaction for every business write and outbox enqueue. Network delivery and independently captured resources are not atomic.

Index

Constants

View Source
const CreateOutcomes migrate.ID = "000001_create_outcomes"
View Source
const IndexRetainedOutcomes migrate.ID = "000002_index_retained_outcomes"

IndexRetainedOutcomes replaces the caller index with one that also orders by expiry, so quota checks read at most MaxRetainedPerCaller unexpired rows. It builds concurrently, outside a transaction, without blocking writes. The build waits for older transactions, so it raises the session lock_timeout to indexBuildLockTimeout for its statements and then restores the runner's value. It first drops a leftover (possibly INVALID) index from an earlier attempt.

View Source
const MaxKeyBytes = 256
View Source
const MaxPrune = 1000

MaxPrune bounds rows removed by one Prune call or pruner statement.

View Source
const MigrationOrigin migrate.Origin = "foundry.idempotency"
View Source
const MinKeyBytes = 16

Variables

This section is empty.

Functions

func KeyPattern

func KeyPattern(maximum int) string

KeyPattern is shared with HTTP metadata and generated clients.

func Migrations

func Migrations() []migrate.Definition

Migrations describes the shared store. Install explicitly with ordinary migrations; constructors never mutate schemas or existing data.

func Module

func Module(name foundation.ProviderID, key foundation.Key[*Store], requires []foundation.ProviderID, construct func(foundation.Resolver) (*Store, error)) foundation.Module

Module owns admission and drains active operations before its borrowed database provider shuts down. It starts the store's expiry pruner (Config.PruneInterval) after registering cleanup, logging failures through the application logger unless the store was constructed WithLogger.

Types

type Code

type Code string

Code contains safe public outcome categories, never stored input or key data.

const (
	BadKey      Code = "idempotency_bad_key"
	Mismatch    Code = "idempotency_mismatch"
	InProgress  Code = "idempotency_in_progress"
	Capacity    Code = "idempotency_capacity"
	Unavailable Code = "idempotency_unavailable"
)

func (Code) Error

func (c Code) Error() string

type Config

type Config struct {
	Schema               string
	MaxActive            int
	MaxRetainedPerCaller int
	MaxKeyBytes          int
	MaxInputBytes        int
	MaxResultBytes       int
	Timeout              time.Duration
	DuplicateWait        time.Duration
	Retention            time.Duration
	PruneInterval        time.Duration
	PruneBatch           int
}

Config bounds one store. MaxRetainedPerCaller counts a caller's unexpired outcomes across its operations; expired outcomes no longer consume quota. The check is a bounded indexed read after the claim and does not serialize a caller's operations, so concurrently executing new operations of one caller can each pass it: retention may exceed the quota by at most that concurrency. Replays do not consume quota. MaxActive bounds this process's in-flight work.

PruneInterval schedules the store-owned pruner that Start (and Module) run; each sweep removes up to PruneBatch expired outcomes per statement. Zero disables automatic pruning; explicit Prune remains available.

func DefaultConfig

func DefaultConfig() Config

func (Config) Validate

func (c Config) Validate() error

type Definition

type Definition struct {
	ID      OperationID
	Version uint32
}

Definition versions both the operation's meaning and its wire fingerprint. A version change creates a new address and can execute an old client key again.

func (Definition) Validate

func (d Definition) Validate() error

type Encoding

type Encoding[T any] struct {
	// contains filtered or unexported fields
}

Encoding retains the concrete operation input/result at persistence boundaries. JSONEncoding reuses generated DTO metadata and codecs. Transport adapters may explicitly compose existing descriptors with DefineEncoding; they must include all semantic input and keep codecs deterministic, bounded and concurrency-safe.

func DefineEncoding

func DefineEncoding[T any](identity string, encode func(context.Context, T, int) ([]byte, error), decode func(context.Context, []byte, int) (T, error)) Encoding[T]

DefineEncoding is the explicit adapter boundary. identity is a versioned wire contract digest recorded with each outcome. An outcome stored under another identity replays only when decode accepts its exact representation, so decode must reject anything the current contract does not describe (unknown fields, missing required values); an incompatible outcome stays unavailable. Encoded values must be complete JSON documents, checked by the shared parser.

func JSONEncoding

func JSONEncoding[T any](descriptor contract.JSON[T]) Encoding[T]

func (Encoding[T]) Format

func (Encoding[T]) Format(s fmt.State, _ rune)

func (Encoding[T]) Validate

func (e Encoding[T]) Validate() error

type Error

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

Error preserves a private operational cause for diagnostics; never log Unwrap.

func (*Error) Error

func (e *Error) Error() string

func (*Error) Format

func (e *Error) Format(s fmt.State, _ rune)

func (*Error) Is

func (e *Error) Is(target error) bool

func (*Error) Unwrap

func (e *Error) Unwrap() error

type Handler

type Handler[I, R any] func(context.Context, *database.Tx, I) (R, error)

Handler receives the transaction that owns claim, writes, outbox and outcome. It must not commit an independently captured pool or call external providers.

type Input

type Input[T any] struct {
	// contains filtered or unexported fields
}

Input is the one-way, concrete fingerprint boundary. It never deserializes stored input because raw request input is not persisted.

func DefineInput

func DefineInput[T any](identity string, encode func(context.Context, T, int) ([]byte, error)) Input[T]

func JSONInput

func JSONInput[T any](descriptor contract.JSON[T]) Input[T]

func (Input[T]) Format

func (Input[T]) Format(s fmt.State, _ rune)

func (Input[T]) Validate

func (i Input[T]) Validate() error

type Key

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

Key is an opaque client submission identity. Parsing does not establish entropy. Use a random UUID or stronger random value, and reuse it for network retries.

func ParseKey

func ParseKey(text string) (Key, error)

func (Key) Format

func (Key) Format(s fmt.State, _ rune)

type Namespace

type Namespace = keyspace.Namespace

Namespace separates application deployments sharing the same store.

type Operation

type Operation[I, R any] struct {
	// contains filtered or unexported fields
}

func Define

func Define[I, R any](store *Store, definition Definition, input Input[I], output Encoding[R]) (Operation[I, R], error)

func (Operation[I, R]) Definition

func (o Operation[I, R]) Definition() Definition

func (Operation[I, R]) Format

func (Operation[I, R]) Format(s fmt.State, _ rune)

func (Operation[I, R]) Run

func (o Operation[I, R]) Run(ctx context.Context, scope Scope, key Key, input I, handler Handler[I, R]) (Result[R], error)

Run never retries the business callback. An uncertain commit is reconciled on the primary, or becomes Unavailable. Retry with the SAME scope/key and input. Authorization belongs before every Run, including those that only replay. Exhausted local admission (fault.Overloaded) returns a retryable Unavailable.

func (Operation[I, R]) Validate

func (o Operation[I, R]) Validate() error

type OperationID

type OperationID string

OperationID names a stable business operation, independently of its URL.

type Option

type Option func(*Store) error

Option configures a Store at construction.

func WithLogger

func WithLogger(logger *slog.Logger) Option

WithLogger borrows a logger for redacted background-pruning failures until Done closes. Module uses the application's configured logger by default.

type Result

type Result[R any] struct {
	// contains filtered or unexported fields
}

Result owns the committed representation. A non-nil error with Committed()==true reports a post-commit operational failure: publish the result and report the failure, but never repeat the callback. Values/bytes are never formatted.

func (Result[R]) Committed

func (r Result[R]) Committed() bool

func (Result[R]) Encoded

func (r Result[R]) Encoded() []byte

func (Result[R]) Format

func (Result[R]) Format(s fmt.State, _ rune)

func (Result[R]) Replayed

func (r Result[R]) Replayed() bool

func (Result[R]) Value

func (r Result[R]) Value() R

type Scope

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

Scope contains trusted tenant and caller identity. HTTP adapters derive this from verified authentication on every request; it is never read from the key. Webhook adapters verify the provider signature before constructing their scope.

func NewScope

func NewScope(tenant, caller string) (Scope, error)

func (Scope) Format

func (Scope) Format(s fmt.State, _ rune)

type Store

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

Store borrows an existing primary database and owns bounded operation lifetimes. New performs no I/O and starts no background task, migration or connection. Start (called by Module) begins the optional expiry pruner; Close stops it.

func New

func New(db *database.DB, namespace Namespace, config Config, options ...Option) (*Store, error)

func (*Store) Close

func (s *Store) Close(ctx context.Context) error

Close rejects new operations, cancels active ones and stops the pruner. A canceled caller stops waiting; owned callbacks and cleanup continue.

func (*Store) Config

func (s *Store) Config() Config

func (*Store) Database

func (s *Store) Database() *database.DB

func (*Store) Done

func (s *Store) Done() <-chan struct{}

Done closes after Close, once operations and the pruner have exited.

func (*Store) Format

func (*Store) Format(s fmt.State, _ rune)

func (*Store) Prune

func (s *Store) Prune(ctx context.Context, limit int) (int64, error)

Prune explicitly removes at most limit completed, expired outcomes belonging to this application namespace. It never removes an uncommitted claim. Parallel callers skip locked rows. A removed key may execute again; schedule deliberately. The count is zero if the transaction did not have a confirmed commit. Stores started by Start/Module also prune automatically (Config.PruneInterval).

func (*Store) Start

func (s *Store) Start(ctx context.Context) error

Start begins automatic pruning when Config.PruneInterval is positive. It performs no I/O; the first sweep runs one interval later. Repeated calls are harmless. Close stops the pruner and waits for any running sweep to exit.

func (*Store) Validate

func (s *Store) Validate() error

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL