andras/vault-link
andras/vault-link
Watch
1
0
Fork
You've already forked vault-link
 0
Code Issues Pull requests 18 Projects Releases 55 Packages Wiki Activity Actions 37

..

Browse source
This commit is contained in:
Andras Schmelczer 2026-09-26 20:22:19 +01:00
commit ca905ac1ce
6 changed files with 181 additions and 202 deletions
Download patch file Download diff file Expand all files Collapse all files

2
frontend/sync-client/src/persistence/database.test.ts
View file

@ -63,7 +63,7 @@ test("old state loads without duplicate flags, full recovery content, or acknowl
deferred: {
base: undefined,
observedHash: undefined,
remote: { ...head, documentId: "deferred" }
remote: { vaultUpdateId: head.vaultUpdateId, contentSize: head.contentSize }
}
});
assert.equal(saved.lastSeenUpdateId, 2);

31
frontend/sync-client/src/persistence/database.ts
View file

@ -1,11 +1,6 @@
import type { PendingRequest, UnconfirmedRequest } from "./requests";
import { upgradeState } from "./upgrade-state";
import type { StoredSnapshot } from "../snapshot";
import type { DocumentUpdateResponse } from "../services/types/DocumentUpdateResponse";
import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent";
import type { FileManifest } from "../services/types/FileManifest";
import type { FileManifestUpdateResponse } from "../services/types/FileManifestUpdateResponse";
import type { PushFileManifest } from "../services/types/PushFileManifest";
import type { PutFileContent } from "../services/types/PutFileContent";
import type { FileManifestEntries } from "../types/file-manifest-entries";
export type VaultUpdateId = number;
@ -17,29 +12,21 @@ export interface DocumentMetadata {
hash: string;
}
export interface ContentHead {
vaultUpdateId: number;
contentSize: number;
}
export interface DocumentState {
/** A null version retains only the clean hash after a server history reset. */
base?: { vaultUpdateId: number | null; hash: string };
/** Last observed disk hash; an empty string means a notification awaits scanning. */
observedHash?: string;
/** A remote version still to be incorporated, including excluded downloads. */
remote?: DocumentVersionWithoutContent;
remote?: ContentHead;
}
export type PendingRequest = { rejection?: string } & (
| {
type: "content";
documentId: string;
request: PutFileContent;
snapshot: StoredSnapshot;
response?: DocumentUpdateResponse;
}
| {
type: "fileManifest";
request: PushFileManifest;
response?: FileManifestUpdateResponse;
}
);
export type { PendingRequest, UnconfirmedRequest } from "./requests";
export interface EngineState {
vaultKey: string;
@ -50,7 +37,7 @@ export interface EngineState {
lastSeenUpdateId?: number;
pending?: PendingRequest;
/** A rejected retry does not prove an earlier attempt failed to commit. */
unconfirmed?: PendingRequest[];
unconfirmed?: UnconfirmedRequest[];
/** Commits consumption of the saved notification queue with the identity map. */
lastAppliedLocalChangeId?: string;
/** Original server paths while local files are excluded; null means absent. */

21
frontend/sync-client/src/persistence/requests.ts Normal file
View file

@ -0,0 +1,21 @@
import type { PutFileContent } from "../services/types/PutFileContent";
import type { PushFileManifest } from "../services/types/PushFileManifest";
import type { FileManifestEntries } from "../types/file-manifest-entries";
/** Exact wire payload and the hash needed to advance the base after acknowledgement. */
export type PendingRequest =
| { type: "content"; documentId: string; request: PutFileContent; hash: string }
| { type: "fileManifest"; request: PushFileManifest };
/** Enough to recognize a late receipt or suppress an unchanged rejected submission. */
export type UnconfirmedRequest = { requestId: string; rejection?: string } & (
| { type: "content"; documentId: string; hash: string }
| { type: "fileManifest"; entries: FileManifestEntries }
);
export function requestReceipt(request: PendingRequest, rejection?: string): UnconfirmedRequest {
const receipt = { requestId: request.request.requestId, rejection };
return request.type === "content"
? { ...receipt, type: "content", documentId: request.documentId, hash: request.hash }
: { ...receipt, type: "fileManifest", entries: request.request.entries };
}

91
frontend/sync-client/src/persistence/upgrade-state.ts
View file

@ -1,57 +1,72 @@
import type { DocumentState, EngineState } from "./database";
import type { ContentHead, DocumentState, EngineState } from "./database";
import { requestReceipt, type PendingRequest, type UnconfirmedRequest } from "./requests";
import type { StoredSnapshot } from "../snapshot";
import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent";
/** Read the previous redundant state once; all subsequent saves use the smaller format. */
type PreviousRequest = PendingRequest & {
snapshot?: StoredSnapshot;
response?: unknown;
rejection?: string;
};
function upgradeRequest(request: PreviousRequest): PendingRequest {
return request.type === "content"
? {
type: "content", documentId: request.documentId,
request: request.request, hash: request.snapshot?.hash ?? request.hash
}
: { type: "fileManifest", request: request.request };
}
/** Normalize earlier stores at the boundary; none of their duplicate data is saved again. */
export function upgradeState(state: EngineState): EngineState {
if (!("initialized" in state)) return state;
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- Read the previous persisted schema at the migration boundary.
const legacy = state as EngineState & {
initialized: boolean;
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- Read previous persisted schemas at the migration boundary.
const legacy = state as Omit<EngineState, "pending" | "unconfirmed" | "documents"> & {
initialized?: boolean;
historyRecovery?: boolean;
remoteHeads?: Record<string, DocumentVersionWithoutContent>;
documents: Record<
string,
DocumentState & {
materialized?: boolean;
bootstrap?: boolean;
recoveryBase?: StoredSnapshot;
}
>;
remoteHeads?: Record<string, ContentHead>;
pending?: PreviousRequest;
unconfirmed?: (UnconfirmedRequest | PreviousRequest)[];
documents: Record<string, DocumentState & {
materialized?: boolean;
bootstrap?: boolean;
recoveryBase?: StoredSnapshot;
}>;
};
const next: EngineState = {
vaultKey: state.vaultKey,
fileManifest: state.fileManifest,
local: state.local,
documents: {},
lastSeenUpdateId:
legacy.initialized && legacy.historyRecovery !== true
? state.lastSeenUpdateId
: undefined,
pending: state.pending,
unconfirmed: state.unconfirmed,
lastSeenUpdateId: "initialized" in legacy &&
(!legacy.initialized || legacy.historyRecovery === true)
? undefined : state.lastSeenUpdateId,
pending: legacy.pending && upgradeRequest(legacy.pending),
unconfirmed: legacy.unconfirmed?.map((request) =>
"request" in request
? requestReceipt(upgradeRequest(request), request.rejection)
: request
),
lastAppliedLocalChangeId: state.lastAppliedLocalChangeId,
excluded: state.excluded
};
for (const [id, doc] of Object.entries(legacy.documents)) {
if (legacy.pending?.rejection && !legacy.pending.response) {
(next.unconfirmed ??= []).push(requestReceipt(upgradeRequest(legacy.pending), legacy.pending.rejection));
delete next.pending;
}
for (const id of new Set([...Object.keys(legacy.documents), ...Object.keys(legacy.remoteHeads ?? {})])) {
const doc = legacy.documents[id] ?? {};
const base = doc.recoveryBase
? { vaultUpdateId: null, hash: doc.recoveryBase.hash }
: doc.base && doc.bootstrap !== true
? { vaultUpdateId: doc.base.vaultUpdateId, hash: doc.base.hash }
: undefined;
next.documents[id] = {
base,
observedHash:
doc.materialized === true ? (doc.observedHash ?? "") : undefined
};
}
for (const [id, remote] of Object.entries(legacy.remoteHeads ?? {})) {
const doc = (next.documents[id] ??= {});
if (
doc.observedHash === undefined ||
remote.vaultUpdateId > (doc.base?.vaultUpdateId ?? 0)
)
doc.remote = remote;
? { vaultUpdateId: doc.base.vaultUpdateId, hash: doc.base.hash }
: undefined;
const observedHash = "materialized" in doc
? doc.materialized === true ? doc.observedHash ?? "" : undefined
: doc.observedHash;
const remote = doc.remote ?? legacy.remoteHeads?.[id];
next.documents[id] = { base, observedHash };
if (remote && (observedHash === undefined || remote.vaultUpdateId > (base?.vaultUpdateId ?? 0)))
next.documents[id].remote = { vaultUpdateId: remote.vaultUpdateId, contentSize: remote.contentSize };
}
return next;
}

6
frontend/sync-client/src/sync-operations/syncer.test.ts
View file

@ -210,10 +210,10 @@ describe("Syncer per-file state and event delivery", () => {
const { syncer } = fixture();
const projected = (
syncer as unknown as {
documentHead: (value: typeof head) => Record<string, unknown>;
contentHead: (value: typeof head) => Record<string, unknown>;
}
).documentHead({ ...head, contentBase64: "secret" } as typeof head);
assert.deepEqual(projected, head);
).contentHead({ ...head, contentBase64: "secret" } as typeof head);
assert.deepEqual(projected, { vaultUpdateId: head.vaultUpdateId, contentSize: head.contentSize });
assert.strictEqual("contentBase64" in projected, false);
});
});

210
frontend/sync-client/src/sync-operations/syncer.ts
View file

@ -1,12 +1,15 @@
import { requestReceipt } from "../persistence/requests";
import type { SyncHistory } from "../tracing/sync-history";
import { SyncStatus, SyncType } from "../tracing/sync-history";
import { v4 as uuid } from "uuid";
import { base64ToBytes } from "byte-base64";
import { diff } from "reconcile-text";
import { diff, undiff } from "reconcile-text";
import type {
Database,
EngineState,
PendingRequest
PendingRequest,
UnconfirmedRequest,
ContentHead
} from "../persistence/database";
import { emptyState } from "../persistence/database";
import { toStoredSnapshot, type StoredSnapshot } from "../snapshot";
@ -504,9 +507,9 @@ export class Syncer {
private async incorporateEventBatch(batch: EventBatch): Promise<void> {
let cursor = this.database.state.lastSeenUpdateId ?? 0;
let manifest = this.database.state.fileManifest;
const heads: Record<string, DocumentVersionWithoutContent> = {};
for (const doc of Object.values(this.database.state.documents))
if (doc.remote) heads[doc.remote.documentId] = doc.remote;
const heads: Record<string, ContentHead> = {};
for (const [id, doc] of Object.entries(this.database.state.documents))
if (doc.remote) heads[id] = doc.remote;
const receipts: EventRecord[] = [];
// Fold every page before touching local files. If fetching is interrupted,
// discard this work and retry from the last committed cursor.
@ -522,7 +525,7 @@ export class Syncer {
if (
this.database.state.unconfirmed?.some(
(pending) =>
pending.request.requestId === event.requestId
pending.requestId === event.requestId
) === true
)
receipts.push(event);
@ -537,7 +540,7 @@ export class Syncer {
heads[event.document.documentId].vaultUpdateId <
event.document.vaultUpdateId
) {
heads[event.document.documentId] = this.documentHead(
heads[event.document.documentId] = this.contentHead(
event.document
);
}
@ -558,7 +561,7 @@ export class Syncer {
let acknowledged = false;
for (const event of receipts) {
const pending = next.unconfirmed?.find(
(candidate) => candidate.request.requestId === event.requestId
(candidate) => candidate.requestId === event.requestId
);
if (!pending) continue;
if (event.type === "content" && pending.type === "content") {
@ -569,14 +572,9 @@ export class Syncer {
) {
doc.base = {
vaultUpdateId: event.eventId,
hash: pending.snapshot.hash
hash: pending.hash
};
this.clearRejections(next, pending.documentId);
await this.cacheContent(
event.eventId,
next.local[pending.documentId],
base64ToBytes(pending.snapshot.contentBase64)
);
}
} else if (
event.type === "fileManifest" &&
@ -599,24 +597,21 @@ export class Syncer {
await this.reconcile(manifest, { cursor, heads });
}
private documentHead(
version: DocumentVersionWithoutContent
): DocumentVersionWithoutContent {
private contentHead(
version: ContentHead
): ContentHead {
return {
vaultUpdateId: version.vaultUpdateId,
documentId: version.documentId,
updatedDate: version.updatedDate,
userId: version.userId,
deviceId: version.deviceId,
contentSize: version.contentSize
};
}
private recordManifestApplication(
before: Readonly<Record<string, string>>,
writes: Readonly<Record<string, FileWrite>>
after: Readonly<Record<string, string>>,
writes: Readonly<Record<string, FileWrite>>,
message: string
): void {
const after = this.database.state.local;
for (const id of new Set([
...Object.keys(before),
...Object.keys(after)
@ -626,7 +621,7 @@ export class Syncer {
if (from === to && !writes[id]) continue;
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
message: "Remote change applied",
message,
details:
from && to && from !== to
? {
@ -660,18 +655,19 @@ export class Syncer {
private async contentWrite(
next: EngineState,
head: DocumentVersionWithoutContent,
id: string,
head: ContentHead,
path: string
): Promise<FileWrite | undefined> {
const document = (next.documents[head.documentId] ??= {});
const document = (next.documents[id] ??= {});
if (
next.excluded?.[head.documentId] !== undefined ||
next.excluded?.[id] !== undefined ||
this.settings.isIgnored(path) ||
this.settings.isOversized(head.contentSize)
) {
return undefined;
}
const currentPath = this.database.state.local[head.documentId];
const currentPath = this.database.state.local[id];
const local =
currentPath &&
(await this.files.fs.stat(currentPath))?.kind === "file"
@ -695,7 +691,7 @@ export class Syncer {
? local
: await this.remoteSnapshot(
{
documentId: head.documentId,
documentId: id,
vaultUpdateId: document.base.vaultUpdateId
},
path
@ -703,7 +699,7 @@ export class Syncer {
: undefined;
// Cache the new head after the older merge base so it remains the most
// recently used version and can back the follow-up local diff.
const remote = await this.remoteSnapshot(head, path);
const remote = await this.remoteSnapshot({ documentId: id, vaultUpdateId: head.vaultUpdateId }, path);
let result = remote;
if (local !== undefined) {
result = await mergeContent(
@ -736,7 +732,7 @@ export class Syncer {
cursor?: number;
initial?: VaultSnapshot;
clearPending?: boolean;
heads?: Record<string, DocumentVersionWithoutContent>;
heads?: Record<string, ContentHead>;
} = {}
): Promise<void> {
validateFileManifest(remote.entries);
@ -763,8 +759,8 @@ export class Syncer {
const previousManifest = next.fileManifest.entries;
next.local = mergeFileManifests(base, next.local, remote.entries);
next.fileManifest = remote;
for (const head of Object.values(heads ?? {}))
(next.documents[head.documentId] ??= {}).remote = head;
for (const [id, head] of Object.entries(heads ?? {}))
(next.documents[id] ??= {}).remote = head;
for (const [id, oldPath] of Object.entries(before)) {
const path = next.local[id];
if (
@ -794,7 +790,7 @@ export class Syncer {
if (initial) {
for (const head of initial.documents)
(next.documents[head.documentId] ??= {}).remote =
this.documentHead(head);
this.contentHead(head);
}
const writes: Record<string, FileWrite> = {};
for (const [id, path] of Object.entries(next.local)) {
@ -825,15 +821,15 @@ export class Syncer {
`File manifest references missing content: ${id}`
);
}
doc.remote = this.documentHead(head);
const write = await this.contentWrite(next, head, path);
doc.remote = this.contentHead(head);
const write = await this.contentWrite(next, id, head, path);
if (write) {
writes[id] = write;
}
}
}
await this.files.apply(next, writes, guard);
this.recordManifestApplication(before, writes);
this.recordManifestApplication(before, this.database.state.local, writes, "Remote change applied");
}
private async preparePush(): Promise<boolean> {
@ -899,7 +895,7 @@ export class Syncer {
(request) =>
request.type === "content" &&
request.documentId === id &&
request.snapshot.hash === snapshot.hash
request.hash === snapshot.hash
)
) {
continue;
@ -910,7 +906,7 @@ export class Syncer {
pending: {
type: "content",
documentId: id,
snapshot,
hash: snapshot.hash,
request: {
requestId: uuid(),
parentVersionId: base?.vaultUpdateId ?? null,
@ -935,7 +931,7 @@ export class Syncer {
!this.rejections.some(
(request) =>
request.type === "fileManifest" &&
sameFileManifest(entries, request.request.entries)
sameFileManifest(entries, request.entries)
)
) {
guard();
@ -988,7 +984,7 @@ export class Syncer {
);
}
private get rejections(): PendingRequest[] {
private get rejections(): UnconfirmedRequest[] {
return (this.database.state.unconfirmed ?? []).filter(
(request) => request.rejection !== undefined
);
@ -1003,8 +999,8 @@ export class Syncer {
const doc = state.documents[id];
if (
path !== undefined &&
doc?.observedHash === request.snapshot.hash &&
doc.base?.hash !== request.snapshot.hash &&
doc?.observedHash === request.hash &&
doc.base?.hash !== request.hash &&
!this.settings.isIgnored(path) &&
state.excluded?.[id] === undefined
)
@ -1012,11 +1008,11 @@ export class Syncer {
} else if (
sameFileManifest(
this.manifestToSubmit(),
request.request.entries
request.entries
) &&
!sameFileManifest(
state.fileManifest.entries,
request.request.entries
request.entries
)
)
return request.rejection;
@ -1043,55 +1039,40 @@ export class Syncer {
await this.database.commit(next);
}
private async finishPending(): Promise<void> {
let { pending } = this.database.state;
if (!pending) {
return;
}
if (!pending.response && !pending.rejection) {
const copy = structuredClone(pending);
try {
if (copy.type === "content") {
copy.response = await this.service.putFileContent(
copy.documentId,
copy.request
);
} else {
copy.response = await this.service.pushFileManifest(
copy.request
);
}
} catch (error) {
if (!(error instanceof PermanentSyncError)) throw error;
copy.rejection = error.message;
}
await this.database.commit({ ...this.next(), pending: copy });
pending = copy;
}
if (pending.rejection && !pending.response) {
private async sendPending<T>(pending: PendingRequest, send: () => Promise<T>): Promise<T | undefined> {
try {
return await send();
} catch (error) {
if (!(error instanceof PermanentSyncError)) throw error;
const next = this.next();
(next.unconfirmed ??= []).push(pending);
(next.unconfirmed ??= []).push(requestReceipt(pending, error.message));
delete next.pending;
await this.database.commit(next);
this.history.addHistoryEntry({
status: SyncStatus.ERROR,
message: pending.rejection,
message: error.message,
details: {
type: SyncType.SKIPPED,
relativePath:
pending.type === "content"
? (next.local[pending.documentId] ??
pending.documentId)
: "File manifest"
relativePath: pending.type === "content"
? (next.local[pending.documentId] ?? pending.documentId)
: "File manifest"
}
});
return;
}
}
private async finishPending(): Promise<void> {
const { pending } = this.database.state;
if (!pending) return;
// Responses are ephemeral. An interrupted application retries the exact
// saved request and gets its original acknowledgement from the server.
if (pending.type === "fileManifest") {
const response = pending.response!;
const response = await this.sendPending(pending, async () =>
this.service.pushFileManifest(pending.request));
if (!response) return;
if (response.type === "StaleBase") {
await this.reconcile(response, { clearPending: true });
} else {
@ -1104,68 +1085,32 @@ export class Syncer {
delete next.pending;
const previous = this.database.state.fileManifest.entries;
await this.database.commit(next);
for (const id of new Set([
...Object.keys(previous),
...Object.keys(next.fileManifest.entries)
])) {
const before = previous[id],
after = next.fileManifest.entries[id];
if (before === after) {
continue;
}
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
message: "File manifest change committed",
details:
after === undefined
? {
type: SyncType.DELETE,
relativePath: before
}
: before === undefined
? {
type: SyncType.CREATE,
relativePath: after
}
: {
type: SyncType.MOVE,
relativePath: after,
movedFrom: before
}
});
}
this.recordManifestApplication(previous, next.fileManifest.entries, {}, "File manifest change committed");
}
} else {
const response = pending.response!;
const response = await this.sendPending(pending, async () =>
this.service.putFileContent(pending.documentId, pending.request));
if (!response) return;
if (response.type === "StaleBase") {
await this.reconcile(this.database.state.fileManifest, {
clearPending: true,
heads: { [pending.documentId]: this.documentHead(response) }
heads: { [pending.documentId]: this.contentHead(response) }
});
} else {
const next = this.next();
const doc = (next.documents[pending.documentId] ??= {});
doc.base = {
vaultUpdateId: response.vaultUpdateId,
hash: pending.snapshot.hash
};
doc.base = { vaultUpdateId: response.vaultUpdateId, hash: pending.hash };
this.clearRejections(next, pending.documentId);
delete doc.remote;
delete next.pending;
await this.database.commit(next);
const path = next.local[pending.documentId];
await this.cacheContent(
response.vaultUpdateId,
path,
base64ToBytes(pending.snapshot.contentBase64)
);
await this.cacheSubmission(pending, response.vaultUpdateId);
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
message: "Content committed",
details: {
type: SyncType.UPDATE,
relativePath:
next.local[pending.documentId] ?? pending.documentId
relativePath: next.local[pending.documentId] ?? pending.documentId
},
author: response.userId,
timestamp: new Date(response.updatedDate)
@ -1173,4 +1118,19 @@ export class Syncer {
}
}
}
/** Warm the optional diff cache using the wire payload, never another saved copy. */
private async cacheSubmission(pending: Extract<PendingRequest, { type: "content" }>, version: number): Promise<void> {
const { content, parentVersionId } = pending.request;
let bytes: Uint8Array;
if (content.type === "Snapshot") bytes = base64ToBytes(content.value);
else {
const parent = parentVersionId === null ? undefined : this.contentCache.get(parentVersionId);
// A restart may have evicted the parent. Future uploads can use snapshots.
if (!parent) return;
const text = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode(parent);
bytes = new TextEncoder().encode(undiff(text, content.value));
}
await this.cacheContent(version, this.database.state.local[pending.documentId], bytes);
}
}