From bd7679285e9a91237256b0e8f0cd7830c654ba1c Mon Sep 17 00:00:00 2001 From: DanieCuevas <43822444+DanielCuevas1208@users.noreply.github.com> Date: Mon, 3 Aug 2026 06:27:54 -0700 Subject: [PATCH] feat: extend local first job queue --- .gitattributes | 14 + .github/workflows/ci.yml | 2 + README.md | 214 ++++++++++++-- internal/cli/demo.go | 184 ++++++++++-- internal/cli/inspect.go | 2 +- internal/cli/metrics.go | 48 ++++ internal/cli/requeue.go | 49 ++++ internal/cli/util.go | 46 ++- internal/cli/util_test.go | 64 +++++ internal/cli/work.go | 35 ++- internal/metrics/metrics.go | 184 ++++++++++++ internal/metrics/metrics_test.go | 169 +++++++++++ internal/queue/models.go | 21 ++ internal/queue/queue.go | 77 +++++- internal/queue/queue_test.go | 461 ++++++++++++++++++++++++++++++- internal/queue/sqlite.go | 188 +++++++++++-- internal/worker/worker_test.go | 94 +++++++ main.go | 6 + 18 files changed, 1775 insertions(+), 83 deletions(-) create mode 100644 .gitattributes create mode 100644 internal/cli/metrics.go create mode 100644 internal/cli/requeue.go create mode 100644 internal/cli/util_test.go create mode 100644 internal/metrics/metrics.go create mode 100644 internal/metrics/metrics_test.go diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..cbcd8c1 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,14 @@ +# Normalize Go source to LF so formatting checks are stable across platforms. +*.go text eol=lf + +# Markdown and shell scripts also normalize to LF. +*.md text eol=lf +*.yml text eol=lf +*.yaml text eol=lf +*.ps1 text eol=crlf + +# Text artifacts that must never be modified by Git. +*.db binary +*.db-wal binary +*.db-shm binary +*.db-journal binary diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6f3e777..7f564e7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,6 +34,8 @@ jobs: run: go vet ./... - name: Test with race detector run: go test -v -count=1 -race -coverprofile=coverage.out ./... + - name: Coverage summary + run: go tool cover -func=coverage.out | tail -n 1 - name: Build run: go build ./... - name: Benchmark queue paths diff --git a/README.md b/README.md index bfa15e9..86bca03 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A small durable background queue built with Go and SQLite. -The project shows leases, retries, idempotency, crash recovery, priority dispatch, and an append-only event log. +The project shows leases, retries, idempotency, crash recovery, priority dispatch, priority aging, a dead-letter queue, Prometheus metrics, and an append-only event log. ## Value @@ -20,6 +20,7 @@ The queue separates durable state from worker execution. - `internal/worker` leases jobs and runs handlers. - `internal/fault` injects deterministic errors, panics, delays, and stalls. - `internal/cli` renders commands, snapshots, history, and the demo. +- `internal/metrics` renders queue state in the Prometheus text format. - `internal/fixture` provides repeatable sample workloads. A job starts as `pending`. @@ -81,11 +82,19 @@ jobqueue enqueue -kind -payload [-priority ] [-idempotency-key ### `work` ```text -jobqueue work -kind [-concurrency ] [-lease ] [-poll ] [-db ] +jobqueue work -kind [-concurrency ] [-lease ] [-poll ] [-aging ] [-metrics-addr ] [-db ] ``` The worker recovers expired leases when it starts. +Priority aging is enabled by default with a 30-second interval. + +A job gains one priority point per interval it waits. + +Use `-aging 0` to disable aging. + +Use `-metrics-addr` to serve Prometheus metrics beside the worker. + ### `inspect` ```text @@ -104,6 +113,18 @@ jobqueue history [-json] [-db ] The command prints one job and its event timeline. +### `requeue` + +```text +jobqueue requeue [-db ] [-max-attempts ] [-payload ] +``` + +The command returns a dead-lettered job to the pending state. + +The job resets its attempt count and keeps its data. + +Use `-payload` to correct the job data before it runs again. + ### `seed` ```text @@ -112,13 +133,27 @@ jobqueue seed [-db ] The command loads three idempotent jobs for each bundled workload. +### `metrics` + +```text +jobqueue metrics [-once] [-addr ] [-db ] +``` + +The command serves queue state in the Prometheus text format. + +The default address is `:9090`. + +Scrape the endpoint with a Prometheus server. + +Use `-once` to print one snapshot and exit. + ### `demo` ```text jobqueue demo [-db ] [-keep] [-run ] [-kind ] ``` -The demo combines priority, retries, panic recovery, crash recovery, and scheduling. +The demo combines priority, retries, panic recovery, crash recovery, scheduling, priority aging, and dead-letter requeue. ## Features @@ -136,11 +171,31 @@ A future job cannot bypass its `run_at` time, even when its priority is higher. Equal priorities use readiness time, creation time, and job ID as deterministic tie breakers. +### Priority aging + +A pending job gains one priority point per aging interval it waits. + +The interval is a store setting; the default is 30 seconds. + +The `work` command enables aging by default. + +Use `-aging 0` to disable it. + +An older low-priority job can overtake a fresher high-priority job. + +The store measures the wait from the job's readiness time. + +A scheduled job starts aging only when its `run_at` time passes. + +Aging prevents a constant high-priority stream from starving other work. + +The `demo` command shows a low-priority job winning after five intervals. + ### Retries A failed handler returns the job to `pending` while attempts remain. -The job enters `failed` after the attempt budget is exhausted. +The job enters the dead-letter queue after the attempt budget is exhausted. ### Idempotency @@ -148,18 +203,56 @@ An idempotency key makes repeated enqueue calls return one durable job. The database enforces the uniqueness rule. +### Dead-letter queue + +A job that exhausts its attempts enters the `dead_letter` state. + +The event log records one `dead_lettered` event per exhausted job. + +Use the `requeue` command to return a dead-lettered job to `pending`. + +The job keeps its data unless the command supplies a new payload. + +A requeued job resets its attempt count and can fail again. + ### Crash recovery Startup recovery finds leases past their deadlines. Recovery consumes an attempt and records a `recovered` event. +A recovered job with no attempts left enters the dead-letter queue. + ### Event log Every state change appends one event row. The `history` command shows one job's complete timeline. +### Metrics + +The exporter renders queue state in the Prometheus text format. + +Each scrape computes a fresh snapshot from the SQLite store. + +The exporter reports four metric families. + +`jobqueue_jobs` counts jobs by state. + +`jobqueue_jobs_by_kind` counts jobs by kind and state. + +`jobqueue_events_total` counts events by type. + +`jobqueue_oldest_pending_seconds` reports the oldest pending job's age. + +Every known state and event type appears with an explicit zero. + +The output order stays stable across scrapes. + +Use the `metrics` command for one snapshot or a live endpoint. + +Use `work -metrics-addr` to serve the same endpoint beside a worker. + ### Scheduling A scheduled job stores its earliest lease time in `run_at`. @@ -176,27 +269,63 @@ Run `jobqueue demo` to see a complete local scenario. == Local-first Durable Job Queue: demo == enqueuing scenario jobs: first-try success alpha priority= 0 - priority retry beta priority=10 - orphaned by a crash delta priority= 0 + priority retry beta priority=10 + exhausts attempts gamma priority= 0 + panic then ok epsilon priority= 0 + orphaned by a crash delta priority= 0 + delayed run omega priority= 0 orphaned job delta was leased and then abandoned. starting worker; it will recover orphans and process jobs. queue drained before the run deadline. +Dead-letter queue +----------------- + kind=demo priority=0 state=dead_letter attempts=3/3 + +operator requeues the dead-lettered job with a corrected payload. +starting worker again; it will process the requeued job. + +queue drained before the run deadline. + +Priority aging +-------------- +aging interval: 100ms; a job gains one priority point per interval it waits. + aged (low priority) priority= 0 waited=5 intervals effective=5 + fresh (high priority) priority= 1 waited=0 intervals effective=1 + +lease order: (aged) then (fresh) +the waiting job outranks the fresher higher-priority job. + Queue state ----------- - completed: 5 - failed: 1 + completed: 6 -Recent events (29) ------------------- - [12:00:00] acknowledged - [12:00:00] recovered attempt 1/3 +Recent events (32) +----------------- + [12:00:00] requeued attempts reset to 0/3 + [12:00:00] dead_lettered attempt 3/3 exhausted: disk full Jobs (6) -------- kind=demo priority=10 state=completed attempts=2/3 + +Metrics +------- +# HELP jobqueue_jobs Number of jobs in each state. +# TYPE jobqueue_jobs gauge +jobqueue_jobs{state="pending"} 0 +jobqueue_jobs{state="leased"} 0 +jobqueue_jobs{state="completed"} 6 +jobqueue_jobs{state="dead_letter"} 0 +jobqueue_jobs{state="failed"} 0 +# HELP jobqueue_events_total Number of events per event type. +# TYPE jobqueue_events_total counter +jobqueue_events_total{type="enqueued"} 5 +jobqueue_events_total{type="retried"} 5 +jobqueue_events_total{type="dead_lettered"} 1 +jobqueue_events_total{type="requeued"} 1 ``` The demo uses generated job IDs and current timestamps. @@ -224,9 +353,9 @@ Run queue benchmarks with this command. go test ./internal/queue -run '^$' -bench Benchmark -benchmem -count=1 ``` -Verification status: tests, vet, build, and benchmarks pass locally on Go 1.25. +Verification status: tests, vet, build, and benchmarks pass locally and in CI. -Race tests run in CI on Ubuntu, where the required C compiler is available. +Race tests run in CI on Ubuntu. ## Limitations @@ -236,27 +365,68 @@ A sustained backlog can exceed the writer's capacity. Jobs and events remain until an operator removes them. -A high-priority stream can delay lower-priority jobs. - -The queue has no dead-letter workflow or priority aging. +A high-priority stream can delay lower-priority jobs until aging lifts them. The worker is one process and does not coordinate across hosts. -The project does not provide Prometheus metrics or a web interface. +The project does not provide a web interface. ## Roadmap - [x] Durable leases, retries, idempotency, crash recovery, and event history. - [x] Scheduled jobs with nanosecond-safe release times. - [x] Priority-aware dispatch with deterministic ordering. -- [ ] Dead-letter queue for permanently failed jobs. -- [ ] Prometheus metrics. +- [x] Priority aging to prevent starvation. +- [x] Dead-letter queue with requeue of permanently failed jobs. +- [x] Prometheus metrics for queue inspection. - [ ] Web UI for queue inspection. - [ ] Horizontal scaling with a shared SQLite file. ### Release notes -This release adds durable priority dispatch. +This release adds Prometheus metrics. + +The new `metrics` command serves the exposition format over HTTP. + +Use `-once` to print one snapshot instead. + +The `work` command can serve the same endpoint beside a worker. + +The demo prints the final metrics snapshot. + +Each scrape reads the SQLite store and reports current state. + +The previous release added priority aging to prevent starvation. + +A pending job gains one priority point per aging interval it waits. + +The default aging interval is 30 seconds. + +The `work` command enables aging by default. + +Use `-aging 0` to disable aging. + +The store measures the wait from the job's readiness time. + +A scheduled job starts aging only when its `run_at` time passes. + +The library keeps aging opt-in, so callers keep their exact ordering. + +The demo now shows a low-priority job overtaking a fresher one. + +The previous release added a dead-letter queue for jobs that exhaust their attempts. + +A job enters the `dead_letter` state after its attempt budget runs out. + +The event log records a `dead_lettered` event for each exhausted job. + +The new `requeue` command returns a dead-lettered job to `pending`. + +The command can supply a new payload and a new attempt budget. + +The demo now shows the full dead-letter workflow. + +The previous release added durable priority dispatch. Jobs store an integer priority with a default of zero. @@ -264,4 +434,4 @@ The lease query selects ready jobs by descending priority. The migration adds `priority` to existing databases before creating its indexes. -The release also preserves sub-second schedule deadlines during SQLite writes. \ No newline at end of file +That release also preserved sub-second schedule deadlines during SQLite writes. \ No newline at end of file diff --git a/internal/cli/demo.go b/internal/cli/demo.go index 8d7731b..7048f28 100644 --- a/internal/cli/demo.go +++ b/internal/cli/demo.go @@ -10,6 +10,7 @@ import ( "time" "github.com/local-first-job-queue/internal/fault" + "github.com/local-first-job-queue/internal/metrics" "github.com/local-first-job-queue/internal/queue" "github.com/local-first-job-queue/internal/worker" ) @@ -21,15 +22,23 @@ import ( // The scenario covers: // - a job that completes on the first attempt, // - a job that fails twice and then succeeds, -// - a job that exhausts its attempts and enters the failed state, +// - a job that exhausts its attempts and enters the dead-letter queue, // - a job that panics on the first attempt and succeeds on the second, // - a job that a previous worker leased and abandoned, to show crash // recovery on the next worker start. +// +// After the first worker run, the demo requeues the dead-lettered job with a +// corrected payload and runs the worker again. The requeued job completes, so +// the output shows the full dead-letter workflow. +// +// A short separate segment shows priority aging: a low-priority job scheduled +// in the past overtakes a fresher higher-priority job because it has waited +// for several aging intervals. func Demo(args []string) error { fs := flag.NewFlagSet("demo", flag.ExitOnError) dbPath := fs.String("db", "", "database path (default: a temp file)") keep := fs.Bool("keep", false, "keep the demo database file after the run") - maxRun := fs.Duration("run", 3*time.Second, "maximum time the worker runs") + maxRun := fs.Duration("run", 3*time.Second, "maximum time each worker run lasts") kind := fs.String("kind", "demo", "job kind used by the demo") fs.Parse(args) @@ -76,14 +85,14 @@ func Demo(args []string) error { }{ {"first-try success", "alpha", `{"name":"alpha"}`, 3, 0, 0}, {"priority retry", "beta", `{"name":"beta","fault":{"mode":"error","fail_until_attempt":2,"message":"rate limited"}}`, 3, 10, 0}, - {"exhausts attempts", "gamma", `{"name":"gamma","fault":{"mode":"error","fail_until_attempt":5,"message":"disk full"}}`, 3, 0, 0}, + {"exhausts attempts", "gamma", `{"name":"gamma","fault":{"mode":"error","fail_until_attempt":3,"message":"disk full"}}`, 3, 0, 0}, {"panic then ok", "epsilon", `{"name":"epsilon","fault":{"mode":"panic","fail_until_attempt":1,"message":"kaboom"}}`, 2, 0, 0}, {"orphaned by a crash", "delta", `{"name":"delta"}`, 3, 0, 0}, {"delayed run", "omega", `{"name":"omega"}`, 3, 0, 40 * time.Millisecond}, } fmt.Println("enqueuing scenario jobs:") - var deltaID string + var deltaID, gammaID string for _, s := range scenario { opts := []queue.EnqueueOption{ queue.WithMaxAttempts(s.maxAttempts), @@ -97,8 +106,11 @@ func Demo(args []string) error { return fmt.Errorf("enqueue %s: %w", s.label, err) } fmt.Printf(" %-22s %-22s priority=%2d %s\n", s.label, s.name, s.priority, job.ID) - if s.name == "delta" { + switch s.name { + case "delta": deltaID = job.ID + case "gamma": + gammaID = job.ID } } @@ -115,39 +127,154 @@ func Demo(args []string) error { log.Printf("handled name=%s attempt=%d state-before=%s", payloadName(job.Payload), job.RetryCount+1, job.State) return nil } - w := worker.NewWorker(q, fault.New(inner, fault.FromPayload).Handle, *kind, - worker.WithConcurrency(1), - worker.WithPollInterval(5*time.Millisecond), - worker.WithLeaseDuration(2*time.Second), - ) + newWorker := func() *worker.Worker { + return worker.NewWorker(q, fault.New(inner, fault.FromPayload).Handle, *kind, + worker.WithConcurrency(1), + worker.WithPollInterval(5*time.Millisecond), + worker.WithLeaseDuration(2*time.Second), + ) + } - ctx, cancel := context.WithTimeout(context.Background(), *maxRun) - defer cancel() - done := make(chan error, 1) - go func() { done <- w.Run(ctx) }() + runWorker := func() bool { + w := newWorker() + ctx, cancel := context.WithTimeout(context.Background(), *maxRun) + defer cancel() + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + idle := waitUntilIdle(q, 20*time.Millisecond, ctx.Done()) + cancel() + <-done + return idle + } + + if runWorker() { + fmt.Println("queue drained before the run deadline.") + } else { + fmt.Println("reached the run deadline before the queue drained.") + } + + // gamma exhausted its attempts during the first run. Show the dead-letter + // queue, then requeue gamma with a corrected payload and run the worker + // again so the job completes. + fmt.Println("\nDead-letter queue") + fmt.Println("-----------------") + snap, err := q.Inspect() + if err != nil { + return fmt.Errorf("inspect: %w", err) + } + deadLetterCount := 0 + for _, j := range snap.Jobs { + if j.State == queue.StateDeadLetter { + fmt.Printf(" %s kind=%s priority=%d state=%s attempts=%d/%d\n", + shortID(j.ID), j.Kind, j.Priority, j.State, j.RetryCount, j.MaxAttempts) + deadLetterCount++ + } + } + if deadLetterCount == 0 { + fmt.Println(" (empty)") + } - idle := waitUntilIdle(q, 20*time.Millisecond, ctx.Done()) - cancel() - <-done + fmt.Println("\noperator requeues the dead-lettered job with a corrected payload.") + if _, err := q.Requeue(gammaID, queue.RequeueWithPayload(`{"name":"gamma","fixed":true}`)); err != nil { + return fmt.Errorf("requeue gamma: %w", err) + } + fmt.Println("starting worker again; it will process the requeued job.") + fmt.Println() - if idle { - fmt.Println("\nqueue drained before the run deadline.") + if runWorker() { + fmt.Println("queue drained before the run deadline.") } else { - fmt.Println("\nreached the run deadline before the queue drained.") + fmt.Println("reached the run deadline before the queue drained.") + } + + if err := showPriorityAging(*kind); err != nil { + return err } fmt.Println() - snap, err := q.Inspect() + snap, err = q.Inspect() if err != nil { return fmt.Errorf("inspect: %w", err) } RenderSnapshot(snap, os.Stdout) + fmt.Println() + fmt.Println("Metrics") + fmt.Println("-------") + if err := metrics.New(store).Write(os.Stdout); err != nil { + return fmt.Errorf("render metrics: %w", err) + } + fmt.Printf("\ninspect again with: jobqueue inspect -db %q\n", path) fmt.Printf("inspect a job with: jobqueue history -db %q\n", path) + fmt.Printf("requeue a dead letter with: jobqueue requeue -db %q\n", path) + return nil +} + +// showPriorityAging demonstrates that a job gains one priority point per aging +// interval it has waited. A low-priority job scheduled in the past has a higher +// effective priority than a fresher higher-priority job, so it leases first. +// The segment uses its own in-memory store so it does not disturb the scenario +// counts. +func showPriorityAging(kind string) error { + const agingInterval = 100 * time.Millisecond + store, err := queue.NewSQLiteStore("file:jobqueue-demo-aging?mode=memory&cache=shared", + queue.WithAgingInterval(agingInterval)) + if err != nil { + return fmt.Errorf("open aging store: %w", err) + } + defer store.Close() + q := queue.NewQueue(store) + + fmt.Println("\nPriority aging") + fmt.Println("--------------") + fmt.Printf("aging interval: %s; a job gains one priority point per interval it waits.\n", agingInterval) + + aged, err := q.Enqueue(kind, `{"name":"aged"}`, queue.WithPriority(0), queue.WithRunAt(time.Now().Add(-5*agingInterval))) + if err != nil { + return fmt.Errorf("enqueue aged: %w", err) + } + fresh, err := q.Enqueue(kind, `{"name":"fresh"}`, queue.WithPriority(1)) + if err != nil { + return fmt.Errorf("enqueue fresh: %w", err) + } + fmt.Printf(" %-22s priority=%2d waited=5 intervals effective=%d\n", + "aged (low priority)", 0, effectivePriority(*aged, agingInterval)) + fmt.Printf(" %-22s priority=%2d waited=0 intervals effective=%d\n", + "fresh (high priority)", 1, effectivePriority(*fresh, agingInterval)) + fmt.Println() + + ctx := context.Background() + first, err := q.Lease(ctx, kind, time.Minute) + if err != nil { + return fmt.Errorf("lease first: %w", err) + } + second, err := q.Lease(ctx, kind, time.Minute) + if err != nil { + return fmt.Errorf("lease second: %w", err) + } + if first == nil || second == nil { + return fmt.Errorf("expected two leases in the aging demo") + } + if first.ID != aged.ID { + return fmt.Errorf("aging demo: expected aged job to lease first") + } + if second.ID != fresh.ID { + return fmt.Errorf("aging demo: expected fresh job to lease second") + } + fmt.Printf("lease order: %s (%s) then %s (%s)\n", + shortID(first.ID), payloadName(first.Payload), shortID(second.ID), payloadName(second.Payload)) + fmt.Println("the waiting job outranks the fresher higher-priority job.") return nil } +func shortID(id string) string { + if len(id) > 8 { + return id[:8] + } + return id +} + func payloadName(payload string) string { const needle = `"name":"` i := strings.Index(payload, needle) @@ -162,6 +289,21 @@ func payloadName(payload string) string { return payload[i : i+j] } +// effectivePriority reports the priority the lease query would use for a job, +// including the aging boost it has earned by waiting. The store measures the +// wait from COALESCE(run_at, created_at), mirroring the SQL ordering clause. +func effectivePriority(j queue.Job, interval time.Duration) int { + ready := j.CreatedAt + if j.RunAt != nil { + ready = *j.RunAt + } + waited := time.Since(ready) + if waited < 0 { + waited = 0 + } + return j.Priority + int(waited/interval) +} + // waitUntilIdle returns true when the queue has no pending and no leased jobs. // It polls until that condition holds or until done is closed. func waitUntilIdle(q *queue.Queue, every time.Duration, done <-chan struct{}) bool { diff --git a/internal/cli/inspect.go b/internal/cli/inspect.go index e1cf5d4..40753fc 100644 --- a/internal/cli/inspect.go +++ b/internal/cli/inspect.go @@ -48,7 +48,7 @@ func RenderSnapshot(snap *queue.QueueSnapshot, w io.Writer) { } else { for _, s := range []queue.JobState{ queue.StatePending, queue.StateLeased, - queue.StateCompleted, queue.StateFailed, + queue.StateCompleted, queue.StateDeadLetter, queue.StateFailed, } { if c, ok := snap.Stats[s]; ok { fmt.Fprintf(w, " %s: %d\n", s, c) diff --git a/internal/cli/metrics.go b/internal/cli/metrics.go new file mode 100644 index 0000000..b0a1084 --- /dev/null +++ b/internal/cli/metrics.go @@ -0,0 +1,48 @@ +package cli + +import ( + "flag" + "fmt" + "io" + "log" + "net/http" + "os" + + "github.com/local-first-job-queue/internal/metrics" + "github.com/local-first-job-queue/internal/queue" +) + +// Metrics exposes the queue state in the Prometheus text exposition format. +// With -once it prints a snapshot and exits, which suits scripts and the demo. +// Without -once it serves an HTTP endpoint that recomputes the snapshot on +// every scrape, so a Prometheus server can poll the running queue. +func Metrics(args []string) error { + fs := flag.NewFlagSet("metrics", flag.ExitOnError) + addr := fs.String("addr", ":9090", "listen address for the HTTP endpoint") + dbPath := fs.String("db", "queue.db", "database path") + once := fs.Bool("once", false, "print one snapshot and exit") + fs.Parse(args) + + store, err := queue.NewSQLiteStore(*dbPath) + if err != nil { + return fmt.Errorf("open store: %w", err) + } + defer store.Close() + + if *once { + return renderMetrics(os.Stdout, store) + } + + log.Printf("metrics listening on %s (db=%s)", *addr, *dbPath) + log.Printf("scrape with: curl -s %s/metrics", *addr) + return http.ListenAndServe(*addr, metrics.Handler(store)) +} + +// renderMetrics writes one queue snapshot in the Prometheus text format to w. +// It is separate from Metrics so tests can capture the output without serving. +func renderMetrics(w io.Writer, store *queue.SQLiteStore) error { + if err := metrics.New(store).Write(w); err != nil { + return fmt.Errorf("render metrics: %w", err) + } + return nil +} diff --git a/internal/cli/requeue.go b/internal/cli/requeue.go new file mode 100644 index 0000000..a7ae1d3 --- /dev/null +++ b/internal/cli/requeue.go @@ -0,0 +1,49 @@ +package cli + +import ( + "flag" + "fmt" + + "github.com/local-first-job-queue/internal/queue" +) + +// Requeue returns a dead-lettered job to the pending state. An operator uses +// this command after fixing the reason the job failed. The job ID may appear +// before or after the flags, like the history command. +func Requeue(args []string) error { + fs := flag.NewFlagSet("requeue", flag.ExitOnError) + dbPath := fs.String("db", "queue.db", "database path") + maxAttempts := fs.Int("max-attempts", 0, "new attempt budget; default keeps the current one") + payload := fs.String("payload", "", "new payload; default keeps the current one") + + ordered := moveFirstPositionalToEnd(args) + fs.Parse(ordered) + + if fs.NArg() == 0 { + return fmt.Errorf("usage: requeue [-db path] [-max-attempts n] [-payload json]") + } + jobID := fs.Arg(0) + + store, err := queue.NewSQLiteStore(*dbPath) + if err != nil { + return fmt.Errorf("open store: %w", err) + } + defer store.Close() + + q := queue.NewQueue(store) + opts := []queue.RequeueOption{} + if *maxAttempts > 0 { + opts = append(opts, queue.RequeueWithMaxAttempts(*maxAttempts)) + } + if *payload != "" { + opts = append(opts, queue.RequeueWithPayload(*payload)) + } + + job, err := q.Requeue(jobID, opts...) + if err != nil { + return fmt.Errorf("requeue: %w", err) + } + fmt.Printf("requeued job %s (%s) as pending with attempts 0/%d\n", + job.ID, job.Kind, job.MaxAttempts) + return nil +} diff --git a/internal/cli/util.go b/internal/cli/util.go index 864ab6a..0c8fd37 100644 --- a/internal/cli/util.go +++ b/internal/cli/util.go @@ -2,19 +2,49 @@ package cli import "strings" -// moveFirstPositionalToEnd moves the first non-flag argument to the end of the -// slice. The Go flag package stops parsing at the first non-flag token, so -// without this a user who writes "history -db path" would see -db ignored. +// valueFlags lists the CLI flags that consume a following value. The helper +// uses this set to tell flag values apart from positional arguments. Boolean +// flags and flags written as -flag=value are not included. +var valueFlags = map[string]bool{ + "-aging": true, + "-concurrency": true, + "-db": true, + "-idempotency-key": true, + "-kind": true, + "-lease": true, + "-max-attempts": true, + "-payload": true, + "-poll": true, + "-priority": true, + "-run-after": true, + "-run-at": true, +} + +// moveFirstPositionalToEnd moves positional arguments to the end of the slice. +// The Go flag package stops parsing at the first non-flag token, so without +// this a user who writes "history -db path" would see -db ignored. The +// helper leaves flag values in place and only relocates true positionals, so +// the id may appear before or after the flags. func moveFirstPositionalToEnd(args []string) []string { - for i, a := range args { + var reordered []string + var positional []string + for i := 0; i < len(args); i++ { + a := args[i] if a == "--" { return args } - if strings.HasPrefix(a, "-") { + if strings.HasPrefix(a, "-") && a != "-" { + reordered = append(reordered, a) + if valueFlags[a] && !strings.Contains(a, "=") && i+1 < len(args) { + reordered = append(reordered, args[i+1]) + i++ + } continue } - // Found the first positional token. Push it to the end. - return append(append(append([]string{}, args[:i]...), args[i+1:]...), a) + positional = append(positional, a) + } + if len(positional) == 0 { + return args } - return args + return append(reordered, positional...) } diff --git a/internal/cli/util_test.go b/internal/cli/util_test.go new file mode 100644 index 0000000..67cd914 --- /dev/null +++ b/internal/cli/util_test.go @@ -0,0 +1,64 @@ +package cli + +import ( + "reflect" + "strings" + "testing" + + "github.com/local-first-job-queue/internal/queue" +) + +func TestMoveFirstPositionalToEnd(t *testing.T) { + cases := []struct { + name string + in []string + want []string + }{ + {"id first", []string{"abc123", "-db", "queue.db"}, []string{"-db", "queue.db", "abc123"}}, + {"db before id", []string{"-db", "queue.db", "abc123"}, []string{"-db", "queue.db", "abc123"}}, + {"json and id first", []string{"abc123", "-json", "-db", "queue.db"}, []string{"-json", "-db", "queue.db", "abc123"}}, + {"json before id", []string{"-json", "abc123", "-db", "queue.db"}, []string{"-json", "-db", "queue.db", "abc123"}}, + {"value flag form", []string{"abc123", "-db=queue.db"}, []string{"-db=queue.db", "abc123"}}, + {"no positional", []string{"-db", "queue.db"}, []string{"-db", "queue.db"}}, + {"empty", nil, nil}, + {"double dash", []string{"--", "abc123", "-db", "queue.db"}, []string{"--", "abc123", "-db", "queue.db"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := moveFirstPositionalToEnd(tc.in) + if !reflect.DeepEqual(got, tc.want) { + t.Errorf("moveFirstPositionalToEnd(%v) = %v, want %v", tc.in, got, tc.want) + } + }) + } +} + +// TestRenderMetricsSnapshot verifies that the metrics -once path renders the +// queue state as Prometheus text. The collector output is deterministic, so a +// fresh store with one pending job yields a stable snapshot. +func TestRenderMetricsSnapshot(t *testing.T) { + s, err := queue.NewSQLiteStore("file:cli_metrics_" + t.Name() + "?mode=memory&cache=shared") + if err != nil { + t.Fatalf("new store: %v", err) + } + defer s.Close() + q := queue.NewQueue(s) + if _, err := q.Enqueue("email", `{"to":"a@example.com"}`); err != nil { + t.Fatalf("enqueue: %v", err) + } + + var b strings.Builder + if err := renderMetrics(&b, s); err != nil { + t.Fatalf("render metrics: %v", err) + } + for _, want := range []string{ + `jobqueue_jobs{state="pending"} 1`, + `jobqueue_jobs_by_kind{kind="email",state="pending"} 1`, + `jobqueue_events_total{type="enqueued"} 1`, + "# TYPE jobqueue_oldest_pending_seconds gauge", + } { + if !strings.Contains(b.String(), want) { + t.Errorf("metrics output missing %q in:\n%s", want, b.String()) + } + } +} diff --git a/internal/cli/work.go b/internal/cli/work.go index 9b558a1..1013445 100644 --- a/internal/cli/work.go +++ b/internal/cli/work.go @@ -2,14 +2,17 @@ package cli import ( "context" + "errors" "flag" "fmt" "log" + "net/http" "os/signal" "syscall" "time" "github.com/local-first-job-queue/internal/fault" + "github.com/local-first-job-queue/internal/metrics" "github.com/local-first-job-queue/internal/queue" "github.com/local-first-job-queue/internal/worker" ) @@ -32,9 +35,11 @@ func Work(args []string) error { concurrency := fs.Int("concurrency", 1, "number of concurrent workers") leaseDuration := fs.Duration("lease", 30*time.Second, "lease duration per job") pollInterval := fs.Duration("poll", time.Second, "time between lease attempts") + aging := fs.Duration("aging", queue.DefaultAgingInterval, "priority aging interval; a job gains one priority point per interval it waits (0 disables)") + metricsAddr := fs.String("metrics-addr", "", "address to serve Prometheus metrics on, e.g. :9090 (empty disables)") fs.Parse(args) - store, err := queue.NewSQLiteStore(*dbPath) + store, err := queue.NewSQLiteStore(*dbPath, queue.WithAgingInterval(*aging)) if err != nil { return fmt.Errorf("open store: %w", err) } @@ -48,10 +53,32 @@ func Work(args []string) error { worker.WithPollInterval(*pollInterval), ) + var srv *http.Server + if *metricsAddr != "" { + srv = &http.Server{Addr: *metricsAddr, Handler: metrics.Handler(store)} + go func() { + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Printf("metrics server: %v", err) + } + }() + log.Printf("metrics listening on %s", *metricsAddr) + } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() - log.Printf("worker started kind=%q concurrency=%d lease=%s poll=%s", - *kind, *concurrency, *leaseDuration, *pollInterval) - return w.Run(ctx) + log.Printf("worker started kind=%q concurrency=%d lease=%s poll=%s aging=%s", + *kind, *concurrency, *leaseDuration, *pollInterval, *aging) + err = w.Run(ctx) + + if srv != nil { + shutdownCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _ = srv.Shutdown(shutdownCtx) + } + if errors.Is(err, context.Canceled) { + // A signal cancelled the run. This is a normal shutdown, not an error. + return nil + } + return err } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 0000000..5baf950 --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,184 @@ +// Package metrics renders queue state in the Prometheus text exposition +// format. An operator can scrape a live endpoint, or print a one-shot snapshot +// with the metrics command. The package reads the shared SQLite store, so it +// observes every job and event without an extra collection pipeline. +package metrics + +import ( + "fmt" + "io" + "net/http" + "time" + + "github.com/local-first-job-queue/internal/queue" +) + +// The canonical state order keeps the exposition stable across scrapes. The +// failed state appears for legacy databases that predate the dead-letter +// queue, so exporters still report it. +var stateOrder = []queue.JobState{ + queue.StatePending, + queue.StateLeased, + queue.StateCompleted, + queue.StateDeadLetter, + queue.StateFailed, +} + +var eventTypeOrder = []queue.EventType{ + queue.EventEnqueued, + queue.EventScheduled, + queue.EventLeased, + queue.EventAcknowledged, + queue.EventFailed, + queue.EventRetried, + queue.EventRecovered, + queue.EventDeadLettered, + queue.EventRequeued, +} + +// Option configures a Collector. +type Option func(*Collector) + +// WithNow overrides the clock used for age calculations. Tests use it to make +// the oldest-pending metric deterministic. +func WithNow(fn func() time.Time) Option { + return func(c *Collector) { + c.now = fn + } +} + +// Collector reads queue state from the store and renders it in the Prometheus +// text format. Each call to Write computes a fresh snapshot, so no counters are +// kept between scrapes. +type Collector struct { + store *queue.SQLiteStore + now func() time.Time +} + +// New returns a Collector that reads from store. +func New(store *queue.SQLiteStore, opts ...Option) *Collector { + c := &Collector{store: store, now: time.Now} + for _, o := range opts { + o(c) + } + return c +} + +// Write renders the current queue state in the Prometheus text exposition +// format. The output is deterministic: families appear in a fixed order and +// label values are sorted. +func (c *Collector) Write(w io.Writer) error { + stats, err := c.store.GetQueueStats() + if err != nil { + return fmt.Errorf("queue stats: %w", err) + } + byKind, err := c.store.GetStateKindCounts() + if err != nil { + return fmt.Errorf("kind counts: %w", err) + } + evCounts, err := c.store.GetEventTypeCounts() + if err != nil { + return fmt.Errorf("event counts: %w", err) + } + + if err := c.writeStateGauge(w, stats); err != nil { + return err + } + if err := c.writeKindGauge(w, byKind); err != nil { + return err + } + if err := c.writeEventCounters(w, evCounts); err != nil { + return err + } + return c.writeOldestPending(w) +} + +func (c *Collector) writeStateGauge(w io.Writer, stats map[queue.JobState]int) error { + if _, err := fmt.Fprintln(w, "# HELP jobqueue_jobs Number of jobs in each state."); err != nil { + return err + } + if _, err := fmt.Fprintln(w, "# TYPE jobqueue_jobs gauge"); err != nil { + return err + } + for _, state := range stateOrder { + if _, err := fmt.Fprintf(w, "jobqueue_jobs{state=%q} %d\n", state, stats[state]); err != nil { + return err + } + } + return nil +} + +func (c *Collector) writeKindGauge(w io.Writer, byKind []queue.KindStateCount) error { + if _, err := fmt.Fprintln(w, "# HELP jobqueue_jobs_by_kind Number of jobs per kind and state."); err != nil { + return err + } + if _, err := fmt.Fprintln(w, "# TYPE jobqueue_jobs_by_kind gauge"); err != nil { + return err + } + for _, c := range byKind { + if _, err := fmt.Fprintf(w, "jobqueue_jobs_by_kind{kind=%q,state=%q} %d\n", c.Kind, c.State, c.Count); err != nil { + return err + } + } + return nil +} + +func (c *Collector) writeEventCounters(w io.Writer, evCounts []queue.EventTypeCount) error { + byType := map[queue.EventType]int{} + for _, c := range evCounts { + byType[c.EventType] = c.Count + } + if _, err := fmt.Fprintln(w, "# HELP jobqueue_events_total Number of events per event type."); err != nil { + return err + } + if _, err := fmt.Fprintln(w, "# TYPE jobqueue_events_total counter"); err != nil { + return err + } + for _, et := range eventTypeOrder { + if _, err := fmt.Fprintf(w, "jobqueue_events_total{type=%q} %d\n", et, byType[et]); err != nil { + return err + } + } + return nil +} + +func (c *Collector) writeOldestPending(w io.Writer) error { + ready, ok, err := c.store.GetOldestPendingReadyTime() + if err != nil { + return err + } + if _, err := fmt.Fprintln(w, "# HELP jobqueue_oldest_pending_seconds Age in seconds of the oldest pending job."); err != nil { + return err + } + if _, err := fmt.Fprintln(w, "# TYPE jobqueue_oldest_pending_seconds gauge"); err != nil { + return err + } + if !ok { + return nil + } + age := c.now().Sub(ready).Seconds() + _, err = fmt.Fprintf(w, "jobqueue_oldest_pending_seconds %g\n", age) + return err +} + +// Handler returns an HTTP handler that serves the queue metrics at /metrics. +// A request to any other path explains how to scrape. Each scrape calls Write, +// so the numbers always reflect the current database state. +func Handler(store *queue.SQLiteStore) http.Handler { + col := New(store) + mux := http.NewServeMux() + mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + if err := col.Write(w); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + }) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + fmt.Fprintf(w, "Local-first Durable Job Queue metrics.\nScrape GET /metrics.\n") + }) + return mux +} diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go new file mode 100644 index 0000000..cae970d --- /dev/null +++ b/internal/metrics/metrics_test.go @@ -0,0 +1,169 @@ +package metrics + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/local-first-job-queue/internal/queue" +) + +func newTestStore(t *testing.T) (*queue.SQLiteStore, *queue.Queue) { + t.Helper() + s, err := queue.NewSQLiteStore("file:metrics_" + t.Name() + "?mode=memory&cache=shared") + if err != nil { + t.Fatalf("new store: %v", err) + } + t.Cleanup(func() { s.Close() }) + return s, queue.NewQueue(s) +} + +func render(t *testing.T, col *Collector) string { + t.Helper() + var b strings.Builder + if err := col.Write(&b); err != nil { + t.Fatalf("write metrics: %v", err) + } + return b.String() +} + +// TestEmptyQueueRendersZeroMetrics verifies that an empty queue still produces +// the full metric families with explicit zero values. Absent families would +// hide state from a dashboard, so every known state and event type appears. +func TestEmptyQueueRendersZeroMetrics(t *testing.T) { + s, _ := newTestStore(t) + got := render(t, New(s)) + + for _, want := range []string{ + "# HELP jobqueue_jobs Number of jobs in each state.", + "# TYPE jobqueue_jobs gauge", + `jobqueue_jobs{state="pending"} 0`, + `jobqueue_jobs{state="leased"} 0`, + `jobqueue_jobs{state="completed"} 0`, + `jobqueue_jobs{state="dead_letter"} 0`, + `jobqueue_jobs{state="failed"} 0`, + "# TYPE jobqueue_events_total counter", + `jobqueue_events_total{type="enqueued"} 0`, + `jobqueue_events_total{type="recovered"} 0`, + "# TYPE jobqueue_oldest_pending_seconds gauge", + } { + if !strings.Contains(got, want) { + t.Errorf("metrics missing %q in:\n%s", want, got) + } + } + // The value line must be absent: an empty queue has nothing to measure. + if strings.Contains(got, "jobqueue_oldest_pending_seconds 0") || + strings.Contains(got, "jobqueue_oldest_pending_seconds\t0") { + t.Errorf("expected no oldest pending value when the queue is empty:\n%s", got) + } +} + +// TestWriteReflectsWorkload verifies that completed, dead-lettered, and pending +// jobs produce the expected gauges and that the event counters match the log. +func TestWriteReflectsWorkload(t *testing.T) { + s, q := newTestStore(t) + ctx := context.Background() + + if _, err := q.Enqueue("email", `{"to":"a@example.com"}`); err != nil { + t.Fatalf("enqueue email: %v", err) + } + // Enqueue the failing report job with a higher priority so the first lease + // of the report kind returns it, independent of enqueue timing. + flaky, err := q.Enqueue("report", `{}`, queue.WithMaxAttempts(1), queue.WithPriority(1)) + if err != nil { + t.Fatalf("enqueue flaky: %v", err) + } + if _, err := q.Enqueue("report", `{}`); err != nil { + t.Fatalf("enqueue report: %v", err) + } + + job, err := q.Lease(ctx, "email", time.Minute) + if err != nil || job == nil { + t.Fatalf("lease email: %v %v", job, err) + } + if err := q.Acknowledge(job.ID); err != nil { + t.Fatalf("ack: %v", err) + } + + leased, err := q.Lease(ctx, "report", time.Minute) + if err != nil || leased == nil || leased.ID != flaky.ID { + t.Fatalf("lease flaky: %v %v", leased, err) + } + if err := q.Fail(leased.ID, "boom"); err != nil { + t.Fatalf("fail: %v", err) + } + + got := render(t, New(s)) + + for _, want := range []string{ + `jobqueue_jobs{state="pending"} 1`, + `jobqueue_jobs{state="completed"} 1`, + `jobqueue_jobs{state="dead_letter"} 1`, + `jobqueue_jobs_by_kind{kind="email",state="completed"} 1`, + `jobqueue_jobs_by_kind{kind="report",state="pending"} 1`, + `jobqueue_jobs_by_kind{kind="report",state="dead_letter"} 1`, + `jobqueue_events_total{type="enqueued"} 3`, + `jobqueue_events_total{type="leased"} 2`, + `jobqueue_events_total{type="acknowledged"} 1`, + `jobqueue_events_total{type="dead_lettered"} 1`, + } { + if !strings.Contains(got, want) { + t.Errorf("metrics missing %q in:\n%s", want, got) + } + } +} + +// TestOldestPendingAgeIsDeterministic verifies that the oldest-pending metric +// uses the ready time and the injected clock. A scheduled job reports its exact +// age instead of a wall-clock value. +func TestOldestPendingAgeIsDeterministic(t *testing.T) { + s, q := newTestStore(t) + + now := time.Now().UTC() + job, err := q.Enqueue("report", `{}`, queue.WithRunAt(now.Add(-30*time.Second))) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + if job.State != queue.StatePending { + t.Fatalf("expected pending, got %s", job.State) + } + + col := New(s, WithNow(func() time.Time { return now })) + got := render(t, col) + if want := "jobqueue_oldest_pending_seconds 30\n"; !strings.Contains(got, want) { + t.Errorf("expected %q in:\n%s", strings.TrimSpace(want), got) + } +} + +// TestHandlerScrapes verifies that the HTTP handler serves the exposition +// format on /metrics and a short landing page at the root. +func TestHandlerScrapes(t *testing.T) { + s, _ := newTestStore(t) + h := Handler(s) + + req := httptest.NewRequest(http.MethodGet, "/metrics", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + if ct := rec.Header().Get("Content-Type"); !strings.Contains(ct, "text/plain") { + t.Errorf("expected text content type, got %q", ct) + } + if !strings.Contains(rec.Body.String(), "# TYPE jobqueue_jobs gauge") { + t.Errorf("expected metrics body, got %q", rec.Body.String()) + } + + req = httptest.NewRequest(http.MethodGet, "/", nil) + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 at root, got %d", rec.Code) + } + if !strings.Contains(rec.Body.String(), "/metrics") { + t.Errorf("expected landing page to mention /metrics, got %q", rec.Body.String()) + } +} diff --git a/internal/queue/models.go b/internal/queue/models.go index 28b7f86..4042af0 100644 --- a/internal/queue/models.go +++ b/internal/queue/models.go @@ -9,6 +9,10 @@ const ( StateLeased JobState = "leased" StateCompleted JobState = "completed" StateFailed JobState = "failed" + // StateDeadLetter is the terminal state for jobs that exhausted their + // attempt budget. A dead-lettered job stays inspectable and an operator can + // requeue it with the Requeue method. + StateDeadLetter JobState = "dead_letter" ) type Job struct { @@ -38,6 +42,8 @@ const ( EventFailed EventType = "failed" EventRetried EventType = "retried" EventRecovered EventType = "recovered" + EventDeadLettered EventType = "dead_lettered" + EventRequeued EventType = "requeued" ) type Event struct { @@ -53,3 +59,18 @@ type QueueSnapshot struct { Events []Event `json:"events"` Stats map[JobState]int `json:"stats"` } + +// KindStateCount reports how many jobs share one kind and state. Metrics and +// inspection tools group jobs by these two dimensions. +type KindStateCount struct { + Kind string `json:"kind"` + State JobState `json:"state"` + Count int `json:"count"` +} + +// EventTypeCount reports how many events share one event type. The count is +// monotonic while the event log keeps every row. +type EventTypeCount struct { + EventType EventType `json:"event_type"` + Count int `json:"count"` +} diff --git a/internal/queue/queue.go b/internal/queue/queue.go index 9a20d93..6e5a927 100644 --- a/internal/queue/queue.go +++ b/internal/queue/queue.go @@ -11,7 +11,7 @@ import ( // DefaultMaxAttempts is the maximum number of attempts allowed for a job, // including the first attempt. A job with this value runs once and may retry -// twice more before it enters the failed state. +// twice more before it enters the dead-letter state. const DefaultMaxAttempts = 3 // DefaultPriority is used when an enqueue call does not set a priority. @@ -75,6 +75,30 @@ func WithRunAfter(d time.Duration) EnqueueOption { } } +type RequeueOption func(*requeueConfig) + +type requeueConfig struct { + maxAttempts int + payload *string +} + +// RequeueWithMaxAttempts sets a new attempt budget for a requeued job. The +// default keeps the budget the job had when it was dead-lettered. +func RequeueWithMaxAttempts(n int) RequeueOption { + return func(c *requeueConfig) { + c.maxAttempts = n + } +} + +// RequeueWithPayload replaces the payload of a requeued job. The default keeps +// the original payload, so an operator can fix the job data before retrying. +func RequeueWithPayload(payload string) RequeueOption { + return func(c *requeueConfig) { + v := payload + c.payload = &v + } +} + type Queue struct { store *SQLiteStore } @@ -212,8 +236,8 @@ func (q *Queue) Fail(jobID string, errMsg string) error { return fmt.Errorf("fail job: %w", err) } - evType := EventFailed - meta := errMsg + evType := EventDeadLettered + meta := fmt.Sprintf("attempt %d/%d exhausted: %s", job.RetryCount+1, job.MaxAttempts, errMsg) if shouldRetry { evType = EventRetried meta = fmt.Sprintf("attempt %d/%d: %s", job.RetryCount+1, job.MaxAttempts, errMsg) @@ -232,7 +256,9 @@ func (q *Queue) Fail(jobID string, errMsg string) error { // Recover returns orphaned leases to the pending state. A lease is orphaned // when its deadline passed and no worker acknowledged it. Recovered jobs get -// one extra attempt. When no attempt remains, the job enters the failed state. +// one extra attempt. When no attempt remains, the job enters the dead-letter +// state. Each recovered job logs one event: recovered, or dead_lettered when +// the attempt budget is gone. func (q *Queue) Recover() (int, error) { recovered, err := q.store.RecoverOrphanedLeases() if err != nil { @@ -246,6 +272,10 @@ func (q *Queue) Recover() (int, error) { Timestamp: now, } meta := fmt.Sprintf("attempt %d/%d", r.RetryCount, r.MaxAttempts) + if r.State == StateDeadLetter { + ev.EventType = EventDeadLettered + meta = fmt.Sprintf("attempt %d/%d exhausted", r.RetryCount, r.MaxAttempts) + } ev.Metadata = &meta if err := q.store.AppendEvent(ev); err != nil { return len(recovered), fmt.Errorf("log recovery event: %w", err) @@ -254,6 +284,45 @@ func (q *Queue) Recover() (int, error) { return len(recovered), nil } +// Requeue returns a dead-lettered job to the pending state with a fresh +// attempt budget. The original job data is kept unless an option overrides it. +// A requeued job can fail again and re-enter the dead-letter queue. +func (q *Queue) Requeue(jobID string, opts ...RequeueOption) (*Job, error) { + cfg := requeueConfig{} + for _, o := range opts { + o(&cfg) + } + + job, err := q.store.GetJob(jobID) + if err != nil { + return nil, fmt.Errorf("get job: %w", err) + } + if job.State != StateDeadLetter { + return nil, fmt.Errorf("job %s is %s, not dead-lettered", jobID, job.State) + } + maxAttempts := job.MaxAttempts + if cfg.maxAttempts >= 1 { + maxAttempts = cfg.maxAttempts + } + + updated, err := q.store.RequeueJob(jobID, maxAttempts, cfg.payload) + if err != nil { + return nil, err + } + + meta := fmt.Sprintf("attempts reset to 0/%d", maxAttempts) + ev := Event{ + JobID: jobID, + EventType: EventRequeued, + Timestamp: time.Now().UTC(), + Metadata: &meta, + } + if err := q.store.AppendEvent(ev); err != nil { + return nil, fmt.Errorf("log event: %w", err) + } + return &updated, nil +} + func (q *Queue) Inspect() (*QueueSnapshot, error) { jobs, err := q.store.GetAllJobs() if err != nil { diff --git a/internal/queue/queue_test.go b/internal/queue/queue_test.go index 135f4df..9ed92df 100644 --- a/internal/queue/queue_test.go +++ b/internal/queue/queue_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "strconv" + "strings" "sync" "testing" "time" @@ -21,6 +22,17 @@ func newTestStore(t *testing.T) *SQLiteStore { return s } +func newTestStoreWithAging(t *testing.T, interval time.Duration) *SQLiteStore { + t.Helper() + s, err := NewSQLiteStore("file:test_"+t.Name()+"?mode=memory&cache=shared", + WithAgingInterval(interval)) + if err != nil { + t.Fatalf("new store: %v", err) + } + t.Cleanup(func() { s.Close() }) + return s +} + func TestMigrationAddsPriorityToLegacyDatabase(t *testing.T) { path := filepath.Join(t.TempDir(), "legacy.db") raw, err := sql.Open("sqlite", path) @@ -170,8 +182,8 @@ func TestFailAndRetry(t *testing.T) { } snap, _ := q.Inspect() - if snap.Stats[StateFailed] != 1 { - t.Errorf("expected 1 failed, got %d", snap.Stats[StateFailed]) + if snap.Stats[StateDeadLetter] != 1 { + t.Errorf("expected 1 dead-lettered, got %d", snap.Stats[StateDeadLetter]) } } @@ -227,6 +239,161 @@ func TestPriorityOrdering(t *testing.T) { } } +// TestPriorityAgingLiftsOldJob verifies that a job which has waited for several +// aging intervals gains enough effective priority to overtake a fresher +// higher-priority job. The store backdates the first job so the outcome does +// not depend on wall-clock timing. +func TestPriorityAgingLiftsOldJob(t *testing.T) { + s := newTestStoreWithAging(t, time.Second) + q := NewQueue(s) + ctx := context.Background() + + old, err := q.Enqueue("test", `{"name":"old"}`, WithPriority(0)) + if err != nil { + t.Fatalf("enqueue old: %v", err) + } + // Backdate the old job by five aging intervals so its effective priority + // becomes 0 + 5 = 5, above the fresh job's priority of 3. + if _, err := s.db.Exec( + `UPDATE jobs SET created_at = ? WHERE id = ?`, + time.Now().UTC().Add(-5*time.Second).Format(sqliteTimeFormat), old.ID); err != nil { + t.Fatalf("backdate old: %v", err) + } + fresh, err := q.Enqueue("test", `{"name":"fresh"}`, WithPriority(3)) + if err != nil { + t.Fatalf("enqueue fresh: %v", err) + } + + got, err := q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease: %v", err) + } + if got == nil || got.ID != old.ID { + t.Fatalf("expected aged job %s to lease first, got %+v", old.ID, got) + } + + got, err = q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease second: %v", err) + } + if got == nil || got.ID != fresh.ID { + t.Fatalf("expected fresh job %s second, got %+v", fresh.ID, got) + } +} + +// TestPriorityAgingDisabledKeepsOrdering verifies that a store without an aging +// interval ignores job age. A backdated low-priority job still leases after a +// fresh high-priority job, so the default behavior is unchanged. +func TestPriorityAgingDisabledKeepsOrdering(t *testing.T) { + s := newTestStore(t) + q := NewQueue(s) + ctx := context.Background() + + old, err := q.Enqueue("test", `{"name":"old"}`, WithPriority(0)) + if err != nil { + t.Fatalf("enqueue old: %v", err) + } + if _, err := s.db.Exec( + `UPDATE jobs SET created_at = ? WHERE id = ?`, + time.Now().UTC().Add(-5*time.Second).Format(sqliteTimeFormat), old.ID); err != nil { + t.Fatalf("backdate old: %v", err) + } + fresh, err := q.Enqueue("test", `{"name":"fresh"}`, WithPriority(3)) + if err != nil { + t.Fatalf("enqueue fresh: %v", err) + } + + got, err := q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease: %v", err) + } + if got == nil || got.ID != fresh.ID { + t.Fatalf("expected fresh job %s first without aging, got %+v", fresh.ID, got) + } + + got, err = q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease second: %v", err) + } + if got == nil || got.ID != old.ID { + t.Fatalf("expected old job %s second, got %+v", old.ID, got) + } +} + +// TestPriorityAgingTiesUseDeterministicOrder verifies that jobs with the same +// effective priority fall back to the deterministic tie breakers. Two equally +// old jobs of the same priority must lease in creation order. +func TestPriorityAgingTiesUseDeterministicOrder(t *testing.T) { + s := newTestStoreWithAging(t, time.Second) + q := NewQueue(s) + ctx := context.Background() + + first, err := q.Enqueue("test", `{"name":"first"}`, WithPriority(0)) + if err != nil { + t.Fatalf("enqueue first: %v", err) + } + second, err := q.Enqueue("test", `{"name":"second"}`, WithPriority(0)) + if err != nil { + t.Fatalf("enqueue second: %v", err) + } + // Both jobs get the same aging boost, so created_at and then id decide. + if _, err := s.db.Exec( + `UPDATE jobs SET created_at = ? WHERE id = ?`, + time.Now().UTC().Add(-3*time.Second).Format(sqliteTimeFormat), first.ID); err != nil { + t.Fatalf("backdate first: %v", err) + } + if _, err := s.db.Exec( + `UPDATE jobs SET created_at = ? WHERE id = ?`, + time.Now().UTC().Add(-3*time.Second).Format(sqliteTimeFormat), second.ID); err != nil { + t.Fatalf("backdate second: %v", err) + } + + got, err := q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease: %v", err) + } + if got == nil || got.ID != first.ID { + t.Fatalf("expected first job %s to lease first on tie, got %+v", first.ID, got) + } +} + +// TestPriorityAgingRespectsSchedule verifies that aging does not let a future +// scheduled job bypass its run_at time. A scheduled job only ages once it is +// ready, so a backdated run_at but far-future creation is not enough to lease. +func TestPriorityAgingRespectsSchedule(t *testing.T) { + s := newTestStoreWithAging(t, time.Second) + q := NewQueue(s) + ctx := context.Background() + + future, err := q.Enqueue("test", `{}`, WithPriority(100), WithRunAt(time.Now().Add(time.Hour))) + if err != nil { + t.Fatalf("enqueue future: %v", err) + } + ready, err := q.Enqueue("test", `{}`, WithPriority(0)) + if err != nil { + t.Fatalf("enqueue ready: %v", err) + } + + got, err := q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease: %v", err) + } + if got == nil || got.ID != ready.ID { + t.Fatalf("expected ready job %s first, got %+v", ready.ID, got) + } + if err := q.Acknowledge(got.ID); err != nil { + t.Fatalf("ack: %v", err) + } + + got, err = q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease after ready job: %v", err) + } + if got != nil { + t.Fatalf("expected scheduled job %s to remain pending, got %+v", future.ID, got) + } +} + func TestFutureHighPriorityJobDoesNotBypassSchedule(t *testing.T) { s := newTestStore(t) q := NewQueue(s) @@ -412,7 +579,7 @@ func TestRecoverOrphanedLeases(t *testing.T) { // TestRecoverExhaustsAttempts verifies that recovery respects the attempt budget. // A job with a single allowed attempt has no room to retry, so an expired lease -// must move it to the failed state instead of looping forever. +// must move it to the dead-letter state instead of looping forever. func TestRecoverExhaustsAttempts(t *testing.T) { s := newTestStore(t) q := NewQueue(s) @@ -439,8 +606,292 @@ func TestRecoverExhaustsAttempts(t *testing.T) { if err != nil { t.Fatalf("get job: %v", err) } - if got.State != StateFailed { - t.Errorf("expected failed after recovery exhausted attempts, got %s", got.State) + if got.State != StateDeadLetter { + t.Errorf("expected dead-letter after recovery exhausted attempts, got %s", got.State) + } + + events, err := s.GetJobEvents(job.ID) + if err != nil { + t.Fatalf("get events: %v", err) + } + last := events[len(events)-1] + if last.EventType != EventDeadLettered { + t.Errorf("expected dead_lettered event, got %s", last.EventType) + } + if last.Metadata == nil || !strings.Contains(*last.Metadata, "exhausted") { + t.Errorf("expected exhausted marker in metadata, got %v", last.Metadata) + } +} + +// TestDeadLetterEventsAndRequeue verifies the full dead-letter workflow. A job +// that exhausts its attempts enters the dead_letter state with a dead_lettered +// event. Requeue returns it to pending with a reset attempt budget and logs a +// requeued event. The requeued job can lease and complete again. +func TestDeadLetterEventsAndRequeue(t *testing.T) { + s := newTestStore(t) + q := NewQueue(s) + ctx := context.Background() + + job, err := q.Enqueue("test", `{"name":"flaky"}`, WithMaxAttempts(2)) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + + // Attempt 1 fails and is retried. + leased, err := q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease 1: %v", err) + } + if err := q.Fail(leased.ID, "boom 1"); err != nil { + t.Fatalf("fail 1: %v", err) + } + + // Attempt 2 fails and exhausts the budget. + leased, err = q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease 2: %v", err) + } + if err := q.Fail(leased.ID, "boom 2"); err != nil { + t.Fatalf("fail 2: %v", err) + } + + got, err := s.GetJob(job.ID) + if err != nil { + t.Fatalf("get job: %v", err) + } + if got.State != StateDeadLetter { + t.Fatalf("expected dead_letter, got %s", got.State) + } + + events, err := s.GetJobEvents(job.ID) + if err != nil { + t.Fatalf("get events: %v", err) + } + want := []EventType{EventEnqueued, EventLeased, EventRetried, EventLeased, EventDeadLettered} + if len(events) != len(want) { + t.Fatalf("expected %d events, got %d: %+v", len(want), len(events), events) + } + for i, et := range want { + if events[i].EventType != et { + t.Errorf("event %d: expected %s, got %s", i, et, events[i].EventType) + } + } + if events[len(events)-1].Metadata == nil || + !strings.Contains(*events[len(events)-1].Metadata, "2/2") { + t.Errorf("expected attempt marker in dead_lettered metadata, got %v", + events[len(events)-1].Metadata) + } + + // Requeue resets the job and logs a requeued event. + requeued, err := q.Requeue(job.ID, RequeueWithPayload(`{"name":"flaky","fixed":true}`)) + if err != nil { + t.Fatalf("requeue: %v", err) + } + if requeued.State != StatePending { + t.Errorf("expected pending after requeue, got %s", requeued.State) + } + if requeued.RetryCount != 0 { + t.Errorf("expected retry_count 0 after requeue, got %d", requeued.RetryCount) + } + if requeued.Payload != `{"name":"flaky","fixed":true}` { + t.Errorf("expected replaced payload, got %q", requeued.Payload) + } + + events, err = s.GetJobEvents(job.ID) + if err != nil { + t.Fatalf("get events after requeue: %v", err) + } + last := events[len(events)-1] + if last.EventType != EventRequeued { + t.Errorf("expected requeued event, got %s", last.EventType) + } + + // The requeued job completes normally. + leased, err = q.Lease(ctx, "test", time.Minute) + if err != nil || leased == nil || leased.ID != job.ID { + t.Fatalf("expected requeued job to lease, got %v %v", leased, err) + } + if err := q.Acknowledge(leased.ID); err != nil { + t.Fatalf("ack requeued: %v", err) + } + snap, _ := q.Inspect() + if snap.Stats[StateCompleted] != 1 { + t.Errorf("expected 1 completed, got %d", snap.Stats[StateCompleted]) + } +} + +// TestRequeueRequiresDeadLetter verifies that Requeue rejects jobs that are not +// in the dead-letter state. A pending job must stay pending. +func TestRequeueRequiresDeadLetter(t *testing.T) { + s := newTestStore(t) + q := NewQueue(s) + + job, err := q.Enqueue("test", `{}`) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + if _, err := q.Requeue(job.ID); err == nil { + t.Fatal("expected error requeueing a pending job") + } + + snap, _ := q.Inspect() + if snap.Stats[StatePending] != 1 { + t.Errorf("expected the pending job to remain, got %+v", snap.Stats) + } +} + +// TestRequeueWithMaxAttempts verifies that the requeue attempt budget override +// is honored and that a requeued job may exhaust its budget again. +func TestRequeueWithMaxAttempts(t *testing.T) { + s := newTestStore(t) + q := NewQueue(s) + ctx := context.Background() + + job, err := q.Enqueue("test", `{}`, WithMaxAttempts(1)) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + leased, err := q.Lease(ctx, "test", time.Minute) + if err != nil { + t.Fatalf("lease: %v", err) + } + if err := q.Fail(leased.ID, "boom"); err != nil { + t.Fatalf("fail: %v", err) + } + + requeued, err := q.Requeue(job.ID, RequeueWithMaxAttempts(5)) + if err != nil { + t.Fatalf("requeue: %v", err) + } + if requeued.MaxAttempts != 5 { + t.Errorf("expected max_attempts 5, got %d", requeued.MaxAttempts) + } + if requeued.RetryCount != 0 { + t.Errorf("expected retry_count 0, got %d", requeued.RetryCount) + } +} + +// TestGetStateKindCountsAndEventTypeCounts verifies the aggregation queries +// used by the metrics exporter. Jobs group by kind and state, events group by +// type, and both results keep a stable order. +func TestGetStateKindCountsAndEventTypeCounts(t *testing.T) { + s := newTestStore(t) + q := NewQueue(s) + ctx := context.Background() + + if _, err := q.Enqueue("email", `{}`); err != nil { + t.Fatalf("enqueue email: %v", err) + } + // Enqueue the failing report job with a higher priority so the first lease + // of the report kind returns it, independent of enqueue timing. + flaky, err := q.Enqueue("report", `{}`, WithMaxAttempts(1), WithPriority(1)) + if err != nil { + t.Fatalf("enqueue flaky: %v", err) + } + if _, err := q.Enqueue("report", `{}`); err != nil { + t.Fatalf("enqueue report: %v", err) + } + + job, err := q.Lease(ctx, "email", time.Minute) + if err != nil || job == nil { + t.Fatalf("lease email: %v %v", job, err) + } + if err := q.Acknowledge(job.ID); err != nil { + t.Fatalf("ack: %v", err) + } + leased, err := q.Lease(ctx, "report", time.Minute) + if err != nil || leased == nil || leased.ID != flaky.ID { + t.Fatalf("lease flaky: %v %v", leased, err) + } + if err := q.Fail(leased.ID, "boom"); err != nil { + t.Fatalf("fail: %v", err) + } + + counts, err := s.GetStateKindCounts() + if err != nil { + t.Fatalf("kind counts: %v", err) + } + wantKinds := []KindStateCount{ + {Kind: "email", State: StateCompleted, Count: 1}, + {Kind: "report", State: StateDeadLetter, Count: 1}, + {Kind: "report", State: StatePending, Count: 1}, + } + if len(counts) != len(wantKinds) { + t.Fatalf("expected %d kind counts, got %d: %+v", len(wantKinds), len(counts), counts) + } + for i, want := range wantKinds { + if counts[i] != want { + t.Errorf("kind count %d: expected %+v, got %+v", i, want, counts[i]) + } + } + + events, err := s.GetEventTypeCounts() + if err != nil { + t.Fatalf("event counts: %v", err) + } + byType := map[EventType]int{} + for _, e := range events { + byType[e.EventType] = e.Count + } + if byType[EventEnqueued] != 3 { + t.Errorf("expected 3 enqueued events, got %d", byType[EventEnqueued]) + } + if byType[EventAcknowledged] != 1 { + t.Errorf("expected 1 acknowledged event, got %d", byType[EventAcknowledged]) + } + if byType[EventDeadLettered] != 1 { + t.Errorf("expected 1 dead_lettered event, got %d", byType[EventDeadLettered]) + } +} + +// TestGetOldestPendingReadyTime verifies that the oldest pending job is found +// by its ready time, and that an empty queue reports no value. +func TestGetOldestPendingReadyTime(t *testing.T) { + s := newTestStore(t) + q := NewQueue(s) + + if _, ok, err := s.GetOldestPendingReadyTime(); err != nil || ok { + t.Fatalf("expected no oldest pending on empty queue, ok=%v err=%v", ok, err) + } + + old, err := q.Enqueue("test", `{}`, WithRunAt(time.Now().Add(-time.Hour))) + if err != nil { + t.Fatalf("enqueue old: %v", err) + } + if _, err := q.Enqueue("test", `{}`); err != nil { + t.Fatalf("enqueue new: %v", err) + } + + ready, ok, err := s.GetOldestPendingReadyTime() + if err != nil { + t.Fatalf("oldest pending: %v", err) + } + if !ok { + t.Fatal("expected an oldest pending job") + } + if old.RunAt == nil || !ready.Equal(*old.RunAt) { + t.Errorf("expected ready time %v, got %v", old.RunAt, ready) + } +} + +// TestGetStateKindCountsEmptyQueue verifies the aggregation queries return +// empty results for a fresh database. +func TestGetStateKindCountsEmptyQueue(t *testing.T) { + s := newTestStore(t) + + counts, err := s.GetStateKindCounts() + if err != nil { + t.Fatalf("kind counts: %v", err) + } + if len(counts) != 0 { + t.Errorf("expected no kind counts, got %+v", counts) + } + events, err := s.GetEventTypeCounts() + if err != nil { + t.Fatalf("event counts: %v", err) + } + if len(events) != 0 { + t.Errorf("expected no event counts, got %+v", events) } } diff --git a/internal/queue/sqlite.go b/internal/queue/sqlite.go index 418d54a..34a7faf 100644 --- a/internal/queue/sqlite.go +++ b/internal/queue/sqlite.go @@ -10,11 +10,40 @@ import ( const sqliteTimeFormat = time.RFC3339Nano +// DefaultAgingInterval is the recommended priority aging interval. A job gains +// one priority point per interval it has waited, which prevents a constant +// high-priority stream from starving lower-priority work. The work command +// uses this value unless the operator overrides it. +const DefaultAgingInterval = 30 * time.Second + +type StoreOption func(*storeConfig) + +type storeConfig struct { + agingInterval time.Duration + agingSet bool +} + +// WithAgingInterval sets the priority aging interval. A pending job gains one +// priority point per interval it has waited, so a lower-priority job eventually +// overtakes a constant stream of higher-priority work. A zero interval keeps +// the plain priority ordering. When the option is absent, aging is disabled. +func WithAgingInterval(d time.Duration) StoreOption { + return func(c *storeConfig) { + c.agingInterval = d + c.agingSet = true + } +} + type SQLiteStore struct { - db *sql.DB + db *sql.DB + agingInterval time.Duration } -func NewSQLiteStore(dbPath string) (*SQLiteStore, error) { +func NewSQLiteStore(dbPath string, opts ...StoreOption) (*SQLiteStore, error) { + cfg := storeConfig{} + for _, o := range opts { + o(&cfg) + } db, err := sql.Open("sqlite", dbPath) if err != nil { return nil, fmt.Errorf("open db: %w", err) @@ -24,7 +53,7 @@ func NewSQLiteStore(dbPath string) (*SQLiteStore, error) { if err := migrate(db); err != nil { return nil, fmt.Errorf("migrate: %w", err) } - return &SQLiteStore{db: db}, nil + return &SQLiteStore{db: db, agingInterval: cfg.agingInterval}, nil } func migrate(db *sql.DB) error { @@ -188,11 +217,12 @@ func (s *SQLiteStore) LeaseJob(kind string, leaseDuration time.Duration) (*Job, } defer tx.Rollback() - row := tx.QueryRow( - `SELECT id, kind, payload, state, retry_count, max_attempts, priority, idempotency_key, created_at, updated_at, leased_until, run_at + orderBy, orderArgs := s.pendingOrderBy(now) + query := `SELECT id, kind, payload, state, retry_count, max_attempts, priority, idempotency_key, created_at, updated_at, leased_until, run_at FROM jobs WHERE kind = ? AND state = 'pending' - AND (run_at IS NULL OR run_at <= ?) - ORDER BY priority DESC, COALESCE(run_at, created_at) ASC, created_at ASC, id ASC LIMIT 1`, kind, now.Format(sqliteTimeFormat)) + AND (run_at IS NULL OR run_at <= ?) ORDER BY ` + orderBy + ` LIMIT 1` + args := append([]any{kind, now.Format(sqliteTimeFormat)}, orderArgs...) + row := tx.QueryRow(query, args...) job, err := scanJob(row) if err == sql.ErrNoRows { return nil, nil @@ -289,18 +319,54 @@ func (s *SQLiteStore) FailJob(id string, retry bool) error { return nil } _, err := s.db.Exec( - `UPDATE jobs SET state = 'failed', leased_until = NULL, updated_at = ? + `UPDATE jobs SET state = 'dead_letter', retry_count = retry_count + 1, leased_until = NULL, updated_at = ? WHERE id = ? AND state = 'leased'`, now.Format(sqliteTimeFormat), id) if err != nil { - return fmt.Errorf("fail job: %w", err) + return fmt.Errorf("dead-letter job: %w", err) } return nil } +// RequeueJob returns a dead-lettered job to the pending state with a fresh +// attempt budget. The caller may supply a new payload and a new attempt limit. +// When the job is not dead-lettered, the update matches no row and an error is +// returned. +func (s *SQLiteStore) RequeueJob(id string, maxAttempts int, payload *string) (Job, error) { + now := time.Now().UTC() + var ( + res sql.Result + err error + ) + if payload != nil { + res, err = s.db.Exec( + `UPDATE jobs SET state = 'pending', retry_count = 0, max_attempts = ?, + payload = ?, leased_until = NULL, updated_at = ? + WHERE id = ? AND state = 'dead_letter'`, + maxAttempts, *payload, now.Format(sqliteTimeFormat), id) + } else { + res, err = s.db.Exec( + `UPDATE jobs SET state = 'pending', retry_count = 0, max_attempts = ?, + leased_until = NULL, updated_at = ? + WHERE id = ? AND state = 'dead_letter'`, + maxAttempts, now.Format(sqliteTimeFormat), id) + } + if err != nil { + return Job{}, fmt.Errorf("requeue job: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return Job{}, fmt.Errorf("rows affected: %w", err) + } + if n == 0 { + return Job{}, fmt.Errorf("job %s is not dead-lettered", id) + } + return s.GetJob(id) +} + // RecoverOrphanedLeases finds jobs whose lease deadline passed and returns // them to the pending state. Each recovery consumes one attempt. When no -// attempt remains, the job enters the failed state instead of looping. +// attempt remains, the job enters the dead-letter state instead of looping. // The returned slice reports the jobs that were touched, with their updated // retry counters, so callers can log a recovery event per job. func (s *SQLiteStore) RecoverOrphanedLeases() ([]Job, error) { @@ -326,18 +392,19 @@ func (s *SQLiteStore) RecoverOrphanedLeases() ([]Job, error) { return nil, nil } - for _, j := range orphans { + for i := range orphans { + j := &orphans[i] nextAttempt := j.RetryCount + 1 if nextAttempt >= j.MaxAttempts { _, err := s.db.Exec( - `UPDATE jobs SET state = 'failed', leased_until = NULL, updated_at = ? + `UPDATE jobs SET state = 'dead_letter', retry_count = ?, leased_until = NULL, updated_at = ? WHERE id = ? AND state = 'leased'`, - now.Format(sqliteTimeFormat), j.ID) + nextAttempt, now.Format(sqliteTimeFormat), j.ID) if err != nil { - return nil, fmt.Errorf("fail orphaned job %s: %w", j.ID, err) + return nil, fmt.Errorf("dead-letter orphaned job %s: %w", j.ID, err) } j.RetryCount = nextAttempt - j.State = StateFailed + j.State = StateDeadLetter continue } _, err := s.db.Exec( @@ -353,8 +420,28 @@ func (s *SQLiteStore) RecoverOrphanedLeases() ([]Job, error) { return orphans, nil } +// pendingOrderBy builds the ORDER BY clause for ready jobs. When aging is +// enabled, a job gains one priority point for each aging interval it has +// waited since it became ready. The boost is added to the stored priority, so +// an older low-priority job can overtake a fresher high-priority job. The +// returned arguments bind the "now" timestamp and the aging interval, and are +// appended to the caller's query arguments. Aging is measured from the earlier +// of run_at and created_at, so a scheduled job only ages once it is ready. +func (s *SQLiteStore) pendingOrderBy(now time.Time) (string, []any) { + if s.agingInterval <= 0 { + return `priority DESC, COALESCE(run_at, created_at) ASC, created_at ASC, id ASC`, nil + } + agingSec := s.agingInterval.Seconds() + orderBy := `(priority + CAST((julianday(?) - julianday(COALESCE(run_at, created_at))) * 86400.0 / ? AS INTEGER)) DESC, + priority DESC, COALESCE(run_at, created_at) ASC, created_at ASC, id ASC` + args := []any{now.Format(sqliteTimeFormat), agingSec} + return orderBy, args +} + func (s *SQLiteStore) GetPendingJobs() ([]Job, error) { - return s.queryJobs(`WHERE state = 'pending' ORDER BY priority DESC, COALESCE(run_at, created_at) ASC, created_at ASC, id ASC`) + orderBy, orderArgs := s.pendingOrderBy(time.Now().UTC()) + where := "WHERE state = 'pending' ORDER BY " + orderBy + return s.queryJobs(where, orderArgs...) } func (s *SQLiteStore) GetLeasedJobs() ([]Job, error) { @@ -365,10 +452,10 @@ func (s *SQLiteStore) GetAllJobs() ([]Job, error) { return s.queryJobs(`ORDER BY created_at DESC`) } -func (s *SQLiteStore) queryJobs(where string) ([]Job, error) { +func (s *SQLiteStore) queryJobs(where string, args ...any) ([]Job, error) { rows, err := s.db.Query( `SELECT id, kind, payload, state, retry_count, max_attempts, priority, idempotency_key, created_at, updated_at, leased_until, run_at - FROM jobs ` + where) + FROM jobs `+where, args...) if err != nil { return nil, fmt.Errorf("query jobs: %w", err) } @@ -404,6 +491,71 @@ func (s *SQLiteStore) GetQueueStats() (map[JobState]int, error) { return stats, nil } +// GetStateKindCounts returns the number of jobs for each (kind, state) pair. +// The result is stable, so exporters can render it in a fixed order. +func (s *SQLiteStore) GetStateKindCounts() ([]KindStateCount, error) { + rows, err := s.db.Query(`SELECT kind, state, COUNT(*) FROM jobs GROUP BY kind, state ORDER BY kind, state`) + if err != nil { + return nil, fmt.Errorf("query kind counts: %w", err) + } + defer rows.Close() + + var counts []KindStateCount + for rows.Next() { + var c KindStateCount + var state string + if err := rows.Scan(&c.Kind, &state, &c.Count); err != nil { + return nil, fmt.Errorf("scan kind count: %w", err) + } + c.State = JobState(state) + counts = append(counts, c) + } + return counts, nil +} + +// GetEventTypeCounts returns the number of events for each event type. The +// event log is append-only, so each count grows and never shrinks. +func (s *SQLiteStore) GetEventTypeCounts() ([]EventTypeCount, error) { + rows, err := s.db.Query(`SELECT event_type, COUNT(*) FROM events GROUP BY event_type ORDER BY event_type`) + if err != nil { + return nil, fmt.Errorf("query event counts: %w", err) + } + defer rows.Close() + + var counts []EventTypeCount + for rows.Next() { + var c EventTypeCount + var et string + if err := rows.Scan(&et, &c.Count); err != nil { + return nil, fmt.Errorf("scan event count: %w", err) + } + c.EventType = EventType(et) + counts = append(counts, c) + } + return counts, nil +} + +// GetOldestPendingReadyTime returns the ready time of the oldest pending job. +// The ready time is the earlier of run_at and created_at, matching the lease +// ordering. ok is false when no job is pending. +func (s *SQLiteStore) GetOldestPendingReadyTime() (ready time.Time, ok bool, err error) { + var raw string + err = s.db.QueryRow( + `SELECT COALESCE(run_at, created_at) FROM jobs + WHERE state = 'pending' ORDER BY COALESCE(run_at, created_at) ASC LIMIT 1`).Scan(&raw) + if err == sql.ErrNoRows { + return time.Time{}, false, nil + } + if err != nil { + return time.Time{}, false, fmt.Errorf("query oldest pending: %w", err) + } + t, err := time.Parse(sqliteTimeFormat, raw) + if err != nil { + return time.Time{}, false, fmt.Errorf("parse oldest pending: %w", err) + } + return t, true, nil +} + func (s *SQLiteStore) AppendEvent(e Event) error { md := sql.NullString{} if e.Metadata != nil { diff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go index 5338aa1..7edcf1f 100644 --- a/internal/worker/worker_test.go +++ b/internal/worker/worker_test.go @@ -122,6 +122,100 @@ func TestWorkerRecoversOrphans(t *testing.T) { } } +// TestWorkerDeadLetterAndRequeue verifies the end-to-end dead-letter workflow. +// A job that always fails exhausts its attempts and enters the dead-letter +// state. After a requeue with a working handler, the same job completes. +func TestWorkerDeadLetterAndRequeue(t *testing.T) { + q, _ := newTestQueue(t) + + failing := func(ctx context.Context, job queue.Job) error { + return context.DeadlineExceeded + } + w := NewWorker(q, failing, "test", + WithPollInterval(10*time.Millisecond), + WithLeaseDuration(time.Minute), + ) + + ctx, cancel := context.WithCancel(context.Background()) + wDone := make(chan struct{}) + go func() { + _ = w.Run(ctx) + close(wDone) + }() + + q.Enqueue("test", `{}`, queue.WithMaxAttempts(3)) + waitFor(t, func() bool { + snap, _ := q.Inspect() + return snap.Stats[queue.StateDeadLetter] == 1 + }, 2*time.Second) + cancel() + <-wDone + + snap, err := q.Inspect() + if err != nil { + t.Fatalf("inspect: %v", err) + } + var job queue.Job + for _, j := range snap.Jobs { + if j.State == queue.StateDeadLetter { + job = j + break + } + } + if job.State != queue.StateDeadLetter { + t.Fatalf("expected dead_letter after exhausting attempts, got %s", job.State) + } + + // Requeue the dead letter with a handler that succeeds, then confirm the + // job completes. + requeued, err := q.Requeue(job.ID) + if err != nil { + t.Fatalf("requeue: %v", err) + } + if requeued.State != queue.StatePending { + t.Fatalf("expected pending after requeue, got %s", requeued.State) + } + + var processed atomic.Int32 + w2 := NewWorker(q, func(ctx context.Context, job queue.Job) error { + processed.Add(1) + return nil + }, "test", + WithPollInterval(10*time.Millisecond), + WithLeaseDuration(time.Minute), + ) + ctx2, cancel2 := context.WithCancel(context.Background()) + w2Done := make(chan struct{}) + go func() { + _ = w2.Run(ctx2) + close(w2Done) + }() + waitFor(t, func() bool { + snap, _ := q.Inspect() + return snap.Stats[queue.StateCompleted] == 1 + }, 2*time.Second) + cancel2() + <-w2Done + + if n := processed.Load(); n != 1 { + t.Errorf("expected 1 processed after requeue, got %d", n) + } +} + +// waitFor polls cond until it returns true or the deadline passes. It reports +// a test failure when the deadline passes first. +func waitFor(t *testing.T, cond func() bool, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("condition not met within %s", timeout) +} + // TestRecoveryEventSequence simulates a worker that leases a job and then // crashes without acknowledging it. A second worker run must recover the // orphaned lease and then process the job. The append-only event log for that diff --git a/main.go b/main.go index df2fdde..9119ad9 100644 --- a/main.go +++ b/main.go @@ -20,7 +20,9 @@ Commands: work Start a worker process inspect View queue state and event log history View the event log for one job + requeue Return a dead-lettered job to the queue seed Load bundled sample data + metrics Expose queue state for Prometheus demo Run a self-contained scenario with fault injection Use -help for command flags.`) @@ -40,8 +42,12 @@ Use -help for command flags.`) err = cli.Inspect(args) case "history": err = cli.History(args) + case "requeue": + err = cli.Requeue(args) case "seed": err = cli.Seed(args) + case "metrics": + err = cli.Metrics(args) case "demo": err = cli.Demo(args) default: