From fad98f680ac09b931c507eccf5b90e2711411819 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 16 Aug 2026 19:45:28 -0400 Subject: [PATCH] fix: finish scheduled chat lifecycle hardening --- api/server/controllers/UserController.js | 14 + api/server/controllers/UserController.spec.js | 1 + .../controllers/__tests__/deleteUser.spec.js | 19 + .../__tests__/request.resumeMetadata.spec.js | 1 + api/server/experimental.js | 14 +- api/server/index.js | 13 +- .../services/UserFinalizationFallback.js | 39 ++ .../services/UserFinalizationFallback.spec.js | 71 ++ api/strategies/openIdJwtStrategy.spec.js | 1 + .../SidePanel/Schedules/ScheduleCard.tsx | 46 +- .../Schedules/ScheduleCardSkeleton.tsx | 24 + .../SidePanel/Schedules/ScheduleDialog.tsx | 52 +- .../SidePanel/Schedules/SchedulePanel.tsx | 86 +-- package-lock.json | 20 + packages/api/src/schedules/FOLLOWUPS.md | 152 +---- .../api/src/stream/GenerationJobManager.ts | 645 ++++++++++++------ .../__tests__/RedisEventTransport.spec.ts | 49 +- .../__tests__/abortSignalHonesty.spec.ts | 495 +++++++++++++- .../stream/__tests__/idempotencyClaim.spec.ts | 8 +- .../api/src/stream/__tests__/startup.spec.ts | 140 +++- .../implementations/RedisEventTransport.ts | 77 ++- .../api/src/stream/interfaces/IJobStore.ts | 13 +- packages/api/src/stream/startup.ts | 25 +- .../client/src/components/ControlCombobox.tsx | 9 + .../methods/user.finalizationFallback.spec.ts | 235 +++++++ packages/data-schemas/src/methods/user.ts | 130 ++++ packages/data-schemas/src/schema/user.ts | 7 + packages/data-schemas/src/types/user.ts | 3 + 28 files changed, 1885 insertions(+), 504 deletions(-) create mode 100644 api/server/services/UserFinalizationFallback.js create mode 100644 api/server/services/UserFinalizationFallback.spec.js create mode 100644 client/src/components/SidePanel/Schedules/ScheduleCardSkeleton.tsx create mode 100644 packages/data-schemas/src/methods/user.finalizationFallback.spec.ts diff --git a/api/server/controllers/UserController.js b/api/server/controllers/UserController.js index a55ebffe67..bea387da4b 100644 --- a/api/server/controllers/UserController.js +++ b/api/server/controllers/UserController.js @@ -594,6 +594,10 @@ const quiesceInteractiveGenerations = async (user) => { user.id, user.tenantId, ); + // A writer whose replica temporarily loses the primary lease store promotes its + // hold into Mongo before the primary TTL can expire. Read that fallback before the + // active scan for the same lease-to-job handoff ordering used above. + const earlyFallbackOwnerWork = await db.countUserFinalizationFallbackLeases(user.id); const activeJobIds = await GenerationJobManager.getActiveJobIdsForUser(user.id, user.tenantId); for (const streamId of activeJobIds ?? []) { // FENCE BEFORE THE CAS. An abort transitions the shared job to terminal, which @@ -659,6 +663,16 @@ const quiesceInteractiveGenerations = async (user) => { ); return false; } + const fallbackOwnerWork = + earlyFallbackOwnerWork > 0 + ? earlyFallbackOwnerWork + : await db.countUserFinalizationFallbackLeases(user.id); + if (fallbackOwnerWork > 0) { + logger.info( + `[quiesceInteractiveGenerations] ${fallbackOwnerWork} fallback finalization(s) still landing for ${user.id}`, + ); + return false; + } const fences = await db.getUserAbortFences(user.id); if (fences.length === 0) { return true; diff --git a/api/server/controllers/UserController.spec.js b/api/server/controllers/UserController.spec.js index 7b3dd7e047..730b262672 100644 --- a/api/server/controllers/UserController.spec.js +++ b/api/server/controllers/UserController.spec.js @@ -32,6 +32,7 @@ jest.mock('~/models', () => { addUserAbortFence: jest.fn().mockResolvedValue(undefined), clearUserAbortFence: jest.fn().mockResolvedValue(undefined), getUserAbortFences: jest.fn().mockResolvedValue([]), + countUserFinalizationFallbackLeases: jest.fn().mockResolvedValue(0), deleteAllUserSessions: jest.fn().mockResolvedValue(undefined), deleteAllSharedLinks: jest.fn().mockResolvedValue(undefined), deleteAllAgentApiKeys: jest.fn().mockResolvedValue(undefined), diff --git a/api/server/controllers/__tests__/deleteUser.spec.js b/api/server/controllers/__tests__/deleteUser.spec.js index 382f6467fe..4a59cf8b74 100644 --- a/api/server/controllers/__tests__/deleteUser.spec.js +++ b/api/server/controllers/__tests__/deleteUser.spec.js @@ -97,6 +97,7 @@ jest.mock('~/models', () => ({ addUserAbortFence: jest.fn().mockResolvedValue(undefined), clearUserAbortFence: jest.fn().mockResolvedValue(undefined), getUserAbortFences: jest.fn().mockResolvedValue([]), + countUserFinalizationFallbackLeases: jest.fn().mockResolvedValue(0), deleteTransactions: jest.fn(), deleteBalances: jest.fn(), deleteAllAgentApiKeys: jest.fn(), @@ -423,6 +424,20 @@ describe('deleteUserController - interactive generation quiesce', () => { expect(mockDeleteUserById).not.toHaveBeenCalled(); }); + it('defers while a Mongo fallback finalization lease is still live', async () => { + const req = { user: { id: 'user1', _id: 'user1', email: 'a@b.com' }, body: {} }; + const res = createRes(); + const db = require('~/models'); + mockGetUserById.mockResolvedValue({ _id: 'user1', twoFactorEnabled: false }); + db.countUserFinalizationFallbackLeases.mockResolvedValueOnce(1); + + await deleteUserController(req, res); + + expect(res.status).toHaveBeenCalledWith(503); + expect(mockDeleteMessages).not.toHaveBeenCalled(); + expect(mockDeleteUserById).not.toHaveBeenCalled(); + }); + it('fences an abort whose signal never left this replica and re-signals it', async () => { const req = { user: { id: 'user1', _id: 'user1', email: 'a@b.com' }, body: {} }; const res = createRes(); @@ -558,6 +573,10 @@ describe('deleteUserController - interactive generation quiesce', () => { expect(GenerationJobManager.countUserFinalizations.mock.invocationCallOrder[0]).toBeLessThan( GenerationJobManager.getActiveJobIdsForUser.mock.invocationCallOrder[0], ); + const db = require('~/models'); + expect(db.countUserFinalizationFallbackLeases.mock.invocationCallOrder[0]).toBeLessThan( + GenerationJobManager.getActiveJobIdsForUser.mock.invocationCallOrder[0], + ); }); it('defers on an admission lease observed before the job scan', async () => { diff --git a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js index 75fe1f915c..e48ecba0cf 100644 --- a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js +++ b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js @@ -2025,6 +2025,7 @@ describe('ResumableAgentController resume metadata', () => { error: 'Attached resources could not be restored', }), 1000, + { preserveForReconcile: false }, ); }); diff --git a/api/server/experimental.js b/api/server/experimental.js index e1fc36cf69..e4eab0cdfd 100644 --- a/api/server/experimental.js +++ b/api/server/experimental.js @@ -53,9 +53,14 @@ const { markEraseAttempted, getActiveRunsForSchedule, recordRunOutcome, + renewUserFinalizationFallbackLease, + clearUserFinalizationFallbackLease, + addUserAbortFence, + clearUserAbortFence, } = require('~/models'); const { checkMigrations } = require('./services/start/migration'); const { configureGenerationStreams } = require('@librechat/api'); +const { createUserFinalizationFallbackStore } = require('./services/UserFinalizationFallback'); const initializeMCPs = require('./services/initializeMCPs'); const configureSocialLogins = require('./socialLogins'); const createSpaFallback = require('./utils/fallback'); @@ -423,7 +428,14 @@ if (cluster.isMaster) { // store even with USE_REDIS_STREAMS — making the multiworker scheduler private, // so cross-worker aborts and orphan recovery could not work. Shared with // api/server/index.js so both topologies initialize identically. - configureGenerationStreams(); + configureGenerationStreams( + createUserFinalizationFallbackStore({ + renewUserFinalizationFallbackLease, + clearUserFinalizationFallbackLease, + addUserAbortFence, + clearUserAbortFence, + }), + ); // Same teardown discipline as api/server/index.js: stop active generations and // close their SSE streams while the HTTP server drains, then release stream // resources. Without this a killed worker leaves its Redis jobs `running` until diff --git a/api/server/index.js b/api/server/index.js index 75f2aefe7f..332a917f5b 100644 --- a/api/server/index.js +++ b/api/server/index.js @@ -50,6 +50,10 @@ const { sweepOrphanedPreviews, getRoleByName, seedDatabase, + renewUserFinalizationFallbackLease, + clearUserFinalizationFallbackLease, + addUserAbortFence, + clearUserAbortFence, } = require('~/models'); const initializeOAuthReconnectManager = require('./services/initializeOAuthReconnectManager'); const { capabilityContextMiddleware } = require('./middleware/roles/capabilities'); @@ -127,7 +131,14 @@ const rejectScheduleWritesUntilReady = (req, res, next) => { const configureGenerationStreams = () => { // Shared with the clustered entrypoint (experimental.js) so both topologies get the // same stream services; returns whether the resulting store is genuinely shared. - const isShared = configureSharedGenerationStreams(); + const isShared = configureSharedGenerationStreams( + createUserFinalizationFallbackStore({ + renewUserFinalizationFallbackLease, + clearUserFinalizationFallbackLease, + addUserAbortFence, + clearUserAbortFence, + }), + ); // Stop active generations and close their SSE streams while the HTTP server drains. registerShutdownTask( 'generation job manager prepare', diff --git a/api/server/services/UserFinalizationFallback.js b/api/server/services/UserFinalizationFallback.js new file mode 100644 index 0000000000..f2404ba5a0 --- /dev/null +++ b/api/server/services/UserFinalizationFallback.js @@ -0,0 +1,39 @@ +const { tenantStorage, runAsSystem } = require('@librechat/data-schemas'); + +/** + * Generation lifecycle callbacks run from timers and Redis handlers, where there + * is no request-scoped tenant context. Re-establish the exact user's tenant for + * tenant-isolated User writes; base users deliberately use the explicit system + * context rather than inheriting an unrelated request context. + */ +const runForUserTenant = (userId, tenantId, operation) => { + const run = async () => operation(); + return tenantId ? tenantStorage.run({ tenantId, userId }, run) : runAsSystem(run); +}; + +/** + * Adapts User model methods to the process-level generation lifecycle store. + * Keeping this adapter shared ensures the standard and clustered entrypoints + * apply identical strict-tenant behavior. + */ +const createUserFinalizationFallbackStore = ({ + renewUserFinalizationFallbackLease, + clearUserFinalizationFallbackLease, + addUserAbortFence, + clearUserAbortFence, +}) => ({ + renew: (userId, tenantId, safeLeaseKey, expiresAt) => + runForUserTenant(userId, tenantId, () => + renewUserFinalizationFallbackLease(userId, safeLeaseKey, expiresAt), + ), + clear: (userId, tenantId, safeLeaseKey) => + runForUserTenant(userId, tenantId, () => + clearUserFinalizationFallbackLease(userId, safeLeaseKey), + ), + retainAbortDelivery: (userId, tenantId, streamId) => + runForUserTenant(userId, tenantId, () => addUserAbortFence(userId, streamId)), + clearAbortDelivery: (userId, tenantId, streamId) => + runForUserTenant(userId, tenantId, () => clearUserAbortFence(userId, streamId)), +}); + +module.exports = { createUserFinalizationFallbackStore }; diff --git a/api/server/services/UserFinalizationFallback.spec.js b/api/server/services/UserFinalizationFallback.spec.js new file mode 100644 index 0000000000..08c9051532 --- /dev/null +++ b/api/server/services/UserFinalizationFallback.spec.js @@ -0,0 +1,71 @@ +const { + getTenantId, + getUserId, + tenantStorage, + SYSTEM_TENANT_ID, +} = require('@librechat/data-schemas'); +const { createUserFinalizationFallbackStore } = require('./UserFinalizationFallback'); + +describe('UserFinalizationFallback startup adapter', () => { + const makeStrictUserMethod = (contexts) => + jest.fn(async () => { + const tenantId = getTenantId(); + if (!tenantId) { + throw new Error('[TenantIsolation] Query attempted without tenant context in strict mode'); + } + contexts.push({ tenantId, userId: getUserId() }); + }); + + test('establishes tenant context for every timer callback with no request context', async () => { + const contexts = []; + const renewUserFinalizationFallbackLease = makeStrictUserMethod(contexts); + const clearUserFinalizationFallbackLease = makeStrictUserMethod(contexts); + const addUserAbortFence = makeStrictUserMethod(contexts); + const clearUserAbortFence = makeStrictUserMethod(contexts); + const store = createUserFinalizationFallbackStore({ + renewUserFinalizationFallbackLease, + clearUserFinalizationFallbackLease, + addUserAbortFence, + clearUserAbortFence, + }); + + expect(tenantStorage.getStore()).toBeUndefined(); + + await store.renew('user-1', 'tenant-a', 'fallback_1', new Date()); + await store.clear('user-1', 'tenant-a', 'fallback_1'); + await store.retainAbortDelivery('user-1', 'tenant-a', 'stream-1'); + await store.clearAbortDelivery('user-1', 'tenant-a', 'stream-1'); + + expect(contexts).toEqual([ + { tenantId: 'tenant-a', userId: 'user-1' }, + { tenantId: 'tenant-a', userId: 'user-1' }, + { tenantId: 'tenant-a', userId: 'user-1' }, + { tenantId: 'tenant-a', userId: 'user-1' }, + ]); + expect(renewUserFinalizationFallbackLease).toHaveBeenCalledWith( + 'user-1', + 'fallback_1', + expect.any(Date), + ); + expect(clearUserFinalizationFallbackLease).toHaveBeenCalledWith('user-1', 'fallback_1'); + expect(addUserAbortFence).toHaveBeenCalledWith('user-1', 'stream-1'); + expect(clearUserAbortFence).toHaveBeenCalledWith('user-1', 'stream-1'); + }); + + test('uses explicit system context for tenantless users instead of ambient request context', async () => { + const contexts = []; + const renewUserFinalizationFallbackLease = makeStrictUserMethod(contexts); + const store = createUserFinalizationFallbackStore({ + renewUserFinalizationFallbackLease, + clearUserFinalizationFallbackLease: jest.fn(), + addUserAbortFence: jest.fn(), + clearUserAbortFence: jest.fn(), + }); + + await tenantStorage.run({ tenantId: 'unrelated-tenant', userId: 'other-user' }, async () => + store.renew('base-user', undefined, 'fallback_base', new Date()), + ); + + expect(contexts).toEqual([{ tenantId: SYSTEM_TENANT_ID, userId: 'other-user' }]); + }); +}); diff --git a/api/strategies/openIdJwtStrategy.spec.js b/api/strategies/openIdJwtStrategy.spec.js index 378fe184b3..bef2e65045 100644 --- a/api/strategies/openIdJwtStrategy.spec.js +++ b/api/strategies/openIdJwtStrategy.spec.js @@ -532,6 +532,7 @@ describe('openIdJwtStrategy – auth user document cache', () => { mockAuthUserDocCacheStore, 'auth-user-doc-key', expect.objectContaining({ id: 'tenant-b-user', tenantId: 'tenant-b' }), + expect.objectContaining({ readAt: expect.any(Number) }), ); }); diff --git a/client/src/components/SidePanel/Schedules/ScheduleCard.tsx b/client/src/components/SidePanel/Schedules/ScheduleCard.tsx index f864c98e6c..03cad7310a 100644 --- a/client/src/components/SidePanel/Schedules/ScheduleCard.tsx +++ b/client/src/components/SidePanel/Schedules/ScheduleCard.tsx @@ -6,6 +6,7 @@ import { Play, Trash, Pencil, Ellipsis } from 'lucide-react'; import { PermissionTypes, Permissions } from 'librechat-data-provider'; import { Label, + Chip, Switch, Spinner, OGDialog, @@ -22,8 +23,8 @@ import { useRunScheduleNowMutation, } from '~/data-provider'; import { useLocalize, useHasAccess } from '~/hooks'; -import { getMessageTimestamp, cn } from '~/utils'; import { useAgentsMapContext } from '~/Providers'; +import { getMessageTimestamp } from '~/utils'; import ScheduleDialog from './ScheduleDialog'; import { describeCadence } from './cadence'; @@ -31,20 +32,16 @@ interface ScheduleCardProps { schedule: TSchedule; } -const CHIP_BASE = 'rounded-full px-2 py-0.5 text-xs font-medium'; -const GREEN_CHIP = 'bg-green-100 text-green-800 dark:bg-green-900 dark:text-green-300'; -const RED_CHIP = 'bg-red-100 text-red-800 dark:bg-red-900 dark:text-red-300'; -const AMBER_CHIP = 'bg-amber-100 text-amber-800 dark:bg-amber-900 dark:text-amber-300'; -const NEUTRAL_CHIP = 'bg-surface-tertiary text-text-secondary'; +type StatusTone = 'neutral' | 'success' | 'warning' | 'error'; -const STATUS_CHIPS: Record = { - success: { label: 'com_ui_schedule_last_run', className: GREEN_CHIP }, - error: { label: 'com_ui_schedule_last_run_failed', className: RED_CHIP }, - interrupted: { label: 'com_ui_schedule_last_run_failed', className: RED_CHIP }, - requires_action: { label: 'com_ui_schedule_needs_approval', className: AMBER_CHIP }, - started: { label: 'com_ui_schedule_run_started', className: NEUTRAL_CHIP }, - skipped_overlap: { label: 'com_ui_schedule_run_skipped', className: NEUTRAL_CHIP }, - skipped_balance: { label: 'com_ui_schedule_run_skipped', className: NEUTRAL_CHIP }, +const STATUS_CHIPS: Record = { + success: { label: 'com_ui_schedule_last_run', tone: 'success' }, + error: { label: 'com_ui_schedule_last_run_failed', tone: 'error' }, + interrupted: { label: 'com_ui_schedule_last_run_failed', tone: 'error' }, + requires_action: { label: 'com_ui_schedule_needs_approval', tone: 'warning' }, + started: { label: 'com_ui_schedule_run_started', tone: 'neutral' }, + skipped_overlap: { label: 'com_ui_schedule_run_skipped', tone: 'neutral' }, + skipped_balance: { label: 'com_ui_schedule_run_skipped', tone: 'neutral' }, }; const DISABLED_REASON_LABELS: Record = { @@ -139,6 +136,7 @@ export default function ScheduleCard({ schedule }: ScheduleCardProps) { label: localize('com_ui_schedule_run_now'), onClick: handleRunNow, hideOnClick: false, + disabled: runSchedule.isLoading, icon: runSchedule.isLoading ? ( ) : ( @@ -173,7 +171,7 @@ export default function ScheduleCard({ schedule }: ScheduleCardProps) { return (
@@ -185,7 +183,7 @@ export default function ScheduleCard({ schedule }: ScheduleCardProps) { checked={schedule.enabled} onCheckedChange={handleToggle} disabled={updateSchedule.isLoading} - aria-label={localize('com_ui_schedule_enabled')} + aria-label={`${localize('com_ui_schedule_enabled')}: ${schedule.name}`} className="shrink-0" /> @@ -221,20 +219,16 @@ export default function ScheduleCard({ schedule }: ScheduleCardProps) { (lastRunConvoId != null && lastRunConvoId !== '' ? ( ) : ( - - {localize(statusChip.label)} - + {localize(statusChip.label)} ))} {schedule.disabledReason != null && ( - - {localize(DISABLED_REASON_LABELS[schedule.disabledReason])} - + {localize(DISABLED_REASON_LABELS[schedule.disabledReason])} )}
)} @@ -267,7 +261,7 @@ export default function ScheduleCard({ schedule }: ScheduleCardProps) { selection={{ selectHandler: confirmDelete, selectClasses: - 'bg-red-700 dark:bg-red-600 hover:bg-red-800 dark:hover:bg-red-800 text-white', + 'bg-surface-destructive text-text-on-status hover:bg-surface-destructive-hover', selectText: localize('com_ui_delete'), }} /> diff --git a/client/src/components/SidePanel/Schedules/ScheduleCardSkeleton.tsx b/client/src/components/SidePanel/Schedules/ScheduleCardSkeleton.tsx new file mode 100644 index 0000000000..5f71c241a6 --- /dev/null +++ b/client/src/components/SidePanel/Schedules/ScheduleCardSkeleton.tsx @@ -0,0 +1,24 @@ +import { Skeleton } from '@librechat/client'; + +/** Mirrors the schedule card's title controls, agent, cadence, and next-run rows. */ +export default function ScheduleCardSkeleton({ count = 5 }: { count?: number }) { + return ( + + ); +} diff --git a/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx b/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx index 70d574b31b..adcf0de2e6 100644 --- a/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx +++ b/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx @@ -7,7 +7,9 @@ import { Input, Label, Button, + FieldMessage, Spinner, + Textarea, Dropdown, OGDialog, ControlCombobox, @@ -123,9 +125,10 @@ export default function ScheduleDialog({ register, watch, handleSubmit, - formState: { dirtyFields }, + formState: { dirtyFields, errors }, } = useForm({ defaultValues: getDefaultValues(schedule), + mode: 'onChange', }); /** The revision the form defaults were built from, captured at the same instant. * The `schedule` prop keeps refreshing while the dialog is open (the schedules @@ -133,9 +136,6 @@ export default function ScheduleDialog({ * refreshed revision onto a cadence rebuilt from this stale snapshot — exactly * the overwrite the fence exists to refuse. */ const openedConfigRevision = useRef(schedule?.configRevision); - const name = watch('name'); - const prompt = watch('prompt'); - const agentId = watch('agent_id'); const frequency = watch('frequency'); const hour12 = watch('hour12'); const minute = watch('minute'); @@ -316,7 +316,6 @@ export default function ScheduleDialog({ resolvePreservedWeeklyDays(frequency), ); const summary = `${describeCadence(summaryCadence, localize, locale)} · ${timezone}`; - const canSubmit = name.trim().length > 0 && prompt.trim().length > 0 && agentId.length > 0; return ( @@ -333,28 +332,34 @@ export default function ScheduleDialog({ +
-