queue

Documentation

Overview

Package queue implements the durable PostgreSQL job queue of ADR-002: enqueue with idempotency keys, FOR UPDATE SKIP LOCKED dequeue, leases with heartbeat, bounded retries with jitter, and a dead letter (§21.3).

Index

• Constants
• Variables
• func Enqueue(ctx context.Context, q EnqueueStore, opts EnqueueOptions) (store.Job, error)
• type EnqueueOptions
• type EnqueueStore
• type HandlerFunc
• type Step
• type StepRecorder
  • func NewStepRecorder(q StepStore, job store.Job) *StepRecorder
  • func (r *StepRecorder) Fail(ctx context.Context, message string)
  • func (r *StepRecorder) Flush(ctx context.Context)
  • func (r *StepRecorder) Skip(ctx context.Context, name, message string)
  • func (r *StepRecorder) Start(ctx context.Context, name string)
  • func (r *StepRecorder) Succeed(ctx context.Context, message string)
• type StepStore
• type Worker
  • func NewWorker(q WorkerStore, concurrency int, logger *slog.Logger) *Worker
  • func (w *Worker) Register(jobType string, h HandlerFunc)
  • func (w *Worker) Run(ctx context.Context)
  • func (w *Worker) Wait(timeout time.Duration)
• type WorkerStore

Constants

const LeaseDuration = 90 * time.Second

LeaseDuration is the job lease TTL (deployment-engine §2.5): unfinished jobs are picked up by another worker only after it expires, and never replayed blindly.

Variables

var KnownQueues = []string{"default", "deploy", "backup", "cleanup", "notify", "maintenance", "webhook", "task"}

KnownQueues is the set of logical queues a worker consumes. Enqueue refuses anything outside it: adding a queue means adding it here, not discovering months later that a job type has been silently piling up.

var RetryBase = 5 * time.Second

RetryBase is the first retry delay; it doubles at each attempt (§22.1). It is a variable, not a constant, so an operator — or a test suite that fails jobs on purpose — can tune it without patching the engine.

Functions

func Enqueue

func Enqueue(ctx context.Context, q EnqueueStore, opts EnqueueOptions) (store.Job, error)

Enqueue inserts a job. When an idempotency key conflicts, the original job is returned instead (INV-004).

Types

type EnqueueOptions

type EnqueueOptions struct {
	Queue          string // logical queue, e.g. deploy, backup, maintenance
	Type           string // e.g. server.validate
	Payload        any    // JSON-serializable; never contains secrets (INV-003)
	Priority       int32
	RunAt          time.Time // zero = now
	MaxAttempts    int32     // 0 = default 5
	IdempotencyKey *string
	LockKey        *string // e.g. server:validate:<uuid>
	TeamID         *int64
	ResourceID     *int64
	RetryOfID      *int64
}

EnqueueOptions describes a job to enqueue.

type EnqueueStore

type EnqueueStore interface {
	EnqueueJob(context.Context, store.EnqueueJobParams) (store.Job, error)
	GetJobByIdempotencyKey(context.Context, *string) (store.Job, error)
}

EnqueueStore is the queue persistence boundary for enqueuing jobs.

type HandlerFunc

type HandlerFunc func(ctx context.Context, job store.Job, rec *StepRecorder) (result any, err error)

HandlerFunc executes one job attempt. It must be idempotent: after a crash, another worker re-runs the job once the lease expires, and the handler inspects the remote effect before redoing work (§21.3, §22.1). The returned result is stored on success (never secrets, INV-003).

type Step

type Step struct {
	Name       string     `json:"name"`
	Status     string     `json:"status"` // pending|running|succeeded|failed|skipped
	Message    *string    `json:"message,omitempty"`
	StartedAt  *time.Time `json:"started_at,omitempty"`
	FinishedAt *time.Time `json:"finished_at,omitempty"`
}

Step mirrors the OpenAPI JobStep schema: every job step is visible, with a remediation message on failure (§20.1, §22.5).

type StepRecorder

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

StepRecorder persists job steps as the handler progresses.

func NewStepRecorder

func NewStepRecorder(q StepStore, job store.Job) *StepRecorder

NewStepRecorder starts an empty step timeline for a job attempt.

func (*StepRecorder) Fail

func (r *StepRecorder) Fail(ctx context.Context, message string)

Fail closes the current step as failed, with a remediation message.

func (*StepRecorder) Flush

func (r *StepRecorder) Flush(ctx context.Context)

Flush persists the timeline; persistence failures are non-fatal for the job itself.

func (*StepRecorder) Skip

func (r *StepRecorder) Skip(ctx context.Context, name, message string)

Skip records a step that did not need to run.

func (*StepRecorder) Start

func (r *StepRecorder) Start(ctx context.Context, name string)

Start opens a new running step.

func (*StepRecorder) Succeed

func (r *StepRecorder) Succeed(ctx context.Context, message string)

Succeed closes the current step as succeeded.

type StepStore

type StepStore interface {
	UpdateJobSteps(context.Context, store.UpdateJobStepsParams) error
}

StepStore is the persistence boundary for a job's step records.

type Worker

type Worker struct {
	// Telemetry is optional: a nil Metrics records nothing, so a worker built
	// without telemetry behaves exactly as before (ADR-008).
	Metrics *telemetry.Metrics
	Tracer  trace.Tracer

	Store       WorkerStore
	Concurrency int
	Queues      []string
	Logger      *slog.Logger
	// contains filtered or unexported fields
}

Worker consumes the queue with a bounded pool of goroutines.

func NewWorker

func NewWorker(q WorkerStore, concurrency int, logger *slog.Logger) *Worker

NewWorker builds a worker consuming the given logical queues.

func (*Worker) Register

func (w *Worker) Register(jobType string, h HandlerFunc)

Register binds a job type to its handler.

func (*Worker) Run

func (w *Worker) Run(ctx context.Context)

Run consumes jobs until ctx is cancelled, then waits for in-flight jobs (the caller bounds the drain with AKERDOCK_SHUTDOWN_TIMEOUT, §6.5).

func (*Worker) Wait

func (w *Worker) Wait(timeout time.Duration)

Wait blocks until in-flight jobs finish or the timeout elapses.

type WorkerStore

type WorkerStore interface {
	StepStore
	PromoteWaitingJobs(context.Context) (int64, error)
	ReapExpiredLeases(context.Context, int32) ([]store.ReapExpiredLeasesRow, error)
	DequeueJob(context.Context, store.DequeueJobParams) (store.Job, error)
	MarkJobRunning(context.Context, store.MarkJobRunningParams) (int64, error)
	HeartbeatJob(context.Context, store.HeartbeatJobParams) (int64, error)
	SucceedJob(context.Context, store.SucceedJobParams) (int64, error)
	FailJob(context.Context, store.FailJobParams) (int64, error)
}

WorkerStore is the queue persistence boundary a worker dequeues through.