diff --git a/.github/workflows/sweep.yml b/.github/workflows/sweep.yml index 58a34e658a..d23ef5190f 100644 --- a/.github/workflows/sweep.yml +++ b/.github/workflows/sweep.yml @@ -1672,12 +1672,16 @@ jobs: NODE )" signature="$(PAYLOAD="$payload" node -e 'const crypto=require("node:crypto"); process.stdout.write(`sha256=${crypto.createHmac("sha256", process.env.CLAWSWEEPER_WEBHOOK_SECRET).update(process.env.PAYLOAD).digest("hex")}`)')" - curl --fail --silent --show-error --connect-timeout 5 --max-time 20 \ + response="$(curl --fail --silent --show-error --connect-timeout 5 --max-time 20 \ --request POST \ --header "content-type: application/json" \ --header "x-clawsweeper-exact-review-signature: $signature" \ --data "$payload" \ - "$queue_url/internal/exact-review/enqueue" | jq -e '.ok == true and (.queued == true or .deduped == true)' + "$queue_url/internal/exact-review/enqueue")" + jq -e '.ok == true and (.queued == true or .deduped == true or .shed == true)' <<< "$response" >/dev/null + if jq -e '.shed == true' <<< "$response" >/dev/null; then + echo "Source-drift recovery shed by exact-review queue backpressure." + fi - name: Release terminal review leases id: release-terminal-review-leases @@ -3437,9 +3441,11 @@ jobs: --header "x-clawsweeper-exact-review-signature: $signature" \ --data "$payload" \ "$queue_url/internal/exact-review/enqueue")" || response="" - if jq -e '.ok == true and (.queued == true or .deduped == true or .accepted == false)' <<<"$response" >/dev/null; then + if jq -e '.ok == true and (.queued == true or .deduped == true or .shed == true or .accepted == false)' <<<"$response" >/dev/null; then if jq -e '.accepted == false' <<<"$response" >/dev/null; then echo "Recovery skipped because the target is disabled." + elif jq -e '.shed == true' <<<"$response" >/dev/null; then + echo "Recovery shed by exact-review queue backpressure." fi queued=true break diff --git a/dashboard/exact-review-health.ts b/dashboard/exact-review-health.ts index c5daa584c1..c8e15a413d 100644 --- a/dashboard/exact-review-health.ts +++ b/dashboard/exact-review-health.ts @@ -35,6 +35,8 @@ export type ExactReviewHandoffHealth = { capacity: number; active: number; available_slots: number; + pending_depth: number; + shed_since_reset: number; phases: Record; }; @@ -47,6 +49,7 @@ export function summarizeExactReviewHandoff({ capacity, dispatchLeaseMs, executionLeaseMs, + shedSinceReset = 0, }: { items: ExactReviewHealthItem[]; dispatcher?: ExactReviewHealthDispatcher; @@ -54,9 +57,11 @@ export function summarizeExactReviewHandoff({ capacity: number; dispatchLeaseMs: number; executionLeaseMs: number; + shedSinceReset?: number; }): ExactReviewHandoffHealth { const safeNow = finiteTimestamp(now, Date.now()); const safeCapacity = Math.max(0, Math.floor(finiteNumber(capacity, 0))); + const safeShedSinceReset = Math.max(0, Math.floor(finiteNumber(shedSinceReset, 0))); const safeLeaseMs = Math.max(1_000, finiteNumber(dispatchLeaseMs, 10 * 60_000)); const safeExecutionLeaseMs = Math.max(1_000, finiteNumber(executionLeaseMs, 130 * 60_000)); const warningMs = Math.min(2 * 60_000, Math.max(30_000, Math.floor(safeLeaseMs / 3))); @@ -101,6 +106,8 @@ export function summarizeExactReviewHandoff({ capacity: safeCapacity, active, available_slots: Math.max(0, safeCapacity - active), + pending_depth: phases.pending.count, + shed_since_reset: safeShedSinceReset, phases, }; if (items.length === 0) { diff --git a/dashboard/worker.ts b/dashboard/worker.ts index 29e0e77f0c..e6516dde11 100644 --- a/dashboard/worker.ts +++ b/dashboard/worker.ts @@ -37,6 +37,12 @@ type WorkflowRunSummary = { const FAILED_REVIEW_SHARD_RECOVERY_SOURCE_ACTION = "failed_review_shard_recovery"; const EXACT_REVIEW_ARTIFACT_PUBLISH_SOURCE_ACTION = "exact_review_artifact_publish"; const EXACT_REVIEW_ARTIFACT_RETENTION_RECOVERY_SOURCE_ACTION = "artifact_retention_recovery"; +const EXACT_REVIEW_SOURCE_DRIFT_REQUEUE_SOURCE_ACTION = "source_drift_requeue"; +const EXACT_REVIEW_LOW_PRIORITY_SOURCE_ACTIONS = new Set([ + FAILED_REVIEW_SHARD_RECOVERY_SOURCE_ACTION, + EXACT_REVIEW_ARTIFACT_RETENTION_RECOVERY_SOURCE_ACTION, + EXACT_REVIEW_SOURCE_DRIFT_REQUEUE_SOURCE_ACTION, +]); type ExactReviewBaseDecision = { targetRepo: string; @@ -99,6 +105,7 @@ type ExactReviewClaimedRun = { }; type ExactReviewQueueState = { items: Record; + shedSinceReset?: number; dispatcher?: { state: "active" | "paused" | "blocked" | "unknown"; reason?: "workflow_not_active" | "workflow_status_unavailable"; @@ -119,6 +126,7 @@ type ExactReviewQueueStorageMeta = { migrated_at: number; storage_generation: number; dispatcher_json: string | null; + shed_since_reset?: number; }; type DurableObjectStub = { fetch: (request: Request) => Promise }; type DurableObjectNamespace = { @@ -164,6 +172,9 @@ const DEFAULT_EXACT_REVIEW_EXECUTION_LEASE_MS = 130 * 60 * 1000; const DEFAULT_EXACT_REVIEW_HEARTBEAT_GRACE_MS = 20 * 60 * 1000; const DEFAULT_EXACT_REVIEW_RETRY_MS = 30_000; const DEFAULT_EXACT_REVIEW_WORKFLOW_PAUSED_RETRY_MS = 60_000; +const DEFAULT_EXACT_REVIEW_DISPATCH_DEBOUNCE_MS = 45_000; +const DEFAULT_EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS = 3 * 60_000; +const DEFAULT_EXACT_REVIEW_PENDING_SOFT_LIMIT = 300; const EXACT_REVIEW_COMPLETION_RETRY_MAX_MS = 2 * 60 * 60 * 1000; const EXACT_REVIEW_ARTIFACT_RETRY_MAX_MS = 80 * 24 * 60 * 60 * 1000; const EXACT_REVIEW_RECONCILE_RUN_LIMIT = 128; @@ -592,7 +603,6 @@ export class ExactReviewQueue { ); const key = exactReviewItemKey(decision); const current = state.items[key]; - const nextAttemptAt = exactReviewQueueEnqueueAttemptAt(state, now); if (current) { const ignoredRecovery = decision.sourceAction === FAILED_REVIEW_SHARD_RECOVERY_SOURCE_ACTION; @@ -608,10 +618,32 @@ export class ExactReviewQueue { : decision; current.revision += 1; current.updatedAt = now; - current.nextAttemptAt = nextAttemptAt; - if (current.state === "pending") current.attempts = 0; + // Immediacy must come from the merged decision: a pending explicit command + // keeps its command marker through the merge, and a later plain webhook + // event must not re-debounce it. + current.nextAttemptAt = + current.state === "pending" + ? exactReviewQueueDebouncedAttemptAt( + state, + current.decision, + now, + current.createdAt, + this.env, + ) + : exactReviewQueueEnqueueAttemptAt(state, now); + if (current.state === "pending") { + current.attempts = 0; + } } } else { + if ( + isLowPriorityExactReviewDecision(decision) && + exactReviewQueuePendingCount(state) >= exactReviewPendingSoftLimit(this.env) + ) { + state.shedSinceReset = exactReviewShedSinceReset(state) + 1; + this.writeStateSync(state); + return { shed: true as const }; + } state.items[key] = { key, decision, @@ -619,7 +651,7 @@ export class ExactReviewQueue { revision: 1, createdAt: now, updatedAt: now, - nextAttemptAt, + nextAttemptAt: exactReviewQueueDebouncedAttemptAt(state, decision, now, now, this.env), attempts: 0, }; } @@ -629,6 +661,9 @@ export class ExactReviewQueue { if (accepted.deduped) { return json({ ok: true, deduped: true, item_key: exactReviewItemKey(decision) }, 202); } + if (accepted.shed) { + return json({ ok: true, shed: true, reason: "backpressure" }, 202); + } await this.scheduleNext(accepted.state, now); return json({ ok: true, queued: true, item_key: accepted.key }, 202); } @@ -1011,7 +1046,7 @@ export class ExactReviewQueue { exactReviewPublicationDispatchLeaseMs(this.env), exactReviewHeartbeatGraceMs(this.env), ); - const expiredSnapshot = expireExactReviewPublicationItems(snapshot, startedAt); + const expiredSnapshot = expireExactReviewPublicationItems(snapshot, startedAt, this.env); const snapshotChanged = reclaimedSnapshot || expiredSnapshot; const capacity = exactReviewQueueCapacity(this.env); const targetCapacity = exactReviewTargetCapacity(this.env); @@ -1049,7 +1084,7 @@ export class ExactReviewQueue { exactReviewPublicationDispatchLeaseMs(this.env), exactReviewHeartbeatGraceMs(this.env), ); - expireExactReviewPublicationItems(state, now); + expireExactReviewPublicationItems(state, now, this.env); const admitted = exactReviewQueueAdmittedItems( state, now, @@ -1198,11 +1233,13 @@ export class ExactReviewQueue { : null; this.storage.sql.exec( `INSERT INTO ${EXACT_REVIEW_QUEUE_META_TABLE} - (singleton_id, schema_version, migrated_at, storage_generation, dispatcher_json) - VALUES (1, ?, ?, 1, ?)`, + (singleton_id, schema_version, migrated_at, storage_generation, dispatcher_json, + shed_since_reset) + VALUES (1, ?, ?, 1, ?, ?)`, EXACT_REVIEW_QUEUE_STORAGE_SCHEMA_VERSION, migratedAt, dispatcherJson, + exactReviewShedSinceReset(legacy || { items: {} }), ); migratedLegacy = true; this.syncLegacyCompatibilitySync(this.readStateSync()); @@ -1236,9 +1273,22 @@ export class ExactReviewQueue { schema_version INTEGER NOT NULL, migrated_at INTEGER NOT NULL, storage_generation INTEGER NOT NULL, - dispatcher_json TEXT + dispatcher_json TEXT, + shed_since_reset INTEGER NOT NULL DEFAULT 0 ) STRICT`, ); + const hasShedCounter = Array.from( + this.storage.sql.exec( + `SELECT name FROM pragma_table_info('${EXACT_REVIEW_QUEUE_META_TABLE}') + WHERE name = 'shed_since_reset'`, + ), + ).length; + if (!hasShedCounter) { + this.storage.sql.exec( + `ALTER TABLE ${EXACT_REVIEW_QUEUE_META_TABLE} + ADD COLUMN shed_since_reset INTEGER NOT NULL DEFAULT 0`, + ); + } this.storage.sql.exec( `CREATE TABLE IF NOT EXISTS ${EXACT_REVIEW_QUEUE_ITEM_TABLE} ( item_key TEXT PRIMARY KEY, @@ -1260,7 +1310,8 @@ export class ExactReviewQueue { private readStorageMetaSync() { return Array.from( this.storage.sql.exec( - `SELECT schema_version, migrated_at, storage_generation, dispatcher_json + `SELECT schema_version, migrated_at, storage_generation, dispatcher_json, + shed_since_reset FROM ${EXACT_REVIEW_QUEUE_META_TABLE} WHERE singleton_id = 1`, ), @@ -1375,13 +1426,15 @@ export class ExactReviewQueue { if (!replaceState && receiptChanges.length === 0) return; this.storage.sql.exec( `UPDATE ${EXACT_REVIEW_QUEUE_META_TABLE} - SET dispatcher_json = ?, storage_generation = storage_generation + 1 + SET dispatcher_json = ?, shed_since_reset = ?, + storage_generation = storage_generation + 1 WHERE singleton_id = 1 AND storage_generation = ?`, replaceState && legacyState.dispatcher ? JSON.stringify(legacyState.dispatcher) : replaceState ? null : meta.dispatcher_json, + replaceState ? exactReviewShedSinceReset(legacyState) : Number(meta.shed_since_reset || 0), sqlGeneration, ); const reconciledGeneration = this.readStorageMetaSync()?.storage_generation; @@ -1398,6 +1451,7 @@ export class ExactReviewQueue { ) as Record; return { items, + shedSinceReset: exactReviewShedSinceReset(legacy), ...(legacy.dispatcher && typeof legacy.dispatcher === "object" ? { dispatcher: legacy.dispatcher } : {}), @@ -1473,7 +1527,11 @@ export class ExactReviewQueue { throw new Error("invalid exact-review queue dispatcher JSON"); } } - const state = { items, dispatcher }; + const state = { + items, + dispatcher, + shedSinceReset: Math.max(0, Number(meta.shed_since_reset || 0)), + }; this.baselines.set(state, { items: baselineItems, dispatcherJson: meta.dispatcher_json, @@ -1512,12 +1570,17 @@ export class ExactReviewQueue { const dispatcherJson = state.dispatcher ? JSON.stringify(state.dispatcher) : null; this.storage.sql.exec( `UPDATE ${EXACT_REVIEW_QUEUE_META_TABLE} - SET dispatcher_json = ?, storage_generation = storage_generation + 1 + SET dispatcher_json = ?, shed_since_reset = ?, + storage_generation = storage_generation + 1 WHERE singleton_id = 1`, dispatcherJson, + exactReviewShedSinceReset(state), ); this.syncLegacyCompatibilitySync(state); - this.baselines.set(state, { items: nextItems, dispatcherJson }); + this.baselines.set(state, { + items: nextItems, + dispatcherJson, + }); } private readStateBaselineSync(): ExactReviewQueueBaseline { @@ -1616,6 +1679,7 @@ export class ExactReviewQueue { }, items: state.items, dispatcher: state.dispatcher, + shedSinceReset: exactReviewShedSinceReset(state), }; const shadowBytes = new TextEncoder().encode(JSON.stringify(shadow)).byteLength; if (shadowBytes > EXACT_REVIEW_QUEUE_LEGACY_SHADOW_MAX_BYTES) { @@ -3114,7 +3178,7 @@ function reclaimExpiredExactReviewLease( return true; } -function expireExactReviewPublicationItems(state: ExactReviewQueueState, now: number) { +function expireExactReviewPublicationItems(state: ExactReviewQueueState, now: number, env) { let changed = false; for (const [key, item] of Object.entries(state.items)) { const publication = item.decision.publication; @@ -3140,9 +3204,24 @@ function expireExactReviewPublicationItems(state: ExactReviewQueueState, now: nu current.decision = mergePendingExactReviewDecision(current.decision, decision); current.revision += 1; current.updatedAt = now; - current.nextAttemptAt = exactReviewQueueEnqueueAttemptAt(state, now); + // Merged decision, not the raw recovery: a pending explicit command must + // keep its immediate attempt time (same rule as the enqueue merge path). + current.nextAttemptAt = exactReviewQueueDebouncedAttemptAt( + state, + current.decision, + now, + current.createdAt, + env, + ); current.attempts = 0; } else if (!current) { + // The expired publication was already deleted above; shedding here only + // suppresses creation of its replacement recovery item. + if (exactReviewQueuePendingCount(state) >= exactReviewPendingSoftLimit(env)) { + state.shedSinceReset = exactReviewShedSinceReset(state) + 1; + changed = true; + continue; + } state.items[recoveryKey] = { key: recoveryKey, decision, @@ -3150,7 +3229,7 @@ function expireExactReviewPublicationItems(state: ExactReviewQueueState, now: nu revision: 1, createdAt: now, updatedAt: now, - nextAttemptAt: exactReviewQueueEnqueueAttemptAt(state, now), + nextAttemptAt: exactReviewQueueDebouncedAttemptAt(state, decision, now, now, env), attempts: 0, }; } @@ -3167,6 +3246,39 @@ function exactReviewQueueEnqueueAttemptAt(state: ExactReviewQueueState, now: num : now; } +function exactReviewQueueDebouncedAttemptAt( + state: ExactReviewQueueState, + decision: ExactReviewDecision, + now: number, + firstEnqueuedAt: number, + env, +) { + const baseAttemptAt = exactReviewQueueEnqueueAttemptAt(state, now); + if (isImmediateExactReviewDecision(decision)) return baseAttemptAt; + const debounceAt = Math.min( + now + exactReviewDispatchDebounceMs(env), + firstEnqueuedAt + exactReviewDispatchDebounceMaxMs(env), + ); + return Math.max(baseAttemptAt, debounceAt); +} + +function isImmediateExactReviewDecision(decision: ExactReviewDecision) { + return Boolean(decision.commandStatusMarker || decision.publication); +} + +function isLowPriorityExactReviewDecision(decision: ExactReviewDecision) { + return EXACT_REVIEW_LOW_PRIORITY_SOURCE_ACTIONS.has(decision.sourceAction); +} + +function exactReviewQueuePendingCount(state: ExactReviewQueueState) { + return Object.values(state.items).filter((item) => item.state === "pending").length; +} + +function exactReviewShedSinceReset(state: Pick) { + const value = Number(state.shedSinceReset || 0); + return Number.isSafeInteger(value) && value > 0 ? value : 0; +} + function exactReviewQueueIsPublication(item: ExactReviewQueueItem) { return item.decision.sourceAction === EXACT_REVIEW_ARTIFACT_PUBLISH_SOURCE_ACTION; } @@ -3244,6 +3356,7 @@ function exactReviewQueueStats( const handoffHealth = summarizeExactReviewHandoff({ items, dispatcher: state.dispatcher, + shedSinceReset: exactReviewShedSinceReset(state), now, capacity, dispatchLeaseMs, @@ -3310,6 +3423,7 @@ function exactReviewQueueStats( items.filter((item) => !exactReviewQueueIsPublication(item)), now, capacity, + exactReviewShedSinceReset(state), ), publication: exactReviewQueueLaneStats( items.filter(exactReviewQueueIsPublication), @@ -3319,6 +3433,7 @@ function exactReviewQueueStats( }; return { pending: handoffHealth.phases.pending.count, + shed_since_reset: exactReviewShedSinceReset(state), dispatching: handoffHealth.phases.dispatching.count, leased: handoffHealth.phases.leased.count, oldest_pending_at: handoffHealth.phases.pending.oldest_at, @@ -3343,7 +3458,12 @@ function exactReviewQueueStats( }; } -function exactReviewQueueLaneStats(items: ExactReviewQueueItem[], now: number, capacity: number) { +function exactReviewQueueLaneStats( + items: ExactReviewQueueItem[], + now: number, + capacity: number, + shedSinceReset = 0, +) { const pendingItems = items.filter((item) => item.state === "pending"); const dispatchingItems = items.filter((item) => item.state === "dispatching"); const leasedItems = items.filter((item) => item.state === "leased"); @@ -3358,6 +3478,8 @@ function exactReviewQueueLaneStats(items: ExactReviewQueueItem[], now: number, c ); return { pending: pendingItems.length, + pending_depth: pendingItems.length, + shed_since_reset: shedSinceReset, ready: pendingItems.filter((item) => item.nextAttemptAt <= now).length, backoff: pendingItems.filter((item) => item.nextAttemptAt > now).length, dispatching: dispatchingItems.length, @@ -3551,6 +3673,39 @@ function exactReviewWorkflowPausedRetryMs(env) { ); } +function exactReviewDispatchDebounceMs(env) { + return Math.max( + 0, + Math.min( + 15 * 60_000, + numberFrom(env.EXACT_REVIEW_DISPATCH_DEBOUNCE_MS, DEFAULT_EXACT_REVIEW_DISPATCH_DEBOUNCE_MS), + ), + ); +} + +function exactReviewDispatchDebounceMaxMs(env) { + return Math.max( + 0, + Math.min( + 60 * 60_000, + numberFrom( + env.EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS, + DEFAULT_EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS, + ), + ), + ); +} + +function exactReviewPendingSoftLimit(env) { + return Math.max( + 1, + Math.min( + 100_000, + numberFrom(env.EXACT_REVIEW_PENDING_SOFT_LIMIT, DEFAULT_EXACT_REVIEW_PENDING_SOFT_LIMIT), + ), + ); +} + async function exactReviewDispatchToken(env) { return exactReviewRepositoryToken(env, { actions: "read", contents: "write" }); } diff --git a/dashboard/wrangler.toml b/dashboard/wrangler.toml index 8f77168673..34f95aa371 100644 --- a/dashboard/wrangler.toml +++ b/dashboard/wrangler.toml @@ -51,3 +51,6 @@ EXACT_REVIEW_DISPATCH_LEASE_MS = "360000" EXACT_REVIEW_EXECUTION_LEASE_MS = "7800000" EXACT_REVIEW_HEARTBEAT_GRACE_MS = "1200000" EXACT_REVIEW_WORKFLOW_PAUSED_RETRY_MS = "60000" +EXACT_REVIEW_DISPATCH_DEBOUNCE_MS = "45000" +EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS = "180000" +EXACT_REVIEW_PENDING_SOFT_LIMIT = "300" diff --git a/docs/limits.md b/docs/limits.md index 2f915e4117..2b84f48373 100644 --- a/docs/limits.md +++ b/docs/limits.md @@ -126,6 +126,14 @@ backlog drain. Exact capacity is consumed only while queue work is pending. As those priority workers start, normal, hot-intake, and commit-review planners count them and reduce their next background wave. +Fresh webhook work waits for `EXACT_REVIEW_DISPATCH_DEBOUNCE_MS` (45 seconds by +default) so rapid edits and pushes coalesce before dispatch. Repeated pending +revisions extend that delay up to `EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS` (three +minutes by default) from the item's first enqueue. Explicit command work and +publication work bypass the delay. When pending depth reaches +`EXACT_REVIEW_PENDING_SOFT_LIMIT` (300 by default), new recovery-only work is +shed; existing items, webhook events, commands, and publications remain admitted. + Exact-review result publication has a separate 24-workflow Actions lane. Its checkout, artifact handling, comment sync, and result routing are deterministic control-plane work: they consume GitHub runners, but not Codex slots. The @@ -189,6 +197,12 @@ hot intake `14`, and commit review `2`. Existing repair lanes keep their ## Runtime Overrides +- `EXACT_REVIEW_DISPATCH_DEBOUNCE_MS` overrides the 45,000 ms coalescing delay + for fresh non-command exact-review events. +- `EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS` overrides the 180,000 ms maximum + coalescing window measured from the item's first enqueue. +- `EXACT_REVIEW_PENDING_SOFT_LIMIT` overrides the pending-depth threshold for + shedding new recovery-only exact-review work; the default is 300. - `EXACT_REVIEW_HEARTBEAT_GRACE_MS` overrides the 1,200,000 ms exact-review worker heartbeat grace. It is clamped to at least 420,000 ms so a configured grace can never dip below the five-minute worker heartbeat interval plus request time and jitter. diff --git a/test/clawsweeper.test.ts b/test/clawsweeper.test.ts index 3a9071a275..c2b31af85f 100644 --- a/test/clawsweeper.test.ts +++ b/test/clawsweeper.test.ts @@ -2575,9 +2575,10 @@ test("sweep review recovery uses explicit failed shard artifacts", () => { assert.match(recoveryJob, /\/internal\/exact-review\/enqueue/); assert.match( recoveryJob, - /\.ok == true and \(\.queued == true or \.deduped == true or \.accepted == false\)/, + /\.ok == true and \(\.queued == true or \.deduped == true or \.shed == true or \.accepted == false\)/, ); assert.match(recoveryJob, /Recovery skipped because the target is disabled/); + assert.match(recoveryJob, /Recovery shed by exact-review queue backpressure/); assert.match(recoveryJob, /for attempt in 1 2 3/); assert.match(recoveryJob, /failed_recovery_dispatches/); assert.match( diff --git a/test/dashboard-worker.test.ts b/test/dashboard-worker.test.ts index 4f1f9fe75c..a703af6ae5 100644 --- a/test/dashboard-worker.test.ts +++ b/test/dashboard-worker.test.ts @@ -53,6 +53,145 @@ test("exact-review publication defaults to 24 bounded publishers", () => { ); }); +test("exact-review queue debounces fresh work and caps pending revision extensions", async () => { + const originalNow = Date.now; + let now = 1_000_000; + Date.now = () => now; + try { + const storage = new MemoryDurableStorage(); + const queue = new ExactReviewQueue( + { storage }, + { + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "1000", + EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS: "1500", + }, + ); + await queue.fetch(buildExactReviewQueueRequest("debounce-1", 750, "edited")); + let state = (await storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(state.items["openclaw/gogcli#750"].nextAttemptAt, 1_001_000); + + now += 500; + await queue.fetch(buildExactReviewQueueRequest("debounce-2", 750, "synchronize")); + state = (await storage.get("exact-review-queue")) as typeof state; + assert.equal(state.items["openclaw/gogcli#750"].nextAttemptAt, 1_001_500); + assert.equal(state.items["openclaw/gogcli#750"].revision, 2); + + now += 900; + await queue.fetch(buildExactReviewQueueRequest("debounce-3", 750, "edited")); + state = (await storage.get("exact-review-queue")) as typeof state; + assert.equal(state.items["openclaw/gogcli#750"].nextAttemptAt, 1_001_500); + assert.equal(state.items["openclaw/gogcli#750"].revision, 3); + } finally { + Date.now = originalNow; + } +}); + +test("exact-review queue bypasses debounce for commands and publications", async () => { + const originalNow = Date.now; + Date.now = () => 2_000_000; + try { + const storage = new MemoryDurableStorage(); + const queue = new ExactReviewQueue({ storage }, {}); + const commandStatusMarker = + ""; + await queue.fetch( + buildExactReviewQueueRequest( + "command-immediate", + 751, + "legacy_dispatch", + "issue", + undefined, + { + commandStatusMarker, + }, + ), + ); + await queue.fetch( + buildExactReviewQueueRequest( + "publication-immediate", + 752, + "exact_review_artifact_publish", + "issue", + undefined, + exactReviewPublicationOverrides(752, "7520"), + ), + ); + const state = (await storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(state.items["openclaw/gogcli#751"].nextAttemptAt, 2_000_000); + assert.equal(state.items["openclaw/gogcli#752@publish:7520:1"].nextAttemptAt, 2_000_000); + + // A later plain webhook event merging into the pending command must not + // re-debounce it: immediacy comes from the merged decision's command marker. + await queue.fetch(buildExactReviewQueueRequest("command-followup", 751, "edited")); + const merged = (await storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(merged.items["openclaw/gogcli#751"].revision, 2); + assert.equal(merged.items["openclaw/gogcli#751"].nextAttemptAt, 2_000_000); + } finally { + Date.now = originalNow; + } +}); + +test("exact-review queue sheds only new recovery work above the pending soft limit", async () => { + const storage = new MemoryDurableStorage(); + const env = { + EXACT_REVIEW_PENDING_SOFT_LIMIT: "1", + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", + }; + const queue = new ExactReviewQueue({ storage }, env); + await queue.fetch(buildExactReviewQueueRequest("ordinary-existing", 760, "edited")); + + const existing = await queue.fetch( + buildExactReviewQueueRequest("existing-recovery", 760, "source_drift_requeue"), + ); + assert.equal(existing.status, 202); + assert.equal((await existing.json()).queued, true); + + for (const [index, sourceAction] of [ + "failed_review_shard_recovery", + "artifact_retention_recovery", + "source_drift_requeue", + ].entries()) { + const shed = await queue.fetch( + buildExactReviewQueueRequest(`shed-${index}`, 761 + index, sourceAction), + ); + assert.equal(shed.status, 202); + assert.deepEqual(await shed.json(), { ok: true, shed: true, reason: "backpressure" }); + } + + const webhook = await queue.fetch( + buildExactReviewQueueRequest("webhook-over-limit", 770, "opened"), + ); + assert.equal((await webhook.json()).queued, true); + const publication = await queue.fetch( + buildExactReviewQueueRequest( + "publication-over-limit", + 771, + "exact_review_artifact_publish", + "issue", + undefined, + exactReviewPublicationOverrides(771, "7710"), + ), + ); + assert.equal((await publication.json()).queued, true); + + const restarted = new ExactReviewQueue({ storage }, env); + const stats = await ( + await restarted.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.pending, 3); + assert.equal(stats.shed_since_reset, 3); + assert.equal(stats.handoff_health.pending_depth, 3); + assert.equal(stats.handoff_health.shed_since_reset, 3); + assert.equal(stats.lanes.review.pending_depth, 2); + assert.equal(stats.lanes.review.shed_since_reset, 3); +}); + test("heartbeated exact-review leases use the heartbeat grace while legacy leases keep execution expiry", () => { const now = 1_000_000; const item = { @@ -285,7 +424,7 @@ test("dashboard status reads the exact-review handoff model from the durable que available_slots: status.lanes.review.available_slots, capacity: status.lanes.review.capacity, }, - { pending: 2, ready: 1, backoff: 1, active: 1, available_slots: 63, capacity: 64 }, + { pending: 2, ready: 0, backoff: 2, active: 1, available_slots: 63, capacity: 64 }, ); assert.deepEqual( { @@ -1849,6 +1988,7 @@ test("exact-review queue coalesces deliveries, dispatches a bound rollout snapsh { CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", EXACT_REVIEW_QUEUE_MAX_CONCURRENT: "1", }, ); @@ -2895,6 +3035,7 @@ test("exact-review queue admits at most one active item per target repository", { CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", EXACT_REVIEW_QUEUE_MAX_CONCURRENT: "2", EXACT_REVIEW_TARGET_MAX_CONCURRENT: "1", }, @@ -2964,6 +3105,7 @@ test("exact-review queue can use the global capacity for one target", async () = { CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", EXACT_REVIEW_QUEUE_MAX_CONCURRENT: "4", EXACT_REVIEW_TARGET_MAX_CONCURRENT: "4", }, @@ -3020,6 +3162,7 @@ test("exact-review queue keeps publication artifacts durable outside review capa { CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", EXACT_REVIEW_QUEUE_MAX_CONCURRENT: "4", EXACT_REVIEW_TARGET_MAX_CONCURRENT: "4", }, @@ -3289,6 +3432,7 @@ test("exact-review queue wakes while target capacity remains", async () => { { CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", EXACT_REVIEW_QUEUE_MAX_CONCURRENT: "4", EXACT_REVIEW_TARGET_MAX_CONCURRENT: "2", }, @@ -3337,6 +3481,7 @@ test("exact-review queue defers retained backlog until a paused dispatcher retry { CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", EXACT_REVIEW_QUEUE_MAX_CONCURRENT: "4", EXACT_REVIEW_TARGET_MAX_CONCURRENT: "2", }, @@ -3527,6 +3672,7 @@ test("exact-review queue retries dispatch failures and reclaims an unclaimed lea { CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", }, ); assert.equal( @@ -3652,6 +3798,7 @@ test("exact-review queue preserves a claimed lease after an ambiguous dispatch f { CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", }, ); assert.equal( diff --git a/test/exact-review-health.test.ts b/test/exact-review-health.test.ts index 8889a87787..4001079a85 100644 --- a/test/exact-review-health.test.ts +++ b/test/exact-review-health.test.ts @@ -35,6 +35,7 @@ test("exact-review handoff health reports an empty queue as idle", () => { test("exact-review handoff health exposes phase counts, ages, and available capacity", () => { const health = summarize({ capacity: 4, + shedSinceReset: 7, dispatcher: { state: "active" }, items: [ { state: "pending", createdAt: NOW - 90_000, updatedAt: NOW - 90_000 }, @@ -57,6 +58,8 @@ test("exact-review handoff health exposes phase counts, ages, and available capa assert.equal(health.reason, "handoff_current"); assert.equal(health.active, 2); assert.equal(health.available_slots, 2); + assert.equal(health.pending_depth, 1); + assert.equal(health.shed_since_reset, 7); assert.deepEqual(health.phases.pending, { count: 1, oldest_at: "2026-07-13T01:58:30.000Z", diff --git a/test/sweep-workflow.test.ts b/test/sweep-workflow.test.ts index 3e6539bd6f..da430027cd 100644 --- a/test/sweep-workflow.test.ts +++ b/test/sweep-workflow.test.ts @@ -490,7 +490,8 @@ test("exact event review hands immutable artifacts to the queue-bounded publishe assert.match(drift.run ?? "", /x-clawsweeper-exact-review-signature/); assert.match(drift.run ?? "", /internal\/exact-review\/enqueue/); assert.match(drift.run ?? "", /decision\.sourceAction === "failed_review_shard_recovery"/); - assert.match(drift.run ?? "", /\.queued == true or \.deduped == true/); + assert.match(drift.run ?? "", /\.queued == true or \.deduped == true or \.shed == true/); + assert.match(drift.run ?? "", /Source-drift recovery shed by exact-review queue backpressure/); const status = step(publisher, "Mark re-review complete"); assert.equal(status.env?.LIVE_TERMINAL_MISSING, undefined); assert.equal(status.env?.LIVE_GUARDED_OPEN, undefined); @@ -2052,8 +2053,9 @@ test("failed review recovery waits for durable exact-review queue acknowledgemen ); assert.match( recoveryBlock, - /\.ok == true and \(\.queued == true or \.deduped == true or \.accepted == false\)/, + /\.ok == true and \(\.queued == true or \.deduped == true or \.shed == true or \.accepted == false\)/, ); + assert.match(recoveryBlock, /Recovery shed by exact-review queue backpressure/); assert.doesNotMatch(recoveryBlock, /workflow run sweep\.yml/); assert.doesNotMatch(recoveryBlock, /repos\/\$GITHUB_REPOSITORY\/dispatches/); assert.match(recoveryBlock, /for attempt in 1 2 3/);