diff --git a/api/server/controllers/agents/__tests__/resume.spec.js b/api/server/controllers/agents/__tests__/resume.spec.js index 6ccb281776..3cc728cd7d 100644 --- a/api/server/controllers/agents/__tests__/resume.spec.js +++ b/api/server/controllers/agents/__tests__/resume.spec.js @@ -1369,6 +1369,56 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => { expect(mockDecrementPendingRequest).toHaveBeenCalledWith(USER_ID); }); + it('finalizes a genuine failure whose message merely mentions abort', async () => { + const job = makeToolApprovalJob(); + mockGenerationJobManager.getJob.mockResolvedValue(job); + // The store still shows OUR generation running: proof no abort finalized the + // job (an actual abort flips it terminal BEFORE the signal fires). Taking the + // abort path here left the job `running` forever with no controller driving it. + mockJobStore.getJob.mockResolvedValue({ createdAt: 1000, status: 'running' }); + mockInitializeClient.mockResolvedValue({ + client: makeClient({ + resumeCompletion: jest.fn().mockRejectedValue(new Error('transaction aborted by server')), + }), + userMCPAuthMap: {}, + }); + + const res = await post(approveBody()); + expect(res.status).toBe(200); + await settled; + await flush(); + + expect(mockGenerationJobManager.emitError).toHaveBeenCalledWith( + CONVO_ID, + 'transaction aborted by server', + 1000, + ); + expect(mockGenerationJobManager.completeJob).toHaveBeenCalled(); + }); + + it('treats an abort-shaped throw as an abort when the job is already finalized', async () => { + const job = makeToolApprovalJob(); + mockGenerationJobManager.getJob.mockResolvedValue(job); + // abortJob deleted the job before the throw unwound: classification must not + // route this to the error path, whose outcome write would walk the failure + // streak for a user stop. + mockJobStore.getJob.mockResolvedValue(null); + mockInitializeClient.mockResolvedValue({ + client: makeClient({ + resumeCompletion: jest.fn().mockRejectedValue(new Error('request aborted')), + }), + userMCPAuthMap: {}, + }); + + const res = await post(approveBody()); + expect(res.status).toBe(200); + await settled; + await flush(); + + expect(mockGenerationJobManager.emitError).not.toHaveBeenCalled(); + expect(mockGenerationJobManager.completeJob).not.toHaveBeenCalled(); + }); + it('resume failure: emits an error, finalizes the job, and prunes the checkpoint', async () => { mockGenerationJobManager.getJob.mockResolvedValue(makeToolApprovalJob()); mockInitializeClient.mockResolvedValue({ diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index 7ddd084146..6493634acc 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -1037,7 +1037,31 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) // resumed generation usually surfaces its abort as a THROW out of // resumeCompletion, so this classification — not the pre-finalize signal check — // is the path most stops actually take. - if (job.abortController.signal.aborted || err?.message?.includes('abort')) { + // + // The signal is authoritative; an abort-SHAPED error alone is not. A genuine + // failure can mention 'abort' (driver errors like 'transaction aborted') with + // nothing having finalized the job, and returning here then left it `running` + // forever with no controller driving it. abortJob flips the job terminal BEFORE + // any signal fires, so a job still running under this generation is proof no + // abort finalized it — send that to the error path below instead. + const abortShaped = err?.name === 'AbortError' || err?.message?.includes('abort'); + let wasAborted = job.abortController.signal.aborted; + if (!wasAborted && abortShaped) { + try { + const liveJob = await GenerationJobManager.getJobStore().getJob(streamId); + wasAborted = !( + liveJob && + liveJob.createdAt === job.createdAt && + liveJob.status === 'running' + ); + } catch (readErr) { + logger.warn( + '[ResumeAgentController] Abort classification read failed; treating as failure', + readErr, + ); + } + } + if (wasAborted) { logger.debug(`[ResumeAgentController] Resume aborted; settling the run: ${streamId}`); await settleAbortedScheduledResume(job, streamId, conversationId); return; diff --git a/api/strategies/openIdJwtStrategy.js b/api/strategies/openIdJwtStrategy.js index 0bdf14b8b6..6ed0e24408 100644 --- a/api/strategies/openIdJwtStrategy.js +++ b/api/strategies/openIdJwtStrategy.js @@ -141,6 +141,21 @@ const openIdJwtLogin = (openIdConfig) => { if (user) { user.id = user._id.toString(); + // A user whose account deletion has begun must not authenticate (or be + // re-cached): the destructive cascade is running, and a served pre-barrier + // document would let this request recreate data behind it. The racing-fill + // case (Mongo read BEFORE the barrier stamped this field) is closed by the + // cache-side tombstone check in setCachedAuthUserDoc. + if (user.deletionRequestedAt != null) { + if (authUserCacheStore && authUserCacheKey) { + await invalidateCachedAuthUserDoc(authUserCacheStore, { + userId: user.id, + cacheKey: authUserCacheKey, + }); + } + done(null, false, { message: 'Account deletion in progress' }); + return; + } /** Absent on the full doc means local user; null skips getUserPrincipals' fallback lookup */ user.idOnTheSource ??= null; diff --git a/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx b/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx index 6c3895c8a2..05c702f2b8 100644 --- a/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx +++ b/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx @@ -210,8 +210,12 @@ export default function ScheduleDialog({ showToast({ message: localize('com_ui_schedule_updated'), status: 'success' }); onOpenChange(false); }, - onError: () => { - showToast({ message: localize('com_ui_error'), status: 'error' }); + onError: (error) => { + const status = (error as { response?: { status?: number } } | undefined)?.response?.status; + showToast({ + message: localize(status === 409 ? 'com_ui_schedule_conflict' : 'com_ui_error'), + status: 'error', + }); }, }); @@ -255,7 +259,15 @@ export default function ScheduleDialog({ } updateSchedule.mutate({ id: schedule.id, - payload, + // Fence on the revision this dialog opened with: cadence is rebuilt whole + // from that snapshot, so an edit from another tab would otherwise be + // silently overwritten — the server answers 409 instead. + payload: { + ...payload, + ...(schedule.configRevision != null + ? { expectedConfigRevision: schedule.configRevision } + : {}), + }, }); return; } diff --git a/client/src/locales/en/translation.json b/client/src/locales/en/translation.json index e67389a2a5..930070e0d6 100644 --- a/client/src/locales/en/translation.json +++ b/client/src/locales/en/translation.json @@ -1673,6 +1673,7 @@ "com_ui_saved": "Saved!", "com_ui_saving": "Saving...", "com_ui_schedule_am": "AM", + "com_ui_schedule_conflict": "This schedule was changed elsewhere. Reopen it to edit the latest version.", "com_ui_schedule_created": "Schedule created", "com_ui_schedule_daily": "Daily", "com_ui_schedule_day": "Day of week", diff --git a/packages/api/src/auth/userDocCache.spec.ts b/packages/api/src/auth/userDocCache.spec.ts index e052b77f33..e49432f735 100644 --- a/packages/api/src/auth/userDocCache.spec.ts +++ b/packages/api/src/auth/userDocCache.spec.ts @@ -5,6 +5,7 @@ import { AUTH_USER_DOC_CACHE_TTL_MS, buildAuthUserDocCacheKey, buildAuthUserDocReverseIndexKey, + buildAuthUserDocTombstoneKey, getAuthUserDocCacheMode, getCachedAuthUserDoc, invalidateCachedAuthUserDoc, @@ -159,6 +160,26 @@ describe('auth user document cache helpers', () => { ); }); + it('unwinds its own write when the deletion tombstone is present', async () => { + const store = makeStore(); + const userId = new Types.ObjectId(); + const cacheKey = 'auth-user-doc:v1:tombstoned'; + // The deletion barrier writes this BEFORE sweeping keys, so a fill whose Mongo + // read predates the barrier but whose cache write lands after the sweep must + // observe it here and delete its own entry — otherwise the deleted user's + // document is served for the full TTL while the destructive cascade runs. + store.values.set(buildAuthUserDocTombstoneKey(userId.toString()), Date.now()); + + await setCachedAuthUserDoc(store, cacheKey, { + _id: userId, + id: userId.toString(), + email: 'user@example.com', + }); + + expect(store.values.has(cacheKey)).toBe(false); + expect(store.values.has(buildAuthUserDocReverseIndexKey(userId.toString()))).toBe(false); + }); + it('deduplicates reverse-index keys and caps the remembered set', async () => { const store = makeStore(); const objectId = new Types.ObjectId(); diff --git a/packages/api/src/auth/userDocCache.ts b/packages/api/src/auth/userDocCache.ts index 95be55fd64..2354681ea7 100644 --- a/packages/api/src/auth/userDocCache.ts +++ b/packages/api/src/auth/userDocCache.ts @@ -1,6 +1,10 @@ import { createHash } from 'crypto'; import { logger } from '@librechat/data-schemas'; -import { AUTH_USER_DOC_BY_ID_PREFIX, CacheKeys } from 'librechat-data-provider'; +import { + AUTH_USER_DOC_TOMBSTONE_PREFIX, + AUTH_USER_DOC_BY_ID_PREFIX, + CacheKeys, +} from 'librechat-data-provider'; import type { IUser } from '@librechat/data-schemas'; import { cacheConfig } from '~/cache/cacheConfig'; @@ -104,6 +108,10 @@ export function buildAuthUserDocReverseIndexKey(userId: string): string { return `${AUTH_USER_DOC_BY_ID_PREFIX}:${userId}`; } +export function buildAuthUserDocTombstoneKey(userId: string): string { + return `${AUTH_USER_DOC_TOMBSTONE_PREFIX}:${userId}`; +} + function sanitizeUserForCache(user: Partial): CachedAuthUser { const id = getUserId(user); const { _id: _ignored, ...rest } = user; @@ -175,6 +183,16 @@ export async function setCachedAuthUserDoc( const userId = getUserId(sanitized); if (userId) { await rememberUserCacheKey(store, userId, cacheKey, AUTH_USER_DOC_CACHE_TTL_MS); + // Checked AFTER the writes above, never before: the deletion barrier writes + // its tombstone and then sweeps keys, so a fill whose Mongo read predates the + // barrier either lands before the sweep (swept) or observes the tombstone + // here and unwinds itself. A pre-write check leaves the sweep-then-write + // interleaving serving a deleted user's document for the full TTL. + const tombstoned = await store.get(buildAuthUserDocTombstoneKey(userId)); + if (tombstoned != null) { + await store.delete(cacheKey); + await store.delete(buildAuthUserDocReverseIndexKey(userId)); + } } } catch (error) { logger.warn('[authUserDocCache] Cache write failed', { diff --git a/packages/api/src/schedules/handlers.spec.ts b/packages/api/src/schedules/handlers.spec.ts index fa50f0dca8..0c32909a1c 100644 --- a/packages/api/src/schedules/handlers.spec.ts +++ b/packages/api/src/schedules/handlers.spec.ts @@ -60,6 +60,9 @@ describe('toWireSchedule', () => { [ 'agent_id', 'cadence', + // Public so the edit dialog can fence its PATCH on the revision it opened + // with (updateSchedulePayloadSchema.expectedConfigRevision). + 'configRevision', 'createdAt', 'disabledReason', 'enabled', @@ -671,6 +674,72 @@ describe('updateSchedule refuses field-less payloads', () => { }); }); +describe('updateSchedule client revision fence', () => { + const existingRow = () => + ({ + id: 'sched-1', + enabled: true, + agent_id: 'agent-1', + cadence: { frequency: 'daily', hour: 8, minute: 0 }, + timezone: 'UTC', + nextRunAt: new Date('2026-07-31T09:00:00Z'), + configRevision: 7, + }) as unknown as ISchedule; + + const makePatchReq = (body: Record) => + ({ + params: { id: 'sched-1' }, + body, + user: { id: 'user-1', tenantId: 't1', role: 'USER' }, + }) as unknown as ServerRequest; + + it('refuses a PATCH computed from a superseded revision', async () => { + const deps = makeCreateDeps({ isUserDeleting: jest.fn(async () => false) }); + (deps.methods.getScheduleById as jest.Mock).mockResolvedValue(existingRow()); + const { res, captured } = makeRes(); + + await createSchedulesHandlers(deps).updateSchedule( + makePatchReq({ name: 'renamed', expectedConfigRevision: 6 }), + res, + ); + + // The dialog rebuilds cadence whole from the snapshot it opened with, so the + // server's fresh-read fence alone cannot see that another tab edited the row. + expect(captured.status).toBe(409); + expect(deps.methods.updateScheduleById).not.toHaveBeenCalled(); + }); + + it('applies the PATCH and strips the fence field when the revision matches', async () => { + const deps = makeCreateDeps({ isUserDeleting: jest.fn(async () => false) }); + (deps.methods.getScheduleById as jest.Mock).mockResolvedValue(existingRow()); + const { res, captured } = makeRes(); + + await createSchedulesHandlers(deps).updateSchedule( + makePatchReq({ name: 'renamed', expectedConfigRevision: 7 }), + res, + ); + + expect(captured.status ?? 200).toBe(200); + const [, , update] = (deps.methods.updateScheduleById as jest.Mock).mock.calls[0]; + // The fence input is not a schedule field; writing it would corrupt the row. + expect(update).not.toHaveProperty('expectedConfigRevision'); + expect(update).toHaveProperty('name', 'renamed'); + }); + + it('still refuses a PATCH that carries only the fence field', async () => { + const deps = makeCreateDeps({ isUserDeleting: jest.fn(async () => false) }); + const { res, captured } = makeRes(); + + await createSchedulesHandlers(deps).updateSchedule( + makePatchReq({ expectedConfigRevision: 7 }), + res, + ); + + expect(captured.status).toBe(400); + expect(deps.methods.updateScheduleById).not.toHaveBeenCalled(); + }); +}); + describe('attachment id deduplication', () => { it('accepts a create whose payload repeats an owned file id', async () => { const deps = makeCreateDeps({ isUserDeleting: jest.fn(async () => false) }); diff --git a/packages/api/src/schedules/handlers.ts b/packages/api/src/schedules/handlers.ts index cf1f8dfb69..daec7c7304 100644 --- a/packages/api/src/schedules/handlers.ts +++ b/packages/api/src/schedules/handlers.ts @@ -180,6 +180,7 @@ export type WireSchedule = Pick< | 'lastRun' | 'runCount' | 'failureCount' + | 'configRevision' | 'createdAt' | 'updatedAt' >; @@ -201,6 +202,7 @@ export function toWireSchedule(schedule: ISchedule): WireSchedule { lastRun: schedule.lastRun, runCount: schedule.runCount, failureCount: schedule.failureCount, + configRevision: schedule.configRevision, createdAt: schedule.createdAt, updatedAt: schedule.updatedAt, }; @@ -580,12 +582,16 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH res.status(400).json({ error: 'Invalid schedule payload', issues: parsed.error.issues }); return; } + // The fence input is not a schedule field: strip it before the emptiness check + // and the update spread below, or it would count as an edit and be written to + // the row. + const { expectedConfigRevision, ...editedFields } = parsed.data; // A field-less PATCH is not a harmless no-op: updateScheduleById rotates the claim // token and bumps the config revision on every write, so an empty update would // fence a legitimate in-flight occurrence — its terminal bookkeeping revision- // fences to a no-op, and a fire in the POST-to-controller window is refused at // the admission boundary without ever running. Refuse before touching fencing. - if (Object.keys(parsed.data).length === 0) { + if (Object.keys(editedFields).length === 0) { res.status(400).json({ error: 'Schedule update must include at least one field' }); return; } @@ -599,6 +605,15 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH res.status(404).json({ error: 'Schedule not found' }); return; } + // Client-side revision fence: the dialog rebuilds compound fields (cadence) + // from the snapshot it opened with, so an edit from another tab is invisible + // to the fresh-read fence below — the payload is internally consistent with a + // row that no longer exists. Refuse before any side effect (file holds) when + // the client says which revision it edited. + if (expectedConfigRevision != null && existing.configRevision !== expectedConfigRevision) { + res.status(409).json({ error: 'Schedule was modified concurrently. Please retry.' }); + return; + } const limits = await deps.getLimits(user); // When the owner's config disables schedules, block edits that keep the // schedule enabled; still allow turning one OFF. @@ -642,7 +657,7 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH // or a failed arm leaves exactly this state; re-arm on ANY edit rather than only a // cadence one, or a name/prompt edit would silently leave it dead. const needsArming = existing.nextRunAt == null; - const update: Partial = { ...parsed.data } as Partial; + const update: Partial = { ...editedFields } as Partial; if (enabled && (cadenceChanged || needsArming)) { const nextRunAt = computeNextRunAt({ cadence, timezone, scheduleId: existing.id }); if (nextRunAt == null) { diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index ae7a366932..5ae4d8eef6 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -1740,10 +1740,35 @@ class GenerationJobManagerClass { runtime.finalEvent = abortFinalEvent; } + // Same bound as the abort publication above: everything between the won CAS + // and abortJob's return must terminate during a transport outage, or the Stop + // route behind this call never saves its partial or resolves its stop stamp. if (runtime?.createdEventPublication) { - await runtime.createdEventPublication; + try { + await withTimeout( + runtime.createdEventPublication, + ABORT_PUBLISH_TIMEOUT_MS, + `Created-event publication timed out during abort for ${streamId}`, + ); + } catch (err) { + logger.error( + `[GenerationJobManager] Created-event wait failed during abort for ${streamId}:`, + err, + ); + } + } + try { + await withTimeout( + Promise.resolve(this.eventTransport.emitDone(streamId, abortFinalEvent, jobData.createdAt)), + ABORT_PUBLISH_TIMEOUT_MS, + `Abort final-event publication timed out for ${streamId}`, + ); + } catch (err) { + logger.error( + `[GenerationJobManager] Failed to publish abort final event for ${streamId}:`, + err, + ); } - await this.eventTransport.emitDone(streamId, abortFinalEvent, jobData.createdAt); if (runtime?.startupTelemetry) { this.recordStartupEvent(runtime, abortFinalEvent); } diff --git a/packages/data-provider/src/config.ts b/packages/data-provider/src/config.ts index 429ebf2598..50dec5f763 100644 --- a/packages/data-provider/src/config.ts +++ b/packages/data-provider/src/config.ts @@ -2494,6 +2494,13 @@ export enum CacheKeys { export const AUTH_USER_DOC_BY_ID_PREFIX = 'auth-user-doc-byid'; +/** + * Deletion-barrier tombstone for the auth user-doc cache. Written BEFORE the + * barrier's key sweep so a cache fill racing the sweep (Mongo read pre-barrier, + * cache write post-sweep) observes it after writing and deletes its own entry. + */ +export const AUTH_USER_DOC_TOMBSTONE_PREFIX = 'auth-user-doc-tombstone'; + /** * Enum for violation types, used to identify, log, and cache violations. */ diff --git a/packages/data-provider/src/types/schedules.ts b/packages/data-provider/src/types/schedules.ts index 6bdc81bee7..e98c70e971 100644 --- a/packages/data-provider/src/types/schedules.ts +++ b/packages/data-provider/src/types/schedules.ts @@ -58,7 +58,17 @@ export type TCreateSchedule = z.infer; /** Idempotency is a property of the CREATE attempt, not of the schedule's config. */ export const updateSchedulePayloadSchema = createSchedulePayloadSchema .omit({ clientRequestId: true }) - .partial(); + .partial() + .extend({ + /** + * The configRevision the client's edit was computed from (captured when the + * dialog opened). The server fences the update on it, so a concurrent edit + * from another tab answers 409 instead of being silently overwritten by a + * payload rebuilt from a stale snapshot (cadence is sent whole, so the + * server-side fresh-read fence alone cannot detect this). + */ + expectedConfigRevision: z.number().int().min(0).optional(), + }); export type TUpdateSchedule = z.infer; export type TScheduleLastRun = { @@ -84,6 +94,7 @@ export type TSchedule = { lastRun?: TScheduleLastRun; runCount: number; failureCount: number; + configRevision?: number; createdAt: string; updatedAt: string; }; diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 969d8b3ac1..0f48c76bea 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -381,6 +381,31 @@ describe('recordRunOutcome', () => { expect(updated.lastRun?.conversationId).toBe('convo-paused'); }); + it('re-affirmed pauses keep the ORIGINAL fire time and do not churn updatedAt', async () => { + const schedule = await methods.createSchedule(scheduleData()); + const originalFiredAt = new Date('2026-07-20T12:00:02Z'); + await methods.insertScheduleRun(runData(schedule, { scheduledFor, firedAt: originalFiredAt })); + const pause = () => + methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'requires_action', + conversationId: 'convo-paused', + autoDisableAfterFailures: 3, + }); + + await pause(); + const afterFirst = await getSchedule(schedule.id); + // Reconciliation re-affirms a long-lived pause every pass; a fresh stamp per + // pass walked the card's timestamp forward and reordered updated-time listings + // for as long as the approval sat waiting. + await pause(); + const afterSecond = await getSchedule(schedule.id); + + expect(afterSecond.lastRun?.firedAt?.toISOString()).toBe(originalFiredAt.toISOString()); + expect(afterSecond.updatedAt?.toISOString()).toBe(afterFirst.updatedAt?.toISOString()); + }); + it('does not write a pause card when no active run matches (spoof guard)', async () => { const schedule = await methods.createSchedule(scheduleData()); // No run row for this occurrence: the card is written only after a matching active @@ -1057,6 +1082,40 @@ describe('deletion barrier fails closed on cache invalidation', () => { }); await expect(userMethods.markUserDeleting(user._id.toString())).resolves.toBeInstanceOf(Date); }); + + it('writes the deletion tombstone BEFORE sweeping the cached keys', async () => { + process.env.AUTH_USER_CACHE_MODE = 'on'; + const ops: string[] = []; + const tracking = createUserMethods(mongoose, { + getCache: () => + ({ + get: async () => undefined, + set: async (key: string) => { + ops.push(`set:${key}`); + return true; + }, + delete: async (key: string) => { + ops.push(`delete:${key}`); + return true; + }, + }) as never, + }); + const user = await mongoose.models.User.create({ + email: `barrier-tombstone-${Date.now()}@test.dev`, + name: 'B', + }); + + await tracking.markUserDeleting(user._id.toString()); + + // A fill racing this barrier checks the tombstone AFTER its own write. That + // only closes the race if the tombstone exists before the sweep begins: a + // sweep-first ordering leaves a window where the fill sees no tombstone and + // its entry was written after the sweep — surviving for the full TTL. + const tombstoneSet = ops.findIndex((op) => op.startsWith('set:auth-user-doc-tombstone:')); + const firstDelete = ops.findIndex((op) => op.startsWith('delete:')); + expect(tombstoneSet).toBeGreaterThanOrEqual(0); + expect(firstDelete).toBeGreaterThan(tombstoneSet); + }); }); describe('schedule deletion is retryable', () => { diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index ee874b9439..7650e91d1e 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -858,6 +858,10 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche ...revisionFilter, }, { $set: { lastRun: { ...lastRun, scheduledFor } } }, + // Run-state bookkeeping, not a config edit: re-affirmed pauses would + // otherwise bump updatedAt (and reorder updated-time listings) every + // reconciliation pass for as long as an approval sits waiting. + { timestamps: false }, ); } @@ -1114,9 +1118,16 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche // and the approval must not have the OLD config's pause stamped on it. Without // this the later terminal write (which IS fenced) could never replace the stale // status, so the card stuck on "Needs approval". + // The ROW's fire time, not the projection time: reconciliation re-affirms a + // long-lived pause every pass, and a fresh stamp per pass walked the card's + // timestamp forward for as long as the approval sat waiting. await projectLastRun( params.scheduleId, - { conversationId: params.conversationId, status: params.status, firedAt }, + { + conversationId: params.conversationId, + status: params.status, + firedAt: paused.firedAt ?? firedAt, + }, params.scheduledFor, paused.configRevision != null ? { configRevision: paused.configRevision } : {}, ); @@ -1161,7 +1172,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche await applyBalanceSkipBookkeeping({ scheduleId: params.scheduleId, scheduledFor: params.scheduledFor, - firedAt, + firedAt: settled.firedAt ?? firedAt, conversationId: params.conversationId, rowRevision: settled.configRevision, balanceSkipDisableThreshold: params.balanceSkipDisableThreshold, @@ -1169,7 +1180,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche } else { await applyTerminalBookkeeping({ ...params, - firedAt, + firedAt: settled.firedAt ?? firedAt, expectConfigRevision: settled.configRevision, }); } diff --git a/packages/data-schemas/src/methods/user.ts b/packages/data-schemas/src/methods/user.ts index a084e9f819..b2327fc0c7 100644 --- a/packages/data-schemas/src/methods/user.ts +++ b/packages/data-schemas/src/methods/user.ts @@ -1,5 +1,6 @@ import mongoose, { FilterQuery } from 'mongoose'; import { + AUTH_USER_DOC_TOMBSTONE_PREFIX, AUTH_USER_DOC_BY_ID_PREFIX, CacheKeys, type RefillIntervalUnit, @@ -290,6 +291,10 @@ export function createUserMethods( * destructive cascade runs. Swallowing the fault there reports a barrier that was * never actually raised. */ + /** Outlives any in-flight auth whose Mongo read predates the barrier; the cached + * doc's own 5s TTL bounds the damage of anything that slips past this window. */ + const AUTH_USER_DOC_TOMBSTONE_TTL_MS = 60_000; + async function invalidateAuthUserDocCache( userId: string, options?: { required?: boolean }, @@ -307,6 +312,17 @@ export function createUserMethods( return; } try { + if (options?.required && cache.set) { + // Tombstone FIRST, sweep second. A fill racing this barrier (Mongo read + // pre-barrier, cache write post-sweep) re-caches the deleted user for its + // full TTL; fills check this key AFTER writing, so either the sweep below + // catches their entry or this tombstone makes them unwind it themselves. + await cache.set( + `${AUTH_USER_DOC_TOMBSTONE_PREFIX}:${userId}`, + Date.now(), + AUTH_USER_DOC_TOMBSTONE_TTL_MS, + ); + } const indexKey = `${AUTH_USER_DOC_BY_ID_PREFIX}:${userId}`; const cachedKeys = await cache.get(indexKey); if (Array.isArray(cachedKeys)) { diff --git a/packages/data-schemas/src/types/cache.ts b/packages/data-schemas/src/types/cache.ts index 2181f83812..72717fdedd 100644 --- a/packages/data-schemas/src/types/cache.ts +++ b/packages/data-schemas/src/types/cache.ts @@ -5,7 +5,7 @@ */ export interface CacheStore { get: (key: string) => Promise; - set: (key: string, value: unknown) => Promise; + set: (key: string, value: unknown, ttlMs?: number) => Promise; delete?: (key: string) => Promise; clear?: () => Promise; /** True when the store is shared across processes (e.g. Redis-backed). */