Promote #1895 to main (catchup backpressure retry) — release-bound for v10.0.9 #1896
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| /** | ||
| * User-requested catch-up must outrank autonomous exact-VM repair (priority | ||
| * 1_000) and ordinary background sync (priority 0). This lets a subscribe or | ||
| * explicit catch-up displace queued background work instead of being marked | ||
| * deferred before it has fetched a byte. | ||
| */ | ||
| export const FOREGROUND_CATCHUP_SYNC_PRIORITY = 2_000; | ||
|
|
||
| /** | ||
| * Admission can still race another foreground catch-up. Retry that local-only | ||
| * outcome briefly; transport, authorization, timeout, and integrity failures | ||
| * are deliberately not retried here. | ||
| */ | ||
| export const CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS = [100, 250, 500] as const; | ||
|
|
||
| export interface CatchupBackpressureResult { | ||
| deferredBackpressure?: number; | ||
| } | ||
|
|
||
| export async function retryCatchupPlaneOnBackpressure<T extends CatchupBackpressureResult>( | ||
| run: () => Promise<T>, | ||
| options?: { | ||
| delaysMs?: readonly number[]; | ||
| wait?: (delayMs: number) => Promise<void>; | ||
| }, | ||
| ): Promise<T> { | ||
| const delaysMs = options?.delaysMs ?? CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS; | ||
| const wait = options?.wait ?? ((delayMs: number) => new Promise<void>((resolve) => { | ||
| setTimeout(resolve, delayMs); | ||
| })); | ||
|
|
||
| let result = await run(); | ||
| for (const delayMs of delaysMs) { | ||
| if ((result.deferredBackpressure ?? 0) === 0) break; | ||
| await wait(delayMs); | ||
| result = await run(); | ||
| } | ||
| return result; | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| import { describe, expect, it, vi } from 'vitest'; | ||
| import { | ||
| CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS, | ||
| retryCatchupPlaneOnBackpressure, | ||
| } from '../src/sync/catchup-backpressure-retry.js'; | ||
|
|
||
| describe('retryCatchupPlaneOnBackpressure', () => { | ||
| it('retries only the local scheduler deferral result', async () => { | ||
| const run = vi.fn() | ||
| .mockResolvedValueOnce({ deferredBackpressure: 1, marker: 'deferred' }) | ||
| .mockResolvedValueOnce({ deferredBackpressure: 0, marker: 'complete' }); | ||
| const waits: number[] = []; | ||
|
|
||
| const result = await retryCatchupPlaneOnBackpressure(run, { | ||
| delaysMs: [3, 5], | ||
| wait: async (delayMs) => { waits.push(delayMs); }, | ||
| }); | ||
|
|
||
| expect(result).toEqual({ deferredBackpressure: 0, marker: 'complete' }); | ||
| expect(run).toHaveBeenCalledTimes(2); | ||
| expect(waits).toEqual([3]); | ||
| }); | ||
|
|
||
| it('returns the final deferred result after the bounded retry budget', async () => { | ||
| const run = vi.fn(async () => ({ deferredBackpressure: 1 })); | ||
|
|
||
| const result = await retryCatchupPlaneOnBackpressure(run, { | ||
| wait: async () => {}, | ||
| }); | ||
|
|
||
| expect(result.deferredBackpressure).toBe(1); | ||
| expect(run).toHaveBeenCalledTimes(CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS.length + 1); | ||
| }); | ||
|
|
||
| it('does not retry a clean result', async () => { | ||
| const run = vi.fn(async () => ({ deferredBackpressure: 0 })); | ||
|
|
||
| await retryCatchupPlaneOnBackpressure(run, { | ||
| wait: async () => { throw new Error('must not wait'); }, | ||
| }); | ||
|
|
||
| expect(run).toHaveBeenCalledTimes(1); | ||
| }); | ||
| }); |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟡 Issue: Model foreground catch-up as one admission policy, not loose flags
What's wrong
The foreground catch-up concept leaks through several layers as an optional number plus an optional boolean. Those options only make sense as a coordinated policy, but the API lets callers assemble mismatched combinations and forces readers to trace multiple flags through durable, changelog, shared-memory, and CLI worker boundaries.
Example
A new explicit catch-up caller can pass
{ priority: FOREGROUND_CATCHUP_SYNC_PRIORITY }withoutretryDeferredBackpressure, or retry deferred work without foreground priority. Both combinations are representable even though the comments describe one foreground admission mode.Suggested direction
Replace the naked
priorityplusretryDeferredBackpressureplumbing with a named mode or typedSyncAdmissionPolicythat carries the intended priority, retry behavior, and single-flight identity together.For Agents
Review
DurableSyncOptions,syncSharedMemoryFromPeerDetailedoptions,syncContextGraphFromConnectedPeersoptions, and the CLI runner calls. Preserve the current foreground priority value and bounded retry behavior, but represent them as one named admission policy or mode and derive the single-flight key fields from that policy.