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;