feat(progress): add idempotent repair-cell persistence

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

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',
+    }))
+  })
+})

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<void> {
+  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<void> {
+  if (!progressJobId) return
+  await upsertRepairCell(em, {
+    jobId: progressJobId,
+    tenantId: scope.tenantId,
+    organizationId: scope.organizationId,
+    cell: `data_sync.${direction}`,
+    dueAt: new Date(),
+    reason,
+  })
+}

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<SyncJobPayload>, ctx: Handle
   try {
     const engine = ctx.resolve<SyncEngine>('dataSyncEngine')
     await engine.runExport(job.payload.runId, job.payload.batchSize, job.payload.scope)
+    const em = ctx.resolve<import('@mikro-orm/postgresql').EntityManager>('em')
+    const syncRunService = ctx.resolve<SyncRunService>('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<SyncRunService>('dataSyncRunService')
+      const em = ctx.resolve<import('@mikro-orm/postgresql').EntityManager>('em')
       const progressService = ctx.resolve<ProgressService>('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)

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<SyncJobPayload>, ctx: Handle
   try {
     const engine = ctx.resolve<SyncEngine>('dataSyncEngine')
     await engine.runImport(job.payload.runId, job.payload.batchSize, job.payload.scope)
+    const em = ctx.resolve<import('@mikro-orm/postgresql').EntityManager>('em')
+    const syncRunService = ctx.resolve<SyncRunService>('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<SyncRunService>('dataSyncRunService')
+      const em = ctx.resolve<import('@mikro-orm/postgresql').EntityManager>('em')
       const progressService = ctx.resolve<ProgressService>('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)

packages/core/src/modules/progress/__tests__/repair-cells.test.ts

@@ -0,0 +1,46 @@
+import { describe, expect, it } from '@jest/globals'
+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 () => {
+    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),
+    })
+
+    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' }))
+  })
+
+  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<unknown>) => 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 }))
+  })
+})

packages/core/src/modules/progress/data/entities.ts

@@ -93,3 +93,45 @@ 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' | 'leaseEpoch' | 'reason' | 'leaseToken' | 'leaseUntil' | '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: 'lease_epoch', type: 'int' })
+  leaseEpoch: 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
+
+  @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()
+}