fix(webv2): stop retrying an enqueue that keeps failing with a 500

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] } };

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<string, Promise<unknown>>();
   const cancelRequestedRunKeys = new Set<string>();
   let reconcileRetryDelayMs = 100;
+  const serverErrorCounts = new Map<string, { count: number; firstAt: number }>();
+  const exhaustedServerErrors = new Map<string, string>();
   let retryTimer: ReturnType<typeof setTimeout> | undefined;
   let renewTimer: ReturnType<typeof setInterval> | 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();