diff --git a/invokeai/frontend/webv2/src/features/queue/runtime.test.ts b/invokeai/frontend/webv2/src/features/queue/runtime.test.ts index 6277aedc8b6..6dc5b9ed388 100644 --- a/invokeai/frontend/webv2/src/features/queue/runtime.test.ts +++ b/invokeai/frontend/webv2/src/features/queue/runtime.test.ts @@ -1664,6 +1664,189 @@ describe('queue runtime', () => { runtime.dispose(); }); + it('fails a run whose enqueue keeps returning a 500 instead of retrying forever', async () => { + vi.useFakeTimers(); + const queueItem = createPendingQueueItem(); + const project = { id: 'project-1', queue: { items: [queueItem] } }; + const commands = createTestCommands(); + const enqueueGenerate = vi.fn().mockRejectedValue(new ApiError('graph validation crashed', 500)); + const runtime = createQueueRuntime({ + ...runtimeServices, + backend: createTestBackend({ enqueueGenerate, listItems: vi.fn().mockResolvedValue([]) }), + history: { + commands, + getSnapshot: () => ({ connectionStatus: 'connected', isHydrated: true, projects: [project] }), + subscribe: vi.fn(() => vi.fn()), + }, + journal: createTestJournal(), + }); + + runtime.start(); + + await vi.advanceTimersByTimeAsync(20_000); + expect(commands.setStatus).toHaveBeenCalledWith( + expect.objectContaining({ projectId: project.id, queueItemId: queueItem.id, status: 'failed' }) + ); + const attempts = enqueueGenerate.mock.calls.length; + expect(attempts).toBeGreaterThanOrEqual(3); + await vi.advanceTimersByTimeAsync(30_000); + expect(enqueueGenerate).toHaveBeenCalledTimes(attempts); + await runtime.dispose(); + vi.useRealTimers(); + }); + + it('forgets earlier 500s once the run settles, so a later run of the item starts fresh', async () => { + vi.useFakeTimers(); + const queueItem = createPendingQueueItem(); + const project = { id: 'project-1', queue: { items: [queueItem] } }; + const listeners = new Set<() => void>(); + const notify = () => listeners.forEach((listener) => listener()); + const commands = createTestCommands({ + setStatus: vi.fn(({ status }) => { + queueItem.status = status; + notify(); + }), + }); + const enqueueGenerate = vi.fn().mockRejectedValue(new ApiError('graph validation crashed', 500)); + const runtime = createQueueRuntime({ + ...runtimeServices, + backend: createTestBackend({ enqueueGenerate, listItems: vi.fn().mockResolvedValue([]) }), + history: { + commands, + getSnapshot: () => ({ connectionStatus: 'connected', isHydrated: true, projects: [project] }), + subscribe: (listener) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + }, + journal: createTestJournal(), + }); + + runtime.start(); + await vi.advanceTimersByTimeAsync(1_000); + expect(enqueueGenerate.mock.calls.length).toBeGreaterThanOrEqual(2); + + // The user cancels the run before the 500s are given up on, which settles it. + queueItem.status = 'cancelled'; + notify(); + await vi.advanceTimersByTimeAsync(15_000); + + // Running the item again must not inherit the earlier run's 500 count and window. + enqueueGenerate.mockClear(); + queueItem.status = 'pending'; + notify(); + await vi.advanceTimersByTimeAsync(1_000); + expect(enqueueGenerate).toHaveBeenCalled(); + expect(commands.setStatus).not.toHaveBeenCalledWith(expect.objectContaining({ status: 'failed' })); + await runtime.dispose(); + vi.useRealTimers(); + }); + + it('recovers from 500s that stop within a second instead of failing the run', async () => { + vi.useFakeTimers(); + const queueItem = createPendingQueueItem(); + const project = { id: 'project-1', queue: { items: [queueItem] } }; + const commands = createTestCommands(); + const startedAt = Date.now(); + const enqueueGenerate = vi.fn(() => + Date.now() - startedAt < 1_000 + ? Promise.reject(new ApiError('transient internal error', 500)) + : Promise.resolve({ batchId: 'backend-batch', enqueued: 1, itemIds: [88], requested: 1 }) + ); + const runtime = createQueueRuntime({ + ...runtimeServices, + backend: createTestBackend({ enqueueGenerate, listItems: vi.fn().mockResolvedValue([]) }), + history: { + commands, + getSnapshot: () => ({ connectionStatus: 'connected', isHydrated: true, projects: [project] }), + subscribe: vi.fn(() => vi.fn()), + }, + journal: createTestJournal(), + }); + + runtime.start(); + + await vi.advanceTimersByTimeAsync(10_000); + expect(enqueueGenerate.mock.calls.length).toBeGreaterThan(3); + expect(commands.markBackendSubmitted).toHaveBeenCalledWith( + expect.objectContaining({ backendItemIds: [88], projectId: project.id, queueItemId: queueItem.id }) + ); + expect(commands.setStatus).not.toHaveBeenCalledWith(expect.objectContaining({ status: 'failed' })); + await runtime.dispose(); + vi.useRealTimers(); + }); + + it('adopts a run the server accepted before its last 500 instead of failing it', async () => { + vi.useFakeTimers(); + const queueItem = createPendingQueueItem(); + const project = { id: 'project-1', queue: { items: [queueItem] } }; + const commands = createTestCommands(); + const startedAt = Date.now(); + let committed = false; + // Once the 500s have lasted long enough to give up, the server commits the next enqueue but still answers 500. + const enqueueGenerate = vi.fn(() => { + committed = Date.now() - startedAt >= 10_000; + return Promise.reject(new ApiError('response failed after commit', 500)); + }); + const acceptedItem = { + batchId: 'backend-batch', + id: 88, + origin: buildQueueItemOrigin(queueItem.id, project.id), + status: 'pending', + }; + const listItems = vi.fn().mockImplementation(() => Promise.resolve(committed ? [acceptedItem] : [])); + const runtime = createQueueRuntime({ + ...runtimeServices, + backend: createTestBackend({ enqueueGenerate, listItems }), + history: { + commands, + getSnapshot: () => ({ connectionStatus: 'connected', isHydrated: true, projects: [project] }), + subscribe: vi.fn(() => vi.fn()), + }, + journal: createTestJournal(), + }); + + runtime.start(); + + await vi.advanceTimersByTimeAsync(20_000); + expect(committed).toBe(true); + expect(commands.markBackendSubmitted).toHaveBeenCalledWith( + expect.objectContaining({ backendItemIds: [88], projectId: project.id, queueItemId: queueItem.id }) + ); + expect(commands.setStatus).not.toHaveBeenCalledWith(expect.objectContaining({ status: 'failed' })); + await runtime.dispose(); + vi.useRealTimers(); + }); + + it('keeps retrying an enqueue that returns 503 while the server is unavailable', async () => { + const queueItem = createPendingQueueItem(); + const project = { id: 'project-1', queue: { items: [queueItem] } }; + const commands = createTestCommands(); + const enqueueGenerate = vi + .fn() + .mockRejectedValueOnce(new ApiError('unavailable', 503)) + .mockRejectedValueOnce(new ApiError('unavailable', 503)) + .mockRejectedValueOnce(new ApiError('unavailable', 503)) + .mockResolvedValue({ batchId: 'backend-batch', enqueued: 1, itemIds: [88], requested: 1 }); + const runtime = createQueueRuntime({ + ...runtimeServices, + backend: createTestBackend({ enqueueGenerate, listItems: vi.fn().mockResolvedValue([]) }), + history: { + commands, + getSnapshot: () => ({ connectionStatus: 'connected', isHydrated: true, projects: [project] }), + subscribe: vi.fn(() => vi.fn()), + }, + journal: createTestJournal(), + }); + + runtime.start(); + + await vi.waitFor(() => expect(commands.markBackendSubmitted).toHaveBeenCalled(), { timeout: 4_000 }); + expect(enqueueGenerate).toHaveBeenCalledTimes(4); + expect(commands.setStatus).not.toHaveBeenCalledWith(expect.objectContaining({ status: 'failed' })); + await runtime.dispose(); + }); + it('keeps reconciling after three consecutive lost enqueue responses', async () => { const queueItem = createPendingQueueItem(); const project = { id: 'project-1', queue: { items: [queueItem] } }; diff --git a/invokeai/frontend/webv2/src/features/queue/runtime.ts b/invokeai/frontend/webv2/src/features/queue/runtime.ts index 71e1ef4a634..3aa86522a10 100644 --- a/invokeai/frontend/webv2/src/features/queue/runtime.ts +++ b/invokeai/frontend/webv2/src/features/queue/runtime.ts @@ -412,6 +412,21 @@ const getJournalSignature = (item: QueueItem): string => item.cancelledBackendItemIds, ]); +/** + * A 5xx answer to an enqueue is usually caused by the request itself (e.g. a graph that + * crashes server-side validation), so retrying the same submission forever never succeeds. + * Gateway and availability errors (502/503/504) stay transient and keep retrying. + * A run only gives up after at least MAX_SERVER_ERROR_SUBMISSIONS 500s spread over at least + * MIN_SERVER_ERROR_WINDOW_MS, so a short-lived internal error can still recover. + * After the last allowed 500 the run is reconciled once more, and only fails if the + * backend queue really has nothing for it. + */ +const MAX_SERVER_ERROR_SUBMISSIONS = 3; +const MIN_SERVER_ERROR_WINDOW_MS = 10_000; +const TRANSIENT_SERVER_STATUSES = [502, 503, 504]; +const isDeterministicServerError = (error: unknown): boolean => + error instanceof ApiError && error.status >= 500 && !TRANSIENT_SERVER_STATUSES.includes(error.status); + export const createQueueRuntime = ({ backend, destinations, @@ -486,6 +501,8 @@ export const createQueueRuntime = ({ const submissionOperations = new Map>(); const cancelRequestedRunKeys = new Set(); let reconcileRetryDelayMs = 100; + const serverErrorCounts = new Map(); + const exhaustedServerErrors = new Map(); let retryTimer: ReturnType | undefined; let renewTimer: ReturnType | undefined; let hasInitialized = false; @@ -795,6 +812,8 @@ export const createQueueRuntime = ({ reconcileRetryRunKeys.delete(key); cancelRequestedRunKeys.delete(key); reportedCancelFailures.delete(key); + serverErrorCounts.delete(key); + exhaustedServerErrors.delete(key); invalidateAttempt(projectId, queueItemId); pendingJournalRecords.delete(key); const ownershipEpoch = ownershipEpochs.get(key) ?? 0; @@ -1805,6 +1824,7 @@ export const createQueueRuntime = ({ await request .then(async ({ batchId, enqueued, itemIds, requested }) => { + serverErrorCounts.delete(runKey); if (!isAttemptCurrent(attempt)) { return; } @@ -1874,6 +1894,21 @@ export const createQueueRuntime = ({ }); return; } + if (isDeterministicServerError(error)) { + const now = Date.now(); + const previous = serverErrorCounts.get(runKey); + const failures = (previous?.count ?? 0) + 1; + const firstAt = previous?.firstAt ?? now; + if (failures >= MAX_SERVER_ERROR_SUBMISSIONS && now - firstAt >= MIN_SERVER_ERROR_WINDOW_MS) { + // The server may have accepted the batch before answering 500, so reconcile before failing the run. + serverErrorCounts.delete(runKey); + exhaustedServerErrors.set(runKey, toErrorMessage(error)); + } else { + serverErrorCounts.set(runKey, { count: failures, firstAt }); + } + } else { + serverErrorCounts.delete(runKey); + } startedRunKeys.delete(runKey); reconcileRetryRunKeys.add(runKey); scheduleRetry(); @@ -1933,6 +1968,8 @@ export const createQueueRuntime = ({ pendingReconcileRunKeys.delete(key); reconcileRetryRunKeys.delete(key); reportedCancelFailures.delete(key); + serverErrorCounts.delete(key); + exhaustedServerErrors.delete(key); pendingJournalRecords.delete(key); releaseRunOwnershipAfterJournal(tracked.projectId, tracked.queueItemId); } @@ -2160,6 +2197,8 @@ export const createQueueRuntime = ({ const cancelRequested = cancelRequestedRunKeys.has(runKey); reconcileRetryRunKeys.delete(runKey); const outcome = outcomes.get(queueItem.id); + const exhaustedServerError = exhaustedServerErrors.get(runKey); + exhaustedServerErrors.delete(runKey); if ((outcome?.kind === 'adopted' || outcome?.kind === 'resumed') && outcome.missingBackendItemIds?.length) { commands.recordError({ @@ -2185,6 +2224,15 @@ export const createQueueRuntime = ({ switch (outcome?.kind) { case 'enqueue': + if (exhaustedServerError !== undefined) { + commands.setStatus({ + error: exhaustedServerError, + projectId: project.id, + queueItemId: currentQueueItem.id, + status: 'failed', + }); + break; + } startedRunKeys.delete(runKey); break; case 'adopted': @@ -2490,6 +2538,8 @@ export const createQueueRuntime = ({ fallbackAckOperations.clear(); reconcileRetryRunKeys.clear(); pendingReconcileRunKeys.clear(); + serverErrorCounts.clear(); + exhaustedServerErrors.clear(); trackedRuns.clear(); activeAttempts.clear(); ownershipEpochs.clear();