queue

package
v0.0.0-...-f02db28 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 23, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Overview

Package queue owns the queue driver abstraction and its Redis Streams MVP implementation (FR-003, FR-014, FR-016; ADR-003 five-mechanism model):

  1. Message delivery — at-least-once via consumer group export-worker (ADR-015) on anvilkit:deployment.export.requested.
  2. Pending recovery — XPENDING/XAUTOCLAIM reclaim of delivered-but-unacked messages; never increments the business attempt counter.
  3. Business retry — a worker decision after a classified retryable failure; attempt 0..3, maxRetries = 3 → four executions max.
  4. Delayed retry — exponential backoff (base 10 s, max 5 m, jitter) enforced by a Redis Hash (payloads) + ZSET (delay index) and a dispatcher loop that removes an envelope only after successful re-enqueue. Envelopes are idempotent by retryEnvelopeId.
  5. Dead-letter routing — anvilkit:deployment.export.dlq after exhaustion or for unparseable input, with full DLQEntry metadata.

Kafka-ready seam (FR-021): job logic imports only the Consumer, Publisher, RetryStore, and DeadLetterer interfaces — never Redis primitives. At GA the Redis driver is swapped for Kafka topics keyed by deploymentId (main, retry, DLQ), with delayed retry moving to a scheduler/delayed-topic pattern; the interfaces stay unchanged.

Index

Constants

View Source 
const (
	StreamMain       = "anvilkit:deployment.export.requested"
	StreamDLQ        = "anvilkit:deployment.export.dlq"
	KeyRetryPayloads = "anvilkit:deployment.export.retry:payloads"
	KeyRetryZSet     = "anvilkit:deployment.export.retry:zset"
	ConsumerGroup    = "export-worker"
)

Canonical Redis key names (ADR-003; consumer group per ADR-015).

View Source 
const (
	StreamArtifactReady = "anvilkit:deployment.artifact.ready"
	StreamExportFailed  = "anvilkit:deployment.export.failed"
)

Outcome-event streams (FR-013). Consumers (cdn-service, observers) are duplicate-tolerant keyed by deploymentId — emission is at-least-once under the CAS-then-emit ordering (ADR-005 default).

Variables

View Source 
var DefaultBackoff = BackoffPolicy{Base: 10 * time.Second, Max: 5 * time.Minute}

DefaultBackoff is the PRD 0008 §12.2 policy.

View Source 
var ErrUnparseable = errors.New("unparseable event: no extractable deploymentId")

ErrUnparseable marks input from which no deploymentId can be extracted — the DLQ-with-alert route (FR-003): the original is acked only after successful DLQ handoff, and no status update is ever attempted.

Functions

func EnqueueTimeOf 

func EnqueueTimeOf(entryID string) time.Time

EnqueueTimeOf derives the original enqueue time from a stream entry ID (epoch-millis prefix).

func EnvelopeID 

func EnvelopeID(deploymentID string, attempt int, lastErrorCode events.ErrorCode) string

EnvelopeID derives the idempotency key (ADR-003): 

retryEnvelopeId = deploymentId + ":" + attempt + ":" + lastErrorCode

where attempt is the retry execution the envelope schedules (1..3). A reclaimed message that fails the same way after a crash-before-ack derives the same id, so the re-write is a harmless overwrite (AC-027).

func ParseEvent 

func ParseEvent(payload []byte) (*events.ExportRequested, error)

ParseEvent validates payload against the frozen v1 inbound schema (embedded contract of record) and decodes it.

  • JSON-invalid payloads, or payloads without a non-empty string deploymentId, return an error wrapping ErrUnparseable.
  • Schema-invalid payloads that do carry a deploymentId return a classified non-retryable VALIDATION_FAILED (§13).

Types

type BackoffPolicy 

type BackoffPolicy struct {
	Base time.Duration
	Max  time.Duration
	// Rand yields uniform [0,1); nil uses math/rand/v2. Injectable for tests.
	Rand func() float64
}

BackoffPolicy computes delayed-retry backoff: exponential from Base capped at Max, with equal jitter (half fixed, half uniform-random) so retry storms decorrelate (PRD 0008 §12.2: base 10 s, max 5 m, jitter true).

func (BackoffPolicy) Delay 

func (p BackoffPolicy) Delay(attempt int) time.Duration

Delay returns the backoff before retry execution attempt (1-based).

type Consumer 

