From e5a533daeae2bc23144d623ba6475bcf7886978c Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Wed, 7 Oct 2026 11:53:57 +0000 Subject: [PATCH] :books: Document the wake-up contract of a worker queue An out-of-JVM process could not learn that work exists without reading the dispatcher: the key shape lives in `app.worker/queue-key`, but the payload format was only in two private places (`encode-payload` in the dispatcher and the runner's local decoder), and nothing said what a pop means or what a consumer owes. The backend memory now states the contract as a public surface: the key with its tenant prefix, the two-item payload and its format, mark-scheduled-then-push and why, the scheduled-at handed back as-is on the claim (what makes the queue a notification and not the truth), the management API as the only authority, no owner of the queue and the runner rule that follows, and the beats as the cancel channel (linked to the heartbeat note, not duplicated). The ephemeral request/reply lane stays out on purpose. The exporter memory points at it as the first external consumer. AI-assisted-by: glm-5.3-flash --- .serena/memories/backend/core.md | 42 +++++++++++++++++++++++++++++++ .serena/memories/exporter/core.md | 2 ++ 2 files changed, 44 insertions(+) diff --git a/.serena/memories/backend/core.md b/.serena/memories/backend/core.md index 1ba0d477fc..133d9ca982 100644 --- a/.serena/memories/backend/core.md +++ b/.serena/memories/backend/core.md @@ -73,6 +73,48 @@ Every background job is a job-def: a plain `(defn execute-X [cfg params] ...)` i - `heartbeat` uses named options `[cfg & {:keys [job-id progress] :as options}]` and returns the number of durable writes, so 0 means there was nothing to write: no job context, or the throttle did not allow it. A write that finds the job no longer active (terminal or gone) raises an `:interrupt` (`code :job-interrupted`, carrying the job id and the status it found) through the shared `jobs/check-active` instead of counting zero, so a task in course stops at its next beat; the interrupt is internal and never reaches HTTP (the management progress report answers `:skip` for it). The job id falls back to `::jobs/job-id` on the cfg and then to the runner-bound `*job-id*`. The `modified_at` touch is throttled to ~1s and the progress event to ~200ms, so the check fires at most once per second with no extra query (the read of the row happens only when a write already found nothing). A progress report is a milestone (`schema:progress`): a required `:stage` keyword plus an optional `:counters` map of `scope -> {:current, :total?}`, and no other key. Counters are non-negative, a `total` is positive and never lower than its `current`, and stage and counter keys are names of at most 64 characters. A milestone without counters is valid (a point of the run with no units to count), the stage vocabulary belongs to the worker and not to the substrate, and a client must degrade for a stage it does not know. Read a stored payload back with `jobs/decode-progress`: the database holds the keywords as strings. The 1s beat (down from 60s, so a cancelled heavy job is seen as gone at its next beat) costs one `modified_at` touch per beating job per second: cheap while the only heavy beater runs alone on its queue, revisit if concurrent beating jobs ever grow. - `job.error` conforms to `app.jobs/schema:job-error` (`type` and `code` keywords, `hint` text, extra details allowed). Read it back with `jobs/decode-job-error`; a plain `db/decode-json-pgobject` leaves the keywords as strings. `complete` takes named options and an optional `resource-id`, which is only ever set, never replaced: it reads the row locked inside its own transaction and applies the rules in a fixed order, skipping a job that is no longer completable before refusing one for its resource, so the answer is a function of the row and not of the order two callers arrive in. +## The wake-up contract of a worker queue (public surface) + +What an out-of-JVM process needs to learn about work WITHOUT reading the +dispatcher's guts. Verified against `app.worker/queue-key`, +`encode-payload` (`app.worker.dispatcher`) and the runner's local +`decode-payload`; a change there is a change of this contract. + +- The key: `penpot.worker.queue::` (`app.worker/queue-key`). + `job.queue` stores the bare name, so the prefix is composed at use. The + tenant prefix keeps instances sharing one Redis from handing each other + jobs. +- The payload: a JSON array of exactly two items, `["", + ""]`, the instant formatted by `ct/format-inst`. Fewer + than two is not a payload; the runner's decoder reads exactly those + two and drops and logs anything it cannot parse as uuid + instant. +- Dispatcher order: the rows are marked `scheduled` FIRST, pushed LAST + (RPUSH). A crash between the two leaves `scheduled` rows that + `reschedule-lost-jobs` re-queues; the reverse order would leave a + duplicate payload nothing deduplicates. +- The `scheduled-at` in the payload must be handed back AS-IS on `claim-job`: + the claim demands an exact match and answers `{:action :skip :status ...}` + for a stale payload. That is what makes a claim idempotent and what turns + the queue into a mere notification: a pop that is not confirmed by a claim + is not work. +- The queue is not the truth: the management API is (`claim-job`, + `report-job-progress`, `complete-job`, `fail-job` in + `app/rpc/management/jobs.clj`), shared-key authenticated by the route + resolver. Everything a consumer decides about a job goes through it. +- The queue has NO owner: anyone with Redis access can BLPOP it. Operational + rule that follows: do NOT wire a JVM runner for a queue an external + process consumes — the dispatcher hands each payload to whoever pops, so + both would start taking the same jobs. +- Cancellation travels the beats: `report-job-progress` answers + `{:action :skip}` once the row stopped being active — the only signal that + distinguishes "cancelled" from "network down". Same mechanism as the + internal `heartbeat` `:interrupt` (see the heartbeat note above); do not + build a second channel. +- Not part of this contract: `app.jobs/request`/`reply` (the ephemeral lane; + in retirement, `P1/F5/T16`). A comment in `app/jobs.clj` (2026-09-17) saying + that lane "waits for a producer" predates that decision and must not guide + anything here. + For worker dispatch, cron, retry semantics (`ex/raise :type ::wrk/retry` with `:delay`/`:strategy`), deduplication, job events, and queue internals: `mem:backend/subtleties`. ## Jobs metrics diff --git a/.serena/memories/exporter/core.md b/.serena/memories/exporter/core.md index 16df0e758e..a0be587c19 100644 --- a/.serena/memories/exporter/core.md +++ b/.serena/memories/exporter/core.md @@ -17,6 +17,8 @@ on this side: the row of the job, its cancel and its result are the backend's. ## The worker anatomy +- The queue wake-up contract (key shape, the two-item payload, mark-then-push, the claim-back rule) is the backend's public surface now: `mem:backend/core`, "The wake-up contract of a worker queue". This process was its first external consumer; read it before writing another. + - Main thread boots pools + K pollers (`core.cljs`): browser pool, wasm pool, temp-file cleaner (`jobs.utils/init`), then `consumer.worker/start!`. Shutdown: browser → wasm pool → pollers → done (the pollers own their connections; the pools unwind while an export is still in flight). - K pollers (`app.consumer.worker`): one Redis connection per poller (connection = slot + backpressure), BLPOP on `penpot.worker.queue::exporter`, payload `[job-id scheduled-at]` JSON; corrupt payload warns and drops, dead Redis delays 1s and retries. `K = PENPOT_EXPORTER_WORKER_CONCURRENCY` default 2, floor 1; replicas × K is the total concurrency. Pools size themselves to K (`browser` and `wasm.pool` read `ccfg/concurrency`); there is no independent pool-max config anymore, only `:wasm-worker-pool-min`. - Management client (`app.consumer.api`): claim/progress/create-job-session/complete-job (JSON and multipart via undici FormData)/fail-job, POST to `/api/management/methods/` with `X-Shared-Key: exporter ` (key = `PENPOT_EXPORTER_SHARED_KEY` or HKDF-derived from the secret); transit bodies, non-2xx errors carry `:status` and the error body.