From 0a85688ac3f12c1c281238dc97973f5b69b09c20 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 20:07:05 +0000 Subject: [PATCH] fix(stream): refresh host-action retention on each retry attempt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex finding: unacknowledged terminal host-action evidence was capped at the 24h pause TTL, so a host dependency (Mongo) unreachable for longer than that let the Redis key — and its pending marker — expire with no generation-fenced acknowledgement, stranding the ScheduleRun where no reconciler is armed. Measure retention from the LAST retry rather than from the terminal transition: enumerating a pending host action IS the retry attempt, so both stores refresh its retention as they hand it to the hook (Redis re-EXPIREs the job key; the in-memory store stamps terminalHostActionRefreshedAt and bounds from it). Evidence therefore survives as long as some replica is still actively retrying, while a deployment that stops sweeping entirely still lets it age out — so this does not reintroduce the unbounded leak the cap existed to prevent. Test: after a failed hook, a later cleanup pass keeps the marker pending and moves its retention basis forward. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq --- .../stream/__tests__/pendingAction.spec.ts | 21 +++++++++++++++++++ .../implementations/InMemoryJobStore.ts | 8 +++++-- .../stream/implementations/RedisJobStore.ts | 10 +++++++++ .../api/src/stream/interfaces/IJobStore.ts | 7 +++++++ 4 files changed, 44 insertions(+), 2 deletions(-) diff --git a/packages/api/src/stream/__tests__/pendingAction.spec.ts b/packages/api/src/stream/__tests__/pendingAction.spec.ts index 0a0c034dd6..cdca10c1e1 100644 --- a/packages/api/src/stream/__tests__/pendingAction.spec.ts +++ b/packages/api/src/stream/__tests__/pendingAction.spec.ts @@ -1085,6 +1085,27 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () expect(await manager.getJobStatus(streamId)).toBe('aborted'); }); + test('retention is refreshed on each retry so a long host outage keeps the evidence', async () => { + const streamId = 'stream-host-refresh'; + await pauseScheduled(streamId); + manager.setApprovalExpiredHandler(jest.fn().mockRejectedValue(new Error('mongo down'))); + expect(await manager.expireApproval(streamId)).toBe(true); + + const before = (await jobStore.getJob(streamId))?.terminalHostActionRefreshedAt; + + // A later cleanup pass re-enumerates it for retry; the hook still fails. + await sweep(manager); + + const after = await jobStore.getJob(streamId); + // Still pending, and its retention basis moved forward — so the bounded retention + // window measures from the LAST retry, not from when the approval expired. + expect(after?.terminalHostActionPending).toBe(true); + expect(after?.terminalHostActionRefreshedAt).toEqual(expect.any(Number)); + if (before != null) { + expect(after!.terminalHostActionRefreshedAt!).toBeGreaterThanOrEqual(before); + } + }); + test('a non-scheduled job with a no-op host hook accumulates no marker', async () => { const streamId = 'stream-host-nonsched'; await manager.createJob(streamId, 'user-1'); // no schedule metadata diff --git a/packages/api/src/stream/implementations/InMemoryJobStore.ts b/packages/api/src/stream/implementations/InMemoryJobStore.ts index 227d711cbb..b449cf0a5d 100644 --- a/packages/api/src/stream/implementations/InMemoryJobStore.ts +++ b/packages/api/src/stream/implementations/InMemoryJobStore.ts @@ -1024,8 +1024,12 @@ export class InMemoryJobStore implements IJobStoreV2 { async getTerminalHostActionJobs(): Promise { const pending: SerializableJobData[] = []; + const now = Date.now(); for (const job of this.jobs.values()) { if (job.terminalHostActionPending === true) { + // Enumerating IS the retry attempt: refresh retention so evidence outlives a host + // dependency that is unreachable for longer than the retention window. + job.terminalHostActionRefreshedAt = now; pending.push(job); } } @@ -1110,8 +1114,8 @@ export class InMemoryJobStore implements IJobStoreV2 { // but only within a bounded window, so a permanently-failing hook cannot leak. const hostActionHeld = job.terminalHostActionPending === true && - job.completedAt != null && - now - job.completedAt <= HOST_ACTION_RETENTION_MS; + now - (job.terminalHostActionRefreshedAt ?? job.completedAt ?? 0) <= + HOST_ACTION_RETENTION_MS; if ( isFinished && job.completedAt && diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index 38e2c2f6eb..270c81d402 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -2722,6 +2722,16 @@ export class RedisJobStore implements IJobStoreV2 { if (stale.length > 0) { await this.redis.srem(KEYS.terminalHostActionJobs, ...stale).catch(() => undefined); } + // Enumerating IS the retry attempt: extend each pending job's TTL so unacknowledged + // host-action evidence outlives a host dependency (e.g. Mongo) that stays unreachable + // longer than the retention window. A deployment that stops sweeping lets it age out. + if (pending.length > 0 && this.ttl.requiresAction > 0) { + await Promise.all( + pending.map((job) => + this.redis.expire(KEYS.job(job.streamId), this.ttl.requiresAction).catch(() => undefined), + ), + ); + } return pending; } diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index 99a60a2af6..05aff508fe 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -140,6 +140,13 @@ export interface SerializableJobData { * no action clears it immediately on its no-op success, so nothing accumulates. */ terminalHostActionPending?: boolean; + /** + * Last time a cleanup pass enumerated this pending host action for retry. Retention is + * measured from this rather than `completedAt`, so evidence survives as long as some + * replica is still actively retrying the hook (e.g. Mongo unreachable for days), while a + * deployment that stops retrying entirely still lets it age out instead of leaking. + */ + terminalHostActionRefreshedAt?: number; /** Serialized final event for replay */ finalEvent?: string;