Documentation
¶
Overview ¶
Package jobs is wowapi's Postgres-backed job runner (D-0047 — a focused queue behind the framework interfaces, NOT River). Modules enqueue a job in the SAME transaction as their business write (so the job commits atomically with the write, or not at all); a worker process claims jobs with FOR UPDATE SKIP LOCKED, executes each in a transaction bound to the job's tenant, and retries with exponential backoff + jitter until success or exhaustion (DLQ). Contract: docs/blueprint/07-platform-services.md §3.
The queue lives in jobs_queue (global; the tenant travels in the row's tenant_id, NULL for global jobs). job_runs is an append-only reporting mirror. jobs_queue is kernel-only: app_rt may only INSERT (enqueue), while the runner connects as app_platform and holds SELECT/INSERT/UPDATE on both tables.
Import boundary (depguard): stdlib + kernel/database + kernel/errors + kernel/model + kernel/config + pgx + google/uuid. Never module/app/adapters/ testkit in production code.
Index ¶
- func CountDead(ctx context.Context, pool *pgxpool.Pool) (int64, error)
- func DiscardDead(ctx context.Context, pool *pgxpool.Pool, id int64) error
- func Enqueue(ctx context.Context, db database.TenantDB, j Job, opts ...Opt) error
- func ExpJitterBackoff(attempt int) time.Duration
- func PublishDLQDepth(ctx context.Context, pool *pgxpool.Pool, m observability.Metrics) error
- func ReplayDead(ctx context.Context, pool *pgxpool.Pool, id int64) error
- type BackoffPolicy
- type DeadJob
- type DeadJobEntry
- type Job
- type Opt
- type Registry
- type RetryPolicy
- type Runner
- type RunnerOpt
- func WithDeadHook(fn func(context.Context, DeadJob)) RunnerOpt
- func WithDrainTimeout(d time.Duration) RunnerOpt
- func WithIDGen(g model.IDGen) RunnerOpt
- func WithJobTimeout(d time.Duration) RunnerOpt
- func WithLogger(l *slog.Logger) RunnerOpt
- func WithPoolSize(n int) RunnerOpt
- func WithReclaimInterval(d time.Duration) RunnerOpt
- func WithReclaimTimeout(d time.Duration) RunnerOpt
- func WithRunnerTracer(tr observability.Tracer) RunnerOpt
- type Scheduler
- func (s *Scheduler) Ensure(ctx context.Context) error
- func (s *Scheduler) OnRun(fn func(name string, lag time.Duration, err error))
- func (s *Scheduler) Register(name string, every time.Duration, run func(ctx context.Context) error)
- func (s *Scheduler) Run(ctx context.Context, poll time.Duration) error
- func (s *Scheduler) Tick(ctx context.Context)
- type Worker
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CountDead ¶
CountDead returns the number of dead-lettered (status='discarded') jobs currently in the queue — the jobs contribution to DLQ depth (roadmap CA-1 / backlog B-8). Runs on the platform pool (jobs_queue is a global kernel table).
func DiscardDead ¶
DiscardDead permanently deletes a discarded job from the queue. Returns KindNotFound if id is not a discarded job.
func Enqueue ¶
Enqueue inserts j into jobs_queue in the caller's tenant transaction. Because the INSERT rides the caller's tx (app_rt has INSERT on jobs_queue), the job is committed atomically with the business write: roll the tx back and the job never exists; commit it and the job is guaranteed queued. The tenant is taken from app_tenant_id() (the tx's SET LOCAL binding), not from Go.
func ExpJitterBackoff ¶
ExpJitterBackoff returns an exponential backoff delay for the given attempt: base 1s doubling each attempt, capped at 5m, plus a deterministic jitter of up to 25% of the (capped) delay. The jitter is a pure function of attempt — no time.Now or rand — so it is safe to call anywhere and never touches package init. The result is non-decreasing in attempt and never exceeds 5m.
func PublishDLQDepth ¶
PublishDLQDepth counts dead-lettered jobs and sets the dlq_depth{queue="jobs"} gauge on m. Drive it from the leader-safe scheduler (a single replica claims each interval) so the depth is counted once, not once per replica. m may be nil (no emission); otherwise it is the shared sink — observability.NoOp when no adapter is wired.
func ReplayDead ¶
ReplayDead resets a discarded job to 'available' for another run: attempts back to 0, run_at now, error/lock cleared. Returns KindNotFound if id is not a discarded job (already replayed, running, or never existed).
Types ¶
type BackoffPolicy ¶
BackoffPolicy maps a (1-based) attempt number to the delay before the next retry. It must be a pure function of attempt — no time.Now or rand at package init (jitter is derived deterministically from the attempt, see ExpJitterBackoff).
type DeadJob ¶
type DeadJob struct {
ID int64
Kind string
Tenant *uuid.UUID // nil for a global job
Attempts int
LastError string
}
DeadJob describes a job that exhausted its attempts and landed in the DLQ (status=discarded). It is handed to the dead-letter hook (WithDeadHook) so a process can emit a metric or alert.
type DeadJobEntry ¶
type DeadJobEntry struct {
ID int64
Kind string
TenantID *uuid.UUID // nil for a global job
Attempts int
MaxAttempts int
LastError string
FinishedAt *time.Time
Payload []byte
}
DeadJob is a dead-lettered (discarded) job row for inspection.
type Job ¶
type Job interface {
// Kind is the stable identifier a Worker registers under. Naming mirrors
// event kinds: "module.resource.verb" (e.g. "notify.email.send").
Kind() string
}
Job is a payload that knows its own kind. Payload structs implement it; the kind selects the registered Worker at execution time. A Job is JSON-marshaled into jobs_queue.payload on enqueue and handed back to the Worker as raw bytes.
type Opt ¶
type Opt func(*enqueueConfig)
Opt customizes a single Enqueue/EnqueueGlobal call.
func WithMaxAttempts ¶
WithMaxAttempts overrides the number of executions before the job is discarded to the DLQ. Values <= 0 are ignored (the table default of 5 applies).
func WithRunAt ¶
WithRunAt schedules the job to become eligible no earlier than t (delayed jobs). Without it the job is eligible immediately (run_at = now()).
func WithTracer ¶
func WithTracer(tr observability.Tracer) Opt
WithTracer wires a tracer so this enqueue captures the current request's W3C traceparent (roadmap O1/CA-9) into the job's trace_context; the runner continues that trace when it later executes the job. Default: NoOpTracer (empty trace context — no behavior change). Mirrors outbox.WithWriterTracer.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry collects the (kind → worker + retry policy) bindings during module boot. It accumulates errors (duplicate kind, empty kind, nil worker) rather than panicking, so RegisterKind reads cleanly at call sites and boot fails once via Err() with every problem reported together (mirrors outbox's HandlerRegistry).
func (*Registry) Err ¶
Err returns the accumulated registration errors joined into one, or nil. Boot calls this after all modules have registered and refuses to start on error.
func (*Registry) RegisterKind ¶
func (r *Registry) RegisterKind(kind string, w Worker, rp RetryPolicy)
RegisterKind binds a worker and retry policy to a job kind. Registering the same kind twice, an empty kind, or a nil worker records an error surfaced by Err(). A zero-value RetryPolicy is filled from DefaultRetry so a caller can register with just a worker.
type RetryPolicy ¶
type RetryPolicy struct {
MaxAttempts int
Backoff BackoffPolicy
}
RetryPolicy governs how a kind is retried. MaxAttempts is the total number of executions before a job is discarded to the DLQ; Backoff spaces the retries. The authoritative attempt ceiling for a specific job is its jobs_queue max_attempts column (set at enqueue, defaulting to 5) — MaxAttempts here is the policy default that DefaultRetry aligns with.
func DefaultRetry ¶
func DefaultRetry() RetryPolicy
DefaultRetry is the blueprint default: 5 attempts, exponential backoff with jitter from 1s up to a 5m cap.
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner consumes jobs. It holds an app_platform pool (claim + status writes, which app_rt is not granted), the tenant TxManager (to execute each worker in a transaction bound to the job's tenant), and the Registry of workers. It runs a bounded fixed-size worker pool — never one goroutine per job (blueprint: no unbounded goroutines; the `go` keyword is permitted in kernel/jobs).
func NewRunner ¶
func NewRunner(platformPool *pgxpool.Pool, txm database.TxManager, reg *Registry, opts ...RunnerOpt) *Runner
NewRunner wires a Runner. platformPool must authenticate as app_platform (the role granted claim/complete on jobs_queue + job_runs); txm runs worker transactions per tenant; reg supplies the workers.
func (*Runner) ClaimOnce ¶
ClaimOnce claims up to poolSize available jobs (marking each 'running' in a committed statement), then executes them concurrently on the bounded worker pool, waiting for the batch to finish. It returns the number of jobs claimed. Per-job outcomes (completed / retry / DLQ) are written by the workers; a DB failure while writing an outcome is logged and leaves the job 'running' for ReclaimStalled — ClaimOnce only returns an error for a failure to claim.
func (*Runner) EnqueueGlobal ¶
EnqueueGlobal inserts a tenant-less (global) job. It is a Runner method because it writes on the app_platform pool (there is no business tx to ride — a global job has no tenant), unlike Enqueue which rides the caller's tenant tx. The row's tenant_id is NULL; at execution the worker runs under the sentinel nil tenant (see execOne).
func (*Runner) ReclaimStalled ¶
ReclaimStalled resets 'running' jobs whose lock is older than olderThan back to 'available', so jobs a crashed worker left mid-flight are retried. It returns the number of jobs reclaimed. make_interval pins the unit unambiguously (a Go duration string like "5m0s" would be misread by Postgres interval parsing, where "m" means months).
func (*Runner) Run ¶
Run drives the runner until ctx is cancelled: ClaimOnce back-to-back while there is work, then poll on the interval, sweeping stalled jobs periodically. Cancellation is graceful — the loop stops claiming new work, and the in-flight batch finishes (bounded by drainTimeout) before Run returns nil.
type RunnerOpt ¶
type RunnerOpt func(*Runner)
RunnerOpt customizes a Runner.
func WithDeadHook ¶
WithDeadHook registers a callback invoked when a job is discarded to the DLQ (the "leave a hook for a metric" seam).
func WithDrainTimeout ¶
WithDrainTimeout bounds how long in-flight jobs may finish after ctx is cancelled before the runner stops waiting. Default 30s.
func WithIDGen ¶
WithIDGen overrides the id generator used for job_runs primary keys (tests inject a deterministic sequence).
func WithJobTimeout ¶
WithJobTimeout bounds a single job's worker runtime, independent of the shutdown drain budget. Default 2m. The reclaim floor is derived from this.
func WithLogger ¶
WithLogger overrides the slog.Logger for internal (non-worker) errors.
func WithPoolSize ¶
WithPoolSize sets the bounded worker-pool size (and the per-claim batch). Default 10.
func WithReclaimInterval ¶
WithReclaimInterval sets how often Run sweeps for stalled jobs. Default 1m.
func WithReclaimTimeout ¶
WithReclaimTimeout sets how old a 'running' job's lock must be before ReclaimStalled resets it to 'available' (crash recovery). Default 5m.
func WithRunnerTracer ¶
func WithRunnerTracer(tr observability.Tracer) RunnerOpt
WithRunnerTracer wires a tracer so the runner continues each job's originating request trace when it executes the job (roadmap O1/CA-9): it extracts the traceparent captured at enqueue and runs the worker under a child span, and it is also the default tracer for EnqueueGlobal. Default: NoOpTracer. Mirrors outbox.WithRelayTracer.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler runs registered maintenance tasks on fixed intervals, leader-safe across worker replicas (roadmap E5 + R3). Each task has a row in `schedules`; a due tick is claimed by an atomic conditional UPDATE (`next_run_at <= now` under FOR UPDATE SKIP LOCKED), so exactly one replica runs a given task per interval — no separate leader election. Tasks run OUTSIDE the claim tx, so a slow task never holds the row lock; because the claim already advanced next_run_at, a failed task simply retries next interval (tasks must be idempotent, which the kernel sweeps are).
func NewScheduler ¶
NewScheduler builds a scheduler over the platform pool.
func (*Scheduler) Ensure ¶
Ensure upserts a schedule row for every registered task, preserving next_run_at for an existing row (so a redeploy does not reset the clock) while updating its interval. Run calls it; tests may call it before Tick.
func (*Scheduler) Register ¶
Register adds a recurring task. every is clamped to >= 1s. Call before Run.
type Worker ¶
Worker executes one job. It receives the tenant-bound database facade (RLS is already scoped to the job's tenant via SET LOCAL) and the raw JSON payload it unmarshals into its own typed struct. A returned error triggers retry/backoff; a nil error marks the job completed.
DELIVERY IS AT-LEAST-ONCE. The worker's DB effect commits in the tenant tx, but the queue 'completed' mark commits in a SEPARATE tx (a different role and pool), so a crash in between — or a reclaim of an over-running job — reruns the worker. Unlike event handlers (which get the processed_events inbox for exactly-once DB effects), jobs have NO framework-provided dedup. Therefore a worker MUST be idempotent by construction: DB-only work should be naturally idempotent (upserts, version checks); a worker with an EXTERNAL side effect (email, webhook, payment) MUST carry its own idempotency key against the provider, or it can double-fire (review finding ARCH-59).
Source Files