Something went wrong. Try again.
wip outbox library design
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253package outpost
import ( "context" "errors" "time")
// Appender persists messages. A backend adapter binds it to a unit of work, so// the core never sees driver types: for example InTx(tx) for a SQL// transaction, InBatch(batch) for CQL, InTxn(txn) for Kafka, or AppenderFunc// for tests and for backends without a transaction.//// Every message must carry a stable, unique ID. Emit generates one when the// caller does not; code appending directly must supply one, ideally derived// from the business fact so a redelivery stays idempotent.type Appender interface { Append(ctx context.Context, msgs ...Message) error}
// AppenderFunc adapts a function to Appender.type AppenderFunc func(ctx context.Context, msgs ...Message) error
// Append implements Appender.func (f AppenderFunc) Append(ctx context.Context, msgs ...Message) error { return f(ctx, msgs...)}
// ErrStaleClaim reports that a lease is no longer current: the message has// been completed or abandoned, or another relay claimed it after this lease// expired. Complete and Fail return it when the token does not match. The// relay treats it as a no-op.var ErrStaleClaim = errors.New("outpost: stale claim")
// Store is the relay's view of a backend. Implementations must guarantee://// - Claim returns at most limit messages, in arrival order, one row per// message. A message is eligible when it is not terminal, its AvailableAt// is zero or due, and it is not currently leased.// - Claim leases every returned row for the requested duration and assigns// each a token unique to that lease.// - Complete marks the message terminal-successful.// - Fail records the cause and increments the message's failure count. A nil// retryAt abandons the message; otherwise the message becomes eligible no// earlier than *retryAt.// - Complete and Fail affect only the row whose current token matches the// ref. On mismatch they return ErrStaleClaim and change nothing.type Store interface { Claim(ctx context.Context, limit int, lease time.Duration) ([]Claim, error) Complete(ctx context.Context, refs ...ClaimRef) error Fail(ctx context.Context, ref ClaimRef, cause error, retryAt *time.Time) error}