Extend Local-first Durable Job Queue

.github/workflows/ci.yml

@@ -38,5 +38,7 @@ jobs:
         run: go tool cover -func=coverage.out | tail -n 1
       - name: Build
         run: go build ./...
+      - name: Demo smoke test
+        run: go run . demo -run 1s
       - name: Benchmark queue paths
-        run: go test ./internal/queue -run '^$' -bench Benchmark -benchmem -count=1
+        run: go test ./internal/queue -run '^$' -bench Benchmark -benchmem -count=1

CHANGELOG.md

@@ -0,0 +1,42 @@
+# Changelog
+
+All notable changes to this project are listed here.
+
+The format follows Keep a Changelog.
+This project uses calendar versioning by release.
+
+## Unreleased
+
+### Added
+
+- Retention with the `purge` command. It removes finished jobs and their events in one transaction.
+- The `purge -dry-run` flag previews a removal without changing the store.
+- The `purge -state` and `purge -before` flags name states and set an age cutoff.
+- The demo shows a retention segment with a dry run and a real purge.
+- Deterministic tests for purge selection, age bounds, and event removal.
+
+## 2026-08-03
+
+### Added
+
+- Prometheus metrics for queue inspection.
+- The `metrics` command serves the exposition format over HTTP.
+- The `work` command can serve the same endpoint beside a worker.
+- Priority aging prevents starvation of low-priority work.
+- The demo shows a low-priority job overtaking a fresher high-priority job.
+
+## 2026-08-01
+
+### Added
+
+- Dead-letter queue with the `requeue` command.
+- Durable priority dispatch with deterministic ordering.
+- Scheduled jobs with nanosecond-safe release times.
+
+## 2026-07-28
+
+### Added
+
+- Durable leases, retries, idempotency, crash recovery, and event history.
+- The `enqueue`, `work`, `inspect`, `history`, `seed`, and `demo` commands.
+- A deterministic fault-injection harness for worker handlers.

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, priority aging, a dead-letter queue, Prometheus metrics, and an append-only event log.
+The project shows leases, retries, idempotency, crash recovery, priority dispatch, priority aging, a dead-letter queue, retention, Prometheus metrics, and an append-only event log.
 
 ## Value
 
@@ -12,6 +12,8 @@ Every lease, retry, recovery, and acknowledgement remains visible in SQLite.
 
 The demo injects repeatable faults, so failure paths are easy to inspect.
 
+The purge command enforces retention, so finished jobs cannot grow without limit.
+
 ## Architecture
 
 The queue separates durable state from worker execution.
@@ -74,6 +76,17 @@ go run . history <id> -db queue.db
 go run . requeue <id> -db queue.db
 ```
 
+Enforce retention with these commands.
+
+```text
+go run . purge -db queue.db -dry-run
+go run . purge -db queue.db -before 720h
+```
+
+The first command previews the removal.
+
+The second command removes jobs not updated in 30 days.
+
 ## Commands
 
 ### `enqueue`
@@ -140,6 +153,30 @@ jobqueue seed [-db <path>]
 
 The command loads three idempotent jobs for each bundled workload.
 
+### `purge`
+
+```text
+jobqueue purge [-state <state>]... [-before <duration>] [-dry-run] [-db <path>]
+```
+
+The command removes finished jobs and their events.
+
+It targets the terminal states by default.
+
+Those states are `completed`, `failed`, and `dead_letter`.
+
+Pending and leased jobs survive a default purge.
+
+Use `-state` to name different states.
+
+Use `-state pending` to clear a backlog on purpose.
+
+Use `-before` to keep recent history.
+
+The command measures age from the job's last update.
+
+Use `-dry-run` to preview the removal without changing the store.
+
 ### `metrics`
 
 ```text
@@ -236,6 +273,26 @@ Every state change appends one event row.
 
 The `history` command shows one job's complete timeline.
 
+### Retention
+
+The `purge` command removes finished jobs and their events in one transaction.
+
+Each removed job takes its event rows with it.
+
+The append-only log stays consistent with the remaining jobs.
+
+The default target set is the terminal states.
+
+A purge never touches pending or leased work unless you name those states.
+
+An operator can clear a stuck backlog with `-state pending`.
+
+Use `-before` to keep jobs updated within a chosen window.
+
+A dry run reports the exact counts before any change.
+
+The removal is a single SQLite transaction, so it is atomic.
+
 ### Metrics
 
 The exporter renders queue state in the Prometheus text format.
@@ -305,6 +362,13 @@ aging interval: 100ms; a job gains one priority point per interval it waits.
 lease order: <id> (aged) then <id> (fresh)
 the waiting job outranks the fresher higher-priority job.
 
+Retention
+---------
+jobs: pending=1 completed=2 dead_letter=1
+  dry run: would remove 3 jobs and 9 events
+  removed 3 jobs and 9 events
+after purge: pending=1 completed=0 dead_letter=0
+
 Queue state
 -----------
   completed: 6
@@ -370,7 +434,7 @@ SQLite serializes writes through one store connection.
 
 A sustained backlog can exceed the writer's capacity.
 
-Jobs and events remain until an operator removes them.
+Jobs and events accumulate until an operator runs the purge command.
 
 A high-priority stream can delay lower-priority jobs until aging lifts them.
 
@@ -386,12 +450,27 @@ The project does not provide a web interface.
 - [x] Priority aging to prevent starvation.
 - [x] Dead-letter queue with requeue of permanently failed jobs.
 - [x] Prometheus metrics for queue inspection.
+- [x] Retention with a purge command for finished jobs and their events.
 - [ ] Web UI for queue inspection.
 - [ ] Horizontal scaling with a shared SQLite file.
 
 ### Release notes
 
-This release adds Prometheus metrics.
+This release adds retention with the `purge` command.
+
+The command removes finished jobs and their events in one transaction.
+
+It targets the terminal states by default.
+
+Use `-state` to name different states.
+
+Use `-before` to keep recent history.
+
+Use `-dry-run` to preview a removal without changing the store.
+
+The demo now shows a retention segment with a dry run and a real purge.
+
+The previous release added Prometheus metrics.
 
 The new `metrics` command serves the exposition format over HTTP.
 

internal/cli/demo.go

@@ -190,6 +190,9 @@ func Demo(args []string) error {
 	if err := showPriorityAging(*kind); err != nil {
 		return err
 	}
+	if err := showRetention(*kind); err != nil {
+		return err
+	}
 
 	fmt.Println()
 	snap, err = q.Inspect()
@@ -208,6 +211,83 @@ func Demo(args []string) error {
 	fmt.Printf("\ninspect again with: jobqueue inspect -db %q\n", path)
 	fmt.Printf("inspect a job with: jobqueue history <id> -db %q\n", path)
 	fmt.Printf("requeue a dead letter with: jobqueue requeue <id> -db %q\n", path)
+	fmt.Printf("purge finished jobs with: jobqueue purge -db %q\n", path)
+	return nil
+}
+
+// showRetention demonstrates the purge workflow: preview the removal with a
+// dry run, apply it, and confirm the queue state. The segment uses its own
+// in-memory store so the scenario counts stay intact.
+func showRetention(kind string) error {
+	store, err := queue.NewSQLiteStore("file:jobqueue-demo-retention?mode=memory&cache=shared")
+	if err != nil {
+		return fmt.Errorf("open retention store: %w", err)
+	}
+	defer store.Close()
+	q := queue.NewQueue(store)
+	ctx := context.Background()
+
+	// complete enqueues one job and finishes it in one step. Each lease is
+	// unambiguous because the future job is not ready yet.
+	complete := func() error {
+		if _, err := q.Enqueue(kind, `{"name":"completed"}`); err != nil {
+			return fmt.Errorf("enqueue completed: %w", err)
+		}
+		job, err := q.Lease(ctx, kind, time.Minute)
+		if err != nil || job == nil {
+			return fmt.Errorf("lease completed: job=%v err=%v", job, err)
+		}
+		if err := q.Acknowledge(job.ID); err != nil {
+			return fmt.Errorf("ack completed: %w", err)
+		}
+		return nil
+	}
+	for i := 0; i < 2; i++ {
+		if err := complete(); err != nil {
+			return err
+		}
+	}
+
+	if _, err := q.Enqueue(kind, `{"name":"doomed"}`, queue.WithMaxAttempts(1)); err != nil {
+		return fmt.Errorf("enqueue doomed: %w", err)
+	}
+	doomed, err := q.Lease(ctx, kind, time.Minute)
+	if err != nil || doomed == nil {
+		return fmt.Errorf("lease doomed: %w", err)
+	}
+	if err := q.Fail(doomed.ID, "boom"); err != nil {
+		return fmt.Errorf("fail doomed: %w", err)
+	}
+	if _, err := q.Enqueue(kind, `{"name":"future"}`, queue.WithRunAt(time.Now().Add(time.Hour))); err != nil {
+		return fmt.Errorf("enqueue future: %w", err)
+	}
+
+	snap, err := q.Inspect()
+	if err != nil {
+		return fmt.Errorf("inspect retention: %w", err)
+	}
+	fmt.Println("\nRetention")
+	fmt.Println("---------")
+	fmt.Printf("jobs: pending=%d completed=%d dead_letter=%d\n",
+		snap.Stats[queue.StatePending], snap.Stats[queue.StateCompleted], snap.Stats[queue.StateDeadLetter])
+
+	var b strings.Builder
+	if err := runPurge(&b, q, nil, nil, true); err != nil {
+		return fmt.Errorf("retention dry run: %w", err)
+	}
+	fmt.Print("  " + b.String())
+	b.Reset()
+	if err := runPurge(&b, q, nil, nil, false); err != nil {
+		return fmt.Errorf("retention purge: %w", err)
+	}
+	fmt.Print("  " + b.String())
+
+	snap, err = q.Inspect()
+	if err != nil {
+		return fmt.Errorf("inspect after purge: %w", err)
+	}
+	fmt.Printf("after purge: pending=%d completed=%d dead_letter=%d\n",
+		snap.Stats[queue.StatePending], snap.Stats[queue.StateCompleted], snap.Stats[queue.StateDeadLetter])
 	return nil
 }