mirror of
https://github.com/arsvendg/Stirling-PDF.git
synced 2026-07-14 10:34:06 +02:00
fix(policies): poll runs to completion with progress, soft-retry when queue is full (#6690)
## 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.
This commit is contained in:
+6
-1
@@ -59,6 +59,10 @@ public class PolicyEngine {
|
|||||||
// files. See ResourceMonitor#shouldQueueJob(int).
|
// files. See ResourceMonitor#shouldQueueJob(int).
|
||||||
private static final int RUN_RESOURCE_WEIGHT = 50;
|
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 PolicyExecutor stepExecutor;
|
||||||
private final TaskManager taskManager;
|
private final TaskManager taskManager;
|
||||||
private final PolicyRunRegistry registry;
|
private final PolicyRunRegistry registry;
|
||||||
@@ -239,7 +243,8 @@ public class PolicyEngine {
|
|||||||
if (!completion.isDone()) {
|
if (!completion.isDone()) {
|
||||||
String message = "Policy run could not be queued: " + ex.getMessage();
|
String message = "Policy run could not be queued: " + ex.getMessage();
|
||||||
log.error("Policy run {} was not admitted: {}", run.getRunId(), 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);
|
taskManager.setError(run.getRunId(), message);
|
||||||
completion.complete(run);
|
completion.complete(run);
|
||||||
}
|
}
|
||||||
|
|||||||
+21
@@ -336,6 +336,27 @@ class PolicyEngineTest {
|
|||||||
assertEquals(PolicyRunStatus.PENDING, registry.get(handle.runId()).getStatus());
|
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<Object> 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
|
@Test
|
||||||
void resumeIsNotYetImplemented() {
|
void resumeIsNotYetImplemented() {
|
||||||
assertThrows(UnsupportedOperationException.class, () -> engine.resume("any", List.of()));
|
assertThrows(UnsupportedOperationException.class, () -> engine.resume("any", List.of()));
|
||||||
|
|||||||
@@ -1,9 +1,11 @@
|
|||||||
import { describe, it, expect, beforeEach } from "vitest";
|
import { describe, it, expect, beforeEach } from "vitest";
|
||||||
import {
|
import {
|
||||||
dispatchKey,
|
dispatchKey,
|
||||||
|
getRun,
|
||||||
isDispatched,
|
isDispatched,
|
||||||
markDispatched,
|
markDispatched,
|
||||||
recordRunStart,
|
recordRunStart,
|
||||||
|
removeRun,
|
||||||
updateRun,
|
updateRun,
|
||||||
resetPolicyRuns,
|
resetPolicyRuns,
|
||||||
type PolicyRunRecord,
|
type PolicyRunRecord,
|
||||||
@@ -70,6 +72,16 @@ describe("policyRunStore", () => {
|
|||||||
expect(read("stirling-policy-runs").runs[0].status).toBe("PENDING");
|
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", () => {
|
it("caps stored runs at 50, newest first", () => {
|
||||||
for (let i = 0; i < 55; i++) {
|
for (let i = 0; i < 55; i++) {
|
||||||
recordRunStart(rec({ runId: `r${i}`, fileId: `f${i}`, startedAt: i }));
|
recordRunStart(rec({ runId: `r${i}`, fileId: `f${i}`, startedAt: i }));
|
||||||
|
|||||||
@@ -19,6 +19,11 @@ export interface PolicyRunRecord {
|
|||||||
fileName: string;
|
fileName: string;
|
||||||
fileSize: number;
|
fileSize: number;
|
||||||
status: PolicyRunStatus;
|
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. */
|
/** Output files (downloadable via /api/v1/general/files/{id}) once done. */
|
||||||
outputs: { fileId: string; fileName: string }[];
|
outputs: { fileId: string; fileName: string }[];
|
||||||
/** True once ALL outputs have been imported into the workspace. */
|
/** True once ALL outputs have been imported into the workspace. */
|
||||||
@@ -33,6 +38,9 @@ export interface PolicyRunRecord {
|
|||||||
error: string | null;
|
error: string | null;
|
||||||
/** Stable backend failure code (e.g. an entitlement sentinel) when FAILED; null otherwise. */
|
/** Stable backend failure code (e.g. an entitlement sentinel) when FAILED; null otherwise. */
|
||||||
errorCode?: string | null;
|
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. */
|
/** Epoch ms when the run was dispatched. */
|
||||||
startedAt: number;
|
startedAt: number;
|
||||||
}
|
}
|
||||||
@@ -146,6 +154,19 @@ export function updateRun(runId: string, patch: Partial<PolicyRunRecord>) {
|
|||||||
emit();
|
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. */
|
/** Reset the store — used by tests to isolate it. */
|
||||||
export function resetPolicyRuns() {
|
export function resetPolicyRuns() {
|
||||||
state = { runs: [], dispatched: [] };
|
state = { runs: [], dispatched: [] };
|
||||||
|
|||||||
@@ -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");
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -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<typeof import("@app/services/policyApi")>()),
|
||||||
|
getPolicyRun: vi.fn(),
|
||||||
|
}));
|
||||||
|
vi.mock("@app/components/policies/policyRunStore", async (orig) => ({
|
||||||
|
...(await orig<typeof import("@app/components/policies/policyRunStore")>()),
|
||||||
|
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" }),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -35,17 +35,68 @@ import type { StirlingFile, StirlingFileStub } from "@app/types/fileContext";
|
|||||||
import { usePolicies } from "@app/hooks/usePolicies";
|
import { usePolicies } from "@app/hooks/usePolicies";
|
||||||
import {
|
import {
|
||||||
dispatchKey,
|
dispatchKey,
|
||||||
|
getRun,
|
||||||
isDispatched,
|
isDispatched,
|
||||||
markDispatched,
|
markDispatched,
|
||||||
recordRunStart,
|
recordRunStart,
|
||||||
|
removeRun,
|
||||||
updateRun,
|
updateRun,
|
||||||
usePolicyRuns,
|
usePolicyRuns,
|
||||||
type PolicyRunRecord,
|
type PolicyRunRecord,
|
||||||
} from "@app/components/policies/policyRunStore";
|
} 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 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
|
/** 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
|
* (20 × 250ms ≈ 5s). The stub can surface in the file list a beat before its
|
||||||
@@ -81,13 +132,65 @@ export function usePolicyAutoRun(): void {
|
|||||||
// run so a folder-watch burst opens the modal once, not once per file.
|
// run so a folder-watch burst opens the modal once, not once per file.
|
||||||
const firedLimitModal = useRef<Set<string>>(new Set());
|
const firedLimitModal = useRef<Set<string>>(new Set());
|
||||||
|
|
||||||
const onRunFinished = useCallback((view: PolicyRunView) => {
|
// 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<Map<string, number>>(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;
|
const code = view.errorCode;
|
||||||
if (code !== "PAYG_LIMIT_REACHED" && code !== "FEATURE_DEGRADED") return;
|
if (code !== "PAYG_LIMIT_REACHED" && code !== "FEATURE_DEGRADED") return;
|
||||||
if (firedLimitModal.current.has(view.runId)) return;
|
if (firedLimitModal.current.has(view.runId)) return;
|
||||||
firedLimitModal.current.add(view.runId);
|
firedLimitModal.current.add(view.runId);
|
||||||
dispatchPaygLimitReached(view.errorSubscribed ?? null);
|
dispatchPaygLimitReached(view.errorSubscribed ?? null);
|
||||||
}, []);
|
},
|
||||||
|
[scheduleQueueRetry],
|
||||||
|
);
|
||||||
|
|
||||||
// Dispatch: for each active policy × each session file not yet run, fire a run.
|
// Dispatch: for each active policy × each session file not yet run, fire a run.
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
@@ -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
|
* 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
|
* 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.
|
* persisted failed run never re-triggers a modal on reload.
|
||||||
*/
|
*/
|
||||||
async function poll(
|
export async function poll(
|
||||||
runId: string,
|
runId: string,
|
||||||
onTerminal?: (view: PolicyRunView) => void,
|
onTerminal?: (view: PolicyRunView) => void,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
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);
|
await delay(POLL_MS);
|
||||||
let view;
|
let view;
|
||||||
try {
|
try {
|
||||||
view = await getPolicyRun(runId);
|
view = await getPolicyRun(runId);
|
||||||
} catch {
|
} catch (err) {
|
||||||
continue; // transient — keep trying within the cap.
|
// 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, {
|
updateRun(runId, {
|
||||||
status: view.status,
|
status: view.status,
|
||||||
|
currentStep: view.currentStep,
|
||||||
|
stepCount: view.stepCount,
|
||||||
outputs: view.outputs,
|
outputs: view.outputs,
|
||||||
error: view.error,
|
error: view.error,
|
||||||
errorCode: view.errorCode ?? null,
|
errorCode: view.errorCode ?? null,
|
||||||
@@ -356,4 +483,7 @@ async function poll(
|
|||||||
return;
|
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.");
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -60,6 +60,42 @@ describe("runsToActivity", () => {
|
|||||||
action: "Step 2 failed",
|
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", () => {
|
describe("runsToStats", () => {
|
||||||
|
|||||||
@@ -49,6 +49,8 @@ function formatBytes(bytes: number): string {
|
|||||||
/** Map a run's lifecycle status to an activity row's display status. */
|
/** Map a run's lifecycle status to an activity row's display status. */
|
||||||
function activityStatus(run: PolicyRunRecord): PolicyActivityItem["status"] {
|
function activityStatus(run: PolicyRunRecord): PolicyActivityItem["status"] {
|
||||||
if (run.status === "COMPLETED") return "enforced";
|
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";
|
if (run.status === "FAILED" || run.status === "CANCELLED") return "flagged";
|
||||||
return "processing";
|
return "processing";
|
||||||
}
|
}
|
||||||
@@ -59,8 +61,15 @@ function activityAction(run: PolicyRunRecord): string {
|
|||||||
return `${formatBytes(run.fileSize)} • enforced`;
|
return `${formatBytes(run.fileSize)} • enforced`;
|
||||||
case "flagged":
|
case "flagged":
|
||||||
return run.error ?? "Enforcement failed";
|
return run.error ?? "Enforcement failed";
|
||||||
default:
|
default: {
|
||||||
return "Enforcing…";
|
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…";
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user