idempotency

Documentation

Overview

Package idempotency prevents repeated execution using atomic persistence.

New constructs a Manager with a Store and Options for event keys, payload validation, expiry and optional local response caching. Typed handler wrapping and operation execution acquire a record before calling business code and replay a persisted response for completed duplicate events.

Persistence guarantees

Stores must atomically reject live claims and provide consistent snapshots. The dynamodb subpackage implements DynamoDB persistence; the independent idempotency/cache module supplies Redis/Valkey persistence. Provision shared storage for coordination across Lambda environments; the optional local cache alone is not a distributed lock.

Record stores expiry in seconds and execution lease expiry in milliseconds. Concurrent live operations produce AlreadyInProgressError. Payload mismatch and storage failures remain distinguishable through typed errors.

For usage and compatibility details, see the user guide.

Index

• Variables
• func CanonicalJSON(value any) ([]byte, error)
• func Execute[R any](ctx context.Context, m *Manager, payload any, ...) (result R, err error)
• func WrapHandler[T, R any](m *Manager, handler func(context.Context, T) (R, error), ...) func(context.Context, T) (R, error)
• type AlreadyExistsError
  • func (*AlreadyExistsError) Error() string
• type AlreadyInProgressError
  • func (e *AlreadyInProgressError) Error() string
• type ConfigurationError
  • func (e *ConfigurationError) Error() string
• type InconsistentStateError
  • func (e *InconsistentStateError) Error() string
• type InvalidStatusError
  • func (e *InvalidStatusError) Error() string
• type KeyError
  • func (e *KeyError) Error() string
  • func (e *KeyError) Unwrap() error
• type Manager
  • func New(store Store, options Options) (*Manager, error)
  • func (m *Manager) ClearCache()
  • func (m *Manager) Key(payload any) (key, validation string, skip bool, err error)
• type Options
• type PersistenceError
  • func (e *PersistenceError) Error() string
  • func (e *PersistenceError) Unwrap() error
• type Query
• type Record
  • func (r Record) Clone() Record
  • func (r Record) CurrentStatus(now time.Time) (Status, error)
  • func (r Record) IsExpired(now time.Time) bool
• type Status
• type Store
• type ValidationError
  • func (*ValidationError) Error() string

Examples

• Record.CurrentStatus

Variables

var ErrNotFound = errors.New("idempotency record not found")

Functions

func CanonicalJSON

func CanonicalJSON(value any) ([]byte, error)

CanonicalJSON implements the pinned deepSort + JSON.stringify contract for JSON-shaped inputs. RawMessage preserves object insertion order for keys whose lowercase forms compare equally; Go maps have no insertion order. Numbers use JavaScript's float64 domain. See docs/IDEMPOTENCY.md for limits.

func Execute

func Execute[R any](ctx context.Context, m *Manager, payload any, handler func(context.Context) (R, error), replayHook ...func(context.Context, R, Record) (R, error)) (result R, err error)

Execute makes one operation idempotent. The payload may be a projection of the handler arguments. The callback must honor context deadlines. A replay hook runs only for stored responses, and receives a private record snapshot.

func WrapHandler

func WrapHandler[T, R any](m *Manager, handler func(context.Context, T) (R, error), replayHook ...func(context.Context, R, Record) (R, error)) func(context.Context, T) (R, error)

Types

type AlreadyExistsError

type AlreadyExistsError struct{ Record *Record }

func (*AlreadyExistsError) Error

func (*AlreadyExistsError) Error() string

type AlreadyInProgressError

type AlreadyInProgressError struct{ Key string }

func (*AlreadyInProgressError) Error

func (e *AlreadyInProgressError) Error() string

type ConfigurationError

type ConfigurationError struct{ Message string }

func (*ConfigurationError) Error

func (e *ConfigurationError) Error() string

type InconsistentStateError

type InconsistentStateError struct{ Key string }

func (*InconsistentStateError) Error

func (e *InconsistentStateError) Error() string

type InvalidStatusError

type InvalidStatusError struct{ Status Status }

