From d40dc005aa1ebf9f87c723449bb5fd4dfb3c9e88 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 16 Aug 2026 20:14:44 -0400 Subject: [PATCH] fix: close scheduled chat review follow-ups --- .../Schedules/__tests__/mutations.spec.tsx | 48 ++++++++++++ .../src/data-provider/Schedules/mutations.ts | 7 +- .../api/src/stream/GenerationJobManager.ts | 43 +++++++++-- .../__tests__/abortSignalHonesty.spec.ts | 76 +++++++++++++++++++ ...protocolRollout.stream_integration.spec.ts | 28 +++++-- 5 files changed, 187 insertions(+), 15 deletions(-) create mode 100644 client/src/data-provider/Schedules/__tests__/mutations.spec.tsx diff --git a/client/src/data-provider/Schedules/__tests__/mutations.spec.tsx b/client/src/data-provider/Schedules/__tests__/mutations.spec.tsx new file mode 100644 index 0000000000..d193acfa10 --- /dev/null +++ b/client/src/data-provider/Schedules/__tests__/mutations.spec.tsx @@ -0,0 +1,48 @@ +import { createElement } from 'react'; +import { QueryKeys } from 'librechat-data-provider'; +import { act, renderHook } from '@testing-library/react'; +import { QueryClient, QueryClientProvider } from '@tanstack/react-query'; +import type { ReactNode } from 'react'; +import { useDeleteScheduleMutation } from '../mutations'; + +const mockDeleteSchedule = jest.fn(); + +jest.mock('librechat-data-provider', () => { + const actual = jest.requireActual('librechat-data-provider'); + return { + ...actual, + dataService: { + ...actual.dataService, + deleteSchedule: (...args: unknown[]) => mockDeleteSchedule(...args), + }, + }; +}); + +const createWrapper = (queryClient: QueryClient) => + function Wrapper({ children }: { children: ReactNode }) { + return createElement(QueryClientProvider, { client: queryClient }, children); + }; + +describe('useDeleteScheduleMutation', () => { + it('invalidates the schedules list when deletion returns an error after mutating state', async () => { + const queryClient = new QueryClient({ + logger: { log: jest.fn(), warn: jest.fn(), error: jest.fn() }, + defaultOptions: { + queries: { retry: false }, + mutations: { retry: false }, + }, + }); + const invalidateQueries = jest.spyOn(queryClient, 'invalidateQueries'); + mockDeleteSchedule.mockRejectedValueOnce(new Error('Could not confirm active run stopped')); + const { result } = renderHook(() => useDeleteScheduleMutation(), { + wrapper: createWrapper(queryClient), + }); + + await act(async () => { + await result.current.mutateAsync('schedule-1').catch(() => undefined); + }); + + expect(invalidateQueries).toHaveBeenCalledWith([QueryKeys.schedules]); + queryClient.clear(); + }); +}); diff --git a/client/src/data-provider/Schedules/mutations.ts b/client/src/data-provider/Schedules/mutations.ts index ea1d461cb8..a16e48e2ef 100644 --- a/client/src/data-provider/Schedules/mutations.ts +++ b/client/src/data-provider/Schedules/mutations.ts @@ -53,9 +53,12 @@ export const useDeleteScheduleMutation = ( (id: string) => dataService.deleteSchedule(id), { ...options, - onSuccess: (...args) => { + // An `unconfirmed` 503 is still a server-side mutation: the schedule has + // already been disabled, marked deleting, and hidden from list reads. + // Refresh on either outcome so the card cannot remain actionable in cache. + onSettled: (...args) => { queryClient.invalidateQueries([QueryKeys.schedules]); - options?.onSuccess?.(...args); + options?.onSettled?.(...args); }, }, ); diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index fd6e4da39e..dd48c2cc27 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -20,6 +20,7 @@ import type { StandardGraph } from '@librechat/agents'; import type { SerializableJobData, CreatedJobData, + ReplacedGeneration, IEventTransport, UsageMetadata, AbortResult, @@ -715,6 +716,14 @@ interface PreparedSubscription { deferDeliveryUntilActivated: boolean; } +interface CommittedCreateRecovery { + job: CreatedJobData; + /** Transaction receipts already delivered during ambiguous-create recovery. + * They may be absent from the repaired job after their fenced ACK, but the + * caller still needs them to classify its stale pre-create snapshot. */ + handledPredecessors: readonly ReplacedGeneration[]; +} + type DeferredDelivery = | { type: 'chunk'; event: t.ServerSentEvent } | { type: 'done'; event: t.ServerSentEvent } @@ -1910,7 +1919,7 @@ class GenerationJobManagerClass { tenantId: string | undefined, options: CreateGenerationJobOptions, creationAttemptId: string, - ): Promise { + ): Promise { const clientRequestId = options.idempotencyClientRequestId; const claimToken = options.idempotencyClaimToken; const expectedConversationId = conversationId ?? streamId; @@ -1924,6 +1933,7 @@ class GenerationJobManagerClass { (job.idempotencyClientRequestId ?? undefined) === clientRequestId; let committedJob = (await this.jobStore.getJob(streamId)) as CreatedJobData | null; + let handledPredecessors: readonly ReplacedGeneration[] = []; if ( committedJob != null && committedJob.userId === userId && @@ -1933,6 +1943,9 @@ class GenerationJobManagerClass { // mutable because later recovery may refresh it, so TypeScript cannot // safely retain a narrowing on that outer binding across the await. const currentCommittedJob = committedJob; + handledPredecessors = + currentCommittedJob.replacedJobs ?? + (currentCommittedJob.replacedJob != null ? [currentCommittedJob.replacedJob] : []); const ownsCommittedCreate = matchesCommittedCreate(currentCommittedJob); const predecessorsDelivered = await this.processDurableReplacementReceipts( streamId, @@ -2048,7 +2061,7 @@ class GenerationJobManagerClass { if (repaired) { const confirmed = await this.jobStore.getJob(streamId); if (matchesCommittedCreate(confirmed)) { - return confirmed; + return { job: confirmed, handledPredecessors }; } return null; } @@ -2211,6 +2224,7 @@ class GenerationJobManagerClass { // so replacement must explicitly close/handoff that old attachment. const observedPredecessorJob = await this.jobStore.getJob(streamId); let jobData: CreatedJobData; + let recoveredPredecessors: readonly ReplacedGeneration[] = []; try { jobData = await this.jobStore.createJob( streamId, @@ -2263,7 +2277,7 @@ class GenerationJobManagerClass { throw error; } - let recovered: CreatedJobData | null = null; + let recovered: CommittedCreateRecovery | null = null; try { recovered = await this.recoverCommittedCreate( streamId, @@ -2282,7 +2296,8 @@ class GenerationJobManagerClass { if (recovered == null) { throw error; } - jobData = recovered; + jobData = recovered.job; + recoveredPredecessors = recovered.handledPredecessors; } if (options.idempotencyClientRequestId != null && options.idempotencyClaimToken != null) { @@ -2397,14 +2412,23 @@ class GenerationJobManagerClass { } const replacementEpochs = new Map(); - const exactPredecessors = + const durableExactPredecessors = jobData.replacedJobs ?? (jobData.replacedJob != null ? [jobData.replacedJob] : []); const exactPredecessorsByEpoch = new Map( - exactPredecessors.map((predecessor) => [predecessor.createdAt, predecessor]), + [...recoveredPredecessors, ...durableExactPredecessors].map((predecessor) => [ + predecessor.createdAt, + predecessor, + ]), ); + const exactPredecessors = [...exactPredecessorsByEpoch.values()]; const observedPredecessorReceipt = observedPredecessorJob ? exactPredecessorsByEpoch.get(observedPredecessorJob.createdAt) : undefined; + const observedPredecessorCoveredByLaterReceipt = + observedPredecessorJob != null && + exactPredecessors.some( + (predecessor) => predecessor.createdAt > observedPredecessorJob.createdAt, + ); if ( observedPredecessorJob && observedPredecessorJob.createdAt !== jobData.createdAt && @@ -2413,7 +2437,12 @@ class GenerationJobManagerClass { // The transaction-time predecessor is authoritative for the same // epoch. A pre-read may say running even though it terminalized before // replacement; do not publish a contradictory replaced/abort pair. - observedPredecessorReceipt == null + observedPredecessorReceipt == null && + // A newer transaction-time receipt also proves this pre-read was already + // superseded before our create committed. If its earlier handoff were + // still unconfirmed, Redis would have inherited that epoch in the same + // retained receipt chain; its absence is the predecessor's fenced ACK. + !observedPredecessorCoveredByLaterReceipt ) { replacementEpochs.set( observedPredecessorJob.createdAt, diff --git a/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts b/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts index a396a29631..9516f8aac7 100644 --- a/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts +++ b/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts @@ -284,6 +284,82 @@ describe('abort signal honesty across replicas', () => { await owner.destroy(); }); + it('does not let a predecessor owner release clear the current generation Stop fence', async () => { + const { GenerationJobManagerClass } = await import('../GenerationJobManager'); + const { InMemoryJobStore } = await import('../implementations/InMemoryJobStore'); + const { InMemoryEventTransport } = await import('../implementations/InMemoryEventTransport'); + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 0 }); + const retainAbortDelivery = jest.fn(async () => undefined); + const clearAbortDelivery = jest.fn(async () => undefined); + const fallback = { + renew: jest.fn(async () => undefined), + clear: jest.fn(async () => undefined), + retainAbortDelivery, + clearAbortDelivery, + }; + const owner = new GenerationJobManagerClass(); + owner.configure({ + jobStore, + eventTransport: new InMemoryEventTransport(), + isRedis: true, + userFinalizationFallback: fallback, + }); + owner.initialize(); + const peerTransport = Object.assign(new InMemoryEventTransport(), { + emitAbortConfirmed: jest.fn(async () => false), + }); + const peer = new GenerationJobManagerClass(); + peer.configure({ + jobStore, + eventTransport: peerTransport, + isRedis: true, + userFinalizationFallback: fallback, + }); + peer.initialize(); + + try { + const predecessor = await owner.createJob( + 'conv-generation-fence', + 'user-1', + 'conv-generation-fence', + ); + const predecessorAbort = await owner.abortJob('conv-generation-fence'); + expect(predecessorAbort.signalDelivered).toBe(true); + + const current = await owner.createJob( + 'conv-generation-fence', + 'user-1', + 'conv-generation-fence', + ); + expect(current.createdAt).toBeGreaterThan(predecessor.createdAt); + const currentAbort = await peer.abortJob('conv-generation-fence'); + expect(currentAbort).toMatchObject({ + success: true, + signalDelivered: false, + signalPublished: false, + }); + expect(retainAbortDelivery).toHaveBeenLastCalledWith( + 'user-1', + undefined, + 'conv-generation-fence', + ); + clearAbortDelivery.mockClear(); + + // Generation A still has its local owner hold while generation B's failed + // Stop owns the stream-level durable recovery fence. A's late catch must not + // clear B's evidence merely because A has a hold for the same stream. + await owner.releaseOwnerLease( + 'user-1', + 'conv-generation-fence', + undefined, + predecessor.createdAt, + ); + expect(clearAbortDelivery).not.toHaveBeenCalled(); + } finally { + await Promise.allSettled([peer.destroy(), owner.destroy()]); + } + }); + it('heartbeats the abort contender while required persistence exceeds the lease TTL', async () => { const { owner, jobStore } = await makeManagers(); await owner.createJob('conv-slow-stop-persistence', 'user-1', 'conv-slow-stop-persistence'); diff --git a/packages/api/src/stream/__tests__/protocolRollout.stream_integration.spec.ts b/packages/api/src/stream/__tests__/protocolRollout.stream_integration.spec.ts index f75a478728..17bfd4bd40 100644 --- a/packages/api/src/stream/__tests__/protocolRollout.stream_integration.spec.ts +++ b/packages/api/src/stream/__tests__/protocolRollout.stream_integration.spec.ts @@ -278,15 +278,22 @@ describe('Redis generation protocol rollout bridge', () => { }); test('recovers a lost create reply, repairs membership, and stops the exact raced predecessor', async () => { - const abortHandlers = new Map void | boolean>>(); + const abortHandlers = new Map< + string, + Set<(generationId?: number) => void | boolean | Promise> + >(); const acknowledgedAborts = new Map>(); + const confirmedAbortRequests: number[] = []; const eventTransport = Object.assign(new InMemoryEventTransport(), { emitAbort(streamId: string, generationId?: number) { for (const handler of abortHandlers.get(streamId) ?? []) { handler(generationId); } }, - async onAbort(streamId: string, handler: (generationId?: number) => void) { + async onAbort( + streamId: string, + handler: (generationId?: number) => void | boolean | Promise, + ) { let handlers = abortHandlers.get(streamId); if (handlers == null) { handlers = new Set(); @@ -296,9 +303,10 @@ describe('Redis generation protocol rollout bridge', () => { return () => handlers?.delete(handler); }, async emitAbortConfirmed(streamId: string, generationId: number) { + confirmedAbortRequests.push(generationId); let acknowledged = acknowledgedAborts.get(streamId)?.has(generationId) === true; for (const handler of abortHandlers.get(streamId) ?? []) { - acknowledged = handler(generationId) === true || acknowledged; + acknowledged = (await handler(generationId)) === true || acknowledged; } if (acknowledged) { let generations = acknowledgedAborts.get(streamId); @@ -385,6 +393,8 @@ describe('Redis generation protocol rollout bridge', () => { expect(racedPredecessor!.createdAt).toBeGreaterThan(stalePredecessor.createdAt); expect(replacement.createdAt).toBeGreaterThan(racedPredecessor!.createdAt); expect(racedPredecessor!.abortController.signal.aborted).toBe(true); + expect(confirmedAbortRequests).toContain(racedPredecessor!.createdAt); + expect(confirmedAbortRequests).not.toContain(stalePredecessor.createdAt); expect(doneEvents).toContainEqual({ event: expect.objectContaining({ final: true, @@ -409,7 +419,10 @@ describe('Redis generation protocol rollout bridge', () => { }); test('a superseded lost-reply helper delivers but does not acknowledge the crashed successor chain', async () => { - const abortHandlers = new Map void | boolean>>(); + const abortHandlers = new Map< + string, + Set<(generationId?: number) => void | boolean | Promise> + >(); const acknowledgedAborts = new Map>(); const eventTransport = Object.assign(new InMemoryEventTransport(), { emitAbort(streamId: string, generationId?: number) { @@ -417,7 +430,10 @@ describe('Redis generation protocol rollout bridge', () => { handler(generationId); } }, - async onAbort(streamId: string, handler: (generationId?: number) => void) { + async onAbort( + streamId: string, + handler: (generationId?: number) => void | boolean | Promise, + ) { let handlers = abortHandlers.get(streamId); if (handlers == null) { handlers = new Set(); @@ -429,7 +445,7 @@ describe('Redis generation protocol rollout bridge', () => { async emitAbortConfirmed(streamId: string, generationId: number) { let acknowledged = acknowledgedAborts.get(streamId)?.has(generationId) === true; for (const handler of abortHandlers.get(streamId) ?? []) { - acknowledged = handler(generationId) === true || acknowledged; + acknowledged = (await handler(generationId)) === true || acknowledged; } if (acknowledged) { let generations = acknowledgedAborts.get(streamId);