From d5d3762bc5415c338a8b1535bb36e801a9250183 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Wed, 22 Jul 2026 02:48:33 -0400 Subject: [PATCH] =?UTF-8?q?fix:=20Codex=20round=2016=20=E2=80=94=20preserv?= =?UTF-8?q?e=20live-expired=20scheduled=20approvals,=20DB-time=20claims?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. ApprovalLifecycle.expire: for a scheduled fire, omit completedAt so the aborted job is preserved (cross-store signal — Redis keeps the running TTL via transitionStatus's preserveTerminal, the in-memory finished-job sweep keys on completedAt), letting reconcile settle the requires_action run in ~2 min instead of the 25-h abandoned-pause cutoff. Covers the runtime expiry path that round-15's store-cleanup fix missed, for both stores. 2. claimDueSchedule: compare nextRunAt/leaseUntil against MongoDB $$NOW (via $expr and an aggregation-pipeline $set for the lease) instead of each worker's process clock, so replica clock skew can't claim future occurrences early or mis-time a lease. Drops the now-unused process-clock override param. --- packages/api/src/stream/ApprovalLifecycle.ts | 15 ++++++++-- packages/data-schemas/src/methods/schedule.ts | 29 +++++++++++++------ 2 files changed, 32 insertions(+), 12 deletions(-) diff --git a/packages/api/src/stream/ApprovalLifecycle.ts b/packages/api/src/stream/ApprovalLifecycle.ts index e117c7bee2..1867e7a35e 100644 --- a/packages/api/src/stream/ApprovalLifecycle.ts +++ b/packages/api/src/stream/ApprovalLifecycle.ts @@ -112,13 +112,22 @@ export class ApprovalLifecycle { * `expectedActionId` for the same stale-decision protection as `resolve`. */ async expire(streamId: string, expectedActionId?: string): Promise { + // A scheduled fire's expired approval must be RETAINED so the schedules + // reconciler can read `jobStatus === 'aborted'` and settle its `requires_action` + // run within ~2 min. Omitting completedAt is the cross-store preserve signal + // (Redis keeps it on the running TTL; the in-memory finished-job sweep keys on + // completedAt), so it isn't reaped before reconciliation — unlike a normal + // expired approval, which sets completedAt so terminal-cleanup can reclaim it. + const job = await this.store.getJob(streamId).catch(() => null); + const preserveForSchedule = job?.scheduleId != null; const ok = await this.store.transitionStatus(streamId, { from: 'requires_action', to: 'aborted', clear: ['pendingAction', 'pendingActionId'], - // completedAt lets the stores' terminal-cleanup reclaim the job; without - // it an expired approval lingers in the in-memory map indefinitely. - patch: { error: 'Approval expired before a decision was made', completedAt: Date.now() }, + patch: { + error: 'Approval expired before a decision was made', + ...(preserveForSchedule ? {} : { completedAt: Date.now() }), + }, expectActionId: expectedActionId, }); if (ok) { diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 347857eecc..aca4a1d54a 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -21,7 +21,6 @@ const COUNTED_FOR_WINDOW = 64; export interface ClaimDueScheduleParams { instanceId: string; leaseMs: number; - now?: Date; } export interface RecordRunOutcomeParams { @@ -172,20 +171,32 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche * due schedule regardless of replica count, with or without Redis. */ async function claimDueSchedule(params: ClaimDueScheduleParams): Promise { - const now = params.now ?? new Date(); + // Compare due-ness and lease expiry against MongoDB's own clock (`$$NOW`), not + // each worker's process clock: all replicas race on the persisted nextRunAt / + // leaseUntil, so a skewed worker must not claim future occurrences early or set + // a mis-timed lease. `nextRunAt` existence is gated by the plain filter (a bare + // $expr $lte would match a missing field as null); a missing leaseUntil is + // treated as epoch so it's always claimable. return Schedule() .findOneAndUpdate( { enabled: true, - nextRunAt: { $lte: now }, - $or: [{ leaseUntil: { $exists: false } }, { leaseUntil: { $lt: now } }], - }, - { - $set: { - leaseUntil: new Date(now.getTime() + params.leaseMs), - leaseBy: params.instanceId, + nextRunAt: { $exists: true, $ne: null }, + $expr: { + $and: [ + { $lte: ['$nextRunAt', '$$NOW'] }, + { $lt: [{ $ifNull: ['$leaseUntil', new Date(0)] }, '$$NOW'] }, + ], }, }, + [ + { + $set: { + leaseUntil: { $add: ['$$NOW', params.leaseMs] }, + leaseBy: params.instanceId, + }, + }, + ], { new: true, sort: { nextRunAt: 1 } }, ) .lean();