func (*InvalidStatusError) Error

func (e *InvalidStatusError) Error() string

type KeyError

type KeyError struct{ Err error }

func (*KeyError) Error

func (e *KeyError) Error() string

func (*KeyError) Unwrap

func (e *KeyError) Unwrap() error

type Manager

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

Manager owns immutable operation configuration and an optional response cache. Multiple managers may share a Store while retaining independent key prefixes, queries, deadlines, and caches. It never stores invocation context globally.

func New

func New(store Store, options Options) (*Manager, error)

func (*Manager) ClearCache

func (m *Manager) ClearCache()

func (*Manager) Key

func (m *Manager) Key(payload any) (key, validation string, skip bool, err error)

Key returns the persistence key, validation hash, and whether an absent query selection should bypass idempotency. False/zero/empty selections are hashed unless ThrowOnNoKey is enabled, matching the reference's two separate checks.

type Options

type Options struct {
	KeyPrefix                 string
	EventKeyJMESPath          string
	PayloadValidationJMESPath string
	KeyQuery                  Query
	ValidationQuery           Query
	ThrowOnNoKey              bool
	ExpiresAfter              *time.Duration
	HashFunction              string
	SerializeKey              func(any) ([]byte, error)
	UseLocalCache             bool
	MaxLocalCacheSize         int
	Disabled                  bool
	Now                       func() time.Time
	Diagnostic                func(string)
}

Options are snapshotted at construction. Query, serializer, clock, and diagnostic callbacks must be concurrency-safe. A nil duration uses one hour.

type PersistenceError

type PersistenceError struct {
	Operation string
	Err       error
}

func (*PersistenceError) Error

func (e *PersistenceError) Error() string

func (*PersistenceError) Unwrap

func (e *PersistenceError) Unwrap() error

type Query

type Query interface{ Search(any) (any, error) }

type Record

type Record struct {
	Key                  string            `json:"id"`
	Status               Status            `json:"status"`
	Expiration           int64             `json:"expiration,omitzero"`
	InProgressExpiration int64             `json:"in_progress_expiration,omitzero"`
	Data                 jsonv1.RawMessage `json:"data,omitempty"`
	Validation           string            `json:"validation,omitempty"`
}

Record uses the reference persistence field names. Expiration is in Unix seconds; InProgressExpiration is in Unix milliseconds. Zero means absent. Data holds JSON response bytes, not a JSON string containing those bytes.

func (Record) Clone

func (r Record) Clone() Record

func (Record) CurrentStatus

func (r Record) CurrentStatus(now time.Time) (Status, error)

Example

package main

import (
	"fmt"
	"time"

	"github.com/rambow-cloud/powertools-lambda-go/idempotency"
)

func main() {
	now := time.Unix(1_700_000_000, 0)
	record := idempotency.Record{Key: "order-123", Status: idempotency.Completed,
		Expiration: now.Add(time.Hour).Unix()}
	status, err := record.CurrentStatus(now)
	fmt.Println(status, err)
	status, err = record.CurrentStatus(now.Add(2 * time.Hour))
	fmt.Println(status, err)
}

Output:
COMPLETED <nil>
EXPIRED <nil>

func (Record) IsExpired

func (r Record) IsExpired(now time.Time) bool

type Status

type Status string

const (
	InProgress Status = "INPROGRESS"
	Completed  Status = "COMPLETED"
	Expired    Status = "EXPIRED"
)

type Store

type Store interface {
	Put(context.Context, Record, time.Time) error
	Get(context.Context, string) (Record, error)
	Update(context.Context, Record) error
	Delete(context.Context, string) error
}

Store owns atomic acquisition, response persistence, and cleanup. Put must reject a live existing record with AlreadyExistsError, optionally carrying its snapshot. Expired records and expired execution leases may be replaced. Get must provide a consistent snapshot. Implementations must be concurrency-safe. Update and Delete follow the reference's unconditional write semantics.

type ValidationError

type ValidationError struct{ Record Record }

func (*ValidationError) Error

func (*ValidationError) Error() string