fix(stream): refresh host-action retention on each retry attempt

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq
This commit is contained in:
Claude 2026-08-18 20:07:05 +00:00
parent e19bb25b2b
commit 0a85688ac3
No known key found for this signature in database
4 changed files with 44 additions and 2 deletions

View file

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

View file

@ -1024,8 +1024,12 @@ export class InMemoryJobStore implements IJobStoreV2 {
async getTerminalHostActionJobs(): Promise<SerializableJobData[]> {
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 &&

View file

@ -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;
}

View file

@ -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;