Documentation
¶
Overview ¶
Package batch provides the shared primitives for moving a batch through its lifecycle states while keeping the queue's per-state membership records (entity.QueueBatchState) in step.
The records are advisory and the Batch entity is authoritative, so the primitives follow one protocol:
- A transition CASes the batch first, then files the record under the new state before removing the one under the old state, so a batch always has at least one record while it is in the queue.
- A crash between the CAS and the record move is repaired by the pipeline's at-least-once redelivery: the retry's "already in target state" branch calls EnsureRecord, and every record write is idempotent.
- Readers treat records as candidate batch IDs only: they hydrate each batch by key and classify it by its own State, never by the bucket the record was found in, so a stale record can misplace a batch but never misreport it.
Index ¶
- func EnsureRecord(ctx context.Context, store storage.Storage, batch entity.Batch) error
- func FindByRequestID(ctx context.Context, store storage.Storage, requestID string) ([]entity.Batch, int, error)
- func ListByStates(ctx context.Context, store storage.Storage, states []entity.BatchState) ([]entity.Batch, error)
- func Transition(ctx context.Context, store storage.Storage, batch entity.Batch, ...) (entity.Batch, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func EnsureRecord ¶
EnsureRecord idempotently files the batch under its current state bucket. It is the repair half of the transition protocol: idempotent redelivery branches that skip the CAS because the batch is already in the target state call this instead, covering a prior attempt that crashed between the CAS and the record move.
func FindByRequestID ¶
func FindByRequestID(ctx context.Context, store storage.Storage, requestID string) ([]entity.Batch, int, error)
FindByRequestID resolves every batch attempt associated with a request, ordered by batch ID. The batches are independent of each other, but a deterministic order stabilizes logs, tests, and first-error selection.
An association whose batch row is missing is skipped and counted in stale rather than failing the call: the batch row and the association are separate writes, so an attempt that died between them leaves the association behind. The count is returned so callers can meter it without re-reading.
Unlike ListByStates, which treats a dangling membership record as store corruption, a dangling association is an expected retry artifact.
func ListByStates ¶
func ListByStates(ctx context.Context, store storage.Storage, states []entity.BatchState) ([]entity.Batch, error)
ListByStates returns the bound queue's batches whose current state is one of the given states, read through the queue's membership records: each requested state bucket is listed, candidate IDs are deduplicated across buckets, every candidate is hydrated by key with bounded concurrency, and the result keeps only batches whose hydrated State is in states. Classification always uses the hydrated state — a record found in a stale bucket can therefore never misreport a batch, only route an extra read. Result order is unspecified.
A candidate ID whose batch does not exist is returned as an error rather than skipped: batch rows are never deleted, so a dangling record means the store is inconsistent, not that the batch concluded.
func Transition ¶
func Transition(ctx context.Context, store storage.Storage, batch entity.Batch, newState entity.BatchState) (entity.Batch, error)
Transition moves a batch to newState: it performs the optimistic-locking CAS on the batch (newVersion = Version+1, assigned in memory only after the store write succeeds), then re-files the queue's membership record — Put under newState first, Delete under the prior state after, so the batch is never without a record. The Delete is skipped when the state is unchanged. It returns the batch as last successfully written.
A storage.ErrVersionMismatch from the CAS is returned wrapped (errors.Is works), with no record writes attempted, so callers keep their existing lost-race semantics. Any other non-nil error means the transition may have partially applied — the CAS may have committed with the record move incomplete — and the caller is expected to let redelivery retry; the retry's already-in-target-state branch repairs the record via EnsureRecord.
Types ¶
This section is empty.
Source Files