Documentation
¶
Index ¶
- Constants
- func Backoff(id string, attempts int, base, max time.Duration, jitter float64) time.Duration
- func QuarantineSnapshot(path string) error
- func ReadGzipJSON(path string, value any) error
- func RemoveSnapshot(path string) error
- func WriteGzipJSON(path string, value any) error
- func WriteSnapshot[T any](path string, snapshot Snapshot[T]) error
- type ClearResult
- type Clock
- type EnqueueResult
- type Item
- type Limits
- type Queue
- func (q *Queue[T]) Ack(ids ...string)
- func (q *Queue[T]) ClearBacklog() ClearResult
- func (q *Queue[T]) Enqueue(value T) EnqueueResult
- func (q *Queue[T]) Generation() uint64
- func (q *Queue[T]) Items(states ...State) []Item[T]
- func (q *Queue[T]) Put(item Item[T], state State) EnqueueResult
- func (q *Queue[T]) PutWithEvicted(item Item[T], state State) (EnqueueResult, []Item[T])
- func (q *Queue[T]) Remove(ids ...string) int
- func (q *Queue[T]) RemoveMatching(match func(Item[T]) bool) int
- func (q *Queue[T]) Restore(snapshot Snapshot[T]) error
- func (q *Queue[T]) Retry(ids []string, cause error)
- func (q *Queue[T]) RetryNow()
- func (q *Queue[T]) RetryWithBackoff(ids []string, cause error, base, max time.Duration)
- func (q *Queue[T]) SetNextAttemptMatching(match func(Item[T]) bool, next time.Time) int
- func (q *Queue[T]) Snapshot() Snapshot[T]
- func (q *Queue[T]) Stats() Stats
- func (q *Queue[T]) TakeReady(limit int) []Item[T]
- func (q *Queue[T]) Update(id string, update func(*Item[T])) bool
- func (q *Queue[T]) UpdateLimits(limits Limits)
- func (q *Queue[T]) UpdateMatching(match func(Item[T]) bool, update func(*Item[T])) int
- type Snapshot
- type SnapshotItem
- type Snapshotter
- type State
- type Stats
Constants ¶
const DefaultSnapshotInterval = 60 * time.Second
const PoisonIsolationAttempts = 3
PoisonIsolationAttempts is the number of failed deliveries after which an item is delivered alone. It remains retryable and never enters a dead-letter collection, so one bad item cannot block later work or grow storage forever.
Variables ¶
This section is empty.
Functions ¶
func Backoff ¶
Backoff returns a capped exponential delay with deterministic bounded jitter. Deterministic jitter keeps retry schedules stable across tests and snapshots while still preventing every item in a failed batch from retrying together.
func QuarantineSnapshot ¶
QuarantineSnapshot moves an unsupported but otherwise decodable snapshot out of the active path using the same durable rename as decode failures.
func ReadGzipJSON ¶
ReadGzipJSON decodes a snapshot and quarantines unreadable content. Missing files are returned as os.ErrNotExist so callers can treat first start as a no-op without hiding real I/O errors.
func RemoveSnapshot ¶
func WriteGzipJSON ¶
WriteGzipJSON atomically persists a caller-owned snapshot envelope. It is exported so protocol adapters can retain their existing on-disk schema while sharing the durability sequence with the generic queue.
Types ¶
type EnqueueResult ¶
type Queue ¶
type Queue[T any] struct { // contains filtered or unexported fields }
func (*Queue[T]) ClearBacklog ¶
func (q *Queue[T]) ClearBacklog() ClearResult
func (*Queue[T]) Enqueue ¶
func (q *Queue[T]) Enqueue(value T) EnqueueResult
func (*Queue[T]) Generation ¶
func (*Queue[T]) Items ¶
Items returns a stable copy in enqueue order. With no states it returns all states; otherwise it returns only the requested states.
func (*Queue[T]) Put ¶
func (q *Queue[T]) Put(item Item[T], state State) EnqueueResult
Put restores or adapts an item with caller-owned delivery metadata. New producers should normally use Enqueue; adapters use Put to preserve stable IDs, retry attempts and retry deadlines from an existing protocol.
func (*Queue[T]) PutWithEvicted ¶ added in v0.0.16
func (q *Queue[T]) PutWithEvicted(item Item[T], state State) (EnqueueResult, []Item[T])
PutWithEvicted has the same admission contract as Put and additionally returns the actual pending/retry items removed while enforcing limits.
func (*Queue[T]) Remove ¶
Remove deletes matching non-inflight items. Inflight work remains protected from cancellation by capacity and management operations.
func (*Queue[T]) RetryWithBackoff ¶
RetryWithBackoff lets a delivery worker apply runtime retry settings while preserving the queue's deterministic jitter and poison isolation rules.
func (*Queue[T]) SetNextAttemptMatching ¶
SetNextAttemptMatching updates retry scheduling metadata without exposing a payload mutation path that could bypass byte-limit enforcement.
func (*Queue[T]) Update ¶
Update applies a small adapter-specific value/metadata update while keeping queue synchronization internal.
func (*Queue[T]) UpdateLimits ¶
UpdateLimits applies runtime capacity settings and evicts the oldest pending/retry items until the queue fits. Inflight deliveries remain owned by their current writer and are never canceled.
type Snapshot ¶
type SnapshotItem ¶
type Snapshotter ¶
type Snapshotter[T any] struct { Queue *Queue[T] Path string Interval time.Duration OnError func(error) // contains filtered or unexported fields }
Snapshotter periodically persists a generic queue and performs one final write after cancellation. Delivery workers remain responsible for ordering their own shutdown drain before canceling this runner.
func (*Snapshotter[T]) Restore ¶
func (s *Snapshotter[T]) Restore() error
func (*Snapshotter[T]) Run ¶
func (s *Snapshotter[T]) Run(ctx context.Context)
func (*Snapshotter[T]) WriteNow ¶
func (s *Snapshotter[T]) WriteNow() error
Source Files