feat: cutover-409 save retry (load-path-rework 0002, 3a)
saveItemBody retries ONCE after a SyncRoomMovedError: invalidates the cached batch-resolved descriptor (onStackMoved clears batchedStacks), closes the stale stack, re-resolves, and retries the write against the room the fresh descriptor names. A persistent refusal fails after the single retry. Bumps pcbjam-shared for the registry wire types + typed error + mux replay. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VLSht9cadprtT2mhynawWu
This commit is contained in:
parent
89b4a7b61c
commit
451833666e
3 changed files with 108 additions and 15 deletions
|
|
@ -1 +1 @@
|
||||||
Subproject commit c83c27be6cf1dcdc3554b9426fc4af06bbc9889f
|
Subproject commit 1b191e69daf37160bb00aae3f607aaeafb9be220
|
||||||
|
|
@ -3,6 +3,8 @@ import {
|
||||||
encodeBundle,
|
encodeBundle,
|
||||||
encodeFrames,
|
encodeFrames,
|
||||||
sha256Hex,
|
sha256Hex,
|
||||||
|
SYNC_ACTION_HEADER,
|
||||||
|
SYNC_ACTION_RELOAD,
|
||||||
type ServerMsg,
|
type ServerMsg,
|
||||||
type SyncManifest,
|
type SyncManifest,
|
||||||
} from "@pcbjam/shared";
|
} from "@pcbjam/shared";
|
||||||
|
|
@ -62,9 +64,15 @@ async function fakeServer(seed: Record<string, string>) {
|
||||||
arrayBuffer: async () => bytes.buffer,
|
arrayBuffer: async () => bytes.buffer,
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Cutover simulation: `movedOnce` refuses the NEXT write with the 409 +
|
||||||
|
// reload-action answer a room gives after it stopped being the namespace's
|
||||||
|
// writer; `resolves` counts sync-stack resolutions (the retry re-resolves).
|
||||||
|
const state = { movedOnce: false, resolves: 0 };
|
||||||
|
|
||||||
const fetchImpl = (async (input: unknown, init?: RequestInit) => {
|
const fetchImpl = (async (input: unknown, init?: RequestInit) => {
|
||||||
const url = String(input);
|
const url = String(input);
|
||||||
if (url.endsWith("/sync-stack")) {
|
if (url.endsWith("/sync-stack")) {
|
||||||
|
state.resolves += 1;
|
||||||
return json({
|
return json({
|
||||||
lib: { id: LIB_ID, name: "My Lib" },
|
lib: { id: LIB_ID, name: "My Lib" },
|
||||||
layers: [
|
layers: [
|
||||||
|
|
@ -85,6 +93,16 @@ async function fakeServer(seed: Record<string, string>) {
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
if (url.startsWith(`${ROOM}/body/`) && init?.method === "PUT") {
|
if (url.startsWith(`${ROOM}/body/`) && init?.method === "PUT") {
|
||||||
|
if (state.movedOnce) {
|
||||||
|
state.movedOnce = false;
|
||||||
|
return {
|
||||||
|
ok: false,
|
||||||
|
status: 409,
|
||||||
|
headers: {
|
||||||
|
get: (h: string) => (h === SYNC_ACTION_HEADER ? SYNC_ACTION_RELOAD : null),
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
const path = decodeURIComponent(url.slice(`${ROOM}/body/`.length));
|
const path = decodeURIComponent(url.slice(`${ROOM}/body/`.length));
|
||||||
const body = new Uint8Array(init.body as ArrayBuffer | Uint8Array);
|
const body = new Uint8Array(init.body as ArrayBuffer | Uint8Array);
|
||||||
bodies.set(path, body);
|
bodies.set(path, body);
|
||||||
|
|
@ -113,7 +131,7 @@ async function fakeServer(seed: Record<string, string>) {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
return { fetchImpl, channel, remotePut };
|
return { fetchImpl, channel, remotePut, state };
|
||||||
}
|
}
|
||||||
|
|
||||||
function makeSource(server: Awaited<ReturnType<typeof fakeServer>>) {
|
function makeSource(server: Awaited<ReturnType<typeof fakeServer>>) {
|
||||||
|
|
@ -239,6 +257,59 @@ describe("syncedLibsSource → editor reload bridge", () => {
|
||||||
* LOCAL caches (peekNamespaces on the backend-named namespaces) + cold-byte
|
* LOCAL caches (peekNamespaces on the backend-named namespaces) + cold-byte
|
||||||
* sums from the list envelope — no stack resolves, no bundle fetches.
|
* sums from the list envelope — no stack resolves, no bundle fetches.
|
||||||
*/
|
*/
|
||||||
|
describe("cutover 409 → re-resolve + retry (load-path-rework 0002)", () => {
|
||||||
|
it("saveItemBody retries once against the re-resolved room and succeeds", async () => {
|
||||||
|
const server = await fakeServer({ "symbol/R": "(r)" });
|
||||||
|
server.state.movedOnce = true;
|
||||||
|
const movedLibs: string[] = [];
|
||||||
|
const source = syncedLibsSource(LIB_ID, {
|
||||||
|
apiBase: API,
|
||||||
|
scope: "s",
|
||||||
|
user: "u",
|
||||||
|
fetchImpl: server.fetchImpl,
|
||||||
|
storeFactory: () => memStore(),
|
||||||
|
channelFactory: () => server.channel,
|
||||||
|
onStackMoved: (id) => movedLibs.push(id),
|
||||||
|
});
|
||||||
|
|
||||||
|
const ok = await source.saveItemBody!(LIB_ID, "symbol", "Mine", "(body)");
|
||||||
|
|
||||||
|
expect(ok).toBe(true);
|
||||||
|
// The refusal invalidated the cached descriptor and re-resolved the stack.
|
||||||
|
expect(movedLibs).toEqual([LIB_ID]);
|
||||||
|
expect(server.state.resolves).toBe(2);
|
||||||
|
source.dispose?.();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("a persistent refusal fails the save after ONE retry", async () => {
|
||||||
|
const server = await fakeServer({ "symbol/R": "(r)" });
|
||||||
|
const alwaysMoved = ((input: unknown, init?: RequestInit) => {
|
||||||
|
// Every write refuses; reads/resolves pass through.
|
||||||
|
if (String(input).includes("/body/") && init?.method === "PUT") {
|
||||||
|
server.state.movedOnce = true;
|
||||||
|
}
|
||||||
|
return (server.fetchImpl as (i: unknown, x?: RequestInit) => unknown)(
|
||||||
|
input,
|
||||||
|
init,
|
||||||
|
);
|
||||||
|
}) as typeof fetch;
|
||||||
|
const source = syncedLibsSource(LIB_ID, {
|
||||||
|
apiBase: API,
|
||||||
|
scope: "s",
|
||||||
|
user: "u",
|
||||||
|
fetchImpl: alwaysMoved,
|
||||||
|
storeFactory: () => memStore(),
|
||||||
|
channelFactory: () => server.channel,
|
||||||
|
});
|
||||||
|
|
||||||
|
const ok = await source.saveItemBody!(LIB_ID, "symbol", "Mine", "(body)");
|
||||||
|
|
||||||
|
expect(ok).toBe(false);
|
||||||
|
expect(server.state.resolves).toBe(2); // initial + the one retry, no loop
|
||||||
|
source.dispose?.();
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("syncedScopeLibsSource.syncState", () => {
|
describe("syncedScopeLibsSource.syncState", () => {
|
||||||
function scoped(libs: LibInfo[]) {
|
function scoped(libs: LibInfo[]) {
|
||||||
const remote: LibsSource = {
|
const remote: LibsSource = {
|
||||||
|
|
|
||||||
|
|
@ -8,6 +8,7 @@ import {
|
||||||
} from "@pcbjam/shared";
|
} from "@pcbjam/shared";
|
||||||
import {
|
import {
|
||||||
peekNamespaces,
|
peekNamespaces,
|
||||||
|
SyncRoomMovedError,
|
||||||
SyncStack,
|
SyncStack,
|
||||||
type ChannelFactory,
|
type ChannelFactory,
|
||||||
type LayerDescriptor,
|
type LayerDescriptor,
|
||||||
|
|
@ -49,6 +50,12 @@ export function syncedLibsSource(
|
||||||
* Returning undefined falls back to resolving this lib on its own.
|
* Returning undefined falls back to resolving this lib on its own.
|
||||||
*/
|
*/
|
||||||
stackFor?: (libId: string) => SyncStackDescriptor | null | undefined;
|
stackFor?: (libId: string) => SyncStackDescriptor | null | undefined;
|
||||||
|
/**
|
||||||
|
* The room refused a write with the cutover 409 (SyncRoomMovedError): the
|
||||||
|
* cached batch-resolved descriptor is stale. The scope source drops it here
|
||||||
|
* so the retry's re-resolve hits the backend for a fresh one.
|
||||||
|
*/
|
||||||
|
onStackMoved?: (libId: string) => void;
|
||||||
/**
|
/**
|
||||||
* SyncStack realtime policy (see SyncStackOptions.realtime). The scope
|
* SyncStack realtime policy (see SyncStackOptions.realtime). The scope
|
||||||
* source passes "shared-only" — a board session warming 150+ libs must not
|
* source passes "shared-only" — a board session warming 150+ libs must not
|
||||||
|
|
@ -195,20 +202,34 @@ export function syncedLibsSource(
|
||||||
return bytes ? new TextDecoder().decode(bytes) : null;
|
return bytes ? new TextDecoder().decode(bytes) : null;
|
||||||
},
|
},
|
||||||
async saveItemBody(_id, kind, name, body): Promise<boolean> {
|
async saveItemBody(_id, kind, name, body): Promise<boolean> {
|
||||||
const { stack } = await ensure();
|
|
||||||
const path = pathOf(kind, name);
|
const path = pathOf(kind, name);
|
||||||
// A successful push fires exactly one change event (the WS self-echo is
|
const bytes = new TextEncoder().encode(body);
|
||||||
// hash-deduped inside the layer), and the stack delivers it AFTER an
|
// Two attempts at most: the second only after a cutover 409 told us the
|
||||||
// async merged read — so the flag must outlive this call; the subscriber
|
// room moved — the stack is re-resolved and the write retried against
|
||||||
// consumes it. A failed push fires none: clear the flag ourselves.
|
// the room the fresh descriptor names (load-path-rework 0002 §3.3).
|
||||||
selfPushed.add(path);
|
for (let attempt = 0; ; attempt++) {
|
||||||
try {
|
const { stack } = await ensure();
|
||||||
await stack.push(path, new TextEncoder().encode(body));
|
// A successful push fires exactly one change event (the WS self-echo
|
||||||
return true;
|
// is hash-deduped inside the layer), and the stack delivers it AFTER
|
||||||
} catch (e) {
|
// an async merged read — so the flag must outlive this call; the
|
||||||
selfPushed.delete(path);
|
// subscriber consumes it. A failed push fires none: clear it ourselves.
|
||||||
log(`[synced] save failed for ${kind}/${name}: ${String(e)}`);
|
selfPushed.add(path);
|
||||||
return false;
|
try {
|
||||||
|
await stack.push(path, bytes);
|
||||||
|
return true;
|
||||||
|
} catch (e) {
|
||||||
|
selfPushed.delete(path);
|
||||||
|
if (e instanceof SyncRoomMovedError && attempt === 0) {
|
||||||
|
log(`[synced] room moved for lib ${libId} — re-resolving stack`);
|
||||||
|
opts.onStackMoved?.(libId);
|
||||||
|
const stale = opened;
|
||||||
|
opened = null;
|
||||||
|
stale?.then((r) => r.stack.close()).catch(() => {});
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
log(`[synced] save failed for ${kind}/${name}: ${String(e)}`);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
dispose(): void {
|
dispose(): void {
|
||||||
|
|
@ -345,6 +366,7 @@ export function syncedScopeLibsSource(
|
||||||
// via the muxed team mirror channel.
|
// via the muxed team mirror channel.
|
||||||
realtime: "shared-only",
|
realtime: "shared-only",
|
||||||
stackFor: (id) => (batchedStacks.has(id) ? batchedStacks.get(id) : undefined),
|
stackFor: (id) => (batchedStacks.has(id) ? batchedStacks.get(id) : undefined),
|
||||||
|
onStackMoved: (id) => batchedStacks.delete(id),
|
||||||
});
|
});
|
||||||
perLib.set(libId, src);
|
perLib.set(libId, src);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue