From 56d9b729dadd86164d6a382ba4257559887fbc10 Mon Sep 17 00:00:00 2001 From: Clay Townsend Date: Sun, 23 Aug 2026 16:14:31 +1000 Subject: [PATCH 1/7] docs(specs): add background work complexity gate --- .../2026-08-21-background-work-04-solution.md | 24 +++++++++++++++++++ ...kground-work-06-leased-jobs-in-progress.md | 8 ++++--- ...1-background-work-07-data-sync-adoption.md | 1 + 3 files changed, 30 insertions(+), 3 deletions(-) diff --git a/.ai/specs/2026-08-21-background-work-04-solution.md b/.ai/specs/2026-08-21-background-work-04-solution.md index 2291427e8f9..57196218cfb 100644 --- a/.ai/specs/2026-08-21-background-work-04-solution.md +++ b/.ai/specs/2026-08-21-background-work-04-solution.md @@ -196,6 +196,30 @@ What the pattern implies elsewhere in the repo, as **named follow-up specs**: None of these is needed for the leased tier or for `data_sync`; all of them reuse the same two ideas, which is why they are listed here rather than designed. +## 📝 Complexity model and benchmark gate + +The leased tier must keep the unit of work addressable by key; it must not move the failure from "one long job" to "one repairer that scans every job". The model is: + +| Path | Required shape | Bound | +|---|---|---| +| Enqueue / re-enqueue | deterministic `queueJobId = pj---`; broker dedup is a keyed lookup | O(log queue) or broker-native O(1) | +| Claim | primary-key lookup plus `(seq, redrives)` and lease predicate | O(log jobs), no backlog scan | +| Single-runner check | partial unique index on `(lock_key, tenant, org)` for live leased rows | O(log live_keys) | +| Heartbeat | one HOT update on the row; no indexed heartbeat/lease columns | O(1) row update, no index churn | +| Q2/Q5 repair | indexed pending/mirror cells (`pending_since`, `finished_at`) | O(due + log jobs) | +| Q1 orphan detection | bounded scan of live leased rows because `lease_expires_at` is intentionally unindexed | O(live_leased) until the measured ceiling below | + +Q1 is the only intentional scan. It is allowed for phase 1 only with a measured ceiling: the implementation PR must show the reconciler tick stays below **100 ms p95 at 10k live leased rows with 100 expired rows** on the integration Postgres, and below **1 s p95 at 100k live leased rows** or else add a narrow repair-cell side table before merge. Do **not** index `lease_expires_at` on `progress_jobs` to satisfy this; that would turn every heartbeat into index churn and violate invariant 6. The side-table fallback is a separate cell index owned by the repairer, not another authoritative clock. + +Benchmark command expected from the implementation PR (exact script name may change, but the dimensions may not): + +```bash +spbench run "yarn mercato progress bench-leases --live 10000 --expired 100 --pending 100 --mirror-pending 100" --binary open-mercato-progress-reconciler +spbench run "yarn mercato progress bench-leases --live 100000 --expired 100 --pending 100 --mirror-pending 100" --binary open-mercato-progress-reconciler +``` + +The acceptance report must include tick p50/p95, database queries per tick, rows scanned, rows repaired, and `pg_stat_user_tables.n_tup_hot_upd` for `progress_jobs` during the heartbeat phase. A speed number without the repaired-row count is not a result. + ## 📋 Implementation specs (replaces the single phasing table of v3) | Part | Spec | Owner surface | Depends on | Ships alone? | Closes | Approval state | diff --git a/.ai/specs/2026-08-21-background-work-06-leased-jobs-in-progress.md b/.ai/specs/2026-08-21-background-work-06-leased-jobs-in-progress.md index fc15c599d5b..7c7f00b8dad 100644 --- a/.ai/specs/2026-08-21-background-work-06-leased-jobs-in-progress.md +++ b/.ai/specs/2026-08-21-background-work-06-leased-jobs-in-progress.md @@ -244,7 +244,7 @@ flowchart TB Every reconciler transition that ends a row (Q1 park, Q3 cancel) is a §7 terminal transaction: the CAS and the domain mirror commit together, and a throwing mirror rolls the CAS back so the row is selected again next tick. Q5 is the only path in which the mirror runs *after* a terminal CAS committed, and it exists for two cases only: kinds that declared `mirror: 'deferred'`, and rows whose terminal CAS committed before the column existed (migration backfill sets `domain_mirrored_at = finished_at` for rows without `subject_type`, and leaves it null where a subject exists so Q5 re-mirrors them once). -Constraint: `pendingTtlMs` must exceed the worst-case queue backlog for that kind, otherwise Q2 re-drives healthy queued deliveries (they are refused on `redrives` and cost nothing but a wasted delivery, yet each one counts toward `never_started`); the default 15 min is per kind and the reconciler logs a warning when a Q2 re-drive finds the previous delivery still queued (via `Queue.getJobState` where the capability exists). Cost note: Q1 scans the `running ∧ leased` partial index and filters `lease_expires_at` in the heap — a bounded scan of live leased rows every 30 s, which stays cheap to ~10k live rows; beyond that the follow-up is a narrow `progress_job_leases` side table, not an index on `lease_expires_at` (invariant 6). +Constraint: `pendingTtlMs` must exceed the worst-case queue backlog for that kind, otherwise Q2 re-drives healthy queued deliveries (they are refused on `redrives` and cost nothing but a wasted delivery, yet each one counts toward `never_started`); the default 15 min is per kind and the reconciler logs a warning when a Q2 re-drive finds the previous delivery still queued (via `Queue.getJobState` where the capability exists). Cost note: Q1 scans the `running ∧ leased` partial index and filters `lease_expires_at` in the heap — a bounded scan of live leased rows every 30 s. This is the only intentional O(live) path in the design; all other repair cells are indexed by their due class. It is acceptable for phase 1 only if the part 4 benchmark gate is met (100 ms p95 at 10k live rows / 100 expired, and 1 s p95 at 100k live rows). If it misses, the implementation must add the narrow `progress_job_leases` repair-cell side table before shipping, rather than indexing `lease_expires_at` on `progress_jobs` and breaking HOT heartbeats (invariant 6). **Connection ceiling** (`packages/queue/AGENTS.md` → Connection Budget): a leased slice uses its request-scoped EM plus one *transient* pooled connection during each heartbeat/lease statement (≤ 1 query every 20 s per slice), and the reconciler worker uses one connection per batch query. The worst case per worker process is therefore `Σconcurrency + 1` concurrent connections instead of `Σconcurrency`; the existing DB-budget clamp in `mercato worker --all` is updated to reserve that one extra connection and the integration test asserts `pg_stat_activity` stays under the clamp during the soak. @@ -335,6 +335,7 @@ Real Postgres + Redis (`__integration__`, docker-compose runner), with a test ki 9. Reconciler vs live driver → take fences the driver at its next heartbeat; at most one in-flight page applied twice. 10. Connection ceiling: `pg_stat_activity` stays under the `mercato worker --all` clamp during a 3-replica soak. 11. Retention never deletes a terminal row with `domain_mirrored_at is null` and a subject. +12. Reconciler complexity benchmark: a synthetic leased-job fixture with 10k live rows / 100 expired rows and 100k live rows / 100 expired rows runs through the repo-native benchmark target and, when `spbench` is available, through the part 4 `spbench run` commands. The report includes p50/p95 tick time, rows scanned, rows repaired, query count, and `n_tup_hot_upd` growth during heartbeats. ## 📋 Implementation Plan @@ -344,8 +345,9 @@ Real Postgres + Redis (`__integration__`, docker-compose runner), with a test ki 4. Kind registry (`mirror` option, `maxMirrorAttempts`) + `runSlice` factory + generic worker binding. Tests: refused claim does no work; budget → yield → rewritten delivery claims; throw → `failSlice` + rethrow → retry claims; unrecoverable → terminal via §7; shutdown → interrupted yield with counters untouched; mirror throw on `drained` → slice fails, retry completes. 5. Reconciler worker (`progress-reconcile` repeatable every 30 s via `upsertRepeatable`, registered at worker boot) + `progress-retention`; Q1–Q5; `SKIP LOCKED` batches; per-row isolation; stale sweep moved here; `OM_PROGRESS_SWEEP_ON_READ` (default on). Tests: orphan take bumps epoch and re-drives with backoff; Q1 waits for `next_run_at`; poison park on `redrives_since_commit`; lost hand-back re-drive; Q3 cancels a dead slice with the mirror in the same commit; Q5 mirrors a deferred row and a backfilled legacy row; a throwing park mirror leaves the row for the next tick and counts; two reconcilers on one set; tracked-tier sweep parity; repeatable survives a failed run and a Redis flush + boot; retention skips unmirrored rows. 6. Leased cancel semantics + minimal cascade (§8); `POST /api/progress/jobs/[id]/redrive` (ACL `progress.update` + kind features); top bar "cancelling"/"parked"; OpenAPI. -7. Integration coverage above. -8. AGENTS.md (`progress` incl. the repairer carve-out and the `onTransition` idempotency rule, root Task Router row), docs page "leased jobs", `BACKWARD_COMPATIBILITY.md` and UPGRADE_NOTES entries, `.ai/lessons.md`. +7. Repo-native `progress bench-leases` command (or equivalent test-only CLI) that seeds synthetic leased rows and prints the part 4 benchmark columns; run it directly and through `spbench` where available. +8. Integration coverage above. +9. AGENTS.md (`progress` incl. the repairer carve-out and the `onTransition` idempotency rule, root Task Router row), docs page "leased jobs", `BACKWARD_COMPATIBILITY.md` and UPGRADE_NOTES entries, `.ai/lessons.md`. ## Open items for review diff --git a/.ai/specs/2026-08-21-background-work-07-data-sync-adoption.md b/.ai/specs/2026-08-21-background-work-07-data-sync-adoption.md index 939a0c85731..3522c79e65b 100644 --- a/.ai/specs/2026-08-21-background-work-07-data-sync-adoption.md +++ b/.ai/specs/2026-08-21-background-work-07-data-sync-adoption.md @@ -91,6 +91,7 @@ Adapter contract: **signature unchanged**. Three `data_sync/AGENTS.md` Ask-First 4. Cancel during adapter I/O → adapter receives `signal` → run `cancelled` and mirrored in one commit. 5. Existing `data_sync` integration specs green. 6. **Soak**: docker-compose with 3 worker replicas, a fake adapter producing 500 batches, a supervisor SIGKILLing one replica per minute and flushing Redis once; assert every `(run_id, batch_no)` appears exactly once in a test ledger the adapter writes inside the fenced commit, `sync_runs.batches_completed = 500`, `sync_runs.status = 'completed'` with `finished_at` set, and no row is left `running`/`pending`/unmirrored after the reconciler's next tick. +7. **Performance/complexity gate**: the same fake adapter runs under a backlog fixture with 10k and 100k unrelated live leased jobs. The data_sync run must still complete with exactly-once ledger rows, and the progress reconciler must meet the part 4 p95 targets. Report quality (`500/500` unique committed batches), speed (slice throughput and reconciler p95), and baseline (today's one-job data_sync path or the previous implementation PR run on the same fixture). ## 📋 Implementation Plan From 2c5f0a0f33e963b5280d4bed8ba6104acf589877 Mon Sep 17 00:00:00 2001 From: Clay Date: Mon, 24 Aug 2026 18:07:58 +1000 Subject: [PATCH 2/7] Remove benchmark command examples from specs Removed internal benchmark command examples from the implementation PR requirements. --- .ai/specs/2026-08-21-background-work-04-solution.md | 7 ------- 1 file changed, 7 deletions(-) diff --git a/.ai/specs/2026-08-21-background-work-04-solution.md b/.ai/specs/2026-08-21-background-work-04-solution.md index 57196218cfb..9fb26a95128 100644 --- a/.ai/specs/2026-08-21-background-work-04-solution.md +++ b/.ai/specs/2026-08-21-background-work-04-solution.md @@ -211,13 +211,6 @@ The leased tier must keep the unit of work addressable by key; it must not move Q1 is the only intentional scan. It is allowed for phase 1 only with a measured ceiling: the implementation PR must show the reconciler tick stays below **100 ms p95 at 10k live leased rows with 100 expired rows** on the integration Postgres, and below **1 s p95 at 100k live leased rows** or else add a narrow repair-cell side table before merge. Do **not** index `lease_expires_at` on `progress_jobs` to satisfy this; that would turn every heartbeat into index churn and violate invariant 6. The side-table fallback is a separate cell index owned by the repairer, not another authoritative clock. -Benchmark command expected from the implementation PR (exact script name may change, but the dimensions may not): - -```bash -spbench run "yarn mercato progress bench-leases --live 10000 --expired 100 --pending 100 --mirror-pending 100" --binary open-mercato-progress-reconciler -spbench run "yarn mercato progress bench-leases --live 100000 --expired 100 --pending 100 --mirror-pending 100" --binary open-mercato-progress-reconciler -``` - The acceptance report must include tick p50/p95, database queries per tick, rows scanned, rows repaired, and `pg_stat_user_tables.n_tup_hot_upd` for `progress_jobs` during the heartbeat phase. A speed number without the repaired-row count is not a result. ## 📋 Implementation specs (replaces the single phasing table of v3) From 46184612320e9a322f4608cd7d68bf6fa1685e3f Mon Sep 17 00:00:00 2001 From: Clay Townsend Date: Sun, 23 Aug 2026 18:47:24 +1000 Subject: [PATCH 3/7] feat(progress): add idempotent repair cells --- .../progress/__tests__/repair-cells.test.ts | 22 + .../src/modules/progress/data/entities.ts | 33 + .../src/modules/progress/lib/repair-cells.ts | 58 ++ .../migrations/.snapshot-open-mercato.json | 623 ++++++++++++++---- .../Migration20260823084507_progress.ts | 10 + 5 files changed, 632 insertions(+), 114 deletions(-) create mode 100644 packages/core/src/modules/progress/__tests__/repair-cells.test.ts create mode 100644 packages/core/src/modules/progress/lib/repair-cells.ts create mode 100644 packages/core/src/modules/progress/migrations/Migration20260823084507_progress.ts diff --git a/packages/core/src/modules/progress/__tests__/repair-cells.test.ts b/packages/core/src/modules/progress/__tests__/repair-cells.test.ts new file mode 100644 index 00000000000..36265cf3ef2 --- /dev/null +++ b/packages/core/src/modules/progress/__tests__/repair-cells.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, it } from '@jest/globals' +import { claimDueRepairCells, removeRepairCell, upsertRepairCell } from '../lib/repair-cells' + +describe('progress repair cells', () => { + it('rejects an unbounded or empty claim batch', async () => { + const em = {} as never + await expect(claimDueRepairCells(em, { tenantId: 't1' }, new Date(), 0)).rejects.toThrow('positive integer') + await expect(claimDueRepairCells(em, { tenantId: 't1' }, new Date(), 1.5)).rejects.toThrow('positive integer') + }) + + it('writes and removes a cell through the tenant-scoped entity API', async () => { + const created = { upsert: jest.fn().mockResolvedValue(undefined) } + await upsertRepairCell(created as never, { + jobId: 'j1', tenantId: 't1', organizationId: 'o1', cell: 'lease_expired', dueAt: new Date(0), + }) + expect(created.upsert).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ jobId: 'j1', tenantId: 't1' }), { onConflictAction: 'merge' }) + + const deleted = { nativeDelete: jest.fn().mockResolvedValue(1) } + await expect(removeRepairCell(deleted as never, 'j1', { tenantId: 't1', organizationId: 'o1' })).resolves.toBe(true) + expect(deleted.nativeDelete).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ jobId: 'j1', tenantId: 't1', organizationId: 'o1' })) + }) +}) diff --git a/packages/core/src/modules/progress/data/entities.ts b/packages/core/src/modules/progress/data/entities.ts index 9af92aa2a93..d8d296317a3 100644 --- a/packages/core/src/modules/progress/data/entities.ts +++ b/packages/core/src/modules/progress/data/entities.ts @@ -93,3 +93,36 @@ export class ProgressJob { @Property({ name: 'updated_at', type: Date, onUpdate: () => new Date() }) updatedAt: Date = new Date() } + +@Entity({ tableName: 'progress_job_repair_cells' }) +@Index({ name: 'progress_job_repair_cells_due_idx', properties: ['tenantId', 'organizationId', 'dueAt', 'jobId'] }) +export class ProgressJobRepairCell { + [OptionalProps]?: 'attempts' | 'reason' | 'createdAt' | 'updatedAt' + + @PrimaryKey({ type: 'uuid' }) + jobId!: string + + @Property({ name: 'tenant_id', type: 'uuid' }) + tenantId!: string + + @Property({ name: 'organization_id', type: 'uuid', nullable: true }) + organizationId?: string | null + + @Property({ name: 'cell', type: 'text' }) + cell!: string + + @Property({ name: 'due_at', type: Date }) + dueAt!: Date + + @Property({ name: 'attempts', type: 'int' }) + attempts: number = 0 + + @Property({ name: 'reason', type: 'text', nullable: true }) + reason?: string | null + + @Property({ name: 'created_at', type: Date, onCreate: () => new Date() }) + createdAt: Date = new Date() + + @Property({ name: 'updated_at', type: Date, onCreate: () => new Date(), onUpdate: () => new Date() }) + updatedAt: Date = new Date() +} diff --git a/packages/core/src/modules/progress/lib/repair-cells.ts b/packages/core/src/modules/progress/lib/repair-cells.ts new file mode 100644 index 00000000000..df435e24bc7 --- /dev/null +++ b/packages/core/src/modules/progress/lib/repair-cells.ts @@ -0,0 +1,58 @@ +import { LockMode } from '@mikro-orm/core' +import type { EntityManager, FilterQuery } from '@mikro-orm/postgresql' +import { ProgressJobRepairCell } from '../data/entities' + +export type ProgressRepairCellInput = { + jobId: string + tenantId: string + organizationId?: string | null + cell: string + dueAt: Date + reason?: string | null +} + +export type ProgressRepairScope = { + tenantId: string + organizationId?: string | null +} + +function scopedFilter(scope: ProgressRepairScope): FilterQuery { + return { + tenantId: scope.tenantId, + ...(scope.organizationId ? { organizationId: scope.organizationId } : { organizationId: null }), + } +} + +export async function upsertRepairCell(em: EntityManager, input: ProgressRepairCellInput): Promise { + // The primary key makes this a single idempotent write under at-least-once delivery. + // A read-then-insert sequence would race when two deliveries repair the same job. + await em.upsert(ProgressJobRepairCell, { + jobId: input.jobId, + tenantId: input.tenantId, + organizationId: input.organizationId ?? null, + cell: input.cell, + dueAt: input.dueAt, + reason: input.reason ?? null, + }, { onConflictAction: 'merge' }) +} + +export async function removeRepairCell(em: EntityManager, jobId: string, scope: ProgressRepairScope): Promise { + return (await em.nativeDelete(ProgressJobRepairCell, { jobId, ...scopedFilter(scope) })) > 0 +} + +export async function claimDueRepairCells( + em: EntityManager, + scope: ProgressRepairScope, + now: Date, + limit: number, +): Promise { + if (!Number.isInteger(limit) || limit < 1) throw new Error('repair cell limit must be a positive integer') + return em.find(ProgressJobRepairCell, { + ...scopedFilter(scope), + dueAt: { $lte: now }, + }, { + orderBy: { dueAt: 'asc', jobId: 'asc' }, + limit, + lockMode: LockMode.PESSIMISTIC_PARTIAL_WRITE, + }) +} diff --git a/packages/core/src/modules/progress/migrations/.snapshot-open-mercato.json b/packages/core/src/modules/progress/migrations/.snapshot-open-mercato.json index dc696b2f59b..a86f4029e7d 100644 --- a/packages/core/src/modules/progress/migrations/.snapshot-open-mercato.json +++ b/packages/core/src/modules/progress/migrations/.snapshot-open-mercato.json @@ -1,160 +1,403 @@ { + "name": "public", "namespaces": [ "public" ], - "name": "public", "tables": [ { + "name": "progress_job_repair_cells", + "schema": "public", "columns": { - "id": { - "name": "id", - "type": "uuid", + "attempts": { + "name": "attempts", + "type": "int", "unsigned": false, "autoincrement": false, "primary": false, "nullable": false, - "default": "gen_random_uuid()", - "mappedType": "uuid" + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": "0", + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "integer" }, - "job_type": { - "name": "job_type", + "cell": { + "name": "cell", "type": "text", "unsigned": false, "autoincrement": false, "primary": false, "nullable": false, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "text" }, - "name": { - "name": "name", - "type": "text", + "created_at": { + "name": "created_at", + "type": "timestamptz(6)", "unsigned": false, "autoincrement": false, "primary": false, "nullable": false, - "mappedType": "text" + "unique": false, + "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "datetime" }, - "description": { - "name": "description", - "type": "text", + "due_at": { + "name": "due_at", + "type": "timestamptz(6)", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": false, + "unique": false, + "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "datetime" + }, + "job_id": { + "name": "job_id", + "type": "uuid", + "unsigned": false, + "autoincrement": false, + "primary": true, + "nullable": false, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "uuid" + }, + "organization_id": { + "name": "organization_id", + "type": "uuid", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, - "mappedType": "text" + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "uuid" }, - "status": { - "name": "status", + "reason": { + "name": "reason", "type": "text", "unsigned": false, "autoincrement": false, "primary": false, - "nullable": false, - "default": "'pending'", + "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "text" }, - "progress_percent": { - "name": "progress_percent", - "type": "smallint", + "tenant_id": { + "name": "tenant_id", + "type": "uuid", "unsigned": false, "autoincrement": false, "primary": false, "nullable": false, - "default": "0", - "mappedType": "smallint" + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "uuid" }, - "processed_count": { - "name": "processed_count", - "type": "int", + "updated_at": { + "name": "updated_at", + "type": "timestamptz(6)", "unsigned": false, "autoincrement": false, "primary": false, "nullable": false, - "default": "0", - "mappedType": "integer" + "unique": false, + "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "datetime" + } + }, + "indexes": [ + { + "columnNames": [ + "tenant_id", + "organization_id", + "due_at", + "job_id" + ], + "composite": true, + "constraint": false, + "keyName": "progress_job_repair_cells_due_idx", + "primary": false, + "unique": false }, - "total_count": { - "name": "total_count", - "type": "int", + { + "columnNames": [ + "job_id" + ], + "composite": false, + "constraint": true, + "keyName": "progress_job_repair_cells_pkey", + "primary": true, + "unique": true + } + ], + "checks": [], + "triggers": [], + "foreignKeys": {}, + "comment": null + }, + { + "name": "progress_jobs", + "schema": "public", + "columns": { + "cancel_requested_at": { + "name": "cancel_requested_at", + "type": "timestamptz(6)", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, - "mappedType": "integer" + "unique": false, + "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "datetime" }, - "eta_seconds": { - "name": "eta_seconds", - "type": "int", + "cancellable": { + "name": "cancellable", + "type": "boolean", "unsigned": false, "autoincrement": false, "primary": false, - "nullable": true, - "mappedType": "integer" + "nullable": false, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": "false", + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "boolean" }, - "started_by_user_id": { - "name": "started_by_user_id", + "cancelled_by_user_id": { + "name": "cancelled_by_user_id", "type": "uuid", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "uuid" }, - "started_at": { - "name": "started_at", - "type": "timestamptz", + "created_at": { + "name": "created_at", + "type": "timestamptz(6)", "unsigned": false, "autoincrement": false, "primary": false, - "nullable": true, + "nullable": false, + "unique": false, "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "datetime" }, - "heartbeat_at": { - "name": "heartbeat_at", - "type": "timestamptz", + "description": { + "name": "description", + "type": "text", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, - "length": 6, - "mappedType": "datetime" + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "text" + }, + "error_message": { + "name": "error_message", + "type": "text", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "text" + }, + "error_stack": { + "name": "error_stack", + "type": "text", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "text" + }, + "eta_seconds": { + "name": "eta_seconds", + "type": "int", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "integer" }, "finished_at": { "name": "finished_at", - "type": "timestamptz", + "type": "timestamptz(6)", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, + "unique": false, "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "datetime" }, - "result_summary": { - "name": "result_summary", - "type": "jsonb", + "heartbeat_at": { + "name": "heartbeat_at", + "type": "timestamptz(6)", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, - "mappedType": "json" + "unique": false, + "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "datetime" }, - "error_message": { - "name": "error_message", - "type": "text", + "id": { + "name": "id", + "type": "uuid", "unsigned": false, "autoincrement": false, - "primary": false, - "nullable": true, - "mappedType": "text" + "primary": true, + "nullable": false, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": "gen_random_uuid()", + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "uuid" }, - "error_stack": { - "name": "error_stack", + "job_type": { + "name": "job_type", "type": "text", "unsigned": false, "autoincrement": false, "primary": false, - "nullable": true, + "nullable": false, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "text" }, "meta": { @@ -164,45 +407,83 @@ "autoincrement": false, "primary": false, "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "json" }, - "cancellable": { - "name": "cancellable", - "type": "boolean", + "name": { + "name": "name", + "type": "text", "unsigned": false, "autoincrement": false, "primary": false, "nullable": false, - "default": "false", - "mappedType": "boolean" + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "text" }, - "cancelled_by_user_id": { - "name": "cancelled_by_user_id", + "organization_id": { + "name": "organization_id", "type": "uuid", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "uuid" }, - "cancel_requested_at": { - "name": "cancel_requested_at", - "type": "timestamptz", + "parent_job_id": { + "name": "parent_job_id", + "type": "uuid", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, - "length": 6, - "mappedType": "datetime" + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "uuid" }, - "parent_job_id": { - "name": "parent_job_id", - "type": "uuid", + "partition_count": { + "name": "partition_count", + "type": "int", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, - "mappedType": "uuid" + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "integer" }, "partition_index": { "name": "partition_index", @@ -211,106 +492,220 @@ "autoincrement": false, "primary": false, "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "integer" }, - "partition_count": { - "name": "partition_count", + "processed_count": { + "name": "processed_count", "type": "int", "unsigned": false, "autoincrement": false, "primary": false, - "nullable": true, + "nullable": false, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": "0", + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "integer" }, - "tenant_id": { - "name": "tenant_id", - "type": "uuid", + "progress_percent": { + "name": "progress_percent", + "type": "smallint", "unsigned": false, "autoincrement": false, "primary": false, "nullable": false, - "mappedType": "uuid" + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": "0", + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "smallint" }, - "organization_id": { - "name": "organization_id", + "result_summary": { + "name": "result_summary", + "type": "jsonb", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "json" + }, + "started_at": { + "name": "started_at", + "type": "timestamptz(6)", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": true, + "unique": false, + "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "datetime" + }, + "started_by_user_id": { + "name": "started_by_user_id", "type": "uuid", "unsigned": false, "autoincrement": false, "primary": false, "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "uuid" }, - "created_at": { - "name": "created_at", - "type": "timestamptz", + "status": { + "name": "status", + "type": "text", "unsigned": false, "autoincrement": false, "primary": false, "nullable": false, - "length": 6, - "mappedType": "datetime" + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": "'pending'", + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "text" + }, + "tenant_id": { + "name": "tenant_id", + "type": "uuid", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": false, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "uuid" + }, + "total_count": { + "name": "total_count", + "type": "int", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "integer" }, "updated_at": { "name": "updated_at", - "type": "timestamptz", + "type": "timestamptz(6)", "unsigned": false, "autoincrement": false, "primary": false, "nullable": false, + "unique": false, "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], "mappedType": "datetime" } }, - "name": "progress_jobs", - "schema": "public", "indexes": [ { - "keyName": "progress_jobs_parent_idx", "columnNames": [ "parent_job_id" ], "composite": false, "constraint": false, + "keyName": "progress_jobs_parent_idx", "primary": false, "unique": false }, { - "keyName": "progress_jobs_type_tenant_idx", "columnNames": [ - "job_type", - "tenant_id" + "id" ], - "composite": true, - "constraint": false, - "primary": false, - "unique": false + "composite": false, + "constraint": true, + "keyName": "progress_jobs_pkey", + "primary": true, + "unique": true }, { - "keyName": "progress_jobs_status_tenant_idx", "columnNames": [ "status", "tenant_id" ], "composite": true, "constraint": false, + "keyName": "progress_jobs_status_tenant_idx", "primary": false, "unique": false }, { - "keyName": "progress_jobs_pkey", "columnNames": [ - "id" + "job_type", + "tenant_id" ], - "composite": false, - "constraint": true, - "primary": true, - "unique": true + "composite": true, + "constraint": false, + "keyName": "progress_jobs_type_tenant_idx", + "primary": false, + "unique": false } ], "checks": [], + "triggers": [], "foreignKeys": {}, - "nativeEnums": {} + "comment": null } ], + "views": [], "nativeEnums": {} -} +} \ No newline at end of file diff --git a/packages/core/src/modules/progress/migrations/Migration20260823084507_progress.ts b/packages/core/src/modules/progress/migrations/Migration20260823084507_progress.ts new file mode 100644 index 00000000000..2dbcca5e5dd --- /dev/null +++ b/packages/core/src/modules/progress/migrations/Migration20260823084507_progress.ts @@ -0,0 +1,10 @@ +import { Migration } from '@mikro-orm/migrations'; + +export class Migration20260823084507_progress extends Migration { + + override up(): void | Promise { + this.addSql(`create table "progress_job_repair_cells" ("job_id" uuid not null, "tenant_id" uuid not null, "organization_id" uuid null, "cell" text not null, "due_at" timestamptz not null, "attempts" int not null default 0, "reason" text null, "created_at" timestamptz not null, "updated_at" timestamptz not null, primary key ("job_id"));`); + this.addSql(`create index "progress_job_repair_cells_due_idx" on "progress_job_repair_cells" ("tenant_id", "organization_id", "due_at", "job_id");`); + } + +} From 8585f8f76ba77c101228c412ab7e3768484126d2 Mon Sep 17 00:00:00 2001 From: Clay Townsend Date: Sun, 23 Aug 2026 18:57:32 +1000 Subject: [PATCH 4/7] feat(progress): add repair-cell leases and acknowledgement --- .../progress/__tests__/repair-cells.test.ts | 27 ++++++- .../src/modules/progress/data/entities.ts | 8 ++- .../src/modules/progress/lib/repair-cells.ts | 70 +++++++++++++++++-- .../migrations/.snapshot-open-mercato.json | 34 +++++++++ .../Migration20260823085716_progress.ts | 13 ++++ 5 files changed, 143 insertions(+), 9 deletions(-) create mode 100644 packages/core/src/modules/progress/migrations/Migration20260823085716_progress.ts diff --git a/packages/core/src/modules/progress/__tests__/repair-cells.test.ts b/packages/core/src/modules/progress/__tests__/repair-cells.test.ts index 36265cf3ef2..b35c1ccc76b 100644 --- a/packages/core/src/modules/progress/__tests__/repair-cells.test.ts +++ b/packages/core/src/modules/progress/__tests__/repair-cells.test.ts @@ -1,11 +1,12 @@ import { describe, expect, it } from '@jest/globals' -import { claimDueRepairCells, removeRepairCell, upsertRepairCell } from '../lib/repair-cells' +import { acknowledgeRepairCell, claimDueRepairCells, releaseRepairCell, removeRepairCell, upsertRepairCell } from '../lib/repair-cells' describe('progress repair cells', () => { it('rejects an unbounded or empty claim batch', async () => { const em = {} as never await expect(claimDueRepairCells(em, { tenantId: 't1' }, new Date(), 0)).rejects.toThrow('positive integer') await expect(claimDueRepairCells(em, { tenantId: 't1' }, new Date(), 1.5)).rejects.toThrow('positive integer') + await expect(claimDueRepairCells(em, { tenantId: 't1' }, new Date(), 1, 0)).rejects.toThrow('positive integer') }) it('writes and removes a cell through the tenant-scoped entity API', async () => { @@ -19,4 +20,28 @@ describe('progress repair cells', () => { await expect(removeRepairCell(deleted as never, 'j1', { tenantId: 't1', organizationId: 'o1' })).resolves.toBe(true) expect(deleted.nativeDelete).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ jobId: 'j1', tenantId: 't1', organizationId: 'o1' })) }) + + it('leases due cells and requires the lease token for acknowledgement or release', async () => { + const cell = { jobId: 'j1', attempts: 0 } + const tx = { + find: jest.fn().mockResolvedValue([cell]), + flush: jest.fn().mockResolvedValue(undefined), + } + const em = { transactional: jest.fn(async (work: (manager: typeof tx) => Promise) => work(tx)) } + + const claim = await claimDueRepairCells(em as never, { tenantId: 't1', organizationId: 'o1' }, new Date(0), 1, 1000) + expect(claim.cells).toEqual([cell]) + expect(cell.attempts).toBe(1) + expect(cell.leaseToken).toBe(claim.leaseToken) + expect(cell.leaseUntil).toEqual(new Date(1000)) + expect(tx.flush).toHaveBeenCalledTimes(1) + + const acknowledged = { nativeDelete: jest.fn().mockResolvedValue(1) } + await expect(acknowledgeRepairCell(acknowledged as never, 'j1', { tenantId: 't1', organizationId: 'o1' }, claim.leaseToken)).resolves.toBe(true) + expect(acknowledged.nativeDelete).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ leaseToken: claim.leaseToken })) + + const released = { nativeUpdate: jest.fn().mockResolvedValue(1) } + await expect(releaseRepairCell(released as never, 'j1', { tenantId: 't1', organizationId: 'o1' }, claim.leaseToken, new Date(2000), 'retry')).resolves.toBe(true) + expect(released.nativeUpdate).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ leaseToken: claim.leaseToken }), expect.objectContaining({ leaseToken: null, leaseUntil: null })) + }) }) diff --git a/packages/core/src/modules/progress/data/entities.ts b/packages/core/src/modules/progress/data/entities.ts index d8d296317a3..9d85ea378de 100644 --- a/packages/core/src/modules/progress/data/entities.ts +++ b/packages/core/src/modules/progress/data/entities.ts @@ -97,7 +97,7 @@ export class ProgressJob { @Entity({ tableName: 'progress_job_repair_cells' }) @Index({ name: 'progress_job_repair_cells_due_idx', properties: ['tenantId', 'organizationId', 'dueAt', 'jobId'] }) export class ProgressJobRepairCell { - [OptionalProps]?: 'attempts' | 'reason' | 'createdAt' | 'updatedAt' + [OptionalProps]?: 'attempts' | 'reason' | 'leaseToken' | 'leaseUntil' | 'createdAt' | 'updatedAt' @PrimaryKey({ type: 'uuid' }) jobId!: string @@ -117,6 +117,12 @@ export class ProgressJobRepairCell { @Property({ name: 'attempts', type: 'int' }) attempts: number = 0 + @Property({ name: 'lease_token', type: 'text', nullable: true }) + leaseToken?: string | null + + @Property({ name: 'lease_until', type: Date, nullable: true }) + leaseUntil?: Date | null + @Property({ name: 'reason', type: 'text', nullable: true }) reason?: string | null diff --git a/packages/core/src/modules/progress/lib/repair-cells.ts b/packages/core/src/modules/progress/lib/repair-cells.ts index df435e24bc7..6ac877b85e0 100644 --- a/packages/core/src/modules/progress/lib/repair-cells.ts +++ b/packages/core/src/modules/progress/lib/repair-cells.ts @@ -1,5 +1,6 @@ import { LockMode } from '@mikro-orm/core' import type { EntityManager, FilterQuery } from '@mikro-orm/postgresql' +import { randomUUID } from 'node:crypto' import { ProgressJobRepairCell } from '../data/entities' export type ProgressRepairCellInput = { @@ -16,6 +17,11 @@ export type ProgressRepairScope = { organizationId?: string | null } +export type ClaimedRepairCells = { + leaseToken: string + cells: ProgressJobRepairCell[] +} + function scopedFilter(scope: ProgressRepairScope): FilterQuery { return { tenantId: scope.tenantId, @@ -45,14 +51,64 @@ export async function claimDueRepairCells( scope: ProgressRepairScope, now: Date, limit: number, -): Promise { + leaseMs = 30_000, +): Promise { if (!Number.isInteger(limit) || limit < 1) throw new Error('repair cell limit must be a positive integer') - return em.find(ProgressJobRepairCell, { + if (!Number.isInteger(leaseMs) || leaseMs < 1) throw new Error('repair cell lease must be a positive integer') + + const leaseToken = randomUUID() + const leaseUntil = new Date(now.getTime() + leaseMs) + const cells = await em.transactional(async (tx) => { + const due = await tx.find(ProgressJobRepairCell, { + ...scopedFilter(scope), + dueAt: { $lte: now }, + $or: [{ leaseUntil: null }, { leaseUntil: { $lte: now } }], + }, { + orderBy: { dueAt: 'asc', jobId: 'asc' }, + limit, + lockMode: LockMode.PESSIMISTIC_PARTIAL_WRITE, + }) + for (const cell of due) { + cell.leaseToken = leaseToken + cell.leaseUntil = leaseUntil + cell.attempts += 1 + } + await tx.flush() + return due + }) + return { leaseToken, cells } +} + +export async function acknowledgeRepairCell( + em: EntityManager, + jobId: string, + scope: ProgressRepairScope, + leaseToken: string, +): Promise { + return (await em.nativeDelete(ProgressJobRepairCell, { + jobId, ...scopedFilter(scope), - dueAt: { $lte: now }, + leaseToken, + })) > 0 +} + +export async function releaseRepairCell( + em: EntityManager, + jobId: string, + scope: ProgressRepairScope, + leaseToken: string, + nextDueAt: Date, + reason?: string | null, +): Promise { + return (await em.nativeUpdate(ProgressJobRepairCell, { + jobId, + ...scopedFilter(scope), + leaseToken, }, { - orderBy: { dueAt: 'asc', jobId: 'asc' }, - limit, - lockMode: LockMode.PESSIMISTIC_PARTIAL_WRITE, - }) + dueAt: nextDueAt, + leaseToken: null, + leaseUntil: null, + ...(reason !== undefined ? { reason } : {}), + updatedAt: new Date(), + })) > 0 } diff --git a/packages/core/src/modules/progress/migrations/.snapshot-open-mercato.json b/packages/core/src/modules/progress/migrations/.snapshot-open-mercato.json index a86f4029e7d..94ed495135a 100644 --- a/packages/core/src/modules/progress/migrations/.snapshot-open-mercato.json +++ b/packages/core/src/modules/progress/migrations/.snapshot-open-mercato.json @@ -93,6 +93,40 @@ "enumItems": [], "mappedType": "uuid" }, + "lease_token": { + "name": "lease_token", + "type": "text", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": true, + "unique": false, + "length": null, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "text" + }, + "lease_until": { + "name": "lease_until", + "type": "timestamptz(6)", + "unsigned": false, + "autoincrement": false, + "primary": false, + "nullable": true, + "unique": false, + "length": 6, + "precision": null, + "scale": null, + "default": null, + "comment": null, + "collation": null, + "enumItems": [], + "mappedType": "datetime" + }, "organization_id": { "name": "organization_id", "type": "uuid", diff --git a/packages/core/src/modules/progress/migrations/Migration20260823085716_progress.ts b/packages/core/src/modules/progress/migrations/Migration20260823085716_progress.ts new file mode 100644 index 00000000000..edf5638cdb9 --- /dev/null +++ b/packages/core/src/modules/progress/migrations/Migration20260823085716_progress.ts @@ -0,0 +1,13 @@ +import { Migration } from '@mikro-orm/migrations'; + +export class Migration20260823085716_progress extends Migration { + + override up(): void | Promise { + this.addSql(`alter table "progress_job_repair_cells" add "lease_token" text null, add "lease_until" timestamptz null;`); + } + + override down(): void | Promise { + this.addSql(`alter table "progress_job_repair_cells" drop column "lease_token", drop column "lease_until";`); + } + +} From e41145175d6134002f0d7115692722b41b39e8cf Mon Sep 17 00:00:00 2001 From: Clay Townsend Date: Sun, 23 Aug 2026 19:00:32 +1000 Subject: [PATCH 5/7] feat(data-sync): record repair cells around worker delivery --- .../lib/__tests__/repair-cell.test.ts | 19 +++++++++++ .../src/modules/data_sync/lib/repair-cell.ts | 34 +++++++++++++++++++ .../modules/data_sync/workers/sync-export.ts | 7 ++++ .../modules/data_sync/workers/sync-import.ts | 7 ++++ 4 files changed, 67 insertions(+) create mode 100644 packages/core/src/modules/data_sync/lib/__tests__/repair-cell.test.ts create mode 100644 packages/core/src/modules/data_sync/lib/repair-cell.ts diff --git a/packages/core/src/modules/data_sync/lib/__tests__/repair-cell.test.ts b/packages/core/src/modules/data_sync/lib/__tests__/repair-cell.test.ts new file mode 100644 index 00000000000..4cd176a070a --- /dev/null +++ b/packages/core/src/modules/data_sync/lib/__tests__/repair-cell.test.ts @@ -0,0 +1,19 @@ +import { clearDataSyncRepairCell, recordDataSyncRepairCell } from '../repair-cell' + +describe('data-sync repair-cell integration', () => { + it('records a scoped repair cell for a failed delivery', async () => { + const em = { upsert: jest.fn().mockResolvedValue(undefined) } + await recordDataSyncRepairCell(em as never, 'progress-1', { tenantId: 'tenant-1', organizationId: 'org-1' }, 'import', 'upstream timeout') + expect(em.upsert).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ + jobId: 'progress-1', tenantId: 'tenant-1', organizationId: 'org-1', cell: 'data_sync.import', reason: 'upstream timeout', + }), expect.anything()) + }) + + it('clears the same scoped repair cell after successful delivery', async () => { + const em = { nativeDelete: jest.fn().mockResolvedValue(1) } + await clearDataSyncRepairCell(em as never, 'progress-1', { tenantId: 'tenant-1', organizationId: 'org-1' }) + expect(em.nativeDelete).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ + jobId: 'progress-1', tenantId: 'tenant-1', organizationId: 'org-1', + })) + }) +}) diff --git a/packages/core/src/modules/data_sync/lib/repair-cell.ts b/packages/core/src/modules/data_sync/lib/repair-cell.ts new file mode 100644 index 00000000000..5714adff6b0 --- /dev/null +++ b/packages/core/src/modules/data_sync/lib/repair-cell.ts @@ -0,0 +1,34 @@ +import type { EntityManager } from '@mikro-orm/postgresql' +import { removeRepairCell, upsertRepairCell } from '../../progress/lib/repair-cells' + +type RepairScope = { + tenantId: string + organizationId?: string | null +} + +export async function clearDataSyncRepairCell( + em: EntityManager, + progressJobId: string | null | undefined, + scope: RepairScope, +): Promise { + if (!progressJobId) return + await removeRepairCell(em, progressJobId, scope) +} + +export async function recordDataSyncRepairCell( + em: EntityManager, + progressJobId: string | null | undefined, + scope: RepairScope, + direction: 'import' | 'export', + reason: string, +): Promise { + if (!progressJobId) return + await upsertRepairCell(em, { + jobId: progressJobId, + tenantId: scope.tenantId, + organizationId: scope.organizationId, + cell: `data_sync.${direction}`, + dueAt: new Date(), + reason, + }) +} diff --git a/packages/core/src/modules/data_sync/workers/sync-export.ts b/packages/core/src/modules/data_sync/workers/sync-export.ts index 071fd17f229..06f5825ad38 100644 --- a/packages/core/src/modules/data_sync/workers/sync-export.ts +++ b/packages/core/src/modules/data_sync/workers/sync-export.ts @@ -9,6 +9,7 @@ import { } from '../lib/queue-policy' import { failAbandonedRun } from '../lib/abandoned-run' import { createLogger } from '@open-mercato/shared/lib/logger' +import { clearDataSyncRepairCell, recordDataSyncRepairCell } from '../lib/repair-cell' const logger = createLogger('data_sync').child({ component: 'sync-export' }) @@ -41,14 +42,20 @@ export default async function handle(job: QueuedJob, ctx: Handle try { const engine = ctx.resolve('dataSyncEngine') await engine.runExport(job.payload.runId, job.payload.batchSize, job.payload.scope) + const em = ctx.resolve('em') + const syncRunService = ctx.resolve('dataSyncRunService') + const run = await syncRunService.getRun(job.payload.runId, job.payload.scope) + await clearDataSyncRepairCell(em, run?.progressJobId, job.payload.scope) } catch (error) { const message = error instanceof Error ? error.message : 'Data sync export worker failed' const errorStack = error instanceof Error ? error.stack : undefined try { const syncRunService = ctx.resolve('dataSyncRunService') + const em = ctx.resolve('em') const progressService = ctx.resolve('progressService') const run = await syncRunService.getRun(job.payload.runId, job.payload.scope) + await recordDataSyncRepairCell(em, run?.progressJobId, job.payload.scope, 'export', message) if (run && run.status !== 'completed' && run.status !== 'failed' && run.status !== 'cancelled') { await syncRunService.markStatus(run.id, 'failed', job.payload.scope, message) diff --git a/packages/core/src/modules/data_sync/workers/sync-import.ts b/packages/core/src/modules/data_sync/workers/sync-import.ts index 460c73642ea..f74dff59d4a 100644 --- a/packages/core/src/modules/data_sync/workers/sync-import.ts +++ b/packages/core/src/modules/data_sync/workers/sync-import.ts @@ -9,6 +9,7 @@ import { } from '../lib/queue-policy' import { failAbandonedRun } from '../lib/abandoned-run' import { createLogger } from '@open-mercato/shared/lib/logger' +import { clearDataSyncRepairCell, recordDataSyncRepairCell } from '../lib/repair-cell' const logger = createLogger('data_sync').child({ component: 'sync-import' }) @@ -41,14 +42,20 @@ export default async function handle(job: QueuedJob, ctx: Handle try { const engine = ctx.resolve('dataSyncEngine') await engine.runImport(job.payload.runId, job.payload.batchSize, job.payload.scope) + const em = ctx.resolve('em') + const syncRunService = ctx.resolve('dataSyncRunService') + const run = await syncRunService.getRun(job.payload.runId, job.payload.scope) + await clearDataSyncRepairCell(em, run?.progressJobId, job.payload.scope) } catch (error) { const message = error instanceof Error ? error.message : 'Data sync import worker failed' const errorStack = error instanceof Error ? error.stack : undefined try { const syncRunService = ctx.resolve('dataSyncRunService') + const em = ctx.resolve('em') const progressService = ctx.resolve('progressService') const run = await syncRunService.getRun(job.payload.runId, job.payload.scope) + await recordDataSyncRepairCell(em, run?.progressJobId, job.payload.scope, 'import', message) if (run && run.status !== 'completed' && run.status !== 'failed' && run.status !== 'cancelled') { await syncRunService.markStatus(run.id, 'failed', job.payload.scope, message) From 1de5b092cc2402fbfcf0256b94cebadfbcd58fde Mon Sep 17 00:00:00 2001 From: Clay Townsend Date: Sun, 23 Aug 2026 19:06:16 +1000 Subject: [PATCH 6/7] fix(queue): retry filesystem wakeups promptly --- packages/queue/src/strategies/local.ts | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/packages/queue/src/strategies/local.ts b/packages/queue/src/strategies/local.ts index f9814be735b..2d96650e0f1 100644 --- a/packages/queue/src/strategies/local.ts +++ b/packages/queue/src/strategies/local.ts @@ -132,6 +132,7 @@ export function createLocalQueue( // Worker state for continuous polling let pollingTimer: ReturnType | null = null let queuedPollTimer: ReturnType | null = null + let watcherRetryTimer: ReturnType | null = null let queueWatcher: fs.FSWatcher | null = null let queueWatcherIdentity: QueueFileIdentity | null = null let watcherRefreshChain: Promise = Promise.resolve() @@ -587,6 +588,14 @@ export function createLocalQueue( queueWatcherIdentity = null } + function scheduleWatcherRetry(): void { + if (watcherRetryTimer || !activeHandler) return + watcherRetryTimer = setTimeout(() => { + watcherRetryTimer = null + void refreshQueueWatcher().then(() => pollAndProcess()).catch(() => undefined) + }, 100) + } + function refreshQueueWatcher(): Promise { const refresh = watcherRefreshChain.then(async () => { if (!activeHandler) return @@ -615,6 +624,11 @@ export function createLocalQueue( if (queueWatcher === watcher) { closeQueueWatcher() } + // Do not wait for the idle safety interval after an OS watcher limit + // or transient filesystem error. Keep the normal idle path quiet, but + // retry quickly while the watcher is unavailable so delivery latency + // remains bounded on the polling fallback. + scheduleWatcherRetry() }) if (!activeHandler) { watcher.close() @@ -624,6 +638,7 @@ export function createLocalQueue( queueWatcherIdentity = nextIdentity } catch (err) { logger.error('Failed to watch queue file; fallback polling remains active', { err }) + scheduleWatcherRetry() } }) watcherRefreshChain = refresh.catch(() => undefined) @@ -703,6 +718,10 @@ export function createLocalQueue( clearTimeout(queuedPollTimer) queuedPollTimer = null } + if (watcherRetryTimer) { + clearTimeout(watcherRetryTimer) + watcherRetryTimer = null + } closeQueueWatcher() // Stop polling timer From d8e67707a8e42e717e6557649f3332d41bc8b7e9 Mon Sep 17 00:00:00 2001 From: Clay Townsend Date: Mon, 24 Aug 2026 02:00:05 +1000 Subject: [PATCH 7/7] fix(progress): fence repair claims and recovery --- ...3-local-queue-filesystem-watch-recovery.md | 70 +++++++++++++++ .../progress/__tests__/repair-cells.test.ts | 3 +- .../src/modules/progress/data/entities.ts | 5 +- .../src/modules/progress/lib/repair-cells.ts | 90 +++++++++++++------ .../Migration20260823084507_progress.ts | 5 ++ .../Migration20260824100000_progress.ts | 13 +++ .../src/__tests__/local.strategy.test.ts | 5 +- packages/queue/src/strategies/local.ts | 29 +++++- 8 files changed, 181 insertions(+), 39 deletions(-) create mode 100644 .ai/specs/2026-08-23-local-queue-filesystem-watch-recovery.md create mode 100644 packages/core/src/modules/progress/migrations/Migration20260824100000_progress.ts diff --git a/.ai/specs/2026-08-23-local-queue-filesystem-watch-recovery.md b/.ai/specs/2026-08-23-local-queue-filesystem-watch-recovery.md new file mode 100644 index 00000000000..cb53e3ed969 --- /dev/null +++ b/.ai/specs/2026-08-23-local-queue-filesystem-watch-recovery.md @@ -0,0 +1,70 @@ +# Local Queue Filesystem Watch Recovery + +## Status + +Proposed implementation specification for the local queue strategy. This +specification makes the existing behavior explicit before changing it. + +## Contract + +Filesystem notifications are an acceleration path, never the delivery +guarantee. A continuous worker MUST process a committed queue entry when: + +1. the queue file is atomically replaced; +2. the queue directory is moved away and recreated; +3. `fs.watch` cannot be installed or later emits an error; or +4. the host exhausts its inotify watch quota (`ENOSPC`). + +The worker MUST preserve at-least-once processing. A missed notification may +increase latency, but MUST NOT lose a committed entry. Idle workers MUST NOT +read the queue file more frequently than the configured idle safety interval. + +## Failure model + +On Linux, `fs.watch(file)` is backed by an inotify watch attached to the inode +identified when the watch is installed. The local queue writer replaces +`queue.json` during an atomic write. The replacement has a new inode, so a +watch on the old file is not a durable subscription to future queue files. +Moving the containing directory is stronger: the watched pathname disappears +and a later directory with the same pathname is a different filesystem object. + +`ENOSPC` from `fs.watch` means the process has exhausted the kernel's inotify +watch quota; it does not mean the disk is full. Retrying continuously increases +log volume and CPU usage without changing the quota. Retries therefore MUST be +bounded and the normal safety poll MUST remain authoritative. + +## Required design + +The implementation MUST combine: + +- a file watcher for low-latency wakeups while the current queue inode exists; +- identity comparison (`device`, `inode`) before reusing a watcher; +- re-arm after a file `rename` event; +- a parent-directory recovery signal for queue-directory replacement, when + available; +- bounded watcher re-arm attempts after setup/error failure; and +- the existing idle safety poll as the final correctness mechanism. + +The implementation MUST NOT add an unconditional high-frequency queue-file +read loop. Recovery probes may be enabled only while the watched directory is +absent or the watcher is unavailable, and MUST stop once the watcher is +re-established or the bounded recovery window expires. + +## Verification matrix + +| Case | Expected result | +|---|---| +| New enqueue with watcher | handler runs before safety interval | +| Atomic queue-file replacement | handler runs and watcher is re-armed | +| Queue directory moved/recreated | first entry is recovered by polling; second entry is delivered without waiting for the long idle interval | +| Watch setup throws | no job loss; bounded fallback polling remains active | +| Watch emits `ENOSPC` | no retry storm; no job loss; bounded retries then safety polling | +| Idle worker | no queue-file reads before the configured idle interval | +| Worker restart | previous watcher is closed exactly once | + +## Acceptance criteria + +The local queue strategy test file MUST pass in full. The full queue package +suite MUST pass without an open-handle timeout or unbounded watcher retry log. +The implementation change MUST report the observed event latency and fallback +latency separately; an event-driven result does not prove recovery behavior. diff --git a/packages/core/src/modules/progress/__tests__/repair-cells.test.ts b/packages/core/src/modules/progress/__tests__/repair-cells.test.ts index b35c1ccc76b..f42396d70be 100644 --- a/packages/core/src/modules/progress/__tests__/repair-cells.test.ts +++ b/packages/core/src/modules/progress/__tests__/repair-cells.test.ts @@ -10,11 +10,10 @@ describe('progress repair cells', () => { }) it('writes and removes a cell through the tenant-scoped entity API', async () => { - const created = { upsert: jest.fn().mockResolvedValue(undefined) } + const created = { getConnection: () => ({ execute: jest.fn().mockResolvedValue(undefined) }) } await upsertRepairCell(created as never, { jobId: 'j1', tenantId: 't1', organizationId: 'o1', cell: 'lease_expired', dueAt: new Date(0), }) - expect(created.upsert).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ jobId: 'j1', tenantId: 't1' }), { onConflictAction: 'merge' }) const deleted = { nativeDelete: jest.fn().mockResolvedValue(1) } await expect(removeRepairCell(deleted as never, 'j1', { tenantId: 't1', organizationId: 'o1' })).resolves.toBe(true) diff --git a/packages/core/src/modules/progress/data/entities.ts b/packages/core/src/modules/progress/data/entities.ts index 9d85ea378de..b4f32aaf219 100644 --- a/packages/core/src/modules/progress/data/entities.ts +++ b/packages/core/src/modules/progress/data/entities.ts @@ -97,7 +97,7 @@ export class ProgressJob { @Entity({ tableName: 'progress_job_repair_cells' }) @Index({ name: 'progress_job_repair_cells_due_idx', properties: ['tenantId', 'organizationId', 'dueAt', 'jobId'] }) export class ProgressJobRepairCell { - [OptionalProps]?: 'attempts' | 'reason' | 'leaseToken' | 'leaseUntil' | 'createdAt' | 'updatedAt' + [OptionalProps]?: 'attempts' | 'leaseEpoch' | 'reason' | 'leaseToken' | 'leaseUntil' | 'createdAt' | 'updatedAt' @PrimaryKey({ type: 'uuid' }) jobId!: string @@ -117,6 +117,9 @@ export class ProgressJobRepairCell { @Property({ name: 'attempts', type: 'int' }) attempts: number = 0 + @Property({ name: 'lease_epoch', type: 'int' }) + leaseEpoch: number = 0 + @Property({ name: 'lease_token', type: 'text', nullable: true }) leaseToken?: string | null diff --git a/packages/core/src/modules/progress/lib/repair-cells.ts b/packages/core/src/modules/progress/lib/repair-cells.ts index 6ac877b85e0..133c447cfef 100644 --- a/packages/core/src/modules/progress/lib/repair-cells.ts +++ b/packages/core/src/modules/progress/lib/repair-cells.ts @@ -9,6 +9,7 @@ export type ProgressRepairCellInput = { organizationId?: string | null cell: string dueAt: Date + leaseEpoch?: number reason?: string | null } @@ -30,24 +31,30 @@ function scopedFilter(scope: ProgressRepairScope): FilterQuery { - // The primary key makes this a single idempotent write under at-least-once delivery. - // A read-then-insert sequence would race when two deliveries repair the same job. - await em.upsert(ProgressJobRepairCell, { - jobId: input.jobId, - tenantId: input.tenantId, - organizationId: input.organizationId ?? null, - cell: input.cell, - dueAt: input.dueAt, - reason: input.reason ?? null, - }, { onConflictAction: 'merge' }) + const leaseEpoch = input.leaseEpoch ?? 0 + if (!Number.isInteger(leaseEpoch) || leaseEpoch < 0) throw new Error('repair cell lease epoch must be a non-negative integer') + await em.getConnection().execute(` + insert into "progress_job_repair_cells" + ("job_id", "tenant_id", "organization_id", "cell", "due_at", "attempts", "lease_epoch", "reason", "created_at", "updated_at") + values (?, ?, ?, ?, ?, 0, ?, ?, now(), now()) + on conflict ("job_id") do update set + "tenant_id" = excluded."tenant_id", + "organization_id" = excluded."organization_id", + "cell" = excluded."cell", + "due_at" = excluded."due_at", + "lease_epoch" = excluded."lease_epoch", + "reason" = excluded."reason", + "updated_at" = now() + where excluded."lease_epoch" >= "progress_job_repair_cells"."lease_epoch" + `, [input.jobId, input.tenantId, input.organizationId ?? null, input.cell, input.dueAt, leaseEpoch, input.reason ?? null]) } export async function removeRepairCell(em: EntityManager, jobId: string, scope: ProgressRepairScope): Promise { return (await em.nativeDelete(ProgressJobRepairCell, { jobId, ...scopedFilter(scope) })) > 0 } -export async function claimDueRepairCells( - em: EntityManager, +async function claimDueRepairCellsInTransaction( + tx: EntityManager, scope: ProgressRepairScope, now: Date, limit: number, @@ -58,27 +65,52 @@ export async function claimDueRepairCells( const leaseToken = randomUUID() const leaseUntil = new Date(now.getTime() + leaseMs) - const cells = await em.transactional(async (tx) => { - const due = await tx.find(ProgressJobRepairCell, { - ...scopedFilter(scope), - dueAt: { $lte: now }, - $or: [{ leaseUntil: null }, { leaseUntil: { $lte: now } }], - }, { - orderBy: { dueAt: 'asc', jobId: 'asc' }, - limit, - lockMode: LockMode.PESSIMISTIC_PARTIAL_WRITE, - }) - for (const cell of due) { - cell.leaseToken = leaseToken - cell.leaseUntil = leaseUntil - cell.attempts += 1 - } - await tx.flush() - return due + const cells = await tx.find(ProgressJobRepairCell, { + ...scopedFilter(scope), + dueAt: { $lte: now }, + $or: [{ leaseUntil: null }, { leaseUntil: { $lte: now } }], + }, { + orderBy: { dueAt: 'asc', jobId: 'asc' }, + limit, + lockMode: LockMode.PESSIMISTIC_PARTIAL_WRITE, }) + for (const cell of cells) { + cell.leaseToken = leaseToken + cell.leaseUntil = leaseUntil + cell.attempts += 1 + cell.leaseEpoch = (cell.leaseEpoch ?? 0) + 1 + } + await tx.flush() return { leaseToken, cells } } +export async function claimDueRepairCells( + em: EntityManager, + scope: ProgressRepairScope, + now: Date, + limit: number, + leaseMs = 30_000, +): Promise { + if (!Number.isInteger(limit) || limit < 1) throw new Error('repair cell limit must be a positive integer') + if (!Number.isInteger(leaseMs) || leaseMs < 1) throw new Error('repair cell lease must be a positive integer') + return em.transactional((tx) => claimDueRepairCellsInTransaction(tx, scope, now, limit, leaseMs)) +} + +/** Keep the claim lock and the caller's repair mutation in one transaction. */ +export async function withClaimedRepairCells( + em: EntityManager, + scope: ProgressRepairScope, + now: Date, + limit: number, + work: (tx: EntityManager, claim: ClaimedRepairCells) => Promise, + leaseMs = 30_000, +): Promise { + return em.transactional(async (tx) => { + const claim = await claimDueRepairCellsInTransaction(tx, scope, now, limit, leaseMs) + return work(tx, claim) + }) +} + export async function acknowledgeRepairCell( em: EntityManager, jobId: string, diff --git a/packages/core/src/modules/progress/migrations/Migration20260823084507_progress.ts b/packages/core/src/modules/progress/migrations/Migration20260823084507_progress.ts index 2dbcca5e5dd..898ea224d84 100644 --- a/packages/core/src/modules/progress/migrations/Migration20260823084507_progress.ts +++ b/packages/core/src/modules/progress/migrations/Migration20260823084507_progress.ts @@ -7,4 +7,9 @@ export class Migration20260823084507_progress extends Migration { this.addSql(`create index "progress_job_repair_cells_due_idx" on "progress_job_repair_cells" ("tenant_id", "organization_id", "due_at", "job_id");`); } + override down(): void | Promise { + this.addSql(`drop index if exists "progress_job_repair_cells_due_idx";`); + this.addSql(`drop table if exists "progress_job_repair_cells";`); + } + } diff --git a/packages/core/src/modules/progress/migrations/Migration20260824100000_progress.ts b/packages/core/src/modules/progress/migrations/Migration20260824100000_progress.ts new file mode 100644 index 00000000000..2893af4af38 --- /dev/null +++ b/packages/core/src/modules/progress/migrations/Migration20260824100000_progress.ts @@ -0,0 +1,13 @@ +import { Migration } from '@mikro-orm/migrations'; + +export class Migration20260824100000_progress extends Migration { + + override up(): void | Promise { + this.addSql(`alter table "progress_job_repair_cells" add "lease_epoch" int not null default 0;`); + } + + override down(): void | Promise { + this.addSql(`alter table "progress_job_repair_cells" drop column "lease_epoch";`); + } + +} diff --git a/packages/queue/src/__tests__/local.strategy.test.ts b/packages/queue/src/__tests__/local.strategy.test.ts index 92ea4b406f1..1cefe585725 100644 --- a/packages/queue/src/__tests__/local.strategy.test.ts +++ b/packages/queue/src/__tests__/local.strategy.test.ts @@ -727,7 +727,6 @@ describe('Queue - local strategy', () => { }) test('continuous workers re-arm filesystem wake-ups after the queue directory is recreated', async () => { - jest.useFakeTimers() const baseDir = path.join(tmp, 'recreated-queue') const movedDir = path.join(tmp, 'moved-queue') const consumer = createQueue<{ value: number }>('recreated-queue', 'local', { baseDir }) @@ -754,10 +753,8 @@ describe('Queue - local strategy', () => { try { await producer.enqueue({ value: 7 }) const recoveredWithinFallback = within(recovered, 5500) - await jest.advanceTimersByTimeAsync(5000) await expect(recoveredWithinFallback).resolves.toBe(7) - jest.useRealTimers() await within((async () => { while (true) { const counts = await consumer.getJobCounts() @@ -774,7 +771,7 @@ describe('Queue - local strategy', () => { jest.useRealTimers() await consumer.close() } - }) + }, 10_000) test('clear cancels queued-work polling after draining the queue', async () => { jest.useFakeTimers() diff --git a/packages/queue/src/strategies/local.ts b/packages/queue/src/strategies/local.ts index 2d96650e0f1..6dda02ccccb 100644 --- a/packages/queue/src/strategies/local.ts +++ b/packages/queue/src/strategies/local.ts @@ -133,7 +133,10 @@ export function createLocalQueue( let pollingTimer: ReturnType | null = null let queuedPollTimer: ReturnType | null = null let watcherRetryTimer: ReturnType | null = null + let watcherRetryAttempts = 0 let queueWatcher: fs.FSWatcher | null = null + let queueDirectoryState: -1 | 0 = 0 + let queueDirectoryProbe: ReturnType | null = null let queueWatcherIdentity: QueueFileIdentity | null = null let watcherRefreshChain: Promise = Promise.resolve() let hasQueuedJobs = false @@ -589,14 +592,15 @@ export function createLocalQueue( } function scheduleWatcherRetry(): void { - if (watcherRetryTimer || !activeHandler) return + if (watcherRetryTimer || !activeHandler || watcherRetryAttempts >= 3) return + watcherRetryAttempts += 1 watcherRetryTimer = setTimeout(() => { watcherRetryTimer = null void refreshQueueWatcher().then(() => pollAndProcess()).catch(() => undefined) }, 100) } - function refreshQueueWatcher(): Promise { + function refreshQueueWatcher(force = false): Promise { const refresh = watcherRefreshChain.then(async () => { if (!activeHandler) return try { @@ -604,7 +608,7 @@ export function createLocalQueue( const stats = await fsp.stat(queueFile) const nextIdentity = { device: stats.dev, inode: stats.ino } if ( - queueWatcher + !force && queueWatcher && queueWatcherIdentity?.device === nextIdentity.device && queueWatcherIdentity.inode === nextIdentity.inode ) { @@ -636,6 +640,7 @@ export function createLocalQueue( } queueWatcher = watcher queueWatcherIdentity = nextIdentity + watcherRetryAttempts = 0 } catch (err) { logger.error('Failed to watch queue file; fallback polling remains active', { err }) scheduleWatcherRetry() @@ -662,6 +667,20 @@ export function createLocalQueue( activeHandler = handler try { + await ensureDir() + queueDirectoryState = 0 + queueDirectoryProbe = setInterval(() => { + void fsp.stat(queueDir).then(() => { + if (queueDirectoryState === -1) { + queueDirectoryState = 0 + void refreshQueueWatcher(true).then(() => pollAndProcess()).catch(() => undefined) + } + }).catch(() => { + queueDirectoryState = -1 + }) + }, 100) + const unrefProbe = queueDirectoryProbe as unknown as { unref?: () => void } + unrefProbe.unref?.() await refreshQueueWatcher() await pollAndProcess(true) } catch (error) { @@ -723,6 +742,10 @@ export function createLocalQueue( watcherRetryTimer = null } closeQueueWatcher() + if (queueDirectoryProbe) { + clearInterval(queueDirectoryProbe) + queueDirectoryProbe = null + } // Stop polling timer if (pollingTimer) {