deliveryqueue

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 New[T any](limits Limits, sizeOf func(T) int64, clock Clock) *Queue[T]
  • 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
  • func ReadSnapshot[T any](path string) (Snapshot[T], error)
• type SnapshotItem
• type Snapshotter
  • func (s *Snapshotter[T]) Restore() error
  • func (s *Snapshotter[T]) Run(ctx context.Context)
  • func (s *Snapshotter[T]) WriteNow() error
• 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.

Functions

func Backoff

func Backoff(id string, attempts int, base, max time.Duration, jitter float64) time.Duration

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

func QuarantineSnapshot(path string) error

QuarantineSnapshot moves an unsupported but otherwise decodable snapshot out of the active path using the same durable rename as decode failures.

func ReadGzipJSON

func ReadGzipJSON(path string, value any) error

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 RemoveSnapshot(path string) error

func WriteGzipJSON

func WriteGzipJSON(path string, value any) error

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.

func WriteSnapshot

func WriteSnapshot[T any](path string, snapshot Snapshot[T]) error

Types

type ClearResult

type ClearResult struct {
	Items   int
	Pending int
	Retry   int
	Bytes   int64
}

type Clock

type Clock interface {
	Now() time.Time
}

type EnqueueResult

type EnqueueResult struct {
	ID       string
	Accepted bool
	Dropped  bool
	Conflict bool
	Error    string
}

type Item

type Item[T any] struct {
	ID          string    `json:"id"`
	Value       T         `json:"value"`
	Bytes       int64     `json:"bytes"`
	Attempts    int       `json:"attempts"`
	NextAttempt time.Time `json:"next_attempt,omitempty"`
}

type Limits

type Limits struct {
	MaxEntries int
	MaxBytes   int64
}

type Queue

type Queue[T any] struct {
	// contains filtered or unexported fields
}

func New

func New[T any](limits Limits, sizeOf func(T) int64, clock Clock) *Queue[T]

func (*Queue[T]) Ack

func (q *Queue[T]) Ack(ids ...string)

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 (q *Queue[T]) Generation() uint64

func (*Queue[T]) Items

func (q *Queue[T]) Items(states ...State) []Item[T]

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

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

func (q *Queue[T]) Remove(ids ...string) int

Remove deletes matching non-inflight items. Inflight work remains protected from cancellation by capacity and management operations.

func (*Queue[T]) RemoveMatching

func (q *Queue[T]) RemoveMatching(match func(Item[T]) bool) int

func (*Queue[T]) Restore

func (q *Queue[T]) Restore(snapshot Snapshot[T]) error

func (*Queue[T]) Retry

func (q *Queue[T]) Retry(ids []string, cause error)

func (*Queue[T]) RetryNow

func (q *Queue[T]) RetryNow()

func (*Queue[T]) RetryWithBackoff

func (q *Queue[T]) RetryWithBackoff(ids []string, cause error, base, max time.Duration)

RetryWithBackoff lets a delivery worker apply runtime retry settings while preserving the queue's deterministic jitter and poison isolation rules.

func (*Queue[T]) SetNextAttemptMatching

func (q *Queue[T]) SetNextAttemptMatching(match func(Item[T]) bool, next time.Time) int

SetNextAttemptMatching updates retry scheduling metadata without exposing a payload mutation path that could bypass byte-limit enforcement.

func (*Queue[T]) Snapshot

func (q *Queue[T]) Snapshot() Snapshot[T]

func (*Queue[T]) Stats

func (q *Queue[T]) Stats() Stats

func (*Queue[T]) TakeReady

func (q *Queue[T]) TakeReady(limit int) []Item[T]

func (*Queue[T]) Update

func (q *Queue[T]) Update(id string, update func(*Item[T])) bool

Update applies a small adapter-specific value/metadata update while keeping queue synchronization internal.

func (*Queue[T]) UpdateLimits

func (q *Queue[T]) UpdateLimits(limits Limits)

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.

func (*Queue[T]) UpdateMatching

func (q *Queue[T]) UpdateMatching(match func(Item[T]) bool, update func(*Item[T])) int

type Snapshot

type Snapshot[T any] struct {
	Version    int               `json:"version"`
	SavedAt    time.Time         `json:"saved_at"`
	NextID     uint64            `json:"next_id"`
	Generation uint64            `json:"generation,omitempty"`
	Items      []SnapshotItem[T] `json:"items"`
}

func ReadSnapshot

func ReadSnapshot[T any](path string) (Snapshot[T], error)

type SnapshotItem

type SnapshotItem[T any] struct {
	Item      Item[T]   `json:"item"`
	State     State     `json:"state"`
	CreatedAt time.Time `json:"created_at"`
}

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

type State

type State string

const (
	Pending  State = "pending"
	Retry    State = "retry"
	Inflight State = "inflight"
)

type Stats

type Stats struct {
	Pending   int
	Retry     int
	Inflight  int
	Bytes     uint64
	Dropped   uint64
	OldestAge time.Duration
	LastError string
}