From f127d4f5757de424b2a02d1d47bc716120c91e71 Mon Sep 17 00:00:00 2001 From: Reece Browne <74901996+reecebrowne@users.noreply.github.com> Date: Tue, 16 Jun 2026 18:32:12 +0100 Subject: [PATCH] fix(policies): poll runs to completion with progress, soft-retry when queue is full (#6690) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What & why Production reports of policy enforcement "hanging" traced to large/many-page documents: the watermark step's flatten-to-image (`convertPDFToImage`) on a 500+ page PDF takes minutes, exceeding both the client poll cap and the backend per-step timeout. This makes the slow case graceful instead of looking broken, and makes load-shedding non-fatal. ### Poll runs to completion (no false "hang") The client poll loop used a flat ~150s cap that was **shorter than the backend's 300s per-step timeout**, so it abandoned long-but-healthy runs mid-flight. The budget is now sized to the backend's real worst case — `stepCount × per-step timeout + grace`, learned from the first status report — so the client always polls long enough to surface the run's **actual** terminal state (success or the backend's real error) rather than a misleading client-side timeout. ### Per-step progress The activity feed now shows `Enforcing… · step n/m` (from `currentStep`/`stepCount`), so a slow step shows movement instead of a dead spinner. ### Soft-retry on queue rejection Under load the shared `JobQueue` rejects runs ("queue full"), which previously surfaced as a hard failure needing a manual Retry. The backend now tags that rejection with a stable `POLICY_QUEUE_FULL` errorCode; the client treats it as transient backpressure and **auto-retries the file in place** with exponential backoff (≈4s→64s, ~2 min), shown as a soft "Busy — retrying…" row, falling back to the manual Retry only once the retry budget is spent. ## Testing - **Frontend unit tests** (30 pass across the policies suite), including a new `usePolicyAutoRun.retry.test.tsx` that drives the real controller orchestration (poll → `POLICY_QUEUE_FULL` → relabel → backoff → in-place re-dispatch), plus poll-budget, step-progress, and activity-feed relabel cases. - **Backend** `PolicyEngineTest` case asserting a queue-rejected run carries the `POLICY_QUEUE_FULL` code. - Typecheck clean on all three flavors (proprietary/saas/core); prettier + spotless clean. - Poll-budget + progress + real-error surfacing were also verified live end-to-end against a 599-page run (survived past the old cap, showed step progress, reported the backend's real 300s-timeout failure, recovered after a simulated network drop). ## Not included (follow-ups) - The underlying flatten-to-image cost itself (bounded-memory/streaming flatten, revisiting `convertPDFToImage` default and the 300s timeout) — the real perf fix, deliberately out of scope here. --- .../policy/engine/PolicyEngine.java | 7 +- .../policy/engine/PolicyEngineTest.java | 21 +++ .../policies/policyRunStore.test.ts | 12 ++ .../components/policies/policyRunStore.ts | 21 +++ .../policies/usePolicyAutoRun.retry.test.tsx | 106 ++++++++++++ .../policies/usePolicyAutoRun.test.ts | 160 ++++++++++++++++++ .../components/policies/usePolicyAutoRun.ts | 156 +++++++++++++++-- .../services/policyLiveData.test.ts | 36 ++++ .../proprietary/services/policyLiveData.ts | 13 +- 9 files changed, 516 insertions(+), 16 deletions(-) create mode 100644 frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.retry.test.tsx create mode 100644 frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.test.ts diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java index 7db5e7a14..196a680f0 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java @@ -59,6 +59,10 @@ public class PolicyEngine { // files. See ResourceMonitor#shouldQueueJob(int). private static final int RUN_RESOURCE_WEIGHT = 50; + // errorCode marking a run that was never admitted (job queue full under load). Transient: the + // client treats it as "busy" and retries, rather than as a terminal processing failure. + private static final String QUEUE_FULL_CODE = "POLICY_QUEUE_FULL"; + private final PolicyExecutor stepExecutor; private final TaskManager taskManager; private final PolicyRunRegistry registry; @@ -239,7 +243,8 @@ public class PolicyEngine { if (!completion.isDone()) { String message = "Policy run could not be queued: " + ex.getMessage(); log.error("Policy run {} was not admitted: {}", run.getRunId(), ex.getMessage()); - run.fail(message); + // Transient admission rejection, not a processing failure (see QUEUE_FULL_CODE). + run.failWithCode(message, QUEUE_FULL_CODE, null); taskManager.setError(run.getRunId(), message); completion.complete(run); } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java index ddb4eced3..78348fa5c 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java @@ -336,6 +336,27 @@ class PolicyEngineTest { assertEquals(PolicyRunStatus.PENDING, registry.get(handle.runId()).getStatus()); } + @Test + void runRejectedWhenQueueFullCarriesTransientErrorCode() { + when(resourceMonitor.shouldQueueJob(anyInt())).thenReturn(true); + // Admission rejected (queue full): the queued future completes exceptionally. + CompletableFuture rejected = new CompletableFuture<>(); + rejected.completeExceptionally( + new RuntimeException("Job queue full, please try again later")); + doReturn(rejected).when(jobQueue).queueJob(anyString(), anyInt(), any(), anyLong()); + + PolicyRunHandle handle = + engine.submit( + definition(new PipelineStep(ROTATE, Map.of())), + PolicyInputs.of(List.of(pdf("input", "input.pdf"))), + PolicyProgressListener.NOOP); + + PolicyRun run = registry.get(handle.runId()); + assertEquals(PolicyRunStatus.FAILED, run.getStatus()); + // Tagged transient so the client backs off and retries instead of hard-failing. + assertEquals("POLICY_QUEUE_FULL", run.getErrorCode()); + } + @Test void resumeIsNotYetImplemented() { assertThrows(UnsupportedOperationException.class, () -> engine.resume("any", List.of())); diff --git a/frontend/editor/src/proprietary/components/policies/policyRunStore.test.ts b/frontend/editor/src/proprietary/components/policies/policyRunStore.test.ts index 8a62cbfcc..922e64650 100644 --- a/frontend/editor/src/proprietary/components/policies/policyRunStore.test.ts +++ b/frontend/editor/src/proprietary/components/policies/policyRunStore.test.ts @@ -1,9 +1,11 @@ import { describe, it, expect, beforeEach } from "vitest"; import { dispatchKey, + getRun, isDispatched, markDispatched, recordRunStart, + removeRun, updateRun, resetPolicyRuns, type PolicyRunRecord, @@ -70,6 +72,16 @@ describe("policyRunStore", () => { expect(read("stirling-policy-runs").runs[0].status).toBe("PENDING"); }); + it("getRun returns the record by id, removeRun drops it but keeps the dispatched key", () => { + recordRunStart(rec({ runId: "abc" })); + expect(getRun("abc")?.fileId).toBe("f1"); + removeRun("abc"); + expect(getRun("abc")).toBeUndefined(); + expect(read("stirling-policy-runs").runs).toHaveLength(0); + // The (policy, file) pair stays dispatched so the auto-run doesn't re-fire on its own. + expect(isDispatched("security", "f1")).toBe(true); + }); + it("caps stored runs at 50, newest first", () => { for (let i = 0; i < 55; i++) { recordRunStart(rec({ runId: `r${i}`, fileId: `f${i}`, startedAt: i })); diff --git a/frontend/editor/src/proprietary/components/policies/policyRunStore.ts b/frontend/editor/src/proprietary/components/policies/policyRunStore.ts index 18206bc44..455efa5f2 100644 --- a/frontend/editor/src/proprietary/components/policies/policyRunStore.ts +++ b/frontend/editor/src/proprietary/components/policies/policyRunStore.ts @@ -19,6 +19,11 @@ export interface PolicyRunRecord { fileName: string; fileSize: number; status: PolicyRunStatus; + /** Pipeline progress reported by the run-status endpoint: the 1-based step + * currently running, and the total step count. Drive the "step X/Y" label + * while a run is in flight. Absent until the first status report. */ + currentStep?: number; + stepCount?: number; /** Output files (downloadable via /api/v1/general/files/{id}) once done. */ outputs: { fileId: string; fileName: string }[]; /** True once ALL outputs have been imported into the workspace. */ @@ -33,6 +38,9 @@ export interface PolicyRunRecord { error: string | null; /** Stable backend failure code (e.g. an entitlement sentinel) when FAILED; null otherwise. */ errorCode?: string | null; + /** Set while an auto-retry is pending after a transient (queue-full) rejection, so the activity + * feed shows a soft "busy" row instead of a hard failure during the backoff window. */ + retrying?: boolean; /** Epoch ms when the run was dispatched. */ startedAt: number; } @@ -146,6 +154,19 @@ export function updateRun(runId: string, patch: Partial) { emit(); } +/** The current record for a run id, if any. */ +export function getRun(runId: string): PolicyRunRecord | undefined { + return state.runs.find((r) => r.runId === runId); +} + +/** Drop a run record (leaving its dispatched key intact). Used when retrying a + * queue-rejected run in place, so the replacement run doesn't stack a second row. */ +export function removeRun(runId: string) { + if (!state.runs.some((r) => r.runId === runId)) return; + state = { ...state, runs: state.runs.filter((r) => r.runId !== runId) }; + emit(); +} + /** Reset the store — used by tests to isolate it. */ export function resetPolicyRuns() { state = { runs: [], dispatched: [] }; diff --git a/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.retry.test.tsx b/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.retry.test.tsx new file mode 100644 index 000000000..16b321c8c --- /dev/null +++ b/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.retry.test.tsx @@ -0,0 +1,106 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { renderHook, act } from "@testing-library/react"; + +// The auto-run hook reaches into several contexts + the network; stub those so we can drive just +// the queue-rejection retry path against the REAL run store. +vi.mock("@app/constants/featureFlags", () => ({ POLICIES_ENABLED: true })); +vi.mock("@app/contexts/FileContext", () => ({ + useAllFiles: () => ({ fileStubs: [] }), + useFileManagement: () => ({ addFiles: vi.fn() }), + useFileContext: () => ({ consumeFiles: vi.fn() }), +})); +vi.mock("@app/hooks/usePolicies", () => ({ + usePolicies: () => ({ + policies: { + security: { + configured: true, + status: "active", + backendId: "backend-1", + runOn: "upload", + }, + }, + }), +})); +vi.mock("@app/services/policyApi", () => ({ + runStoredPolicy: vi.fn(), + getPolicyRun: vi.fn(), + downloadPolicyOutput: vi.fn(), +})); +vi.mock("@app/services/fileStorage", () => ({ + fileStorage: { getStirlingFile: vi.fn() }, +})); + +import { usePolicyAutoRun } from "@app/components/policies/usePolicyAutoRun"; +import { + recordRunStart, + getRun, + resetPolicyRuns, + usePolicyRuns, +} from "@app/components/policies/policyRunStore"; +import { runStoredPolicy, getPolicyRun } from "@app/services/policyApi"; +import { fileStorage } from "@app/services/fileStorage"; + +const getRunApi = vi.mocked(getPolicyRun); +const runStored = vi.mocked(runStoredPolicy); +const getFile = vi.mocked(fileStorage.getStirlingFile); + +const queueFullView = { + runId: "run-1", + status: "FAILED", + currentStep: 0, + stepCount: 2, + error: "Policy run could not be queued: Job queue full", + errorCode: "POLICY_QUEUE_FULL", + outputs: [], +} as never; + +beforeEach(() => { + vi.useFakeTimers(); + localStorage.clear(); + resetPolicyRuns(); + getRunApi.mockReset(); + runStored.mockReset(); + getFile.mockReset(); +}); +afterEach(() => vi.useRealTimers()); + +describe("auto-run queue-rejection retry", () => { + it("relabels a queue-rejected run as retrying, then re-dispatches it in place", async () => { + // The polled run comes back queue-rejected; the retry resolves the file + fires a fresh run. + getRunApi.mockResolvedValue(queueFullView); + getFile.mockResolvedValue({ size: 1234 } as never); + runStored.mockResolvedValue("run-2"); + + recordRunStart({ + runId: "run-1", + categoryId: "security", + fileId: "file-1", + fileName: "doc.pdf", + fileSize: 1234, + status: "RUNNING", + outputs: [], + error: null, + startedAt: 0, + }); + + renderHook(() => { + usePolicyAutoRun(); + return usePolicyRuns(); + }); + + // First poll (2s cadence) sees the rejection → relabel as a soft "retrying" row. + await act(async () => { + await vi.advanceTimersByTimeAsync(2000); + }); + expect(getRun("run-1")?.retrying).toBe(true); + expect(runStored).not.toHaveBeenCalled(); + + // After the first backoff window (BASE 4s) the rejected record is dropped and a fresh run fires. + await act(async () => { + await vi.advanceTimersByTimeAsync(4000); + }); + expect(runStored).toHaveBeenCalledWith("backend-1", [{ size: 1234 }]); + expect(getRun("run-1")).toBeUndefined(); + expect(getRun("run-2")?.status).toBe("PENDING"); + }); +}); diff --git a/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.test.ts b/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.test.ts new file mode 100644 index 000000000..5f4235f2f --- /dev/null +++ b/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.test.ts @@ -0,0 +1,160 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; + +// Mock the network + the run store so we can drive the poll loop deterministically. +vi.mock("@app/services/policyApi", async (orig) => ({ + ...(await orig()), + getPolicyRun: vi.fn(), +})); +vi.mock("@app/components/policies/policyRunStore", async (orig) => ({ + ...(await orig()), + updateRun: vi.fn(), +})); + +import { poll } from "@app/components/policies/usePolicyAutoRun"; +import { getPolicyRun } from "@app/services/policyApi"; +import { updateRun } from "@app/components/policies/policyRunStore"; + +const getRun = vi.mocked(getPolicyRun); +const update = vi.mocked(updateRun); + +const POLL_MS = 2000; +// Must match the poll loop's budget formula (stepCount × per-step timeout + grace). +// With the 2-step `view()` below that's 2 × 300_000 + 30_000 = 630_000ms. +const STEP_TIMEOUT_MS = 300_000; +const POLL_GRACE_MS = 30_000; +const STEP_COUNT = 2; +const BUDGET_MS = STEP_COUNT * STEP_TIMEOUT_MS + POLL_GRACE_MS; +const view = (status: string) => + ({ + runId: "r1", + status, + currentStep: 1, + stepCount: STEP_COUNT, + outputs: [], + error: null, + }) as never; + +beforeEach(() => { + vi.useFakeTimers(); + getRun.mockReset(); + update.mockReset(); +}); +afterEach(() => vi.useRealTimers()); + +/** Run the poll loop for n cadence ticks, flushing the awaited fetch each tick. */ +async function tick(n: number) { + for (let i = 0; i < n; i++) await vi.advanceTimersByTimeAsync(POLL_MS); +} + +describe("policy run poll loop", () => { + it("stops and marks FAILED after repeated run-not-found (404)", async () => { + getRun.mockRejectedValue({ code: "ERR_NOT_FOUND" }); + const p = poll("r1"); + await tick(3); // MAX_NOT_FOUND consecutive 404s + await p; + expect(update).toHaveBeenCalledWith( + "r1", + expect.objectContaining({ status: "FAILED" }), + ); + // It gave up after the not-found streak, not after the full time budget. + expect(getRun.mock.calls.length).toBe(3); + }); + + it("also detects 404 via axios-style response.status", async () => { + getRun.mockRejectedValue({ response: { status: 404 } }); + const p = poll("r1"); + await tick(3); + await p; + expect(update).toHaveBeenCalledWith( + "r1", + expect.objectContaining({ status: "FAILED" }), + ); + }); + + it("does NOT fail on a brief not-found blip that then recovers", async () => { + getRun + .mockRejectedValueOnce({ code: "ERR_NOT_FOUND" }) + .mockResolvedValue(view("COMPLETED")); + const p = poll("r1"); + await tick(2); + await p; + const statuses = update.mock.calls.map( + (c) => (c[1] as { status: string }).status, + ); + expect(statuses).toContain("COMPLETED"); + expect(statuses).not.toContain("FAILED"); + }); + + it("transient non-404 errors don't count toward the not-found streak", async () => { + getRun + .mockRejectedValueOnce({ code: "ERR_NOT_FOUND" }) + .mockRejectedValueOnce({ response: { status: 500 } }) + .mockRejectedValueOnce({ code: "ERR_NOT_FOUND" }) + .mockResolvedValue(view("COMPLETED")); + const p = poll("r1"); + await tick(4); + await p; + const statuses = update.mock.calls.map( + (c) => (c[1] as { status: string }).status, + ); + expect(statuses).toContain("COMPLETED"); + expect(statuses).not.toContain("FAILED"); + }); + + it("marks FAILED when the run never reaches a terminal state within the budget", async () => { + getRun.mockResolvedValue(view("RUNNING")); + const p = poll("r1"); + // Advance past the full step-count-derived budget in one go. + await vi.advanceTimersByTimeAsync(BUDGET_MS + POLL_MS); + await p; + expect(update).toHaveBeenLastCalledWith( + "r1", + expect.objectContaining({ status: "FAILED" }), + ); + }); + + it("keeps polling a long run for the whole step budget (no premature giveup)", async () => { + getRun.mockResolvedValue(view("RUNNING")); + const p = poll("r1"); + // 200s in: a single step may run up to the per-step timeout, so the run is + // still legitimately in flight and must keep being polled, not failed. + await tick(100); + expect(update).not.toHaveBeenCalledWith( + "r1", + expect.objectContaining({ status: "FAILED" }), + ); + // Let it run out so the dangling promise doesn't leak into other tests. + await vi.advanceTimersByTimeAsync(BUDGET_MS); + await p; + }); + + it("records pipeline progress (currentStep/stepCount) while running", async () => { + getRun + .mockResolvedValueOnce(view("RUNNING")) + .mockResolvedValue(view("COMPLETED")); + const p = poll("r1"); + await tick(2); + await p; + expect(update).toHaveBeenCalledWith( + "r1", + expect.objectContaining({ currentStep: 1, stepCount: STEP_COUNT }), + ); + }); + + it("finishes cleanly on a terminal status and fires onTerminal", async () => { + getRun.mockResolvedValue(view("COMPLETED")); + const onTerminal = vi.fn(); + const p = poll("r1", onTerminal); + await tick(1); + await p; + expect(update).toHaveBeenCalledWith( + "r1", + expect.objectContaining({ status: "COMPLETED" }), + ); + expect(onTerminal).toHaveBeenCalledTimes(1); + expect(update).not.toHaveBeenCalledWith( + "r1", + expect.objectContaining({ status: "FAILED" }), + ); + }); +}); diff --git a/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.ts b/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.ts index 49c7f084b..2103745ee 100644 --- a/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.ts +++ b/frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.ts @@ -35,17 +35,68 @@ import type { StirlingFile, StirlingFileStub } from "@app/types/fileContext"; import { usePolicies } from "@app/hooks/usePolicies"; import { dispatchKey, + getRun, isDispatched, markDispatched, recordRunStart, + removeRun, updateRun, usePolicyRuns, type PolicyRunRecord, } from "@app/components/policies/policyRunStore"; -/** Poll cadence + cap for a single run's status (≈2.5 min worst case). */ +/** Status poll cadence. */ const POLL_MS = 2000; -const MAX_POLLS = 75; + +/** The server aborts any single tool step that runs longer than its internal-API + * read timeout, then fails the run — so a run can legitimately stay in flight + * for up to this long per step. The client must keep polling at least that long, + * or it abandons a run the server is still working on (which reads as a hang). */ +const STEP_TIMEOUT_MS = 300_000; + +/** Slack on top of the per-step budget: queueing before the first step starts and + * output handling after the last one finishes. */ +const POLL_GRACE_MS = 30_000; + +/** Step count assumed before the first status report reveals the real pipeline + * length — only governs the budget for those first couple of polls. */ +const DEFAULT_STEP_COUNT = 4; + +/** errorCode the backend sets when a run is rejected at admission (job queue full under load). + * Transient, not a real processing failure — we back off and retry rather than surfacing it. */ +const POLICY_QUEUE_FULL = "POLICY_QUEUE_FULL"; + +/** Auto-retry budget + exponential backoff for a queue-rejected run, to ride out a busy period + * before giving up to a manual retry. Delays are BASE × 2^attempt (≈4s, 8s … 64s, ~2min total). */ +const MAX_QUEUE_RETRIES = 5; +const QUEUE_RETRY_BASE_MS = 4000; + +/** Consecutive "run not found" responses before giving up. The run state lives + * in memory on the server, so a restart or a second instance behind the load + * balancer makes a live run's status return 404 — and it won't come back. We + * tolerate a brief blip (e.g. a poll racing a just-dispatched run, or one hop + * to an instance that hasn't seen it) then fail, rather than polling forever. */ +const MAX_NOT_FOUND = 3; + +/** A 404 from the run-status endpoint, across the web (axios) and desktop + * (tauri http client → {@code code: "ERR_NOT_FOUND"}) builds. */ +function isRunNotFound(err: unknown): boolean { + const e = err as + | { code?: string; status?: number; response?: { status?: number } } + | null + | undefined; + return ( + e?.code === "ERR_NOT_FOUND" || + e?.status === 404 || + e?.response?.status === 404 + ); +} + +/** Mark a run terminal-failed so it stops being polled (and re-polled on reload) + * and the activity feed offers Retry, instead of the file enforcing forever. */ +function failRun(runId: string, message: string): void { + updateRun(runId, { status: "FAILED", error: message, errorCode: null }); +} /** How long to wait for an upload's bytes to land in IndexedDB before giving up * (20 × 250ms ≈ 5s). The stub can surface in the file list a beat before its @@ -81,14 +132,66 @@ export function usePolicyAutoRun(): void { // run so a folder-watch burst opens the modal once, not once per file. const firedLimitModal = useRef>(new Set()); - const onRunFinished = useCallback((view: PolicyRunView) => { - const code = view.errorCode; - if (code !== "PAYG_LIMIT_REACHED" && code !== "FEATURE_DEGRADED") return; - if (firedLimitModal.current.has(view.runId)) return; - firedLimitModal.current.add(view.runId); - dispatchPaygLimitReached(view.errorSubscribed ?? null); + // Latest policies, read from inside the stable retry callback (which has no deps). + const policiesRef = useRef(policies); + policiesRef.current = policies; + // Per-file (dispatchKey) count of consecutive queue-rejection retries, so backoff escalates and + // eventually gives up. Survives the run-id changing on each retry; reset on any real outcome. + const queueRetries = useRef>(new Map()); + + // A queue-rejected run is just backpressure — drop the rejected record and fire a fresh run in + // its place after a growing backoff (one feed row, not a new one per attempt). Once the budget is + // spent, leave the last failure standing so the activity feed offers a manual Retry. + const scheduleQueueRetry = useCallback((runId: string) => { + const rec = getRun(runId); + if (!rec) return; + const key = dispatchKey(rec.categoryId, rec.fileId); + const attempts = queueRetries.current.get(key) ?? 0; + const backendId = policiesRef.current[rec.categoryId]?.backendId; + if (attempts >= MAX_QUEUE_RETRIES || !backendId) { + queueRetries.current.delete(key); + return; + } + queueRetries.current.set(key, attempts + 1); + // Soft-label the row as busy through the backoff window (it's still FAILED underneath). + updateRun(runId, { retrying: true }); + setTimeout( + () => { + removeRun(runId); + void runPolicyOnFile( + rec.categoryId, + backendId, + rec.fileId as FileId, + rec.fileName, + ); + }, + QUEUE_RETRY_BASE_MS * 2 ** attempts, + ); }, []); + const onRunFinished = useCallback( + (view: PolicyRunView) => { + // Transient admission rejection (queue full): back off and retry instead of failing. + if (view.errorCode === POLICY_QUEUE_FULL) { + scheduleQueueRetry(view.runId); + return; + } + // Any genuine terminal outcome clears the file's retry budget so a later run starts fresh. + const finished = getRun(view.runId); + if (finished) { + queueRetries.current.delete( + dispatchKey(finished.categoryId, finished.fileId), + ); + } + const code = view.errorCode; + if (code !== "PAYG_LIMIT_REACHED" && code !== "FEATURE_DEGRADED") return; + if (firedLimitModal.current.has(view.runId)) return; + firedLimitModal.current.add(view.runId); + dispatchPaygLimitReached(view.errorSubscribed ?? null); + }, + [scheduleQueueRetry], + ); + // Dispatch: for each active policy × each session file not yet run, fire a run. useEffect(() => { if (!POLICIES_ENABLED) return; @@ -328,25 +431,49 @@ export async function runPolicyOnFile( } /** - * Poll a run's status until it reaches a terminal state (or the cap). Calls {@code onTerminal} once + * Poll a run's status until it reaches a terminal state (or the budget). Calls {@code onTerminal} once * with the final view when it terminates — the caller uses that to pop the usage-limit modal when a * run was blocked. Only runs polled this session fire it (terminal runs aren't re-polled), so a * persisted failed run never re-triggers a modal on reload. */ -async function poll( +export async function poll( runId: string, onTerminal?: (view: PolicyRunView) => void, ): Promise { - for (let i = 0; i < MAX_POLLS; i++) { + let notFoundStreak = 0; + // Sized to the server's worst case: each step may run up to STEP_TIMEOUT_MS + // before the server itself aborts it, so the budget tracks the real pipeline + // length (learned from the first status report) rather than a flat cap that + // would quit while a long step is still legitimately running. + let budgetMs = DEFAULT_STEP_COUNT * STEP_TIMEOUT_MS + POLL_GRACE_MS; + const startedAt = Date.now(); + while (Date.now() - startedAt < budgetMs) { await delay(POLL_MS); let view; try { view = await getPolicyRun(runId); - } catch { - continue; // transient — keep trying within the cap. + } catch (err) { + // The server lost the run's (in-memory) state — a restart, or a poll that + // hopped to an instance without it. Tolerate a brief blip, then fail so + // the file stops enforcing forever; the user can retry. + if (isRunNotFound(err)) { + if (++notFoundStreak >= MAX_NOT_FOUND) { + failRun(runId, "The enforcement run could no longer be found."); + return; + } + } else { + notFoundStreak = 0; // a non-404 error doesn't confirm the run is gone. + } + continue; // keep trying within the budget. + } + notFoundStreak = 0; + if (view.stepCount > 0) { + budgetMs = view.stepCount * STEP_TIMEOUT_MS + POLL_GRACE_MS; } updateRun(runId, { status: view.status, + currentStep: view.currentStep, + stepCount: view.stepCount, outputs: view.outputs, error: view.error, errorCode: view.errorCode ?? null, @@ -356,4 +483,7 @@ async function poll( return; } } + // Budget exhausted without a terminal status — stop here and fail it, so the + // file doesn't enforce forever and reloads don't re-poll it. + failRun(runId, "Enforcement timed out — the run didn't finish in time."); } diff --git a/frontend/editor/src/proprietary/services/policyLiveData.test.ts b/frontend/editor/src/proprietary/services/policyLiveData.test.ts index 0a60feeae..38b96c845 100644 --- a/frontend/editor/src/proprietary/services/policyLiveData.test.ts +++ b/frontend/editor/src/proprietary/services/policyLiveData.test.ts @@ -60,6 +60,42 @@ describe("runsToActivity", () => { action: "Step 2 failed", }); }); + + it("shows pipeline progress on a running run once the step is known", () => { + const [withStep, noStep] = runsToActivity([ + run({ runId: "a", status: "RUNNING", currentStep: 1, stepCount: 2 }), + run({ runId: "b", status: "RUNNING" }), + ]); + expect(withStep.action).toBe("Enforcing… · step 1/2"); + // Before the first status report (no step yet) it stays the plain label. + expect(noStep.action).toBe("Enforcing…"); + }); + + it("shows a queue-rejected run awaiting retry as busy, not a failure", () => { + const [item] = runsToActivity([ + run({ + status: "FAILED", + error: "Policy run could not be queued: Job queue full", + errorCode: "POLICY_QUEUE_FULL", + retrying: true, + }), + ]); + expect(item.status).toBe("processing"); + expect(item.action).toBe("Busy — retrying…"); + }); + + it("shows a queue rejection that has exhausted its retries as a failure", () => { + const [item] = runsToActivity([ + run({ + status: "FAILED", + error: "Policy run could not be queued: Job queue full", + errorCode: "POLICY_QUEUE_FULL", + // no `retrying` flag — the auto-retry budget is spent. + }), + ]); + expect(item.status).toBe("flagged"); + expect(item.action).toContain("Job queue full"); + }); }); describe("runsToStats", () => { diff --git a/frontend/editor/src/proprietary/services/policyLiveData.ts b/frontend/editor/src/proprietary/services/policyLiveData.ts index 48d2ba2b7..67d53f9c5 100644 --- a/frontend/editor/src/proprietary/services/policyLiveData.ts +++ b/frontend/editor/src/proprietary/services/policyLiveData.ts @@ -49,6 +49,8 @@ function formatBytes(bytes: number): string { /** Map a run's lifecycle status to an activity row's display status. */ function activityStatus(run: PolicyRunRecord): PolicyActivityItem["status"] { if (run.status === "COMPLETED") return "enforced"; + // A queue-rejected run awaiting auto-retry is transient backpressure, not a failure. + if (run.retrying) return "processing"; if (run.status === "FAILED" || run.status === "CANCELLED") return "flagged"; return "processing"; } @@ -59,8 +61,15 @@ function activityAction(run: PolicyRunRecord): string { return `${formatBytes(run.fileSize)} • enforced`; case "flagged": return run.error ?? "Enforcement failed"; - default: - return "Enforcing…"; + default: { + if (run.retrying) return "Busy — retrying…"; + // Show pipeline progress while running, once the status endpoint reports + // it — turns a static "Enforcing…" into visible movement on slow steps. + const { currentStep, stepCount } = run; + return currentStep && stepCount + ? `Enforcing… · step ${currentStep}/${stepCount}` + : "Enforcing…"; + } } }