outbox

type InMemoryStore ¶

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

InMemoryStore is a process-local, goroutine-safe reference Store backed by a map — for tests and single-process demos. It has NO real transaction: Save just appends, so it cannot deliver the atomic-with-the-business-write guarantee a production store gives; use a database-backed adapter in production.

It still faithfully models the relay contract: rows are pending until MarkPublished, FetchUnpublished returns them oldest-first, and MarkFailed bumps the attempt count and stores the last reason while leaving the row pending for retry. It does not implement claim/locking (FetchUnpublished's documented adapter concern) — it is single-process.

func NewInMemoryStore() *InMemoryStore

func (s *InMemoryStore) AttemptsOf(id string) int

func (s *InMemoryStore) FetchUnpublished(limit int) ([]Record, error)

func (s *InMemoryStore) LastErrorOf(id string) string

func (s *InMemoryStore) MarkFailed(id, reason string) error

func (s *InMemoryStore) MarkPublished(ids []string) error

func (s *InMemoryStore) PendingCount() int

func (s *InMemoryStore) Save(encoded []byte, queue string) (string, error)

type Options ¶

type Options struct {
	// BatchSize is how many rows to reserve and publish per [Relay.Flush] (default 100).
	BatchSize int
	// BackoffStep is the base backoff added per prior attempt (default 50ms).
	BackoffStep time.Duration
	// BackoffCap is the upper bound on a single backoff sleep (default 5s).
	BackoffCap time.Duration
	// Sleeper sleeps for d; defaults to time.Sleep. Inject a no-op (or recorder) in tests
	// so they stay instant.
	Sleeper func(d time.Duration)
}

Options configures a Relay. The zero value selects sensible defaults: a batch of 100, a 50ms backoff step, a 5s backoff cap, and time.Sleep as the sleeper.

type Outbox ¶

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

Outbox is the write side of the transactional outbox: it turns a babelqueue envelope into a stored outbox row, so the message is persisted atomically with the business data and a separate Relay publishes it later.

It is intentionally tiny and dependency-free: Outbox.Write encodes via the frozen envelope codec (GR-1 — the bytes are stored unchanged; the outbox never adds an envelope field) and delegates persistence to the injected Store, which the caller binds to their own DB (GR-7). It does NOT begin or commit anything — the caller owns the transaction boundary, which is the whole point of the pattern.

func New(store Store) *Outbox

func (o *Outbox) Write(env babelqueue.Envelope) (id string, err error)

type Record ¶

type Record struct {
	ID       string // the outbox row id (the store's primary key, not meta.id)
	Body     string // the frozen, encoded envelope JSON, byte-for-byte as stored
	Queue    string // the logical queue the relay should publish to
	Attempts int    // how many times the relay has already tried to publish this row
}

Record is one pending row read back from a Store for a Relay to publish. It pairs the store's bookkeeping (ID, Attempts) with the verbatim, frozen wire envelope (Body) and the queue it should go to. Body is the exact Encode output handed to Store.Save — the relay publishes these bytes unchanged (GR-1/GR-5), so trace_id is preserved end-to-end (GR-4) without the relay ever decoding the envelope.

type Relay ¶

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

Relay is the read/publish side of the transactional outbox (ADR-0029): it drains the pending rows an Outbox committed and forwards each onto the broker through the frozen babelqueue babelqueue.Transport seam — the same publish-only contract every framework-less producer uses — marking every row published or failed.

Run it on a short interval (a worker loop, a scheduled command) after the business transaction commits. It only ever reads already-durable rows, so it never invents work.

Semantics — at-least-once handoff:

• A row is marked published only AFTER babelqueue.Transport.Publish returns; if the process dies between publish and Store.MarkPublished, the row stays pending and is published again on the next pass. That is at-least-once: a downstream consumer must dedupe on the canonical meta.id (github.com/babelqueue/babelqueue-go/idempotency is that guard, the consumer-side mirror of this producer-side helper — ADR-0022).
• A publish that errors is caught, Store.MarkFailed records the reason and bumps the attempt count, and the row stays pending for a later retry. One poison row never blocks the rest of the batch.
• trace_id is preserved end-to-end (GR-4): the relay publishes the stored bytes verbatim — it never decodes, rebuilds or re-encodes the envelope — so the body that reaches the broker is byte-identical to what was stored (GR-1/GR-5).

Backoff: between a failed publish and continuing within the same pass, the relay sleeps for a bounded, linearly-growing delay (capped), to avoid hammering a broker that is briefly down. The sleeper is injectable so tests stay instant.

func NewRelay(transport babelqueue.Transport, store Store, opts Options) *Relay

func (r *Relay) Drain(ctx context.Context, maxPasses int) (Result, error)

func (r *Relay) Flush(ctx context.Context) (Result, error)

type Result ¶

type Result struct {
	Published int
	Failed    int
}

Result summarizes a relay pass: how many pending rows were published and how many failed (and were left pending for a later retry).

func (r Result) Attempted() int

type Store ¶

type Store interface {
	// Save persists one encoded envelope into the outbox, within the transaction the
	// caller has already opened around its business write, and returns the new row's
	// outbox id (the store's own primary key, NOT meta.id). The body is stored verbatim.
	Save(encoded []byte, queue string) (id string, err error)

	// FetchUnpublished reserves up to limit rows that are pending publish, oldest
	// first, so a relay can forward them. Implementations SHOULD lock/claim the rows
	// they return (e.g. SELECT … FOR UPDATE SKIP LOCKED, or a picked_at claim) so two
	// concurrent relays do not both publish the same row; at-least-once tolerates a rare
	// double send. The slice is empty when the outbox is drained.
	FetchUnpublished(limit int) ([]Record, error)

	// MarkPublished marks the given rows as successfully published (so they are never
	// relayed again). The relay calls it only after the transport accepted the message.
	MarkPublished(ids []string) error

	// MarkFailed records a failed publish attempt for one row: it increments the row's
	// attempt counter and stores the last reason, leaving the row pending so a later
	// relay pass retries it (at-least-once). A store MAY park a row that exceeds a
	// max-attempts threshold, but that policy is the adapter's, not the core's.
	MarkFailed(id, reason string) error
}

Store is the persistence seam for the transactional outbox (ADR-0029) — the durable "outbox" table an Outbox writer fills and a Relay drains. The core defines it and binds to NO database driver (GR-7); the caller binds a concrete adapter to their own connection, and the reference InMemoryStore backs tests.

Save runs inside the transaction the caller has already opened around its business write, so the message commits atomically with the business data. The body it stores is the frozen wire envelope, byte-for-byte unchanged — implementations must not re-encode or mutate it.