andras/vault-link
andras/vault-link
1
0
Fork 0
Code Issues Pull requests 19 Projects Releases 55 Packages Wiki Activity Actions 24

Bug fixes

Browse source
This commit is contained in:
Andras Schmelczer 2026-03-15 13:27:18 +00:00
commit df37e6c236
15 changed files with 628 additions and 153 deletions
Download patch file Download diff file Expand all files Collapse all files

8
frontend/sync-client/src/errors/http-client-error.ts Normal file
View file

@ -0,0 +1,8 @@
export class HttpClientError extends Error {
public readonly status: number;
public constructor(status: number, message: string) {
super(message);
this.name = "HttpClientError";
this.status = status;
}
}

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

@ -64,32 +64,45 @@ export class Database {
) {
initialState ??= {};
this.documents =
initialState.documents?.map(({ relativePath, ...metadata }) => ({
const validDocuments = (initialState.documents ?? []).filter(
(doc) =>
this.validateStoredField(doc, "relativePath", "string") &&
this.validateStoredField(doc, "documentId", "string") &&
this.validateStoredField(doc, "parentVersionId", "number")
);
this.documents = validDocuments.map(
({ relativePath, ...metadata }) => ({
relativePath,
metadata,
isDeleted: false,
parallelVersion: 0
})) ?? [];
})
);
if (initialState.pendingDocuments) {
for (const pending of initialState.pendingDocuments) {
const existing =
this.getLatestDocumentByRelativePath(
pending.relativePath
);
this.documents.push({
relativePath: pending.relativePath,
metadata: undefined,
isDeleted: false,
parallelVersion:
existing !== undefined
? existing.parallelVersion + 1
: 0,
originalCreationPath: pending.originalCreationPath,
idempotencyKey: pending.idempotencyKey
});
}
const validPendingDocuments = (
initialState.pendingDocuments ?? []
).filter(
(doc) =>
this.validateStoredField(doc, "relativePath", "string") &&
this.validateStoredField(doc, "idempotencyKey", "string")
);
for (const pending of validPendingDocuments) {
const existing = this.getLatestDocumentByRelativePath(
pending.relativePath
);
this.documents.push({
relativePath: pending.relativePath,
metadata: undefined,
isDeleted: false,
parallelVersion:
existing !== undefined
? existing.parallelVersion + 1
: 0,
originalCreationPath: pending.originalCreationPath,
idempotencyKey: pending.idempotencyKey
});
}
this.ensureConsistency();
@ -106,6 +119,25 @@ export class Database {
});
}
private validateStoredField(
doc: object,
field: string,
expectedType: "string" | "number"
): boolean {
const value = (doc as Record<string, unknown>)[field];
if (
typeof value !== expectedType ||
(expectedType === "string" && !value) ||
(expectedType === "number" && isNaN(value as number))
) {
this.logger.warn(
`Skipping stored document with invalid ${field}: ${JSON.stringify(doc)}`
);
return false;
}
return true;
}
public get length(): number {
return this.documents.length;
}
@ -301,7 +333,7 @@ export class Database {
({ relativePath, metadata }) => ({
relativePath,
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
...metadata! // `resolvedDocuments` only returns docs with metadata set
...metadata! // filtered to only docs with metadata set
})
),
pendingDocuments: this.pendingDocuments.map(
@ -316,6 +348,25 @@ export class Database {
}
private ensureConsistency(): void {
// Check for duplicate documentIds across ALL documents with metadata,
// not just the deduplicated resolvedDocuments view. A duplicate on a
// lower-parallelVersion record would otherwise go undetected.
const allWithMetadata = this.documents
// eslint-disable-next-line no-restricted-syntax -- Type narrowing, not removing a specific item
.filter((d) => d.metadata !== undefined);
const documentIdSet = new Set<string>();
for (const doc of allWithMetadata) {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const docId = doc.metadata!.documentId;
if (documentIdSet.has(docId)) {
throw new Error(
`Duplicate documentId ${docId} found in database`
);
}
documentIdSet.add(docId);
}
// Also check the deduplicated view for path-level invariants
const idToPath = new Map<string, string[]>();
this.resolvedDocuments.forEach(({ relativePath, metadata }) => {

75
frontend/sync-client/src/services/sync-service.ts
View file

@ -9,6 +9,7 @@ import type { Settings } from "../persistence/settings";
import type { FetchController } from "./fetch-controller";
import { sleep } from "../utils/sleep";
import { SyncResetError } from "../errors/sync-reset-error";
import { HttpClientError } from "../errors/http-client-error";
import type { SerializedError } from "./types/SerializedError";
import type { DocumentVersionWithoutContent } from "./types/DocumentVersionWithoutContent";
import type { DocumentUpdateResponse } from "./types/DocumentUpdateResponse";
@ -65,6 +66,17 @@ export class SyncService {
return result;
}
private static async throwHttpError(
response: Response,
context: string
): Promise<never> {
const message = `${context}: ${await SyncService.errorFromResponse(response)}`;
if (response.status >= 400 && response.status < 500) {
throw new HttpClientError(response.status, message);
}
throw new Error(message);
}
public async create({
relativePath,
contentBytes,
@ -98,10 +110,9 @@ export class SyncService {
});
if (!response.ok) {
throw new Error(
`Failed to create document: ${await SyncService.errorFromResponse(
response
)}`
await SyncService.throwHttpError(
response,
"Failed to create document"
);
}
@ -146,10 +157,9 @@ export class SyncService {
);
if (!response.ok) {
throw new Error(
`Failed to update document: ${await SyncService.errorFromResponse(
response
)}`
await SyncService.throwHttpError(
response,
"Failed to update document"
);
}
@ -157,8 +167,7 @@ export class SyncService {
(await response.json()) as DocumentUpdateResponse; // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
this.logger.debug(
`Updated document ${JSON.stringify(result)} with id ${
result.documentId
`Updated document ${JSON.stringify(result)} with id ${result.documentId
}}`
);
@ -199,10 +208,9 @@ export class SyncService {
);
if (!response.ok) {
throw new Error(
`Failed to update document: ${await SyncService.errorFromResponse(
response
)}`
await SyncService.throwHttpError(
response,
"Failed to update document"
);
}
@ -210,8 +218,7 @@ export class SyncService {
(await response.json()) as DocumentUpdateResponse; // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
this.logger.debug(
`Updated document ${JSON.stringify(result)} with id ${
result.documentId
`Updated document ${JSON.stringify(result)} with id ${result.documentId
}}`
);
@ -245,10 +252,9 @@ export class SyncService {
);
if (!response.ok) {
throw new Error(
`Failed to delete document: ${await SyncService.errorFromResponse(
response
)}`
await SyncService.throwHttpError(
response,
"Failed to delete document"
);
}
@ -279,10 +285,9 @@ export class SyncService {
);
if (!response.ok) {
throw new Error(
`Failed to get document: ${await SyncService.errorFromResponse(
response
)}`
await SyncService.throwHttpError(
response,
"Failed to get document"
);
}
@ -317,10 +322,9 @@ export class SyncService {
);
if (!response.ok) {
throw new Error(
`Failed to get document: ${await SyncService.errorFromResponse(
response
)}`
await SyncService.throwHttpError(
response,
"Failed to get document"
);
}
@ -338,7 +342,7 @@ export class SyncService {
return this.retryForever(async () => {
this.logger.debug(
"Getting all documents" +
(since != null ? ` since ${since}` : "")
(since != null ? ` since ${since}` : "")
);
const url = new URL(this.getUrl("/documents"));
@ -350,10 +354,9 @@ export class SyncService {
});
if (!response.ok) {
throw new Error(
`Failed to get documents: ${await SyncService.errorFromResponse(
response
)}`
await SyncService.throwHttpError(
response,
"Failed to get documents"
);
}
@ -464,6 +467,12 @@ export class SyncService {
throw e;
}
// Don't retry 4xx client errors — the request itself is wrong
// and retrying won't help
if (e instanceof HttpClientError) {
throw e;
}
const retryInterval =
this.settings.getSettings().networkRetryIntervalMs;
this.logger.error(

4
frontend/sync-client/src/services/websocket-manager.ts
View file

@ -109,6 +109,10 @@ export class WebSocketManager {
await awaitAll(this.outstandingPromises);
}
public hasOutstandingWork(): boolean {
return this.outstandingPromises.length > 0;
}
public sendHandshakeMessage(
message: WebSocketClientMessage & { type: "handshake" }
): void {

50
frontend/sync-client/src/sync-client.ts
View file

@ -18,6 +18,7 @@ import type { NetworkConnectionStatus } from "./types/network-connection-status"
import { DocumentSyncStatus } from "./types/document-sync-status";
import { WebSocketManager } from "./services/websocket-manager";
import { createClientId } from "./utils/create-client-id";
import { SyncResetError } from "./errors/sync-reset-error";
import { CursorTracker } from "./sync-operations/cursor-tracker";
import type { CursorSpan } from "./services/types/CursorSpan";
import type { MaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
@ -424,8 +425,21 @@ export class SyncClient {
public async waitUntilFinished(): Promise<void> {
this.checkIfDestroyed("waitUntilIdle");
await this.syncer.waitUntilFinished();
await this.webSocketManager.waitUntilFinished();
// Loop until both sync queue and WebSocket handlers are
// simultaneously idle. WS handlers can enqueue new sync
// operations, and completed sync operations can trigger
// broadcasts that create new WS handler promises.
let iteration = 0;
while (true) {
iteration++;
this.logger.info(`waitUntilFinished: iteration ${iteration}`);
await this.webSocketManager.waitUntilFinished();
await this.syncer.waitUntilFinished();
// Check if anything new arrived while we were waiting
if (!this.webSocketManager.hasOutstandingWork()) {
break;
}
}
await this.database.save(); // flush all changes to disk
}
@ -476,10 +490,40 @@ export class SyncClient {
this.hasFinishedOfflineSync = true;
}
/**
* Hard pause: aborts all in-flight HTTP operations via FetchController reset.
* Used when the SyncClient is being destroyed or fully reset (connection
* settings changed). This is the nuclear option — every outstanding fetch
* is rejected with SyncResetError so the queue drains immediately.
*/
private async pause(): Promise<void> {
this.hasFinishedOfflineSync = false;
this.fetchController.startReset();
try {
await this.webSocketManager.stop();
await this.waitUntilFinished();
} catch (e) {
// SyncResetError is expected here — we just called startReset()
// which rejects in-flight fetches. Only re-throw non-reset errors
// (after ensuring the FetchController is left in a usable state).
this.fetchController.finishReset();
if (!(e instanceof SyncResetError)) {
throw e;
}
}
}
/**
* Soft pause: stops the WebSocket and clears the sync queue, but lets
* in-flight HTTP operations complete naturally. Used when the user toggles
* sync off — we don't want to abort creates/updates that are mid-flight
* because they'd just be re-queued on re-enable, potentially leading to
* an infinite retry loop with flaky connections.
*/
private async softPause(): Promise<void> {
this.hasFinishedOfflineSync = false;
await this.webSocketManager.stop();
this.syncer.reset();
await this.waitUntilFinished();
}
@ -509,7 +553,7 @@ export class SyncClient {
if (newSettings.isSyncEnabled) {
await this.startSyncing();
} else {
await this.pause();
await this.softPause();
}
}

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

@ -71,6 +71,10 @@ export class Syncer {
if (isConnected) {
// The JS WebSocket API doesn't support setting headers, so we have to send the token as a message
this.sendHandshakeMessage();
} else {
// Clear so that the next reconnect re-runs scheduleSyncForOfflineChanges
// instead of returning the stale resolved promise.
this.runningScheduleSyncForOfflineChanges = undefined;
}
});
this.webSocketManager.onRemoteVaultUpdateReceived.add(
@ -267,7 +271,7 @@ export class Syncer {
public async waitUntilFinished(): Promise<void> {
await this.runningScheduleSyncForOfflineChanges;
await this.syncQueue.onIdle(); // Wait for queue to be empty and running tasks to finish
await this.syncQueue.onIdle();
}
public async syncRemotelyUpdatedFile(
@ -330,19 +334,19 @@ export class Syncer {
remoteVersion.documentId
);
await this.enqueueSyncOperation(
async () =>
this.unrestrictedSyncer.unrestrictedSyncRemotelyUpdatedFile(
async () => {
await this.unrestrictedSyncer.unrestrictedSyncRemotelyUpdatedFile(
remoteVersion,
document
),
);
this.database.addSeenUpdateId(remoteVersion.vaultUpdateId);
},
[
document?.relativePath,
remoteVersion.relativePath,
remoteVersion.documentId
]
);
this.database.addSeenUpdateId(remoteVersion.vaultUpdateId);
}
private async internalScheduleSyncForOfflineChanges(): Promise<void> {
@ -371,9 +375,12 @@ export class Syncer {
}
const instructions: (Instruction | undefined)[] = await awaitAll(
allLocalFiles.map(async (relativePath) => {
if (
const existingMetadata =
this.database.getLatestDocumentByRelativePath(relativePath)
?.metadata !== undefined
?.metadata;
if (
existingMetadata !== undefined &&
existingMetadata.parentVersionId > 0
) {
this.logger.debug(
`Document ${relativePath} might have been updated locally, scheduling sync to validate and update it`
@ -382,12 +389,27 @@ export class Syncer {
return { type: "update", relativePath } as Instruction;
}
// Perhaps the file has been moved; let's check by looking at the deleted files
const contentHash = await this.syncQueue.add(async () => {
// Perhaps the file has been moved; let's check by looking at the deleted files.
// Skip reading oversized files into memory for hash computation —
// they can't participate in move detection and will be scheduled as creates.
const hashResult = await this.syncQueue.add(async () => {
try {
const sizeInBytes =
await this.operations.getFileSize(relativePath);
const sizeInMB = Math.ceil(
sizeInBytes / 1024 / 1024
);
const { maxFileSizeMB } =
this.settings.getSettings();
if (sizeInMB > maxFileSizeMB) {
// File exceeds size limit — skip hash-based move
// detection and schedule as a create instead
return { skippedOversized: true } as const;
}
const contentBytes =
await this.operations.read(relativePath); // this can throw FileNotFoundError
return hash(contentBytes);
return { hash: hash(contentBytes) } as const;
} catch (e) {
if (
e instanceof Error &&
@ -399,15 +421,21 @@ export class Syncer {
}
});
if (contentHash == undefined) {
if (hashResult == undefined) {
// The file was deleted before we had a chance to read it, no need to sync it here
return;
}
const originalFile = findMatchingFile(
contentHash,
locallyPossiblyDeletedFiles
);
const contentHash =
"hash" in hashResult ? hashResult.hash : undefined;
const originalFile =
contentHash != undefined
? findMatchingFile(
contentHash,
locallyPossiblyDeletedFiles
)
: undefined;
if (originalFile !== undefined) {
// `originalFile` hasn't been deleted but it got moved instead
/* eslint-disable no-restricted-syntax -- Comparing by property, not direct equality */
@ -505,12 +533,25 @@ export class Syncer {
//
// The result type needs special handling since syncQueue.add() can
// return undefined when the queue is paused/cleared.
const result = await this.syncQueue.add(async () =>
this.updatedDocumentsByPathAndKeysLocks.withLock(
filteredKeys,
operation
)
);
const result = await this.syncQueue.add(async () => {
try {
return await this.updatedDocumentsByPathAndKeysLocks.withLock(
filteredKeys,
operation
);
} catch (e) {
// Catch all errors to prevent unhandled promise rejections.
// SyncResetError: lock waiter rejected during reset (expected).
// Other errors: logged by executeSync's history entry, will
// be retried on the next scheduleSyncForOfflineChanges cycle.
if (!(e instanceof SyncResetError)) {
this.logger.info(
`Sync operation failed, will retry on next cycle: ${e}`
);
}
return undefined;
}
});
return result as T;
}
}

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

@ -83,6 +83,15 @@ export class UnrestrictedSyncer {
doc.idempotencyKey !== undefined &&
resolved.has(doc.idempotencyKey)
) {
// Check if document was removed by a concurrent operation
// (e.g., a delete) between the snapshot and now
if (!this.database.containsDocument(doc)) {
this.logger.info(
`Pending doc at ${doc.relativePath} was removed during key resolution, skipping`
);
continue;
}
const documentId = resolved.get(doc.idempotencyKey)!; // eslint-disable-line @typescript-eslint/no-non-null-assertion
// Skip if this documentId is already assigned to another document
@ -160,7 +169,14 @@ export class UnrestrictedSyncer {
let response: DocumentVersion | DocumentUpdateResponse | undefined =
undefined;
if (document.metadata === undefined) {
if (
document.metadata === undefined ||
document.metadata.parentVersionId === 0
) {
// parentVersionId === 0 occurs when resolveIdempotencyKeys
// assigned a documentId but hasn't synced yet. Treat as a
// create — the server will recognise the idempotency key
// and return the existing document.
response = await this.syncService.create({
relativePath: originalRelativePath,
contentBytes,
@ -188,16 +204,22 @@ export class UnrestrictedSyncer {
(await this.serverConfig.getConfig())
.mergeableFileExtensions
);
// Snapshot parentVersionId atomically with the cache
// lookup. document.metadata is a mutable shared
// reference — a concurrent operation could update
// parentVersionId between the cache lookup and the
// putText call, causing a diff/version mismatch.
const parentVersionIdForUpdate =
document.metadata.parentVersionId;
const cachedVersion = this.contentCache.get(
document.metadata.parentVersionId
parentVersionIdForUpdate
);
response =
isText && cachedVersion !== undefined
? await this.syncService.putText({
documentId: document.metadata.documentId,
parentVersionId:
document.metadata.parentVersionId,
parentVersionId: parentVersionIdForUpdate,
relativePath: document.relativePath,
content: diff(
new TextDecoder().decode(cachedVersion),
@ -206,8 +228,7 @@ export class UnrestrictedSyncer {
})
: await this.syncService.putBinary({
documentId: document.metadata.documentId,
parentVersionId:
document.metadata.parentVersionId,
parentVersionId: parentVersionIdForUpdate,
relativePath: document.relativePath,
contentBytes
});
@ -522,6 +543,31 @@ export class UnrestrictedSyncer {
this.logger.info(
`Document ${document.relativePath} has been deleted before we could finish updating it`
);
// Assign metadata so the pending delete can inform the server
if (document.metadata === undefined) {
const existingWithSameId =
this.database.getDocumentByDocumentId(
response.documentId
);
if (
existingWithSameId !== undefined &&
existingWithSameId !== document
) {
// Another doc already has this documentId — the server
// knows about it. Just remove this stale pending doc.
this.database.removeDocument(document);
} else {
this.database.updateDocumentMetadata(
{
documentId: response.documentId,
parentVersionId: response.vaultUpdateId,
hash: contentHash,
remoteRelativePath: response.relativePath
},
document
);
}
}
this.database.addSeenUpdateId(response.vaultUpdateId);
return;
}
@ -615,18 +661,9 @@ export class UnrestrictedSyncer {
if (!("type" in response) || response.type === "MergingUpdate") {
const responseBytes = base64ToBytes(response.contentBase64);
contentHash = hash(responseBytes);
this.database.updateDocumentMetadata(
{
documentId: response.documentId,
parentVersionId: response.vaultUpdateId,
hash: contentHash,
remoteRelativePath: response.relativePath
},
document
);
// Write file BEFORE updating metadata so that if the write fails,
// metadata doesn't point to a version whose content was never written.
await this.operations.write(
actualPath,
originalContentBytes,
@ -642,27 +679,90 @@ export class UnrestrictedSyncer {
);
}
// Re-read and re-hash after write because the 3-way merge in
// operations.write() may produce content different from responseBytes.
const actualContent = await this.operations.read(actualPath);
const actualHash = hash(actualContent);
// The document may have been removed by a concurrent operation
// (e.g., a delete) during the awaited file write/read above.
// The file is safely on disk; recovery will re-detect it.
if (!this.database.containsDocument(document)) {
this.logger.info(
`Document ${document.relativePath} was removed during sync, skipping metadata update`
);
return;
}
this.database.updateDocumentMetadata(
{
documentId: response.documentId,
parentVersionId: response.vaultUpdateId,
hash: actualHash,
remoteRelativePath: response.relativePath
},
document
);
// Cache the SERVER's content (responseBytes), not the local
// content (actualContent). The cache is used to compute diffs
// for subsequent updates: diff(cached, newFileContent). The
// server applies this diff against its content at
// parentVersionId, which is responseBytes. Using actualContent
// would produce diffs that don't match the server's state.
await this.updateCache(
response.vaultUpdateId,
responseBytes,
actualPath
);
} else {
this.database.updateDocumentMetadata(
{
documentId: response.documentId,
parentVersionId: response.vaultUpdateId,
hash: contentHash,
remoteRelativePath: response.relativePath
},
document
);
await this.updateCache(
response.vaultUpdateId,
originalContentBytes,
actualPath
);
// FastForwardUpdate — the server accepted our content as-is,
// UNLESS this was an idempotent create return (the server
// returned the original version, whose content may differ from
// what we sent). Detect this by comparing contentSize.
const serverContentMatchesLocal =
!("contentSize" in response) ||
response.contentSize === originalContentBytes.length;
if (serverContentMatchesLocal) {
this.database.updateDocumentMetadata(
{
documentId: response.documentId,
parentVersionId: response.vaultUpdateId,
hash: contentHash,
remoteRelativePath: response.relativePath
},
document
);
await this.updateCache(
response.vaultUpdateId,
originalContentBytes,
actualPath
);
} else {
// The server returned a stale idempotent version. Fetch
// the actual content so the cache stays consistent, then
// the hash mismatch will trigger a follow-up update sync.
const serverContent =
await this.syncService.getDocumentVersionContent({
documentId: response.documentId,
vaultUpdateId: response.vaultUpdateId
});
this.database.updateDocumentMetadata(
{
documentId: response.documentId,
parentVersionId: response.vaultUpdateId,
hash: hash(serverContent),
remoteRelativePath: response.relativePath
},
document
);
await this.updateCache(
response.vaultUpdateId,
serverContent,
actualPath
);
}
}
this.database.addSeenUpdateId(response.vaultUpdateId);
@ -672,9 +772,10 @@ export class UnrestrictedSyncer {
sizeInBytes: number,
relativePath: RelativePath
): CommonHistoryEntry | undefined {
const sizeInMB = Math.round(sizeInBytes / 1024 / 1024);
const { maxFileSizeMB } = this.settings.getSettings();
if (sizeInMB > maxFileSizeMB) {
const maxFileSizeBytes = maxFileSizeMB * 1024 * 1024;
if (sizeInBytes > maxFileSizeBytes) {
const sizeInMB = (sizeInBytes / 1024 / 1024).toFixed(1);
return {
status: SyncStatus.SKIPPED,
details: {

6
frontend/sync-client/src/utils/data-structures/locks.ts
View file

@ -85,7 +85,11 @@ export class Locks<T> {
reject(new SyncResetError());
}
}
this.locked.clear();
// Do NOT clear this.locked — let running operations release their own
// locks via the finally block in withLock. Clearing this.locked would
// allow new operations to acquire locks on keys still held by in-flight
// operations, breaking mutual exclusion.
this.waiters.clear();
}