diff --git a/.planning/STATE.md b/.planning/STATE.md index b9a6875..480aefd 100644 --- a/.planning/STATE.md +++ b/.planning/STATE.md @@ -28,7 +28,7 @@ See: .planning/PROJECT.md (updated 2026-05-31) Phase: Parity / Phase 1 — Game-Type-Aware Statistics — COMPLETE (5/5 plans + review fixes) Plan: 01-01..01-05 done; code-review BLOCK→fixed→APPROVE Status: Phase implemented + reviewed; landing to master. Migration 0008 (game_type dimension + nullable rotation_id + NULLS NOT DISTINCT + is_show) & 0009 (stale NULL-type cleanup); set-based classification; per-type + all-time recalc; per-type legacy-export/parity-sql; audit path made game-type-correct. pnpm verify green, 100% cov, OpenAPI diff empty. Deferred: large-bucket perf pass (review findings 3/4/5 + parity-driver flag). -Last activity: 2026-06-15 — Completed quick task 260615-u06: F9 excludePlayers (player-leaderboard exclusion, squad-parity-safe) +Last activity: 2026-06-15 — Landed quick tasks 260615-u06 (F9 excludePlayers) and 260615-v6m (F5 orphaned-published reconciler) to master via PRs #22/#23; pnpm verify green, 100% cov ## Performance Metrics @@ -134,6 +134,7 @@ Decisions are logged in PROJECT.md Key Decisions table. Recent decisions affecti | 260614-fw2 | Set-based canonical player identity resolution in per-rotation recalc (behavior-preserving) | 2026-06-14 | fa7c54b | [260614-fw2-perf-set-based-canonical-player-identity](./quick/260614-fw2-perf-set-based-canonical-player-identity/) | | 260614-r9k | Guard all-time recalc against NULL replay_timestamp (toISOString crash) | 2026-06-14 | b0275e0 | [260614-r9k-recalc-null-timestamp-guard](./quick/260614-r9k-recalc-null-timestamp-guard/) | | 260615-u06 | F9 — apply the legacy excludePlayers exclusion to the player leaderboard | 2026-06-15 | b80a235 | [260615-u06-f9-excludeplayers-apply-the-legacy-exclu](./quick/260615-u06-f9-excludeplayers-apply-the-legacy-exclu/) | +| 260615-v6m | F5 — reconciler re-queues orphaned `published` parse_jobs (self-healing ingest) | 2026-06-15 | f4e0c1b | [260615-v6m-f5-reconciler-for-orphaned-published-par](./quick/260615-v6m-f5-reconciler-for-orphaned-published-par/) | ## Deferred Items diff --git a/.planning/quick/260615-v6m-f5-reconciler-for-orphaned-published-par/260615-v6m-PLAN.md b/.planning/quick/260615-v6m-f5-reconciler-for-orphaned-published-par/260615-v6m-PLAN.md new file mode 100644 index 0000000..ab84ea4 --- /dev/null +++ b/.planning/quick/260615-v6m-f5-reconciler-for-orphaned-published-par/260615-v6m-PLAN.md @@ -0,0 +1,201 @@ +--- +phase: quick-260615-v6m +plan: 01 +type: execute +wave: 1 +depends_on: [] +files_modified: + - src/infra/db/migrations/0012_parse_job_history_reconciled_action.sql + - src/modules/ingest/types.ts + - src/modules/ingest/repository/repository.ts + - src/modules/ingest/reconciler.ts + - src/modules/ingest/reconciler.test.ts + - src/modules/ingest/runtime.ts + - src/modules/ingest/runtime.test.ts + - src/modules/ingest/repository/tests/postgres.test.ts + - src/config/env.ts + - src/server.ts + - src/test/app.test.ts +autonomous: true +requirements: [F5] +must_haves: + truths: + - "A parse_job stuck in status=published with published_at older than the staleness threshold is re-queued (status→queued) by the reconcile interval task, after which the existing publisher re-publishes it." + - "A freshly-published job (published_at within the threshold) and a queued/retryable/succeeded/failed job are left untouched by reconciliation." + - "Each reclaim is recorded in parse_job_history with a new `reconciled` action (status_from=published, status_to=queued)." + - "Re-running reconciliation is idempotent: a job already re-queued is not reclaimed again, and a late parser result still supersedes correctly via the existing parser_results path." + - "Staleness threshold and reconcile batch size are configurable via env and surfaced on config.ingest, wired through createIngestRuntime and server.ts." + artifacts: + - path: "src/infra/db/migrations/0012_parse_job_history_reconciled_action.sql" + provides: "Adds 'reconciled' value to parse_job_history_action enum" + contains: "alter type parse_job_history_action add value" + - path: "src/modules/ingest/reconciler.ts" + provides: "ParseJobReconciler class mirroring ParseJobPublisher shape (repository contract + observer/logger + noop defaults)" + min_lines: 40 + - path: "src/modules/ingest/repository/repository.ts" + provides: "reclaimStalePublishedJobs repository method (FOR UPDATE SKIP LOCKED re-queue + history)" + contains: "reclaimStalePublishedJobs" + key_links: + - from: "src/modules/ingest/runtime.ts" + to: "src/modules/ingest/reconciler.ts" + via: "createIngestRuntime constructs ParseJobReconciler and runs it as a third IntervalTask" + pattern: "ParseJobReconciler" + - from: "src/modules/ingest/reconciler.ts" + to: "src/modules/ingest/repository/repository.ts" + via: "reconciler calls repository.reclaimStalePublishedJobs" + pattern: "reclaimStalePublishedJobs" + - from: "src/server.ts" + to: "src/config/env.ts" + via: "createIngestRuntime receives staleAfterMs and reconcileBatchSize from config.ingest" + pattern: "config.ingest" +--- + + +Add a server-2 reconciler that recovers orphaned `published` parse_jobs whose parser result never returned. Stale `published` jobs (where `published_at` is older than a configurable threshold) are re-queued so the existing `ParseJobPublisher` re-publishes them. Closes F5 (75 orphaned jobs on staging, oldest 33 days; `listPublishableJobs` only selects `queued`/`retryable`, so a `published` orphan is never re-published). + +Purpose: A `published` job whose downstream parser ack was lost is currently a permanent dead-end (attempts stays 1, queue empty, no recovery path). This restores self-healing for the ingest pipeline. + +Output: A new `0012` migration adding a `reconciled` history action, a `reclaimStalePublishedJobs` repository method, a `ParseJobReconciler` class, a third reconcile `IntervalTask` in `createIngestRuntime`, two new env vars surfaced on `config.ingest`, and full coverage (real-pg repository integration + factory-injection reconciler unit + updated runtime/app/env constructors). + +Scope boundary: The parser-side root cause (ack-after-persist in replay-parser-2) is a SEPARATE change and is OUT OF SCOPE here. This plan only adds the server-2 recovery loop. + + + +@$HOME/.claude/gsd-core/workflows/execute-plan.md +@$HOME/.claude/gsd-core/templates/summary.md + + + +@.planning/PROJECT.md +@.planning/STATE.md +@AGENTS.md + +# Project skills — read before writing any server-2 code (the agent_skills config is stale; read these directly) +@.claude/skills/solidstats-server-ts-conventions/SKILL.md +@.claude/skills/solidstats-server-ts-tests/SKILL.md +@.claude/skills/solidstats-shared-backend-ts-standards/SKILL.md +@.claude/skills/solidstats-shared-ts-standards/SKILL.md + +# Mirror these shapes exactly — do not re-derive the lifecycle +@src/modules/ingest/publisher.ts +@src/modules/ingest/runtime.ts +@src/infra/runtime/interval-task.ts +@src/modules/ingest/repository/repository.ts +@src/modules/ingest/types.ts +@src/config/env.ts +@src/server.ts +@src/modules/ingest/runtime.test.ts +@src/modules/ingest/repository/tests/postgres.test.ts +@src/infra/db/migrations/0003_parse_job_history.sql + + + + + + Task 1: Migration + repository reclaim method + history action + src/infra/db/migrations/0012_parse_job_history_reconciled_action.sql, src/modules/ingest/types.ts, src/modules/ingest/repository/repository.ts + +Add the data layer for reconciliation (F5). + +1. Create migration `0012_parse_job_history_reconciled_action.sql`: a single statement `alter type parse_job_history_action add value 'reconciled';`. `ALTER TYPE ... ADD VALUE` runs inside the migrate runner's per-file transaction on PG17 (the runner wraps each file in begin/commit — see migrate.ts), so no special handling is needed. Do not modify the existing `0003` enum definition (the migrate runner checksums applied files and throws on drift). + +2. In `types.ts`, add `"reconciled"` to the `ParseJobHistoryAction` union (place it adjacent to the other operator-driven transitions, keeping the existing member order otherwise intact). + +3. In `repository.ts`, add a public method `reclaimStalePublishedJobs(staleAfterMs: number, batchSize: number): Promise` on `PgIngestRepository`. Implement it as ONE `withTransaction` block using a CTE that claims stale rows with `for update skip locked` and re-queues them in a single round-trip, mirroring the `claimPendingStagingRecords` CTE pattern already in this file. The claim CTE selects `id` from `parse_jobs` where `status = 'published'` and `published_at < now() - ($1 * interval '1 millisecond')`, ordered by `published_at, id`, limited to `$2`, `for update skip locked`. The update sets `status = 'queued', published_at = null, started_at = null, finished_at = null, error = null, updated_at = now()` from the claimed CTE and uses `returning parse_jobs.*` — mirror the reset column list from `retryParseJob`. Map rows via the existing `mapParseJobRow`. Then, for each reclaimed row, call the existing `recordParseJobHistory(client, ...)` helper with `action: "reconciled"`, `statusFrom: "published"`, `statusTo: "queued"`, `details: { stale_after_ms: staleAfterMs }` (snake_case detail keys per existing rows; the file already has `eslint-disable camelcase`). Keep all of it inside the one transaction so the re-queue and history insert commit atomically. Use `interval '1 millisecond'` multiplication for the threshold rather than constructing interval strings, so the comparison stays parameterized. + +Idempotency note: because the claim filters on `status = 'published'`, a job already re-queued by a prior pass is no longer eligible; a concurrent late parser result transitions the job to `succeeded` (terminal) and is likewise excluded. `skip locked` lets multiple replicas claim disjoint batches. + + + pnpm exec tsc --noEmit + + Migration 0012 exists and adds the `reconciled` enum value; `ParseJobHistoryAction` includes `"reconciled"`; `reclaimStalePublishedJobs` exists on `PgIngestRepository` re-queuing only stale `published` rows with `for update skip locked`, resetting published_at/started_at/finished_at/error, and recording a `reconciled` history row per reclaimed job; `tsc --noEmit` passes. + + + + Task 2: ParseJobReconciler class + factory-injection unit test + src/modules/ingest/reconciler.ts, src/modules/ingest/reconciler.test.ts + + - Given a repository stub whose `reclaimStalePublishedJobs` returns N records, calling `reconcileStale({ staleAfterMs, batchSize })` calls the repository once with those exact args and returns the reclaimed records. + - For each reclaimed job, the observer's `jobReconciled(job)` is called and the logger's `info` is called once with job bindings. + - When the repository returns an empty array, no observer/logger job calls happen and the result is empty. + - Default noop observer/logger are used when runtimeOptions omit them (constructing with only the repository does not throw and reconcile still returns reclaimed records) — covers the noop branches for 100% coverage. + + +Create `ParseJobReconciler` in `reconciler.ts`, mirroring `ParseJobPublisher` in publisher.ts exactly: a narrow `ParseJobReconcilerRepository` contract interface declaring only `reclaimStalePublishedJobs(staleAfterMs: number, batchSize: number): Promise`; a `ParseJobReconcilerOptions` ({ batchSize: number; staleAfterMs: number }); a `ParseJobReconcilerObserver` with a `jobReconciled(job: ParseJobRecord): void` method; a `ParseJobReconcilerLogger` with `error`/`info`; a `ParseJobReconcilerRuntimeOptions` ({ logger?, observer? }); module-level `noopObserver`/`noopLogger` defaults using the same `void`-the-args style; and the class with a constructor `(repository, runtimeOptions = {})` that falls back to the noop defaults. + +The single public method `reconcileStale(options: ParseJobReconcilerOptions): Promise` calls `this.repository.reclaimStalePublishedJobs(options.staleAfterMs, options.batchSize)`, then for each returned job calls `this.observer.jobReconciled(job)` and `this.logger.info(bindings, "parse job reconciled")` with a small `parseJobLogBindings`-style helper (job_id, replay_id, published_at), and returns the reclaimed array. No try/catch around the repository call (the `IntervalTask` wrapper already logs task failures — match how `publishQueued` lets failures propagate where appropriate). Keep the file under the `/* eslint-disable camelcase */` header like publisher.ts if you emit snake_case log bindings. + +Write `reconciler.test.ts` as a factory/constructor-injection unit test (no Fastify, no mock framework beyond `vi.fn`), following the per-layer testing map (service-equivalent logic = unit). Construct the class with a stub satisfying `ParseJobReconcilerRepository` and `vi.fn` observer/logger; cover every behavior above including the noop-default branches. Aim for full reachable-source coverage of reconciler.ts. + + + pnpm test -- src/modules/ingest/reconciler.test.ts + + `ParseJobReconciler` exists mirroring the publisher shape with noop observer/logger defaults and a `reconcileStale` method; `reconciler.test.ts` passes and exercises the populated path, the empty path, the observer/logger calls, and the noop-default construction. + + + + Task 3: Wire reconcile IntervalTask + config + update existing constructors + src/modules/ingest/runtime.ts, src/config/env.ts, src/server.ts, src/modules/ingest/runtime.test.ts, src/test/app.test.ts, src/modules/ingest/repository/tests/postgres.test.ts + +Wire the reconciler into the runtime and config, and update every constructor/test the new fields touch. + +1. `env.ts`: add two envalid vars in the existing `cleanEnv` block — `PARSE_JOB_STALE_AFTER_MS: num({ default: 3_600_000 })` (1 hour — generous, well above worst-case parse + queue-drain) and `PARSE_JOB_RECONCILE_BATCH_SIZE: num({ default: 25 })`. Surface them on `config.ingest` as `staleAfterMs` and `reconcileBatchSize`, and add both to the `AppConfig['ingest']` interface. `redactConfigForLogs` already spreads `config.ingest` wholesale, so no redaction change is needed. + +2. `runtime.ts`: extend `IngestRuntimeOptions` with `reconcileBatchSize: number` and `staleAfterMs: number`. In `createIngestRuntime`, construct `const reconciler = new ParseJobReconciler(options.repository, { logger: options.logger });` and add a third `IntervalTask` named `"parse-job-reconciler"` reusing `options.pollIntervalMs`, whose `task` calls `await reconciler.reconcileStale({ batchSize: options.reconcileBatchSize, staleAfterMs: options.staleAfterMs });`. Add `reconcileTask` to both the `start()` (start it alongside the others) and `close()` (`Promise.all` list) of the returned runtime. Import `ParseJobReconciler` from `./reconciler.js`. Note: `options.repository` (a `PgIngestRepository`) already satisfies the narrow `ParseJobReconcilerRepository` contract once Task 1 lands. + +3. `server.ts`: pass `reconcileBatchSize: config.ingest.reconcileBatchSize` and `staleAfterMs: config.ingest.staleAfterMs` into the existing `createIngestRuntime({ ... })` call. + +4. `runtime.test.ts`: add `reconcileBatchSize` and `staleAfterMs` to BOTH `createIngestRuntime({ ... })` option objects, add `reclaimStalePublishedJobs: vi.fn(async () => [])` to `repositoryDouble()`, and extend the existing started-loop assertions to also assert `repository.reclaimStalePublishedJobs` was called with the configured args (mirror the existing `listPublishableJobs`/`claimPendingStagingRecords` assertions, including the post-close "called once" assertion). + +5. `app.test.ts`: add `reconcileBatchSize: 25` and `staleAfterMs: 3_600_000` to the inline `ingest: { ... }` config object at ~line 83 so the `AppConfig` shape stays satisfied. + +6. `postgres.test.ts` (real-pg integration): add a new `it(...)` covering `reclaimStalePublishedJobs`. Insert a replay + parse job, drive it to `published` via the existing `repository.markJobPublished(...)`, then backdate its `published_at` with a raw `update parse_jobs set published_at = now() - interval '2 hours' where id = $1` so it is stale; also create a second fresh-published job (not backdated) and a third left `queued`. Call `repository.reclaimStalePublishedJobs(3_600_000, 10)` and assert: the stale job is returned and now `status = 'queued'` with `published_at`/`started_at`/`finished_at` reset to null; the fresh-published job is still `published`; the queued job is untouched; `listParseJobHistory(staleJobId)` includes a `reconciled` entry with `statusFrom: "published"`, `statusTo: "queued"`. Then assert idempotency: a second `reclaimStalePublishedJobs(3_600_000, 10)` returns an empty array (the re-queued job is no longer `published`). Reuse the existing `insertStaging`/`withTransaction`/`createReplay`/`createParseJob` helpers in that file. + +Run the full verify suite (lint + typecheck + tests + coverage) to honor the 100% reachable-source coverage gate. + + + pnpm verify + + `config.ingest` exposes `staleAfterMs` (default 3_600_000) and `reconcileBatchSize` (default 25); `createIngestRuntime` runs a third `parse-job-reconciler` IntervalTask that calls `reconcileStale` with the configured args and is started/closed with the others; server.ts passes both new fields; runtime.test.ts and app.test.ts construct the updated options/config; postgres.test.ts proves stale `published`→`queued` re-queue + `reconciled` history + fresh/queued untouched + idempotent second pass; `pnpm verify` is green with 100% reachable-source coverage. + + + + + +## Trust Boundaries + +| Boundary | Description | +|----------|-------------| +| reconcile task → parse_jobs table | Background loop mutates job status; concurrent with the publisher, parser-result consumer, and other replicas. | +| env → config.ingest | Operator-supplied staleness/batch values cross into the reconcile loop. | + +## STRIDE Threat Register + +| Threat ID | Category | Component | Disposition | Mitigation Plan | +|-----------|----------|-----------|-------------|-----------------| +| T-v6m-01 | Tampering / race | reclaimStalePublishedJobs concurrent with parser-result consumer or another replica | mitigate | `for update skip locked` claims disjoint batches; re-queue and history insert are one transaction; a late parser result still supersedes via the existing terminal-state guard in recordParserCompleted (status becomes succeeded → no longer `published`, so never double-reclaimed). | +| T-v6m-02 | Denial of Service | too-low staleAfterMs re-queuing in-flight jobs into a publish/parse storm | accept | Default is generous (1h, well above worst-case parse+queue-drain); value is operator-controlled config, not user input. Documented intent in the env default. | +| T-v6m-03 | Repudiation | reclaim transition not auditable | mitigate | Every reclaim writes a `reconciled` parse_job_history row (status_from=published, status_to=queued, stale_after_ms detail). | +| T-v6m-SC | Tampering | npm/pip/cargo installs | mitigate | No new dependencies are added by this plan (pure repository/runtime/config change); no package-install tasks, so the legitimacy gate is not engaged. | + + + +- `pnpm exec tsc --noEmit` passes (types/enum union consistent). +- `pnpm test -- src/modules/ingest/reconciler.test.ts` passes (reconciler unit). +- Real-pg integration in postgres.test.ts proves stale `published`→`queued`, `reconciled` history, fresh/queued untouched, and idempotent second pass. +- `pnpm verify` is green end-to-end including the 100% reachable-source coverage gate. +- OpenAPI is unaffected (no route/schema change); no contract diff expected. + + + +- A stale `published` parse_job is re-queued by the reconcile IntervalTask and subsequently re-published by the existing publisher; orphans can no longer accumulate indefinitely. +- Fresh-published / queued / terminal jobs are never reclaimed. +- Every reclaim is audited via a `reconciled` parse_job_history row. +- Staleness threshold and batch size are configurable env-driven settings on config.ingest. +- Parser-side ack-after-persist remains explicitly out of scope. +- `pnpm verify` green with 100% reachable-source coverage; clean git tree on completion. + + + +Create `.planning/quick/260615-v6m-f5-reconciler-for-orphaned-published-par/260615-v6m-SUMMARY.md` when done. + diff --git a/.planning/quick/260615-v6m-f5-reconciler-for-orphaned-published-par/260615-v6m-SUMMARY.md b/.planning/quick/260615-v6m-f5-reconciler-for-orphaned-published-par/260615-v6m-SUMMARY.md new file mode 100644 index 0000000..6cb40e2 --- /dev/null +++ b/.planning/quick/260615-v6m-f5-reconciler-for-orphaned-published-par/260615-v6m-SUMMARY.md @@ -0,0 +1,90 @@ +--- +phase: quick-260615-v6m +plan: 01 +subsystem: ingest +status: complete +tags: [reconciler, parse-jobs, ingest, self-healing, F5] +requires: + - parse_jobs lifecycle + parse_job_history audit (migration 0003) + - ParseJobPublisher / IntervalTask ingest runtime +provides: + - reclaimStalePublishedJobs repository method + - ParseJobReconciler class + - parse-job-reconciler IntervalTask + - config.ingest.staleAfterMs / reconcileBatchSize +affects: + - src/modules/ingest + - src/config/env.ts + - src/server.ts +tech-stack: + added: [] + patterns: + - "FOR UPDATE SKIP LOCKED claim CTE re-queue in one withTransaction round-trip" + - "narrow repository contract + noop observer/logger defaults (mirrors ParseJobPublisher)" +key-files: + created: + - src/infra/db/migrations/0012_parse_job_history_reconciled_action.sql + - src/modules/ingest/reconciler.ts + - src/modules/ingest/reconciler.test.ts + modified: + - src/modules/ingest/types.ts + - src/modules/ingest/repository/repository.ts + - src/modules/ingest/runtime.ts + - src/modules/ingest/runtime.test.ts + - src/config/env.ts + - src/server.ts + - src/test/app.test.ts + - src/modules/ingest/repository/tests/postgres.test.ts + - src/infra/queue/rabbitmq.test.ts + - src/infra/storage/client.test.ts +decisions: + - "noopLogger.error is a contract stub never invoked by reconcileStale (failures propagate to the IntervalTask wrapper); covered with a narrow v8-ignore rather than a fabricated error path." +metrics: + tasks: 3 + files: 13 + completed: "2026-06-15" +--- + +# Phase quick-260615-v6m Plan 01: F5 Reconciler for Orphaned Published Parse Jobs Summary + +A self-healing reconcile loop that re-queues `parse_jobs` stuck in `status=published` (parser ack lost) so the existing publisher re-publishes them, closing F5 (orphaned jobs accumulating indefinitely because `listPublishableJobs` only selects `queued`/`retryable`). + +## What Was Built + +- **Task 1 — data layer:** migration `0012` adds a `reconciled` value to the `parse_job_history_action` enum; `ParseJobHistoryAction` union extended; `reclaimStalePublishedJobs(staleAfterMs, batchSize)` re-queues stale `published` rows via a `FOR UPDATE SKIP LOCKED` claim CTE in one `withTransaction` round-trip (resetting `published_at`/`started_at`/`finished_at`/`error`), and writes a `reconciled` history row (`status_from=published`, `status_to=queued`, `stale_after_ms` detail) per reclaimed job. Threshold compared via parameterized `($1 * interval '1 millisecond')`. +- **Task 2 (TDD) — `ParseJobReconciler`:** mirrors `ParseJobPublisher` exactly — narrow `ParseJobReconcilerRepository` contract, options/observer/logger interfaces, module-level `noopObserver`/`noopLogger` defaults, and a single `reconcileStale` method that calls the repository, then per job notifies `observer.jobReconciled` and logs `parse job reconciled` bindings, returning the reclaimed array. No try/catch — failures propagate to the `IntervalTask` wrapper. Factory-injection unit test covers populated / empty / noop-default paths. +- **Task 3 — wiring:** two new envalid vars (`PARSE_JOB_STALE_AFTER_MS` default `3_600_000`, `PARSE_JOB_RECONCILE_BATCH_SIZE` default `25`) surfaced on `config.ingest`; `createIngestRuntime` constructs the reconciler and runs a third `parse-job-reconciler` `IntervalTask` (reusing `pollIntervalMs`), started/closed with the promotion/publish tasks; `server.ts` passes both fields through. Real-pg integration proves stale `published`→`queued` re-queue + `reconciled` history, fresh-published/queued untouched, and idempotent second pass. + +## Verification + +`pnpm verify` green end-to-end: format + lint + typecheck + unit tests + real-pg integration + openapi + ops checks + **100% reachable-source coverage** (3339/3339 stmts, 1564/1564 branches, 1064/1064 funcs, 3285/3285 lines), 854 tests passing. OpenAPI contract unchanged (no route/schema change). + +## Deviations from Plan + +### Auto-fixed Issues + +**1. [Rule 3 - Blocking] Two extra inline `AppConfig['ingest']` fixtures needed the new fields** +- **Found during:** Task 3 typecheck. +- **Issue:** `src/infra/queue/rabbitmq.test.ts` and `src/infra/storage/client.test.ts` build inline `ingest` config objects (beyond the plan-listed `app.test.ts`); adding the two interface fields broke their typecheck. +- **Fix:** Added `reconcileBatchSize: 25` and `staleAfterMs: 3_600_000` to both fixtures. +- **Commit:** f4e0c1b + +**2. [Rule 1 - Coverage gate] `noopLogger.error` unreachable** +- **Found during:** Task 3 coverage gate. +- **Issue:** Unlike the publisher (whose `logger.error` is reached on a publish-failure path), `reconcileStale` never logs errors, leaving the noop `error` stub body uncovered. +- **Fix:** Narrow `/* v8 ignore next 3 -- @preserve */` on the noop error body with a justification, per the tests-skill sanctioned suppression. The `error` member is retained to satisfy the `IntervalTaskLogger` contract passed in at runtime. +- **Commit:** efeb9d7 + +## Known Stubs + +None. The reconciler is fully wired end-to-end (config → runtime → repository → DB) and proven by the real-pg integration test. + +## Threat Flags + +None. No new network endpoints, auth paths, or trust boundaries; the only new surface is operator-controlled config and an internal background loop already covered by the plan's threat register. + +## Self-Check: PASSED + +- Files exist: `0012_parse_job_history_reconciled_action.sql`, `reconciler.ts`, `reconciler.test.ts` — FOUND. +- Commits exist: `2891abd` (Task 1), `efeb9d7` (Task 2), `f4e0c1b` (Task 3) — FOUND. +- `pnpm verify` green with 100% reachable-source coverage. diff --git a/src/config/env.ts b/src/config/env.ts index 25d5285..9e29ba3 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -14,6 +14,8 @@ export interface AppConfig { pollIntervalMs: number; promotionBatchSize: number; publishBatchSize: number; + reconcileBatchSize: number; + staleAfterMs: number; }; auth: { bootstrapAdminSteamId: string; @@ -44,6 +46,8 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { RABBITMQ_URL: url(), INGEST_PROMOTION_BATCH_SIZE: num({ default: 25 }), PARSE_JOB_PUBLISH_BATCH_SIZE: num({ default: 25 }), + PARSE_JOB_RECONCILE_BATCH_SIZE: num({ default: 25 }), + PARSE_JOB_STALE_AFTER_MS: num({ default: 3_600_000 }), PARSER_CONTRACT_VERSION: str({ default: "3.0.0" }), RUNTIME_POLL_INTERVAL_MS: num({ default: 5000 }), BOOTSTRAP_ADMIN_STEAM_ID: str({ default: "" }), @@ -70,6 +74,8 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { pollIntervalMs: parsed.RUNTIME_POLL_INTERVAL_MS, promotionBatchSize: parsed.INGEST_PROMOTION_BATCH_SIZE, publishBatchSize: parsed.PARSE_JOB_PUBLISH_BATCH_SIZE, + reconcileBatchSize: parsed.PARSE_JOB_RECONCILE_BATCH_SIZE, + staleAfterMs: parsed.PARSE_JOB_STALE_AFTER_MS, }, auth: { bootstrapAdminSteamId: parsed.BOOTSTRAP_ADMIN_STEAM_ID, diff --git a/src/infra/db/migrations/0012_parse_job_history_reconciled_action.sql b/src/infra/db/migrations/0012_parse_job_history_reconciled_action.sql new file mode 100644 index 0000000..03ece03 --- /dev/null +++ b/src/infra/db/migrations/0012_parse_job_history_reconciled_action.sql @@ -0,0 +1 @@ +alter type parse_job_history_action add value 'reconciled'; diff --git a/src/infra/queue/rabbitmq.test.ts b/src/infra/queue/rabbitmq.test.ts index ecf0ece..fd1f864 100644 --- a/src/infra/queue/rabbitmq.test.ts +++ b/src/infra/queue/rabbitmq.test.ts @@ -66,6 +66,8 @@ const config: AppConfig = { pollIntervalMs: 5000, promotionBatchSize: 25, publishBatchSize: 25, + reconcileBatchSize: 25, + staleAfterMs: 3_600_000, }, logLevel: "silent", port: 3000, diff --git a/src/infra/storage/client.test.ts b/src/infra/storage/client.test.ts index 4a0a1bb..8539fe3 100644 --- a/src/infra/storage/client.test.ts +++ b/src/infra/storage/client.test.ts @@ -48,6 +48,8 @@ const config: AppConfig = { pollIntervalMs: 5000, promotionBatchSize: 25, publishBatchSize: 25, + reconcileBatchSize: 25, + staleAfterMs: 3_600_000, }, logLevel: "silent", port: 3000, diff --git a/src/modules/ingest/reconciler.test.ts b/src/modules/ingest/reconciler.test.ts new file mode 100644 index 0000000..12cb8f9 --- /dev/null +++ b/src/modules/ingest/reconciler.test.ts @@ -0,0 +1,94 @@ +/* eslint-disable @typescript-eslint/unbound-method, camelcase, no-magic-numbers, unicorn/no-null */ +import { describe, expect, it, vi } from "vitest"; + +import { + ParseJobReconciler, + type ParseJobReconcilerObserver, + type ParseJobReconcilerRepository, +} from "./reconciler.js"; + +import type { ParseJobRecord } from "./types.js"; + +describe("ParseJobReconciler", () => { + it("reclaims stale published jobs, notifies the observer, and logs each one", async () => { + const jobs = [parseJob("job-1"), parseJob("job-2")], + repository: ParseJobReconcilerRepository = { + reclaimStalePublishedJobs: vi.fn(async () => jobs), + }, + observer = observerDouble(), + logger = { error: vi.fn(), info: vi.fn() }, + reconciler = new ParseJobReconciler(repository, { logger, observer }), + reclaimed = await reconciler.reconcileStale({ + batchSize: 25, + staleAfterMs: 3_600_000, + }); + + expect(repository.reclaimStalePublishedJobs).toHaveBeenCalledWith( + 3_600_000, + 25, + ); + expect(reclaimed).toBe(jobs); + expect(observer.jobReconciled).toHaveBeenCalledTimes(2); + expect(observer.jobReconciled).toHaveBeenCalledWith(jobs[0]); + expect(observer.jobReconciled).toHaveBeenCalledWith(jobs[1]); + expect(logger.info).toHaveBeenCalledTimes(2); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ job_id: "job-1" }), + "parse job reconciled", + ); + }); + + it("does nothing per-job when the repository returns no stale jobs", async () => { + const repository: ParseJobReconcilerRepository = { + reclaimStalePublishedJobs: vi.fn(async () => []), + }, + observer = observerDouble(), + logger = { error: vi.fn(), info: vi.fn() }, + reconciler = new ParseJobReconciler(repository, { logger, observer }), + reclaimed = await reconciler.reconcileStale({ + batchSize: 10, + staleAfterMs: 1000, + }); + + expect(reclaimed).toEqual([]); + expect(observer.jobReconciled).not.toHaveBeenCalled(); + expect(logger.info).not.toHaveBeenCalled(); + }); + + it("uses noop observer/logger defaults when runtime options are omitted", async () => { + const jobs = [parseJob("job-3")], + repository: ParseJobReconcilerRepository = { + reclaimStalePublishedJobs: vi.fn(async () => jobs), + }, + reconciler = new ParseJobReconciler(repository), + reclaimed = await reconciler.reconcileStale({ + batchSize: 5, + staleAfterMs: 500, + }); + + expect(reclaimed).toBe(jobs); + expect(repository.reclaimStalePublishedJobs).toHaveBeenCalledWith(500, 5); + }); +}); + +function observerDouble(): ParseJobReconcilerObserver { + return { jobReconciled: vi.fn() }; +} + +function parseJob(id: string): ParseJobRecord { + return { + attempts: 1, + checksum: "a".repeat(64), + createdAt: "2026-05-09T00:00:00.000Z", + error: null, + finishedAt: null, + id, + objectKey: `raw/${id}.ocap.json`, + parserContractVersion: "3.0.0", + publishedAt: "2026-05-09T00:00:00.000Z", + replayId: `replay-${id}`, + startedAt: null, + status: "queued", + updatedAt: "2026-05-09T00:00:00.000Z", + }; +} diff --git a/src/modules/ingest/reconciler.ts b/src/modules/ingest/reconciler.ts new file mode 100644 index 0000000..19f413b --- /dev/null +++ b/src/modules/ingest/reconciler.ts @@ -0,0 +1,83 @@ +/* eslint-disable camelcase */ +import type { ParseJobRecord } from "./types.js"; + +export interface ParseJobReconcilerRepository { + reclaimStalePublishedJobs( + staleAfterMs: number, + batchSize: number, + ): Promise; +} + +export interface ParseJobReconcilerOptions { + batchSize: number; + staleAfterMs: number; +} + +export interface ParseJobReconcilerObserver { + jobReconciled(job: ParseJobRecord): void; +} + +export interface ParseJobReconcilerLogger { + error(bindings: Record, message: string): void; + info(bindings: Record, message: string): void; +} + +export interface ParseJobReconcilerRuntimeOptions { + logger?: ParseJobReconcilerLogger; + observer?: ParseJobReconcilerObserver; +} + +const noopObserver: ParseJobReconcilerObserver = { + jobReconciled(job) { + void job; + }, + }, + noopLogger: ParseJobReconcilerLogger = { + /* v8 ignore next 3 -- @preserve reconcileStale never logs errors; the noop error is a contract stub */ + error(bindings, message) { + void bindings; + void message; + }, + info(bindings, message) { + void bindings; + void message; + }, + }; + +export class ParseJobReconciler { + private readonly logger: ParseJobReconcilerLogger; + + private readonly observer: ParseJobReconcilerObserver; + + public constructor( + private readonly repository: ParseJobReconcilerRepository, + runtimeOptions: ParseJobReconcilerRuntimeOptions = {}, + ) { + this.logger = runtimeOptions.logger ?? noopLogger; + this.observer = runtimeOptions.observer ?? noopObserver; + } + + public async reconcileStale( + options: ParseJobReconcilerOptions, + ): Promise { + const reclaimed = await this.repository.reclaimStalePublishedJobs( + options.staleAfterMs, + options.batchSize, + ); + + for (const job of reclaimed) { + this.observer.jobReconciled(job); + this.logger.info(parseJobLogBindings(job), "parse job reconciled"); + } + + return reclaimed; + } +} + +function parseJobLogBindings(job: ParseJobRecord): Record { + return { + job_id: job.id, + published_at: job.publishedAt, + replay_id: job.replayId, + }; +} diff --git a/src/modules/ingest/repository/repository.ts b/src/modules/ingest/repository/repository.ts index ffeba2c..786526c 100644 --- a/src/modules/ingest/repository/repository.ts +++ b/src/modules/ingest/repository/repository.ts @@ -312,6 +312,49 @@ export class PgIngestRepository { }); } + public async reclaimStalePublishedJobs( + staleAfterMs: number, + batchSize: number, + ): Promise { + return this.withTransaction(async (client) => { + const result = await client.query( + ` + with stale as ( + select id + from parse_jobs + where status = 'published' + and published_at < now() - ($1 * interval '1 millisecond') + order by published_at, id + limit $2 + for update skip locked + ) + update parse_jobs + set status = 'queued', + published_at = null, + started_at = null, + finished_at = null, + error = null, + updated_at = now() + from stale + where parse_jobs.id = stale.id + returning parse_jobs.* + `, + [staleAfterMs, batchSize], + ); + const reclaimed = result.rows.map((row) => mapParseJobRow(row)); + for (const job of reclaimed) { + await recordParseJobHistory(client, { + action: "reconciled", + details: { stale_after_ms: staleAfterMs }, + jobId: job.id, + statusFrom: "published", + statusTo: "queued", + }); + } + return reclaimed; + }); + } + public async retryParseJob( jobId: string, actorUserId: string, diff --git a/src/modules/ingest/repository/tests/postgres.test.ts b/src/modules/ingest/repository/tests/postgres.test.ts index 2289a9f..3b8684a 100644 --- a/src/modules/ingest/repository/tests/postgres.test.ts +++ b/src/modules/ingest/repository/tests/postgres.test.ts @@ -458,6 +458,83 @@ describe("PgIngestRepository", () => { ]); }); + it("reclaims only stale published jobs, audits them, and is idempotent", async () => { + const seed = async ( + sourceSystem: string, + sourceReplayId: string, + checksum: string, + ): Promise => { + await insertStaging(sourceSystem, sourceReplayId, checksum); + return repository.withTransaction(async (client) => { + const [claimed] = await repository.claimPendingStagingRecords(1), + replay = await repository.createReplay( + client, + requiredRecord(claimed), + ), + job = await repository.createParseJob(client, replay, "3.0.0"); + return job.id; + }); + }; + + const staleJobId = await seed( + "source-stale", + "replay-stale", + "1".repeat(64), + ), + freshJobId = await seed("source-fresh", "replay-fresh", "2".repeat(64)), + queuedJobId = await seed( + "source-queued", + "replay-queued", + "3".repeat(64), + ); + + await repository.markJobPublished(staleJobId); + await repository.markJobPublished(freshJobId); + await pool.query( + "update parse_jobs set published_at = now() - interval '2 hours' where id = $1", + [staleJobId], + ); + + const reclaimed = await repository.reclaimStalePublishedJobs(3_600_000, 10); + + expect(reclaimed).toHaveLength(1); + expect(reclaimed[0]).toMatchObject({ + finishedAt: null, + id: staleJobId, + publishedAt: null, + startedAt: null, + status: "queued", + }); + expect(await repository.getParseJob(staleJobId)).toMatchObject({ + publishedAt: null, + status: "queued", + }); + expect(await repository.getParseJob(freshJobId)).toMatchObject({ + status: "published", + }); + expect(await repository.getParseJob(queuedJobId)).toMatchObject({ + status: "queued", + }); + + const history = await repository.listParseJobHistory(staleJobId); + expect(history.map((entry) => entry.action)).toEqual([ + "created", + "published", + "reconciled", + ]); + expect(history.at(-1)).toMatchObject({ + action: "reconciled", + statusFrom: "published", + statusTo: "queued", + }); + + const secondPass = await repository.reclaimStalePublishedJobs( + 3_600_000, + 10, + ); + expect(secondPass).toEqual([]); + }); + it("rolls back transactions and reports missing insert rows defensively", async () => { await expect( repository.withTransaction(async () => { diff --git a/src/modules/ingest/runtime.test.ts b/src/modules/ingest/runtime.test.ts index 72be8df..3a90d25 100644 --- a/src/modules/ingest/runtime.test.ts +++ b/src/modules/ingest/runtime.test.ts @@ -42,7 +42,9 @@ describe("ingest runtime", () => { publishBatchSize: 3, queue, recalculation, + reconcileBatchSize: 4, repository: repository as unknown as PgIngestRepository, + staleAfterMs: 3_600_000, }); expect(queue.consumeParserResults).toHaveBeenCalledOnce(); @@ -65,11 +67,16 @@ describe("ingest runtime", () => { await vi.advanceTimersByTimeAsync(0); expect(repository.claimPendingStagingRecords).toHaveBeenCalledWith(2); expect(repository.listPublishableJobs).toHaveBeenCalledWith(3); + expect(repository.reclaimStalePublishedJobs).toHaveBeenCalledWith( + 3_600_000, + 4, + ); await runtime.close(); await vi.advanceTimersByTimeAsync(100); expect(repository.claimPendingStagingRecords).toHaveBeenCalledTimes(1); expect(repository.listPublishableJobs).toHaveBeenCalledTimes(1); + expect(repository.reclaimStalePublishedJobs).toHaveBeenCalledTimes(1); } finally { vi.useRealTimers(); } @@ -104,7 +111,9 @@ describe("ingest runtime", () => { publishBatchSize: 3, queue, recalculation, + reconcileBatchSize: 4, repository: repository as unknown as PgIngestRepository, + staleAfterMs: 3_600_000, }); await queue.handlers?.completed(completedMessage); @@ -151,6 +160,7 @@ function repositoryDouble() { return { claimPendingStagingRecords: vi.fn(async () => []), listPublishableJobs: vi.fn(async () => []), + reclaimStalePublishedJobs: vi.fn(async () => []), recordParserCompleted: vi.fn( async (): Promise => "parser-result-1", ), diff --git a/src/modules/ingest/runtime.ts b/src/modules/ingest/runtime.ts index b55873d..ada8a06 100644 --- a/src/modules/ingest/runtime.ts +++ b/src/modules/ingest/runtime.ts @@ -4,6 +4,7 @@ import { } from "../../infra/runtime/interval-task.js"; import { ParseJobPublisher } from "./publisher.js"; +import { ParseJobReconciler } from "./reconciler.js"; import { IngestPromotionService } from "./service.js"; import type { PgIngestRepository } from "./repository/repository.js"; @@ -24,7 +25,9 @@ export interface IngestRuntimeOptions { pollIntervalMs: number; queue: RabbitMqParserRuntime; recalculation: ParserResultRecalculator; + reconcileBatchSize: number; repository: PgIngestRepository; + staleAfterMs: number; } export interface ParserArtifactLoader { @@ -53,6 +56,9 @@ export async function createIngestRuntime( publisher = new ParseJobPublisher(options.repository, options.queue, { logger: options.logger, }), + reconciler = new ParseJobReconciler(options.repository, { + logger: options.logger, + }), promotionTask = new IntervalTask({ intervalMs: options.pollIntervalMs, logger: options.logger, @@ -71,6 +77,17 @@ export async function createIngestRuntime( task: async () => { await publisher.publishQueued({ batchSize: options.publishBatchSize }); }, + }), + reconcileTask = new IntervalTask({ + intervalMs: options.pollIntervalMs, + logger: options.logger, + name: "parse-job-reconciler", + task: async () => { + await reconciler.reconcileStale({ + batchSize: options.reconcileBatchSize, + staleAfterMs: options.staleAfterMs, + }); + }, }); await options.queue.consumeParserResults({ @@ -96,11 +113,16 @@ export async function createIngestRuntime( return { async close() { - await Promise.all([promotionTask.close(), publishTask.close()]); + await Promise.all([ + promotionTask.close(), + publishTask.close(), + reconcileTask.close(), + ]); }, start() { promotionTask.start(); publishTask.start(); + reconcileTask.start(); }, }; } diff --git a/src/modules/ingest/types.ts b/src/modules/ingest/types.ts index 8214163..b1ceb1d 100644 --- a/src/modules/ingest/types.ts +++ b/src/modules/ingest/types.ts @@ -65,6 +65,7 @@ export type ParseJobHistoryAction = | "manual_reparse" | "publish_failed" | "published" + | "reconciled" | "retried" | "parser_completed" | "parser_failed"; diff --git a/src/server.ts b/src/server.ts index ac23a63..6c98a67 100644 --- a/src/server.ts +++ b/src/server.ts @@ -108,7 +108,9 @@ const config = loadConfig(), publishBatchSize: config.ingest.publishBatchSize, queue: queueRuntime, recalculation, + reconcileBatchSize: config.ingest.reconcileBatchSize, repository: ingestRepository, + staleAfterMs: config.ingest.staleAfterMs, }); let shuttingDown = false; diff --git a/src/test/app.test.ts b/src/test/app.test.ts index 65ddf5b..9636915 100644 --- a/src/test/app.test.ts +++ b/src/test/app.test.ts @@ -85,6 +85,8 @@ test("createLoggerOptions should build credential redaction paths when config is pollIntervalMs: 5000, promotionBatchSize: 25, publishBatchSize: 25, + reconcileBatchSize: 25, + staleAfterMs: 3_600_000, }, logLevel: "debug", port: 3000,