type Consumer interface {
	// Fetch blocks up to block for at most max new messages.
	Fetch(ctx context.Context, max int, block time.Duration) ([]Message, error)
	// Ack acknowledges processed messages.
	Ack(ctx context.Context, ids ...string) error
	// Reclaim takes over messages delivered to any consumer but idle
	// (unacked) for at least minIdle — mechanism 2, XAUTOCLAIM.
	Reclaim(ctx context.Context, minIdle time.Duration, count int) ([]Message, error)
	// PendingCount reports delivered-but-unacked messages in the group.
	PendingCount(ctx context.Context) (int64, error)
}

Consumer delivers messages at-least-once and acks them. Ack is called only under the ADR-003 ack rule: successful completion, confirmed terminal/non-actionable deployment state, or successful write-then-ack handoff to retry storage or the DLQ.

type DLQEntry 

type DLQEntry struct {
	Payload     []byte
	ErrorCode   events.ErrorCode
	FailedStage events.FailedStage
	Attempt     int
	TraceID     string
	WorkerID    string
	EnqueuedAt  time.Time // original main-stream enqueue time
	FailedAt    time.Time // terminal failure time
}

DLQEntry preserves everything PRD 0010 §10.3.3 requires for inspection and manual replay.

type DeadLetterer 

type DeadLetterer interface {
	SendToDLQ(ctx context.Context, entry DLQEntry) error
}

DeadLetterer routes terminally failed or unparseable messages to the DLQ.

type Dispatcher 

type Dispatcher struct {
	Store    RetryStore
	Pub      Publisher
	Log      *slog.Logger
	Metrics  *obs.Metrics
	Interval time.Duration // default 1s
	Batch    int           // default 100
}

Dispatcher is the delayed-retry dispatcher loop (EW-QUEUE-006): every Interval it looks up due envelopes, re-enqueues them to the main stream, and removes each envelope only after its re-enqueue succeeded. A failed re-enqueue leaves the envelope in place for the next tick — an envelope can be dispatched twice under crash timing, which is safe: redelivery is at-least-once and processing is idempotent by deploymentId.

func (*Dispatcher) Run 

func (d *Dispatcher) Run(ctx context.Context)

Run loops until ctx is done.

func (*Dispatcher) Tick 

func (d *Dispatcher) Tick(ctx context.Context, now time.Time)

Tick performs one dispatch pass (exported for deterministic tests).

type Message 

type Message struct {
	ID            string // driver-native id (Redis stream entry ID)
	Payload       []byte // raw deployment.export.requested JSON
	Attempt       int
	LastErrorCode string // error code that scheduled this retry ("" on originals)
	TraceID       string // trace continuity across retries ("" on originals)
}

Message is one delivered queue message. Attempt is the business retry counter carried in entry metadata: 0 on original publishes, N on dispatcher re-enqueues. Pending reclaim redelivers the entry unchanged, so reclaim never increments Attempt (mechanism 2 vs 3).

type OutgoingMessage 

type OutgoingMessage struct {
	Payload       []byte
	Attempt       int
	LastErrorCode string
	TraceID       string
}

OutgoingMessage is a publish request to the main stream.

type Publisher 

type Publisher interface {
	Publish(ctx context.Context, msg OutgoingMessage) (string, error)
}

Publisher appends messages to the main stream (dispatcher re-enqueues and tooling).

type RedisDriver 

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

RedisDriver implements Consumer, Publisher, and DeadLetterer on Redis Streams (the MVP driver behind the FR-021 seam).

func NewRedisDriver 

func NewRedisDriver(rdb redis.UniversalClient, consumerName string) *RedisDriver

NewRedisDriver builds the driver. Call EnsureGroup once before consuming.

func (*RedisDriver) Ack 

func (d *RedisDriver) Ack(ctx context.Context, ids ...string) error

Ack acknowledges processed messages (ADR-003 ack rule applies at the call site, never here).

func (*RedisDriver) AppendOutcome 

func (d *RedisDriver) AppendOutcome(ctx context.Context, stream string, payload []byte) (string, error)

AppendOutcome appends one outcome-event payload to the given stream. It is the emitter's transport (internal/emit); payloads are schema-validated before they reach here.

func (*RedisDriver) EnsureGroup 

func (d *RedisDriver) EnsureGroup(ctx context.Context) error

EnsureGroup creates the consumer group (and stream) if absent.

func (*RedisDriver) Fetch 

func (d *RedisDriver) Fetch(ctx context.Context, max int, block time.Duration) ([]Message, error)

Fetch reads new messages for this consumer (XREADGROUP >).

func (*RedisDriver) PendingCount 

