This commit is contained in:
Andras Schmelczer 2026-09-27 08:29:17 +01:00
commit 3dab48f58b
14 changed files with 1141 additions and 1376 deletions

View file

@ -124,7 +124,7 @@ export async function assertCanonical(
do {
const batch = await getJson<{
headEventId: number;
endEventId?: number;
endEventId: number;
events: { eventId: number; requestId: string }[];
}>(`${vaultUrl}/events-since?after=${after}`, token);
assert.equal(
@ -132,7 +132,7 @@ export async function assertCanonical(
snapshot.headEventId,
"Server changed while checking convergence"
);
const end = batch.endEventId ?? batch.headEventId;
const end = batch.endEventId;
assert.equal(
end,
batch.events.at(-1)?.eventId ?? after,

View file

@ -75,7 +75,7 @@ test("stale content acknowledgements contain metadata without the excluded paylo
test("highly fragmented local edits use a snapshot within the server operation budget", async () => {
const { fixture } = await import("./sync-fixture");
const { toStoredSnapshot } = await import("../sync-client/src/snapshot");
const { hashSnapshot } = await import("../sync-client/src/snapshot");
const f = await fixture({});
f.contentCache.resize(1024 * 1024);
const original = Array.from({ length: 12_000 }, (_, i) => `word${i}`).join(
@ -95,7 +95,7 @@ test("highly fragmented local edits use a snapshot within the server operation b
const request = await syncer.pushContent(
"a.md",
1,
await toStoredSnapshot({ content: bytes(edited) })
await hashSnapshot({ content: bytes(edited) })
);
assert.equal(request.type, "Snapshot");
});

View file

@ -4,7 +4,7 @@ import { test } from "node:test";
import { Database, emptyState } from "../sync-client/src/persistence/database";
import { Syncer } from "../sync-client/src/sync-operations/syncer";
import { SyncClient, type StoredClient } from "../sync-client/src/sync-client";
import { toStoredSnapshot } from "../sync-client/src/snapshot";
import { hashSnapshot } from "../sync-client/src/snapshot";
import { allocatePortablePath } from "../sync-client/src/utils/portable-path";
import { MemoryDisk, MemoryPersistence } from "./storage";
@ -18,7 +18,7 @@ test("offline content notifications need no saves and restart reads the current
initial.local = { a: "a.md" };
initial.documents.a = {
observedHash: (await toStoredSnapshot({ content: bytes("original") }))
observedHash: (await hashSnapshot({ content: bytes("original") }))
.hash
};
const store = new MemoryPersistence({
@ -67,7 +67,7 @@ test("offline content notifications need no saves and restart reads the current
}
});
test("legacy content notifications preserve the acknowledged namespace prefix on restart", async () => {
test("namespace notifications preserve the acknowledged prefix on restart", async () => {
const disk = new MemoryDisk();
await disk.userWrite("d.md", bytes("A"));
await disk.userWrite("b.md", bytes("B"));
@ -81,10 +81,10 @@ test("legacy content notifications preserve the acknowledged namespace prefix on
])
initial.documents[id] = {
observedHash: (await toStoredSnapshot({ content: bytes(content) }))
observedHash: (await hashSnapshot({ content: bytes(content) }))
.hash
};
initial.lastAppliedLocalChangeId = "consumed-update";
initial.lastAppliedLocalChangeId = "consumed-move";
const saved: StoredClient = {
database: initial,
settings: { remoteUri: "http://offline.test", vaultName: "test" },
@ -97,8 +97,6 @@ test("legacy content notifications preserve the acknowledged namespace prefix on
identities: { a: "a.md" },
changeId: "consumed-move"
},
{ type: "update", path: "c.md", changeId: "consumed-update" },
{ type: "update", path: "b.md", changeId: "new-update" },
{
type: "move",
oldPath: "c.md",
@ -132,7 +130,7 @@ test("legacy content notifications preserve the acknowledged namespace prefix on
test("remote binary replacement and deletion do not create backup artifacts", async () => {
const f = await fixture({ "a.bin": "private edit" });
await f.files.apply(structuredClone(f.database.state), {
a: { replacement: await toStoredSnapshot({ content: bytes("remote") }) }
a: { replacement: await hashSnapshot({ content: bytes("remote") }) }
});
assert.deepEqual(await f.disk.listFilesRecursively(), ["a.bin"]);
assert.deepEqual(f.disk.userFiles(), new Map([["a.bin", bytes("remote")]]));
@ -155,7 +153,7 @@ test("rejected content recovery remains idempotent across a rename and network f
const snapshot = (await f.files.snapshot("a.md"))!;
f.database.state.documents.a.base = {
...head("a", 1, "base"),
hash: (await toStoredSnapshot({ content: bytes("base") })).hash
hash: (await hashSnapshot({ content: bytes("base") })).hash
};
f.database.state.pending = {
type: "content",
@ -164,7 +162,7 @@ test("rejected content recovery remains idempotent across a rename and network f
request: {
requestId: "request",
parentVersionId: 1,
content: { type: "Snapshot", value: snapshot.contentBase64 }
content: { type: "Snapshot", value: Buffer.from(snapshot.content).toString("base64") }
}
};
await f.database.save();
@ -253,38 +251,55 @@ test("unfinished bootstrap cannot be rebound to another vault", async () => {
}
});
test("a rejected upload is retained locally and does not block another file or a corrected snapshot", async () => {
test("a rejected upload pauses all sync across restarts until an explicit retry", async (context) => {
let reject = true;
const sent: string[] = [];
const f = await fixture(
{ "a.md": "too large", "b.md": "other" },
{
putFileContent: async (id) => {
sent.push(id);
if (id === "a" && reject)
throw new PermanentSyncError("HTTP 413");
return {
type: "Accepted",
...head(id, sent.length + 2, "small")
};
}
let eventReads = 0;
const sent: { id: string; request: unknown }[] = [];
const service = {
getEvents: async () => {
eventReads++;
return { headEventId: 0, endEventId: 0, events: [] };
},
putFileContent: async (id: string, request: unknown) => {
sent.push({ id, request: structuredClone(request) });
if (reject) throw new PermanentSyncError("HTTP 413");
return { type: "Accepted" as const, ...head(id, sent.length + 2, "small") };
}
);
f.internals.stopped = false;
assert(await f.internals.preparePush());
await f.internals.finishPending();
assert.equal(f.database.state.pending, undefined);
assert.match(f.database.state.unconfirmed![0].rejection!, /413/);
assert(await f.internals.preparePush());
await f.internals.finishPending();
assert.deepEqual(sent, ["a", "b"]);
assert.equal(await f.internals.preparePush(), false);
};
let f = await fixture({ "a.md": "too large", "b.md": "other" }, service);
context.after(async () => f.syncer.stop());
for (const doc of Object.values(f.database.state.documents))
doc.base = { vaultUpdateId: 1, hash: "previous" };
await f.database.save();
f.syncer.start();
await assert.rejects(f.syncer.waitUntilFinished(), /413/);
const pending = structuredClone(f.database.state.pending!);
assert.equal(pending.rejection, "HTTP 413");
assert.deepEqual(sent.map(({ id }) => id), ["a"]);
const reads = eventReads;
await f.disk.userWrite("a.md", bytes("small"));
await f.syncer.syncLocallyUpdatedFile({ relativePath: "a.md" });
f.syncer.wake();
await assert.rejects(f.syncer.waitUntilFinished(), /413/);
assert.equal(eventReads, reads);
assert.equal(sent.length, 1);
await f.syncer.stop();
f = await fixture({}, service, f);
f.syncer.start();
await assert.rejects(f.syncer.waitUntilFinished(), /413/);
assert.equal(eventReads, reads);
assert.equal(sent.length, 1);
reject = false;
assert(await f.internals.preparePush());
await f.internals.finishPending();
assert.deepEqual(sent, ["a", "b", "a"]);
assert.equal(f.database.state.unconfirmed?.find((request) => request.rejection), undefined);
await f.syncer.retryRejectedRequests();
f.syncer.wake();
await f.syncer.waitUntilFinished();
assert.deepEqual(sent.map(({ id }) => id), ["a", "a", "a", "b"]);
assert.deepEqual(sent[1], sent[0], "resolve the old request before uploading corrected bytes");
assert.notDeepEqual(sent[2].request, sent[0].request);
assert.equal(f.database.state.pending, undefined);
assert.deepEqual((await f.disk.read("a.md"))?.content, bytes("small"));
await f.syncer.stop();
});
test("a transient upload failure retries the exact request and recovers its acknowledgement", async () => {
@ -304,11 +319,10 @@ test("a transient upload failure retries the exact request and recovers its ackn
assert.equal(requests.length, 2);
assert.deepEqual(requests[1], requests[0]);
assert.equal(f.database.state.documents.a.base!.vaultUpdateId, 2);
assert.equal(f.database.state.unconfirmed?.find((request) => request.rejection), undefined);
assert.equal(f.database.state.pending, undefined);
});
test("a persisted upload rejection finishes without filesystem writes", async () => {
test("a persisted upload rejection pauses without filesystem writes", async () => {
let uploads = 0;
const f = await fixture(undefined, {
putFileContent: async () => {
@ -322,41 +336,28 @@ test("a persisted upload rejection finishes without filesystem writes", async ()
if (!label.startsWith("read:"))
throw new Error("Unexpected filesystem mutation");
};
await f.internals.finishPending();
await assert.rejects(f.internals.finishPending(), /413/);
assert.equal(uploads, 1);
assert.equal(f.database.state.pending, undefined);
assert.equal(f.database.state.unconfirmed![0].rejection, "HTTP 413");
assert.equal(f.database.state.pending?.rejection, "HTTP 413");
assert.deepEqual(f.disk.userFiles(), new Map([["a.md", bytes("local")]]));
});
test("event replay incorporates an earlier accepted upload after a permanent retry rejection", async () => {
test("an explicit retry recovers an earlier accepted upload after a permanent rejection", async () => {
const requests: unknown[] = [];
const f = await fixture(undefined, {
putFileContent: async () => {
throw new PermanentSyncError(
"HTTP 413 after proxy reconfiguration"
);
},
getDocumentVersionContent: async () => bytes("local")
putFileContent: async (_id, request) => {
requests.push(structuredClone(request));
if (requests.length === 1) throw new PermanentSyncError("HTTP 413 after proxy reconfiguration");
return { type: "Accepted", ...head("a", 2, "local") };
}
});
f.internals.stopped = false;
await f.internals.preparePush();
const { pending } = f.database.state;
assert(pending);
const { requestId } = pending.request;
await f.internals.finishPending();
await assert.rejects(f.internals.finishPending(), /413/);
await assert.rejects(f.syncer.waitUntilFinished(), /413/);
f.database.state.lastSeenUpdateId = 1;
await f.internals.incorporateEventBatch({
headEventId: 2,
events: [
{
eventId: 2,
requestId,
type: "content",
document: head("a", 2, "local")
}
]
});
await f.syncer.retryRejectedRequests();
await f.internals.finishPending();
assert.deepEqual(requests[1], requests[0]);
assert.equal(f.database.state.documents.a.base?.vaultUpdateId, 2);
assert.equal(f.database.state.pending, undefined);
assert.equal(await f.internals.preparePush(), false);
@ -364,7 +365,7 @@ test("event replay incorporates an earlier accepted upload after a permanent ret
assert.deepEqual(f.disk.userFiles(), new Map([["a.md", bytes("local")]]));
});
test("rejected manifests can be replaced after editing the local namespace", async () => {
test("rejected manifests are resolved before publishing later namespace changes", async () => {
let reject = true;
const f = await fixture(
{},
@ -389,11 +390,14 @@ test("rejected manifests can be replaced after editing the local namespace", asy
base: { ...head("a", 1, "x"), hash: "hash" }
};
await f.database.save();
await f.internals.finishPending();
await assert.rejects(f.internals.finishPending(), /413/);
f.internals.stopped = false;
assert.equal(await f.internals.preparePush(), false);
f.database.state.local = {};
reject = false;
await f.syncer.retryRejectedRequests();
await f.internals.finishPending();
assert.deepEqual(f.database.state.fileManifest.entries, { a: "new.md" });
assert(await f.internals.preparePush());
await f.internals.finishPending();
assert.deepEqual(f.database.state.fileManifest.entries, {});
@ -509,7 +513,7 @@ for (const kind of ["path", "content"] as const) {
f.database.state.lastSeenUpdateId = 1;
f.database.state.documents.a.base = {
...head("a", 1, "base"),
hash: (await toStoredSnapshot({ content: bytes("base") })).hash
hash: (await hashSnapshot({ content: bytes("base") })).hash
};
if (kind === "path") {
await f.disk.userRename("a.md", "c.md");
@ -520,6 +524,7 @@ for (const kind of ["path", "content"] as const) {
}
await f.internals.incorporateEventBatch({
headEventId: 3,
endEventId: 3,
events: [2, 3].map((id) =>
kind === "path"
? {
@ -674,11 +679,12 @@ test("rejected content is reported to callers and an explicit retry clears the r
});
f.internals.stopped = false;
await f.internals.preparePush();
await f.internals.finishPending();
await assert.rejects(f.internals.finishPending(), /413/);
await assert.rejects(f.syncer.waitUntilFinished(), /413/);
assert(f.syncer.isBusy);
await f.syncer.retryRejectedRequests();
assert(await f.internals.preparePush());
assert.equal(f.database.state.pending?.rejection, undefined);
assert(f.database.state.pending, "explicit retry must retain the exact request");
});
test("notification saves cannot roll back an uncertain engine commit, even if the first reload fails", async () => {

View file

@ -91,6 +91,7 @@ for (const [count, delivery] of [
fetchedAfter.push(after);
return {
headEventId: available,
endEventId: available,
events: events.filter(
(event) =>
event.eventId > after &&
@ -108,7 +109,6 @@ for (const [count, delivery] of [
>()
};
const syncer = new Syncer(
"device",
logger,
database,
settings,
@ -122,6 +122,7 @@ for (const [count, delivery] of [
const deliver = async (event: EventRecord) => {
await websocket.onRemoteVaultUpdateReceived.triggerAsync({
headEventId: event.eventId,
endEventId: event.eventId,
// Payloads are deliberately inconsistent with HTTP. The socket
// only wakes the loop; it cannot supply reconciliation state.
events: [

View file

@ -1,7 +1,7 @@
import assert from "node:assert/strict";
import { test } from "node:test";
import { fixture, bytes } from "./sync-fixture";
import { toStoredSnapshot } from "../sync-client/src/snapshot";
import { hashSnapshot } from "../sync-client/src/snapshot";
import { LocalChangesDuringReconciliation } from "../sync-client/src/errors/errors";
for (const equal of [false, true])
@ -78,7 +78,7 @@ test("content edited after planning is merged with the incoming snapshot", async
await f.files.apply(structuredClone(f.database.state), {
a: {
expected,
replacement: await toStoredSnapshot({
replacement: await hashSnapshot({
content: bytes("remote\nbase\n")
})
}
@ -95,7 +95,7 @@ test("deleting the source after planning does not resurrect it", async () => {
next.local.a = "remote.md";
await f.disk.userDelete("a.md");
await f.files.apply(next, {
a: { replacement: await toStoredSnapshot({ content: bytes("remote") }) }
a: { replacement: await hashSnapshot({ content: bytes("remote") }) }
});
assert.deepEqual(f.disk.userFiles(), new Map());
assert.deepEqual(f.database.state.local, {});
@ -111,7 +111,7 @@ for (const protect of ["ignore", "size"] as const)
else await f.settings.setSetting("maxFileSizeMB", 0);
await f.files.apply(next, {
a: {
replacement: await toStoredSnapshot({
replacement: await hashSnapshot({
content: bytes("remote")
})
}
@ -224,7 +224,7 @@ for (const phase of ["before", "durable"])
const original = structuredClone(f.database.state);
const write = {
expected: (await f.files.snapshot("a.md"))!,
replacement: await toStoredSnapshot({
replacement: await hashSnapshot({
content: bytes("remote\nbase\n")
})
};
@ -272,7 +272,7 @@ test("atomic updater includes an edit made after the last snapshot without unlin
await f.files.apply(structuredClone(f.database.state), {
a: {
expected,
replacement: await toStoredSnapshot({
replacement: await hashSnapshot({
content: bytes("remote\nbase\n")
})
}
@ -298,7 +298,7 @@ test("a failed atomic update leaves database metadata unchanged", async () => {
f.files.apply(structuredClone(original), {
a: {
expected: (await f.files.snapshot("a.md"))!,
replacement: await toStoredSnapshot({
replacement: await hashSnapshot({
content: bytes("remote")
})
}

View file

@ -4,7 +4,7 @@ import { fixture } from "./sync-fixture";
const f = await fixture(
{},
{
getEvents: async () => ({ headEventId: 0, events: [] })
getEvents: async () => ({ headEventId: 0, endEventId: 0, events: [] })
}
);
let delivered = 0;

View file

@ -61,6 +61,7 @@ test("canonical checks independently verify both persisted content hashes", asyn
: url.includes("/events-since")
? {
headEventId: 2,
endEventId: 2,
events: [
{ eventId: 1, requestId: "create" },
{ eventId: 2, requestId: "manifest" }

View file

@ -8,7 +8,7 @@ import {
import { Settings } from "../sync-client/src/persistence/settings";
import { FileOperations } from "../sync-client/src/file-operations/file-operations";
import { Syncer } from "../sync-client/src/sync-operations/syncer";
import { toStoredSnapshot } from "../sync-client/src/snapshot";
import { hashSnapshot } from "../sync-client/src/snapshot";
import { Logger } from "../sync-client/src/tracing/logger";
import { SyncHistory } from "../sync-client/src/tracing/sync-history";
import { FixedSizeDocumentCache } from "../sync-client/src/utils/data-structures/fix-sized-cache";
@ -56,6 +56,7 @@ async function fixture(syncIntervalMs: number | undefined) {
const batch = (after = 2): EventBatch =>
structuredClone({
headEventId,
endEventId: headEventId,
events: events.filter((event) => event.eventId > after)
});
const service: Partial<SyncService> = {
@ -90,7 +91,7 @@ async function fixture(syncIntervalMs: number | undefined) {
initial.lastSeenUpdateId = 2;
initial.local = { [documentId]: "note.bin" };
initial.fileManifest = { fileManifestId: 2, entries: { ...initial.local } };
const hash = (await toStoredSnapshot({ content: bytes("base") })).hash;
const hash = (await hashSnapshot({ content: bytes("base") })).hash;
initial.documents[documentId] = {
remote: latest,
observedHash: hash,
@ -123,7 +124,6 @@ async function fixture(syncIntervalMs: number | undefined) {
>()
};
const syncer = new Syncer(
"device",
logger,
database,
settings,

View file

@ -265,7 +265,9 @@ test(
mode = "reject-retry";
await disk.userWrite("a.md", bytes("base"));
await client.syncLocallyUpdatedFile({ relativePath: "a.md" });
await client.waitUntilFinished();
await assert.rejects(client.waitUntilFinished(), /413/);
assert.deepEqual((await disk.read("a.md"))?.content, bytes("base"));
await client.reset();
const result = Buffer.from(
(await disk.read("a.md"))!.content
).toString();

View file

@ -7,7 +7,7 @@ import { NodeFileSystemOperations } from "../local-client-cli/src/node-filesyste
import { Database, emptyState } from "../sync-client/src/persistence/database";
import { FileOperations } from "../sync-client/src/file-operations/file-operations";
import type { ServerConfig } from "../sync-client/src/services/server-config";
import { toStoredSnapshot } from "../sync-client/src/snapshot";
import { hashSnapshot } from "../sync-client/src/snapshot";
test(
"real filesystem: direct updates, unsafe entries, and file swaps",
@ -103,7 +103,7 @@ test(
initial.documents[id] = {
observedHash: (
await toStoredSnapshot(
await hashSnapshot(
(await disk.read(relative))!
)
).hash

View file

@ -7,7 +7,7 @@ import {
type StoredDatabase
} from "../sync-client/src/persistence/database";
import { FileOperations } from "../sync-client/src/file-operations/file-operations";
import { toStoredSnapshot } from "../sync-client/src/snapshot";
import { hashSnapshot } from "../sync-client/src/snapshot";
import type { ServerConfig } from "../sync-client/src/services/server-config";
import { mergeFileManifests } from "../sync-client/src/sync-operations/file-manifest";
import { assertManifest, pathDecision } from "./oracles";
@ -37,7 +37,7 @@ async function fixture(
initial.documents[id] = {
observedHash: (
await toStoredSnapshot((await disk.read(path))!)
await hashSnapshot((await disk.read(path))!)
).hash
};
const persistence = new MemoryPersistence({ database: initial });
@ -75,7 +75,7 @@ async function swapFixture() {
next.local = { a: "b.md", b: "a.md", c: "nested/c.md" };
next.documents.c = {};
const writes = {
c: { replacement: await toStoredSnapshot({ content: bytes("C") }) }
c: { replacement: await hashSnapshot({ content: bytes("C") }) }
};
const expected = new Map([
["a.md", bytes("B")],
@ -129,8 +129,8 @@ async function mergeFixture() {
next: structuredClone(current.initial),
writes: {
a: {
expected: await toStoredSnapshot({ content: bytes(base) }),
replacement: await toStoredSnapshot({ content: bytes(remote) })
expected: await hashSnapshot({ content: bytes(base) }),
replacement: await hashSnapshot({ content: bytes(remote) })
}
},
expected: new Map([
@ -154,7 +154,7 @@ async function deletionFixture() {
next,
writes: {
a: {
replacement: await toStoredSnapshot({
replacement: await hashSnapshot({
content: bytes("remote binary\0")
})
}
@ -178,8 +178,8 @@ async function permanentDestinationFixture() {
},
writes: {
a: {
expected: await toStoredSnapshot({ content: bytes("A") }),
replacement: await toStoredSnapshot({
expected: await hashSnapshot({ content: bytes("A") }),
replacement: await hashSnapshot({
content: bytes("REMOTE")
})
}
@ -418,8 +418,8 @@ for (const phase of ["before", "durable"]) {
await current.disk.userDelete("a.md");
await current.disk.userWrite("a.md", bytes("new incarnation"));
const changes: LocalChange[] = [
{ type: "delete", path: "a.md" },
{ type: "create", path: "a.md" }
{ type: "delete", path: "a.md", changeId: "delete" },
{ type: "create", path: "a.md", changeId: "recreate" }
];
const scan = async () => {
const runtime = current.restart();
@ -475,7 +475,12 @@ for (const editedContent of ["new bytes", "B"]) {
await current.disk.userRename("a.md", "moved.md");
await current.disk.userWrite("moved.md", bytes(editedContent));
const changes: LocalChange[] = [
{ type: "move", oldPath: "a.md", relativePath: "moved.md" }
{
type: "move",
oldPath: "a.md",
relativePath: "moved.md",
changeId: "move"
}
];
await scanChanges(runtime, changes);
assert.deepEqual(
@ -509,7 +514,12 @@ for (const equalContent of [false, true]) {
["temporary.md", "b.md"]
]) {
await current.disk.userRename(oldPath, relativePath);
changes.push({ type: "move", oldPath, relativePath });
changes.push({
type: "move",
oldPath,
relativePath,
changeId: oldPath
});
}
const expected = { a: "b.md", b: "a.md" };
const expectedFiles = new Map([

View file

@ -9,7 +9,6 @@ for (const mode of [
"behind",
"dirty",
"pending",
"legacy-state",
...["before:save", "durable:save"].flatMap((boundary) =>
[1, 2, 3, 4, 5].map((ordinal) => `${boundary}:${ordinal}`)
)
@ -124,16 +123,6 @@ for (const mode of [
await f.reopenClient();
await f.client.start();
}
if (mode === "legacy-state") {
await f.client.destroy();
const stored = f.store.snapshot() as ReturnType<
typeof f.store.snapshot
> & { historyCheckpoint?: unknown };
delete stored.historyCheckpoint;
await f.store.save(stored);
await f.reopenClient();
await f.client.start();
}
await f.client.setSetting("isSyncEnabled", true);
const actual = Buffer.from(
(await f.disk.read("a.md"))!.content

View file

@ -5,7 +5,7 @@ import { test } from "node:test";
import type { Database } from "../sync-client/src/persistence/database";
import { emptyState } from "../sync-client/src/persistence/database";
import { SyncClient } from "../sync-client/src/sync-client";
import { toStoredSnapshot } from "../sync-client/src/snapshot";
import { hashSnapshot } from "../sync-client/src/snapshot";
import { EventListeners } from "../sync-client/src/utils/data-structures/event-listeners";
import type { SyncService } from "../sync-client/src/services/sync-service";
import type { WebSocketManager } from "../sync-client/src/services/websocket-manager";
@ -81,11 +81,13 @@ test("oversized untracked file must remain in place", async () => {
});
test("rejected unknown success must use submitted content as merge base", async () => {
let reject = true;
const f = await fixture(
{ "a.md": "base" },
{
putFileContent: async () => {
throw new PermanentSyncError("HTTP 413");
if (reject) throw new PermanentSyncError("HTTP 413");
return { type: "Accepted", ...head("a", 2, "base A") };
},
getDocumentVersionContent: async ({ vaultUpdateId }) =>
bytes(
@ -106,10 +108,14 @@ test("rejected unknown success must use submitted content as merge base", async
f.internals.stopped = false;
await f.internals.preparePush();
const req = f.database.state.pending!.request.requestId;
await f.internals.finishPending();
await assert.rejects(f.internals.finishPending(), /413/);
await f.disk.userWrite("a.md", bytes("base"));
reject = false;
await f.syncer.retryRejectedRequests();
await f.internals.finishPending();
await f.internals.incorporateEventBatch({
headEventId: 3,
endEventId: 3,
events: [
{
eventId: 2,
@ -166,7 +172,7 @@ test("oversized remote head must not cause identical stale uploads forever", asy
);
f.database.state.documents.a.base = {
...head("a", 1, "base"),
hash: (await toStoredSnapshot({ content: bytes("base") })).hash
hash: (await hashSnapshot({ content: bytes("base") })).hash
};
(f.database.state.documents.a ??= {}).remote = head("a", 2, remoteText);
await f.settings.setSettings({ maxFileSizeMB: 10 / (1024 * 1024) });
@ -178,11 +184,13 @@ test("oversized remote head must not cause identical stale uploads forever", asy
});
test("local undo of accepted-but-rejected manifest must survive replay", async () => {
let reject = true;
const f = await fixture(
{ "a.md": "A" },
{
pushFileManifest: async () => {
throw new PermanentSyncError("HTTP 413");
if (reject) throw new PermanentSyncError("HTTP 413");
return { type: "Accepted", fileManifestId: 2 };
},
getDocumentVersionContent: async () => bytes("A")
}
@ -202,14 +210,18 @@ test("local undo of accepted-but-rejected manifest must survive replay", async (
await f.internals.preparePush();
const req = f.database.state.pending!.request.requestId;
f.internals.stopped = true;
await f.internals.finishPending();
await assert.rejects(f.internals.finishPending(), /413/);
await f.disk.userRename("b.md", "a.md");
await f.syncer.syncLocallyUpdatedFile({
oldPath: "b.md",
relativePath: "a.md"
});
reject = false;
await f.syncer.retryRejectedRequests();
await f.internals.finishPending();
await f.internals.incorporateEventBatch({
headEventId: 2,
endEventId: 2,
events: [
{
eventId: 2,
@ -244,9 +256,6 @@ test("a dirty cursor update must invalidate old precise positions", async () =>
const { CursorTracker } = await import(
"../sync-client/src/sync-operations/cursor-tracker"
);
const { FileChangeNotifier } = await import(
"../sync-client/src/sync-operations/file-change-notifier"
);
const f = await fixture({ "a.md": "base" });
f.database.state.documents.a.base = {
...head("a", 1, "base"),
@ -261,7 +270,7 @@ test("a dirty cursor update must invalidate old precise positions", async () =>
f.database,
ws as unknown as WebSocketManager,
f.files,
new FileChangeNotifier()
new EventListeners()
);
let latest: import("../sync-client/src/types/maybe-outdated-client-cursors").MaybeOutdatedClientCursors[] =
[];
@ -335,21 +344,35 @@ test("silent open WebSocket must eventually trigger catchup or reconnect", async
fetch: async (input) => {
const path = String(input);
if (path.endsWith("/ping"))
return Response.json({
supportedApiVersion: SUPPORTED_API_VERSION,
isAuthenticated: true,
mergeableFileExtensions: ["md"],
serverVersion: "test"
});
return Response.json(
{
supportedApiVersion: SUPPORTED_API_VERSION,
isAuthenticated: true,
mergeableFileExtensions: ["md"],
serverVersion: "test"
},
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
if (path.endsWith("/file-manifest"))
return Response.json(
{ type: "Accepted", fileManifestId: 0 },
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
if (path.endsWith("/vault-snapshot"))
return Response.json({
headEventId: 0,
fileManifest: { fileManifestId: 0, entries: {} },
documents: []
});
return Response.json(
{
headEventId: 0,
fileManifest: { fileManifestId: 0, entries: {} },
documents: []
},
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
if (path.includes("/events-since")) {
eventReads++;
return Response.json({ headEventId: 0, events: [] });
return Response.json(
{ headEventId: 0, endEventId: 0, events: [] },
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
}
throw new Error("Unexpected request " + path);
}
@ -478,7 +501,7 @@ for (const interrupted of [false, true]) {
f = await fixture({ "a.md": "base LOCAL" }, service);
f.database.state.documents.a.base = {
...head("a", 1, "base"),
hash: (await toStoredSnapshot({ content: bytes("base") })).hash
hash: (await hashSnapshot({ content: bytes("base") })).hash
};
(f.database.state.documents.a ??= {}).remote = head("a", 1, "base");
f.database.state.lastSeenUpdateId = 1;
@ -554,14 +577,16 @@ test("content notifications during every snapshot do not starve a complete names
});
for (const boundary of ["before:save", "durable:save"]) {
test(`receipt incorporation after paged replay survives ${boundary} and process replacement`, async () => {
test(`retried acknowledgement before paged replay survives ${boundary} and process replacement`, async () => {
let reject = true;
const service: Partial<SyncService> = {
getEvents: async (after) => {
assert.equal(after, 2);
return page;
},
putFileContent: async () => {
throw new PermanentSyncError("HTTP 413");
if (reject) throw new PermanentSyncError("HTTP 413");
return { type: "Accepted", ...head("a", 2, "base A") };
},
getDocumentVersionContent: async ({ vaultUpdateId }) =>
bytes(
@ -582,7 +607,7 @@ for (const boundary of ["before:save", "durable:save"]) {
f.internals.stopped = false;
await f.internals.preparePush();
const { requestId } = f.database.state.pending!.request;
await f.internals.finishPending();
await assert.rejects(f.internals.finishPending(), /413/);
await f.disk.userWrite("a.md", bytes("base"));
const first = {
headEventId: 3,
@ -608,17 +633,20 @@ for (const boundary of ["before:save", "durable:save"]) {
}
]
};
reject = false;
await f.syncer.retryRejectedRequests();
f.persistence.boundary = (label) => {
if (label === boundary) throw new Error("injected crash");
};
await assert.rejects(
f.internals.incorporateEventBatch(first),
f.internals.finishPending(),
/injected crash/
);
f.persistence.boundary = () => { };
f.disk.crash(true);
f = await fixture({}, service, f);
assert.equal(f.database.state.lastSeenUpdateId, 1);
await f.internals.finishPending();
await f.internals.incorporateEventBatch(first);
assert.equal(f.database.state.lastSeenUpdateId, 3);
assert.deepEqual(
@ -653,19 +681,28 @@ test("initial vault settings bind the engine and survive restart", async () => {
webSocket: Socket as unknown as typeof WebSocket,
fetch: async (input: RequestInfo | URL) => {
if (String(input).endsWith("/ping"))
return Response.json({
supportedApiVersion: SUPPORTED_API_VERSION,
isAuthenticated: true,
mergeableFileExtensions: ["md"],
serverVersion: "test"
});
return Response.json(
{
supportedApiVersion: SUPPORTED_API_VERSION,
isAuthenticated: true,
mergeableFileExtensions: ["md"],
serverVersion: "test"
},
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
if (String(input).endsWith("/vault-snapshot"))
return Response.json({
headEventId: 0,
fileManifest: { fileManifestId: 0, entries: {} },
documents: []
});
return Response.json({ headEventId: 0, events: [] });
return Response.json(
{
headEventId: 0,
fileManifest: { fileManifestId: 0, entries: {} },
documents: []
},
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
return Response.json(
{ headEventId: 0, endEventId: 0, events: [] },
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
}
};
let client = await SyncClient.create(options);
@ -695,13 +732,14 @@ test("disabling sync persists across reset without resuming the transport", asyn
}
});
for (const receipt of [false, true]) {
test(`${receipt ? "replayed" : "direct"} upload acknowledgement survives a local deletion`, async () => {
for (const retry of [false, true]) {
test(`${retry ? "retried" : "direct"} upload acknowledgement survives a local deletion`, async () => {
let reject = retry;
const f = await fixture(
{ "a.md": "base" },
{
putFileContent: async () => {
if (receipt)
if (reject)
throw new PermanentSyncError(
"HTTP 413 after lost success"
);
@ -717,30 +755,19 @@ for (const receipt of [false, true]) {
await f.disk.userWrite("a.md", bytes("edited"));
f.internals.stopped = false;
await f.internals.preparePush();
const { requestId } = f.database.state.pending!.request;
if (receipt) await f.internals.finishPending();
if (retry) await assert.rejects(f.internals.finishPending(), /413/);
// Recovery folds queued deletions into the durable namespace before
// an outstanding request or its delayed receipt is handled.
// an outstanding request is retried.
f.internals.stopped = true;
await f.disk.userDelete("a.md");
await f.syncer.syncLocallyDeletedFile("a.md");
await f.internals.scan();
assert.deepEqual(f.database.state.local, {});
if (receipt) {
await f.internals.incorporateEventBatch({
headEventId: 2,
events: [
{
eventId: 2,
requestId,
type: "content",
document: head("a", 2, "edited")
}
]
});
assert.equal(f.database.state.lastSeenUpdateId, 2);
assert.equal(f.database.state.unconfirmed?.length ?? 0, 0);
} else await f.internals.finishPending();
if (retry) {
reject = false;
await f.syncer.retryRejectedRequests();
}
await f.internals.finishPending();
assert.equal(f.database.state.pending, undefined);
assert.deepEqual(f.database.state.local, {});
assert.equal(await f.disk.read("a.md"), undefined);
@ -781,21 +808,35 @@ async function settingsLifecycleFixture() {
fetch: async (input) => {
const path = String(input);
if (path.endsWith("/ping"))
return Response.json({
supportedApiVersion: SUPPORTED_API_VERSION,
isAuthenticated: true,
mergeableFileExtensions: ["md"],
serverVersion: "test"
});
return Response.json(
{
supportedApiVersion: SUPPORTED_API_VERSION,
isAuthenticated: true,
mergeableFileExtensions: ["md"],
serverVersion: "test"
},
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
if (path.endsWith("/file-manifest"))
return Response.json(
{ type: "Accepted", fileManifestId: 0 },
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
if (path.endsWith("/vault-snapshot"))
return Response.json({
headEventId: 0,
fileManifest: { fileManifestId: 0, entries: {} },
documents: []
});
return Response.json(
{
headEventId: 0,
fileManifest: { fileManifestId: 0, entries: {} },
documents: []
},
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
if (path.includes("/events-since")) {
eventReads++;
return Response.json({ headEventId: 0, events: [] });
return Response.json(
{ headEventId: 0, endEventId: 0, events: [] },
{ headers: { "X-Vault-Link-History": "0:empty" } }
);
}
throw new Error("Unexpected request " + path);
}
@ -818,12 +859,11 @@ for (const boundary of ["before:save", "durable:save"]) {
const { database } = f.client as unknown as { database: Database };
await database.commit({
...database.state,
unconfirmed: [{
pending: {
type: "fileManifest",
requestId: "rejected",
entries: { retired: "old.md" },
rejection: "retired rejection"
}]
request: { requestId: "rejected", parentFileManifestId: 0, entries: {} },
rejection: "upload rejected"
}
});
let saves = 0,
failed = false;
@ -840,6 +880,10 @@ for (const boundary of ["before:save", "durable:save"]) {
);
assert(failed);
assert.equal(f.client.getSettings().maxFileSizeMB, 20);
if (boundary === "before:save") {
await assert.rejects(f.client.waitUntilFinished(), /upload rejected/);
await f.client.reset();
}
const before = f.reads();
await f.client.syncLocallyUpdatedFile({
relativePath: "absent.md"

View file

@ -6,7 +6,7 @@ import {
import { Settings } from "../sync-client/src/persistence/settings";
import { FileOperations } from "../sync-client/src/file-operations/file-operations";
import { Syncer } from "../sync-client/src/sync-operations/syncer";
import { toStoredSnapshot } from "../sync-client/src/snapshot";
import { hashSnapshot } from "../sync-client/src/snapshot";
import { Logger } from "../sync-client/src/tracing/logger";
import { SyncHistory } from "../sync-client/src/tracing/sync-history";
import { FixedSizeDocumentCache } from "../sync-client/src/utils/data-structures/fix-sized-cache";
@ -45,7 +45,7 @@ export async function fixture(
initial.documents[id] = {
observedHash: (
await toStoredSnapshot({ content: bytes(content) })
await hashSnapshot({ content: bytes(content) })
).hash
};
}
@ -88,7 +88,6 @@ export async function fixture(
} as unknown as WebSocketManager;
const contentCache = new FixedSizeDocumentCache(1024);
const syncer = new Syncer(
"device",
logger,
database,
settings,