authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-03 22:05:33-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-03 22:05:33-07:00
log6de4b47ecb5fc1cee11a3f3ef1b8279c35764cf6
treefd2e62eec0d0336b308ff14df8931b484a2ec573
parent27c2b145f8d214b5c578eefc463278d0916fb048
signature Signed by PGP key B5690EEEBB952194

Return replay retry state from orchestration recovery (#1728)

3 files changed, 285 insertions(+), 12 deletions(-)

apps/web/src/orchestrationRecovery.test.ts+177-4
...@@ -1,6 +1,9 @@...@@ -1,6 +1,9 @@
1import { describe, expect, it } from "vitest";1import { describe, expect, it } from "vitest";
22
3import { createOrchestrationRecoveryCoordinator } from "./orchestrationRecovery";3import {
4 createOrchestrationRecoveryCoordinator,
5 deriveReplayRetryDecision,
6} from "./orchestrationRecovery";
47
5describe("createOrchestrationRecoveryCoordinator", () => {8describe("createOrchestrationRecoveryCoordinator", () => {
6 it("defers live events until bootstrap completes and then requests replay", () => {9 it("defers live events until bootstrap completes and then requests replay", () => {
...@@ -59,10 +62,13 @@ describe("createOrchestrationRecoveryCoordinator", () => {...@@ -59,10 +62,13 @@ describe("createOrchestrationRecoveryCoordinator", () => {
59 coordinator.classifyDomainEvent(7);62 coordinator.classifyDomainEvent(7);
60 coordinator.markEventBatchApplied([{ sequence: 4 }, { sequence: 5 }, { sequence: 6 }]);63 coordinator.markEventBatchApplied([{ sequence: 4 }, { sequence: 5 }, { sequence: 6 }]);
6164
62 expect(coordinator.completeReplayRecovery()).toBe(true);65 expect(coordinator.completeReplayRecovery()).toEqual({
66 replayMadeProgress: true,
67 shouldReplay: true,
68 });
63 });69 });
6470
65 it("does not immediately replay again when replay returns no new events", () => {71 it("retries replay when no progress was made but higher live sequences were observed", () => {
66 const coordinator = createOrchestrationRecoveryCoordinator();72 const coordinator = createOrchestrationRecoveryCoordinator();
6773
68 coordinator.beginSnapshotRecovery("bootstrap");74 coordinator.beginSnapshotRecovery("bootstrap");
...@@ -70,7 +76,10 @@ describe("createOrchestrationRecoveryCoordinator", () => {...@@ -70,7 +76,10 @@ describe("createOrchestrationRecoveryCoordinator", () => {
70 coordinator.classifyDomainEvent(5);76 coordinator.classifyDomainEvent(5);
71 coordinator.beginReplayRecovery("sequence-gap");77 coordinator.beginReplayRecovery("sequence-gap");
7278
73 expect(coordinator.completeReplayRecovery()).toBe(false);79 expect(coordinator.completeReplayRecovery()).toEqual({
80 replayMadeProgress: false,
81 shouldReplay: true,
82 });
74 expect(coordinator.getState()).toMatchObject({83 expect(coordinator.getState()).toMatchObject({
75 latestSequence: 3,84 latestSequence: 3,
76 highestObservedSequence: 5,85 highestObservedSequence: 5,
...@@ -79,6 +88,19 @@ describe("createOrchestrationRecoveryCoordinator", () => {...@@ -79,6 +88,19 @@ describe("createOrchestrationRecoveryCoordinator", () => {
79 });88 });
80 });89 });
8190
91 it("does not request another replay when a replay made no progress and nothing newer was observed", () => {
92 const coordinator = createOrchestrationRecoveryCoordinator();
93
94 coordinator.beginSnapshotRecovery("bootstrap");
95 coordinator.completeSnapshotRecovery(3);
96 coordinator.beginReplayRecovery("sequence-gap");
97
98 expect(coordinator.completeReplayRecovery()).toEqual({
99 replayMadeProgress: false,
100 shouldReplay: false,
101 });
102 });
103
82 it("marks replay failure as unbootstrapped so snapshot fallback is recovery-only", () => {104 it("marks replay failure as unbootstrapped so snapshot fallback is recovery-only", () => {
83 const coordinator = createOrchestrationRecoveryCoordinator();105 const coordinator = createOrchestrationRecoveryCoordinator();
84106
...@@ -131,3 +153,154 @@ describe("createOrchestrationRecoveryCoordinator", () => {...@@ -131,3 +153,154 @@ describe("createOrchestrationRecoveryCoordinator", () => {
131 });153 });
132 });154 });
133});155});
156
157describe("deriveReplayRetryDecision", () => {
158 it("retries immediately when replay made progress", () => {
159 expect(
160 deriveReplayRetryDecision({
161 previousTracker: {
162 attempts: 2,
163 latestSequence: 3,
164 highestObservedSequence: 5,
165 },
166 completion: {
167 replayMadeProgress: true,
168 shouldReplay: true,
169 },
170 recoveryState: {
171 latestSequence: 5,
172 highestObservedSequence: 5,
173 },
174 baseDelayMs: 100,
175 maxNoProgressRetries: 3,
176 }),
177 ).toEqual({
178 shouldRetry: true,
179 delayMs: 0,
180 tracker: null,
181 });
182 });
183
184 it("caps no-progress retries for the same frontier", () => {
185 const first = deriveReplayRetryDecision({
186 previousTracker: null,
187 completion: {
188 replayMadeProgress: false,
189 shouldReplay: true,
190 },
191 recoveryState: {
192 latestSequence: 3,
193 highestObservedSequence: 5,
194 },
195 baseDelayMs: 100,
196 maxNoProgressRetries: 3,
197 });
198
199 const second = deriveReplayRetryDecision({
200 previousTracker: first.tracker,
201 completion: {
202 replayMadeProgress: false,
203 shouldReplay: true,
204 },
205 recoveryState: {
206 latestSequence: 3,
207 highestObservedSequence: 5,
208 },
209 baseDelayMs: 100,
210 maxNoProgressRetries: 3,
211 });
212
213 const third = deriveReplayRetryDecision({
214 previousTracker: second.tracker,
215 completion: {
216 replayMadeProgress: false,
217 shouldReplay: true,
218 },
219 recoveryState: {
220 latestSequence: 3,
221 highestObservedSequence: 5,
222 },
223 baseDelayMs: 100,
224 maxNoProgressRetries: 3,
225 });
226
227 const fourth = deriveReplayRetryDecision({
228 previousTracker: third.tracker,
229 completion: {
230 replayMadeProgress: false,
231 shouldReplay: true,
232 },
233 recoveryState: {
234 latestSequence: 3,
235 highestObservedSequence: 5,
236 },
237 baseDelayMs: 100,
238 maxNoProgressRetries: 3,
239 });
240
241 expect(first).toEqual({
242 shouldRetry: true,
243 delayMs: 100,
244 tracker: {
245 attempts: 1,
246 latestSequence: 3,
247 highestObservedSequence: 5,
248 },
249 });
250 expect(second).toEqual({
251 shouldRetry: true,
252 delayMs: 200,
253 tracker: {
254 attempts: 2,
255 latestSequence: 3,
256 highestObservedSequence: 5,
257 },
258 });
259 expect(third).toEqual({
260 shouldRetry: true,
261 delayMs: 400,
262 tracker: {
263 attempts: 3,
264 latestSequence: 3,
265 highestObservedSequence: 5,
266 },
267 });
268 expect(fourth).toEqual({
269 shouldRetry: false,
270 delayMs: 0,
271 tracker: null,
272 });
273 });
274
275 it("resets the retry budget when the replay frontier changes", () => {
276 const exhausted = {
277 attempts: 3,
278 latestSequence: 3,
279 highestObservedSequence: 5,
280 };
281
282 expect(
283 deriveReplayRetryDecision({
284 previousTracker: exhausted,
285 completion: {
286 replayMadeProgress: false,
287 shouldReplay: true,
288 },
289 recoveryState: {
290 latestSequence: 3,
291 highestObservedSequence: 6,
292 },
293 baseDelayMs: 100,
294 maxNoProgressRetries: 3,
295 }),
296 ).toEqual({
297 shouldRetry: true,
298 delayMs: 100,
299 tracker: {
300 attempts: 1,
301 latestSequence: 3,
302 highestObservedSequence: 6,
303 },
304 });
305 });
306});
apps/web/src/orchestrationRecovery.ts+72-6
...@@ -13,8 +13,74 @@ export interface OrchestrationRecoveryState {...@@ -13,8 +13,74 @@ export interface OrchestrationRecoveryState {
13 inFlight: OrchestrationRecoveryPhase | null;13 inFlight: OrchestrationRecoveryPhase | null;
14}14}
1515
16export interface ReplayRecoveryCompletion {
17 replayMadeProgress: boolean;
18 shouldReplay: boolean;
19}
20
21export interface ReplayRetryTracker {
22 attempts: number;
23 latestSequence: number;
24 highestObservedSequence: number;
25}
26
27export interface ReplayRetryDecision {
28 shouldRetry: boolean;
29 delayMs: number;
30 tracker: ReplayRetryTracker | null;
31}
32
16type SequencedEvent = Readonly<{ sequence: number }>;33type SequencedEvent = Readonly<{ sequence: number }>;
1734
35export function deriveReplayRetryDecision(input: {
36 previousTracker: ReplayRetryTracker | null;
37 completion: ReplayRecoveryCompletion;
38 recoveryState: Pick<OrchestrationRecoveryState, "latestSequence" | "highestObservedSequence">;
39 baseDelayMs: number;
40 maxNoProgressRetries: number;
41}): ReplayRetryDecision {
42 if (!input.completion.shouldReplay) {
43 return {
44 shouldRetry: false,
45 delayMs: 0,
46 tracker: null,
47 };
48 }
49
50 if (input.completion.replayMadeProgress) {
51 return {
52 shouldRetry: true,
53 delayMs: 0,
54 tracker: null,
55 };
56 }
57
58 const previousTracker = input.previousTracker;
59 const sameFrontier =
60 previousTracker !== null &&
61 previousTracker.latestSequence === input.recoveryState.latestSequence &&
62 previousTracker.highestObservedSequence === input.recoveryState.highestObservedSequence;
63
64 const attempts = sameFrontier && previousTracker !== null ? previousTracker.attempts + 1 : 1;
65 if (attempts > input.maxNoProgressRetries) {
66 return {
67 shouldRetry: false,
68 delayMs: 0,
69 tracker: null,
70 };
71 }
72
73 return {
74 shouldRetry: true,
75 delayMs: input.baseDelayMs * 2 ** (attempts - 1),
76 tracker: {
77 attempts,
78 latestSequence: input.recoveryState.latestSequence,
79 highestObservedSequence: input.recoveryState.highestObservedSequence,
80 },
81 };
82}
83
18export function createOrchestrationRecoveryCoordinator() {84export function createOrchestrationRecoveryCoordinator() {
19 let state: OrchestrationRecoveryState = {85 let state: OrchestrationRecoveryState = {
20 latestSequence: 0,86 latestSequence: 0,
...@@ -120,16 +186,16 @@ export function createOrchestrationRecoveryCoordinator() {...@@ -120,16 +186,16 @@ export function createOrchestrationRecoveryCoordinator() {
120 return true;186 return true;
121 },187 },
122188
123 completeReplayRecovery(): boolean {189 completeReplayRecovery(): ReplayRecoveryCompletion {
124 const replayMadeProgress =190 const replayMadeProgress =
125 replayStartSequence !== null && state.latestSequence > replayStartSequence;191 replayStartSequence !== null && state.latestSequence > replayStartSequence;
126 replayStartSequence = null;192 replayStartSequence = null;
127 state.inFlight = null;193 state.inFlight = null;
128 if (!replayMadeProgress) {194 const replayResolution = resolveReplayNeedAfterRecovery();
129 state.pendingReplay = false;195 return {
130 return false;196 replayMadeProgress,
131 }197 shouldReplay: replayResolution.shouldReplay,
132 return resolveReplayNeedAfterRecovery().shouldReplay;198 };
133 },199 },
134200
135 failReplayRecovery(): void {201 failReplayRecovery(): void {
apps/web/src/routes/__root.tsx+36-2
...@@ -42,6 +42,7 @@ import { projectQueryKeys } from "../lib/projectReactQuery";...@@ -42,6 +42,7 @@ import { projectQueryKeys } from "../lib/projectReactQuery";
42import { collectActiveTerminalThreadIds } from "../lib/terminalStateCleanup";42import { collectActiveTerminalThreadIds } from "../lib/terminalStateCleanup";
43import { deriveOrchestrationBatchEffects } from "../orchestrationEventEffects";43import { deriveOrchestrationBatchEffects } from "../orchestrationEventEffects";
44import { createOrchestrationRecoveryCoordinator } from "../orchestrationRecovery";44import { createOrchestrationRecoveryCoordinator } from "../orchestrationRecovery";
45import { deriveReplayRetryDecision } from "../orchestrationRecovery";
45import { getWsRpcClient } from "~/wsRpcClient";46import { getWsRpcClient } from "~/wsRpcClient";
4647
47export const Route = createRootRouteWithContext<{48export const Route = createRootRouteWithContext<{
...@@ -188,6 +189,9 @@ function coalesceOrchestrationUiEvents(...@@ -188,6 +189,9 @@ function coalesceOrchestrationUiEvents(
188 return coalesced;189 return coalesced;
189}190}
190191
192const REPLAY_RECOVERY_RETRY_DELAY_MS = 100;
193const MAX_NO_PROGRESS_REPLAY_RETRIES = 3;
194
191function ServerStateBootstrap() {195function ServerStateBootstrap() {
192 useEffect(() => startServerStateSync(getWsRpcClient().server), []);196 useEffect(() => startServerStateSync(getWsRpcClient().server), []);
193197
...@@ -309,6 +313,7 @@ function EventRouter() {...@@ -309,6 +313,7 @@ function EventRouter() {
309 let disposed = false;313 let disposed = false;
310 disposedRef.current = false;314 disposedRef.current = false;
311 const recovery = createOrchestrationRecoveryCoordinator();315 const recovery = createOrchestrationRecoveryCoordinator();
316 let replayRetryTracker: import("../orchestrationRecovery").ReplayRetryTracker | null = null;
312 let needsProviderInvalidation = false;317 let needsProviderInvalidation = false;
313 const pendingDomainEvents: OrchestrationEvent[] = [];318 const pendingDomainEvents: OrchestrationEvent[] = [];
314 let flushPendingDomainEventsScheduled = false;319 let flushPendingDomainEventsScheduled = false;
...@@ -435,13 +440,42 @@ function EventRouter() {...@@ -435,13 +440,42 @@ function EventRouter() {
435 applyEventBatch(events);440 applyEventBatch(events);
436 }441 }
437 } catch {442 } catch {
443 replayRetryTracker = null;
438 recovery.failReplayRecovery();444 recovery.failReplayRecovery();
439 void fallbackToSnapshotRecovery();445 void fallbackToSnapshotRecovery();
440 return;446 return;
441 }447 }
442448
443 if (!disposed && recovery.completeReplayRecovery()) {449 if (!disposed) {
444 void recoverFromSequenceGap();450 const replayCompletion = recovery.completeReplayRecovery();
451 const retryDecision = deriveReplayRetryDecision({
452 previousTracker: replayRetryTracker,
453 completion: replayCompletion,
454 recoveryState: recovery.getState(),
455 baseDelayMs: REPLAY_RECOVERY_RETRY_DELAY_MS,
456 maxNoProgressRetries: MAX_NO_PROGRESS_REPLAY_RETRIES,
457 });
458 replayRetryTracker = retryDecision.tracker;
459
460 if (retryDecision.shouldRetry) {
461 if (retryDecision.delayMs > 0) {
462 await new Promise<void>((resolve) => {
463 setTimeout(resolve, retryDecision.delayMs);
464 });
465 if (disposed) {
466 return;
467 }
468 }
469 void recoverFromSequenceGap();
470 } else if (replayCompletion.shouldReplay && import.meta.env.MODE !== "test") {
471 console.warn(
472 "[orchestration-recovery]",
473 "Stopping replay recovery after no-progress retries.",
474 {
475 state: recovery.getState(),
476 },
477 );
478 }
445 }479 }
446 };480 };
447481