func (d *RedisDriver) PendingCount(ctx context.Context) (int64, error)

PendingCount reports delivered-but-unacked messages in the group.

func (*RedisDriver) Publish 

func (d *RedisDriver) Publish(ctx context.Context, msg OutgoingMessage) (string, error)

Publish appends to the main stream. Attempt metadata travels in entry fields so business retries (mechanism 3) are distinguishable from pending reclaims (mechanism 2), which redeliver the entry unchanged.

func (*RedisDriver) Reclaim 

func (d *RedisDriver) Reclaim(ctx context.Context, minIdle time.Duration, count int) ([]Message, error)

Reclaim takes over messages idle for at least minIdle (XAUTOCLAIM), redelivering them unchanged — the business attempt counter is untouched.

func (*RedisDriver) SendToDLQ 

func (d *RedisDriver) SendToDLQ(ctx context.Context, entry DLQEntry) error

SendToDLQ appends the full DLQEntry to the dead-letter stream (mechanism 5). The caller acks the original only after this returns nil (write-then-ack, PRD 0008 G-13).

func (*RedisDriver) TrimMain 

func (d *RedisDriver) TrimMain(ctx context.Context, horizon time.Time) (int64, error)

TrimMain trims the main stream to the retention horizon, capped so that no delivered-but-unacked (pending) entry and no undelivered entry is ever removed: the effective bound is the smallest of the horizon, the oldest pending entry, and the first entry past the group's last-delivered ID.

func (*RedisDriver) TrimStream 

func (d *RedisDriver) TrimStream(ctx context.Context, stream string, horizon time.Time) (int64, error)

TrimStream removes entries older than horizon from stream. Entries at or after the horizon are never removed. Returns entries removed.

type RedisRetryStore 

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

RedisRetryStore implements RetryStore on a Redis Hash (payloads) + ZSET (delay index) per ADR-003.

func NewRedisRetryStore 

func NewRedisRetryStore(rdb redis.UniversalClient) *RedisRetryStore

func (*RedisRetryStore) Due 

func (s *RedisRetryStore) Due(ctx context.Context, now time.Time, batch int) ([]RetryEnvelope, error)

Due returns up to batch envelopes whose NextAttemptAt has passed.

func (*RedisRetryStore) OldestDueLag 

func (s *RedisRetryStore) OldestDueLag(ctx context.Context, now time.Time) (time.Duration, bool, error)

OldestDueLag reports the age of the oldest due envelope.

func (*RedisRetryStore) Remove 

func (s *RedisRetryStore) Remove(ctx context.Context, retryEnvelopeID string) error

Remove deletes the envelope — called only after successful re-enqueue (ADR-003 removal-after-success rule).

func (*RedisRetryStore) Schedule 

func (s *RedisRetryStore) Schedule(ctx context.Context, env RetryEnvelope) error

Schedule writes the envelope: HSET payload + ZADD delay index, atomically. Repeated writes of the same RetryEnvelopeID overwrite in place.

type RetryEnvelope 

type RetryEnvelope struct {
	RetryEnvelopeID string           `json:"retryEnvelopeId"`
	DeploymentID    string           `json:"deploymentId"`
	Attempt         int              `json:"attempt"`
	NextAttemptAt   int64            `json:"nextAttemptAt"` // epoch millis
	LastErrorCode   events.ErrorCode `json:"lastErrorCode"`
	TraceID         string           `json:"traceId"`
	Payload         json.RawMessage  `json:"payload"` // original event payload
}

RetryEnvelope schedules one business retry execution. Attempt is the execution it schedules (1..3). Idempotent by RetryEnvelopeID: a repeated write of the same envelope is a harmless overwrite, never a second envelope (AC-027).

type RetryStore 

type RetryStore interface {
	// Schedule writes the envelope (HSET + ZADD), idempotent by
	// RetryEnvelopeID.
	Schedule(ctx context.Context, env RetryEnvelope) error
	// Due returns up to batch envelopes with NextAttemptAt <= now.
	Due(ctx context.Context, now time.Time, batch int) ([]RetryEnvelope, error)
	// Remove deletes an envelope — called only after successful re-enqueue.
	Remove(ctx context.Context, retryEnvelopeID string) error
	// OldestDueLag reports how long the oldest due envelope has waited
	// (0, false when none is due). Feeds the retry-dispatch-lag gauge.
	OldestDueLag(ctx context.Context, now time.Time) (time.Duration, bool, error)
}

RetryStore is the delayed-retry storage (mechanism 4): Hash payloads + ZSET delay index.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.