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, undiff } from "reconcile-text"; import type { Database, EngineState, PendingRequest, UnconfirmedRequest, ContentHead } from "../persistence/database"; import { emptyState } from "../persistence/database"; import { toStoredSnapshot, type StoredSnapshot } from "../snapshot"; import type { FileOperations, FileWrite } from "../file-operations/file-operations"; import type { Settings } from "../persistence/settings"; import type { SyncService } from "../services/sync-service"; import type { ServerConfig } from "../services/server-config"; import type { WebSocketManager } from "../services/websocket-manager"; import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent"; import type { EventBatch } from "../services/types/EventBatch"; import type { EventRecord } from "../services/types/EventRecord"; import type { FileManifest } from "../services/types/FileManifest"; import type { VaultSnapshot } from "../services/types/VaultSnapshot"; import type { PushContent } from "../services/types/PushContent"; import type { Logger } from "../tracing/logger"; import { EventListeners } from "../utils/data-structures/event-listeners"; import { VaultHistoryStatus } from "../types/vault-history-status"; import { mergeContent } from "./content"; import { scanLocalFiles, type LocalChange } from "./scan"; import { mergeFileManifests, resolvePaths, sameFileManifest, validateFileManifest } from "./file-manifest"; import { removeFromArray } from "../utils/remove-from-array"; import { isInternalPath } from "../utils/portable-path"; import { isBinary } from "../utils/is-binary"; import { isFileTypeMergeable } from "../utils/is-file-type-mergeable"; import type { FixedSizeDocumentCache } from "../utils/data-structures/fix-sized-cache"; import { AuthenticationError, LocalChangesDuringReconciliation, PermanentSyncError, ServerVersionMismatchError, SyncResetError, ServerHistoryChangedError } from "../errors/errors"; export class Syncer { public readonly onReadyForEvents = new EventListeners<() => void>(); public readonly onServerHistoryChanged = new EventListeners<() => void>(); public readonly onRemainingOperationsCountChanged = new EventListeners< (count: number) => unknown >(); private running?: Promise; private timer?: ReturnType; private stopped = true; private dirty = false; private localGeneration = 0; private failure?: unknown; private readonly changes: LocalChange[]; private notificationSaving: Promise = Promise.resolve(); private notificationSaveFailed = false; private readonly pendingPaths = new Set(); private unsyncablePaths = new Set(); private hasScanned = false; public constructor( deviceId: string, private readonly logger: Logger, private readonly database: Database, private readonly settings: Settings, private readonly service: SyncService, websocket: WebSocketManager, private readonly files: FileOperations, private readonly serverConfig: ServerConfig, private readonly history: SyncHistory, private readonly contentCache: FixedSizeDocumentCache, private readonly localChangeStorage: { entries: LocalChange[]; save: (entries: LocalChange[]) => Promise; } = { entries: [], save: async () => {} } ) { this.changes = localChangeStorage.entries; websocket.onWebSocketStatusChanged.add((connected) => { if (connected) { websocket.sendHandshakeMessage({ type: "handshake", token: settings.getSettings().token, deviceId }); } // Failed upgrades and lost transports still need HTTP catchup, // including when the optional polling interval is disabled. this.wake(); }); websocket.onRemoteVaultUpdateReceived.add(async () => { this.wake(); }); } public get isFirstSyncComplete(): boolean { return this.database.state.lastSeenUpdateId !== undefined; } public get hasLocalChanges(): boolean { return this.changes.length > 0; } public get isBusy(): boolean { return ( !!this.running || this.dirty || !!this.database.state.pending || !!this.failure || !!this.rejectionMessage ); } public isDocumentUpToDate(path: string): boolean { if ( this.database.state.lastSeenUpdateId === undefined || !this.hasScanned || this.pendingPaths.has(path) || this.unsyncablePaths.has(path) || this.settings.isIgnored(path) ) { return false; } const record = this.database.getLatestDocumentByRelativePath(path); if (!record) return false; const document = this.database.state.documents[record.documentId]; if ( document?.observedHash === undefined || !document.base || document.observedHash !== document.base.hash || this.database.state.fileManifest.entries[record.documentId] !== path ) { return false; } const remote = document.remote; if (remote && remote.vaultUpdateId > (document.base.vaultUpdateId ?? 0)) return false; const { pending } = this.database.state; if ( pending?.type === "content" && pending.documentId === record.documentId ) { return false; } if ( pending?.type === "fileManifest" && pending.request.entries[record.documentId] !== this.database.state.fileManifest.entries[record.documentId] ) { return false; } return true; } public start(): void { this.stopped = false; this.failure = undefined; this.hasScanned = false; this.wake(); } public async stop(): Promise { this.stopped = true; clearTimeout(this.timer); await this.running; } public wake(): void { if (this.stopped) { return; } this.dirty = true; if (this.running) { return; } clearTimeout(this.timer); this.running = this.run().finally(() => { this.running = undefined; this.onRemainingOperationsCountChanged.trigger( this.failure || this.database.state.pending || this.rejectionMessage ? 1 : 0 ); if (this.stopped) { return; } if (this.dirty && !this.failure) { this.wake(); return; } if ( this.failure instanceof PermanentSyncError || this.failure instanceof AuthenticationError || this.failure instanceof ServerVersionMismatchError ) { return; } const intervalMs = this.failure === undefined ? this.settings.getSettings().syncIntervalMs : this.settings.getSettings().networkRetryIntervalMs; if (intervalMs === undefined || intervalMs === 0) { return; } this.timer = setTimeout(() => { this.wake(); }, intervalMs); }); } public async waitUntilFinished(): Promise { while (this.running) { await this.running; } if (this.failure) { throw this.failure; } if (this.rejectionMessage) throw new PermanentSyncError(this.rejectionMessage); } public async syncLocallyCreatedFile(_path: string): Promise { this.localGeneration++; this.pendingPaths.add(_path); await this.recordLocalChange({ type: "create", path: _path }); } public async syncLocallyDeletedFile(_path: string): Promise { this.localGeneration++; this.pendingPaths.add(_path); await this.recordLocalChange({ type: "delete", path: _path }); } public async syncLocallyUpdatedFile({ oldPath, relativePath }: { oldPath?: string; relativePath: string; }): Promise { // Content snapshots are re-read before writing. Only namespace // notifications invalidate an identity plan; keystrokes must not starve // a complete scan of unrelated documents. if (oldPath !== undefined && oldPath !== "") this.localGeneration++; this.pendingPaths.add(relativePath); if (oldPath) this.pendingPaths.add(oldPath); if ( oldPath && !isInternalPath(oldPath) && !isInternalPath(relativePath) ) { await this.recordLocalChange({ type: "move", oldPath, relativePath }); } this.wake(); } /** Reload an uncertain metadata save before starting another attempt. */ public async reloadLocalState(): Promise { await this.flushLocalChanges(); await this.database.recoverPersistence(); } private async recordLocalChange(change: LocalChange): Promise { change.changeId = uuid(); const previous = [...this.changes]; this.changes.push(change); const saving = this.notificationSaving.then(async () => { await this.files.captureChange(change, previous); await this.localChangeStorage.save(structuredClone(this.changes)); this.notificationSaveFailed = false; }); this.notificationSaving = saving.catch(() => { this.notificationSaveFailed = true; }); try { await saving; } finally { this.wake(); } } public async flushLocalChanges(): Promise { await this.notificationSaving; if (this.notificationSaveFailed) { await this.localChangeStorage.save(structuredClone(this.changes)); this.notificationSaveFailed = false; } } private async run(): Promise { this.onRemainingOperationsCountChanged.trigger(1); this.failure = undefined; try { await this.reloadLocalState(); if ( this.isFirstSyncComplete && this.service.vaultHistoryStatus === VaultHistoryStatus.MISSING ) await this.recoverServerHistory(); if (!this.isFirstSyncComplete) await this.service.resetHistory(); await this.serverConfig.initialize(); if (!this.isFirstSyncComplete) await this.bootstrap(); if (!this.stopped) this.onReadyForEvents.trigger(); do { this.dirty = false; // An unknown accepted request must be resolved before observing a // newer base, otherwise we can merge our own edit a second time. if (this.database.state.pending) { await this.finishPending(); } const batch = await this.service.getEvents( this.database.state.lastSeenUpdateId! ); await this.incorporateEventBatch(batch); if (await this.preparePush()) { await this.finishPending(); this.dirty = true; } } while (this.dirty && !this.stopped); if (!this.dirty && !this.database.state.pending) this.pendingPaths.clear(); } catch (error) { if (error instanceof ServerHistoryChangedError && !this.stopped) { try { await this.recoverServerHistory(); this.dirty = true; return; } catch (recoveryError) { error = recoveryError; } } if (error instanceof LocalChangesDuringReconciliation) { this.dirty = true; return; } if (!(error instanceof SyncResetError && this.stopped)) { this.failure = error; this.serverConfig.reset(); this.logger.error(`Sync paused for retry: ${String(error)}`); } } } private async recoverServerHistory(): Promise { this.onServerHistoryChanged.trigger(); this.contentCache.reset(); this.hasScanned = false; await this.scan(); const guard = this.localGuard(); const previous = this.database.state; const next = emptyState(previous.vaultKey); next.lastAppliedLocalChangeId = previous.lastAppliedLocalChangeId; for (const [id, path] of Object.entries(previous.local)) { const info = await this.files.fs.stat(path); if (info?.kind !== "file") continue; const doc = previous.documents[id]; if ( this.settings.isIgnored(path) || this.settings.isOversized(info.size) ) { next.local[id] = path; next.documents[id] = { observedHash: doc?.observedHash ?? "" }; (next.excluded ??= {})[id] = previous.fileManifest.entries[id] ?? null; continue; } const snapshot = await this.files.snapshot(path); if (!snapshot) throw new LocalChangesDuringReconciliation(); const clean = doc?.base?.hash === snapshot.hash; const localId = doc?.base && !clean ? uuid() : id; next.local[localId] = path; next.documents[localId] = { observedHash: snapshot.hash, base: clean ? { vaultUpdateId: null, hash: snapshot.hash } : undefined }; } guard(); // An absent cursor durably abandons all old requests and version IDs. // The next bootstrap clears the transport checkpoint before any request. await this.database.commit(next); } private localGuard(): () => void { const generation = this.localGeneration; return () => { if (generation !== this.localGeneration) { throw new LocalChangesDuringReconciliation(); } }; } private async cacheContent( updateId: number, path: string | undefined, content: Uint8Array ): Promise { if ( path !== undefined && !isBinary(content) && isFileTypeMergeable( path, (await this.serverConfig.getConfig()).mergeableFileExtensions ) ) { this.contentCache.put(updateId, content); } } private async pushContent( path: string, parentVersionId: number | undefined, snapshot: StoredSnapshot ): Promise { const content = base64ToBytes(snapshot.contentBase64); const parent = parentVersionId === undefined ? undefined : this.contentCache.get(parentVersionId); if ( parent !== undefined && !isBinary(content) && !isBinary(parent) && isFileTypeMergeable( path, (await this.serverConfig.getConfig()).mergeableFileExtensions ) ) { const decode = (bytes: Uint8Array): string => new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode(bytes); const value = diff(decode(parent), decode(content)); // Match the server's work bound; large edit sets remain syncable. if (value.length <= 10_000) return { type: "Diff", value }; } return { type: "Snapshot", value: snapshot.contentBase64 }; } private next(): EngineState { return structuredClone(this.database.state); } private async bootstrap(): Promise { const vaultSnapshot = await this.service.getVaultSnapshot(); await this.reconcile(vaultSnapshot.fileManifest, { cursor: vaultSnapshot.headEventId, initial: vaultSnapshot }); } private async scan(initial?: VaultSnapshot): Promise { await this.flushLocalChanges(); const hadChanges = this.changes.length > 0; this.unsyncablePaths = await scanLocalFiles( { next: this.next(), guard: this.localGuard(), changes: this.changes, files: this.files, ignored: (path) => this.settings.isIgnored(path), oversized: (size) => this.settings.isOversized(size), commit: async (next) => this.database.commit(next) }, initial ); this.hasScanned = true; if (hadChanges) await this.localChangeStorage.save(structuredClone(this.changes)); } private async incorporateEventBatch(batch: EventBatch): Promise { let cursor = this.database.state.lastSeenUpdateId ?? 0; let manifest = this.database.state.fileManifest; const heads: Record = {}; 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. for (;;) { const after = cursor; for (const event of batch.events) { if (event.eventId <= cursor) continue; if (event.eventId !== cursor + 1) throw new PermanentSyncError( "Non-contiguous event history; refusing to skip changes" ); cursor = event.eventId; if ( this.database.state.unconfirmed?.some( (pending) => pending.requestId === event.requestId ) === true ) receipts.push(event); if (event.type === "fileManifest") { if ( event.fileManifest.fileManifestId > manifest.fileManifestId ) manifest = event.fileManifest; } else if ( !heads[event.document.documentId] || heads[event.document.documentId].vaultUpdateId < event.document.vaultUpdateId ) { heads[event.document.documentId] = this.contentHead( event.document ); } } const end = batch.endEventId ?? batch.headEventId; if ( end !== cursor || end > batch.headEventId || (end === after && end < batch.headEventId) ) throw new PermanentSyncError("Incomplete event batch"); if (end === batch.headEventId) break; batch = await this.service.getEvents(cursor); } // Recover our submitted bases before folding later remote edits. A // permanent retry failure cannot cancel an earlier in-flight attempt. const next = this.next(); let acknowledged = false; for (const event of receipts) { const pending = next.unconfirmed?.find( (candidate) => candidate.requestId === event.requestId ); if (!pending) continue; if (event.type === "content" && pending.type === "content") { const doc = next.documents[pending.documentId]; if ( !doc.base || (doc.base.vaultUpdateId ?? 0) < event.eventId ) { doc.base = { vaultUpdateId: event.eventId, hash: pending.hash }; this.clearRejections(next, pending.documentId); } } else if ( event.type === "fileManifest" && pending.type === "fileManifest" ) { if (next.fileManifest.fileManifestId < event.eventId) { next.fileManifest = event.fileManifest; this.clearRejections(next, "manifest"); } } else throw new PermanentSyncError( "Request receipt type does not match submission" ); if (next.unconfirmed) removeFromArray(next.unconfirmed, pending); acknowledged = true; } if (acknowledged) await this.database.commit(next); // Three-way merging compares snapshots. Replaying intermediate remote // edits can discard a local edit even when the remote ends at its base. await this.reconcile(manifest, { cursor, heads }); } private contentHead( version: ContentHead ): ContentHead { return { vaultUpdateId: version.vaultUpdateId, contentSize: version.contentSize }; } private recordManifestApplication( before: Readonly>, after: Readonly>, writes: Readonly>, message: string ): void { for (const id of new Set([ ...Object.keys(before), ...Object.keys(after) ])) { const from = before[id]; const to = after[id]; if (from === to && !writes[id]) continue; this.history.addHistoryEntry({ status: SyncStatus.SUCCESS, message, details: from && to && from !== to ? { type: SyncType.MOVE, relativePath: to, movedFrom: from } : !from && to ? { type: SyncType.CREATE, relativePath: to } : from && !to ? { type: SyncType.DELETE, relativePath: from } : { type: SyncType.UPDATE, relativePath: to ?? from } }); } } private async remoteSnapshot( head: Pick< DocumentVersionWithoutContent, "documentId" | "vaultUpdateId" >, path: string ): Promise { const content = await this.service.getDocumentVersionContent(head); await this.cacheContent(head.vaultUpdateId, path, content); return toStoredSnapshot({ content }); } private async contentWrite( next: EngineState, id: string, head: ContentHead, path: string ): Promise { const document = (next.documents[id] ??= {}); if ( next.excluded?.[id] !== undefined || this.settings.isIgnored(path) || this.settings.isOversized(head.contentSize) ) { return undefined; } const currentPath = this.database.state.local[id]; const local = currentPath && (await this.files.fs.stat(currentPath))?.kind === "file" ? await this.files.snapshot(currentPath) : undefined; if ( local !== undefined && this.settings.isOversized(base64ToBytes(local.contentBase64).length) ) { return undefined; } if ( document.base?.vaultUpdateId === null && local && local.hash !== document.base.hash ) throw new LocalChangesDuringReconciliation(); const base = local && document.base ? document.base.vaultUpdateId === null ? local : await this.remoteSnapshot( { documentId: id, vaultUpdateId: document.base.vaultUpdateId }, path ) : 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({ documentId: id, vaultUpdateId: head.vaultUpdateId }, path); let result = remote; if (local !== undefined) { result = await mergeContent( path, base, local, remote, (await this.serverConfig.getConfig()).mergeableFileExtensions ); } document.base = { vaultUpdateId: head.vaultUpdateId, hash: remote.hash }; delete document.remote; document.observedHash = result.hash; return local?.contentBase64 === result.contentBase64 ? undefined : { expected: local, replacement: result }; } private async reconcile( remote: FileManifest, { cursor, initial, clearPending = false, heads }: { cursor?: number; initial?: VaultSnapshot; clearPending?: boolean; heads?: Record; } = {} ): Promise { validateFileManifest(remote.entries); await this.scan(initial); const before = { ...this.database.state.local }; const guard = this.localGuard(); const next = this.next(); if (cursor !== undefined) { next.lastSeenUpdateId = cursor; } if (clearPending) { delete next.pending; } if (remote.fileManifestId < next.fileManifest.fileManifestId) { await this.database.commit(next); return; } const base = { ...next.fileManifest.entries }; for (const [id, path] of Object.entries(next.excluded ?? {})) { if (path === null) delete base[id]; else base[id] = path; } const previousManifest = next.fileManifest.entries; next.local = mergeFileManifests(base, next.local, remote.entries); next.fileManifest = remote; 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 ( this.settings.isIgnored(oldPath) || this.unsyncablePaths.has(oldPath) || (path !== undefined && this.settings.isIgnored(path)) ) { next.excluded ??= {}; if (next.excluded[id] === undefined) next.excluded[id] = previousManifest[id] ?? null; next.local[id] = oldPath; } else if (next.excluded) delete next.excluded[id]; } for (const [id, path] of Object.entries(next.excluded ?? {})) { if ( before[id] === undefined && (path === null || !this.settings.isIgnored(path)) ) delete next.excluded![id]; } next.local = resolvePaths( next.local, remote.entries, Object.keys(next.excluded ?? {}), [...this.unsyncablePaths] ); if (initial) { for (const head of initial.documents) (next.documents[head.documentId] ??= {}).remote = this.contentHead(head); } const writes: Record = {}; for (const [id, path] of Object.entries(next.local)) { const doc = (next.documents[id] ??= {}); if (before[id] === undefined) delete doc.observedHash; if ( doc.observedHash !== undefined && doc.remote && doc.remote.vaultUpdateId <= (doc.base?.vaultUpdateId ?? 0) ) delete doc.remote; if ( next.excluded?.[id] === undefined && !this.settings.isIgnored(path) && (remote.entries[id] !== undefined || doc.remote !== undefined) && (!this.database.state.local[id] || !doc.base || doc.observedHash === undefined || (doc.remote?.vaultUpdateId ?? 0) > (doc.base.vaultUpdateId ?? 0)) ) { const knownHead = doc.remote; const head = knownHead ?? (await this.service.getDocumentMetadata(id)); if (!head) { throw new PermanentSyncError( `File manifest references missing content: ${id}` ); } 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, this.database.state.local, writes, "Remote change applied"); } private async preparePush(): Promise { if (this.stopped) { return false; } let guard = this.localGuard(); const canonical = resolvePaths( this.database.state.local, this.database.state.fileManifest.entries, Object.keys(this.database.state.excluded ?? {}), [...this.unsyncablePaths] ); if (!sameFileManifest(canonical, this.database.state.local)) { await this.files.apply( { ...this.next(), local: canonical }, {}, guard ); await this.scan(); guard = this.localGuard(); } // Resolve the map after restoring excluded server paths, then make // those allocations visible locally before submitting a manifest. const submitted = this.manifestToSubmit(); const local = { ...this.database.state.local }; for (const [id, path] of Object.entries(submitted)) if (this.database.state.excluded?.[id] === undefined) local[id] = path; if (!sameFileManifest(local, this.database.state.local)) { await this.files.apply({ ...this.next(), local }, {}, guard); await this.scan(); guard = this.localGuard(); } for (const [id, path] of Object.entries(this.database.state.local)) { if ( this.settings.isIgnored(path) || this.database.state.excluded?.[id] !== undefined || this.database.state.documents[id]?.observedHash === undefined || this.unsyncablePaths.has(path) || this.settings.isOversized( // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- Document heads are a sparse ID map at runtime. this.database.state.documents[id]?.remote?.contentSize ?? 0 ) ) { continue; } const snapshot = await this.files.snapshot(path); if ( !snapshot || this.settings.isOversized( base64ToBytes(snapshot.contentBase64).length ) ) { continue; } const { base } = this.database.state.documents[id]; if ( base?.hash === snapshot.hash || this.rejections.some( (request) => request.type === "content" && request.documentId === id && request.hash === snapshot.hash ) ) { continue; } guard(); await this.database.commit({ ...this.next(), pending: { type: "content", documentId: id, hash: snapshot.hash, request: { requestId: uuid(), parentVersionId: base?.vaultUpdateId ?? null, content: await this.pushContent( path, base?.vaultUpdateId ?? undefined, snapshot ) } } }); return true; } const entries = this.manifestToSubmit(); validateFileManifest(entries); if ( !sameFileManifest( entries, this.database.state.fileManifest.entries ) && !this.rejections.some( (request) => request.type === "fileManifest" && sameFileManifest(entries, request.entries) ) ) { guard(); await this.database.commit({ ...this.next(), pending: { type: "fileManifest", request: { requestId: uuid(), parentFileManifestId: this.database.state.fileManifest.fileManifestId, entries } } }); return true; } return false; } private manifestToSubmit(): Record { // Do not publish a new document until its initial content is acknowledged. const entries = { ...this.database.state.local }; for (const id of Object.keys(this.database.state.excluded ?? {})) { const remotePath = this.database.state.fileManifest.entries[id]; if (remotePath === undefined) delete entries[id]; else entries[id] = remotePath; } for (const id of Object.keys(entries)) { if ( !this.database.state.documents[id]?.base && this.database.state.fileManifest.entries[id] === undefined ) { delete entries[id]; } } return resolvePaths( entries, this.database.state.fileManifest.entries, Object.keys(this.database.state.excluded ?? {}), [ ...this.unsyncablePaths, ...Object.keys(this.database.state.excluded ?? {}).flatMap( (id) => this.database.state.local[id] ? [this.database.state.local[id]] : [] ) ] ); } private get rejections(): UnconfirmedRequest[] { return (this.database.state.unconfirmed ?? []).filter( (request) => request.rejection !== undefined ); } private get rejectionMessage(): string | undefined { const { state } = this.database; for (const request of this.rejections) { if (request.type === "content") { const id = request.documentId; const path = state.local[id]; const doc = state.documents[id]; if ( path !== undefined && doc?.observedHash === request.hash && doc.base?.hash !== request.hash && !this.settings.isIgnored(path) && state.excluded?.[id] === undefined ) return request.rejection; } else if ( sameFileManifest( this.manifestToSubmit(), request.entries ) && !sameFileManifest( state.fileManifest.entries, request.entries ) ) return request.rejection; } } private clearRejections(state: EngineState, id?: string): void { for (const request of state.unconfirmed ?? []) if ( id === undefined || (request.type === "content" ? request.documentId : "manifest") === id ) delete request.rejection; } /** An explicit reset or settings change permits fresh attempts after rejection. */ public async retryRejectedRequests(): Promise { await this.database.recoverPersistence(); if (!this.rejections.length) return; const next = this.next(); this.clearRejections(next); await this.database.commit(next); } private async sendPending(pending: PendingRequest, send: () => Promise): Promise { try { return await send(); } catch (error) { if (!(error instanceof PermanentSyncError)) throw error; const next = this.next(); (next.unconfirmed ??= []).push(requestReceipt(pending, error.message)); delete next.pending; await this.database.commit(next); this.history.addHistoryEntry({ status: SyncStatus.ERROR, message: error.message, details: { type: SyncType.SKIPPED, relativePath: pending.type === "content" ? (next.local[pending.documentId] ?? pending.documentId) : "File manifest" } }); return; } } private async finishPending(): Promise { 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 = await this.sendPending(pending, async () => this.service.pushFileManifest(pending.request)); if (!response) return; if (response.type === "StaleBase") { await this.reconcile(response, { clearPending: true }); } else { const next = this.next(); this.clearRejections(next, "manifest"); next.fileManifest = { fileManifestId: response.fileManifestId, entries: pending.request.entries }; delete next.pending; const previous = this.database.state.fileManifest.entries; await this.database.commit(next); this.recordManifestApplication(previous, next.fileManifest.entries, {}, "File manifest change committed"); } } else { 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.contentHead(response) } }); } else { const next = this.next(); const doc = (next.documents[pending.documentId] ??= {}); doc.base = { vaultUpdateId: response.vaultUpdateId, hash: pending.hash }; this.clearRejections(next, pending.documentId); delete doc.remote; delete next.pending; await this.database.commit(next); 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 }, author: response.userId, timestamp: new Date(response.updatedDate) }); } } } /** Warm the optional diff cache using the wire payload, never another saved copy. */ private async cacheSubmission(pending: Extract, version: number): Promise { 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); } }