diff --git a/conformance/spec/folders.md b/conformance/spec/folders.md index c350cf9de..8c19c3833 100644 --- a/conformance/spec/folders.md +++ b/conformance/spec/folders.md @@ -3626,7 +3626,7 @@ When the person changes or moves a folder file whose row the copy no longer hold While a folder file whose row the copy no longer holds stays unchanged at its path, a device MUST report the file as lost. -**Tests:** `device/folders-identity.test.ts › queues a file whose row the copy lost again once it changes, and says so`, `device/folders-identity.test.ts › keeps a file the folder never wrote whose create was refused, inside one push`. +**Tests:** `device/folders-identity.test.ts › queues a file whose row the copy lost again once it changes, and says so`, `device/folders-identity.test.ts › keeps a file the folder never wrote whose create was refused, inside one push`, `device/folders.test.ts › pulls again after a rebuild at the drain after the pull, writing what the new view holds and keeping a file it no longer reads`. ### `folders/lost-create-not-repeated` @@ -3690,6 +3690,44 @@ When a folder’s cursor has aged out of the server log, a device MUST hydrate a **Tests:** `device/folders.test.ts › hydrates again at a push whose cursor the log has aged past`, `device/folders.test.ts › hydrates again while watching when the log ages past its cursor`. +### `folders/view-changed-push-rebuilds` + +When any drain of a folder push or a send-only run, the drain after the push's pull included, ends `copy_expired` with the reason `read_view_changed`, a device MUST hydrate the copy again and drain once more in the same run, sending each write still waiting once and reporting the verdict of every write that either drain settled. + +**Reason:** A key that is replaced or narrowed changes the read view the copy was built under (`device/view-changed`), and a run that stopped there would leave every write behind it waiting for no cause the person can act on. The rebuild reads under the new view, so nothing the new key cannot read stays in the copy. A verdict the first drain settled is final in the queue, and a report without it would tell the person that a refused write was taken. + +**Tests:** `device/folders.test.ts › builds the copy again and delivers every waiting write in the same $run`, `device/folders.test.ts › reports a refusal from before the view changed with the writes sent after the rebuild`, `device/folders.test.ts › builds the copy again where the drain after the pull meets the changed view, and finishes the push`, `device/folders.test.ts › reports a placement refused where its own read meets the changed view, and does not send it again`. + +### `folders/view-changed-push-pulls-again` + +When the drain after a folder push's pull hydrates the copy again under `folders/view-changed-push-rebuilds`, a device MUST pull once more in the same run, and drain once more where that pull queued the placement of a file it wrote. + +**Reason:** The first pull wrote the directory out from the copy built under the old read view. Pulling from the rebuilt copy writes what the new view holds in the same push, rather than leaving it for the next one. A file bound to a row the new key cannot read is not removed by that pull; the next scan reports it as lost (`folders/lost-row-unchanged-held`). + +**Tests:** `device/folders.test.ts › pulls again after a rebuild at the drain after the pull, writing what the new view holds and keeping a file it no longer reads`. + +### `folders/view-changed-push-once` + +If a drain of a folder push or a send-only run that has already rebuilt its copy for a changed read view also ends `copy_expired` with the reason `read_view_changed`, then a device MUST end the run `copy_expired` without hydrating again, keeping the writes still waiting. + +**Reason:** A view that changes again at once is changing under the run, and rebuilding without end would spend the key's rate limit and never finish. + +**Tests:** `device/folders.test.ts › builds the copy again only once in a push, and ends it copy_expired where the view changes again`. + +### `folders/view-changed-rebuild-fails` + +If the hydration a folder push takes for a changed read view fails, then the CLI MUST end the push with that hydration's own error and exit code, keeping the writes still waiting for the next push. + +**Reason:** The person acts on why the rebuild failed, such as a server that cannot be reached, not on the changed view the rebuild was answering. + +**Tests:** `device/folders.test.ts › ends the push with the rebuild's own failure, keeping the writes for the next push`. + +### `folders/view-changed-watch-rebuilds` + +When a folder watch's drain ends `copy_expired` with the reason `read_view_changed`, a device MUST hydrate the copy again and send the writes still waiting without ending the watch. + +**Tests:** `device/folders.test.ts › builds the copy again while watching and delivers every waiting write, still watching`. + ### `folders/failed-hydration-retried` When a folder push previously failed to hydrate a usable copy, a device MUST retry hydration at the next push. @@ -4996,6 +5034,14 @@ When a folder watch's pass leaves a write undelivered and its drain reports `ret **Tests:** `device/folders.test.ts › sends again when the Retry-After a rate limit named is over, without waiting for a change`. +### `folders/watch-rate-limit-handover` + +When a folder watch takes back a folder it let go to a command, a device MUST NOT start a pass of that folder before the time named by the `retry_after_seconds` of the most recent drain of that folder that left a write undelivered, whether the watch or the command drained, unless a later drain of the folder has had writes answered with no `retry_after_seconds`. + +**Reason:** A hand-over does not reset the server's rate limit, so a watch that forgot the time would send at once and be refused again (`folders/watch-rate-limit-pause`). + +**Tests:** `device/folders-handover.test.ts › keeps the server's wait across handing a folder over from $name`. + ### `folders/push-exit-undelivered` When a folder push ends with its drain reporting `unavailable` or a write `undelivered`, and no refused credential stopped it, the CLI MUST exit 3. diff --git a/conformance/src/device/cli-adapter.ts b/conformance/src/device/cli-adapter.ts index 9c8e6eaf0..711575094 100644 --- a/conformance/src/device/cli-adapter.ts +++ b/conformance/src/device/cli-adapter.ts @@ -1260,6 +1260,10 @@ export interface PushReport { hydrated?: HydrateReport | null; failed?: string; }; + /** The hydration the drain took where the server said the copy's read + * view had changed, before it sent the writes still waiting + * (`folders/view-changed-push-rebuilds`). */ + rebuilt: HydrateReport | null; /** `null` where a failed catch-up left no copy to pull from. */ pull: PullReport | null; /** The exit the command ended with: 0; 3 where it left a write @@ -1277,6 +1281,7 @@ export interface SendOnlyReport { settings: SettingsFileReport; scan: ScanReport; drain: PushReport["drain"]; + rebuilt: HydrateReport | null; /** The exit the command ended with, added by the adapter. */ exit: number; } diff --git a/conformance/src/suites/device/folders-handover.test.ts b/conformance/src/suites/device/folders-handover.test.ts index 27f3f21d1..0f66b4391 100644 --- a/conformance/src/suites/device/folders-handover.test.ts +++ b/conformance/src/suites/device/folders-handover.test.ts @@ -488,6 +488,88 @@ describe("handing a folder to a command", () => { }, WAIT); }); + it.each([ + { + name: "its own watch, to a command that sends nothing", + every: false, + command: "scan", + waits: [8], + }, + { + name: "the watch of every folder, to a push the server names a later wait", + every: true, + command: "push", + waits: [4, 8], + }, + ] as const)( + "keeps the server's wait across handing a folder over from $name", + async ({ every, command, waits }) => { + // A monotonic clock, so an adjustment of the wall clock cannot move the + // gaps this measures. + const attempts: number[] = []; + const harness = await oneFolder( + `handover-rate-limit-${command}`, + (answer) => { + attempts.push(performance.now()); + const wait = waits[attempts.length - 1]; + return wait === undefined + ? answer + : { + kind: "json", + status: 429, + headers: { "retry-after": String(wait) }, + body: { + error: { code: "rate_limited", message: "Too many requests" }, + }, + }; + }, + ); + const watch = hold( + every ? harness.folder.watchEvery() : harness.folder.watch(), + ); + await vi.waitFor( + () => + expect( + lines(watch, every ? harness.dir : undefined).some( + (line) => line.watching === true, + ) || watch.stderr.includes("watching"), + watch.stdout + watch.stderr, + ).toBe(true), + WAIT, + ); + put(harness, "One.md", note("One")); + await vi.waitFor( + () => expect(attempts, watch.stderr).toHaveLength(1), + WAIT, + ); + const worked = + command === "scan" + ? await harness.folder.scan() + : await harness.folder.push(); + expect(worked.ok, JSON.stringify(worked)).toBe(true); + // The push sent the write once more, and was told to wait longer. + expect(attempts).toHaveLength(waits.length); + await vi.waitFor(() => { + expect(handovers(watch, every ? harness.dir : undefined)).toEqual([ + "let go", + "taken back", + ]); + }, WAIT); + await vi.waitFor( + () => expect(attempts, watch.stderr).toHaveLength(waits.length + 1), + { timeout: 60_000, interval: 100 }, + ); + const last = waits.length - 1; + expect( + attempts[last + 1]! - attempts[last]!, + "the watch taken back sent before the wait the server last named", + ).toBeGreaterThanOrEqual(waits[last]! * 1000 - 100); + expect(sentTitles(harness)).toEqual( + Array.from({ length: waits.length + 1 }, () => "One"), + ); + }, + ); + it("lets a folder go only once the write its pass is sending is answered", async () => { let release: () => void = () => undefined; const gate = new Promise((resolve) => { diff --git a/conformance/src/suites/device/folders.shared.ts b/conformance/src/suites/device/folders.shared.ts index c686c5ff8..6b9f0720b 100644 --- a/conformance/src/suites/device/folders.shared.ts +++ b/conformance/src/suites/device/folders.shared.ts @@ -155,6 +155,9 @@ export function scriptFolderWrites( /** The answer a create is given in place of the door's own, which it is * handed: a gated one, say. */ creating?: (answer: Answer, request: RecordedRequest) => Answer; + /** The answer the read of a row is given in place of the door's own, + * which it is handed: a refusal under a read view that changed, say. */ + reading?: (answer: Answer, request: RecordedRequest) => Answer; } = {}, ): Map { // What the hydration served, so a write to one of those rows is answered @@ -203,7 +206,12 @@ export function scriptFolderWrites( }, ], // Reconciliation reads the current server row after each settled write. - read: [(request) => door.read(request.pathname.split("/").at(-1) ?? "")], + read: [ + (request) => { + const answer = door.read(request.pathname.split("/").at(-1) ?? ""); + return options.reading?.(answer, request) ?? answer; + }, + ], // The tags a row holds move as the tag doors are told. tags: [ (request) => { @@ -267,6 +275,9 @@ export class EdgeDoor { /** An answer standing in for the server's to a listing of an item's * edges, a refusal say; the door answers where it gives `undefined`. */ listing?: (source: string) => Answer | undefined; + /** The same, for the read of one edge by id: a refusal under a read view + * that changed, say. */ + reading?: (id: string) => Answer | undefined; /** Edges an item's listing serves as they were, once each: a read answered * before a change the copy has since taken (`device.md`, What the real * server cannot be made to produce). */ @@ -373,7 +384,10 @@ export class EdgeDoor { this.failListings -= 1; return refusal(503, "unavailable", "Try again"); } - const edge = this.edges.get(request.pathname.split("/").at(-1) ?? ""); + const id = request.pathname.split("/").at(-1) ?? ""; + const standIn = this.reading?.(id); + if (standIn !== undefined) return standIn; + const edge = this.edges.get(id); return edge === undefined ? refusal(404, "edge_not_found", "No such edge") : writeAnswers.edge(edge, 200); diff --git a/conformance/src/suites/device/folders.test.ts b/conformance/src/suites/device/folders.test.ts index b14e7a20c..66463f94d 100644 --- a/conformance/src/suites/device/folders.test.ts +++ b/conformance/src/suites/device/folders.test.ts @@ -2588,6 +2588,431 @@ describe("files and items", () => { } }); + describe("a copy whose read view changes under a push", () => { + /** + * A folder of notes whose server, at the read after the first write it + * answers, says the copy's read view has changed, as it does once the + * folder's key is replaced. A hydration lists every note created since, + * as the server would, and reads the catalog through `catalog` once the + * view has changed. Where `again`, every read by id is refused. The create + * of the note titled `refused` is refused `validation_error`, and the read + * its refusal owes is answered. + */ + async function rekeyed( + label: string, + { + catalog = () => typeCatalog(), + again = false, + refused, + }: { catalog?: () => Answer; again?: boolean; refused?: string } = {}, + ): Promise<{ harness: FolderHarness; changed: () => boolean }> { + const notes: Array<{ item: WireItemOptions }> = []; + const made = await folderHarness(label, { rows: { "core.note": notes } }); + let changed = false; + // The refused create's row, which the door must not hold, and whose + // read the refusal owes is answered before the view changes. + const refusedIds = new Set(); + const held = scriptFolderWrites(made, { + creating: (answer, request) => { + const { id, properties } = JSON.parse(request.body) as { + id: string; + properties: { title?: unknown }; + }; + if (refused !== undefined && properties.title === refused) { + held.delete(id); + refusedIds.add(id); + return refusal( + 400, + "validation_error", + "That title is not allowed", + ); + } + if (answer.kind === "json" && answer.status === 201) { + const { item } = answer.body as { item: WireItemOptions }; + notes.push({ item }); + } + return answer; + }, + reading: (answer, request) => { + if (changed && !again) return answer; + // A bulk read resolves each id through this door too, and a + // hydration's reads are answered: only a read by id is refused. + if (!made.server.requests.includes(request)) return answer; + if (refusedIds.has(request.pathname.split("/").at(-1) ?? "")) { + return answer; + } + changed = true; + return refusal(409, "read_view_changed", "The read view has changed"); + }, + }); + made.server.copyAnswer("GET", "/types", () => + changed ? catalog() : typeCatalog(), + ); + return { harness: made, changed: () => changed }; + } + + /** The copy-stream bootstraps a hydration starts with. */ + function hydrations(made: FolderHarness): number { + return made.server.requests.filter( + (request) => + request.method === "GET" && + request.pathname === "/events" && + request.headers["last-event-id"] === undefined && + request.headers["x-marfa-read-view"] === undefined, + ).length; + } + + it.each([ + { run: "push", sendOnly: false }, + { run: "send-only push", sendOnly: true }, + ])( + "builds the copy again and delivers every waiting write in the same $run", + async ({ sendOnly }) => { + const made = await rekeyed( + `folder-push-view-changed${sendOnly ? "-send-only" : ""}`, + ); + harness = made.harness; + for (const title of ["One", "Two"]) { + put(harness, `${title}.md`, `---\ntitle: ${title}\n---\n${title}\n`); + } + const before = hydrations(harness); + const pushed = sendOnly + ? await harness.folder.pushSendOnly() + : await harness.folder.push(); + expect(pushed.ok, JSON.stringify(pushed)).toBe(true); + if (!pushed.ok) return; + // The witness: the server did say the view changed, under this push. + expect(made.changed()).toBe(true); + expect( + hydrations(harness) - before, + "the push did not build the copy again", + ).toBe(1); + expect(pushed.value.exit, JSON.stringify(pushed.value)).toBe(0); + expect(pushed.value.drain.undelivered).toBe(0); + expect(pushed.value.rebuilt?.items).toBeGreaterThan(0); + expect( + sentTitles(harness).sort(), + "a write was lost, or sent twice, across the rebuild", + ).toEqual(["One", "Two"]); + + const again = await harness.folder.push(); + expect(again.ok, JSON.stringify(again)).toBe(true); + if (!again.ok) return; + expect(again.value.exit).toBe(0); + expect(again.value.rebuilt).toBeNull(); + expect(sentTitles(harness).sort()).toEqual(["One", "Two"]); + }, + ); + + it("reports a refusal from before the view changed with the writes sent after the rebuild", async () => { + const made = await rekeyed("folder-push-view-changed-after-refusal", { + refused: "Alpha", + }); + harness = made.harness; + for (const title of ["Alpha", "Beta", "Gamma"]) { + put(harness, `${title}.md`, `---\ntitle: ${title}\n---\n${title}\n`); + } + const before = hydrations(harness); + const pushed = await harness.folder.push(); + expect(pushed.ok, JSON.stringify(pushed)).toBe(true); + if (!pushed.ok) return; + // The witness: the view changed under this push, after the refusal. + expect(made.changed()).toBe(true); + expect(hydrations(harness) - before).toBe(1); + const refusals = pushed.value.drain.verdicts.filter( + (verdict) => + verdict.kind === "create_item" && verdict.verdict === "refused", + ); + expect( + refusals.map((verdict) => verdict.refusal?.code), + `the refusal from before the rebuild left the push's report: ${JSON.stringify(pushed.value.drain)}`, + ).toEqual(["validation_error"]); + // A refusal is the write's verdict, not a write left waiting. + expect(pushed.value.exit, JSON.stringify(pushed.value)).toBe(0); + expect(pushed.value.drain.undelivered).toBe(0); + expect( + sentTitles(harness).sort(), + "a write was lost, or sent twice, across the rebuild", + ).toEqual(["Alpha", "Beta", "Gamma"]); + const queued = await harness.folder.device().queue(); + expect(queued.ok, JSON.stringify(queued)).toBe(true); + if (!queued.ok) return; + expect( + queued.value + .filter( + (write) => + write.kind === "create_item" && write.verdict === "refused", + ) + .map((write) => write.reason), + ).toEqual(["validation_error"]); + }); + + it("builds the copy again where the drain after the pull meets the changed view, and finishes the push", async () => { + const edges = new EdgeDoor(); + harness = await folderHarness("folder-push-view-changed-after-pull", { + rows: { + "core.note": [ + { + item: { + id: "01a00000-0000-7000-8000-0000000099a1", + properties: { title: "One", body: "made elsewhere\n" }, + }, + }, + { + item: { + id: "01a00000-0000-7000-8000-0000000099a2", + properties: { title: "Two", body: "made elsewhere\n" }, + }, + }, + ], + }, + }); + // The read of the first placement the pull queued is refused, as the + // server does once the folder's key is replaced. + let changed = false; + edges.reading = () => { + if (changed) return undefined; + changed = true; + return refusal(409, "read_view_changed", "The read view has changed"); + }; + scriptFolderWrites(harness, { edges }); + const before = hydrations(harness); + const pushed = await harness.folder.push(); + expect(pushed.ok, JSON.stringify(pushed)).toBe(true); + if (!pushed.ok) return; + expect(changed).toBe(true); + expect(pushed.value.pull?.written).toBe(2); + expect( + hydrations(harness) - before, + "the push did not build the copy again after the pull", + ).toBe(1); + expect(pushed.value.rebuilt).not.toBeNull(); + expect(pushed.value.exit, JSON.stringify(pushed.value)).toBe(0); + expect(pushed.value.drain.undelivered).toBe(0); + expect(placedPaths(edges)).toEqual(["One.md", "Two.md"]); + expect( + harness.server.requests.filter( + (request) => + request.method === "POST" && request.pathname === "/edges", + ), + "a placement was sent twice across the rebuild", + ).toHaveLength(2); + + const again = await harness.folder.push(); + expect(again.ok, JSON.stringify(again)).toBe(true); + if (!again.ok) return; + expect(again.value.exit).toBe(0); + expect(again.value.rebuilt).toBeNull(); + expect([again.value.drain.answered, again.value.pull?.written]).toEqual([ + 0, 0, + ]); + }); + + it("builds the copy again while watching and delivers every waiting write, still watching", async () => { + const made = await rekeyed("folder-watch-view-changed"); + harness = made.harness; + for (const title of ["One", "Two"]) { + put(harness, `${title}.md`, `---\ntitle: ${title}\n---\n${title}\n`); + } + const before = hydrations(harness); + const watching = harness.folder.watch(); + try { + await vi.waitFor( + () => + expect(sentTitles(harness!).sort(), watching.stderr).toEqual([ + "One", + "Two", + ]), + { timeout: 30_000, interval: 100 }, + ); + expect(made.changed()).toBe(true); + expect(hydrations(harness) - before).toBeGreaterThanOrEqual(1); + expect(watching.running(), watching.stderr).toBe(true); + } finally { + await watching.stop(); + } + expect(sentTitles(harness).sort()).toEqual(["One", "Two"]); + }); + + it("builds the copy again only once in a push, and ends it copy_expired where the view changes again", async () => { + const made = await rekeyed("folder-push-view-changes-twice", { + again: true, + }); + harness = made.harness; + for (const title of ["One", "Two", "Three"]) { + put(harness, `${title}.md`, `---\ntitle: ${title}\n---\n${title}\n`); + } + const before = hydrations(harness); + const pushed = await harness.folder.push(); + expect(pushed.ok, JSON.stringify(pushed)).toBe(false); + if (pushed.ok) return; + expect(pushed.refusal.code).toBe("copy_expired"); + expect(pushed.refusal.raw).toContain("read_view_changed"); + expect(hydrations(harness) - before).toBe(1); + // The witness: the drain after the rebuild went, and read the answered + // write back under the new view, which was refused again. + const settingsId = harness.settings.id; + expect( + harness.server.requests.filter( + (request) => + request.method === "GET" && + /^\/items\/[^/]+$/.test(request.pathname) && + request.pathname !== `/items/${settingsId}` && + request.headers["x-marfa-read-view"] !== undefined, + ), + ).toHaveLength(2); + expect(sentTitles(harness)).toEqual(["One"]); + }); + + it("ends the push with the rebuild's own failure, keeping the writes for the next push", async () => { + let failing = true; + const made = await rekeyed("folder-push-view-changed-rebuild-fails", { + catalog: () => (failing ? answers.serverFault() : typeCatalog()), + }); + harness = made.harness; + for (const title of ["One", "Two"]) { + put(harness, `${title}.md`, `---\ntitle: ${title}\n---\n${title}\n`); + } + const pushed = await harness.folder.push(); + expect(pushed.ok, JSON.stringify(pushed)).toBe(false); + if (pushed.ok) return; + expect(made.changed()).toBe(true); + // The witness for the code: the drain alone ends `copy_expired`. + expect( + pushed.refusal.code, + "the push said the copy expired rather than why the rebuild failed", + ).toBe("server"); + expect(pushed.refusal.raw).toContain('"exit":3'); + // The first create was taken before the view changed; the second waits. + expect(sentTitles(harness)).toHaveLength(1); + const queued = await harness.folder.device().queue(); + expect(queued.ok, JSON.stringify(queued)).toBe(true); + if (!queued.ok) return; + expect( + queued.value.filter( + (write) => write.kind === "create_item" && write.verdict === null, + ), + "the write waiting behind the change was not kept", + ).toHaveLength(1); + + failing = false; + const next = await harness.folder.push(); + expect(next.ok, JSON.stringify(next)).toBe(true); + if (!next.ok) return; + expect(next.value.exit).toBe(0); + expect( + sentTitles(harness).sort(), + "a write was lost, or sent twice, after the rebuild failed", + ).toEqual(["One", "Two"]); + }); + + it("reports a placement refused where its own read meets the changed view, and does not send it again", async () => { + const edges = new EdgeDoor(); + edges.placing = (edge) => + edge.edge_type === "in-folder" + ? refusal( + 403, + "edge_permission_denied", + "Write access to edge type denied", + ) + : undefined; + // The read the refusal owes, to learn whether the edge was made, is + // refused, as the server does once the folder's key is replaced. + let changed = false; + edges.reading = () => { + if (changed) return undefined; + changed = true; + return refusal(409, "read_view_changed", "The read view has changed"); + }; + harness = await folderHarness("folder-push-view-changed-refused-read", { + rows: { + "core.note": [ + { + item: { + id: "01a00000-0000-7000-8000-0000000099b1", + properties: { title: "Elsewhere", body: "made elsewhere\n" }, + }, + }, + ], + }, + }); + scriptFolderWrites(harness, { edges }); + const pushed = await harness.folder.push(); + expect(pushed.ok, JSON.stringify(pushed)).toBe(true); + if (!pushed.ok) return; + // The witness: the refusal's own read was refused under the new view. + expect(changed).toBe(true); + expect(pushed.value.rebuilt).not.toBeNull(); + expect( + pushed.value.drain.verdicts.map((entry) => [ + entry.kind, + entry.verdict, + entry.refusal?.code, + ]), + "the refusal whose read met the changed view left the push's report", + ).toEqual([["create_edge", "refused", "edge_permission_denied"]]); + expect(placedPaths(edges)).toEqual([]); + + const next = await harness.folder.push(); + expect(next.ok, JSON.stringify(next)).toBe(true); + if (!next.ok) return; + expect(postedEdges(harness), "a refused placement was sent again").toBe( + 1, + ); + }); + + it("pulls again after a rebuild at the drain after the pull, writing what the new view holds and keeping a file it no longer reads", async () => { + const edges = new EdgeDoor(); + const one = "01a00000-0000-7000-8000-0000000099c1"; + const two = "01a00000-0000-7000-8000-0000000099c2"; + const three = "01a00000-0000-7000-8000-0000000099c3"; + const notes: Array<{ item: WireItemOptions }> = [ + { item: { id: one, properties: { title: "One", body: "one\n" } } }, + { item: { id: two, properties: { title: "Two", body: "two\n" } } }, + ]; + harness = await folderHarness("folder-push-view-changed-pull-again", { + rows: { "core.note": notes }, + }); + const rows = scriptFolderWrites(harness, { edges }); + // At the read of the first placement the pull queued, the view changes: + // the new key reads Three and no longer reads Two. + let changed = false; + edges.reading = () => { + if (changed) return undefined; + changed = true; + const properties = { title: "Three", body: "three\n" }; + notes.splice(1, 1, { item: { id: three, properties } }); + const held = rows.get(two); + rows.delete(two); + if (held !== undefined) rows.set(three, { ...held, properties }); + return refusal(409, "read_view_changed", "The read view has changed"); + }; + const pushed = await harness.folder.push(); + expect(pushed.ok, JSON.stringify(pushed)).toBe(true); + if (!pushed.ok) return; + expect(changed).toBe(true); + expect(pushed.value.rebuilt).not.toBeNull(); + expect(pushed.value.exit, JSON.stringify(pushed.value)).toBe(0); + expect( + existsSync(join(harness.dir, "Three.md")), + "the push did not pull what the rebuilt copy holds", + ).toBe(true); + expect(placedPaths(edges)).toContain("Three.md"); + expect(readFileSync(join(harness.dir, "Two.md"), "utf8")).toContain( + "two\n", + ); + + const next = await harness.folder.push(); + expect(next.ok, JSON.stringify(next)).toBe(true); + if (!next.ok) return; + expect(next.value.scan.lost).toBe(1); + expect(sentTitles(harness)).toEqual([]); + expect(readFileSync(join(harness.dir, "Two.md"), "utf8")).toContain( + "two\n", + ); + }); + }); + describe("a copy that falls behind the log", () => { const id = "01a00000-0000-7000-8000-0000000000c3"; const agedOut: Answer = { diff --git a/core/README.md b/core/README.md index 259e411a8..02e7ed43e 100644 --- a/core/README.md +++ b/core/README.md @@ -42,7 +42,7 @@ A watch is the one source that an app's listeners share. `Core::watch` (`watch` Reading commands on an absent store, including `--reader` and `device changes`, report `no_store` and exit 2. Queue JSON retains dependency history; plain “waiting on” names only unresolved dependencies. -A move of an edge's source changes the server's read view, so a drain that sends one the server takes ends `copy_expired`, but only after it has sent the writes queued behind it. Each is judged as it would have been, the reads their answers owe are left for the hydration, and any other change to the read view, or one met by a read an earlier drain left owed, still ends the drain at once. One hydration then covers every move the drain sent. A drain that also meets a wait, an unavailable server, a credential stop or a write it cannot deliver returns its report with the copy expired, and the next call is refused `copy_expired`; an error of its own, such as a stop or a redirect, is the one it ends with. +A move of an edge's source changes the server's read view, so a drain that sends one the server takes ends `copy_expired`, but only after it has sent the writes queued behind it. Each is judged as it would have been, the reads their answers owe are left for the hydration, and any other change to the read view, or one met by a read an earlier drain left owed, still ends the drain at once. One hydration then covers every move the drain sent. A drain that also meets a wait, an unavailable server, a credential stop or a write it cannot deliver returns its report with the copy expired, and the next call is refused `copy_expired`; an error of its own, such as a stop or a redirect, is the one it ends with. A `folders push`, with or without `--send-only`, any of whose drains, the one after the pull included, ends `copy_expired` because the read view changed, as it does once the folder's key is replaced, hydrates the copy again and drains once more, so the writes still waiting are sent in the same run, each once; its report names the hydration as `rebuilt`, and tells the verdicts of both drains, a refusal from before the change included, and a verdict whose own read met the change. Where the drain after the pull rebuilt the copy, the push pulls once more, so the directory shows what the new view holds, and drains again where that pull placed a file; a file whose row the new key cannot read stays, and the next scan reports it lost. It rebuilds once a run: a view that changes again ends the run `copy_expired`, and a hydration that fails ends it with that failure. Either way the writes still waiting stay queued. A folder's watch does the same and keeps watching. ## Folder directories @@ -58,7 +58,7 @@ A `marfa folders add` that is refused or fails removes what it made: each direct `marfa folders watch` with no directory watches every folder the registry lists, each with a watch of its own on a thread of its own, in one process. Each folder uses the server its copy was hydrated from: with the command's own credential where that is the server the command names, and otherwise with the credential the keychain keeps for that server, so a key named for one server is never sent to another. A folder that has never hydrated uses the server the command names. The registry is read again every 2 seconds: a folder listed since joins the watch, and the watch of one no longer listed ends. A registry that cannot be read leaves the folders watched as they are. A folder that cannot be opened, because its first sync waits, another process works it, or there is no credential for its server, is reported once and tried again every 5 seconds. The folders whose directories cannot be found, as a deleted directory or a volume that is away, are reported together in one line once that set has stayed the same for 2 seconds, and again only when another folder joins it; each is looked for again 5 seconds later, then at a wait that doubles up to 5 minutes, on its own thread, so a volume that does not answer holds up no other folder. The record of folder watches gives each the reason `missing`. They stay listed, since a volume that is away cannot be told from a deleted directory and the entry lets another folder hold a delete for a file that may have moved into it; `folders remove` takes one off the list. A folder whose watch ends, as when the server refuses its credential or its changes stop reaching it, is reported once and tried again after 60 seconds; one whose credential was refused is opened again only with another credential. A reason is reported again only when it changes, or after the folder was watched in between. Every line about one folder names its directory, and with `--json` each such line is one JSON object with a `dir` field. A folder whose directory is gone can be removed, and leaves the watch. -One process works a folder at a time. A command that works a folder, such as `folders add` over an existing folder, `folders confirm`, `folders restore`, `folders remove`, `folders scan`, `folders pull`, `folders push`, `folders hydrate` or `folders unmatched --remove`, and finds it held asks the process holding it to let it go, and waits up to 10 seconds. A watch, of one folder or of every folder, sees the request within a fraction of a second, finishes the step of its pass under way (sending the settings file's edit, the scan, the drain or the pull, so no write is cut off), lets the folder go and says so, and goes on with the other folders it watches. It tries the folder again every second, with the server and credential it began with, and takes it back, saying so, once the command ends. A folder the command removed or took off the registry leaves the watch, and a watch of that folder alone ends. A folder held by another process, with no command asking for it, for more than 10 seconds is said once to be held by another process, and watched again once that process lets it go. A process that does not let go in time, such as a suspended watch, one whose step waits on a slow server, or another command still working the folder, leaves the command refused with `reading_handle`. The request is a lock the asking command holds in the folder's `.marfa` directory while it waits and works, which the system lets go however the command ends, so a command that crashes never leaves the folder blocked. A request that cannot be made, as in a read-only `.marfa`, leaves the command waiting all the same and then refused with the reason. A watch never asks, so a second watch of a folder is refused at once. The Swift binding does the same for an app: its `add`, `sync`, `confirm_first_sync`, `confirm`, `restore` and `remove` ask and wait up to 10 seconds, and are refused with `ReadingHandle`, naming the folder, past that. A watch an app runs through the binding lets the folder go to a command or another call, and takes it back, as the command line's watch does, and tells its listener with the events `HandedOver`, `HeldElsewhere`, `TakenBack`, `Removed` and `Unlisted`; a folder removed or taken off the registry ends that watch. `status` asks no one. +One process works a folder at a time. A command that works a folder, such as `folders add` over an existing folder, `folders confirm`, `folders restore`, `folders remove`, `folders scan`, `folders pull`, `folders push`, `folders hydrate` or `folders unmatched --remove`, and finds it held asks the process holding it to let it go, and waits up to 10 seconds. A watch, of one folder or of every folder, sees the request within a fraction of a second, finishes the step of its pass under way (sending the settings file's edit, the scan, the drain or the pull, so no write is cut off), lets the folder go and says so, and goes on with the other folders it watches. It tries the folder again every second, with the server and credential it began with, and takes it back, saying so, once the command ends. A wait the server named to the last drain of that folder that left a write undelivered, the watch's or the command's, still holds when the watch takes the folder back, or a watch of it starts, until it is over or a later drain has writes taken with no wait. A folder the command removed or took off the registry leaves the watch, and a watch of that folder alone ends. A folder held by another process, with no command asking for it, for more than 10 seconds is said once to be held by another process, and watched again once that process lets it go. A process that does not let go in time, such as a suspended watch, one whose step waits on a slow server, or another command still working the folder, leaves the command refused with `reading_handle`. The request is a lock the asking command holds in the folder's `.marfa` directory while it waits and works, which the system lets go however the command ends, so a command that crashes never leaves the folder blocked. A request that cannot be made, as in a read-only `.marfa`, leaves the command waiting all the same and then refused with the reason. A watch never asks, so a second watch of a folder is refused at once. The Swift binding does the same for an app: its `add`, `sync`, `confirm_first_sync`, `confirm`, `restore` and `remove` ask and wait up to 10 seconds, and are refused with `ReadingHandle`, naming the folder, past that. A watch an app runs through the binding lets the folder go to a command or another call, and takes it back, as the command line's watch does, and tells its listener with the events `HandedOver`, `HeldElsewhere`, `TakenBack`, `Removed` and `Unlisted`; a folder removed or taken off the registry ends that watch. `status` asks no one. ### Sync folders from login diff --git a/core/marfa-cli/src/folders.rs b/core/marfa-cli/src/folders.rs index bd0531eb1..aef53da97 100644 --- a/core/marfa-cli/src/folders.rs +++ b/core/marfa-cli/src/folders.rs @@ -405,6 +405,7 @@ pub fn run( settings, scan: scanned, drain: drained, + rebuilt, } = folder.send_only()?; exit = output::drain_exit(&drained.report); output::report( @@ -415,6 +416,7 @@ pub fn run( "settings": settings, "scan": scanned, "drain": drained, + "rebuilt": rebuilt, }), json, || { @@ -426,6 +428,7 @@ pub fn run( lines.push(describe_scan(&scanned)); lines.extend(flagged_lines(&scanned.flagged)); lines.extend(embed_lines(&scanned.embeds)); + lines.extend(rebuilt_line(rebuilt.as_ref())); lines.extend(drain_lines(&folder, &drained)); lines.push( "send-only run: nothing was pulled, so no item's file was written or removed; the next `folders push` pulls what the server holds" @@ -445,6 +448,7 @@ pub fn run( settings, scan: scanned, drain: drained, + rebuilt, catch_up, pull: pulled, } = match folder.sync()? { @@ -472,6 +476,7 @@ pub fn run( "settings": settings, "scan": scanned, "drain": drained, + "rebuilt": rebuilt, "catch_up": caught, "pull": pulled, }), @@ -503,6 +508,7 @@ pub fn run( .iter() .chain(pulled.iter().flat_map(|pulled| &pulled.embeds)), )); + lines.extend(rebuilt_line(rebuilt.as_ref())); lines.extend(drain_lines(&folder, &drained)); if let Some(error) = &failed { lines.push(format!("could not catch up: {error}")); @@ -649,6 +655,15 @@ fn asks_to_remove(dir: &Path, going: &[String]) -> Result { Ok(matches!(answer.trim().to_lowercase().as_str(), "y" | "yes")) } +fn rebuilt_line(rebuilt: Option<&marfa_core::HydrateReport>) -> Option { + rebuilt.map(|rebuilt| { + format!( + "the server's read view changed, so the copy was hydrated again ({} item(s)) before the waiting writes were sent", + rebuilt.items + ) + }) +} + /// What a drain sent and what came of it, said once. fn drain_lines(folder: &Folder, drained: &marfa_core::Drained) -> Vec { let mut lines = vec![output::counts(&drained.report)]; diff --git a/core/marfa-core/src/drain.rs b/core/marfa-core/src/drain.rs index 5140a6dce..925e26835 100644 --- a/core/marfa-core/src/drain.rs +++ b/core/marfa-core/src/drain.rs @@ -53,6 +53,21 @@ pub struct DrainReport { } impl DrainReport { + pub(crate) fn empty() -> DrainReport { + DrainReport { + answered: 0, + held: 0, + undelivered: 0, + unsent: 0, + unmade: 0, + unavailable: None, + verdicts: Vec::new(), + stopped: None, + unclaimed_sources: Vec::new(), + retry_after_seconds: None, + } + } + /// Whether the pass sent what it could and has no wait, outage or /// credential stop to tell. fn has_nothing_left(&self) -> bool { @@ -83,6 +98,24 @@ impl DrainReport { } } +/// A drain that ended in an error, with what it settled before it: the queue +/// holds each of those verdicts already, so a caller that goes on, as a +/// folder push does after rebuilding its copy, tells them with its own. +#[derive(Debug)] +pub(crate) struct Cut { + pub error: CoreError, + pub settled: Box, +} + +impl From for Cut { + fn from(error: CoreError) -> Cut { + Cut { + error, + settled: Box::new(DrainReport::empty()), + } + } +} + #[derive(Debug, Clone, PartialEq, Serialize)] pub struct DrainVerdict { pub id: String, @@ -857,6 +890,14 @@ pub(crate) fn would_send(core: &Core) -> Result { } pub fn drain(core: &Core, stop: &AtomicBool) -> Result { + drain_keeping(core, stop).map_err(|cut| cut.error) +} + +/// `drain`, keeping what a drain that ends in an error settled before it. +pub(crate) fn drain_keeping( + core: &Core, + stop: &AtomicBool, +) -> std::result::Result { let op = telemetry::operation!("drain", answered, held, undelivered, unsent, unmade); let result = drain_checked(core, stop); // A renewal whose callback found the connection held waits in memory; @@ -868,7 +909,7 @@ pub fn drain(core: &Core, stop: &AtomicBool) -> Result { if let Ok(report) = &result { core.owe_drain(report.undelivered > 0 || report.unavailable.is_some()); } - telemetry::ended!(op, result.as_ref(), |report| { + telemetry::ended!(op, result.as_ref().map_err(|cut| &cut.error), |report| { outcome = if report.stopped.is_some() { "credential_refused" } else if report.unavailable.is_some() { @@ -885,14 +926,14 @@ pub fn drain(core: &Core, stop: &AtomicBool) -> Result { result } -fn drain_checked(core: &Core, stop: &AtomicBool) -> Result { +fn drain_checked(core: &Core, stop: &AtomicBool) -> std::result::Result { // The handle before the server, so a second opener with no server is // told the real reason it may not write. core.lock.refuse_unless_writer()?; // Whatever opened the store, a folder's queue holds only what its scan // read until the person agrees. if crate::folder::waiting(&*core.conn()?)? { - return Err(crate::error::CoreError::FirstSyncWaiting); + return Err(crate::error::CoreError::FirstSyncWaiting.into()); } // Each answer is read back under the copy's read view, so a copy without // one would send a write and then fail to settle it, on every pass. @@ -900,27 +941,29 @@ fn drain_checked(core: &Core, stop: &AtomicBool) -> Result { let result = drain_inner(core, stop); // An answer that came in is settled before a stop raised during its // request is honored, including for the last write in the queue. - crate::catch_up::unless_stopped(result, stop) + match crate::catch_up::refuse_if_stopped(stop) { + Ok(()) => result, + Err(error) => Err(Cut { + error, + settled: result.map_or_else(|cut| cut.settled, Box::new), + }), + } } -fn drain_inner(core: &Core, stop: &AtomicBool) -> Result { +fn drain_inner(core: &Core, stop: &AtomicBool) -> std::result::Result { crate::catch_up::refuse_if_stopped(stop)?; // Queued writes and their natural-key lookups name the copy's instance // so a replacement refuses them even after the pass confirmed the origin. // Credential renewal uses the exempt authentication endpoints. let (http, generation) = crate::catch_up::naming_instance(core, core.http()?)?; let mut pass = ViewChange::default(); - let result = match send_queue(core, &http, generation, stop, &mut pass) { + let mut report = DrainReport::empty(); + let result = match send_queue(core, &http, generation, stop, &mut pass, &mut report) { // From a request that settled nothing, a lookup among them: a write // the refusal answered expired the copy as it settled. Err(CoreError::CopyExpired { reason }) if reason == INSTANCE_CHANGED => { - let mut conn = core.conn()?; - let tx = conn.transaction()?; - if read_view::generation(&tx)? == generation { - crate::catch_up::expire_for_another_instance(&tx, generation)?; - } - tx.commit()?; - Err(CoreError::CopyExpired { reason }) + expire_for_another_instance(core, generation) + .and(Err(CoreError::CopyExpired { reason })) } result => result, }; @@ -929,33 +972,38 @@ fn drain_inner(core: &Core, stop: &AtomicBool) -> Result { // a report that has something to say a caller must act on (a wait, an // unavailable server, a credential stopped, a write left undelivered), and // the next call is refused `copy_expired`. - let expiry = pass.expire(core); - let report = result?; - match expiry? { - Some(expired) if report.has_nothing_left() => Err(expired), - _ => Ok(report), + let ended = match (result, pass.expire(core)) { + (Err(error), _) | (Ok(()), Err(error)) => Some(error), + (Ok(()), Ok(Some(expired))) if report.has_nothing_left() => Some(expired), + (Ok(()), Ok(_)) => None, + }; + match ended { + Some(error) => Err(Cut { + error, + settled: Box::new(report), + }), + None => Ok(report), } } +fn expire_for_another_instance(core: &Core, generation: u64) -> Result<()> { + let mut conn = core.conn()?; + let tx = conn.transaction()?; + if read_view::generation(&tx)? == generation { + crate::catch_up::expire_for_another_instance(&tx, generation)?; + } + tx.commit()?; + Ok(()) +} + fn send_queue( core: &Core, http: &Http, generation: u64, stop: &AtomicBool, pass: &mut ViewChange, -) -> Result { - let mut report = DrainReport { - answered: 0, - held: 0, - undelivered: 0, - unsent: 0, - unmade: 0, - unavailable: None, - verdicts: Vec::new(), - stopped: None, - unclaimed_sources: Vec::new(), - retry_after_seconds: None, - }; + report: &mut DrainReport, +) -> Result<()> { let waiting = store::waiting_writes(&*core.conn()?)?; let owed = !store::owed_read_backs(&*core.conn()?)?.is_empty(); let names_wait = crate::body::rule::waiting(&*core.conn()?)?; @@ -963,7 +1011,7 @@ fn send_queue( if waiting.is_empty() && !owed && !names_wait && !body_answers { // This pass has no work. Return before later reads could observe a // concurrent edit and apply its deferred work without admission. - return Ok(report); + return Ok(()); } { // A relocation must be admitted before this pass changes local queue @@ -977,8 +1025,8 @@ fn send_queue( "the server could not say which instance it is, so nothing was sent: {error}" )); report.undelivered = waiting.len(); - waited(&mut report, error.retry_after().map(|wait| wait.as_secs())); - return Ok(report); + waited(report, error.retry_after().map(|wait| wait.as_secs())); + return Ok(()); } Err(error) => return Err(error), } @@ -1019,10 +1067,10 @@ fn send_queue( }; // A server that cannot be read cannot be written to either, so nothing // is sent. - let unconfirmed = owed && !confirm(&mut report)?; + let unconfirmed = owed && !confirm(report)?; if !unconfirmed && let Some(why) = read_owed_backs(core, stop, pass)? { report.unavailable = Some(why.reason); - waited(&mut report, why.retry_after_seconds); + waited(report, why.retry_after_seconds); } // Before the queue is read, so an edge a name now resolves to goes out in @@ -1033,7 +1081,7 @@ fn send_queue( // pass: they wait for the next, and a write sent later confirms the // instance again. let (unavailable, retry_after) = (report.unavailable.clone(), report.retry_after_seconds); - if confirm(&mut report)? { + if confirm(report)? { core.settle_bodies(stop)?; } else { report.unavailable = unavailable; @@ -1069,7 +1117,7 @@ fn send_queue( if report.unavailable.is_some() && let Some(stuck) = suspect.take() { - leave_waiting(&stuck, &mut report); + leave_waiting(&stuck, report); } let mut role = Role::Walk; let mut resending: Option = None; @@ -1104,7 +1152,7 @@ fn send_queue( None => { // No write could test it, so the pass ends here. if let Some(stuck) = suspect.take() { - leave_waiting(&stuck, &mut report); + leave_waiting(&stuck, report); } } } @@ -1157,7 +1205,7 @@ fn send_queue( // since is not for a send again to decide: it waits as it was. _ if role == Role::Resend && !matches!(ready, Readiness::Ready) => { if let Some(stuck) = &resending { - leave_waiting(stuck, &mut report); + leave_waiting(stuck, report); } continue; } @@ -1200,7 +1248,7 @@ fn send_queue( report.undelivered += 1; continue; } - Readiness::RefusedWith(_) | Readiness::Ready if !confirm(&mut report)? => { + Readiness::RefusedWith(_) | Readiness::Ready if !confirm(report)? => { waiting.insert(row.id.clone()); report.undelivered += 1; continue; @@ -1226,10 +1274,10 @@ fn send_queue( tx.commit()?; } answers.insert(row.id.clone(), Some(Verdict::Refused)); - let unread = reconcile(core, row, pass)?; + // Told before the read, which can end the pass. report.unsent += 1; put_verdict( - &mut report, + report, replacing, verdict_of( row, @@ -1237,9 +1285,9 @@ fn send_queue( None, ), ); - if let Some(unread) = unread { + if let Some(unread) = reconcile(core, row, pass)? { report.unavailable = Some(unread.reason); - waited(&mut report, unread.retry_after_seconds); + waited(report, unread.retry_after_seconds); } continue; } @@ -1263,7 +1311,7 @@ fn send_queue( row.id, unavailable_by(&error).unwrap_or_else(|| error.to_string()) )); - waited(&mut report, error.retry_after().map(|wait| wait.as_secs())); + waited(report, error.retry_after().map(|wait| wait.as_secs())); waiting.insert(row.id.clone()); report.undelivered += 1; continue; @@ -1299,7 +1347,7 @@ fn send_queue( answers.insert(row.id.clone(), Some(Verdict::Refused)); report.unsent += 1; put_verdict( - &mut report, + report, replacing, verdict_of( row, @@ -1314,7 +1362,7 @@ fn send_queue( Err(error) => { report.undelivered += 1; put_verdict( - &mut report, + report, replacing, verdict_of( row, @@ -1410,34 +1458,56 @@ fn send_queue( answer => answer, }; if let Ok(answer) = &answer { - waited(&mut report, answer.retry_after_seconds); + waited(report, answer.retry_after_seconds); } let class = refine(row, &payload, &answer, classify(&answer)); let environmental = class == Classified::Environmental; + let landed = matches!(class, Classified::Landed { .. } | Classified::Delivered); let taken = match store_fault(row) { Some(error) => Err(error), None => settle(core, row, &answer, class, shape, pass), }; let settled = match taken { Ok(settled) => settled, - Err( - error @ (CoreError::Redirected { .. } - | CoreError::ContractMismatch { .. } - | CoreError::RenewalFailed(_) - | CoreError::SignedOut { .. } - | CoreError::NoKeychain(_) - | CoreError::StorageFull(_) - | CoreError::CopyExpired { .. } - | CoreError::StreamIncomplete { .. }), - ) => return Err(error), - // The server answered and the copy could not take the answer: a - // further attempt may clear it, nothing clears it on its own. Err(error) => { - if store::queued_write(&*core.conn()?, &row.id)? - .is_some_and(|current| current.verdict.is_some()) + // A verdict is committed before the read it owes, and that + // read can end the pass, as a changed read view does. The + // drain began by clearing every verdict a row is sent with, so + // one held now is this answer's, and is told. + if let Some(current) = store::queued_write(&*core.conn()?, &row.id)? + && current.verdict.is_some() { + if answer.is_ok() { + report.answered += 1; + } else { + report.unmade += 1; + } + put_verdict( + report, + replacing, + verdict_of( + ¤t, + &committed(¤t, answer.as_ref().ok(), shape, landed), + current.answer.as_deref(), + ), + ); + return Err(error); + } + if matches!( + error, + CoreError::Redirected { .. } + | CoreError::ContractMismatch { .. } + | CoreError::RenewalFailed(_) + | CoreError::SignedOut { .. } + | CoreError::NoKeychain(_) + | CoreError::StorageFull(_) + | CoreError::CopyExpired { .. } + | CoreError::StreamIncomplete { .. } + ) { return Err(error); } + // The server answered and the copy could not take the answer: + // a further attempt may clear it, nothing clears it on its own. let refusals = store::count_refusal(&*core.conn()?, &row.id)?; let mut settled = finish_counted( core, @@ -1481,7 +1551,7 @@ fn send_queue( } else if answer.is_err() && settled.unavailable.is_none() { report.unmade += 1; } - waited(&mut report, settled.retry_after_seconds); + waited(report, settled.retry_after_seconds); answers.insert(row.id.clone(), settled.verdict); if row.kind == WriteKind::CreateItem && let Some(target) = store::meta_get(&*core.conn()?, &receipt_key(&row.id))? @@ -1522,7 +1592,7 @@ fn send_queue( }; let answered = settled_row.as_ref().unwrap_or(row); put_verdict( - &mut report, + report, replacing, verdict_of(answered, &settled, answered.answer.as_deref()), ); @@ -1552,7 +1622,7 @@ fn send_queue( // Sent again after the test landed and met the failure again: set // aside, and the pass goes on. if let (Some(failure), Some(stuck)) = (&met_again, &resending) { - set_aside(core, stuck, failure, &mut report, &mut answers)?; + set_aside(core, stuck, failure, report, &mut answers)?; carry_on = true; } if stop { @@ -1587,7 +1657,7 @@ fn send_queue( } crate::body::rule::after_answers(&mut *core.conn()?)?; - Ok(report) + Ok(()) } /// What a lookup this pass said of a natural key. @@ -2200,6 +2270,36 @@ fn verdict_of(row: &QueuedWrite, settled: &Settled, answer: Option<&str>) -> Dra } } +/// The verdict the queue holds for a write whose settling ended the pass +/// after it was committed, with what its answer adds. A write `landed`, on a +/// natural key or a receipt the server no longer keeps whole, was replayed. +fn committed(row: &QueuedWrite, answer: Option<&Answer>, shape: Shape, landed: bool) -> Settled { + let item = answer + .filter(|_| shape == Shape::Item) + .and_then(|answer| serde_json::from_str::(&answer.body).ok()); + let edge = answer + .filter(|_| shape == Shape::Edge) + .and_then(|answer| serde_json::from_str::(&answer.body).ok()); + let succeeded = matches!( + row.verdict, + Some(Verdict::Accepted | Verdict::Merged | Verdict::Conflicted) + ); + Settled { + conflicted_copy_id: row.conflicted_copy_id.clone(), + replayed: succeeded + && (landed + || answer.is_some_and(|answer| answer.replayed) + || item.as_ref().is_some_and(|item| item.acknowledged) + || edge.as_ref().is_some_and(|edge| edge.acknowledged)), + merged_fields: item + .and_then(|item| item.conflict_resolution) + .filter(|_| succeeded) + .map(|resolution| resolution.fields) + .unwrap_or_default(), + ..Settled::plain(row.verdict, row.reason.clone(), row.refusals) + } +} + struct Settled { verdict: Option, reason: Option, @@ -4279,6 +4379,106 @@ mod tests { assert!(read_view::expired(&conn).unwrap()); } + #[test] + fn a_verdict_whose_own_read_meets_a_changed_view_is_told_with_the_cut() { + // The witness: the same answers with the read let through end well + // and tell the verdict as any drain does. + for (verdict, answer) in [ + (Verdict::Accepted, edge_answered("e1", "p1", 2)), + ( + Verdict::Refused, + crate::scripted::refusal(422, "validation_failed"), + ), + ] { + for changed in [false, true] { + let server = crate::scripted::Scripted::start(); + let (_dir, core) = holding_edges(&server, &["e1"]); + let row = rank_edit(&core, "e1"); + let read = if changed { + crate::scripted::refusal(409, "read_view_changed") + } else { + crate::scripted::certified(edge_answered( + "e1", + "p1", + 1 + i64::from(verdict == Verdict::Accepted), + )) + }; + server.on("/edges/e1", vec![answer.clone(), read]); + let report = match drain_keeping(&core, &AtomicBool::new(false)) { + Ok(report) => { + assert!( + !changed, + "{verdict:?}: the changed view did not end the drain" + ); + report + } + Err(cut) => { + assert!(changed, "{verdict:?}: {:?}", cut.error); + assert_eq!(cut.error, view_changed()); + *cut.settled + } + }; + assert_eq!( + store::queued_write(&core.conn().unwrap(), &row.id) + .unwrap() + .unwrap() + .verdict, + Some(verdict) + ); + assert_eq!( + report.answered, 1, + "{verdict:?}, view changed {changed}: {report:?}" + ); + assert_eq!( + report + .verdicts + .iter() + .map(|entry| (entry.id.as_str(), entry.verdict)) + .collect::>(), + vec![(row.id.as_str(), Some(verdict))], + "{verdict:?}, view changed {changed}: the verdict the queue holds left the report" + ); + if verdict == Verdict::Refused { + assert_eq!( + report.verdicts[0] + .refusal + .as_ref() + .and_then(|refusal| refusal.code.as_deref()), + Some("validation_failed") + ); + } + } + } + } + + #[test] + fn a_write_refused_unsent_whose_read_meets_a_changed_view_is_told_with_the_cut() { + let server = crate::scripted::Scripted::start(); + server.on( + "/items/a", + vec![crate::scripted::refusal(409, "read_view_changed")], + ); + let (_dir, core) = deleting(&server, &["a"]); + let row = core.queue().unwrap().pop().unwrap(); + // A delete with no server version is refused before it is sent. + core.conn() + .unwrap() + .execute("UPDATE queue SET base_version = 0 WHERE id = ?1", [&row.id]) + .unwrap(); + let cut = drain_keeping(&core, &AtomicBool::new(false)).unwrap_err(); + assert_eq!(cut.error, view_changed()); + // The witness: the read the refusal owes was made, and met the change. + assert_eq!(server.seen("/items/a").len(), 1); + assert_eq!( + (cut.settled.unsent, cut.settled.verdicts.len()), + (1, 1), + "{:?}", + cut.settled + ); + assert_eq!(cut.settled.verdicts[0].id, row.id); + assert_eq!(cut.settled.verdicts[0].verdict, Some(Verdict::Refused)); + } + fn instance_changed() -> CoreError { CoreError::CopyExpired { reason: INSTANCE_CHANGED.into(), diff --git a/core/marfa-core/src/folder/mod.rs b/core/marfa-core/src/folder/mod.rs index 8c7fc1d75..3a5d745f0 100644 --- a/core/marfa-core/src/folder/mod.rs +++ b/core/marfa-core/src/folder/mod.rs @@ -62,6 +62,7 @@ use serde_json::{Map, Value}; use crate::catalog::Catalog; use crate::directory::Directory; use crate::directory::{Mode, OFlags}; +use crate::drain::Cut; use crate::error::CoreError; use crate::model::{BlockedReason, Draft, Edit, Item, ItemState, WriteKind}; use crate::{Core, Result, Server}; @@ -1590,26 +1591,85 @@ impl Folder { /// `drain`, for a watch where `watched`, which waits out /// `placement::GIVE_WAY_REREAD` before it reads again for a give-way. pub(super) fn drain_in(&self, watched: bool) -> Result { + self.drain_keeping(watched).map_err(|cut| cut.error) + } + + /// `drain_in`, keeping what a drain that ends in an error settled before + /// it. The refusals among them are remembered and their files held, as + /// for a drain that ends well; nothing is read from the server for them, + /// since the error may say the copy can no longer be read from. + pub(super) fn drain_keeping( + &self, + watched: bool, + ) -> std::result::Result> { let one = self.core.one_drain(); - let mut report = self.core.drain_held(&one)?; + let mut report = crate::DrainReport::empty(); let mut rebased = 0; - while report.stopped.is_none() && report.unavailable.is_none() { - let now = self.rebase_thinned()? + self.make_edges_gone_before_their_move()?; - if now == 0 { - break; - } - rebased += now; - report.absorb(self.core.drain_held(&one)?); - } - // Queued even where the drain stopped, so the copy holds the edge the line asks for. - self.make_edges_gone_before_their_move()?; - let gave_way = self.settle_placements(&mut report, watched)?; - self.hold_refused()?; - Ok(Drained { + let ended = self.drain_rebasing(&one, &mut report, &mut rebased).err(); + if let Some(drain_error) = ended { + // A refusal that could not be remembered is told in place of the + // drain's own error, since a caller that went on would lose it. + let error = self + .remember_refusals(&report) + .and_then(|()| self.hold_refused()) + .err() + .unwrap_or(drain_error); + return Err(Cut { + error, + settled: Box::new(Drained { + report, + rebased, + gave_way: 0, + }), + }); + } + let settled = (|| { + // Queued even where the drain stopped, so the copy holds the edge the line asks for. + self.make_edges_gone_before_their_move()?; + let gave_way = self.settle_placements(&mut report, watched)?; + self.keep_wait(&report)?; + self.hold_refused()?; + Ok(gave_way) + })(); + let drained = |report, gave_way| Drained { report, rebased, gave_way, - }) + }; + match settled { + Ok(gave_way) => Ok(drained(report, gave_way)), + Err(error) => Err(Cut { + error, + settled: Box::new(drained(report, 0)), + }), + } + } + + /// Drains, and again for as long as the edits the server could not + /// apply are rebased, taking each drain's report into `report`. + fn drain_rebasing( + &self, + one: &std::sync::MutexGuard<'_, ()>, + report: &mut crate::DrainReport, + rebased: &mut usize, + ) -> Result<()> { + loop { + match self.core.drain_held(one) { + Ok(later) => report.absorb(later), + Err(cut) => { + report.absorb(*cut.settled); + return Err(cut.error); + } + } + if report.stopped.is_some() || report.unavailable.is_some() { + return Ok(()); + } + let now = self.rebase_thinned()? + self.make_edges_gone_before_their_move()?; + if now == 0 { + return Ok(()); + } + *rebased += now; + } } /// Read from the queue rather than this drain's report, so a refusal a plain diff --git a/core/marfa-core/src/folder/placement.rs b/core/marfa-core/src/folder/placement.rs index e1f51883a..b4bbb3624 100644 --- a/core/marfa-core/src/folder/placement.rs +++ b/core/marfa-core/src/folder/placement.rs @@ -243,12 +243,19 @@ impl Folder { } } + /// The version of the settings the copy holds, read whether or not the + /// copy can be read: a drain that a changed read view ends remembers its + /// refusals before the copy is built again. + fn settings_version(&self) -> Result> { + store::held_version(&*self.core.conn()?, &self.folder) + } + fn refused(&self) -> Result> { let Some(json) = store::meta_get(&*self.core.conn()?, META_REFUSED)? else { return Ok(None); }; let refused: Option = serde_json::from_str(&json).ok(); - let settings = self.core.get(&self.folder)?.map(|row| row.version); + let settings = self.settings_version()?; let Some(mut refused) = refused.filter(|refused| Some(refused.settings) == settings) else { store::meta_delete(&*self.core.conn()?, META_REFUSED)?; return Ok(None); @@ -304,6 +311,20 @@ impl Folder { report: &mut DrainReport, watched: bool, ) -> Result { + self.remember_refusals(report)?; + if report.unavailable.is_some() || report.stopped.is_some() { + return Ok(0); + } + let (gave_way, withdrawn) = self.give_way(watched)?; + report + .verdicts + .retain(|verdict| !withdrawn.contains(&verdict.id)); + Ok(gave_way) + } + + /// Remembers each placement the server refused among the verdicts a + /// drain reports, and reads nothing. + pub(super) fn remember_refusals(&self, report: &DrainReport) -> Result<()> { let mut refused: Vec<(String, Held)> = Vec::new(); for verdict in &report.verdicts { let row = { @@ -343,14 +364,7 @@ impl Folder { if !refused.is_empty() { self.remember_refused(refused)?; } - if report.unavailable.is_some() || report.stopped.is_some() { - return Ok(0); - } - let (gave_way, withdrawn) = self.give_way(watched)?; - report - .verdicts - .retain(|verdict| !withdrawn.contains(&verdict.id)); - Ok(gave_way) + Ok(()) } /// Gives way from the latest of this folder's placement writes to each @@ -516,7 +530,7 @@ impl Folder { Some(record) => record, None => Refused { key: self.key_state().ok(), - settings: self.core.get(&self.folder)?.map_or(0, |row| row.version), + settings: self.settings_version()?.unwrap_or(0), placements: Withheld::new(), }, }; diff --git a/core/marfa-core/src/folder/pull.rs b/core/marfa-core/src/folder/pull.rs index 16fb328dc..3cd785b93 100644 --- a/core/marfa-core/src/folder/pull.rs +++ b/core/marfa-core/src/folder/pull.rs @@ -145,7 +145,7 @@ enum NotWritten { } impl PullReport { - fn add_writes(&mut self, other: &PullReport) { + pub(super) fn add_writes(&mut self, other: &PullReport) { self.written += other.written; self.rewritten += other.rewritten; self.moved += other.moved; diff --git a/core/marfa-core/src/folder/sync.rs b/core/marfa-core/src/folder/sync.rs index e1f6d44a1..f197f5451 100644 --- a/core/marfa-core/src/folder/sync.rs +++ b/core/marfa-core/src/folder/sync.rs @@ -8,6 +8,7 @@ use super::{ }; use crate::Result; use crate::directory::Directory; +use crate::drain::Cut; use crate::error::CoreError; use crate::model::HydrateReport; @@ -21,11 +22,16 @@ pub struct SyncReport { /// Every write the sync sent, those the pull queued after the first /// drain included. pub drain: Drained, + /// The hydration the drain took where the server said the copy's read + /// view had changed, before it sent the writes still waiting. + pub rebuilt: Option, /// Why the catch-up could not reach the server, where it could not; the /// sync goes on to write out the copy it holds. pub catch_up: std::result::Result, /// `None` where the catch-up failed after the copy expired, which leaves - /// nothing to pull from until a later sync hydrates. + /// nothing to pull from until a later sync hydrates. Where the sync + /// pulled again after a rebuild, the later pull's, with the files both + /// wrote. pub pull: Option, } @@ -37,13 +43,16 @@ pub struct SendReport { pub settings: SettingsFileReport, pub scan: ScanReport, pub drain: Drained, + /// As for `SyncReport`. + pub rebuilt: Option, } impl Folder { /// Everything a folder does, once: resumes an unfinished hydration, sends /// an edit of the settings file, scans, drains, catches up, pulls, and /// drains again where the pull queued the placement of a file it wrote, or - /// the end of one. + /// the end of one. Where that drain rebuilt the copy for a changed read + /// view, pulls once more, and drains again where that pull placed a file. /// A folder whose first sync waits to be confirmed is only read, and says /// what the sync will do. pub fn sync(&self) -> Result { @@ -117,12 +126,14 @@ impl Folder { let hydrated = self.resume()?; let settings = self.send_settings_edit()?; let scan = self.scan_keeping_files()?; - let drain = self.drain()?; + let mut rebuilt = None; + let drain = self.drain_rebuilding(&mut rebuilt)?; Ok(SendReport { hydrated, settings, scan, drain, + rebuilt, }) } @@ -141,34 +152,89 @@ impl Folder { // First, so the rest of the sync works on the new settings. let settings = self.send_settings_edit()?; let scan = self.scan()?; - let mut drain = self.drain()?; + let mut rebuilt = None; + let mut drain = self.drain_rebuilding(&mut rebuilt)?; let catch_up = match self.catch_up() { Ok(caught) => Ok(caught), Err(error) if error.is_environmental() => Err(error), Err(error) => return Err(error), }; - let pull = match self.pull() { + let mut pull = match self.pull() { Ok(pulled) => Some(pulled), Err(CoreError::HydrationIncomplete) if catch_up.is_err() => None, Err(error) => return Err(error), }; - if sends_again(catch_up.is_ok(), &drain.report) - && pull - .as_ref() - .is_some_and(|pulled| pulled.placed + pulled.ended > 0) - { - drain.absorb(self.drain()?); + if sends_again(catch_up.is_ok(), &drain.report) && pull.as_ref().is_some_and(places) { + let rebuilt_before = rebuilt.is_some(); + let later = self.drain_rebuilding(&mut rebuilt)?; + self.absorb(&mut drain, later)?; + // The rebuild read the copy under the new view after the pull + // wrote it out, so the directory is brought to it once more. The + // run has rebuilt already, so a change at the drain after this + // pull ends it. + if !rebuilt_before && rebuilt.is_some() { + let mut again = self.pull()?; + let places_again = places(&again); + if let Some(first) = &pull { + again.add_writes(first); + } + if sends_again(true, &drain.report) && places_again { + let last = self.drain_rebuilding(&mut rebuilt)?; + self.absorb(&mut drain, last)?; + } + pull = Some(again); + } } Ok(SyncReport { hydrated, settings, scan, drain, + rebuilt, catch_up, pull, }) } + /// A drain that the server ends by saying the copy's read view changed, + /// as it does once the folder's key is replaced or narrowed, builds the + /// copy again into `rebuilt` and drains once more, so the writes still + /// waiting go out in this run rather than the next. What the first drain + /// settled is told with what the second sent. Once a run: where `rebuilt` + /// holds a rebuild already, a change ends the run. A rebuild that fails + /// ends it with its own error, the queue kept. + fn drain_rebuilding(&self, rebuilt: &mut Option) -> Result { + match self.drain_keeping(false) { + Ok(drained) => Ok(drained), + Err(Cut { + error: CoreError::CopyExpired { reason }, + settled, + }) if reason == "read_view_changed" && rebuilt.is_none() => { + *rebuilt = Some(self.hydrate()?); + let mut drained = *settled; + self.absorb(&mut drained, self.drain()?)?; + Ok(drained) + } + Err(cut) => Err(cut.error), + } + } + + /// Takes `later`, a later drain of the same run, into `drained`. A write + /// that drain gave way from has left the queue, and the verdict an + /// earlier drain told of it goes with it. + fn absorb(&self, drained: &mut Drained, later: Drained) -> Result<()> { + drained.absorb(later); + let conn = self.core.conn()?; + let mut kept = Vec::with_capacity(drained.report.verdicts.len()); + for verdict in std::mem::take(&mut drained.report.verdicts) { + if crate::store::queued_write(&conn, &verdict.id)?.is_some() { + kept.push(verdict); + } + } + drained.report.verdicts = kept; + Ok(()) + } + /// Takes the folder at `dir` off this machine: its own state under /// `.marfa` goes, and its files stay. Refused while writes wait, unless /// the first sync still waits to be confirmed. No symbolic link is @@ -305,6 +371,11 @@ fn stopped_by(drain: &crate::DrainReport) -> Option { }) } +/// Whether a pull queued the placement of a file it wrote, or the end of one. +fn places(pulled: &PullReport) -> bool { + pulled.placed + pulled.ended > 0 +} + /// Not where the first drain or the catch-up shows the server cannot take /// writes (unreachable, credential refused, a write left undelivered): the /// report says why, and the placements wait for the next sync. @@ -331,6 +402,68 @@ mod tests { } } + #[test] + fn a_verdict_an_earlier_drain_told_leaves_with_a_write_a_later_drain_withdrew() { + let dir = tempfile::tempdir_in(std::env::temp_dir().canonicalize().unwrap()).unwrap(); + let db = dir.path().join(STATE_DIR).join("core.sqlite"); + std::fs::create_dir_all(db.parent().unwrap()).unwrap(); + let core = super::super::working(crate::Core::open(&db, None).unwrap()).unwrap(); + super::super::settings_file::bind(&core, "folder").unwrap(); + drop(core); + let folder = Folder::open( + dir.path(), + Some(crate::Server { + url: "http://127.0.0.1:9".into(), + key: "fixture".into(), + }), + ) + .unwrap(); + let queued = folder + .core + .create_item(&crate::model::Draft { + r#type: "core.note".into(), + properties: serde_json::json!({ "title": "Kept", "body": "" }) + .as_object() + .unwrap() + .clone(), + ..Default::default() + }) + .unwrap(); + let told = |id: &str| crate::drain::DrainVerdict { + id: id.into(), + kind: crate::model::WriteKind::CreateEdge, + item_id: None, + edge_id: None, + verdict: Some(crate::model::Verdict::Refused), + reason: Some("edge_duplicate".into()), + refusal: None, + conflicted_copy_id: None, + refusals: 0, + replayed: false, + merged_fields: Vec::new(), + }; + let drained = |verdicts| Drained { + report: crate::DrainReport { + verdicts, + ..drained() + }, + rebased: 0, + gave_way: 0, + }; + let mut first = drained(vec![told(&queued.id), told("withdrawn")]); + folder.absorb(&mut first, drained(Vec::new())).unwrap(); + assert_eq!( + first + .report + .verdicts + .iter() + .map(|verdict| verdict.id.as_str()) + .collect::>(), + vec![queued.id.as_str()], + "a write no longer queued kept its verdict, or a queued one lost it" + ); + } + #[test] fn a_second_drain_goes_only_where_nothing_showed_the_server_unable_to_take_writes() { assert!(sends_again(true, &drained()), "the witness: it goes"); diff --git a/core/marfa-core/src/folder/watch.rs b/core/marfa-core/src/folder/watch.rs index 069553973..50188ea2d 100644 --- a/core/marfa-core/src/folder/watch.rs +++ b/core/marfa-core/src/folder/watch.rs @@ -259,7 +259,7 @@ impl Folder { let mut gate = Gate::new(crate::clock::now()); let mut telling = Telling::default(); let mut last_full: Option = None; - let mut send_again_at: Option = None; + let mut send_again_at = self.kept_wait(); let mut lists = self.settings().and_then(|settings| settings.lists()).ok(); loop { if stop.load(Ordering::SeqCst) { @@ -564,6 +564,49 @@ fn look(mut requests: Requests, ended: &AtomicBool, asked: &AtomicBool, wakes: m } } +/// Where the folder's store keeps, in milliseconds since the Unix epoch on the +/// engine's wall clock, the time before which the server asked for no write. +/// Kept in the store rather than the watch, so a watch that hands its folder +/// to a command, and is started again, waits out what the server last said. +const SEND_AGAIN_AT: &str = "folder_send_again_at"; + +impl Folder { + /// Keeps the wait a drain of this folder was told, whoever drained: a + /// later answer replaces an earlier one, and a later drain with no wait + /// ends it where the server took writes or writes were left undelivered + /// (`folders/watch-rate-limit-handover`). + pub(super) fn keep_wait(&self, drain: &crate::DrainReport) -> Result<(), CoreError> { + let conn = self.core.conn()?; + if let Some(wait) = rate_limit_wait(drain) { + let until = crate::clock::system_time() + wait; + let millis = until + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |since| since.as_millis()); + crate::store::meta_set(&conn, SEND_AGAIN_AT, &millis.to_string()) + } else if drain.retry_after_seconds.is_none() + && (drain.answered > 0 || drain.undelivered > 0) + { + crate::store::meta_delete(&conn, SEND_AGAIN_AT) + } else { + Ok(()) + } + } + + /// When the wait the store keeps ends on the clock the watch runs by. + /// Never further off than the longest wait a drain keeps, so a wall + /// clock set back cannot hold the watch. + fn kept_wait(&self) -> Option { + let conn = self.core.conn().ok()?; + let millis: u64 = crate::store::meta_get(&conn, SEND_AGAIN_AT) + .ok()?? + .parse() + .ok()?; + let until = std::time::UNIX_EPOCH + Duration::from_millis(millis); + let left = until.duration_since(crate::clock::system_time()).ok()?; + Some(crate::clock::now() + left.min(crate::error::RETRY_AFTER_MOST)) + } +} + /// How long the server asked the watch to leave its writes alone, where a /// pass left some undelivered: no pass runs until it is over. fn rate_limit_wait(drain: &crate::DrainReport) -> Option { @@ -989,6 +1032,53 @@ mod tests { assert!(full_due(Some(now), now + FULL_PASS)); } + #[test] + fn a_later_drain_that_leaves_writes_undelivered_with_no_wait_ends_the_kept_wait() { + let dir = tempfile::tempdir_in(std::env::temp_dir().canonicalize().unwrap()).unwrap(); + let db = dir.path().join(super::super::STATE_DIR).join("core.sqlite"); + std::fs::create_dir_all(db.parent().unwrap()).unwrap(); + let core = super::super::working(crate::Core::open(&db, None).unwrap()).unwrap(); + super::super::settings_file::bind(&core, "folder").unwrap(); + drop(core); + let folder = Folder::open( + dir.path(), + Some(crate::Server { + url: "http://127.0.0.1:9".into(), + key: "fixture".into(), + }), + ) + .unwrap(); + let limited = || { + let mut report = drained().report; + report.undelivered = 2; + report.retry_after_seconds = Some(30); + report + }; + folder.keep_wait(&limited()).unwrap(); + assert!( + folder.kept_wait().is_some(), + "a rate limit's wait was not kept" + ); + // A drain that sent nothing says nothing of the wait. + folder.keep_wait(&drained().report).unwrap(); + assert!(folder.kept_wait().is_some()); + + let mut unavailable = drained().report; + unavailable.undelivered = 1; + unavailable.unavailable = Some("the server could not be reached".into()); + folder.keep_wait(&unavailable).unwrap(); + assert!( + folder.kept_wait().is_none(), + "a later drain left writes undelivered with no wait, and the older wait was kept" + ); + + folder.keep_wait(&limited()).unwrap(); + let mut answered = drained().report; + answered.answered = 1; + folder.keep_wait(&answered).unwrap(); + assert!(folder.kept_wait().is_none()); + } + fn drained() -> Drained { Drained { report: crate::DrainReport { diff --git a/core/marfa-core/src/lib.rs b/core/marfa-core/src/lib.rs index 4fe397f55..a9a1b2e66 100644 --- a/core/marfa-core/src/lib.rs +++ b/core/marfa-core/src/lib.rs @@ -1225,8 +1225,12 @@ impl Core { .unwrap_or_else(|poisoned| poisoned.into_inner()) } - pub(crate) fn drain_held(&self, _one: &MutexGuard<'_, ()>) -> Result { - drain::drain(self, &NEVER_STOPPED) + /// A drain that ends in an error keeps what it settled before it. + pub(crate) fn drain_held( + &self, + _one: &MutexGuard<'_, ()>, + ) -> std::result::Result { + drain::drain_keeping(self, &NEVER_STOPPED) } /// Under a fresh idempotency key. Answers `false` for a row that is not