perf(jobworker): cache claim prepared statement and add claim index

.github/workflows/go.yml

@@ -16,7 +16,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v6
       with:
-        go-version: 1.25.5
+        go-version: 1.26.0
 
     - name: Build
       run: go build -v ./...

CHANGELOG.md

@@ -5,6 +5,105 @@ All notable changes to this project are documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [v0.7.0] - 2026-06-18
+
+Faster job claiming via a cached prepared statement, a dedicated claim index,
+clock-skew-immune claim timestamps, and dependency updates.
+
+### Added
+
+- Partial index **`worker_job_claim_idx`** on
+  `worker.job ("type", priority desc, created_at asc) where started_at is null`,
+  backing the `StartNextJobOrNil` claim query — the hottest query in the queue.
+  The partial predicate confines the index to the pending backlog (so it stays
+  small as finished jobs accumulate), the leading `"type"` prunes to the
+  registered types in the `in (...)` list, and the `priority desc, created_at asc`
+  ordering matches the claim's `ORDER BY` so the scan yields rows in claim order:
+  for a single registered type `limit 1` stops at the first unlocked row with no
+  sort step, and for several types Postgres merges the per-type index streams
+  instead of sorting the whole backlog.
+
+### Changed
+
+- **BREAKING (API):** `jobworker.RegisteredJobTypes()` now returns
+  `(jobTypes []string, generation uint64)` instead of `notnull.StringArray`. The
+  returned slice is a **cached, sorted, shared** value and **must not be mutated**
+  by the caller. The `generation` counter is bumped on every `Register`/
+  `Unregister` call (normally only at startup), so a consumer can cache a value
+  derived from the types and rebuild only when the generation differs — no
+  element-by-element comparison of the type sets.
+- `jobworker.Register`, `RegisterFunc`, `RegisterFuncForJobType`, and `Unregister`
+  now **panic if called while worker threads are running** (after `StartThreads`).
+  Registration is startup-only: it mutates the registered-type set the cached claim
+  statement is keyed on, so changing it mid-run would race that cache. Code that
+  registers all workers before `StartThreads` is unaffected.
+- `jobworkerdb` now claims and starts the next job in a **single CTE statement**
+  (`with claimed as (select … for update skip locked) update worker.job … from
+  claimed … returning worker.job.*`) with **no surrounding transaction**, run as a
+  **cached prepared statement**. The registered job types are inlined as SQL string
+  literals (`"type" in ('a','b')`, escaped via the connection's
+  `FormatStringLiteral`) instead of an `= any($n)` array parameter, so PostgreSQL
+  plans against the actual values. The statement is prepared once per process and
+  re-prepared only when workers are registered or unregistered (normally
+  startup-only). The per-heartbeat
+  `SetJobWorkerAlive` update is likewise a cached prepared statement, and both are
+  released on `Close`.
+- Job-claim timestamps now use the **database clock** (`now()`): `started_at`,
+  `updated_at`, `worker_alive_at`, and the `start_at <= now()` comparison are all
+  evaluated server-side within the one claim statement, making job claiming immune
+  to clock skew between worker processes and the database (consistent with the
+  crash-recovery reaper, which already compares against `now()`). Previously
+  `started_at` and `updated_at` were stamped from the worker process clock.
+
+### Dependencies
+
+- `github.com/domonda/go-sqldb` and `github.com/domonda/go-sqldb/pqconn`
+  v1.0.2 → v1.3.0 (provides the prepared-statement helpers `QueryRowAsStmt` /
+  `ExecStmt` and `FormatStringLiteral` this release builds on).
+- `github.com/domonda/go-errs` v1.0.1 → v1.0.3, `github.com/domonda/golog`
+  v1.0.5 → v1.1.1, and `github.com/domonda/go-types` updated, plus indirect
+  dependency updates.
+- Go toolchain **1.25.5 → 1.26.0** (the `go` directive in `go.mod` and
+  `tools/go.mod`, and the CI `setup-go` version), so building this module now
+  requires Go 1.26.0 or newer.
+
+### For contributors
+
+- The registered-job-types set is cached in `jobworker` (`workerTypes` +
+  `workerTypesGeneration`, guarded by `workersMtx`) and invalidated on
+  `Register`/`Unregister`, so the common claim path takes a single read lock with
+  no allocation. `SetDataBase` and its backing `db` variable moved from
+  `jobworker/database.go` to `jobworker/config.go` (no API change).
+- `jobworkerdb` context keys switched from address-of-`int` sentinels to unexported
+  empty-struct key types, and `reflect.Ptr` usages were updated to
+  `reflect.Pointer`.
+- Static analysis (`go vet`, `revive`, `gosec`) now runs before the database setup
+  in `scripts/run-tests.sh` (skip with `-s`). `revive` and `gosec` are pinned in a
+  separate `tools/go.mod` module (kept out of importers' dependency graph) and
+  invoked via `go tool -modfile=tools/go.mod …`; `revive.toml` is report-only
+  (`warningCode=0`/`errorCode=0`). The test run now covers `./...`, not just `tests/`.
+
+### Migration
+
+Upgrading an existing database from `v0.6.0` requires adding the new claim index.
+No column or data changes are involved, so the code runs correctly without it —
+only the `StartNextJobOrNil` claim query is slower until the index exists. Apply
+it out-of-band:
+
+```sql
+-- Migration: v0.6.0 -> v0.7.0 (dedicated StartNextJobOrNil claim index)
+
+-- Partial index backing the StartNextJobOrNil claim query (the hottest query in
+-- the queue). The predicate confines it to the pending backlog, the leading
+-- "type" prunes to the registered types, and priority desc, created_at asc
+-- matches the claim's ORDER BY.
+-- Use CREATE INDEX CONCURRENTLY (outside a transaction block) so building it on a
+-- large, live table does not lock out job claims and inserts.
+create index concurrently if not exists worker_job_claim_idx
+    on worker.job ("type", priority desc, created_at asc)
+    where started_at is null;
+```
+
 ## [v0.6.0] - 2026-06-17
 
 Worker liveness heartbeat and safe, multi-process crash recovery.
@@ -128,5 +227,6 @@ commit;
 Last release before the worker liveness heartbeat work. See the git history for
 details of `v0.5.4` and earlier releases.
 
+[v0.7.0]: https://github.com/domonda/go-jobqueue/compare/v0.6.0...v0.7.0
 [v0.6.0]: https://github.com/domonda/go-jobqueue/compare/v0.5.4...v0.6.0
 [v0.5.4]: https://github.com/domonda/go-jobqueue/releases/tag/v0.5.4

CLAUDE.md

@@ -8,28 +8,33 @@ PostgreSQL-backed job queue for Go with support for job bundles, retries, worker
 
 ## Common Commands
 
-- **Run tests**: `./scripts/run-tests.sh` (creates temp DB, applies schema, runs tests, drops DB)
+- **Run tests**: `./scripts/run-tests.sh` (runs static analysis — `go vet`, `revive`, `gosec` — then creates a temp DB, applies the schema, runs `go test ./...`, drops the DB)
+- **Run tests, skip static analysis**: `./scripts/run-tests.sh -s`
 - **Run specific tests**: `./scripts/run-tests.sh -- -run TestReset`
 - **Verbose tests**: `./scripts/run-tests.sh -v`
 - **Build check**: `go build -v ./...`
 
+`revive` and `gosec` are pinned in a separate `tools/go.mod` module (kept out of importers' dependency graph) and invoked via `go tool -modfile=tools/go.mod …`; `revive.toml` is report-only.
+
 Tests require PostgreSQL. The script auto-starts one via `docker compose` if none is reachable at `POSTGRES_HOST:POSTGRES_PORT` (defaults from `.env.example`: `127.0.0.1:5432`, user `postgres`).
 
 ## Architecture
 
 Three packages with clear layering:
 
 - **`jobqueue`** (root): Core types (`Job`, `JobBundle`, `Status`), `Service` interface, and package-level functions that delegate to a context-or-default service. No database dependency.
-- **`jobworker`**: Worker registration (`Register`, `RegisterFunc`, `RegisterFuncForJobType`), job execution logic, thread pool management (`StartThreads`, `FinishThreads`), and retry scheduling.
+- **`jobworker`**: Worker registration (`Register`, `RegisterFunc`, `RegisterFuncForJobType`), job execution logic, thread pool management (`StartThreads`, `FinishThreads`), and retry scheduling. Registration (and `Unregister`) is startup-only: it mutates the registered-type set that `jobworkerdb` caches a prepared claim statement against, so it panics if called while worker threads are running.
 - **`jobworkerdb`**: PostgreSQL implementation of both `jobqueue.Service` and `jobworker.DataBase`. All SQL lives here. Handles LISTEN/NOTIFY for `job_available`, `job_stopped`, and `job_bundle_stopped` channels.
 
 ### Service wiring
 
 `jobworkerdb.InitJobQueue(ctx)` (or `InitJobQueueResetInterruptedJobs(ctx, deadFor)`, which additionally resets jobs abandoned by a worker that crashed at least `deadFor` ago) creates the DB-backed service, registers it as default for both `jobqueue` and `jobworker`, and sets up LISTEN/NOTIFY listeners.
 
+A process runs a single active job-queue service at a time. The claim and heartbeat prepared statements are cached in package-level state (not per service instance) and released by `Close`, so to swap the underlying database connection, `Close` the current service before calling `InitJobQueue` again; do not run two services against different connections concurrently, as the second init would reuse the first's cached statements.
+
 ### Worker liveness heartbeat
 
-While a worker processes a job, `jobworker` runs a goroutine (started in `doJobAndSaveResultInDB`) that updates `worker.job.worker_alive_at` every `jobworker.HeartbeatInterval` (default 10s) via `DataBase.SetJobWorkerAlive`. `worker_alive_at` is set on job claim (only when heartbeats are enabled), cleared on every stop/reset/retry transition, and stopped synchronously when the job ends. A job that is started but not stopped and whose `worker_alive_at` is stale was abandoned by a crashed worker — see `jobqueue.Job.WorkerAlive(deadFor)`. With heartbeats disabled (`HeartbeatInterval <= 0`) `worker_alive_at` stays NULL at claim, so the heartbeat-staleness reset branch is inert and no in-progress job is reclaimed (avoids resetting a still-running long job that has no liveness signal).
+While a worker processes a job, `jobworker` runs a goroutine (started in `doJobAndSaveResultInDB`) that updates `worker.job.worker_alive_at` every `jobworker.HeartbeatInterval` (default 10s) via `DataBase.SetJobWorkerAlive`. `worker_alive_at` is set on job claim (only when heartbeats are enabled), cleared on every stop/reset/retry transition, and stopped synchronously when the job ends. A job that is started but not stopped and whose `worker_alive_at` is stale was abandoned by a crashed worker — see `jobqueue.Job.WorkerAlive(deadFor)`. With heartbeats disabled (`HeartbeatInterval <= 0`) `worker_alive_at` stays NULL at claim, so the heartbeat-staleness reset branch is inert and no in-progress job is reclaimed (avoids resetting a still-running long job that has no liveness signal). `HeartbeatInterval` must be set during startup, before `StartThreads`, and treated as immutable afterward: whether a claim stamps `worker_alive_at` is baked into the cached claim statement the first time a job is claimed (it reads `HeartbeatInterval > 0` then), so toggling it across the `0` boundary at runtime would desync the claim path from the heartbeat/reaper logic.
 
 `InitJobQueueResetInterruptedJobs(ctx, deadFor)` is the multi-process-safe reaper: it only resets jobs whose worker is provably dead (`worker_alive_at` stale by ≥ `deadFor` for in-progress jobs, or `stopped_at` older than `deadFor` for jobs a *pre-heartbeat* worker left stuck between `SetJobError` and `ScheduleRetry` during a rolling upgrade — current-version `SetJobError` clamps `current_retry_count` to `max_retry_count`, making such failures terminal so the reaper never resurrects them). The `deadFor` cutoff is evaluated with the DB clock (`now() - make_interval`) so it is immune to clock skew between worker processes. A live worker keeps `worker_alive_at` fresh and finishes its error→retry transition well within `deadFor` (with a fast retry scheduler), so its jobs never fall inside the `deadFor` window and are never rug-pulled. `deadFor` must be comfortably larger than `HeartbeatInterval` (plus any retry-scheduler delay); `InitJobQueueResetInterruptedJobs` returns an error if `deadFor < 3 × HeartbeatInterval` (when heartbeats are enabled) or `deadFor <= 0`.
 

README.md

@@ -88,6 +88,11 @@ jobworker.Register("custom-job", func(ctx context.Context, job *jobqueue.Job) (r
 })
 ```
 
+> **Register all workers during startup, before `StartThreads`.** `Register`,
+> `RegisterFunc`, `RegisterFuncForJobType`, and `Unregister` mutate the set of job
+> types that `jobworkerdb` caches a prepared claim statement against, so they
+> **panic if called while worker threads are running**.
+
 ### 4. Add Jobs to the Queue
 
 ```go
@@ -242,12 +247,16 @@ if err != nil {
 
 Worker pools can run in multiple processes against the same database — this is how you scale
 horizontally. Each process calls `jobworker.StartThreads` independently and they all compete for
-the same queue; every job is claimed atomically (`SELECT ... FOR UPDATE SKIP LOCKED`), so two
-workers never pick up the same available job at once. (A failed job is still retried, and a job
+the same queue; every job is claimed atomically by a single statement that row-locks the next
+job with `FOR UPDATE SKIP LOCKED`, so two workers never pick up the same available job at once. (A failed job is still retried, and a job
 abandoned by a crashed worker is reclaimed, but never run concurrently.)
 
 While a worker processes a job it advances the job's `worker_alive_at` heartbeat every
 `jobworker.HeartbeatInterval` (default 10s). If that worker crashes, the heartbeat goes stale.
+Set `HeartbeatInterval` once during startup, before `StartThreads`, and treat it as immutable:
+whether a claim records `worker_alive_at` is baked into the cached claim statement the first time
+a job is claimed, so toggling it across the `0` (heartbeats-off) boundary at runtime would desync
+the claim path from the reaper.
 
 To reclaim jobs abandoned by a crashed worker, initialize with the reaper variant and a grace
 period:
@@ -303,6 +312,11 @@ set worker_alive_at=started_at, updated_at=now()
 where started_at is not null and stopped_at is null and worker_alive_at is null;
 ```
 
+This release also adds an optional performance index, `worker_job_claim_idx`, for the
+`StartNextJobOrNil` claim query. It is **not required for correctness** — the code runs without
+it, only the claim query is slower until it exists. See the [CHANGELOG](CHANGELOG.md) migration
+note for the `create index concurrently` statement.
+
 ### Queue Management
 
 ```go
@@ -410,15 +424,19 @@ All errors are wrapped using `github.com/domonda/go-errs` for enhanced stack tra
 Tests require a PostgreSQL instance. The test script `scripts/run-tests.sh` handles
 everything automatically:
 
-1. **Connects to PostgreSQL** — checks if an instance is already running at
+1. **Runs static analysis** — `go vet`, `revive`, and `gosec` run first, before any database
+   setup, so lint failures fail fast (skip with `-s`). `revive` and `gosec` are pinned in a
+   separate `tools/go.mod` module and invoked via `go tool -modfile=tools/go.mod …`; `revive` is
+   report-only (`revive.toml`).
+2. **Connects to PostgreSQL** — checks if an instance is already running at
    `POSTGRES_HOST:POSTGRES_PORT` (from `.env.example`, default `127.0.0.1:5432`)
    with user `POSTGRES_USER` (default `postgres`). The user must have `CREATE DATABASE` privileges.
-2. **Starts Docker Compose** — if no PostgreSQL is reachable, starts one via `docker compose up`.
-3. **Creates a temporary database** — named `test-jobqueue-XXXXXXXX` (random suffix to prevent
+3. **Starts Docker Compose** — if no PostgreSQL is reachable, starts one via `docker compose up`.
+4. **Creates a temporary database** — named `test-jobqueue-XXXXXXXX` (random suffix to prevent
    collisions between parallel runs).
-4. **Applies the schema** — runs `schema/worker.sql` against the temporary database.
-5. **Runs the Go tests** — executes `go test` against the `tests/` package.
-6. **Drops the temporary database** — always cleaned up, even on test failure.
+5. **Applies the schema** — runs `schema/worker.sql` against the temporary database.
+6. **Runs the Go tests** — executes `go test ./...` (all packages, not just `tests/`).
+7. **Drops the temporary database** — always cleaned up, even on test failure.
 
 ```bash
 # Run all tests
@@ -430,6 +448,9 @@ everything automatically:
 # Destroy Docker Compose after tests (used in CI)
 ./scripts/run-tests.sh -d
 
+# Skip static analysis (go vet, revive, gosec), run tests only
+./scripts/run-tests.sh -s
+
 # Run specific tests
 ./scripts/run-tests.sh -- -run TestReset
 ```

VERSION

@@ -1 +1 @@
-v0.6.0
+v0.7.0

go.mod

@@ -1,35 +1,35 @@
 module github.com/domonda/go-jobqueue
 
-go 1.25.5
+go 1.26.0
 
 require (
-	github.com/domonda/go-errs v1.0.1
-	github.com/domonda/go-sqldb v1.0.2
-	github.com/domonda/go-sqldb/pqconn v1.0.2
-	github.com/domonda/go-types v0.0.0-20260327082518-11ac2cfe4cdf
-	github.com/domonda/golog v1.0.5
+	github.com/domonda/go-errs v1.0.3
+	github.com/domonda/go-sqldb v1.3.0
+	github.com/domonda/go-sqldb/pqconn v1.3.0
+	github.com/domonda/go-types v0.0.0-20260618113239-de4558565e32
+	github.com/domonda/golog v1.1.1
 	github.com/stretchr/testify v1.11.1
 )
 
 require (
-	github.com/DataDog/go-sqllexer v0.2.1 // indirect
+	github.com/DataDog/go-sqllexer v0.2.2 // indirect
 	github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect
 	github.com/bahlo/generic-list-go v0.2.0 // indirect
-	github.com/buger/jsonparser v1.1.2 // indirect
+	github.com/buger/jsonparser v1.2.0 // indirect
 	github.com/corazawaf/libinjection-go v0.3.2 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/domonda/go-encjson v1.0.0 // indirect
 	github.com/domonda/go-pretty v1.0.0 // indirect
-	github.com/invopop/jsonschema v0.13.0 // indirect
-	github.com/lib/pq v1.12.1 // indirect
+	github.com/invopop/jsonschema v0.14.0 // indirect
+	github.com/lib/pq v1.12.3 // indirect
 	github.com/lucasb-eyer/go-colorful v1.4.0 // indirect
-	github.com/mailru/easyjson v0.9.2 // indirect
-	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mattn/go-isatty v0.0.22 // indirect
 	github.com/muesli/termenv v0.16.0 // indirect
+	github.com/pb33f/ordered-map/v2 v2.3.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
-	github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect
-	golang.org/x/sys v0.42.0 // indirect
-	golang.org/x/term v0.41.0 // indirect
+	go.yaml.in/yaml/v4 v4.0.0-rc.4 // indirect
+	golang.org/x/sys v0.46.0 // indirect
+	golang.org/x/term v0.44.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )