feat(ingest): reconcile orphaned published parse_jobs (F5)

.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
 

.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.

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,

src/infra/db/migrations/0012_parse_job_history_reconciled_action.sql

@@ -0,0 +1 @@
+alter type parse_job_history_action add value 'reconciled';

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,

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,

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",
+  };
+}

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<ParseJobRecord[]>;
+}
+
+export interface ParseJobReconcilerOptions {
+  batchSize: number;
+  staleAfterMs: number;
+}
+
+export interface ParseJobReconcilerObserver {
+  jobReconciled(job: ParseJobRecord): void;
+}
+
+export interface ParseJobReconcilerLogger {
+  error(bindings: Record<string, unknown>, message: string): void;
+  info(bindings: Record<string, unknown>, 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<ParseJobRecord[]> {
+    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<string, unknown> {
+  return {
+    job_id: job.id,
+    published_at: job.publishedAt,
+    replay_id: job.replayId,
+  };
+}

src/modules/ingest/repository/repository.ts

@@ -312,6 +312,49 @@ export class PgIngestRepository {
     });
   }
 
+  public async reclaimStalePublishedJobs(
+    staleAfterMs: number,
+    batchSize: number,
+  ): Promise<ParseJobRecord[]> {
+    return this.withTransaction(async (client) => {
+      const result = await client.query<ParseJobRow>(
+        `
+          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,