diff --git a/specs/constitution.md b/specs/constitution.md index bdcef6bc..653eb50a 100644 --- a/specs/constitution.md +++ b/specs/constitution.md @@ -740,6 +740,18 @@ passing, on the record — issue #80.) `repeat:`) and `removeJobScheduler` — or alert — past a threshold. **BullMQ will never do this for us.** A cron silently re-running a wedging job is exactly the runaway `CONST-BUDGET-BEFORE-TOKENS` exists to prevent, except unattended and overnight. + **A scheduled job that stalls AFTER its run was recorded is returned, never run.** A stall is not only a + dead worker. When Valkey is unreachable for longer than the lock renewal window while a run finishes, + the processor writes the run record, BullMQ refuses the completion ("Missing lock"), and the stall check + moves the scheduled job back to `wait`. Re-processing it is a second paid run of a job that already + ended, and its record overwrites the first. So the processor's first gate asks, for a job with + `stalledCounter > 0`, whether its record settles this attempt (`attempt` equal to `attemptsMade + 1`, an + outcome other than `failed`, started no earlier than the job less a 5 minute clock tolerance). If it + does, the job returns that recorded result without a container, a reservation or a new record, and + logs `job_lost_lock_after_completion`. No record (a worker that really died mid-run) keeps today's + behaviour, a record found and refused is logged as `job_lost_lock_record_rejected` before the job runs + again, and the per-scheduler stall guard still counts the stall either way, because the queue really + did lose the lock. **The rule binds the HOST side too, and one whole class of refusal was on the wrong side of it until issue #310.** A configuration fault is determinate by definition: the operator has to fix it, and the same job with the same inputs refuses identically for as long as they do not. Such errors are tagged @@ -1002,6 +1014,7 @@ passing, on the record — issue #80.) | Date | Change | |---|---| +| 2026-10-04 | Found this round (no issue): a scheduled job whose run finished while Valkey was unreachable longer than the lock renewal window was run again, paid. **`CONST-RETRY-INFRA-ONLY` AMENDED**, the Why only: a scheduled job that stalls after its completion was recorded is returned as its record says, without running (the processor's lost-lock gate: `stalledCounter > 0`, `attempt` equal to `attemptsMade + 1`, outcome not `failed`, started no earlier than the job less a 5 minute clock tolerance; a record found and refused is logged as `job_lost_lock_record_rejected`); a job with no such record runs as before, and the stall guard still counts the stall. The Statement and Acceptance are UNCHANGED, checked: a re-run of a completed job was a retry of a determinate outcome, which the Statement already forbids. **Code evidence**: worker/src/index.mjs -> makeProcessor (the first gate); worker/src/run-history.mjs -> makeSettledRecord. | | 2026-10-04 | Issue #504, part A (delegated allocation: the doctrine and the pure modules). **No article changed.** **`CONST-BUDGET-BEFORE-TOKENS` UNCHANGED, checked**: the ordering is untouched; a delegated allocation changes only the values the dollar reserve compares against (a project's cap becomes the smaller of its row and its allocation), never when the reserve runs. **`CONST-ISOLATION-CONTAINER-PER-JOB` UNCHANGED, checked**: the operator's session still processes no adversarial input by premise; a model there that reads issue text breaks that premise, and `DES-DELEGATED-ALLOCATION-INSIDE-ENVELOPE` names it as a residual of the session tool, whose judgement belongs in a portfolio job (issue #505). | | 2026-10-02 | Issue #501, parts 3 and 4, and #503 part 7 (dollar windows, reserved before the run and settled after it). **`CONST-BUDGET-BEFORE-TOKENS` AMENDED**: the Statement is the issue's: every spend cap is checked and reserved before a run, there are two ledgers (a job count and, when configured, dollars), a dollar reservation is the job's per-job cap held against every window that applies and replaced after the run by the metered cost or kept whole when the cost is not fully known, and the per-job cap is enforced in the container before each call; the constraint governs the ordering, the same for both ledgers. The Why gains why dollars reserve after both job-count reserves and before the container, why the amount is the per-job cap, why a refused dollar reservation is given back (the one departure from "a refused slot still counts"), and why a zero-rated local job reserves nothing yet keeps the ordering (a cap of 0 before each call). The Acceptance gains: given a dollar window is exhausted, a new trigger consumes zero provider tokens. **`CONST-RETRY-INFRA-ONLY` UNCHANGED, checked**: `dollar-cap` is a returned policy refusal; a Valkey fault in the dollar reserve is a never-started retry that refunds both ledgers. **Code evidence**: worker/src/dollar-budget.mjs; worker/src/processor.mjs -> runJob, isNeverStartedExit. The Statement says at least the reservation when the cost is not fully known, and that the per-job cap bounds what a job can spend before it runs, with the residuals named in `DES-DOLLAR-RESERVE-AND-SETTLE`, rather than that a job cannot spend more than it reserved. | | 2026-10-02 | Issues #501 and #502, the shared seams for policy stops. **No article changed.** **`CONST-BUDGET-BEFORE-TOKENS` UNCHANGED, checked**: the new refusal of a cost cap or a model list the runner cannot enforce before a call is free and runs inside the container after the meter installs and before the session exists, so nothing is sent to a provider, and no worker gate moved. **`CONST-RETRY-INFRA-ONLY` UNCHANGED, checked**: that refusal is a determinate exit `2`, not retried. | diff --git a/specs/design.md b/specs/design.md index 691fd7b3..4bf9cb7c 100644 --- a/specs/design.md +++ b/specs/design.md @@ -6391,12 +6391,38 @@ a tunnel. comment, bounded at one; a whole-deployment death between the terminal transition and the listener loses both channels (a single crashed worker does NOT -- the stall path fails its job at next pickup and the observing worker's listener fires); a graceful drain's exit can orphan an in-flight POST or - hook child. The DUPLICATION direction exists too, narrow and named (review finding): a worker that - comments a policy stop and then loses its job LOCK before BullMQ commits the return (crash, or a - partition outliving lock renewal) has its job stall-swept back to wait, and the re-pickup's - UnrecoverableError lands a second, `finishedOn`-guarded FAILED comment on the same issue -- a - pre-existing crash-consistency property of the queue that these comments make externally visible, - bounded at one duplicate, and not worth the idempotence store this entry already rejects. `OQ-023`'s + hook child. The DUPLICATION direction is now closed by the run record, which is the store that already + exists. A job that finishes and then loses its LOCK before BullMQ commits the return (a Valkey outage + longer than the lock renewal window, a partition, or a crash between the record and the commit) is taken + back by the stall check. Its completion is refused ("Missing lock"), and the next pickup fails it with + BullMQ's stall reason. That put the FAILED comment and the infra page on a job that had completed (or + on the same issue after a policy stop's own comment), and it ran a scheduled job again instead + (`CONST-RETRY-INFRA-ONLY`). So the failed listener, on a terminal failure whose message is exactly that + reason (`STALLED_FAILED_REASON` in `index.mjs`, pinned by a test against the installed bullmq's + `moveStalledJobsToWait`), first reads the job's record: this host's file, then the run mirror where one + is armed (`PI_WORKER_NAME`). When the record settles this attempt (same job id, `attempt` equal to + `attemptsMade`, which BullMQ has already raised by one, an outcome other than `failed`, and a `startedAt` + no earlier than the job's creation less `RECORD_CLOCK_SKEW_MS`, 5 minutes, so a reused id cannot match + an old record while a producer clock ahead of the worker's does not turn the check off), the terminal + line is `job_lost_lock_after_completion`, nothing is posted, and the hook fires only when the record is + a paid policy stop the completed listener would have paged for (that listener never saw the first + finish). A missing record, a mismatch or a lookup fault keeps the comment and the page, and a record + that was found and refused is logged as `job_lost_lock_record_rejected` with a fixed reason + (`unreadable`, `id-mismatch`, `other-attempt`, `failed`, `older-than-job`) and its source (`local` or + `mirror`; a mirror value that is there but is not a record is `unreadable` too), so an operator can see + why the comment was posted. The processor's gate says it ONCE per job id, stall count, attempt and + source (a bounded set of `REJECTED_SEEN_MAX`, 1000, oldest out first): it runs before every deferral + and BullMQ never resets a stall count, so a held job meets it on each pickup with the same verdict. + Residuals, each bounded at one comment + and one page unless said otherwise: on a fleet WITHOUT a shared `PI_LOGS_DIR`, a runtime-queue job + failed by a host that did not run it while the mirror holds no copy. The mirror write is not lost to + the outage itself (the shared client's offline queue sends it on reconnect, measured), so this is the + writing host exiting before Valkey returns (its offline queue is lost), or another host failing the job + before the writer has reconnected. A clock skew between producer and worker larger than the tolerance + (logged as `older-than-job`). A run that dies after its terminal comment but before its record. And a + suppressed job is NOT completed in the queue: it stays in BullMQ's failed set with the stall reason, so + the admin's views and `getJobCounts` show it failed while the worker log and the run record say it + completed; the record is the authority (`INT-RUN-HISTORY-FILE-CONTRACT`). `OQ-023`'s prepare-policy silence stands untouched -- #288 is post-spend only, and that row's own hazard (a `.pi/` cap breach commenting on every delivery) is exactly why. - **Traces to**: `REQ-JOB-STATUS-COMMENTS`, `REQ-OPERATOR-FAILURE-NOTIFICATION`, @@ -8169,3 +8195,4 @@ a tunnel. | 2026-10-04 | Issue #504, part B. **`DES-DELEGATED-ALLOCATION-INSIDE-ENVELOPE` AMENDED**: part B is named as shipped (`worker/src/allocation.mjs`); the compare-and-set is spelled out (`CAS_SCRIPT` decodes with `cjson`, a null id is the empty string, the neutral seed is `SEED_SCRIPT`, which on a fresh seed sets `alloc:envelope:expected` to the seed's digest unconditionally in the same step, so a stale one left by a turn-off never survives a new seed, and on replacing a state that does not decode sets it only when absent, so a stale host cannot take the fleet through an unreadable plan); the audit order is split, the system's own changes (seed, expiry, re-base) written by the CAS winner after it won, a writer's change (plan, revert) written before the CAS with an `apply-failed` row after a lost one; the ladder gains `envelope-mismatch` after the writer rung and names `plan-invalid` before it; the host role in a re-base: every host reconciles at boot, on every envelope reload and at each pickup, so setting `alloc:envelope:expected` heals a mismatched host at its next job; `expected` names the APPLIED split's envelope (set by the seed's winner, by a host that agrees, and by a differing host that finds it absent), PR #574's review having found that the lazy seed let the first differing host re-base the fleet; the first host to reconcile at a first start or after the keys were deleted defines the split, and doctor names every host that differs; a changed-externally row is remembered only after it was written; a neutral state re-bases to the new envelope's neutral split, a plan keeps its id, `lastPlanAt` and expiry; a host with no envelope in a governed fleet refuses its jobs (one `EXISTS` per pickup), so delegation is turned off for a fleet by removing the envelope everywhere and deleting `alloc:plan` and `alloc:envelope:expected`; a fault while the pickup reads the split is `InfraRetry`, never a dropped job; two audit residuals are named (a system change whose row the file refuses stands, logged as `allocation_audit_row_lost` and still pushed to `alloc:log`; a writer's CAS that throws leaves its applied row with no `apply-failed` row); a stored state whose amounts are not safe non-negative integers is corrupt, one a newer build wrote is neither trusted nor overwritten; enforcement names the synthetic ledger and `_other`'s own `otherDollars` slot (a deviation from the plan's "through the project slot": a project the envelope does not name keeps its own row and counts in `_other`), with `capSource`; the containment check is wired with every load and reload; a local job's folder is judged where it is mounted: named inside a job path by text or by an ancestor's identity, a symlink farm refused with the configuration to use instead, the envelope's folder refused, a resolved folder in another project refused, a chained child enqueued on the named folder. Three Rejected entries: `_other` through the project slot, refusing a triggers reload whose new job path holds the envelope, and holding a local folder to the run roots by a flag set at enqueue. **`DES-DOLLAR-RESERVE-AND-SETTLE` AMENDED**: item 9 names `_other`'s ledger and `allocation-cap`; a new item 10, the worker half under an envelope: `min` per window, never replace; the synthetic project, repo and `_other` ledgers join the job-cost prefixes and settle through `holdPart`; `capSource` per ledger and `dollarCapSource` for the deployment decide `allocation-cap` against `dollar-cap` (a tie is the operator's); a shrink refuses new starts only; `envelope-mismatch` is a free gate before the mint. **`DES-FLEET-LEASES-FOR-SHARED-BOUNDS` AMENDED**: `alloc:lock` borrows the `SET NX PX` idiom and `RELEASE_IF_MINE` and bounds nothing; the leases are UNCHANGED, checked. **`DES-TERMINAL-COMMENTS-AND-FAILURE-HOOK`** UNCHANGED, checked: the new refusals comment once, pre-spend, and stay out of the hook's policy set. **Code evidence**: `worker/src/allocation.mjs`, `worker/src/processor.mjs` (the envelope gate, the resolved folder's project, `capSourceOf`), `worker/src/index.mjs` (`governedDollars` and the `InfraRetry` at pickup), `worker/src/start.mjs` (`reloadEnvelope`, `afterCommit`, `fpEnvelope`, the reaper), `worker/src/prepare-local.mjs` (`judgePlacement`), `worker/src/outbox.mjs`; tests `allocation-apply`, `allocation.integration`, `index-allocation`, `processor-allocation`, `prepare-local`, `outbox`, `start-wiring`, `doctor`. | | 2026-10-04 | Issue #505, part A. **`DES-DELEGATED-ALLOCATION-INSIDE-ENVELOPE` AMENDED**: a portfolio job (`run.portfolio: true`, confirmed against the live triggers file at pickup) on a host whose envelope could not apply its plan (none, delegation off, or `portfolio-job` not a writer) is refused as `portfolio-no-envelope`, a free gate after `envelope-mismatch` and before the mint and the clone. Rejected: running it and recording the plan's refusal, which pays for a container whose one output is known to be refused. A job whose live entry lost the flag runs as an ordinary cron job. `DES-JOB-OUTBOX-CHAINING` and `DES-CRON-VIA-BULLMQ-SCHEDULER` UNCHANGED, checked (the snapshot and the plan channel are part B; a trigger fired by hand is one ordinary job under a `manual:` id). | | 2026-10-04 | Issue #505, part B. **`DES-JOB-OUTBOX-CHAINING` AMENDED**: a second kind of request, the priorities plan in `/outbox/priorities.json`, collected on the completed-only path after the chain requests and enqueueing nothing, with the facts coming in as `/job/portfolio.json` and its own baked persona `guardrails/PORTFOLIO_PROTOCOL.md`. Rejected: a `type` discriminator in `request-.json` (it would share the chain count cap, and an older worker would refuse it as `chain-bad-flow-name`), and Valkey, the envelope or an admin tool in the container. **`DES-DELEGATED-ALLOCATION-INSIDE-ENVELOPE` AMENDED**: the portfolio-job writer path (the snapshot, the collector, `recordRefusal` rows for the collector's own refusals, `plan-duplicate` on re-collection and `plan-too-soon` for a retried attempt, a race won by the first compare-and-set, a hand fire queued beside a tick and deferred behind it by the folder mutex, the rule that keeps a never-confirmed job's refusal out of the audit file and `alloc:log`, and `plan-collect-error` as an unknown outcome), and three rejected alternatives (the flag on job data alone, the live file alone at collection, an allocation module in the image). `REQ-AI-TRIGGERED-RUNS` UNCHANGED, checked: a plan enqueues nothing. | +| 2026-10-04 | Found this round (no issue): a job that completed while Valkey was unreachable was reported failed, or run again. **`DES-TERMINAL-COMMENTS-AND-FAILURE-HOOK` AMENDED**, Residuals: the duplication residual ("bounded at one duplicate, and not worth the idempotence store") is replaced by the record-read suppression. The run record is the existing store: on a terminal failure whose message is exactly BullMQ's stall reason (`STALLED_FAILED_REASON`, pinned against the installed bullmq), the failed listener reads the record (local file, then the run mirror on a declared fleet) and, when it settles this attempt (id, `attempt` equal to `attemptsMade`, outcome not `failed`, started no earlier than the job), logs `job_lost_lock_after_completion`, posts nothing, and pages only for a paid policy stop the completed listener would have paged for. `startedAt` may sit up to `RECORD_CLOCK_SKEW_MS` (5 minutes) before the job's creation, because the two instants come from two hosts' clocks; a record found and refused (a mirror value that is not a record included) is logged as `job_lost_lock_record_rejected` with a fixed reason and its source, by the processor's gate once per job id, stall count, attempt and source (a bounded set of 1000), since the gate runs before every deferral. The remaining residuals are named: a fleet without a shared logs dir whose mirror holds no copy (the writing host exits before Valkey returns, or another host fails the job before the writer reconnects; the offline queue otherwise delivers the mirror write), a skew beyond the tolerance, a run that dies between its comment and its record, and a suppressed job still sitting in BullMQ's failed set with the stall reason while the record says it completed. The Decision, Rejected list and Acceptance are UNCHANGED, checked. **Code evidence**: worker/src/start.mjs (the failed listener), worker/src/run-history.mjs (`makeReadRecord`, `recordSettlesAttempt`, `makeSettledRecord`), worker/src/run-mirror.mjs (`readMirroredRecord`), worker/src/index.mjs (`STALLED_FAILED_REASON`, the processor's lost-lock gate). | diff --git a/specs/interfaces.md b/specs/interfaces.md index 6ff2d8b5..2af22332 100644 --- a/specs/interfaces.md +++ b/specs/interfaces.md @@ -7364,3 +7364,4 @@ onFailureTimeoutMs; worker/test/on-failure.test.mjs; worker/test/start-wiring.te | 2026-10-04 | Issue #504, part B. **`INT-ENVELOPE-FILE-CONTRACT` AMENDED**: the worker reads the file at boot with `loadEnvelopeChecked` (the merged per-job cap, before any Valkey contact) and holds it in a watched ref with a last good copy; a reload is judged against the live projects and scoped limits, and a projects or scoped-limits reload that commits re-judges the envelope from disk, so an edit needing both files applies in either save order; the containment check runs with every load and every reload against the job paths read from the triggers FILE (`envelopeJobPaths`), a reload that fails it keeps the last good envelope, and a triggers reload that adds a job path around it logs `envelope_inside_job_path`; every host must agree, a host without the key in a fleet with an applied split refusing its jobs; a named residual: an edit to projects, scoped limits or the per-job cap can leave the live envelope invalid (last good kept, doctor fails, the next boot refuses; part C's writers cross-check); doctor loads it through the same function, `up` never writes the key and says so when its line is blank, `init` scaffolds no file. **`INT-RUN-HISTORY-FILE-CONTRACT` AMENDED**: five reasons, `allocation-cap` (the dollar refusal whose binding cap was the split or the envelope total; a tie is `dollar-cap`; its own comment, never "raise the budget"; it joins the `refunded` lists), `envelope-mismatch` (a free gate before the mint, a no-envelope host in a governed fleet included; a fault reading the split is `container-never-started` instead), `local-folder-escaped`, `local-folder-holds-envelope` and `local-folder-project-changed` (prepare-stage refusals of a local job); each pages nobody. **`INT-SCOPED-LIMITS-FILE-CONTRACT` AMENDED**: the dollar hold names `_other`'s ledger (log `ledger: "other"`) and the `allocation-cap` refusal with `source: "allocation"`. **`INT-HOST-REGISTRY-CONTRACT` AMENDED**: a host row gains `fpEnvelope`, the live envelope's digest or `none`, read by doctor only, never by the refusal, which compares the host's digest with `alloc:plan`'s; doctor reads the applied split whenever it may talk to the Valkey, prints its digest, names the hosts matching it (a warning when none does), fails for each host that does not (a host with no envelope included), and counts a key that does not decode as governed (`appliedSplitChecks`), else it warns on disagreeing peers and on a peer that publishes no `fpEnvelope` field (`fleetEnvelopeChecks`). **`INT-CONTAINER-JOB-INPUTS` AMENDED**: a local job's `/workspace` is its folder resolved at prepare (`realpathSync.native`), the path everything at prepare reads and the bind mount names; the raw string used to be mounted, so a link a job swapped below a job path after the folder was checked mounted its target read-write; a folder named inside a run root or a cron folder (by text or by an ancestor's identity) that resolves outside it, a symlink farm included, one that is the envelope's folder or above it, and one whose resolved folder is another project's member are refused before the reserve; a chained child is enqueued on the folder as its parent named it; the residual (a link above the resolved folder swapped before the mount) is named. **`INT-ON-FAILURE-HOOK-CONTRACT`** UNCHANGED, checked: `allocation-cap`, `envelope-mismatch`, `local-folder-escaped`, `local-folder-holds-envelope` and `local-folder-project-changed` are pre-spend refusals and stay out of `HOOK_POLICY_REASONS`. **`INT-PRIORITIES-PLAN-CONTRACT`** UNCHANGED, checked: the apply path reuses `parsePlan` and `planRefusal` as they are. **Code evidence**: `worker/src/allocation.mjs`, `worker/src/envelope.mjs` (`loadEnvelopeChecked`, `envelopeProtectedIdentities`), `worker/src/start.mjs` (`reloadEnvelope`, `fpEnvelope`), `worker/src/prepare-local.mjs` (`judgePlacement`), `worker/src/outbox.mjs`, `worker/src/doctor.mjs` (`appliedSplitChecks`, `fleetEnvelopeChecks`, `BOOT_FILES`). | | 2026-10-04 | Issue #505, part A. **`INT-TRIGGERS-FILE-CONTRACT` AMENDED**: a `run.portfolio` bullet and contract line (cron only, strictly boolean, refused at load on every webhook type with `false` included and when `true` sits beside `run.command`; carried at job level, never inside `trigger`; not inherited by an outbox child; re-read from the live file at pickup by `on.id`, a flag no longer there or an unreadable file making the job an ordinary cron job; reviewed-file-only: no tool parameter, `buildTriggerEntry` cannot produce it, the panel never asks, both edits keep it), and `pi-dispatch run --trigger ` (the same path rule as the one-shot checks, an unknown or webhook id refused, the scheduler data passed through whole, not retried, as a tick is not, id `manual::` because BullMQ takes a custom id with `:` only in three parts, a second call in the same minute queues nothing, queued on this host's cron queue, another host's folder refused). The cron byte-match acceptance admits `portfolio` on `command`'s terms. **`INT-RUN-HISTORY-FILE-CONTRACT` AMENDED**: the reason enum gains `portfolio-no-envelope`, a free pre-spend refusal after `envelope-mismatch` (no envelope, delegation off, or `portfolio-job` not among the writers), never retried, paging nobody. **`INT-CONTAINER-JOB-INPUTS` AMENDED**: a trigger fired by hand is a `cron` event with `scheduledFor` and `previousRunAt` both null. Review of PR #575: the live check holds the entry found by id to the job's `folder`, `flow`, `command` and `task`; `run --trigger` reads `PI_TRIGGERS_FILE`, `PI_WORKER_NAME` and `PI_MAX_COST_USD` (the loader's config inputs) by issue #471's service-key rule (a disagreement, or an unreadable `.env` the shell does not cover, refused), as `run ` now reads `PI_WORKER_NAME`, looks the cron trigger up before any webhook entry spelling its id, and refuses a folder that is not a git repository; a hand fire counts as a run of its trigger for the next tick's `previousRunAt`. `INT-OUTBOX-CONTRACT`, `INT-ENVELOPE-FILE-CONTRACT`, `INT-PRIORITIES-PLAN-CONTRACT` and `INT-ON-FAILURE-HOOK-CONTRACT` UNCHANGED, checked. | | 2026-10-04 | Issue #505, part B. **`INT-CONTAINER-JOB-INPUTS` AMENDED**: `/job/portfolio.json`, a confirmed portfolio job's snapshot, `0444` beside `event.json` on the existing `/job` mount, built at prepare only when the pickup decided portfolio AND the live triggers file still flags the entry; its shape, its content rule (ids, digests, integers, ISO instants, enum tokens, project ids and member labels, a label with a control, format or bidi character replaced by its ref, never issue text, a title, a plan reason or a path), money from the fleet-wide counters of the envelope's window (a member's key the one enforcement uses, `memberDollarKeyPrefix`, so a bare row's key; `spentMicros` is spent plus held, the counter the next job is admitted against; the issue's sketch also named a `reservedMicros`, which the counters cannot give), `runs7d` from the local history merged with the run mirror and `runsComplete` false without one, InfraRetry on a Valkey fault, and the portfolio persona it composes. **`INT-OUTBOX-CONTRACT` AMENDED**: a second file, `/outbox/priorities.json`, with its own ladder (no file, `plan-not-portfolio` from the pickup decision, prepare and the live file together, recorded only in the run record and the log line for a job the pickup never confirmed, `plan-oversize` and `plan-not-regular-file` by `lstat` and again on the `O_NOFOLLOW` and `O_NONBLOCK` descriptor's `fstat`, `plan-unreadable`, `plan-parse-error`, then `applyPlan` with the `portfolio-job` writer); every refusal recorded, completed-only, never throws (`plan-collect-error`, outcome unknown, the plan id carried), and the runner's log-only pre-check. **`INT-RUN-HISTORY-FILE-CONTRACT` AMENDED**: the tail field `plan` (`outcome`, `reason`, `planId`, `clamped`) after `project`, rebuilt and enum-checked, and the reason `portfolio-snapshot-oversize` (prepare-stage policy, before the reserve, `HOOK_POLICY_REASONS` UNCHANGED, checked: nothing is spent to page about). **`INT-PRIORITIES-PLAN-CONTRACT` AMENDED**, one sentence: the runner's pre-check passes the snapshot's `maxPlanDays` as the caller's. `INT-TRIGGERS-FILE-CONTRACT`, `INT-ENVELOPE-FILE-CONTRACT` and `INT-RUNNER-EXIT-CODE-PROTOCOL` UNCHANGED, checked (the pre-check changes neither the exit code nor the exit line). | +| 2026-10-04 | Found this round (no issue), the lost-lock fix. No contract changed. **`INT-RUN-HISTORY-FILE-CONTRACT` UNCHANGED, checked**: the worker now reads a job's own record back (`makeReadRecord`, by the writer's `.json` name; a record whose id differs from the job's is refused as `id-mismatch`) and, on a declared fleet, the mirror's copy by the same sanitized id; the shape, the name, the last-write-wins rule and the PII-free property are untouched, and a read never throws. **`INT-ON-FAILURE-HOOK-CONTRACT` UNCHANGED, checked**: a lost-lock job pages with the argv the completed listener would have used for its record (`policy` and the reason), or not at all. | diff --git a/specs/requirements.md b/specs/requirements.md index d832fa43..f04da7dd 100644 --- a/specs/requirements.md +++ b/specs/requirements.md @@ -512,7 +512,12 @@ and nothing about the box itself (`INT-CONTAINER-RUNTIME-CONTRACT`). - **Traces to**: `CONST-PI-VERSION-PINNED`, `DES-CLI-TRIGGER-FOR-LOCAL`, `INT-RUNNER-EXIT-CODE-PROTOCOL` - **Acceptance**: Given a local job reaching a terminal state, the worker console shows exactly one completion or failure line carrying the job id and outcome; during the run, the container's output is - visible there. + visible there. A job whose run finished but whose completion the queue refused after losing its lock + (a Valkey outage longer than the lock renewal window) ends with `job_lost_lock_after_completion` + (`jobId`, `outcome`, `reason` when present) in place of `job_failed`, when the queue then failed it; a + scheduled job the queue handed back logs that line and then its one `job_completed` line, with the + recorded outcome, and is not run again. A record that was found and refused is said on its own line, + `job_lost_lock_record_rejected` (`jobId`, a fixed `reason`, `source`), before the failure line. ## REQ-CRON-SCHEDULED-JOBS @@ -3124,6 +3129,7 @@ instead of drifting. | Date | Change | |---|---| +| 2026-10-04 | Found this round (no issue): a job that completed while Valkey was unreachable was reported failed (`job_failed`, a failure comment and a page), or run again. **`REQ-LOCAL-JOB-VISIBILITY` AMENDED**, Acceptance: `job_lost_lock_after_completion` (`jobId`, `outcome`, `reason` when present) is the terminal line of a job the queue failed after it lost the lock of a finished run, in place of `job_failed`; a scheduled job handed back logs it and then its one `job_completed` line. A record found and refused is logged as `job_lost_lock_record_rejected` (`jobId`, a fixed `reason`, `source`) before the failure line. The Statement is UNCHANGED, checked. **Code evidence**: worker/src/start.mjs (the failed listener), worker/src/index.mjs (the processor's lost-lock gate). | | 2026-10-04 | Issue #504, part B (apply, state, enforcement and the fleet). **`REQ-DELEGATED-ALLOCATION` AMENDED**: the Statement adds `envelope-mismatch` to the plan ladder after the writer rung, `allocation-cap` and `envelope-mismatch` as pre-spend refusals, the shared applied split and the audit file; the Scope says what part B ships, and that a host with no envelope in a fleet with an applied split refuses its jobs (turning delegation off for a fleet is removing the envelope everywhere and deleting `alloc:plan` and `alloc:envelope:expected`); the Acceptance adds the persisted neutral seed, the live two-writer race (one applies, the other `plan-stale` or `plan-busy`), the `apply-failed` row after a lost compare-and-set, one `expired` row across two hosts, the re-base gated on `alloc:envelope:expected` with one `envelope-changed-externally` row otherwise, `expected` seeded with the applied digest whichever host looks first (PR #574's review: the lazy seed let the first differing host re-base the fleet), the no-envelope host, a pickup fault retried as infrastructure, a corrupt or newer stored state, the synthetic ledger for a project with no row and `_other`'s own ledger, `allocation-cap` against `dollar-cap` (a tie is `dollar-cap`), a shrink that never touches a running job, the operator revert, the last good envelope kept on a reload inside a job path, and the local folder resolved at prepare (`local-folder-escaped` judged by identity, `local-folder-holds-envelope`, `local-folder-project-changed`, a chained child on the named folder). **`REQ-DEPLOYMENT-BOOTSTRAP` AMENDED**: doctor's list of the worker keys it resolves from `.env` names `PI_DISPATCH_RUN_ROOTS`, read for the envelope's containment check. **`REQ-SCOPED-LIMITS`** UNCHANGED, checked: its Why already bounds a delegated allocation by the operator's own limits, which `min(row, allocation)` keeps. **`REQ-SPEND-CAPS-MULTI-WINDOW`** UNCHANGED, checked: the operator's windows behave as before; a deployment window narrowed by an envelope total refuses `allocation-cap` under `REQ-DELEGATED-ALLOCATION`, not under this entry. **`REQ-OPERATOR-FAILURE-NOTIFICATION`** UNCHANGED, checked: every new reason is a pre-spend refusal and pages nobody. **`REQ-ADMIN-VIA-PI-EXTENSION`** UNCHANGED, checked: no tool lands in this part. **`CONST-BUDGET-BEFORE-TOKENS`** UNCHANGED, checked: `envelope-mismatch` is a free gate placed with the others before the mint, `local-folder-project-changed` is decided after prepare and before the token-cap read and every reserve, and `allocation-cap` is decided where `dollar-cap` is. | | 2026-10-04 | Issue #504, part A (the doctrine and the pure modules). **`REQ-DELEGATED-ALLOCATION` NEW**: the worker applies a priorities plan with no keypress, inside an operator envelope, under fixed rules (floors first, the rest by weight with largest remainder and ties by id, a step of at most `floor(total * maxStepPct / 100)` from the applied vector, all weights 0 leaving the headroom unallocated), refuses in a fixed order (delegation off, writer not allowed, duplicate, stale, too soon, incomplete, busy) and records every attempt; the worked cases are stated with `_other` at its default weight of 1 (neutral 36,666,666, 36,666,667 and 26,666,667, and 3:1:0 from it clamped to 61,666,666, 31,666,667 and 6,666,667) and with `_other` at 0 where the issue's 70/30 and 75/25 apply; for the envelope's window a project's cap becomes the smaller of its row and its allocation. **`REQ-ADMIN-VIA-PI-EXTENSION` AMENDED**: the Why keeps the daily cap sentence and adds that one model-callable write needs no keypress, `dispatch_priorities_set`, bounded by the floors, the step and interval rules and the envelope total; the Statement describes the three tools issue #504 adds in prose only, because the scan in `admin/test/wiring.test.mjs` matches the registered tools exactly and their names join the list in part C, which registers them; the Acceptance reads a settings, trigger, limit or envelope write tool, and adds that the priorities write applies or refuses with or without an operator and never shows a confirm. **`REQ-SCOPED-LIMITS` AMENDED**, the Why: the sentence that a limit only ever narrows what may spend was already loose (a confirm-gated limit edit can raise a cap), and now says an operator edit may narrow or widen behind a keypress, and a delegated allocation moves headroom with none but never above the envelope total, below a floor, or above a limit the operator wrote. **`CONST-BUDGET-BEFORE-TOKENS` UNCHANGED, checked**: the ordering is untouched, and only the values the dollar reserve compares against change. | | 2026-10-04 | Issue #571. **`REQ-TOKEN-ACCOUNTING-AND-CAPS` AMENDED**: a call whose answer carries broken usage is not priced; the meter counts it `costUnreported` on every run, capped or not, a child's included, a zero-rated model and a forwarding router's own call never; Acceptance clauses for the capped and uncapped cases, the output-0 answer, the zero-rated model, the async router and the guard-less child. **`REQ-COST-ANALYTICS` AMENDED**, rule (d) and Acceptance: a run with `costUnreported` above 0 is a floor with or without a cap, so is a capped run that lacks `costUnreported` or one of the guard's three short-count counters, and so is a present floor counter that is not a whole number at least 0; an uncapped record with no `costUnreported` stays exact. **`REQ-SPEND-CAPS-MULTI-WINDOW` AMENDED**, Acceptance only: `costUnreported` joins the floor counters. **`REQ-EGRESS-ALLOWLIST` UNCHANGED, checked**: the new doctor line is a cost warning, recorded in `INT-MODEL-ENDPOINTS-FILE-CONTRACT`. | diff --git a/worker/src/index.mjs b/worker/src/index.mjs index 06f8e6f0..776cecda 100644 --- a/worker/src/index.mjs +++ b/worker/src/index.mjs @@ -26,6 +26,17 @@ export const QUEUE = "pi-jobs"; /** The key the host-wide in-flight count lives under. One machine, one counter, whatever the queue. */ export const HOST_SLOT_KEY = "host"; export const JOB_TIMEOUT_MS = 30 * 60 * 1000; // REQ-JOB-TIMEOUT-30M + +/** + * The failed reason BullMQ gives a job its stall check failed (`maxStalledCount: 0` below): the literal in the pinned + * bullmq's `moveStalledJobsToWait`, stored as the job's deferred failure and thrown as an UnrecoverableError at the + * next pickup. Matched EXACTLY by start.mjs's failed listener, and pinned against the installed bullmq source by a + * test, so a bullmq bump that rewords it fails that test rather than silently turning the lost-lock check off. + */ +export const STALLED_FAILED_REASON = "job stalled more than allowable limit"; + +/** How many (job, stall count, attempt, source) keys the processor remembers having logged a refused record for. */ +export const REJECTED_SEEN_MAX = 1000; // The scope-busy re-check (issue #242): a held scope has no natural "until" (the holder may run to // JOB_TIMEOUT_MS), so a deferred job re-tests on a fixed cadence. 5s keeps the worst case trivial // (<=360 wakes across a 30-minute hold, each ~1ms of synchronous predicate briefly occupying a slot) @@ -175,8 +186,44 @@ export function effectiveJobOf(data, settings, allowedModels = null, log = () => }; } -export function makeProcessor({ cancelJob, stopContainer, containerName = (job) => jobContainerName(job.id), redis, getSettings, applyConcurrency = () => {}, pauseUntil = () => null, scopedLimits = () => [], projects = () => [], allocation = null, inFlight = makeInFlight(), hostBound = null, checkLease = null, scopeLease = null, endpointSlots = makeInFlight(), endpointLease = null, modelEndpoints = null, overlayModels = () => null, endpointSetFor = mainModelEndpoints, deps, recordRun = () => {}, timeoutMs = JOB_TIMEOUT_MS, cancelPollMs = 2_000, cancelStopBoundMs = CANCEL_STOP_BOUND_MS, hostName = "", now = () => Date.now(), waitState = makeWaitState({ redis, now }), afterMaxMs = () => WAIT_AFTER_MAX_DEFAULT_MS, checkSlots = makeInFlight(), checkSlotCount = () => 1, checkTimeoutMs = () => 10_000, concurrencyNow = () => 3, intervalMs = () => WAIT_INTERVAL_FLOOR_MS * 2, maxWaitMs = () => 24 * 3600 * 1000, maxChecks = () => 96, maxFaults = () => 5, random = Math.random }) { +export function makeProcessor({ cancelJob, stopContainer, containerName = (job) => jobContainerName(job.id), redis, getSettings, applyConcurrency = () => {}, pauseUntil = () => null, scopedLimits = () => [], projects = () => [], allocation = null, inFlight = makeInFlight(), hostBound = null, checkLease = null, scopeLease = null, endpointSlots = makeInFlight(), endpointLease = null, modelEndpoints = null, overlayModels = () => null, endpointSetFor = mainModelEndpoints, deps, recordRun = () => {}, settledRecord = null, timeoutMs = JOB_TIMEOUT_MS, cancelPollMs = 2_000, cancelStopBoundMs = CANCEL_STOP_BOUND_MS, hostName = "", now = () => Date.now(), waitState = makeWaitState({ redis, now }), afterMaxMs = () => WAIT_AFTER_MAX_DEFAULT_MS, checkSlots = makeInFlight(), checkSlotCount = () => 1, checkTimeoutMs = () => 10_000, concurrencyNow = () => 3, intervalMs = () => WAIT_INTERVAL_FLOOR_MS * 2, maxWaitMs = () => 24 * 3600 * 1000, maxChecks = () => 96, maxFaults = () => 5, random = Math.random }) { + // The lost-lock gate's rejection lines, said ONCE per job id, stall count and attempt. The gate runs before every + // deferral (pause window, wait, scope or endpoint busy), and BullMQ never resets a job's stall count, so a stalled + // scheduled job whose record is refused meets the gate again on every deferred pickup: a 30-minute scope-busy hold + // is about 360 of them. The verdict cannot change between them (same record, same attempt), so one line says it. + // Bounded, oldest out first, so a long-lived worker holds at most REJECTED_SEEN_MAX keys. + const rejectedSeen = new Set(); + const firstRejection = (key) => { + if (rejectedSeen.has(key)) return false; + rejectedSeen.add(key); + if (rejectedSeen.size > REJECTED_SEEN_MAX) rejectedSeen.delete(rejectedSeen.values().next().value); + return true; + }; return async function processor(job, token, signal) { + // THE LOST-LOCK GATE, first because it is free and because it can only ever stop a run (CONST-RETRY-INFRA-ONLY). + // BullMQ hands a job to the processor again after its stall check took it back. That happens to a job whose + // processor FINISHED when Valkey was unreachable for longer than the lock renewal window: the record was + // written, the completion was refused ("Missing lock"), and the job stayed active without a lock. A plain job + // then carries a deferred failure and never reaches here (start.mjs's failed listener handles it), but a + // scheduled job is moved back to wait and would run again, PAID, with its second record overwriting the + // first. So a job that has stalled (`stalledCounter > 0`) and whose record says this same attempt finished + // without failing ends as that record says, without a container. `budgetReserved` is the record's own: the + // completed listener then pages exactly when it would have for the first finish, which never reached it. + // No record (a worker that died mid-run, a lookup fault) keeps today's path: the job runs, and the stall + // guard bounds how often. + if (Number(job.stalledCounter) > 0 && typeof settledRecord === "function") { + const attempt = (Number.isInteger(job.attemptsMade) && job.attemptsMade >= 0 ? job.attemptsMade : 0) + 1; + // A record found and refused is said, with its fixed reason, because the job then RUNS again (paid). + const onReject = (reason, source) => { + if (firstRejection(`${job.id}\u0000${job.stalledCounter}\u0000${attempt}\u0000${source}`)) deps?.log?.("job_lost_lock_record_rejected", { jobId: job.id, reason, source }); + }; + const record = await settledRecord(job.id, { attempt, since: job.timestamp, onReject }); + if (record) { + deps?.log?.("job_lost_lock_after_completion", { jobId: job.id, outcome: record.outcome, ...(record.reason ? { reason: record.reason } : {}) }); + return { outcome: record.outcome, reason: record.reason ?? null, exitCode: record.exitCode ?? null, turns: record.turns ?? null, tokens: record.tokens ?? null, budgetReserved: record.budgetReserved ?? null }; + } + } + // Scoped pause windows (REQ-SCOPED-PAUSE-WINDOWS): if this job's folder/repo is inside an active pause // window, DEFER it to the window end via BullMQ's delayed set -- the job keeps its identity/dedup and // auto-resumes when re-picked. This is FIRST, before the kill timer, the settings read, and the budget @@ -1322,7 +1369,7 @@ export function keeperHoldState(job, at) { return { at, since, startedMs }; } -export function createWorker({ connection, name, stopContainer, containerName, hostQueue = null, checkLease = null, scopeLease = null, checkTimeoutMs, concurrency, getSettings, redis, deps, recordRun, limiter, pauseUntil, scopedLimits, projects, allocation = null, inFlight = makeInFlight(), waitState, afterMaxMs, checkSlots = makeInFlight(), checkSlotCount, concurrencyNow, intervalMs, maxWaitMs, maxChecks, maxFaults, hostSlots = makeInFlight(), endpointSlots = makeInFlight(), endpointLease = null, modelEndpoints = null, overlayModels, extraClosers = [] }) { +export function createWorker({ connection, name, stopContainer, containerName, hostQueue = null, checkLease = null, scopeLease = null, checkTimeoutMs, concurrency, getSettings, redis, deps, recordRun, settledRecord = null, limiter, pauseUntil, scopedLimits, projects, allocation = null, inFlight = makeInFlight(), waitState, afterMaxMs, checkSlots = makeInFlight(), checkSlotCount, concurrencyNow, intervalMs, maxWaitMs, maxChecks, maxFaults, hostSlots = makeInFlight(), endpointSlots = makeInFlight(), endpointLease = null, modelEndpoints = null, overlayModels, extraClosers = [] }) { // One Worker per queue name (issue #57). A host-affine job -- one whose folder, secret resolver or wait // check lives on THIS machine -- is enqueued to `pi-jobs@` rather than filtered for at pickup, // because BullMQ has no selective pop and the put-it-back alternative does not work: promotion out of @@ -1407,6 +1454,9 @@ export function createWorker({ connection, name, stopContainer, containerName, h maxFaults, deps, recordRun, + // The run-record lookup a stalled job is checked against before it can run again (see the processor's + // first gate). `null` in a bare wiring, which keeps today's behaviour: the job runs. + settledRecord, }); // Issue #464: only a connection `parseConnection` built, which judges and pins the Valkey it dials. diff --git a/worker/src/run-history.mjs b/worker/src/run-history.mjs index 30201853..507dc9c6 100644 --- a/worker/src/run-history.mjs +++ b/worker/src/run-history.mjs @@ -945,6 +945,132 @@ export function makeFindPreviousRun({ logsDir, fs = nodeFs }) { }; } +/** + * What `makeReadRecord` returns for a record file that exists but cannot be read as a record (unreadable, bad JSON, + * not an object). Distinct from `null` (no file), so the lost-lock check can say which it met. + */ +export const UNREADABLE_RECORD = Object.freeze({ unreadable: true }); + +/** + * Read one job's run record back, by the name `makeRecordWriter` gave it: `/.json` + * (INT-RUN-HISTORY-FILE-CONTRACT, unchanged). Returns the parsed object, `null` when there is no file, or + * `UNREADABLE_RECORD` when there is one that is not a record. + * + * Why the worker reads its own record: the queue can lose a job's lock after the processor finished and wrote + * this file, when Valkey was unreachable for longer than BullMQ's lock renewal window. BullMQ then refuses the + * completion ("Missing lock") and its stall check takes the job back, so the queue alone says the job failed or + * must run again. The record is the store that already knows it finished (DES-TERMINAL-COMMENTS-AND-FAILURE-HOOK). + * + * The id inside is NOT checked here: `sanitizeJobId` maps `a:b` and `a_b` to one name, and `recordVerdict` refuses + * the other job's record as `id-mismatch`, which is the one place that can also say so. NEVER throws. + */ +export function makeReadRecord({ logsDir, fs = nodeFs }) { + return function readRecord(jobId) { + if (jobId === null || jobId === undefined || jobId === "") return null; + let text; + try { + text = fs.readFileSync(join(logsDir, `${sanitizeJobId(jobId)}.json`), "utf8"); + } catch (err) { + return err?.code === "ENOENT" || err?.code === "ENOTDIR" ? null : UNREADABLE_RECORD; + } + try { + const record = JSON.parse(text); + return record !== null && typeof record === "object" && !Array.isArray(record) ? record : UNREADABLE_RECORD; + } catch { + return UNREADABLE_RECORD; + } + }; +} + +/** + * How far a record's `startedAt` may sit BEFORE the job's creation and still be this job's record: 5 minutes. + * + * The two instants come from two clocks. `job.timestamp` is stamped by whoever ENQUEUED the job (the receiver, the + * CLI, another worker) and `startedAt` by the worker that ran it, so a producer whose clock runs ahead of the + * worker's would make every genuine record look older than its job and silently turn this check off. The + * tolerance absorbs ordinary skew between hosts; what it must still refuse is a record of an OLDER job under a + * reused id, and such a record is older by at least that job's whole run, its removal from the queue and a new + * enqueue, which in practice is far more than 5 minutes. A skew larger than this is an operator problem the + * `job_lost_lock_record_rejected` line (`older-than-job`) makes visible. + */ +export const RECORD_CLOCK_SKEW_MS = 5 * 60 * 1000; + +/** + * Does this record say that THIS attempt of THIS job finished without failing? Returns `null` when it does, + * `"absent"` when there is no record, and otherwise the reason it does not, from a fixed set: + * `unreadable`, `id-mismatch`, `other-attempt`, `failed`, `older-than-job`. Pure; never throws. + * + * `attempt` is the 1-based attempt number the record would carry, which is BullMQ's `attemptsMade + 1` while the + * job is processing (see `buildRecord`) and `attemptsMade` after BullMQ failed it (its moveToFailed adds one + * before the failed event). A record of an earlier attempt does not answer for a later one. + * + * `since` is the job's creation time in millis (BullMQ's `job.timestamp`). A job id can be used again once the + * old job is gone (a fixed `--trigger` id, a redelivered webhook), while its record lives for the retention + * window; a record that started before this job existed (less `RECORD_CLOCK_SKEW_MS`) belongs to the old one. A + * missing or unparseable `startedAt` is not trusted for the same reason. + * + * `failed` (an outcome of `failed`, or none) is refused on purpose: a record written by the catch path says the run + * threw, and the queue's verdict (failed, or run again) is then the right one. + */ +export function recordVerdict(record, { jobId, attempt, since } = {}) { + try { + if (record === null || record === undefined) return "absent"; + if (record === UNREADABLE_RECORD || typeof record !== "object" || Array.isArray(record)) return "unreadable"; + if (record.jobId !== jobId) return "id-mismatch"; + if (!Number.isInteger(attempt) || record.attempt !== attempt) return "other-attempt"; + if (typeof record.outcome !== "string" || record.outcome === "failed") return "failed"; + if (Number.isFinite(since)) { + const started = Date.parse(record.startedAt ?? ""); + if (!Number.isFinite(started) || started < since - RECORD_CLOCK_SKEW_MS) return "older-than-job"; + } + return null; + } catch { + return "unreadable"; + } +} + +/** `recordVerdict` as a yes or no. */ +export const recordSettlesAttempt = (record, opts) => recordVerdict(record, opts) === null; + +/** + * The lookup the worker asks when the queue lost a job's lock: the local record first, then the fleet's copy. + * + * On a fleet without a shared `PI_LOGS_DIR`, the host that meets the stalled job is not always the host that ran + * it, so the local file can be absent or hold an older attempt. The run mirror (`run-mirror.mjs`) holds the same + * record's bytes, so it is asked when the local file does not settle the attempt. `readMirrored` is `null` on a + * deployment that declared no worker name, which has no mirror and is one host. + * + * `onReject(reason, source)` is told about every record it FOUND and refused (`source` is `local` or `mirror`, + * `reason` one of `recordVerdict`'s tokens, never `absent`), so the caller can log why the failure path stood. + * + * Resolves to the record or `null`, and NEVER rejects: a mirror fault is no record, which fails toward today's + * behaviour (a failure comment, or a run). + */ +export function makeSettledRecord({ readRecord, readMirrored = null }) { + return async function settledRecord(jobId, { attempt, since, onReject = () => {} } = {}) { + const judge = (record, source) => { + const why = recordVerdict(record, { jobId, attempt, since }); + if (why !== null && why !== "absent") { + try { + onReject(why, source); + } catch { + // a reporting fault must not change the verdict + } + } + return why === null; + }; + try { + const local = readRecord(jobId); + if (judge(local, "local")) return local; + if (typeof readMirrored !== "function") return null; + const mirrored = await readMirrored(jobId); + return judge(mirrored, "mirror") ? mirrored : null; + } catch { + return null; + } + }; +} + /** * The durable log reaper: an age sweep that deletes `.log` and `.json` history files older than the * retention window, keeping the logs directory bounded. Runs at boot AND on the retention timer since diff --git a/worker/src/run-mirror.mjs b/worker/src/run-mirror.mjs index 589b1220..a6a16286 100644 --- a/worker/src/run-mirror.mjs +++ b/worker/src/run-mirror.mjs @@ -1,3 +1,5 @@ +import { UNREADABLE_RECORD } from "./run-history.mjs"; + /** * A fleet-visible copy of the run history (issue #57, Gap 3). * @@ -186,6 +188,34 @@ export async function readMirroredRuns(redis, { limit = 50, sinceMs = 0, now = ( return { runs, degraded: runs.length >= limit ? "truncated" : "ok" }; } +/** + * One run's mirrored record, by its sanitized id: the parsed object, `null`, or `UNREADABLE_RECORD`. Never throws, + * never rejects. + * + * The by-id read the lost-lock check needs (`makeSettledRecord` in `run-history.mjs`): one bounded `GET`, not the + * index scan `readMirroredRuns` does for the panel. Unreachable and absent are `null` (nothing found); a value that + * is there but is not a record (unparseable, empty, not an object) is `UNREADABLE_RECORD`, exactly as the local + * reader says it, so the check can log `unreadable` from `mirror`. Neither is ever accepted: the caller keeps + * today's path. + */ +export async function readMirroredRecord(redis, sanitizedJobId, { timeoutMs = OP_TIMEOUT_MS } = {}) { + if (!redis || !sanitizedJobId) return null; + try { + const raw = await bounded(redis.get(runRecordKey(sanitizedJobId)), timeoutMs); + if (raw === null || raw === undefined) return null; + if (typeof raw !== "string" || raw === "") return UNREADABLE_RECORD; + let rec; + try { + rec = JSON.parse(raw); + } catch { + return UNREADABLE_RECORD; + } + return rec !== null && typeof rec === "object" && !Array.isArray(rec) ? rec : UNREADABLE_RECORD; + } catch { + return null; // unreachable or timed out: nothing was read + } +} + /** * One list from two sources. * diff --git a/worker/src/start.mjs b/worker/src/start.mjs index 6c7530e1..5e3c1d6b 100644 --- a/worker/src/start.mjs +++ b/worker/src/start.mjs @@ -26,7 +26,7 @@ import { capabilityTokens, serializeCaps } from "./capabilities.mjs"; import { cronFingerprint } from "./fingerprint.mjs"; import { makeHostRegistry } from "./host-registry.mjs"; import { makeImagePreflight } from "./image-preflight.mjs"; -import { createWorker, JOB_TIMEOUT_MS } from "./index.mjs"; +import { createWorker, JOB_TIMEOUT_MS, STALLED_FAILED_REASON } from "./index.mjs"; import { BOOT_REFUSING_JOB_USER_CAUSES, DAEMON_FACTS_TIMEOUT_MS, jobUserRefusal, makeDaemonFactsReader, makeJobUserResolver, relabelsPrivateMounts, resolveImageUser } from "./job-user.mjs"; import { makeCollectChain } from "./outbox.mjs"; import { makeCollectPlan } from "./outbox-plan.mjs"; @@ -60,8 +60,8 @@ import { PODMAN_RESTART_HOLD_EXPIRED, makePodmanServiceReader, onceFs, makeRootf import { makeRunContainer } from "./run-container.mjs"; import { resolveProviderCredential } from "./env-allowlist.mjs"; import { makeSecretsResolver } from "./secrets.mjs"; -import { buildRecord, makeFindPreviousRun, makeLogReaper, makeLogSink, makeRecordWriter, RUNNER_POLICY_REASONS, sanitizeJobId } from "./run-history.mjs"; -import { makeRunMirror } from "./run-mirror.mjs"; +import { buildRecord, makeFindPreviousRun, makeLogReaper, makeLogSink, makeReadRecord, makeRecordWriter, makeSettledRecord, RUNNER_POLICY_REASONS, sanitizeJobId } from "./run-history.mjs"; +import { makeRunMirror, readMirroredRecord } from "./run-mirror.mjs"; import { readOverlay, resolveSettings } from "./runtime-settings.mjs"; import { usdFingerprint } from "./dollar-fingerprint.mjs"; import { authoredCron, envelopeJobPaths, loadSchedules, servedSchedules } from "./schedules.mjs"; @@ -584,6 +584,8 @@ export async function startWorker( makeLogSink: makeLogSinkFn = makeLogSink, makeRecordWriter: makeRecordWriterFn = makeRecordWriter, makeRunMirror: makeRunMirrorFn = makeRunMirror, + // The fleet copy's by-id read, seamed so a wiring test can answer it without a live mirror. + readMirroredRecord: readMirroredRecordFn = readMirroredRecord, makeLogReaper: makeLogReaperFn = makeLogReaper, makeSandboxReaper: makeSandboxReaperFn = makeSandboxReaper, makeSandboxNetworkSweeper: makeSandboxNetworkSweeperFn = makeSandboxNetworkSweeper, @@ -1345,6 +1347,13 @@ export async function startWorker( // the disarm -- the same chosen direction, met at shutdown instead of a crash. void disarmOnce({ job, endedAt }); }; + // The record read back, for a job the queue lost the lock of after it finished (DES-TERMINAL-COMMENTS-AND-FAILURE-HOOK, + // CONST-RETRY-INFRA-ONLY): this host's own file first, then the fleet's copy where a mirror is armed, because the + // host that meets the stalled job need not be the one that ran it. A single host has no mirror and needs none. + const settledRecord = makeSettledRecord({ + readRecord: makeReadRecord({ logsDir: config.logsDir }), + readMirrored: runMirror ? (jobId) => readMirroredRecordFn(redis, sanitizeJobId(jobId)) : null, + }); // INT-CONFIG-OVERLAY-CONTRACT: the worker reads the runtime-settings overlay at EACH job start, so this // closure -- not a value frozen at boot -- is what the processor calls per job. It resolves the fourteen @@ -1889,6 +1898,9 @@ export async function startWorker( // `operator-cancel`, because the operator initiated it and a push telling them what they just did is // noise with a pager attached. const HOOK_POLICY_REASONS = new Set(["worker-abort", "runner-policy", ...RUNNER_POLICY_REASONS]); + // One predicate for the completed listener and the lost-lock path below, so a record replays exactly the page its + // result would have sent. + const pagesAsPolicy = (result) => Boolean(onFailure) && result?.outcome === "policy" && HOOK_POLICY_REASONS.has(result.reason) && result.budgetReserved !== false; // The infra-terminal sentence (issue #288). FIXED, never err.message: the message classes that reach // a failedReason carry host paths and library words (the #310 record), and for a local job this text // lands verbatim in the service log through the adapter's stdout fallthrough. The worker log already @@ -1925,6 +1937,7 @@ export async function startWorker( getSettings, redis, recordRun, + settledRecord, extraClosers, // REQ-SCOPED-PAUSE-WINDOWS: the processor defers a job whose folder/repo is inside an active window. // Reads the live-reloaded ref, so an operator edit takes effect on the next job without a restart. @@ -2198,11 +2211,13 @@ export async function startWorker( // `budgetReserved !== false` (issue #502): `model-not-allowed` is both a runner stop (paid, pages) and a // pre-spend refusal of a main model outside the job's list (free, comments, pages nobody), and the reason // alone cannot tell them apart. Every paid terminal above carries `budgetReserved: true`. - if (onFailure && result?.outcome === "policy" && HOOK_POLICY_REASONS.has(result.reason) && result.budgetReserved !== false) { + if (pagesAsPolicy(result)) { onFailure({ jobId: job?.id, outcome: "policy", reason: result.reason }); } }); - for (const w of allWorkers) w.on("failed", (job, err) => { + // The failed listener's body, for a failure the queue decided. Split out only so the lost-lock check below can + // run it after an await; a failure of any other reason runs it synchronously, exactly as before. + const onFailed = (job, err) => { log("job_failed", { jobId: job?.id, attempt: job?.attemptsMade, reason: String(err?.message ?? err).slice(0, 120) }); // The one reason whose sentence is the fix itself (issue #458): logged WHOLE, beside the cut line above. if (err?.reason === NETNS_KEEPER_NOT_HOLDING || err?.reason === NETNS_KEEPER_CRASH_LOOP) log("job_failed_netns_keeper", { jobId: job?.id, attempt: job?.attemptsMade, reason: String(err?.message ?? "") }); @@ -2217,6 +2232,29 @@ export async function startWorker( void comment({ ...job.data, id: job.id }, (typeof err?.reason === "string" && Object.hasOwn(FAILED_COMMENT_BY_REASON, err.reason) ? FAILED_COMMENT_BY_REASON[err.reason] : null) ?? FAILED_COMMENT); onFailure?.({ jobId: job?.id, outcome: "failed", reason: typeof err?.reason === "string" ? err.reason : "infra" }); } + }; + // A job that FINISHED, then lost its lock (DES-TERMINAL-COMMENTS-AND-FAILURE-HOOK). When Valkey is unreachable + // for longer than the lock renewal window while a processor finishes, the record is written, BullMQ refuses the + // completion ("Missing lock"), and its stall check (`maxStalledCount: 0`) fails the job at the next pickup with + // STALLED_FAILED_REASON. Without this check that job posted the failure comment and paged the operator for a run + // that ended. So a terminal stall failure first asks the run record: when the record says this attempt finished + // without failing, the job's terminal line is `job_lost_lock_after_completion` and nothing is posted. The page + // the completed listener would have sent for that record (a paid policy stop) is sent here instead, because + // that listener never saw the first finish. No record, an earlier attempt's, a `failed` one, or a lookup fault + // keeps the failure path below: the comment and the page. + for (const w of allWorkers) w.on("failed", (job, err) => { + if (!(job?.finishedOn && err?.message === STALLED_FAILED_REASON)) return onFailed(job, err); + void (async () => { + // `attemptsMade` is read AFTER BullMQ's moveToFailed added one, so it equals the attempt number the + // record carries (`buildRecord`'s `attemptsMade + 1`, written while the job was processing). + // A record found and refused is said, with its fixed reason, so an operator reading the failure comment + // can see why the record did not suppress it (a clock skew past the tolerance reads `older-than-job`). + const onReject = (reason, source) => log("job_lost_lock_record_rejected", { jobId: job.id, reason, source }); + const record = await settledRecord(job.id, { attempt: job.attemptsMade, since: job.timestamp, onReject }); + if (!record) return onFailed(job, err); + log("job_lost_lock_after_completion", { jobId: job.id, outcome: record.outcome, ...(record.reason ? { reason: record.reason } : {}) }); + if (pagesAsPolicy(record)) onFailure({ jobId: job.id, outcome: "policy", reason: record.reason }); + })().catch(() => {}); // `settledRecord` never rejects; this only keeps a throwing log line from going unhandled }); // CONST-RETRY-INFRA-ONLY money backstop: BullMQ's maxStalledCount does not bound scheduler jobs, so a diff --git a/worker/test/lost-lock.test.mjs b/worker/test/lost-lock.test.mjs new file mode 100644 index 00000000..d5ce385e --- /dev/null +++ b/worker/test/lost-lock.test.mjs @@ -0,0 +1,190 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { createRequire } from "node:module"; +import { dirname, join } from "node:path"; +import { test } from "node:test"; +import { makeSettledRecord } from "../src/run-history.mjs"; + +// The processor's lost-lock gate (DES-TERMINAL-COMMENTS-AND-FAILURE-HOOK, CONST-RETRY-INFRA-ONLY): a job BullMQ hands +// back after its stall check, whose run record says this attempt already finished, ends as the record says and is +// never run again. index.mjs imports bullmq, so this skips below the node floor / without deps and runs in CI, where +// PI_DISPATCH_REQUIRE_WORKER_TESTS=1 turns a skip into a hard failure. +let mod; +let importError; +try { + mod = await import("../src/index.mjs"); +} catch (error) { + importError = error; +} +if (!mod && process.env.PI_DISPATCH_REQUIRE_WORKER_TESTS === "1") { + throw new Error(`lost-lock tests are REQUIRED here but bullmq could not import.\n${importError}`); +} +const skip = mod ? false : `bullmq not installed (node ${process.version} < 22.19.0); CI runs these`; + +const CREATED = Date.parse("2026-10-04T10:00:00.000Z"); + +/** A scheduled job as BullMQ hands it back after a stall: same id, same attemptsMade, `stalledCounter` raised. */ +const stalledJob = (over = {}) => ({ + id: "repeat:nightly:1759572000000", + name: "local", + attemptsMade: 0, + stalledCounter: 1, + timestamp: CREATED, + data: { kind: "local", folder: "/proj", flow: "tidy", provider: "anthropic", model: "m", maxTurns: 7, trigger: { id: "nightly", pattern: "0 3 * * *" } }, + ...over, +}); + +/** The record the first run wrote before its completion was refused. */ +const completedRecord = (over = {}) => ({ + jobId: "repeat:nightly:1759572000000", + outcome: "completed", + reason: null, + exitCode: 0, + turns: 4, + tokens: { input: 10, output: 5, total: 15, cost: 0.01 }, + budgetReserved: true, + attempt: 1, + startedAt: "2026-10-04T10:00:01.000Z", + endedAt: "2026-10-04T10:03:00.000Z", + ...over, +}); + +function harness({ record = completedRecord(), settledRecord } = {}) { + const seen = { containerCalls: 0, incr: 0, records: [], logs: [], asked: [] }; + // `null` is a bare wiring (no lookup at all); only an absent option takes the real lookup over `record`. + const lookup = + settledRecord !== undefined + ? settledRecord + : makeSettledRecord({ + readRecord: (id) => (seen.asked.push(id), record), + }); + const processor = mod.makeProcessor({ + cancelJob: () => {}, + stopContainer: () => {}, + redis: { incr: async () => (seen.incr++, 1), decr: async () => 0, expire: async () => {} }, + getSettings: () => ({ provider: "anthropic", model: "m", maxTurns: 30, dailyCap: 10, weeklyCap: null, monthlyCap: null, concurrency: 3, softHoldPct: null }), + recordRun: (rec) => seen.records.push(rec), + settledRecord: lookup, + timeoutMs: 100000, + // The clock the fixtures name (issue #284's rule): the gate reads no clock, and the gates after it read this one. + now: () => CREATED + 60_000, + deps: { + mintToken: async () => null, + isDefaultBranchProtected: async () => true, + prepareWorkspace: async () => ({ workspaceDir: "/w", jobDir: "/j" }), + runContainer: async () => (seen.containerCalls++, { code: 0, aborted: false, turns: 3 }), + cleanup: async () => {}, + comment: async () => {}, + log: (event, fields) => seen.logs.push({ event, ...fields }), + }, + }); + return { processor, seen }; +} + +test("a stalled job whose record says this attempt completed is returned as recorded and never run again", { skip }, async () => { + const { processor, seen } = harness(); + const result = await processor(stalledJob(), "tok", new AbortController().signal); + assert.equal(seen.containerCalls, 0, "no container: the run already happened and was paid for"); + assert.equal(seen.incr, 0, "no budget reservation either"); + assert.deepEqual(seen.records, [], "the first run's record is kept, never overwritten by a second one"); + assert.deepEqual(result, { outcome: "completed", reason: null, exitCode: 0, turns: 4, tokens: { input: 10, output: 5, total: 15, cost: 0.01 }, budgetReserved: true }); + assert.deepEqual(seen.logs.filter((l) => l.event === "job_lost_lock_after_completion"), [{ event: "job_lost_lock_after_completion", jobId: "repeat:nightly:1759572000000", outcome: "completed" }]); +}); + +test("the gate asks for attempt attemptsMade + 1 and the job's creation time", { skip }, async () => { + const asked = []; + const { processor, seen } = harness({ settledRecord: async (id, { attempt, since }) => (asked.push({ id, attempt, since }), null) }); + await processor(stalledJob({ attemptsMade: 1 }), "tok", new AbortController().signal); + assert.deepEqual(asked, [{ id: "repeat:nightly:1759572000000", attempt: 2, since: CREATED }]); + assert.equal(seen.containerCalls, 1, "no record: the job runs, today's path"); +}); + +test("a recorded policy stop is returned with its reason and its own budgetReserved, so the completed listener pages exactly as it would have", { skip }, async () => { + const { processor, seen } = harness({ record: completedRecord({ outcome: "policy", reason: "runner-policy", exitCode: 2, budgetReserved: true }) }); + const result = await processor(stalledJob(), "tok", new AbortController().signal); + assert.equal(seen.containerCalls, 0); + assert.equal(result.outcome, "policy"); + assert.equal(result.reason, "runner-policy"); + assert.equal(result.budgetReserved, true); + assert.deepEqual(seen.logs.find((l) => l.event === "job_lost_lock_after_completion"), { event: "job_lost_lock_after_completion", jobId: "repeat:nightly:1759572000000", outcome: "policy", reason: "runner-policy" }); +}); + +test("a recorded FREE refusal keeps budgetReserved false, so the completed listener pages nobody", { skip }, async () => { + // `model-not-allowed` is both the runner's paid stop and the worker's free pre-spend refusal; only + // `budgetReserved` tells them apart (issue #502), so the gate must hand on the record's own value. + const { processor, seen } = harness({ record: completedRecord({ outcome: "policy", reason: "model-not-allowed", exitCode: null, budgetReserved: false }) }); + const result = await processor(stalledJob(), "tok", new AbortController().signal); + assert.equal(seen.containerCalls, 0); + assert.equal(result.reason, "model-not-allowed"); + assert.equal(result.budgetReserved, false, "the record's own value, not a constant"); +}); + +test("a record the gate refuses is logged with its fixed reason before the job runs again", { skip }, async () => { + const { processor, seen } = harness({ record: completedRecord({ attempt: 3 }) }); + await processor(stalledJob(), "tok", new AbortController().signal); + assert.equal(seen.containerCalls, 1); + assert.deepEqual(seen.logs.filter((l) => l.event === "job_lost_lock_record_rejected"), [{ event: "job_lost_lock_record_rejected", jobId: "repeat:nightly:1759572000000", reason: "other-attempt", source: "local" }]); +}); + +test("a refused record is logged once per job, stall count and attempt, however often the job is picked up", { skip }, async () => { + // The gate runs before every deferral, and BullMQ never resets a job's stall count, so a held job meets it on + // every deferred pickup; the verdict cannot change between them. + const { processor, seen } = harness({ record: completedRecord({ attempt: 3 }) }); + for (let i = 0; i < 5; i++) await processor(stalledJob(), "tok", new AbortController().signal); + const rejected = () => seen.logs.filter((l) => l.event === "job_lost_lock_record_rejected"); + assert.equal(rejected().length, 1, "five pickups, one line"); + await processor(stalledJob({ stalledCounter: 2 }), "tok", new AbortController().signal); + assert.equal(rejected().length, 2, "a new stall is a new fact"); + await processor(stalledJob({ id: "repeat:other:1", stalledCounter: 2 }), "tok", new AbortController().signal); + assert.equal(rejected().length, 3, "another job is its own line"); +}); + +test("the remembered rejections are bounded at REJECTED_SEEN_MAX, oldest out first", { skip }, async () => { + const { processor, seen } = harness({ record: completedRecord({ attempt: 3, jobId: "x" }) }); + const pick = (n) => processor(stalledJob({ id: `job-${n}` }), "tok", new AbortController().signal); + for (let n = 0; n <= mod.REJECTED_SEEN_MAX; n++) await pick(n); + const count = () => seen.logs.filter((l) => l.event === "job_lost_lock_record_rejected").length; + assert.equal(count(), mod.REJECTED_SEEN_MAX + 1); + await pick(mod.REJECTED_SEEN_MAX); + assert.equal(count(), mod.REJECTED_SEEN_MAX + 1, "the newest is still remembered"); + await pick(0); + assert.equal(count(), mod.REJECTED_SEEN_MAX + 2, "the oldest was let go, so it may be said again"); + assert.equal(mod.REJECTED_SEEN_MAX, 1000); +}); + +test("the gate stays out of the way when it cannot show the attempt finished: the job runs", { skip }, async () => { + const cases = [ + ["a job that never stalled", { job: stalledJob({ stalledCounter: 0 }) }], + ["a record of another attempt", { record: completedRecord({ attempt: 2 }) }], + ["a failed record", { record: completedRecord({ outcome: "failed" }) }], + ["a record older than the job (a reused id)", { record: completedRecord({ startedAt: "2026-10-04T09:00:00.000Z" }) }], + ["no record", { record: null }], + ]; + for (const [label, { job = stalledJob(), record = completedRecord() }] of cases) { + const { processor, seen } = harness({ record }); + await processor(job, "tok", new AbortController().signal); + assert.equal(seen.containerCalls, 1, `${label}: runs`); + assert.ok(!seen.logs.some((l) => l.event === "job_lost_lock_after_completion"), `${label}: no lost-lock line`); + } + // A bare wiring (no lookup) keeps today's behaviour too. + const bare = harness({ settledRecord: null }); + await bare.processor(stalledJob(), "tok", new AbortController().signal); + assert.equal(bare.seen.containerCalls, 1); +}); + +test("a stalled job is not read for a record at all when it never stalled", { skip }, async () => { + const { processor, seen } = harness(); + await processor(stalledJob({ stalledCounter: 0 }), "tok", new AbortController().signal); + assert.deepEqual(seen.asked, [], "the first pickup of every job costs no file read"); +}); + +test("STALLED_FAILED_REASON is the literal the pinned bullmq's stall check writes", { skip }, () => { + // start.mjs's failed listener matches this EXACTLY. A bullmq bump that rewords it must fail here, not turn the + // lost-lock check off in silence. + const require = createRequire(import.meta.url); + const root = dirname(require.resolve("bullmq/package.json")); + const lua = readFileSync(join(root, "dist/esm/commands/moveStalledJobsToWait-9.lua"), "utf8"); + assert.equal(mod.STALLED_FAILED_REASON, "job stalled more than allowable limit"); + assert.ok(lua.includes(`local failedReason = "${mod.STALLED_FAILED_REASON}"`), "the stall check's deferred failure reason"); + assert.ok(lua.includes('rcall("HSET", jobKey, "defa", failedReason)'), "stored as the deferred failure the next pickup throws"); +}); diff --git a/worker/test/run-history.test.mjs b/worker/test/run-history.test.mjs index 73124071..32063e93 100644 --- a/worker/test/run-history.test.mjs +++ b/worker/test/run-history.test.mjs @@ -3,9 +3,10 @@ import { readFileSync } from "node:fs"; import { basename, join } from "node:path"; import { test } from "node:test"; import { createHmac } from "node:crypto"; -import { authenticExitLines, buildRecord, makeFindPreviousRun, makeLogReaper, makeLogSink, makeRecordWriter, parseExitCode, parseExitContext, parseExitReason, parseExitSession, parseExitTokens, parseExitTurns, parseExitUsage, PLAN_RECORD_REASONS, RUNNER_POLICY_REASONS, sanitizeJobId, TOKEN_KEYS } from "../src/run-history.mjs"; +import { authenticExitLines, buildRecord, makeFindPreviousRun, makeLogReaper, makeLogSink, makeReadRecord, makeRecordWriter, makeSettledRecord, RECORD_CLOCK_SKEW_MS, recordVerdict, UNREADABLE_RECORD, parseExitCode, parseExitContext, parseExitReason, parseExitSession, parseExitTokens, parseExitTurns, parseExitUsage, PLAN_RECORD_REASONS, recordSettlesAttempt, RUNNER_POLICY_REASONS, sanitizeJobId, TOKEN_KEYS } from "../src/run-history.mjs"; import { MODEL_REF_PATTERN } from "../src/model-ref.mjs"; import { FORGE_KINDS } from "../src/forges.mjs"; +import { tempDir } from "./helpers/temp-dir.mjs"; /** * A fake writable that records chunks and lets a test drive `finish`/`error` timing. @@ -1687,3 +1688,110 @@ test("PLAN_RECORD_REASONS is the collector's rungs, plan-invalid, envelope-misma const { ENVELOPE_MISMATCH_REASON } = await import("../src/allocation.mjs"); assert.deepEqual([...PLAN_RECORD_REASONS].sort(), [...new Set([...PLAN_COLLECT_REASONS, PLAN_INVALID, ENVELOPE_MISMATCH_REASON, ...PLAN_LADDER])].sort()); }); + +// ---- the lost-lock check reads a job's own record back (DES-TERMINAL-COMMENTS-AND-FAILURE-HOOK) ------------ + +test("makeReadRecord reads back exactly what makeRecordWriter wrote, by the same sanitized name", () => { + const logsDir = tempDir("pi-read-record-"); + const job = { id: "repeat:nightly:1759572000000", name: "local", attemptsMade: 0, data: { kind: "local", folder: "/srv/proj", flow: "tidy" } }; + const record = buildRecord({ job, result: { outcome: "completed", exitCode: 0, budgetReserved: true }, startedAt: "2026-10-04T10:00:01.000Z", endedAt: "2026-10-04T10:03:00.000Z" }); + makeRecordWriter({ logsDir })(record); + assert.deepEqual(makeReadRecord({ logsDir })(job.id), record); + assert.equal(makeReadRecord({ logsDir })("repeat:nightly:1"), null, "another job has no record"); + // `repeat_nightly_1759572000000` sanitizes to the same file name; its record must not answer for it. + // `repeat_nightly_1759572000000` sanitizes to the same file name: the read returns that file, and the verdict refuses it. + const collided = makeReadRecord({ logsDir })("repeat_nightly_1759572000000"); + assert.equal(recordVerdict(collided, { jobId: "repeat_nightly_1759572000000", attempt: 1 }), "id-mismatch", "a colliding sanitized name is not this job's record"); +}); + +test("makeReadRecord never throws: no file is null, a file that is not a record is UNREADABLE_RECORD", () => { + const failing = (code) => ({ readFileSync: () => { throw Object.assign(new Error(code), { code }); } }); + assert.equal(makeReadRecord({ logsDir: "/nope", fs: failing("ENOENT") })("j1"), null, "no file"); + assert.equal(makeReadRecord({ logsDir: "/nope", fs: failing("ENOTDIR") })("j1"), null, "no directory"); + assert.equal(makeReadRecord({ logsDir: "/l", fs: failing("EACCES") })("j1"), UNREADABLE_RECORD, "a file this worker cannot read"); + for (const body of ["{nope", "null", "[1]", "7"]) { + assert.equal(makeReadRecord({ logsDir: "/l", fs: { readFileSync: () => body } })("j1"), UNREADABLE_RECORD, body); + } + assert.equal(makeReadRecord({ logsDir: "/l", fs: { readFileSync: () => "{}" } })(), null, "no id"); + assert.equal(makeReadRecord({ logsDir: "/l", fs: { readFileSync: () => "{}" } })(null), null); +}); + +const SETTLED = { jobId: "j1", outcome: "completed", attempt: 1, startedAt: "2026-10-04T10:00:01.000Z" }; +const SINCE = Date.parse("2026-10-04T10:00:00.000Z"); + +test("recordSettlesAttempt: this job, this attempt, not failed, started after the job existed", () => { + const ok = (rec, over = {}) => recordSettlesAttempt(rec, { jobId: "j1", attempt: 1, since: SINCE, ...over }); + assert.equal(ok(SETTLED), true); + assert.equal(ok({ ...SETTLED, outcome: "policy", reason: "runner-policy" }), true, "a policy stop also finished"); + assert.equal(ok({ ...SETTLED, outcome: "failed" }), false, "a failed record keeps the queue's verdict"); + assert.equal(ok({ ...SETTLED, outcome: null }), false); + assert.equal(ok(SETTLED, { attempt: 2 }), false, "a record of attempt 1 does not answer for attempt 2"); + assert.equal(ok(SETTLED, { attempt: 0 }), false, "nor the other way"); + assert.equal(ok(SETTLED, { attempt: undefined }), false, "no attempt, no match"); + assert.equal(ok(SETTLED, { jobId: "j2" }), false); + assert.equal(ok({ ...SETTLED, startedAt: new Date(SINCE - RECORD_CLOCK_SKEW_MS - 1).toISOString() }), false, "a record of an older job under a reused id"); + assert.equal(ok({ ...SETTLED, startedAt: new Date(SINCE - RECORD_CLOCK_SKEW_MS).toISOString() }), true, "a producer clock ahead of the worker's, within the tolerance, still matches"); + assert.equal(RECORD_CLOCK_SKEW_MS, 5 * 60 * 1000); + assert.equal(ok({ ...SETTLED, startedAt: null }), false, "an undated record is not trusted when the job's age is known"); + assert.equal(ok(SETTLED, { since: undefined }), true, "no creation time: the id and attempt decide"); + assert.equal(ok(null), false); + assert.equal(ok("j1"), false); + assert.equal(recordSettlesAttempt(SETTLED), false, "no question, no match"); +}); + +test("recordVerdict names why a record was refused, from a fixed set", () => { + const v = (rec, over = {}) => recordVerdict(rec, { jobId: "j1", attempt: 1, since: SINCE, ...over }); + assert.equal(v(SETTLED), null); + assert.equal(v(null), "absent"); + assert.equal(v(undefined), "absent"); + assert.equal(v(UNREADABLE_RECORD), "unreadable"); + assert.equal(v("j1"), "unreadable"); + assert.equal(v([SETTLED]), "unreadable"); + assert.equal(v({ ...SETTLED, jobId: "j2" }), "id-mismatch"); + assert.equal(v(SETTLED, { attempt: 2 }), "other-attempt"); + assert.equal(v({ ...SETTLED, outcome: "failed" }), "failed"); + assert.equal(v({ ...SETTLED, outcome: undefined }), "failed"); + assert.equal(v({ ...SETTLED, startedAt: "2026-10-04T09:00:00.000Z" }), "older-than-job"); + assert.equal(v({ ...SETTLED, startedAt: "garbage" }), "older-than-job"); +}); + +test("makeSettledRecord: the local record first, the mirror only when the local one does not settle the attempt, and never a rejection", async () => { + const at = { attempt: 1, since: SINCE }; + const mirrorAsked = []; + const mirror = (rec) => async (id) => (mirrorAsked.push(id), rec); + assert.deepEqual(await makeSettledRecord({ readRecord: () => SETTLED, readMirrored: mirror(null) })("j1", at), SETTLED); + assert.deepEqual(mirrorAsked, [], "a local hit asks no mirror"); + assert.deepEqual(await makeSettledRecord({ readRecord: () => null, readMirrored: mirror(SETTLED) })("j1", at), SETTLED, "the host that meets the job need not be the one that ran it"); + assert.deepEqual( + await makeSettledRecord({ readRecord: () => ({ ...SETTLED, outcome: "failed" }), readMirrored: mirror(SETTLED) })("j1", at), + SETTLED, + "this host holds an earlier failed attempt, another host ran the one that finished", + ); + assert.equal(await makeSettledRecord({ readRecord: () => null, readMirrored: mirror({ ...SETTLED, attempt: 2 }) })("j1", at), null, "the mirror's record is held to the same test"); + assert.equal(await makeSettledRecord({ readRecord: () => null })("j1", at), null, "no mirror armed: a single host"); + assert.equal(await makeSettledRecord({ readRecord: () => null, readMirrored: async () => { throw new Error("ECONNREFUSED"); } })("j1", at), null, "a mirror fault is no record"); + assert.equal(await makeSettledRecord({ readRecord: () => { throw new Error("boom"); } })("j1", at), null, "a reader fault is no record"); +}); + +test("an unreadable MIRROR value is reported as unreadable from mirror, and never accepted", async () => { + const { readMirroredRecord } = await import("../src/run-mirror.mjs"); + const seen = []; + const lookup = makeSettledRecord({ readRecord: () => null, readMirrored: (id) => readMirroredRecord({ get: async () => "{garbage" }, id) }); + assert.equal(await lookup("j1", { attempt: 1, since: SINCE, onReject: (reason, source) => seen.push([reason, source]) }), null); + assert.deepEqual(seen, [["unreadable", "mirror"]]); +}); + +test("makeSettledRecord reports every record it found and refused, with its source, and never one it did not find", async () => { + const seen = []; + const onReject = (reason, source) => seen.push([reason, source]); + const at = { attempt: 1, since: SINCE, onReject }; + await makeSettledRecord({ readRecord: () => ({ ...SETTLED, attempt: 2 }), readMirrored: async () => UNREADABLE_RECORD })("j1", at); + assert.deepEqual(seen, [["other-attempt", "local"], ["unreadable", "mirror"]]); + seen.length = 0; + await makeSettledRecord({ readRecord: () => null, readMirrored: async () => null })("j1", at); + assert.deepEqual(seen, [], "absent everywhere is not a refusal"); + assert.deepEqual(await makeSettledRecord({ readRecord: () => ({ ...SETTLED, outcome: "failed" }), readMirrored: async () => SETTLED })("j1", at), SETTLED); + assert.deepEqual(seen, [["failed", "local"]], "a refused local record is reported even when the mirror then settles it"); + const throwing = makeSettledRecord({ readRecord: () => ({ ...SETTLED, attempt: 9 }) }); + assert.equal(await throwing("j1", { attempt: 1, since: SINCE, onReject: () => { throw new Error("log down"); } }), null, "a throwing reporter changes nothing"); +}); diff --git a/worker/test/run-mirror.test.mjs b/worker/test/run-mirror.test.mjs index 6fe30eb0..79324aba 100644 --- a/worker/test/run-mirror.test.mjs +++ b/worker/test/run-mirror.test.mjs @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; import { test } from "node:test"; -import { MIRROR_MAX_DAYS, RUNS_INDEX, hostsIn, makeRunMirror, mergeRuns, mirrorWindowMs, readMirroredRuns, runRecordKey } from "../src/run-mirror.mjs"; +import { UNREADABLE_RECORD } from "../src/run-history.mjs"; +import { MIRROR_MAX_DAYS, RUNS_INDEX, hostsIn, makeRunMirror, mergeRuns, mirrorWindowMs, readMirroredRecord, readMirroredRuns, runRecordKey } from "../src/run-mirror.mjs"; const DAY = 24 * 60 * 60 * 1000; @@ -236,3 +237,22 @@ test("hosts come from the RECORDS, not from where a row was read", () => { assert.deepEqual(hostsIn([record("a", "x", { host: "m2" }), record("b", "y", { host: "m1" }), record("c", "z", { host: "m1" })]), ["m1", "m2"]); assert.deepEqual(hostsIn([{ jobId: "a" }]), [], "a pre-#57 record names no host and invents none"); }); + +// ---- readMirroredRecord: the by-id read the lost-lock check falls back to on a fleet ---------------------- + +test("readMirroredRecord reads one record by its sanitized id; absent or unreachable is null, a value that is not a record is UNREADABLE_RECORD, never a throw", async () => { + const rec = { jobId: "repeat:n:1", outcome: "completed", attempt: 1 }; + const asked = []; + const redis = (value) => ({ get: async (k) => (asked.push(k), typeof value === "function" ? value() : value) }); + assert.deepEqual(await readMirroredRecord(redis(JSON.stringify(rec)), "repeat_n_1"), rec); + assert.deepEqual(asked, [runRecordKey("repeat_n_1")], "one GET of the writer's own key, no index scan"); + assert.equal(await readMirroredRecord(redis(null), "x"), null, "absent (expired, or never mirrored)"); + assert.equal(await readMirroredRecord(redis("{nope"), "x"), UNREADABLE_RECORD, "unparseable: there, but not a record"); + assert.equal(await readMirroredRecord(redis("[1]"), "x"), UNREADABLE_RECORD, "not a record"); + assert.equal(await readMirroredRecord(redis("null"), "x"), UNREADABLE_RECORD); + assert.equal(await readMirroredRecord(redis(""), "x"), UNREADABLE_RECORD, "an empty value"); + assert.equal(await readMirroredRecord(redis(() => Promise.reject(new Error("ECONNREFUSED"))), "x"), null, "unreachable"); + assert.equal(await readMirroredRecord({ get: () => new Promise(() => {}) }, "x", { timeoutMs: 5 }), null, "a server that never answers is bounded, not awaited forever"); + assert.equal(await readMirroredRecord(null, "x"), null, "no mirror armed"); + assert.equal(await readMirroredRecord(redis(JSON.stringify(rec)), ""), null, "no id"); +}); diff --git a/worker/test/start-wiring.test.mjs b/worker/test/start-wiring.test.mjs index 15569017..f96785f4 100644 --- a/worker/test/start-wiring.test.mjs +++ b/worker/test/start-wiring.test.mjs @@ -115,7 +115,7 @@ const IDLE_PODMAN_SERVICE = async () => ({ read: true, loaded: true, running: fa /** A running service's answer for the default test socket, started at `startedAtMs`. */ const RUNNING_PODMAN_SERVICE = (startedAtMs) => async () => ({ read: true, loaded: true, running: true, startedAtMs, environment: {}, environmentFiles: [], unitPaths: [], modules: [], manager: { read: true, environment: {}, modules: [] }, listen: ["/test.sock"] }); -async function runStart({ env = {}, makeAuth, makeHost, makeGitLabAuth, makeGitLabHost, makeReaper, makeLogSink, makeRecordWriter, makeLogReaper, makeSandboxReaper, makeSandboxNetworkSweeper, makeRetentionSweep, makeRunContainer, makeHostRegistry, makeScopeClaimSweeper, makeClaimSweeper, makeBackendRegistry, extraBackends, order, stops, now, authResolveTimeoutMs, resolveDockerEndpoint, readDaemonFacts, jobUserIdentity, bootImage, observationFs, loadConfig, readPodmanInfo, makePodmanReaper, makePodmanBackend, listRunningSandboxes, ensureJobsDir, ensureUnderAccountRoot, ensureSandboxDir, judgeValkey, readPodmanService, sleep, watchScopedLimits, watchProjects } = {}) { +async function runStart({ env = {}, makeAuth, makeHost, makeGitLabAuth, makeGitLabHost, makeReaper, makeLogSink, makeRecordWriter, makeLogReaper, makeSandboxReaper, makeSandboxNetworkSweeper, makeRetentionSweep, makeRunContainer, makeHostRegistry, makeScopeClaimSweeper, makeClaimSweeper, makeBackendRegistry, extraBackends, order, stops, now, authResolveTimeoutMs, resolveDockerEndpoint, readDaemonFacts, jobUserIdentity, bootImage, observationFs, loadConfig, readPodmanInfo, makePodmanReaper, makePodmanBackend, listRunningSandboxes, ensureJobsDir, ensureUnderAccountRoot, ensureSandboxDir, judgeValkey, readPodmanService, sleep, watchScopedLimits, watchProjects, readMirroredRecord } = {}) { const secretsResolverCalls = []; const calls = []; const registered = {}; @@ -283,6 +283,7 @@ async function runStart({ env = {}, makeAuth, makeHost, makeGitLabAuth, makeGitL ...(sleep ? { sleep } : {}), ...(watchScopedLimits ? { watchScopedLimits } : {}), ...(watchProjects ? { watchProjects } : {}), + ...(readMirroredRecord ? { readMirroredRecord } : {}), ...(authResolveTimeoutMs ? { authResolveTimeoutMs } : {}), ...(ensureJobsDir ? { ensureJobsDir } : {}), ...(ensureUnderAccountRoot ? { ensureUnderAccountRoot } : {}), @@ -2658,6 +2659,228 @@ test("with PI_ON_FAILURE unset, a terminal failure produces no on_failure line a assert.deepEqual(Object.keys(lines.find((l) => l.event === "job_completed")), ["event", "jobId", "outcome", "reason", "host"]); }); +// --- a job that finished, then lost its lock (DES-TERMINAL-COMMENTS-AND-FAILURE-HOOK) --------------------------------- + +/** The record the first run wrote before Valkey came back and BullMQ refused its completion. */ +const LOST_LOCK_CREATED = Date.parse("2026-10-04T10:00:00.000Z"); +const lostLockRecord = (id, over = {}) => ({ jobId: id, kind: "github", outcome: "completed", reason: null, exitCode: 0, budgetReserved: true, attempt: 1, startedAt: "2026-10-04T10:00:01.000Z", endedAt: "2026-10-04T10:04:00.000Z", ...over }); +/** The job as BullMQ emits it after the stall check's deferred failure: `attemptsMade` already counts this attempt. */ +const lostLockJob = (id, over = {}) => ({ id, data: { kind: "github", repo: "o/r", target: { type: "issue", number: 7 } }, attemptsMade: 1, finishedOn: 9, timestamp: LOST_LOCK_CREATED, ...over }); +const stalledError = async () => { + const { STALLED_FAILED_REASON } = await import("../src/index.mjs"); + return new Error(STALLED_FAILED_REASON); +}; + +async function lostLockBoot({ env = {}, readMirroredRecord } = {}) { + const posted = []; + const host = fakeHost({ postStatusComment: async (_job, _target, text) => void posted.push(text) }); + const makeAuth = async () => ({ mintToken: async () => "tok", selfId: 1, source: "gh" }); + const out = await runStart({ env: { PI_ON_FAILURE: "/nope/pi-notify-does-not-exist", ...env }, makeAuth, makeHost: () => host, ...(readMirroredRecord ? { readMirroredRecord } : {}) }); + const drive = async (job, err) => { + const from = bootLines.length; + out.handlers.failed(job, err); + await settleListeners(); + return parseLines(bootLines.slice(from)); + }; + return { ...out, posted, drive }; +} + +test("a terminal stall failure of a job whose record says this attempt completed: one lost-lock line, no failure comment, no page", { skip }, async () => { + const logsDir = tempDir("pi-lost-lock-"); + writeFileSync(join(logsDir, "gh-41.json"), `${JSON.stringify(lostLockRecord("gh-41"))}\n`); + const { drive, posted } = await lostLockBoot({ env: { PI_LOGS_DIR: logsDir } }); + const lines = await drive(lostLockJob("gh-41"), await stalledError()); + assert.deepEqual(posted, [], "the job completed: nobody is told it failed"); + assert.ok(!lines.some((l) => l.event === "on_failure"), "and nobody is paged"); + assert.ok(!lines.some((l) => l.event === "job_failed"), "job_failed is not the terminal line of a job that completed"); + const terminal = lines.filter((l) => l.event === "job_lost_lock_after_completion"); + assert.equal(terminal.length, 1, "exactly one terminal line (REQ-LOCAL-JOB-VISIBILITY)"); + assert.deepEqual(Object.keys(terminal[0]), ["event", "jobId", "outcome", "host"]); + assert.equal(terminal[0].jobId, "gh-41"); + assert.equal(terminal[0].outcome, "completed"); +}); + +test("the lost-lock check keeps the failure comment and the page unless the record settles THIS attempt, for THIS reason", { skip }, async () => { + const logsDir = tempDir("pi-lost-lock-"); + const write = (id, over) => writeFileSync(join(logsDir, `${id}.json`), `${JSON.stringify(lostLockRecord(id, over))}\n`); + write("gh-a", { attempt: 1 }); + write("gh-f", { outcome: "failed" }); + write("gh-o"); + write("gh-old", { startedAt: "2026-10-04T09:00:00.000Z" }); + const { drive, posted } = await lostLockBoot({ env: { PI_LOGS_DIR: logsDir } }); + const rejected = []; + const cases = [ + ["a record of an earlier attempt", lostLockJob("gh-a", { attemptsMade: 2 }), await stalledError()], + ["a failed record", lostLockJob("gh-f"), await stalledError()], + ["any other failure reason", lostLockJob("gh-o"), new Error("boom")], + ["a record older than the job (a reused id)", lostLockJob("gh-old"), await stalledError()], + ["no record at all", lostLockJob("gh-none"), await stalledError()], + ]; + for (const [label, job, err] of cases) { + posted.length = 0; + const lines = await drive(job, err); + assert.deepEqual(posted, ["Failed: an error stopped this job and it will not be retried further. Ask the operator to check the worker log."], `${label}: comments`); + assert.equal(lines.filter((l) => l.event === "on_failure").length, 1, `${label}: pages`); + assert.equal(lines.filter((l) => l.event === "job_failed").length, 1, `${label}: job_failed is the terminal line`); + assert.ok(!lines.some((l) => l.event === "job_lost_lock_after_completion"), `${label}: no lost-lock line`); + rejected.push(lines.filter((l) => l.event === "job_lost_lock_record_rejected").map((l) => [l.jobId, l.reason, l.source])); + } + // Why the record did not suppress the comment, said with a fixed token (never for a reason the check does not + // apply to, and never for a record that is not there). + assert.deepEqual(rejected, [[["gh-a", "other-attempt", "local"]], [["gh-f", "failed", "local"]], [], [["gh-old", "older-than-job", "local"]], []]); +}); + +test("a producer clock ahead of the worker's, within RECORD_CLOCK_SKEW_MS, still suppresses; an unreadable record is named", { skip }, async () => { + const logsDir = tempDir("pi-lost-lock-"); + // The record started 4 minutes BEFORE the job's creation stamp: two hosts' clocks, not an older job. + writeFileSync(join(logsDir, "gh-skew.json"), `${JSON.stringify(lostLockRecord("gh-skew", { startedAt: new Date(LOST_LOCK_CREATED - 4 * 60_000).toISOString() }))}\n`); + writeFileSync(join(logsDir, "gh-bad.json"), "{not json"); + const { drive, posted } = await lostLockBoot({ env: { PI_LOGS_DIR: logsDir } }); + const skewed = await drive(lostLockJob("gh-skew"), await stalledError()); + assert.deepEqual(posted, []); + assert.equal(skewed.filter((l) => l.event === "job_lost_lock_after_completion").length, 1); + const bad = await drive(lostLockJob("gh-bad"), await stalledError()); + assert.equal(posted.length, 1, "an unreadable record keeps the failure path"); + assert.deepEqual(bad.filter((l) => l.event === "job_lost_lock_record_rejected").map((l) => [l.reason, l.source]), [["unreadable", "local"]]); +}); + +test("a lost-lock job whose record is a paid policy stop pages as that stop, once, and comments nothing more", { skip }, async () => { + const logsDir = tempDir("pi-lost-lock-"); + writeFileSync(join(logsDir, "gh-p.json"), `${JSON.stringify(lostLockRecord("gh-p", { outcome: "policy", reason: "runner-policy", exitCode: 2 }))}\n`); + writeFileSync(join(logsDir, "gh-q.json"), `${JSON.stringify(lostLockRecord("gh-q", { outcome: "policy", reason: "over-budget", budgetReserved: false }))}\n`); + const { drive, posted } = await lostLockBoot({ env: { PI_LOGS_DIR: logsDir } }); + const paid = await drive(lostLockJob("gh-p"), await stalledError()); + assert.deepEqual(posted, [], "the processor already posted the stop's own sentence before it returned"); + const pages = paid.filter((l) => l.event === "on_failure"); + assert.equal(pages.length, 1, "the page the completed listener never got to send"); + assert.deepEqual(paid.find((l) => l.event === "job_lost_lock_after_completion"), { event: "job_lost_lock_after_completion", jobId: "gh-p", outcome: "policy", reason: "runner-policy", host: paid.find((l) => l.event === "job_lost_lock_after_completion").host }); + const free = await drive(lostLockJob("gh-q"), await stalledError()); + assert.ok(!free.some((l) => l.event === "on_failure"), "a free refusal pages nobody, exactly as the completed listener decides"); + assert.deepEqual(posted, []); +}); + +test("on a fleet the lost-lock check falls back to the run mirror, and a single host never asks one", { skip }, async () => { + const logsDir = tempDir("pi-lost-lock-"); + const asked = []; + const fromMirror = async (_redis, sanitizedId) => (asked.push(sanitizedId), lostLockRecord("repeat:n:5")); + const fleet = await lostLockBoot({ env: { PI_LOGS_DIR: logsDir, PI_WORKER_NAME: `ll-${process.pid}` }, readMirroredRecord: fromMirror }); + const lines = await fleet.drive(lostLockJob("repeat:n:5"), await stalledError()); + assert.deepEqual(asked, ["repeat_n_5"], "this host has no file for the job; the mirror is asked by the writer's own sanitized id"); + assert.deepEqual(fleet.posted, []); + assert.equal(lines.filter((l) => l.event === "job_lost_lock_after_completion").length, 1); + assert.ok(!lines.some((l) => l.event === "on_failure" || l.event === "job_failed")); + + asked.length = 0; + const single = await lostLockBoot({ env: { PI_LOGS_DIR: logsDir }, readMirroredRecord: fromMirror }); + const alone = await single.drive(lostLockJob("repeat:n:5"), await stalledError()); + assert.deepEqual(asked, [], "no worker name, no mirror, no Valkey read"); + assert.equal(single.posted.length, 1, "and with no record anywhere the failure path stands"); + assert.equal(alone.filter((l) => l.event === "job_failed").length, 1); +}); + +test("LIVE: a job whose lock is lost while its processor runs is run once, neither commented failed nor paged, plain and scheduled", { skip }, async () => { + const { Queue, Worker } = await import("bullmq"); + const { parseConnection, makeRedisClient } = await import("../src/connection.mjs"); + const { makeProcessor } = await import("../src/index.mjs"); + const { makeRecordWriter: realRecordWriter } = await import("../src/run-history.mjs"); + const logsDir = tempDir("pi-lost-lock-live-"); + const posted = []; + const host = fakeHost({ postStatusComment: async (_job, _target, text) => void posted.push(text) }); + const makeAuth = async () => ({ mintToken: async () => "tok", selfId: 1, source: "gh" }); + // The REAL record writer and the REAL listeners and lookup, from a real boot; only the queue below is this test's. + const { captured, handlers } = await runStart({ env: { PI_LOGS_DIR: logsDir, PI_ON_FAILURE: "/nope/pi-notify-does-not-exist" }, makeAuth, makeHost: () => host, makeRecordWriter: realRecordWriter }); + + const name = `pi-jobs-lostlock-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; + const conn = parseConnection(VALKEY_URL); + const redis = makeRedisClient(VALKEY_URL); + const queue = new Queue(name, { connection: conn }); + const fixedNow = new Date(Date.UTC(2099, 0, 5, 12, 0, 0)); + const budgetKey = "budget:2099-01-05"; + const runs = []; + const processorLogs = []; + const events = []; + let current = null; + let worker; + const waitFor = async (fn, ms = 15_000) => { + const deadline = Date.now() + ms; + for (;;) { + const v = await fn(); + if (v) return v; + if (Date.now() > deadline) throw new Error("waitFor timed out"); + await new Promise((r) => setTimeout(r, 50)); + } + }; + try { + await redis.del(budgetKey); + const base = makeProcessor({ + cancelJob: () => {}, + stopContainer: () => {}, + redis, + getSettings: () => ({ provider: "anthropic", model: "m", maxTurns: 30, dailyCap: 100, concurrency: 1 }), + recordRun: captured.recordRun, + settledRecord: captured.settledRecord, + deps: { + mintToken: async () => null, + isDefaultBranchProtected: async () => true, + prepareWorkspace: async () => ({ workspace: "/tmp/ws", jobDir: "/tmp/job" }), + // The container "runs", and while it does the queue loses the job's lock: the state a Valkey outage + // longer than the renewal window leaves (b1's repro), made deterministic by deleting the key. + runContainer: async () => { + const id = current; + runs.push(id); + // A second run of one id (the defect) returns at once, so it fails on the run count below, not on a timeout. + if (runs.filter((r) => r === id).length > 1) return { code: 0, aborted: false }; + await redis.del(`bull:${name}:${id}:lock`); + await waitFor(async () => !(await redis.lrange(`bull:${name}:active`, 0, -1)).includes(id)); + return { code: 0, aborted: false }; + }, + cleanup: async () => {}, + comment: async () => {}, + log: (event, fields) => processorLogs.push({ event, ...fields }), + now: fixedNow, + }, + }); + const processor = (job, token, signal) => ((current = job.id), base(job, token, signal)); + worker = new Worker(name, processor, { connection: { ...conn, maxRetriesPerRequest: null }, concurrency: 1, maxStalledCount: 0, stalledInterval: 250 }); + worker.on("error", () => {}); + const from = bootLines.length; + worker.on("failed", (job, err) => (events.push(["failed", job.id]), handlers.failed(job, err))); + worker.on("completed", (job, result) => (events.push(["completed", job.id]), handlers.completed(job, result))); + + // PLAIN: the stall check stores a deferred failure, and the next pickup fails the job without the processor. + await queue.add("local", { kind: "local", folder: "/proj", flow: "tidy", provider: "anthropic", model: "m", maxTurns: 7 }, { jobId: "manual:lostlock:1" }); + await waitFor(() => events.some(([e, id]) => e === "failed" && id === "manual:lostlock:1")); + // Either terminal line, so a regression fails on the assertion below rather than on a timeout. + const plainEnd = await waitFor(() => parseLines(bootLines.slice(from)).find((l) => (l.event === "job_lost_lock_after_completion" || l.event === "job_failed") && l.jobId === "manual:lostlock:1")); + assert.equal(plainEnd.event, "job_lost_lock_after_completion", "the plain job's terminal line says it completed and lost its lock"); + assert.deepEqual(runs, ["manual:lostlock:1"], "the plain job ran once"); + + // SCHEDULED: the stall check skips the failure for a live scheduler and moves the job back to wait. + await queue.upsertJobScheduler("lostlock", { every: 3_600_000 }, { name: "local", data: { kind: "local", folder: "/proj", flow: "tidy", provider: "anthropic", model: "m", maxTurns: 7, trigger: { id: "lostlock", pattern: "every" } } }); + await waitFor(() => events.some(([e, id]) => e === "completed" && id.startsWith("repeat:lostlock:"))); + const schedId = events.find(([e, id]) => e === "completed" && id.startsWith("repeat:lostlock:"))[1]; + assert.deepEqual(runs, ["manual:lostlock:1", schedId], "the scheduled job ran once: its re-delivery was returned, not run (CONST-RETRY-INFRA-ONLY)"); + assert.ok(processorLogs.some((l) => l.event === "job_lost_lock_after_completion" && l.jobId === schedId && l.outcome === "completed")); + + await settleListeners(); + const lines = parseLines(bootLines.slice(from)); + assert.deepEqual(posted, [], "no failure comment for either job"); + assert.ok(!lines.some((l) => l.event === "comment"), "nothing through the local comment fallthrough either"); + assert.ok(!lines.some((l) => l.event === "on_failure"), "nobody paged"); + assert.ok(!lines.some((l) => l.event === "job_failed"), "no job_failed line"); + assert.deepEqual(lines.filter((l) => l.event === "job_completed").map((l) => [l.jobId, l.outcome]), [[schedId, "completed"]], "the scheduled job ends completed, as its record says"); + const record = JSON.parse(readFileSync(join(logsDir, `${schedId.replace(/[^A-Za-z0-9._-]/g, "_")}.json`), "utf8")); + assert.equal(record.attempt, 1, "the first run's record stands, not a second run's"); + } finally { + await queue.removeJobScheduler("lostlock").catch(() => {}); + await worker?.close(true).catch(() => {}); + await queue.obliterate({ force: true }).catch(() => {}); + await queue.close().catch(() => {}); + await redis.del(budgetKey).catch(() => {}); + redis.disconnect(); + } +}); + test("a terminal LOCAL failure comments into the log fallthrough with the fixed sentence, and posts nothing to any forge (review finding)", { skip }, async () => { const forgeCalls = []; const host = fakeHost({ postStatusComment: async () => void forgeCalls.push(1) });