From 6ea23406d5e14ffa6d4e9fbbd652237a0885d9e4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E8=B4=B5=E8=90=8D0668001030?= Date: Mon, 20 Jul 2026 20:49:57 +0800 Subject: [PATCH] fix: prevent stale review results from delaying publication --- dashboard/exact-review-queue.ts | 131 ++++++++++++++++++-- test/dashboard-worker.test.ts | 209 +++++++++++++++++++++++++++++++- 2 files changed, 323 insertions(+), 17 deletions(-) diff --git a/dashboard/exact-review-queue.ts b/dashboard/exact-review-queue.ts index f555bda7c9..84956bee07 100644 --- a/dashboard/exact-review-queue.ts +++ b/dashboard/exact-review-queue.ts @@ -362,6 +362,22 @@ export class ExactReviewQueue { exactReviewHeartbeatGraceMs(this.env), ); const key = exactReviewItemKey(decision); + const publicationCoalescing = coalesceExactReviewPublications(state, decision); + if (publicationCoalescing.incomingStale) { + this.writeStateSync(state); + if (publicationCoalescing.superseded) { + this.incrementQueueMetricsSync({ + publicationCompleted: publicationCoalescing.superseded, + publicationSuperseded: publicationCoalescing.superseded, + }); + } + return { + deduped: true as const, + key: publicationCoalescing.freshestKey, + state, + stateChanged: publicationCoalescing.superseded > 0, + }; + } const current = state.items[key]; if (current) { const ignoredRecovery = @@ -421,10 +437,20 @@ export class ExactReviewQueue { }; } this.writeStateSync(state); + if (publicationCoalescing.superseded) { + this.incrementQueueMetricsSync({ + publicationCompleted: publicationCoalescing.superseded, + publicationSuperseded: publicationCoalescing.superseded, + }); + } return { deduped: false as const, key, state }; }); if (accepted.deduped) { - return json({ ok: true, deduped: true, item_key: exactReviewItemKey(decision) }, 202); + if (accepted.stateChanged) await this.scheduleNext(accepted.state, now); + return json( + { ok: true, deduped: true, item_key: accepted.key || exactReviewItemKey(decision) }, + 202, + ); } if (accepted.shed) { return json({ ok: true, shed: true, reason: "backpressure" }, 202); @@ -2032,17 +2058,9 @@ export class ExactReviewQueue { const state = this.readStateSync(); let superseded = 0; for (const candidate of candidates) { - const item = state.items[candidate.itemKey]; - if ( - !item || - item.revision !== candidate.revision || - !exactReviewQueueIsPublication(item) || - (item.state !== "pending" && item.state !== "parked") - ) { - continue; + if (supersedeExactReviewPublicationItem(state, candidate.itemKey, candidate.revision)) { + superseded += 1; } - delete state.items[item.key]; - superseded += 1; } if (superseded) { await this.writeState(state, { @@ -3401,6 +3419,97 @@ function exactReviewItemKey(decision: ExactReviewDecision) { : base; } +function coalesceExactReviewPublications( + state: ExactReviewQueueState, + incoming: ExactReviewDecision, +) { + const incomingPublication = incoming.publication; + const incomingKey = exactReviewItemKey(incoming); + if (!incomingPublication) { + return { incomingStale: false, freshestKey: incomingKey, superseded: 0 }; + } + + let freshestKey = incomingKey; + let freshestPublication = incomingPublication; + const matchingItems: ExactReviewQueueItem[] = []; + for (const item of Object.values(state.items)) { + const publication = item.decision.publication; + if (!publication || publication.itemKey !== incomingPublication.itemKey) continue; + matchingItems.push(item); + if (compareExactReviewPublicationFreshness(publication, freshestPublication) > 0) { + freshestKey = item.key; + freshestPublication = publication; + } + } + + let superseded = 0; + for (const item of matchingItems) { + const publication = item.decision.publication; + if ( + publication && + compareExactReviewPublicationFreshness(publication, freshestPublication) < 0 && + supersedeExactReviewPublicationItem(state, item.key) + ) { + superseded += 1; + } + } + return { + incomingStale: + compareExactReviewPublicationFreshness(incomingPublication, freshestPublication) < 0, + freshestKey, + superseded, + }; +} + +function compareExactReviewPublicationFreshness( + left: ExactReviewPublication, + right: ExactReviewPublication, +) { + // Review revisions are monotonic for one item. Within a revision, GitHub's + // run tuple orders redispatches; claim generation is only a final tie-breaker + // because clearing an expired lease resets it for the next dispatch. + if (left.protocolVersion === 2 && right.protocolVersion === 2) { + const revision = Number(left.leaseRevision) - Number(right.leaseRevision); + if (revision) return revision; + } + const run = compareUnsignedDecimalStrings(left.producerRunId, right.producerRunId); + if (run) return run; + const attempt = left.producerRunAttempt - right.producerRunAttempt; + if (attempt) return attempt; + if (left.protocolVersion === 2 && right.protocolVersion === 2) { + const generation = Number(left.claimGeneration) - Number(right.claimGeneration); + if (generation) return generation; + } + return left.protocolVersion - right.protocolVersion; +} + +function compareUnsignedDecimalStrings(left: string, right: string) { + const normalizedLeft = left.replace(/^0+(?=\d)/, ""); + const normalizedRight = right.replace(/^0+(?=\d)/, ""); + if (normalizedLeft.length !== normalizedRight.length) { + return normalizedLeft.length - normalizedRight.length; + } + return normalizedLeft === normalizedRight ? 0 : normalizedLeft < normalizedRight ? -1 : 1; +} + +function supersedeExactReviewPublicationItem( + state: ExactReviewQueueState, + itemKey: string, + revision?: number, +) { + const item = state.items[itemKey]; + if ( + !item || + (revision !== undefined && item.revision !== revision) || + !exactReviewQueueIsPublication(item) || + (item.state !== "pending" && item.state !== "parked") + ) { + return false; + } + delete state.items[item.key]; + return true; +} + function isExactReviewQueueTargetEnabled(decision: ExactReviewDecision, env) { return ( decision.targetRepo !== "openclaw/clawhub" || diff --git a/test/dashboard-worker.test.ts b/test/dashboard-worker.test.ts index 9f2f449d41..f41d174be2 100644 --- a/test/dashboard-worker.test.ts +++ b/test/dashboard-worker.test.ts @@ -266,6 +266,154 @@ test("exact-review queue bypasses debounce for commands and publications", async } }); +test("exact-review queue keeps only the newest pending or parked publication per item", async () => { + const storage = new MemoryDurableStorage(); + const queue = new ExactReviewQueue({ storage }, {}); + const targetRepo = "openclaw/openclaw"; + + await enqueueExactReviewPublication(queue, { + deliveryId: "publication-pending-old", + itemNumber: 753, + producerRunId: "7530", + targetRepo, + }); + await enqueueExactReviewPublication(queue, { + deliveryId: "publication-parked-old", + itemNumber: 754, + producerRunId: "7540", + targetRepo, + }); + await enqueueExactReviewPublication(queue, { + deliveryId: "publication-unrelated", + itemNumber: 758, + producerRunId: "7580", + targetRepo, + }); + const parked = (await storage.get("exact-review-queue")) as { + items: Record; + }; + parked.items["openclaw/openclaw#754@publish:7540:1"].state = "parked"; + parked.items["openclaw/openclaw#754@publish:7540:1"].parkedReason = "dead_letter_capacity"; + await storage.put("exact-review-queue", parked); + + for (const [itemNumber, runId] of [ + [753, "7531"], + [754, "7541"], + ] as const) { + const response = await enqueueExactReviewPublication(queue, { + deliveryId: `publication-new-${itemNumber}`, + itemNumber, + producerRunId: runId, + producerSourceAction: "synchronize", + targetRepo, + leaseRevision: 2, + }); + assert.equal(response.status, 202); + assert.equal((await response.json()).queued, true); + } + + const state = (await storage.get("exact-review-queue")) as { + items: Record; + }; + assert.deepEqual(Object.keys(state.items).sort(), [ + "openclaw/openclaw#753@publish:7531:1", + "openclaw/openclaw#754@publish:7541:1", + "openclaw/openclaw#758@publish:7580:1", + ]); + assert.equal( + state.items["openclaw/openclaw#753@publish:7531:1"].decision.publication.leaseRevision, + 2, + ); + assert.equal( + state.items["openclaw/openclaw#754@publish:7541:1"].decision.publication.leaseRevision, + 2, + ); + + const stats = await ( + await queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.lanes.publication.enqueued_total, 5); + assert.equal(stats.lanes.publication.completed_total, 2); + assert.equal(stats.lanes.publication.superseded_total, 2); +}); + +test("exact-review queue rejects a delayed stale publication without displacing newer work", async () => { + const storage = new MemoryDurableStorage(); + const queue = new ExactReviewQueue({ storage }, {}); + const targetRepo = "openclaw/openclaw"; + const newer = await enqueueExactReviewPublication(queue, { + deliveryId: "publication-new-first", + itemNumber: 755, + producerRunId: "7551", + producerSourceAction: "synchronize", + targetRepo, + }); + assert.equal((await newer.json()).queued, true); + const delayed = await enqueueExactReviewPublication(queue, { + deliveryId: "publication-old-delayed", + itemNumber: 755, + producerRunId: "7550", + producerSourceAction: "synchronize", + targetRepo, + claimGeneration: 2, + }); + assert.equal(delayed.status, 202); + assert.deepEqual(await delayed.json(), { + ok: true, + deduped: true, + item_key: "openclaw/openclaw#755@publish:7551:1", + }); + + const state = (await storage.get("exact-review-queue")) as { + items: Record; + }; + assert.deepEqual(Object.keys(state.items), ["openclaw/openclaw#755@publish:7551:1"]); + const stats = await ( + await queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.lanes.publication.enqueued_total, 1); + assert.equal(stats.lanes.publication.completed_total, 0); + assert.equal(stats.lanes.publication.superseded_total, 0); +}); + +test("exact-review queue does not delete active publications while coalescing newer results", async () => { + const storage = new MemoryDurableStorage(); + const leased = leasedExactReviewPublicationItem(756, "7560"); + const dispatching = leasedExactReviewPublicationItem(757, "7570"); + dispatching.state = "dispatching"; + dispatching.claimedRunId = undefined; + dispatching.claimedRunAttempt = undefined; + dispatching.claimGeneration = undefined; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { [leased.key]: leased, [dispatching.key]: dispatching }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + + for (const [itemNumber, runId] of [ + [756, "7561"], + [757, "7571"], + ] as const) { + await enqueueExactReviewPublication(queue, { + deliveryId: `publication-active-new-${itemNumber}`, + itemNumber, + producerRunId: runId, + producerSourceAction: "edited", + targetRepo: "openclaw/openclaw", + itemKind: "issue", + leaseRevision: 2, + }); + } + + const state = (await storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(state.items[leased.key].state, "leased"); + assert.equal(state.items[dispatching.key].state, "dispatching"); + assert.equal(state.items["openclaw/openclaw#756@publish:7561:1"].state, "pending"); + assert.equal(state.items["openclaw/openclaw#757@publish:7571:1"].state, "pending"); +}); + test("exact-review queue sheds only new recovery work above the pending soft limit", async () => { const storage = new MemoryDurableStorage(); const env = { @@ -11245,13 +11393,23 @@ function exactReviewPublicationOverrides( itemNumber: number, producerRunId: string, producerSourceAction = "opened", + options: { + targetRepo?: string; + itemKind?: "issue" | "pull_request"; + leaseRevision?: number | null; + claimGeneration?: number | null; + } = {}, ) { + const targetRepo = options.targetRepo || "openclaw/gogcli"; + const itemKind = options.itemKind || "issue"; + const leaseRevision = options.leaseRevision === undefined ? 1 : options.leaseRevision; + const claimGeneration = options.claimGeneration === undefined ? 1 : options.claimGeneration; const producerDecision = { - targetRepo: "openclaw/gogcli", + targetRepo, targetBranch: "main", itemNumber, - itemKind: "issue", - sourceEvent: "issues", + itemKind, + sourceEvent: itemKind === "issue" ? "issues" : "pull_request", sourceAction: producerSourceAction, supersedesInProgress: false, }; @@ -11261,10 +11419,10 @@ function exactReviewPublicationOverrides( producerRunId, producerRunAttempt: 1, sourceSha: "a".repeat(40), - itemKey: `openclaw/gogcli#${itemNumber}`, + itemKey: `${targetRepo}#${itemNumber}`, protocolVersion: 2, - leaseRevision: 1, - claimGeneration: 1, + leaseRevision, + claimGeneration, liveProceeded: true, liveTerminalNoop: false, liveTerminalMissing: false, @@ -11274,6 +11432,45 @@ function exactReviewPublicationOverrides( }; } +function enqueueExactReviewPublication( + queue: ExactReviewQueue, + { + deliveryId, + itemNumber, + producerRunId, + producerSourceAction = "opened", + targetRepo = "openclaw/openclaw", + itemKind = "pull_request", + leaseRevision = 1, + claimGeneration = 1, + }: { + deliveryId: string; + itemNumber: number; + producerRunId: string; + producerSourceAction?: string; + targetRepo?: string; + itemKind?: "issue" | "pull_request"; + leaseRevision?: number; + claimGeneration?: number; + }, +) { + return queue.fetch( + buildExactReviewQueueRequest( + deliveryId, + itemNumber, + "exact_review_artifact_publish", + itemKind, + targetRepo, + exactReviewPublicationOverrides(itemNumber, producerRunId, producerSourceAction, { + targetRepo, + itemKind, + leaseRevision, + claimGeneration, + }), + ), + ); +} + function leasedExactReviewQueueItem(itemNumber: number, runId: string, runAttempt = 1) { const now = Date.now(); const decision = {