split: sync-engine rewrite (sync-operations + sync-client.ts)
Replace the single unrestricted-syncer.ts with a two-loop architecture: - syncer.ts drains the FIFO wire queue (HTTP + WS handlers). - reconciler.ts moves files to make localPath match remoteRelativePath (topo-sorted move graph, in-memory cycle resolution with crash-safe swap markers). - sync-event-queue.ts holds the byDocId / byLocalPath indexes and the pending-create promise chain. - offline-change-detector.ts, expected-fs-events.ts, types.ts, and a rewritten cursor-tracker.ts / file-change-notifier.ts round it out. Plus sync-client.ts wiring, tracing/sync-history.ts updates, index.ts re-exports, and sync-client tsconfig/webpack/package.json.
This commit is contained in:
parent
0fda95ff8e
commit
42c9d55489
18 changed files with 5068 additions and 1257 deletions
|
|
@ -14,19 +14,17 @@
|
|||
},
|
||||
"devDependencies": {
|
||||
"byte-base64": "^1.1.0",
|
||||
"minimatch": "^10.0.1",
|
||||
"p-queue": "^8.1.0",
|
||||
"minimatch": "^10.1.1",
|
||||
"p-queue": "^9.0.1",
|
||||
"reconcile-text": "^0.8.0",
|
||||
"uuid": "^13.0.0",
|
||||
"@types/node": "^24.8.1",
|
||||
"ts-loader": "^9.5.2",
|
||||
"@types/node": "^25.0.2",
|
||||
"ts-loader": "^9.5.4",
|
||||
"tslib": "2.8.1",
|
||||
"tsx": "^4.20.6",
|
||||
"typescript": "5.8.3",
|
||||
"webpack": "^5.99.9",
|
||||
"tsx": "^4.21.0",
|
||||
"typescript": "5.9.3",
|
||||
"webpack": "^5.103.0",
|
||||
"webpack-cli": "^6.0.1",
|
||||
"webpack-merge": "^6.0.1",
|
||||
"@sentry/browser": "^10.8.0",
|
||||
"ws": "^8.18.3"
|
||||
"@sentry/browser": "^10.30.0"
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import { awaitAll } from "./utils/await-all";
|
|||
import { logToConsole } from "./utils/debugging/log-to-console";
|
||||
import { slowFetchFactory } from "./utils/debugging/slow-fetch-factory";
|
||||
import { slowWebSocketFactory } from "./utils/debugging/slow-web-socket-factory";
|
||||
import { InMemoryFileSystem } from "./utils/debugging/in-memory-file-system";
|
||||
import { getRandomColor } from "./utils/get-random-color";
|
||||
import { lineAndColumnToPosition } from "./utils/line-and-column-to-position";
|
||||
import { positionToLineAndColumn } from "./utils/position-to-line-and-column";
|
||||
|
|
@ -21,14 +22,19 @@ export {
|
|||
export { Logger, LogLevel, LogLine } from "./tracing/logger";
|
||||
export { type SyncSettings, DEFAULT_SETTINGS } from "./persistence/settings";
|
||||
export { rateLimit } from "./utils/rate-limit";
|
||||
export type { RelativePath, StoredDatabase } from "./persistence/database";
|
||||
export type {
|
||||
RelativePath,
|
||||
StoredSyncState as StoredDatabase,
|
||||
DocumentRecord
|
||||
} from "./sync-operations/types";
|
||||
export type { FileSystemOperations } from "./file-operations/filesystem-operations";
|
||||
export type { PersistenceProvider } from "./persistence/persistence";
|
||||
export type { CursorSpan } from "./services/types/CursorSpan";
|
||||
export type { ClientCursors } from "./services/types/ClientCursors";
|
||||
export type { NetworkConnectionStatus } from "./types/network-connection-status";
|
||||
export type { ServerVersionMismatchError } from "./services/server-version-mismatch-error";
|
||||
export type { AuthenticationError } from "./services/authentication-error";
|
||||
export type { ServerVersionMismatchError } from "./errors/server-version-mismatch-error";
|
||||
export type { AuthenticationError } from "./errors/authentication-error";
|
||||
export { SyncResetError } from "./errors/sync-reset-error";
|
||||
export type { MaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
|
||||
export { DocumentSyncStatus } from "./types/document-sync-status";
|
||||
export { SyncClient } from "./sync-client";
|
||||
|
|
@ -37,7 +43,8 @@ export type { TextWithCursors, CursorPosition } from "reconcile-text";
|
|||
export const debugging = {
|
||||
slowFetchFactory,
|
||||
slowWebSocketFactory,
|
||||
logToConsole
|
||||
logToConsole,
|
||||
InMemoryFileSystem
|
||||
};
|
||||
|
||||
export const utils = {
|
||||
|
|
|
|||
|
|
@ -2,8 +2,12 @@ import type { PersistenceProvider } from "./persistence/persistence";
|
|||
import type { HistoryEntry, HistoryStats } from "./tracing/sync-history";
|
||||
import { SyncHistory } from "./tracing/sync-history";
|
||||
import { Logger, LogLevel, LogLine } from "./tracing/logger";
|
||||
import type { RelativePath, StoredDatabase } from "./persistence/database";
|
||||
import { Database } from "./persistence/database";
|
||||
import type {
|
||||
DocumentId,
|
||||
RelativePath,
|
||||
StoredSyncState
|
||||
} from "./sync-operations/types";
|
||||
import { SyncEventQueue } from "./sync-operations/sync-event-queue";
|
||||
import * as Sentry from "@sentry/browser";
|
||||
import type { SyncSettings } from "./persistence/settings";
|
||||
import { DEFAULT_SETTINGS, Settings } from "./persistence/settings";
|
||||
|
|
@ -12,7 +16,6 @@ import { Syncer } from "./sync-operations/syncer";
|
|||
import type { FileSystemOperations } from "./file-operations/filesystem-operations";
|
||||
import { FileOperations } from "./file-operations/file-operations";
|
||||
import { FetchController } from "./services/fetch-controller";
|
||||
import { UnrestrictedSyncer } from "./sync-operations/unrestricted-syncer";
|
||||
import { rateLimit } from "./utils/rate-limit";
|
||||
import type { NetworkConnectionStatus } from "./types/network-connection-status";
|
||||
import { DocumentSyncStatus } from "./types/document-sync-status";
|
||||
|
|
@ -24,42 +27,46 @@ import type { MaybeOutdatedClientCursors } from "./types/maybe-outdated-client-c
|
|||
import { FileChangeNotifier } from "./sync-operations/file-change-notifier";
|
||||
import { FixedSizeDocumentCache } from "./utils/data-structures/fix-sized-cache";
|
||||
import { setUpTelemetry } from "./utils/set-up-telemetry";
|
||||
import { DIFF_CACHE_SIZE_MB } from "./consts";
|
||||
import { ServerConfig } from "./services/server-config";
|
||||
import type { EventListeners } from "./utils/data-structures/event-listeners";
|
||||
import { Lock } from "./utils/data-structures/locks";
|
||||
import { ExpectedFsEvents } from "./sync-operations/expected-fs-events";
|
||||
|
||||
export class SyncClient {
|
||||
private hasStartedOfflineSync = false;
|
||||
private hasFinishedOfflineSync = false;
|
||||
private hasStarted = false;
|
||||
private hasBeenDestroyed = false;
|
||||
private unloadTelemetry?: () => void;
|
||||
private isDestroying = false;
|
||||
private readonly eventUnsubscribers: (() => void)[] = [];
|
||||
private readonly settingsChangeLock = new Lock(
|
||||
"SyncClient.onSettingsChange"
|
||||
);
|
||||
|
||||
private constructor(
|
||||
public readonly logger: Logger,
|
||||
private readonly history: SyncHistory,
|
||||
private readonly settings: Settings,
|
||||
private readonly database: Database,
|
||||
private readonly syncEventQueue: SyncEventQueue,
|
||||
private readonly syncer: Syncer,
|
||||
private readonly webSocketManager: WebSocketManager,
|
||||
public readonly logger: Logger,
|
||||
private readonly fetchController: FetchController,
|
||||
private readonly cursorTracker: CursorTracker,
|
||||
private readonly fileChangeNotifier: FileChangeNotifier,
|
||||
private readonly contentCache: FixedSizeDocumentCache,
|
||||
private readonly fileOperations: FileOperations,
|
||||
private readonly serverConfig: ServerConfig,
|
||||
private readonly syncService: SyncService,
|
||||
private readonly expectedFsEvents: ExpectedFsEvents,
|
||||
private readonly persistence: PersistenceProvider<
|
||||
Partial<{
|
||||
settings: Partial<SyncSettings>;
|
||||
database: Partial<StoredDatabase>;
|
||||
database: Partial<StoredSyncState>;
|
||||
}>
|
||||
>
|
||||
) {}
|
||||
|
||||
public get documentCount(): number {
|
||||
return this.database.length;
|
||||
public get syncedDocumentCount(): number {
|
||||
return this.syncEventQueue.syncedDocumentCount;
|
||||
}
|
||||
|
||||
public get isWebSocketConnected(): boolean {
|
||||
|
|
@ -73,6 +80,27 @@ export class SyncClient {
|
|||
return this.history.onHistoryUpdated;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fires whenever a tracked document's local file moves on disk —
|
||||
* watcher-driven user renames, post-create deconflicts placed by
|
||||
* the reconciler, lost-rename replays in offline scan, slot
|
||||
* displacements when another record claims a path. Both
|
||||
* `oldPath` and `newPath` may be `undefined` (placement-pending
|
||||
* state). Useful for callers that mirror disk-side path state
|
||||
* — e.g. test harnesses tracking which paths are safe to mutate
|
||||
* — and need a signal beyond the user-facing history.
|
||||
*/
|
||||
public get onDocumentPathChanged(): EventListeners<
|
||||
(
|
||||
documentId: DocumentId,
|
||||
oldPath: RelativePath | undefined,
|
||||
newPath: RelativePath | undefined
|
||||
) => unknown
|
||||
> {
|
||||
this.checkIfDestroyed("onDocumentPathChanged getter");
|
||||
return this.syncEventQueue.onDocumentPathChanged;
|
||||
}
|
||||
|
||||
public get onSettingsChanged(): EventListeners<
|
||||
(newSettings: SyncSettings, oldSettings: SyncSettings) => unknown
|
||||
> {
|
||||
|
|
@ -101,6 +129,13 @@ export class SyncClient {
|
|||
return this.cursorTracker.onRemoteCursorsUpdated;
|
||||
}
|
||||
|
||||
public get hasPendingWork(): boolean {
|
||||
return (
|
||||
this.syncEventQueue.pendingUpdateCount > 0 ||
|
||||
this.webSocketManager.hasOutstandingWork
|
||||
);
|
||||
}
|
||||
|
||||
public static async create({
|
||||
fs,
|
||||
persistence,
|
||||
|
|
@ -112,7 +147,8 @@ export class SyncClient {
|
|||
persistence: PersistenceProvider<
|
||||
Partial<{
|
||||
settings: Partial<SyncSettings>;
|
||||
database: Partial<StoredDatabase>;
|
||||
database: Partial<StoredSyncState>;
|
||||
deviceId: string;
|
||||
}>
|
||||
>;
|
||||
fetch?: typeof globalThis.fetch;
|
||||
|
|
@ -121,39 +157,46 @@ export class SyncClient {
|
|||
}): Promise<SyncClient> {
|
||||
const logger = new Logger();
|
||||
|
||||
const deviceId = createClientId();
|
||||
|
||||
logger.info(`Creating SyncClient with client id ${deviceId}`);
|
||||
|
||||
const history = new SyncHistory(logger);
|
||||
|
||||
let state = (await persistence.load()) ?? {
|
||||
settings: undefined,
|
||||
database: undefined
|
||||
database: undefined,
|
||||
deviceId: undefined
|
||||
};
|
||||
|
||||
// Persist deviceId across destroy + init so the server's
|
||||
// lost-create dedup (which scopes by device_id) can recognise
|
||||
// a retry as belonging to the same client. Without this,
|
||||
// every fresh `SyncClient` after a destroy would generate a
|
||||
// new deviceId, the server-side query would miss, and the
|
||||
// pending-but-lost create would deconflict instead of
|
||||
// binding to the doc its content was already absorbed into.
|
||||
let deviceId = state.deviceId;
|
||||
if (deviceId === undefined) {
|
||||
deviceId = createClientId();
|
||||
state = { ...state, deviceId };
|
||||
await persistence.save(state);
|
||||
}
|
||||
|
||||
logger.info(`Creating SyncClient with client id ${deviceId}`);
|
||||
|
||||
const settings = new Settings(
|
||||
logger,
|
||||
state.settings,
|
||||
async (data): Promise<void> => {
|
||||
state = { ...state, settings: data };
|
||||
// we're not rate-limiting settings saves as (1) we need to initialise the settings to know the rate limit
|
||||
// and (2) settings changes are infrequent enough that rate-limiting is not necessary
|
||||
await persistence.save(state);
|
||||
}
|
||||
);
|
||||
|
||||
const rateLimitedSave = rateLimit(
|
||||
persistence.save,
|
||||
() => settings.getSettings().minimumSaveIntervalMs
|
||||
);
|
||||
|
||||
const database = new Database(
|
||||
const syncEventQueue = new SyncEventQueue(
|
||||
settings,
|
||||
logger,
|
||||
state.database,
|
||||
async (data): Promise<void> => {
|
||||
state = { ...state, database: data };
|
||||
await rateLimitedSave(state);
|
||||
await persistence.save(state);
|
||||
}
|
||||
);
|
||||
|
||||
|
|
@ -170,32 +213,23 @@ export class SyncClient {
|
|||
fetch
|
||||
);
|
||||
|
||||
const serverConfig = new ServerConfig(syncService);
|
||||
const serverConfig = new ServerConfig(syncService, settings);
|
||||
|
||||
const expectedFsEvents = new ExpectedFsEvents();
|
||||
|
||||
const fileOperations = new FileOperations(
|
||||
logger,
|
||||
database,
|
||||
fs,
|
||||
serverConfig,
|
||||
expectedFsEvents,
|
||||
nativeLineEndings
|
||||
);
|
||||
|
||||
const contentCache = new FixedSizeDocumentCache(
|
||||
1024 * 1024 * DIFF_CACHE_SIZE_MB
|
||||
);
|
||||
const unrestrictedSyncer = new UnrestrictedSyncer(
|
||||
logger,
|
||||
database,
|
||||
settings,
|
||||
syncService,
|
||||
fileOperations,
|
||||
history,
|
||||
contentCache,
|
||||
serverConfig
|
||||
1024 * 1024 * settings.getSettings().diffCacheSizeMB
|
||||
);
|
||||
|
||||
const webSocketManager = new WebSocketManager(
|
||||
deviceId,
|
||||
logger,
|
||||
settings,
|
||||
webSocket
|
||||
|
|
@ -204,34 +238,38 @@ export class SyncClient {
|
|||
const syncer = new Syncer(
|
||||
deviceId,
|
||||
logger,
|
||||
database,
|
||||
settings,
|
||||
syncService,
|
||||
webSocketManager,
|
||||
fileOperations,
|
||||
unrestrictedSyncer
|
||||
syncService,
|
||||
history,
|
||||
contentCache,
|
||||
serverConfig,
|
||||
syncEventQueue
|
||||
);
|
||||
|
||||
const fileChangeNotifier = new FileChangeNotifier();
|
||||
const cursorTracker = new CursorTracker(
|
||||
database,
|
||||
logger,
|
||||
syncEventQueue,
|
||||
webSocketManager,
|
||||
fileOperations,
|
||||
fileChangeNotifier
|
||||
);
|
||||
const client = new SyncClient(
|
||||
logger,
|
||||
history,
|
||||
settings,
|
||||
database,
|
||||
syncEventQueue,
|
||||
syncer,
|
||||
webSocketManager,
|
||||
logger,
|
||||
fetchController,
|
||||
cursorTracker,
|
||||
fileChangeNotifier,
|
||||
contentCache,
|
||||
fileOperations,
|
||||
serverConfig,
|
||||
syncService,
|
||||
expectedFsEvents,
|
||||
persistence
|
||||
);
|
||||
|
||||
|
|
@ -285,10 +323,10 @@ export class SyncClient {
|
|||
}
|
||||
|
||||
/**
|
||||
* Reload settings from disk overriding current in-memory settings.
|
||||
* Missing values will be filled in from DEFAULT_SETTINGS rather than
|
||||
* retaining current in-memory settings.
|
||||
*/
|
||||
* Reload settings from disk overriding current in-memory settings.
|
||||
* Missing values will be filled in from DEFAULT_SETTINGS rather than
|
||||
* retaining current in-memory settings.
|
||||
*/
|
||||
public async reloadSettings(): Promise<void> {
|
||||
this.checkIfDestroyed("reloadSettings");
|
||||
|
||||
|
|
@ -320,10 +358,10 @@ export class SyncClient {
|
|||
}
|
||||
|
||||
/**
|
||||
* Wait for the in-flight operations to finish, reset all tracking,
|
||||
* and the local database but retain the settings.
|
||||
* The SyncClient can be used again after calling this method.
|
||||
*/
|
||||
* Wait for the in-flight operations to finish, reset all tracking,
|
||||
* and the local state but retain the settings.
|
||||
* The SyncClient can be used again after calling this method.
|
||||
*/
|
||||
public async reset(): Promise<void> {
|
||||
this.checkIfDestroyed("reset");
|
||||
|
||||
|
|
@ -332,16 +370,16 @@ export class SyncClient {
|
|||
);
|
||||
await this.pause();
|
||||
|
||||
// clear all local state
|
||||
this.logger.info("Resetting SyncClient's local state");
|
||||
this.database.reset();
|
||||
await this.database.save(); // ensure the new database reads as empty
|
||||
await this.syncEventQueue.clearAllState();
|
||||
await this.syncEventQueue.save();
|
||||
this.resetInMemoryState();
|
||||
this.hasStartedOfflineSync = false;
|
||||
this.hasFinishedOfflineSync = false;
|
||||
this.serverConfig.reset();
|
||||
|
||||
await this.startSyncing();
|
||||
if (this.settings.getSettings().isSyncEnabled) {
|
||||
await this.startSyncing();
|
||||
}
|
||||
}
|
||||
|
||||
public getSettings(): SyncSettings {
|
||||
|
|
@ -363,40 +401,48 @@ export class SyncClient {
|
|||
await this.settings.setSettings(value);
|
||||
}
|
||||
|
||||
public async syncLocallyCreatedFile(
|
||||
relativePath: RelativePath
|
||||
): Promise<void> {
|
||||
public syncLocallyCreatedFile(relativePath: RelativePath): void {
|
||||
this.checkIfDestroyed("syncLocallyCreatedFile");
|
||||
|
||||
this.fileChangeNotifier.notifyOfFileChange(relativePath);
|
||||
return this.syncer.syncLocallyCreatedFile(relativePath);
|
||||
this.fileChangeNotifier.notifyOfFileChange(relativePath); // this is for updating cursors
|
||||
if (this.expectedFsEvents.matchCreate(relativePath)) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.syncer.syncLocallyCreatedFile(relativePath);
|
||||
}
|
||||
|
||||
public async syncLocallyDeletedFile(
|
||||
relativePath: RelativePath
|
||||
): Promise<void> {
|
||||
this.checkIfDestroyed("syncLocallyDeletedFile");
|
||||
|
||||
this.fileChangeNotifier.notifyOfFileChange(relativePath);
|
||||
return this.syncer.syncLocallyDeletedFile(relativePath);
|
||||
}
|
||||
|
||||
public async syncLocallyUpdatedFile({
|
||||
public syncLocallyUpdatedFile({
|
||||
oldPath,
|
||||
relativePath
|
||||
}: {
|
||||
oldPath?: RelativePath;
|
||||
relativePath: RelativePath;
|
||||
}): Promise<void> {
|
||||
}): void {
|
||||
this.checkIfDestroyed("syncLocallyUpdatedFile");
|
||||
|
||||
this.fileChangeNotifier.notifyOfFileChange(relativePath);
|
||||
return this.syncer.syncLocallyUpdatedFile({
|
||||
this.fileChangeNotifier.notifyOfFileChange(relativePath); // this is for updating cursors
|
||||
if (this.expectedFsEvents.matchUpdate(relativePath, oldPath)) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.syncer.syncLocallyUpdatedFile({
|
||||
oldPath,
|
||||
relativePath
|
||||
});
|
||||
}
|
||||
|
||||
public syncLocallyDeletedFile(relativePath: RelativePath): void {
|
||||
this.checkIfDestroyed("syncLocallyDeletedFile");
|
||||
|
||||
this.fileChangeNotifier.notifyOfFileChange(relativePath); // this is for updating cursors
|
||||
if (this.expectedFsEvents.matchDelete(relativePath)) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.syncer.syncLocallyDeletedFile(relativePath);
|
||||
}
|
||||
|
||||
public getDocumentSyncingStatus(
|
||||
relativePath: RelativePath
|
||||
): DocumentSyncStatus {
|
||||
|
|
@ -406,16 +452,11 @@ export class SyncClient {
|
|||
return DocumentSyncStatus.SYNCING_IS_DISABLED;
|
||||
}
|
||||
|
||||
if (!this.syncer.isFirstSyncComplete || !this.hasFinishedOfflineSync) {
|
||||
if (!this.hasFinishedOfflineSync) {
|
||||
return DocumentSyncStatus.SYNCING;
|
||||
}
|
||||
|
||||
const document =
|
||||
this.database.getLatestDocumentByRelativePath(relativePath);
|
||||
if (document === undefined) {
|
||||
return DocumentSyncStatus.SYNCING;
|
||||
}
|
||||
return document.updates.length > 0
|
||||
return this.syncEventQueue.hasPendingEventsForPath(relativePath)
|
||||
? DocumentSyncStatus.SYNCING
|
||||
: DocumentSyncStatus.UP_TO_DATE;
|
||||
}
|
||||
|
|
@ -429,20 +470,20 @@ export class SyncClient {
|
|||
}
|
||||
|
||||
public async waitUntilFinished(): Promise<void> {
|
||||
this.checkIfDestroyed("waitUntilIdle");
|
||||
await this.syncer.waitUntilFinished();
|
||||
await this.webSocketManager.waitUntilFinished();
|
||||
await this.database.save(); // flush all changes to disk
|
||||
this.checkIfDestroyed("waitUntilFinished");
|
||||
await this.waitUntilFinishedInternal();
|
||||
}
|
||||
|
||||
/**
|
||||
* Completely destroy the SyncClient, cancelling all in-progress operations.
|
||||
* After calling this method, the SyncClient cannot be used again.
|
||||
*/
|
||||
* Completely destroy the SyncClient, cancelling all in-progress operations.
|
||||
* After calling this method, the SyncClient cannot be used again.
|
||||
*/
|
||||
public async destroy(): Promise<void> {
|
||||
this.checkIfDestroyed("destroy");
|
||||
|
||||
// Prevent concurrent destroy calls
|
||||
if (this.hasBeenDestroyed) {
|
||||
throw new Error(
|
||||
"SyncClient has been destroyed and can no longer be used; called from destroy"
|
||||
);
|
||||
}
|
||||
if (this.isDestroying) {
|
||||
this.logger.warn(
|
||||
"destroy() called while already destroying, ignoring"
|
||||
|
|
@ -451,52 +492,92 @@ export class SyncClient {
|
|||
}
|
||||
this.isDestroying = true;
|
||||
|
||||
// cancel everything that's in progress
|
||||
await this.pause();
|
||||
// Run cleanup in `finally` so a thrown pause() — or anything else
|
||||
// mid-shutdown — still leaves the client in the disposed state
|
||||
// instead of bricked with subscribers/telemetry hanging on.
|
||||
try {
|
||||
await this.pause();
|
||||
} finally {
|
||||
this.hasBeenDestroyed = true;
|
||||
|
||||
this.hasBeenDestroyed = true;
|
||||
this.resetInMemoryState();
|
||||
|
||||
this.resetInMemoryState();
|
||||
this.eventUnsubscribers.forEach((unsubscribe) => {
|
||||
unsubscribe();
|
||||
});
|
||||
this.eventUnsubscribers.length = 0;
|
||||
|
||||
// Clean up event listeners to prevent memory leaks
|
||||
this.eventUnsubscribers.forEach((unsubscribe) => {
|
||||
unsubscribe();
|
||||
});
|
||||
this.eventUnsubscribers.length = 0;
|
||||
this.logger.info("SyncClient has been successfully disposed");
|
||||
|
||||
this.logger.info("SyncClient has been successfully disposed");
|
||||
this.unloadTelemetry?.();
|
||||
}
|
||||
}
|
||||
|
||||
this.unloadTelemetry?.();
|
||||
/**
|
||||
* The actual drain — separated from `waitUntilFinished` so internal
|
||||
* shutdown paths (`pause` / `destroy`) can wait for in-flight work
|
||||
* without tripping the public `checkIfDestroyed` guard, which exists
|
||||
* only to keep external callers from continuing to use a disposed
|
||||
* client.
|
||||
*
|
||||
* Loops because a WebSocket message handler completing is what enqueues
|
||||
* a `RemoteChange` into the syncer; if we awaited the syncer first and
|
||||
* the WS handler second, a message arriving mid-wait would leave a fresh
|
||||
* drain pending while `save()` ran. Each iteration waits for both, then
|
||||
* re-checks; we exit only once both report idle in the same pass.
|
||||
*/
|
||||
private async waitUntilFinishedInternal(): Promise<void> {
|
||||
while (
|
||||
this.webSocketManager.hasOutstandingWork ||
|
||||
this.syncer.hasPendingWork
|
||||
) {
|
||||
await this.webSocketManager.waitUntilFinished();
|
||||
await this.syncer.waitUntilFinished();
|
||||
}
|
||||
await this.syncEventQueue.save();
|
||||
}
|
||||
|
||||
private async startSyncing(): Promise<void> {
|
||||
this.checkIfDestroyed("startSyncing");
|
||||
this.fetchController.finishReset();
|
||||
// Undo any earlier `pause()` stop so retryForever keeps retrying.
|
||||
this.syncService.resume();
|
||||
|
||||
await this.serverConfig.initialize();
|
||||
await this.serverConfig.getConfig();
|
||||
|
||||
await this.syncer.scheduleSyncForOfflineChanges();
|
||||
this.syncer.resumeDraining();
|
||||
this.webSocketManager.start();
|
||||
|
||||
if (!this.hasStartedOfflineSync) {
|
||||
this.hasStartedOfflineSync = true;
|
||||
await this.syncer.scheduleSyncForOfflineChanges();
|
||||
}
|
||||
|
||||
this.hasFinishedOfflineSync = true;
|
||||
}
|
||||
|
||||
private async pause(): Promise<void> {
|
||||
this.hasFinishedOfflineSync = false;
|
||||
this.syncer.pauseDraining();
|
||||
this.fetchController.startReset();
|
||||
// Signal the service so any `retryForever` loop exits at its next
|
||||
// iteration instead of continuing to retry a network request while
|
||||
// the rest of the client is winding down.
|
||||
this.syncService.stop();
|
||||
await this.webSocketManager.stop();
|
||||
await this.waitUntilFinished();
|
||||
await this.waitUntilFinishedInternal();
|
||||
// Clear the offline-scan gate so a subsequent `startSyncing()`
|
||||
// re-runs the scan; otherwise any local changes made while sync was
|
||||
// paused (offline edits, deletes, renames) wouldn't be detected, and
|
||||
// an incoming remote update would silently overwrite them.
|
||||
this.syncer.clearOfflineScanGate();
|
||||
// Drop any expected fs events that were registered but never matched
|
||||
// (e.g. an op aborted by SyncResetError). Otherwise a real user edit
|
||||
// at the same path after re-enable would be swallowed.
|
||||
this.expectedFsEvents.clear();
|
||||
}
|
||||
|
||||
private resetInMemoryState(): void {
|
||||
this.history.reset();
|
||||
this.contentCache.reset();
|
||||
// don't reset the logger
|
||||
this.cursorTracker.reset();
|
||||
this.syncer.reset();
|
||||
this.fileOperations.reset();
|
||||
}
|
||||
|
||||
private async onSettingsChange(
|
||||
|
|
@ -505,36 +586,55 @@ export class SyncClient {
|
|||
): Promise<void> {
|
||||
this.checkIfDestroyed("onSettingsChange");
|
||||
|
||||
if (
|
||||
newSettings.vaultName !== oldSettings.vaultName ||
|
||||
newSettings.remoteUri !== oldSettings.remoteUri
|
||||
) {
|
||||
await this.reset();
|
||||
}
|
||||
|
||||
if (newSettings.isSyncEnabled !== oldSettings.isSyncEnabled) {
|
||||
if (newSettings.isSyncEnabled) {
|
||||
await this.startSyncing();
|
||||
} else {
|
||||
await this.pause();
|
||||
// Serialize listener invocations so back-to-back settings updates
|
||||
// can't run reset()/pause()/startSyncing() concurrently.
|
||||
await this.settingsChangeLock.withLock(async () => {
|
||||
// The lock is FIFO, so by the time we run the client may have
|
||||
// been destroyed in a queued invocation ahead of us.
|
||||
if (this.hasBeenDestroyed) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (newSettings.diffCacheSizeMB !== oldSettings.diffCacheSizeMB) {
|
||||
this.contentCache.resize(newSettings.diffCacheSizeMB * 1024 * 1024);
|
||||
}
|
||||
const connectionChanged =
|
||||
newSettings.vaultName !== oldSettings.vaultName ||
|
||||
newSettings.remoteUri !== oldSettings.remoteUri;
|
||||
|
||||
if (newSettings.enableTelemetry !== oldSettings.enableTelemetry) {
|
||||
if (newSettings.enableTelemetry) {
|
||||
this.unloadTelemetry = setUpTelemetry();
|
||||
} else {
|
||||
this.unloadTelemetry?.();
|
||||
if (connectionChanged) {
|
||||
// reset() pauses, clears state, then starts iff isSyncEnabled
|
||||
// — so any concurrent isSyncEnabled change is already applied.
|
||||
await this.reset();
|
||||
} else if (
|
||||
newSettings.isSyncEnabled !== oldSettings.isSyncEnabled
|
||||
) {
|
||||
if (newSettings.isSyncEnabled) {
|
||||
await this.startSyncing();
|
||||
} else {
|
||||
await this.pause();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (newSettings.diffCacheSizeMB !== oldSettings.diffCacheSizeMB) {
|
||||
this.contentCache.resize(
|
||||
newSettings.diffCacheSizeMB * 1024 * 1024
|
||||
);
|
||||
}
|
||||
|
||||
if (newSettings.enableTelemetry !== oldSettings.enableTelemetry) {
|
||||
if (newSettings.enableTelemetry) {
|
||||
this.unloadTelemetry = setUpTelemetry();
|
||||
} else {
|
||||
this.unloadTelemetry?.();
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private checkIfDestroyed(origin: string): void {
|
||||
if (this.hasBeenDestroyed) {
|
||||
// Reject new public-API entries the moment destroy() is called,
|
||||
// not after `pause()` returns. Otherwise an external caller could
|
||||
// pass the guard and start mutating state while destroy() is
|
||||
// tearing down the websocket / clearing caches.
|
||||
if (this.hasBeenDestroyed || this.isDestroying) {
|
||||
throw new Error(
|
||||
`SyncClient has been destroyed and can no longer be used; called from ${origin}`
|
||||
);
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import type { FileOperations } from "../file-operations/file-operations";
|
||||
import type { Database, RelativePath } from "../persistence/database";
|
||||
import type { RelativePath } from "./types";
|
||||
import type { SyncEventQueue } from "./sync-event-queue";
|
||||
import type { ClientCursors } from "../services/types/ClientCursors";
|
||||
import type { CursorSpan } from "../services/types/CursorSpan";
|
||||
import type { DocumentWithCursors } from "../services/types/DocumentWithCursors";
|
||||
|
|
@ -10,6 +11,7 @@ import { hash } from "../utils/hash";
|
|||
import type { FileChangeNotifier } from "./file-change-notifier";
|
||||
import { Lock } from "../utils/data-structures/locks";
|
||||
import { EventListeners } from "../utils/data-structures/event-listeners";
|
||||
import type { Logger } from "../tracing/logger";
|
||||
|
||||
// Cursor positions are updated separately from documents. However, a given cursor position is only
|
||||
// valid within a certain version of the document it belongs to. This class tracks previous and the latest
|
||||
|
|
@ -22,22 +24,29 @@ export class CursorTracker {
|
|||
(cursors: MaybeOutdatedClientCursors[]) => unknown
|
||||
>();
|
||||
|
||||
private readonly updateLock = new Lock();
|
||||
private readonly updateLock: Lock;
|
||||
|
||||
private knownRemoteCursors: (ClientCursors & {
|
||||
upToDateness: DocumentUpToDateness;
|
||||
})[] = [];
|
||||
|
||||
private lastLocalCursorState: DocumentWithCursors[] = [];
|
||||
private lastLocalCursorStateWithoutDirtyDocuments: DocumentWithCursors[] =
|
||||
[];
|
||||
// Cache the previously sent state as a JSON string rather than as the
|
||||
// array. We mutate `documentsWithCursors` in-place after the cache check
|
||||
// (setting `vaultUpdateId = null` for dirty docs); storing the array would
|
||||
// alias and the next call's equality check would compare against
|
||||
// post-mutation state.
|
||||
private lastLocalCursorStateJson = "[]";
|
||||
private lastLocalCursorStateWithoutDirtyDocumentsJson = "[]";
|
||||
|
||||
public constructor(
|
||||
private readonly database: Database,
|
||||
logger: Logger,
|
||||
private readonly queue: SyncEventQueue,
|
||||
private readonly webSocketManager: WebSocketManager,
|
||||
private readonly fileOperations: FileOperations,
|
||||
private readonly fileChangeNotifier: FileChangeNotifier
|
||||
) {
|
||||
this.updateLock = new Lock(CursorTracker.name, logger);
|
||||
|
||||
this.webSocketManager.onRemoteCursorsUpdateReceived.add(
|
||||
async (clientCursors) => {
|
||||
await this.updateLock.withLock(async () => {
|
||||
|
|
@ -53,7 +62,7 @@ export class CursorTracker {
|
|||
|
||||
for (const cursor of clientCursors.filter((client) =>
|
||||
client.documentsWithCursors.every(
|
||||
(doc) => doc.vault_update_id != null
|
||||
(doc) => doc.vaultUpdateId != null
|
||||
)
|
||||
)) {
|
||||
updatedKnownRemoteCursors.push({
|
||||
|
|
@ -77,14 +86,20 @@ export class CursorTracker {
|
|||
for (const clientCursor of this.knownRemoteCursors) {
|
||||
if (
|
||||
clientCursor.documentsWithCursors.some(
|
||||
(document) =>
|
||||
document.relative_path === relativePath
|
||||
(document) => document.relativePath === relativePath
|
||||
)
|
||||
) {
|
||||
clientCursor.upToDateness =
|
||||
await this.getDocumentsUpToDateness(clientCursor);
|
||||
}
|
||||
}
|
||||
// Drop the local-cursor send-cache so the next call re-reads
|
||||
// the file. The first cache key is the editor's input, which
|
||||
// doesn't change when the file content does — without this,
|
||||
// a remote update flipping the file from dirty back to clean
|
||||
// would never re-send the cursor with a fresh `vaultUpdateId`.
|
||||
this.lastLocalCursorStateJson = "";
|
||||
this.lastLocalCursorStateWithoutDirtyDocumentsJson = "";
|
||||
})
|
||||
);
|
||||
}
|
||||
|
|
@ -95,70 +110,67 @@ export class CursorTracker {
|
|||
public async sendLocalCursorsToServer(
|
||||
documentToCursors: Record<RelativePath, CursorSpan[]>
|
||||
): Promise<void> {
|
||||
const documentsWithCursors: DocumentWithCursors[] = [];
|
||||
// Serialise concurrent senders so they don't interleave on the
|
||||
// disk reads + state mutations and emit out-of-order cursor messages.
|
||||
await this.updateLock.withLock(async () => {
|
||||
const documentsWithCursors: DocumentWithCursors[] = [];
|
||||
|
||||
for (const [relativePath, cursors] of Object.entries(
|
||||
documentToCursors
|
||||
)) {
|
||||
const record =
|
||||
this.database.getLatestDocumentByRelativePath(relativePath);
|
||||
for (const [relativePath, cursors] of Object.entries(
|
||||
documentToCursors
|
||||
)) {
|
||||
const record = this.queue.getRecordByLocalPath(relativePath);
|
||||
|
||||
if (!record) {
|
||||
continue; // Let's wait for the file to be created before sending cursors
|
||||
if (!record) {
|
||||
continue; // Let's wait for the file to be created before sending cursors
|
||||
}
|
||||
|
||||
documentsWithCursors.push({
|
||||
relativePath: relativePath,
|
||||
documentId: record.documentId,
|
||||
vaultUpdateId: record.parentVersionId,
|
||||
cursors: cursors.map(({ start, end }) => ({
|
||||
start: Math.min(start, end),
|
||||
end: Math.max(start, end)
|
||||
})) // the client might send directional selections
|
||||
});
|
||||
}
|
||||
|
||||
if (!record.metadata) {
|
||||
continue; // this is a new document, no need to sync the cursors
|
||||
const beforeJson = JSON.stringify(documentsWithCursors);
|
||||
if (this.lastLocalCursorStateJson === beforeJson) {
|
||||
// Caching step to avoid reading the edited files all the time
|
||||
return;
|
||||
}
|
||||
this.lastLocalCursorStateJson = beforeJson;
|
||||
|
||||
for (const doc of documentsWithCursors) {
|
||||
const readContent = await this.fileOperations.read(
|
||||
doc.relativePath
|
||||
);
|
||||
const record = this.queue.getRecordByLocalPath(
|
||||
doc.relativePath
|
||||
);
|
||||
if (record?.remoteHash !== (await hash(readContent))) {
|
||||
doc.vaultUpdateId = null;
|
||||
}
|
||||
}
|
||||
|
||||
documentsWithCursors.push({
|
||||
relative_path: relativePath,
|
||||
document_id: record.documentId,
|
||||
vault_update_id: record.metadata.parentVersionId,
|
||||
cursors: cursors.map(({ start, end }) => ({
|
||||
start: Math.min(start, end),
|
||||
end: Math.max(start, end)
|
||||
})) // the client might send directional selections
|
||||
});
|
||||
}
|
||||
|
||||
if (
|
||||
JSON.stringify(this.lastLocalCursorState) ===
|
||||
JSON.stringify(documentsWithCursors)
|
||||
) {
|
||||
// Caching step to avoid reading the edited files all the time
|
||||
return;
|
||||
}
|
||||
this.lastLocalCursorState = documentsWithCursors;
|
||||
|
||||
for (const doc of documentsWithCursors) {
|
||||
const readContent = await this.fileOperations.read(
|
||||
doc.relative_path
|
||||
);
|
||||
const record = this.database.getLatestDocumentByRelativePath(
|
||||
doc.relative_path
|
||||
);
|
||||
if (record?.metadata?.hash !== hash(readContent)) {
|
||||
doc.vault_update_id = null;
|
||||
const afterJson = JSON.stringify(documentsWithCursors);
|
||||
if (
|
||||
this.lastLocalCursorStateWithoutDirtyDocumentsJson === afterJson
|
||||
) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
JSON.stringify(this.lastLocalCursorStateWithoutDirtyDocuments) ===
|
||||
JSON.stringify(documentsWithCursors)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
this.lastLocalCursorStateWithoutDirtyDocumentsJson = afterJson;
|
||||
|
||||
this.lastLocalCursorStateWithoutDirtyDocuments = documentsWithCursors;
|
||||
|
||||
this.webSocketManager.updateLocalCursors({ documentsWithCursors });
|
||||
this.webSocketManager.updateLocalCursors({ documentsWithCursors });
|
||||
});
|
||||
}
|
||||
|
||||
public reset(): void {
|
||||
this.knownRemoteCursors = [];
|
||||
this.lastLocalCursorState = [];
|
||||
this.lastLocalCursorStateWithoutDirtyDocuments = [];
|
||||
this.lastLocalCursorStateJson = "[]";
|
||||
this.lastLocalCursorStateWithoutDirtyDocumentsJson = "[]";
|
||||
this.updateLock.reset();
|
||||
}
|
||||
|
||||
|
|
@ -223,35 +235,28 @@ export class CursorTracker {
|
|||
private async getDocumentUpToDateness(
|
||||
document: DocumentWithCursors
|
||||
): Promise<DocumentUpToDateness> {
|
||||
const record = this.database.getLatestDocumentByRelativePath(
|
||||
document.relative_path
|
||||
);
|
||||
const record = this.queue.getRecordByLocalPath(document.relativePath);
|
||||
|
||||
if (!record) {
|
||||
// the document of the cursor must be from the future
|
||||
return DocumentUpToDateness.Later;
|
||||
}
|
||||
|
||||
if (
|
||||
(record.metadata?.parentVersionId ?? 0) <
|
||||
(document.vault_update_id ?? 0)
|
||||
) {
|
||||
if (record.parentVersionId < (document.vaultUpdateId ?? 0)) {
|
||||
return DocumentUpToDateness.Later;
|
||||
} else if (
|
||||
(document.vault_update_id ?? 0) <
|
||||
(record.metadata?.parentVersionId ?? 0)
|
||||
) {
|
||||
} else if ((document.vaultUpdateId ?? 0) < record.parentVersionId) {
|
||||
// the document of the cursor must be from the past
|
||||
return DocumentUpToDateness.Prior;
|
||||
}
|
||||
|
||||
const currentContent = await this.fileOperations.read(
|
||||
document.relative_path
|
||||
document.relativePath
|
||||
);
|
||||
|
||||
return this.database.getLatestDocumentByRelativePath(
|
||||
document.relative_path
|
||||
)?.metadata?.hash === hash(currentContent)
|
||||
const currentRecord = this.queue.getRecordByLocalPath(
|
||||
document.relativePath
|
||||
);
|
||||
return currentRecord?.remoteHash === (await hash(currentContent))
|
||||
? DocumentUpToDateness.UpToDate
|
||||
: DocumentUpToDateness.Prior;
|
||||
}
|
||||
|
|
|
|||
138
frontend/sync-client/src/sync-operations/expected-fs-events.ts
Normal file
138
frontend/sync-client/src/sync-operations/expected-fs-events.ts
Normal file
|
|
@ -0,0 +1,138 @@
|
|||
import type { RelativePath } from "./types";
|
||||
|
||||
/**
|
||||
* Counter-based registry of filesystem events the syncer is about to
|
||||
* cause. The syncer's own writes/renames/deletes go through
|
||||
* `FileOperations`, which calls into the host filesystem; the host then
|
||||
* fires watcher events that come back through `SyncClient.syncLocallyXxx`.
|
||||
* Without filtering, those echo events would be re-uploaded to the server
|
||||
* and broadcast back, producing an unbounded loop.
|
||||
*
|
||||
* The fix: every fs call in `FileOperations` registers the event it is
|
||||
* about to provoke; the matching `syncLocallyXxx` handler consumes it.
|
||||
* User-initiated edits never register, so they pass through unchanged.
|
||||
*
|
||||
* Counts are per (kind, path) so back-to-back syncer ops on the same path
|
||||
* (e.g. apply remote update then re-apply during convergence) match
|
||||
* one-for-one. If the watcher never fires for a registered op (e.g. the
|
||||
* fs throws before notifying), the entry is left behind; `clear()` is
|
||||
* called on pause/destroy to drop those before they collide with a real
|
||||
* user event later.
|
||||
*/
|
||||
export class ExpectedFsEvents {
|
||||
private readonly creates = new Map<RelativePath, number>();
|
||||
private readonly updates = new Map<RelativePath, number>();
|
||||
private readonly deletes = new Map<RelativePath, number>();
|
||||
// Renames are keyed by `JSON.stringify({oldPath, newPath})` so the
|
||||
// delimiter cannot occur inside either path.
|
||||
private readonly renames = new Map<RelativePath, number>();
|
||||
|
||||
private static renameKey(
|
||||
oldPath: RelativePath,
|
||||
newPath: RelativePath
|
||||
): string {
|
||||
return JSON.stringify({ oldPath, newPath });
|
||||
}
|
||||
|
||||
public expectCreate(path: RelativePath): void {
|
||||
this.bump(this.creates, path);
|
||||
}
|
||||
|
||||
public expectUpdate(path: RelativePath): void {
|
||||
this.bump(this.updates, path);
|
||||
}
|
||||
|
||||
public expectDelete(path: RelativePath): void {
|
||||
this.bump(this.deletes, path);
|
||||
}
|
||||
|
||||
public expectRename(oldPath: RelativePath, newPath: RelativePath): void {
|
||||
this.bump(this.renames, ExpectedFsEvents.renameKey(oldPath, newPath));
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel a previously-registered expectation when the fs op that registered
|
||||
* it failed before any watcher event could fire. Without this, a leaked
|
||||
* expectation silently swallows the next genuine user event at the same
|
||||
* path (or, for renames, the same `oldPath → newPath` pair).
|
||||
*
|
||||
* Floored at zero: if the watcher *did* fire (op partially completed) and
|
||||
* already consumed the entry, the unexpect is a no-op. The fallback is
|
||||
* acceptable — at worst we re-upload a real edit we'd otherwise filter.
|
||||
*/
|
||||
public unexpectCreate(path: RelativePath): void {
|
||||
this.decrement(this.creates, path);
|
||||
}
|
||||
|
||||
public unexpectUpdate(path: RelativePath): void {
|
||||
this.decrement(this.updates, path);
|
||||
}
|
||||
|
||||
public unexpectDelete(path: RelativePath): void {
|
||||
this.decrement(this.deletes, path);
|
||||
}
|
||||
|
||||
public unexpectRename(oldPath: RelativePath, newPath: RelativePath): void {
|
||||
this.decrement(
|
||||
this.renames,
|
||||
ExpectedFsEvents.renameKey(oldPath, newPath)
|
||||
);
|
||||
}
|
||||
|
||||
public matchCreate(path: RelativePath): boolean {
|
||||
return this.consume(this.creates, path);
|
||||
}
|
||||
|
||||
public matchUpdate(
|
||||
path: RelativePath,
|
||||
oldPath: RelativePath | undefined
|
||||
): boolean {
|
||||
if (oldPath !== undefined) {
|
||||
return this.consume(
|
||||
this.renames,
|
||||
ExpectedFsEvents.renameKey(oldPath, path)
|
||||
);
|
||||
}
|
||||
return this.consume(this.updates, path);
|
||||
}
|
||||
|
||||
public matchDelete(path: RelativePath): boolean {
|
||||
return this.consume(this.deletes, path);
|
||||
}
|
||||
|
||||
public clear(): void {
|
||||
this.creates.clear();
|
||||
this.updates.clear();
|
||||
this.deletes.clear();
|
||||
this.renames.clear();
|
||||
}
|
||||
|
||||
private bump(map: Map<RelativePath, number>, key: RelativePath): void {
|
||||
map.set(key, (map.get(key) ?? 0) + 1);
|
||||
}
|
||||
|
||||
private consume(
|
||||
map: Map<RelativePath, number>,
|
||||
key: RelativePath
|
||||
): boolean {
|
||||
const count = map.get(key) ?? 0;
|
||||
if (count === 0) {
|
||||
return false;
|
||||
}
|
||||
if (count === 1) {
|
||||
map.delete(key);
|
||||
} else {
|
||||
map.set(key, count - 1);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private decrement(map: Map<RelativePath, number>, key: RelativePath): void {
|
||||
const count = map.get(key) ?? 0;
|
||||
if (count <= 1) {
|
||||
map.delete(key);
|
||||
} else {
|
||||
map.set(key, count - 1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
import type { RelativePath } from "../persistence/database";
|
||||
import type { RelativePath } from "./types";
|
||||
import { EventListeners } from "../utils/data-structures/event-listeners";
|
||||
|
||||
export class FileChangeNotifier {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,185 @@
|
|||
import { describe, it } from "node:test";
|
||||
import assert from "node:assert";
|
||||
import { Logger } from "../tracing/logger";
|
||||
import { Settings } from "../persistence/settings";
|
||||
import { STORED_STATE_SCHEMA_VERSION, SyncEventQueue } from "./sync-event-queue";
|
||||
import { scheduleOfflineChanges } from "./offline-change-detector";
|
||||
import type { FileOperations } from "../file-operations/file-operations";
|
||||
import type { RelativePath } from "./types";
|
||||
|
||||
const makeQueue = async (): Promise<SyncEventQueue> => {
|
||||
const logger = new Logger();
|
||||
const settings = new Settings(logger, {}, async () => {
|
||||
/* no-op */
|
||||
});
|
||||
return new SyncEventQueue(
|
||||
settings,
|
||||
logger,
|
||||
{ schemaVersion: STORED_STATE_SCHEMA_VERSION },
|
||||
async () => {
|
||||
/* no-op */
|
||||
}
|
||||
);
|
||||
};
|
||||
|
||||
const makeOperations = (
|
||||
files: Record<string, Uint8Array>
|
||||
): FileOperations => {
|
||||
return {
|
||||
listFilesRecursively: async () => Object.keys(files),
|
||||
read: async (path: RelativePath) => {
|
||||
const data = files[path];
|
||||
if (data === undefined) {
|
||||
throw new Error(`File not found: ${path}`);
|
||||
}
|
||||
return data;
|
||||
}
|
||||
} as unknown as FileOperations;
|
||||
};
|
||||
|
||||
describe("scheduleOfflineChanges", () => {
|
||||
it("does not bind a local file to a placement-pending record whose remoteRelativePath was persisted before the doc moved on the server", async () => {
|
||||
// The bug: persisted byDocId can carry a placement-pending record
|
||||
// whose `remoteRelativePath` was saved before the doc was moved
|
||||
// server-side. After restart, offline-scan running before WS
|
||||
// catch-up would bind an unrelated local file at that stale path
|
||||
// to the moved doc and push the user's content as an update —
|
||||
// silently corrupting the moved doc and stranding the local file.
|
||||
const queue = await makeQueue();
|
||||
|
||||
// Stale placement-pending record: server has moved this doc
|
||||
// away from "stale-X.md" since this snapshot was saved.
|
||||
await queue.upsertRecord({
|
||||
documentId: "MOVED-DOC",
|
||||
parentVersionId: 5,
|
||||
remoteRelativePath: "stale-X.md" as RelativePath,
|
||||
remoteHash: "hash-from-old-state",
|
||||
localPath: undefined
|
||||
});
|
||||
|
||||
// User has an unrelated local file at the stale path.
|
||||
const operations = makeOperations({
|
||||
"stale-X.md": new TextEncoder().encode(
|
||||
"user's unrelated local content"
|
||||
)
|
||||
});
|
||||
|
||||
const enqueued: { kind: string; path: string }[] = [];
|
||||
await scheduleOfflineChanges(
|
||||
new Logger(),
|
||||
operations,
|
||||
queue,
|
||||
(path) => enqueued.push({ kind: "create", path }),
|
||||
(args) => enqueued.push({ kind: "update", path: args.relativePath }),
|
||||
(path) => enqueued.push({ kind: "delete", path })
|
||||
);
|
||||
|
||||
// The local file must become a fresh CREATE — never a hostile
|
||||
// UPDATE on the moved doc.
|
||||
assert.deepStrictEqual(enqueued, [
|
||||
{ kind: "create", path: "stale-X.md" }
|
||||
]);
|
||||
|
||||
// The placement-pending record must remain placement-pending —
|
||||
// its localPath must not have been bound to the unrelated user
|
||||
// file. The reconciler will place it correctly once WS catch-up
|
||||
// updates `remoteRelativePath` to the doc's current location.
|
||||
const record = queue.getDocumentByDocumentId("MOVED-DOC");
|
||||
assert.notStrictEqual(record, undefined);
|
||||
assert.strictEqual(record?.localPath, undefined);
|
||||
});
|
||||
|
||||
it("schedules an update for a local file that matches a settled record's localPath", async () => {
|
||||
const queue = await makeQueue();
|
||||
await queue.upsertRecord({
|
||||
documentId: "SETTLED-DOC",
|
||||
parentVersionId: 2,
|
||||
remoteRelativePath: "doc.md" as RelativePath,
|
||||
remoteHash: "hash",
|
||||
localPath: "doc.md" as RelativePath
|
||||
});
|
||||
|
||||
const operations = makeOperations({
|
||||
"doc.md": new TextEncoder().encode("content")
|
||||
});
|
||||
|
||||
const enqueued: { kind: string; path: string }[] = [];
|
||||
await scheduleOfflineChanges(
|
||||
new Logger(),
|
||||
operations,
|
||||
queue,
|
||||
(path) => enqueued.push({ kind: "create", path }),
|
||||
(args) => enqueued.push({ kind: "update", path: args.relativePath }),
|
||||
(path) => enqueued.push({ kind: "delete", path })
|
||||
);
|
||||
|
||||
assert.deepStrictEqual(enqueued, [
|
||||
{ kind: "update", path: "doc.md" }
|
||||
]);
|
||||
});
|
||||
|
||||
it("schedules a delete for a settled record whose local file is missing", async () => {
|
||||
const queue = await makeQueue();
|
||||
await queue.upsertRecord({
|
||||
documentId: "VANISHED-DOC",
|
||||
parentVersionId: 4,
|
||||
remoteRelativePath: "gone.md" as RelativePath,
|
||||
remoteHash: "hash",
|
||||
localPath: "gone.md" as RelativePath
|
||||
});
|
||||
|
||||
const operations = makeOperations({});
|
||||
|
||||
const enqueued: { kind: string; path: string }[] = [];
|
||||
await scheduleOfflineChanges(
|
||||
new Logger(),
|
||||
operations,
|
||||
queue,
|
||||
(path) => enqueued.push({ kind: "create", path }),
|
||||
(args) => enqueued.push({ kind: "update", path: args.relativePath }),
|
||||
(path) => enqueued.push({ kind: "delete", path })
|
||||
);
|
||||
|
||||
assert.deepStrictEqual(enqueued, [
|
||||
{ kind: "delete", path: "gone.md" }
|
||||
]);
|
||||
});
|
||||
|
||||
it("detects an offline rename when an untracked file matches a deleted record's content hash", async () => {
|
||||
const queue = await makeQueue();
|
||||
const content = new TextEncoder().encode("body");
|
||||
const contentHash = await (await import("../utils/hash")).hash(content);
|
||||
|
||||
await queue.upsertRecord({
|
||||
documentId: "DOC-1",
|
||||
parentVersionId: 5,
|
||||
remoteRelativePath: "old.md" as RelativePath,
|
||||
remoteHash: contentHash,
|
||||
localPath: "old.md" as RelativePath
|
||||
});
|
||||
const operations = makeOperations({ "new.md": content });
|
||||
|
||||
const enqueued: {
|
||||
kind: string;
|
||||
path: string;
|
||||
oldPath?: string;
|
||||
}[] = [];
|
||||
await scheduleOfflineChanges(
|
||||
new Logger(),
|
||||
operations,
|
||||
queue,
|
||||
(path) => enqueued.push({ kind: "create", path }),
|
||||
(args) =>
|
||||
enqueued.push({
|
||||
kind: "update",
|
||||
path: args.relativePath,
|
||||
oldPath: args.oldPath
|
||||
}),
|
||||
(path) => enqueued.push({ kind: "delete", path })
|
||||
);
|
||||
|
||||
assert.deepStrictEqual(enqueued, [
|
||||
{ kind: "update", path: "new.md", oldPath: "old.md" }
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,188 @@
|
|||
import type { DocumentRecord, RelativePath } from "./types";
|
||||
import type { Logger } from "../tracing/logger";
|
||||
import { hash } from "../utils/hash";
|
||||
import type { FileOperations } from "../file-operations/file-operations";
|
||||
import { findMatchingFile } from "../utils/find-matching-file";
|
||||
import type { SyncEventQueue } from "./sync-event-queue";
|
||||
import { removeFromArray } from "../utils/remove-from-array";
|
||||
import { FileNotFoundError } from "../errors/file-not-found-error";
|
||||
|
||||
/**
|
||||
* Scans the local filesystem and the document database to determine
|
||||
* which files were created, updated, moved, or deleted while the
|
||||
* client was offline, then enqueues the appropriate sync events.
|
||||
*
|
||||
* Placement-pending records (`localPath === undefined`) are deliberately
|
||||
* NOT bound to local files at the same `remoteRelativePath` here. The
|
||||
* persisted byDocId snapshot can be stale — a doc's server-side path
|
||||
* may have changed since the last save, so binding by stored path would
|
||||
* fold an unrelated user file into a moved doc and silently corrupt it.
|
||||
* Local files at those paths fall through to the LocalCreate flow below;
|
||||
* the server's create_document handler dedupes by path+freshness when
|
||||
* the doc really is at that path, and otherwise creates a new doc that
|
||||
* the reconciler places correctly once catch-up updates the stale
|
||||
* record's `remoteRelativePath`.
|
||||
*/
|
||||
export async function scheduleOfflineChanges(
|
||||
logger: Logger,
|
||||
operations: FileOperations,
|
||||
queue: SyncEventQueue,
|
||||
enqueueCreate: (path: RelativePath) => void,
|
||||
enqueueUpdate: (args: {
|
||||
oldPath?: RelativePath;
|
||||
relativePath: RelativePath;
|
||||
}) => void,
|
||||
enqueueDelete: (path: RelativePath) => void
|
||||
): Promise<void> {
|
||||
const allLocalFiles = new Set(await operations.listFilesRecursively());
|
||||
logger.info(`Scheduling sync for ${allLocalFiles.size} local files`);
|
||||
// `allSettledDocuments()` skips records with `localPath === undefined`
|
||||
// — those have no local file by definition and don't participate in
|
||||
// the disk-vs-record diff. The reconciler will place them on its
|
||||
// next pass.
|
||||
const allDocuments = queue.allSettledDocuments();
|
||||
|
||||
// A doc is "possibly deleted" only if it has no local file. Including
|
||||
// docs that still exist locally would queue a spurious delete alongside
|
||||
// the update below.
|
||||
const locallyPossiblyDeletedFiles: DocumentRecord[] = [];
|
||||
for (const record of allDocuments.values()) {
|
||||
// `localPath` is guaranteed non-undefined for entries in
|
||||
// `allSettledDocuments()`, but narrow explicitly for the type
|
||||
// checker (and so a future change to that helper doesn't
|
||||
// silently break this loop).
|
||||
if (
|
||||
record.localPath !== undefined &&
|
||||
!allLocalFiles.has(record.localPath)
|
||||
) {
|
||||
locallyPossiblyDeletedFiles.push(record);
|
||||
}
|
||||
}
|
||||
|
||||
const locallyPossibleCreatedFiles: RelativePath[] = [];
|
||||
const syncedLocalFiles: RelativePath[] = [];
|
||||
|
||||
for (const localFile of allLocalFiles) {
|
||||
if (allDocuments.has(localFile)) {
|
||||
syncedLocalFiles.push(localFile);
|
||||
} else if (queue.hasPendingCreateForPath(localFile)) {
|
||||
// A LocalCreate for this path is still in flight (no
|
||||
// record yet — its docId is a Promise). Re-enqueueing
|
||||
// would fire a second HTTP create that the server then
|
||||
// deconflicts to a sibling path, leaving the same bytes
|
||||
// in two docs. Skip; the in-flight create owns this slot.
|
||||
continue;
|
||||
} else {
|
||||
locallyPossibleCreatedFiles.push(localFile);
|
||||
}
|
||||
}
|
||||
|
||||
const renamedPaths = new Set<RelativePath>();
|
||||
// Track paths that were in `allLocalFiles` at scan-start but have
|
||||
// since disappeared. The scan awaits between `listFilesRecursively`
|
||||
// and each `read`, so a concurrent delete (slow file events, real
|
||||
// user activity) can vacate a slot mid-scan. Throwing would abort
|
||||
// the whole scan; nothing to sync for a file that's already gone.
|
||||
const disappearedPaths = new Set<RelativePath>();
|
||||
for (const path of locallyPossibleCreatedFiles) {
|
||||
let content: Uint8Array;
|
||||
try {
|
||||
content = await operations.read(path);
|
||||
} catch (e) {
|
||||
if (e instanceof FileNotFoundError) {
|
||||
logger.debug(
|
||||
`File ${path} disappeared before offline-scan could read it; skipping`
|
||||
);
|
||||
disappearedPaths.add(path);
|
||||
continue;
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
const contentHash = await hash(content);
|
||||
|
||||
const matchingDeletedFile = await findMatchingFile(
|
||||
contentHash,
|
||||
locallyPossiblyDeletedFiles
|
||||
);
|
||||
if (matchingDeletedFile !== undefined) {
|
||||
// localPath is guaranteed defined for records in
|
||||
// locallyPossiblyDeletedFiles (we filtered above).
|
||||
const oldPath = matchingDeletedFile.localPath;
|
||||
if (oldPath === undefined) {
|
||||
continue;
|
||||
}
|
||||
logger.debug(
|
||||
`File ${path} might have been moved from ${oldPath} while offline, scheduling sync to move it`
|
||||
);
|
||||
enqueueUpdate({
|
||||
oldPath,
|
||||
relativePath: path
|
||||
});
|
||||
removeFromArray(locallyPossiblyDeletedFiles, matchingDeletedFile);
|
||||
renamedPaths.add(path);
|
||||
}
|
||||
}
|
||||
|
||||
for (const path of locallyPossibleCreatedFiles) {
|
||||
if (renamedPaths.has(path) || disappearedPaths.has(path)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
logger.info(
|
||||
`File ${path} was created while offline, scheduling sync to create it`
|
||||
);
|
||||
|
||||
enqueueCreate(path);
|
||||
}
|
||||
|
||||
for (const item of locallyPossiblyDeletedFiles) {
|
||||
if (item.localPath === undefined) {
|
||||
continue;
|
||||
}
|
||||
logger.info(
|
||||
`File ${item.localPath} was deleted while offline, scheduling sync to delete it`
|
||||
);
|
||||
enqueueDelete(item.localPath);
|
||||
}
|
||||
|
||||
for (const path of syncedLocalFiles) {
|
||||
const record = allDocuments.get(path);
|
||||
if (
|
||||
record !== undefined &&
|
||||
record.localPath !== undefined &&
|
||||
record.localPath !== record.remoteRelativePath &&
|
||||
!allLocalFiles.has(record.remoteRelativePath) &&
|
||||
queue.byLocalPath.get(record.remoteRelativePath) === undefined
|
||||
) {
|
||||
// Lost local-rename recovery. The record's `localPath`
|
||||
// (where the user has the file now) and
|
||||
// `remoteRelativePath` (where the server still thinks it
|
||||
// lives) disagree, which means a queued user-rename's
|
||||
// LocalUpdate never reached the server before the queue
|
||||
// was wiped (typically a sync reset). Without this
|
||||
// branch the next `enqueueUpdate({ relativePath: path })`
|
||||
// is a content-only update — server keeps the doc at the
|
||||
// old path, the user's file at the new path orphans, and
|
||||
// other clients never see the rename. Replay the rename
|
||||
// by restoring the OLD localPath so the queue's enqueue
|
||||
// can find the record by `oldPath`, then enqueueUpdate
|
||||
// moves it back to the new path with `isUserRename`.
|
||||
// Only fires when the old slot is genuinely empty
|
||||
// (neither on disk nor claimed by another tracked
|
||||
// record) — otherwise the rename target is occupied and
|
||||
// we'd be confusing the byLocalPath index.
|
||||
const oldPath = record.remoteRelativePath;
|
||||
const newPath = record.localPath;
|
||||
logger.info(
|
||||
`Lost local rename detected: doc ${record.documentId} at ${oldPath} (server) vs ${newPath} (local); replaying rename to server`
|
||||
);
|
||||
await queue.setLocalPath(record.documentId, oldPath);
|
||||
enqueueUpdate({ oldPath, relativePath: newPath });
|
||||
continue;
|
||||
}
|
||||
logger.info(
|
||||
`File ${path} may have been updated while offline, scheduling sync to update it`
|
||||
);
|
||||
enqueueUpdate({ relativePath: path });
|
||||
}
|
||||
}
|
||||
69
frontend/sync-client/src/sync-operations/reconciler.test.ts
Normal file
69
frontend/sync-client/src/sync-operations/reconciler.test.ts
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
import { describe, it } from "node:test";
|
||||
import assert from "node:assert";
|
||||
import { Logger, LogLevel } from "../tracing/logger";
|
||||
import { Settings } from "../persistence/settings";
|
||||
import { STORED_STATE_SCHEMA_VERSION, SyncEventQueue } from "./sync-event-queue";
|
||||
import { Reconciler } from "./reconciler";
|
||||
import { SyncResetError } from "../errors/sync-reset-error";
|
||||
import type { FileOperations } from "../file-operations/file-operations";
|
||||
import type { SyncService } from "../services/sync-service";
|
||||
import type { RelativePath } from "./types";
|
||||
|
||||
describe("Reconciler", () => {
|
||||
it("does not emit an error when placement fetch is interrupted by reset", async () => {
|
||||
const logger = new Logger();
|
||||
const settings = new Settings(logger, {}, async () => {
|
||||
/* no-op */
|
||||
});
|
||||
const queue = new SyncEventQueue(
|
||||
settings,
|
||||
logger,
|
||||
{ schemaVersion: STORED_STATE_SCHEMA_VERSION },
|
||||
async () => {
|
||||
/* no-op */
|
||||
}
|
||||
);
|
||||
|
||||
await queue.upsertRecord({
|
||||
documentId: "DOC-1",
|
||||
parentVersionId: 1,
|
||||
remoteHash: "hash",
|
||||
remoteRelativePath: "remote.md" as RelativePath,
|
||||
localPath: undefined
|
||||
});
|
||||
|
||||
const operations = {
|
||||
exists: async () => false,
|
||||
create: async () => {
|
||||
assert.fail("reset-interrupted placement should not write");
|
||||
}
|
||||
} as unknown as FileOperations;
|
||||
|
||||
const syncService = {
|
||||
getDocumentVersionContent: async () => {
|
||||
throw new SyncResetError();
|
||||
}
|
||||
} as unknown as SyncService;
|
||||
|
||||
const reconciler = new Reconciler(
|
||||
logger,
|
||||
operations,
|
||||
syncService,
|
||||
queue,
|
||||
new Map()
|
||||
);
|
||||
|
||||
await reconciler.run();
|
||||
|
||||
assert.deepStrictEqual(logger.getMessages(LogLevel.ERROR), []);
|
||||
assert.ok(
|
||||
logger
|
||||
.getMessages(LogLevel.INFO)
|
||||
.some((line) =>
|
||||
line.message.includes(
|
||||
"content fetch for DOC-1 interrupted by sync reset"
|
||||
)
|
||||
)
|
||||
);
|
||||
});
|
||||
});
|
||||
1020
frontend/sync-client/src/sync-operations/reconciler.ts
Normal file
1020
frontend/sync-client/src/sync-operations/reconciler.ts
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -0,0 +1,907 @@
|
|||
import { describe, it } from "node:test";
|
||||
import assert from "node:assert";
|
||||
import {
|
||||
STORED_STATE_SCHEMA_VERSION,
|
||||
SyncEventQueue
|
||||
} from "./sync-event-queue";
|
||||
import { Settings } from "../persistence/settings";
|
||||
import { Logger } from "../tracing/logger";
|
||||
import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent";
|
||||
import { SyncEventType } from "./types";
|
||||
import type { DocumentRecord, RelativePath, StoredSyncState } from "./types";
|
||||
|
||||
interface QueueHarness {
|
||||
queue: SyncEventQueue;
|
||||
settings: Settings;
|
||||
saveCalls: StoredSyncState[];
|
||||
}
|
||||
|
||||
function createHarness(
|
||||
options: {
|
||||
ignorePatterns?: string[];
|
||||
initialState?: Partial<StoredSyncState>;
|
||||
omitSchemaVersion?: boolean;
|
||||
} = {}
|
||||
): QueueHarness {
|
||||
const logger = new Logger();
|
||||
const settings = new Settings(
|
||||
logger,
|
||||
{ ignorePatterns: options.ignorePatterns ?? [] },
|
||||
async () => {
|
||||
/* no-op */
|
||||
}
|
||||
);
|
||||
|
||||
const saveCalls: StoredSyncState[] = [];
|
||||
const initialState: Partial<StoredSyncState> | undefined =
|
||||
options.initialState === undefined && options.omitSchemaVersion !== true
|
||||
? { schemaVersion: STORED_STATE_SCHEMA_VERSION }
|
||||
: options.initialState;
|
||||
|
||||
const queue = new SyncEventQueue(
|
||||
settings,
|
||||
logger,
|
||||
initialState,
|
||||
async (data) => {
|
||||
saveCalls.push(data);
|
||||
}
|
||||
);
|
||||
return { queue, settings, saveCalls };
|
||||
}
|
||||
|
||||
function createQueue(ignorePatterns: string[] = []): SyncEventQueue {
|
||||
return createHarness({ ignorePatterns }).queue;
|
||||
}
|
||||
|
||||
function fakeRemoteVersion(
|
||||
documentId: string,
|
||||
overrides: Partial<DocumentVersionWithoutContent> = {}
|
||||
): DocumentVersionWithoutContent {
|
||||
return {
|
||||
vaultUpdateId: 1,
|
||||
documentId,
|
||||
relativePath: `${documentId}.md`,
|
||||
updatedDate: "2026-01-01",
|
||||
isDeleted: false,
|
||||
userId: "user",
|
||||
deviceId: "device",
|
||||
contentSize: 100,
|
||||
isNewFile: true,
|
||||
...overrides
|
||||
};
|
||||
}
|
||||
|
||||
function fakeRecord(
|
||||
documentId: string,
|
||||
overrides: Partial<DocumentRecord> = {}
|
||||
): DocumentRecord {
|
||||
const path = `${documentId.toLowerCase()}.md`;
|
||||
return {
|
||||
documentId,
|
||||
parentVersionId: 1,
|
||||
remoteHash: `hash-${documentId}`,
|
||||
remoteRelativePath: path,
|
||||
localPath: path,
|
||||
...overrides
|
||||
};
|
||||
}
|
||||
|
||||
describe("SyncEventQueue", () => {
|
||||
it("returns enqueued events in FIFO order with no coalescing", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "b.md" });
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "c.md" });
|
||||
await queue.enqueue({ type: SyncEventType.LocalDelete, path: "a.md" });
|
||||
|
||||
const first = await queue.next();
|
||||
assert.strictEqual(first?.type, SyncEventType.LocalCreate);
|
||||
|
||||
const second = await queue.next();
|
||||
assert.strictEqual(second?.type, SyncEventType.LocalCreate);
|
||||
|
||||
const third = await queue.next();
|
||||
assert.strictEqual(third?.type, SyncEventType.LocalDelete);
|
||||
assert.strictEqual(third.documentId, "A");
|
||||
|
||||
assert.strictEqual(await queue.next(), undefined);
|
||||
});
|
||||
|
||||
it("create events are returned FIFO", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "b.md" });
|
||||
|
||||
const first = await queue.next();
|
||||
assert.strictEqual(first?.type, SyncEventType.LocalCreate);
|
||||
assert.strictEqual(first.path, "a.md");
|
||||
|
||||
const second = await queue.next();
|
||||
assert.strictEqual(second?.type, SyncEventType.LocalCreate);
|
||||
assert.strictEqual(second.path, "b.md");
|
||||
});
|
||||
|
||||
it("delete resolves documentId from path", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalDelete, path: "a.md" });
|
||||
|
||||
const event = await queue.next();
|
||||
assert.strictEqual(event?.type, SyncEventType.LocalDelete);
|
||||
assert.strictEqual(event.documentId, "A");
|
||||
});
|
||||
|
||||
it("delete for unknown path is silently ignored", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalDelete,
|
||||
path: "unknown.md"
|
||||
});
|
||||
assert.strictEqual(queue.pendingUpdateCount, 0);
|
||||
});
|
||||
|
||||
it("delete clears the localPath of the affected record", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalDelete, path: "a.md" });
|
||||
|
||||
const record = queue.getDocumentByDocumentId("A");
|
||||
assert.ok(record !== undefined);
|
||||
assert.strictEqual(record.localPath, undefined);
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
});
|
||||
|
||||
it("document store CRUD operations work correctly", async () => {
|
||||
const queue = createQueue();
|
||||
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
assert.strictEqual(queue.syncedDocumentCount, 0);
|
||||
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
assert.strictEqual(queue.syncedDocumentCount, 1);
|
||||
|
||||
const settled = queue.getRecordByLocalPath("a.md" as RelativePath);
|
||||
assert.strictEqual(settled?.documentId, "A");
|
||||
assert.strictEqual(settled.localPath, "a.md");
|
||||
assert.strictEqual(settled.remoteRelativePath, "a.md");
|
||||
|
||||
const found = queue.getDocumentByDocumentId("A");
|
||||
assert.strictEqual(found?.localPath, "a.md");
|
||||
assert.strictEqual(found.documentId, "A");
|
||||
|
||||
await queue.removeDocumentById("A");
|
||||
assert.strictEqual(queue.syncedDocumentCount, 0);
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
assert.strictEqual(queue.getDocumentByDocumentId("A"), undefined);
|
||||
});
|
||||
|
||||
it("LocalUpdate with oldPath moves the document on disk", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalUpdate,
|
||||
path: "b.md",
|
||||
oldPath: "a.md"
|
||||
});
|
||||
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
const moved = queue.getRecordByLocalPath("b.md" as RelativePath);
|
||||
assert.strictEqual(moved?.documentId, "A");
|
||||
assert.strictEqual(moved.localPath, "b.md");
|
||||
|
||||
// The doc's remoteRelativePath is owned by the wire loop, not the
|
||||
// watcher path — a local rename does not move the server-side path.
|
||||
assert.strictEqual(moved.remoteRelativePath, "a.md");
|
||||
});
|
||||
|
||||
it("LocalUpdate rename onto a tracked slot enqueues a delete for the displaced doc", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
await queue.upsertRecord(fakeRecord("B"));
|
||||
|
||||
// User renames a.md onto b.md, clobbering b.md on disk.
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalUpdate,
|
||||
path: "b.md",
|
||||
oldPath: "a.md"
|
||||
});
|
||||
|
||||
// Doc A now lives at b.md.
|
||||
const aRecord = queue.getDocumentByDocumentId("A");
|
||||
assert.strictEqual(aRecord?.localPath, "b.md");
|
||||
const slot = queue.getRecordByLocalPath("b.md" as RelativePath);
|
||||
assert.strictEqual(slot?.documentId, "A");
|
||||
|
||||
// Doc B has no local file anymore (its bytes were overwritten).
|
||||
const bRecord = queue.getDocumentByDocumentId("B");
|
||||
assert.strictEqual(bRecord?.localPath, undefined);
|
||||
|
||||
// Two events should be queued: the LocalDelete for B, then the
|
||||
// LocalUpdate for A (push order in `enqueue`).
|
||||
assert.strictEqual(queue.pendingUpdateCount, 2);
|
||||
|
||||
const first = await queue.next();
|
||||
assert.strictEqual(first?.type, SyncEventType.LocalDelete);
|
||||
assert.strictEqual(first.documentId, "B");
|
||||
assert.strictEqual(first.path, "b.md");
|
||||
|
||||
const second = await queue.next();
|
||||
assert.strictEqual(second?.type, SyncEventType.LocalUpdate);
|
||||
assert.strictEqual(second.documentId, "A");
|
||||
assert.strictEqual(second.path, "b.md");
|
||||
assert.strictEqual(second.isUserRename, true);
|
||||
});
|
||||
|
||||
it("settled record owns a path over a stale pending create", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A", { localPath: "b.md" }));
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "b.md" });
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalUpdate,
|
||||
path: "c.md",
|
||||
oldPath: "b.md"
|
||||
});
|
||||
|
||||
const aRecord = queue.getDocumentByDocumentId("A");
|
||||
assert.strictEqual(aRecord?.localPath, "c.md");
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("b.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("c.md" as RelativePath)?.documentId,
|
||||
"A"
|
||||
);
|
||||
|
||||
const create = await queue.next();
|
||||
assert.strictEqual(create?.type, SyncEventType.LocalCreate);
|
||||
assert.strictEqual(create.path, "b.md");
|
||||
|
||||
const update = await queue.next();
|
||||
assert.strictEqual(update?.type, SyncEventType.LocalUpdate);
|
||||
assert.strictEqual(update.documentId, "A");
|
||||
assert.strictEqual(update.path, "c.md");
|
||||
});
|
||||
|
||||
it("byLocalPath stays consistent across upsertRecord, setLocalPath, and rename", async () => {
|
||||
const queue = createQueue();
|
||||
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
assert.strictEqual(queue.byLocalPath.size, 1);
|
||||
assert.strictEqual(
|
||||
queue.byLocalPath.get("a.md" as RelativePath)?.documentId,
|
||||
"A"
|
||||
);
|
||||
|
||||
// upsertRecord on an existing record with a non-undefined
|
||||
// localPath does NOT rewrite localPath. The watcher path and the
|
||||
// reconciler are the only authorities on localPath of an
|
||||
// already-placed record; letting the wire loop re-key here would
|
||||
// race a user rename that landed during an HTTP roundtrip.
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("A", { localPath: "renamed.md" as RelativePath })
|
||||
);
|
||||
assert.strictEqual(queue.byLocalPath.size, 1);
|
||||
assert.strictEqual(
|
||||
queue.byLocalPath.get("a.md" as RelativePath)?.documentId,
|
||||
"A"
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.byLocalPath.get("renamed.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
assert.strictEqual(queue.getDocumentByDocumentId("A")?.localPath, "a.md");
|
||||
|
||||
// setLocalPath does re-key — it's the explicit path-mutation API.
|
||||
await queue.setLocalPath("A", "later.md" as RelativePath);
|
||||
assert.strictEqual(queue.byLocalPath.size, 1);
|
||||
assert.strictEqual(
|
||||
queue.byLocalPath.get("a.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.byLocalPath.get("later.md" as RelativePath)?.documentId,
|
||||
"A"
|
||||
);
|
||||
|
||||
// setLocalPath to undefined should drop the entry.
|
||||
await queue.setLocalPath("A", undefined);
|
||||
assert.strictEqual(queue.byLocalPath.size, 0);
|
||||
assert.strictEqual(
|
||||
queue.byLocalPath.get("later.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
|
||||
// The record is still tracked by docId.
|
||||
assert.strictEqual(
|
||||
queue.getDocumentByDocumentId("A")?.localPath,
|
||||
undefined
|
||||
);
|
||||
});
|
||||
|
||||
it("upsertRecord installs localPath only when the existing record has none (placement-pending → placed)", async () => {
|
||||
const queue = createQueue();
|
||||
|
||||
// Same-docId-collapse shape: a placement-pending record (created
|
||||
// earlier by a remote-create handler when the slot was occupied)
|
||||
// gets resolved by a LocalCreate that returns the same docId.
|
||||
// The watcher hasn't touched localPath since the record is
|
||||
// placement-pending, so installing the now-known path is correct.
|
||||
await queue.upsertRecord(fakeRecord("A", { localPath: undefined }));
|
||||
assert.strictEqual(queue.byLocalPath.size, 0);
|
||||
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("A", { localPath: "fresh.md" as RelativePath })
|
||||
);
|
||||
assert.strictEqual(queue.byLocalPath.size, 1);
|
||||
assert.strictEqual(
|
||||
queue.byLocalPath.get("fresh.md" as RelativePath)?.documentId,
|
||||
"A"
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.getDocumentByDocumentId("A")?.localPath,
|
||||
"fresh.md"
|
||||
);
|
||||
});
|
||||
|
||||
it("upsertRecord ignores stale localPath from the wire loop after a watcher rename", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
|
||||
// Watcher renames a.md -> renamed.md while the wire loop is
|
||||
// mid-roundtrip. The wire loop captured an earlier snapshot of
|
||||
// localPath and now tries to write it back through upsertRecord.
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalUpdate,
|
||||
path: "renamed.md",
|
||||
oldPath: "a.md"
|
||||
});
|
||||
assert.strictEqual(
|
||||
queue.getDocumentByDocumentId("A")?.localPath,
|
||||
"renamed.md"
|
||||
);
|
||||
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("A", {
|
||||
parentVersionId: 2,
|
||||
remoteRelativePath: "a.md",
|
||||
remoteHash: "hash-A-v2",
|
||||
localPath: "a.md" as RelativePath
|
||||
})
|
||||
);
|
||||
|
||||
// The watcher's rename wins: localPath stays at renamed.md.
|
||||
const record = queue.getDocumentByDocumentId("A");
|
||||
assert.strictEqual(record?.localPath, "renamed.md");
|
||||
assert.strictEqual(record.parentVersionId, 2);
|
||||
assert.strictEqual(record.remoteRelativePath, "a.md");
|
||||
assert.strictEqual(record.remoteHash, "hash-A-v2");
|
||||
assert.strictEqual(
|
||||
queue.byLocalPath.get("renamed.md" as RelativePath)?.documentId,
|
||||
"A"
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.byLocalPath.get("a.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
});
|
||||
|
||||
it("create can be re-enqueued after being dequeued", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
await queue.next();
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
assert.strictEqual(queue.pendingUpdateCount, 1);
|
||||
});
|
||||
|
||||
it("silently ignores create events matching ignore patterns", async () => {
|
||||
const queue = createQueue(["*.tmp", ".hidden/**"]);
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: "scratch.tmp"
|
||||
});
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: ".hidden/secret.md"
|
||||
});
|
||||
assert.strictEqual(queue.pendingUpdateCount, 0);
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: "notes-new.md"
|
||||
});
|
||||
assert.strictEqual(queue.pendingUpdateCount, 1);
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.RemoteChange,
|
||||
remoteVersion: fakeRemoteVersion("N")
|
||||
});
|
||||
assert.strictEqual(queue.pendingUpdateCount, 2);
|
||||
});
|
||||
|
||||
it("addInternalIgnorePattern hides paths from enqueue and survives settings reload", async () => {
|
||||
const harness = createHarness({ ignorePatterns: ["*.tmp"] });
|
||||
const { queue, settings } = harness;
|
||||
|
||||
queue.addInternalIgnorePattern(".vaultlink/**");
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: ".vaultlink/swap"
|
||||
});
|
||||
assert.strictEqual(queue.pendingUpdateCount, 0);
|
||||
|
||||
// User-pattern matching still works alongside the internal pattern.
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: "scratch.tmp"
|
||||
});
|
||||
assert.strictEqual(queue.pendingUpdateCount, 0);
|
||||
|
||||
// Settings reload must not forget the internal pattern.
|
||||
await settings.setSettings({ ignorePatterns: ["*.bak"] });
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: ".vaultlink/another"
|
||||
});
|
||||
assert.strictEqual(queue.pendingUpdateCount, 0);
|
||||
|
||||
// The new user pattern took effect.
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: "old.bak"
|
||||
});
|
||||
assert.strictEqual(queue.pendingUpdateCount, 0);
|
||||
|
||||
// And paths outside both pattern sets still pass through.
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: "notes.md"
|
||||
});
|
||||
assert.strictEqual(queue.pendingUpdateCount, 1);
|
||||
});
|
||||
|
||||
it("clearPending removes events but keeps documents", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "b.md" });
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "c.md" });
|
||||
|
||||
assert.strictEqual(queue.pendingUpdateCount, 2);
|
||||
|
||||
queue.clearPending();
|
||||
|
||||
assert.strictEqual(queue.pendingUpdateCount, 0);
|
||||
assert.strictEqual(queue.syncedDocumentCount, 1);
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath)?.documentId,
|
||||
"A"
|
||||
);
|
||||
});
|
||||
|
||||
it("allSettledDocuments returns all tracked documents that have a localPath", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
await queue.upsertRecord(fakeRecord("B"));
|
||||
// A doc with no local file (e.g. a remote create whose slot was
|
||||
// occupied) should not appear in the localPath-keyed view.
|
||||
await queue.upsertRecord(fakeRecord("C", { localPath: undefined }));
|
||||
|
||||
const docs = queue.allSettledDocuments();
|
||||
assert.strictEqual(docs.size, 2);
|
||||
const paths = Array.from(docs.keys()).sort();
|
||||
assert.deepStrictEqual(paths, ["a.md", "b.md"]);
|
||||
});
|
||||
|
||||
it("loads initial state from persistence", () => {
|
||||
const harness = createHarness({
|
||||
initialState: {
|
||||
schemaVersion: STORED_STATE_SCHEMA_VERSION,
|
||||
documents: [
|
||||
fakeRecord("A", { parentVersionId: 5 }),
|
||||
fakeRecord("B", { parentVersionId: 3 })
|
||||
],
|
||||
lastSeenUpdateId: 4
|
||||
}
|
||||
});
|
||||
const { queue } = harness;
|
||||
|
||||
assert.strictEqual(queue.syncedDocumentCount, 2);
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath)?.documentId,
|
||||
"A"
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("b.md" as RelativePath)?.documentId,
|
||||
"B"
|
||||
);
|
||||
assert.strictEqual(queue.lastSeenUpdateId, 4);
|
||||
});
|
||||
|
||||
it("constructor with mismatched schema version wipes state and saves the new version", () => {
|
||||
const harness = createHarness({
|
||||
initialState: {
|
||||
schemaVersion: 0,
|
||||
documents: [fakeRecord("A"), fakeRecord("B")],
|
||||
lastSeenUpdateId: 7
|
||||
}
|
||||
});
|
||||
|
||||
// Persisted documents and watermark were discarded.
|
||||
assert.strictEqual(harness.queue.syncedDocumentCount, 0);
|
||||
assert.strictEqual(harness.queue.lastSeenUpdateId, 0);
|
||||
|
||||
// The constructor scheduled a save (don't await — fire-and-forget),
|
||||
// but we synchronously enqueued it so it should have landed by now.
|
||||
// The recorded save uses the current schema version.
|
||||
assert.ok(harness.saveCalls.length >= 1);
|
||||
const last = harness.saveCalls[harness.saveCalls.length - 1];
|
||||
assert.strictEqual(last.schemaVersion, STORED_STATE_SCHEMA_VERSION);
|
||||
assert.deepStrictEqual(last.documents, []);
|
||||
assert.strictEqual(last.lastSeenUpdateId, 0);
|
||||
});
|
||||
|
||||
it("constructor with missing schema version also wipes state", () => {
|
||||
const harness = createHarness({
|
||||
initialState: {
|
||||
documents: [fakeRecord("A")],
|
||||
lastSeenUpdateId: 3
|
||||
}
|
||||
});
|
||||
|
||||
assert.strictEqual(harness.queue.syncedDocumentCount, 0);
|
||||
assert.strictEqual(harness.queue.lastSeenUpdateId, 0);
|
||||
assert.ok(harness.saveCalls.length >= 1);
|
||||
assert.strictEqual(
|
||||
harness.saveCalls[harness.saveCalls.length - 1].schemaVersion,
|
||||
STORED_STATE_SCHEMA_VERSION
|
||||
);
|
||||
});
|
||||
|
||||
it("resolveCreate settles the document and resolves the create promise", async () => {
|
||||
const queue = createQueue();
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
|
||||
const event = await queue.next(); // dequeue the create
|
||||
assert.ok(event?.type === SyncEventType.LocalCreate);
|
||||
const createPromise = event.resolvers.promise;
|
||||
|
||||
await queue.resolveCreate(
|
||||
event,
|
||||
fakeRecord("DOC-1", {
|
||||
parentVersionId: 5,
|
||||
localPath: "a.md" as RelativePath,
|
||||
remoteRelativePath: "a.md" as RelativePath
|
||||
})
|
||||
);
|
||||
|
||||
// Document is now settled
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath)?.documentId,
|
||||
"DOC-1"
|
||||
);
|
||||
|
||||
// Promise was resolved
|
||||
assert.strictEqual(await createPromise, "DOC-1");
|
||||
});
|
||||
|
||||
it("delete collapses a pending create that has not started processing", async () => {
|
||||
const queue = createQueue();
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
const create = queue.peekFront();
|
||||
assert.ok(create?.type === SyncEventType.LocalCreate);
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalDelete, path: "a.md" });
|
||||
|
||||
assert.strictEqual(queue.pendingUpdateCount, 0);
|
||||
assert.strictEqual(await queue.next(), undefined);
|
||||
await assert.rejects(create.resolvers.promise, /cancelled/);
|
||||
});
|
||||
|
||||
it("resolveCreate does not claim a localPath after an in-flight pending create was deleted", async () => {
|
||||
const queue = createQueue();
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
const create = queue.peekFront();
|
||||
assert.ok(create?.type === SyncEventType.LocalCreate);
|
||||
create.isProcessing = true;
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalDelete, path: "a.md" });
|
||||
|
||||
await queue.resolveCreate(
|
||||
create,
|
||||
fakeRecord("DOC-1", {
|
||||
localPath: "a.md" as RelativePath,
|
||||
remoteRelativePath: "a.md" as RelativePath
|
||||
})
|
||||
);
|
||||
|
||||
assert.strictEqual(
|
||||
queue.getDocumentByDocumentId("DOC-1")?.localPath,
|
||||
undefined
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
|
||||
const deleteEvent = await queue.next();
|
||||
assert.strictEqual(deleteEvent?.type, SyncEventType.LocalDelete);
|
||||
assert.strictEqual(deleteEvent.documentId, "DOC-1");
|
||||
});
|
||||
|
||||
it("resolveCreate only clears localPath for a pending delete of that path", async () => {
|
||||
const queue = createQueue();
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: "old.md"
|
||||
});
|
||||
const create = queue.peekFront();
|
||||
assert.ok(create?.type === SyncEventType.LocalCreate);
|
||||
create.isProcessing = true;
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalDelete,
|
||||
path: "old.md"
|
||||
});
|
||||
|
||||
await queue.resolveCreate(
|
||||
create,
|
||||
fakeRecord("DOC-1", {
|
||||
localPath: "new.md" as RelativePath,
|
||||
remoteRelativePath: "new.md" as RelativePath
|
||||
})
|
||||
);
|
||||
|
||||
assert.strictEqual(
|
||||
queue.getDocumentByDocumentId("DOC-1")?.localPath,
|
||||
"new.md"
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("new.md" as RelativePath)?.documentId,
|
||||
"DOC-1"
|
||||
);
|
||||
|
||||
const deleteEvent = await queue.next();
|
||||
assert.strictEqual(deleteEvent?.type, SyncEventType.LocalDelete);
|
||||
assert.strictEqual(deleteEvent.documentId, "DOC-1");
|
||||
assert.strictEqual(deleteEvent.path, "old.md");
|
||||
});
|
||||
|
||||
it("pending create owns a same-path delete over a stale deleting record", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("OLD", { localPath: "a.md" as RelativePath })
|
||||
);
|
||||
queue.markServerDeletePending("OLD");
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
const create = queue.peekFront();
|
||||
assert.ok(create?.type === SyncEventType.LocalCreate);
|
||||
create.isProcessing = true;
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalDelete, path: "a.md" });
|
||||
|
||||
assert.strictEqual(
|
||||
queue.getDocumentByDocumentId("OLD")?.localPath,
|
||||
undefined
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
|
||||
const createEvent = await queue.next();
|
||||
assert.strictEqual(createEvent, create);
|
||||
|
||||
const deleteEvent = await queue.next();
|
||||
assert.strictEqual(deleteEvent?.type, SyncEventType.LocalDelete);
|
||||
assert.strictEqual(deleteEvent.documentId, create.resolvers.promise);
|
||||
});
|
||||
|
||||
it("rename of a queued create drains same-path deletes first", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("OLD", { localPath: "target.md" as RelativePath })
|
||||
);
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalCreate,
|
||||
path: "source.md"
|
||||
});
|
||||
const create = queue.peekFront();
|
||||
assert.ok(create?.type === SyncEventType.LocalCreate);
|
||||
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalDelete,
|
||||
path: "target.md"
|
||||
});
|
||||
await queue.enqueue({
|
||||
type: SyncEventType.LocalUpdate,
|
||||
oldPath: "source.md",
|
||||
path: "target.md"
|
||||
});
|
||||
|
||||
const deleteEvent = await queue.next();
|
||||
assert.strictEqual(deleteEvent?.type, SyncEventType.LocalDelete);
|
||||
assert.strictEqual(deleteEvent.documentId, "OLD");
|
||||
assert.strictEqual(deleteEvent.path, "target.md");
|
||||
|
||||
const createEvent = await queue.next();
|
||||
assert.strictEqual(createEvent, create);
|
||||
assert.strictEqual(createEvent.path, "target.md");
|
||||
|
||||
const updateEvent = await queue.next();
|
||||
assert.strictEqual(updateEvent?.type, SyncEventType.LocalUpdate);
|
||||
assert.strictEqual(updateEvent.documentId, create.resolvers.promise);
|
||||
assert.strictEqual(updateEvent.path, "target.md");
|
||||
});
|
||||
|
||||
it("findLatestCreateForPath returns the pending create", async () => {
|
||||
const queue = createQueue();
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "b.md" });
|
||||
|
||||
const found = queue.findLatestCreateForPath("a.md" as RelativePath);
|
||||
assert.ok(found !== undefined);
|
||||
assert.strictEqual(found.path, "a.md");
|
||||
|
||||
const missing = queue.findLatestCreateForPath("c.md" as RelativePath);
|
||||
assert.strictEqual(missing, undefined);
|
||||
});
|
||||
|
||||
it("hasPendingEventsForPath reflects pending events", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
|
||||
assert.strictEqual(
|
||||
queue.hasPendingEventsForPath("a.md" as RelativePath),
|
||||
false
|
||||
);
|
||||
|
||||
await queue.enqueue({ type: SyncEventType.LocalDelete, path: "a.md" });
|
||||
// After a delete the localPath is cleared; an unknown path is treated
|
||||
// as "must be pending creation", so this still returns true.
|
||||
assert.strictEqual(
|
||||
queue.hasPendingEventsForPath("a.md" as RelativePath),
|
||||
true
|
||||
);
|
||||
});
|
||||
|
||||
it("setLocalPath displaces a previous holder of the same path", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("B", { localPath: "b.md" as RelativePath })
|
||||
);
|
||||
|
||||
// Move B onto a.md — the slot already held by A. The invariant
|
||||
// requires A's localPath to be cleared (placement-pending),
|
||||
// and byLocalPath["a.md"] === B.
|
||||
await queue.setLocalPath("B", "a.md" as RelativePath);
|
||||
|
||||
const a = queue.getDocumentByDocumentId("A");
|
||||
const b = queue.getDocumentByDocumentId("B");
|
||||
assert.strictEqual(a?.localPath, undefined);
|
||||
assert.strictEqual(b?.localPath, "a.md");
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath)?.documentId,
|
||||
"B"
|
||||
);
|
||||
// B's old slot is now empty — nothing else moved into it.
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("b.md" as RelativePath),
|
||||
undefined
|
||||
);
|
||||
});
|
||||
|
||||
it("upsertRecord displaces a previous holder of the same path", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
|
||||
// A new record (different docId) claims a.md. The prior holder
|
||||
// (A) must be displaced — its localPath cleared, and
|
||||
// byLocalPath["a.md"] now points at the new record.
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("B", { localPath: "a.md" as RelativePath })
|
||||
);
|
||||
|
||||
const a = queue.getDocumentByDocumentId("A");
|
||||
const b = queue.getDocumentByDocumentId("B");
|
||||
assert.strictEqual(a?.localPath, undefined);
|
||||
assert.strictEqual(b?.localPath, "a.md");
|
||||
assert.strictEqual(
|
||||
queue.getRecordByLocalPath("a.md" as RelativePath)?.documentId,
|
||||
"B"
|
||||
);
|
||||
});
|
||||
|
||||
it("the localPath/byLocalPath invariant holds across rename + recreate cycles", async () => {
|
||||
// Construct the exact same-path create cycle that produces the
|
||||
// bug-D race: docA at P, then docB created at P (via
|
||||
// upsertRecord), and finally a setLocalPath that would move a
|
||||
// third doc onto P. The invariant must hold at every step:
|
||||
// exactly one record has localPath===P at any given time, and
|
||||
// byLocalPath.get(P) returns it.
|
||||
const queue = createQueue();
|
||||
|
||||
const path = "p.md" as RelativePath;
|
||||
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("A", { localPath: path, remoteRelativePath: path })
|
||||
);
|
||||
|
||||
// Sanity: A holds the slot.
|
||||
assert.strictEqual(queue.getRecordByLocalPath(path)?.documentId, "A");
|
||||
assert.strictEqual(queue.getDocumentByDocumentId("A")?.localPath, path);
|
||||
|
||||
// docB created at P via upsertRecord (e.g. a remote create
|
||||
// that races A's local file onto the same slot). A must be
|
||||
// displaced.
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("B", { localPath: path, remoteRelativePath: path })
|
||||
);
|
||||
assert.strictEqual(
|
||||
queue.getDocumentByDocumentId("A")?.localPath,
|
||||
undefined
|
||||
);
|
||||
assert.strictEqual(queue.getDocumentByDocumentId("B")?.localPath, path);
|
||||
assert.strictEqual(queue.getRecordByLocalPath(path)?.documentId, "B");
|
||||
|
||||
// Now setLocalPath moves a third doc C onto P. B must in turn
|
||||
// be displaced; the invariant still holds.
|
||||
await queue.upsertRecord(
|
||||
fakeRecord("C", { localPath: "c.md" as RelativePath })
|
||||
);
|
||||
await queue.setLocalPath("C", path);
|
||||
assert.strictEqual(
|
||||
queue.getDocumentByDocumentId("B")?.localPath,
|
||||
undefined
|
||||
);
|
||||
assert.strictEqual(queue.getDocumentByDocumentId("C")?.localPath, path);
|
||||
assert.strictEqual(queue.getRecordByLocalPath(path)?.documentId, "C");
|
||||
|
||||
// Across the whole cycle exactly one record holds the slot.
|
||||
const holders = Array.from(queue.allRecords()).filter(
|
||||
(r) => r.localPath === path
|
||||
);
|
||||
assert.strictEqual(holders.length, 1);
|
||||
assert.strictEqual(holders[0].documentId, "C");
|
||||
});
|
||||
|
||||
it("clearAllState clears everything", async () => {
|
||||
const queue = createQueue();
|
||||
await queue.upsertRecord(fakeRecord("A"));
|
||||
await queue.enqueue({ type: SyncEventType.LocalCreate, path: "b.md" });
|
||||
|
||||
await queue.clearAllState();
|
||||
|
||||
assert.strictEqual(queue.syncedDocumentCount, 0);
|
||||
assert.strictEqual(queue.pendingUpdateCount, 0);
|
||||
assert.strictEqual(queue.byLocalPath.size, 0);
|
||||
});
|
||||
});
|
||||
1000
frontend/sync-client/src/sync-operations/sync-event-queue.ts
Normal file
1000
frontend/sync-client/src/sync-operations/sync-event-queue.ts
Normal file
File diff suppressed because it is too large
Load diff
File diff suppressed because it is too large
Load diff
74
frontend/sync-client/src/sync-operations/types.ts
Normal file
74
frontend/sync-client/src/sync-operations/types.ts
Normal file
|
|
@ -0,0 +1,74 @@
|
|||
import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent";
|
||||
|
||||
export type VaultUpdateId = number;
|
||||
export type DocumentId = string;
|
||||
export type RelativePath = string;
|
||||
|
||||
export interface DocumentRecord {
|
||||
documentId: DocumentId;
|
||||
parentVersionId: VaultUpdateId;
|
||||
// Hash of the last server version this client has observed for the doc.
|
||||
// `undefined` means we have a record but haven't actually seen content
|
||||
// yet — typically a remote-create whose target slot was occupied at
|
||||
// receive time, where we deliberately defer the fetch to the reconciler.
|
||||
// Consumers should treat undefined as "no comparison possible" (the
|
||||
// fast-skip in `processLocalUpdate` falls through to a real upload).
|
||||
remoteHash: string | undefined;
|
||||
remoteRelativePath: RelativePath;
|
||||
// Where the doc's file currently lives on disk. `undefined` means the doc
|
||||
// has no local file yet — happens for a remote create whose
|
||||
// `remoteRelativePath` slot was occupied at receive time. The reconciler
|
||||
// will place the file once the slot frees, fetching content from the
|
||||
// server on demand.
|
||||
localPath: RelativePath | undefined;
|
||||
}
|
||||
|
||||
export interface StoredSyncState {
|
||||
schemaVersion: number;
|
||||
documents: DocumentRecord[] | undefined;
|
||||
lastSeenUpdateId: VaultUpdateId | undefined;
|
||||
}
|
||||
|
||||
export enum SyncEventType {
|
||||
LocalCreate = "local-create",
|
||||
LocalUpdate = "local-update", // includes both content and path changes
|
||||
LocalDelete = "local-delete",
|
||||
RemoteChange = "remote-change" // includes every type of create/update/delete coming from the server
|
||||
}
|
||||
|
||||
export type FileSyncEvent =
|
||||
| { type: SyncEventType.LocalCreate; path: RelativePath }
|
||||
| {
|
||||
type: SyncEventType.LocalUpdate;
|
||||
path: RelativePath;
|
||||
oldPath?: RelativePath; // oldPath is undefined for content changes
|
||||
}
|
||||
| { type: SyncEventType.LocalDelete; path: RelativePath }
|
||||
| {
|
||||
type: SyncEventType.RemoteChange;
|
||||
remoteVersion: DocumentVersionWithoutContent;
|
||||
};
|
||||
|
||||
export type SyncEvent =
|
||||
| {
|
||||
type: SyncEventType.LocalCreate;
|
||||
path: RelativePath; // current path on disk; mutated in place by `updatePendingCreatePath` when the user renames mid-flight
|
||||
isProcessing: boolean; // true once the wire loop has started this create; deletes after that must wait for the server ack
|
||||
resolvers: PromiseWithResolvers<DocumentId>;
|
||||
}
|
||||
| {
|
||||
type: SyncEventType.LocalUpdate;
|
||||
documentId: DocumentId | Promise<DocumentId>; // if it's a promise, the promise is fulfilled once the document's create event is processed
|
||||
path: RelativePath; // current path on disk
|
||||
originalPath: RelativePath; // original path on disk when the event was queued
|
||||
isUserRename: boolean; // true iff this event was queued because the user renamed the file
|
||||
}
|
||||
| {
|
||||
type: SyncEventType.LocalDelete;
|
||||
documentId: DocumentId | Promise<DocumentId>; // if it's a promise, the promise is fulfilled once the document's create event is processed
|
||||
path: RelativePath; // only used for showing on the UI
|
||||
}
|
||||
| {
|
||||
type: SyncEventType.RemoteChange;
|
||||
remoteVersion: DocumentVersionWithoutContent;
|
||||
};
|
||||
|
|
@ -1,596 +0,0 @@
|
|||
import type {
|
||||
Database,
|
||||
DocumentRecord,
|
||||
RelativePath
|
||||
} from "../persistence/database";
|
||||
|
||||
import { diff } from "reconcile-text";
|
||||
import type { SyncService } from "../services/sync-service";
|
||||
import type { Logger } from "../tracing/logger";
|
||||
import type {
|
||||
CommonHistoryEntry,
|
||||
SyncCreateDetails,
|
||||
SyncDeleteDetails,
|
||||
SyncDetails,
|
||||
SyncHistory,
|
||||
SyncMovedDetails,
|
||||
SyncUpdateDetails
|
||||
} from "../tracing/sync-history";
|
||||
import { SyncStatus, SyncType } from "../tracing/sync-history";
|
||||
import { EMPTY_HASH, hash } from "../utils/hash";
|
||||
|
||||
import { base64ToBytes } from "byte-base64";
|
||||
import type { Settings } from "../persistence/settings";
|
||||
import type { FileOperations } from "../file-operations/file-operations";
|
||||
import { createPromise } from "../utils/create-promise";
|
||||
import { FileNotFoundError } from "../file-operations/file-not-found-error";
|
||||
import { SyncResetError } from "../services/sync-reset-error";
|
||||
import { globsToRegexes } from "../utils/globs-to-regexes";
|
||||
import type { DocumentVersion } from "../services/types/DocumentVersion";
|
||||
import type { DocumentUpdateResponse } from "../services/types/DocumentUpdateResponse";
|
||||
import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent";
|
||||
import type { FixedSizeDocumentCache } from "../utils/data-structures/fix-sized-cache";
|
||||
import { isFileTypeMergable } from "../utils/is-file-type-mergable";
|
||||
import { isBinary } from "../utils/is-binary";
|
||||
import type { ServerConfig } from "../services/server-config";
|
||||
|
||||
export class UnrestrictedSyncer {
|
||||
private ignorePatterns: RegExp[];
|
||||
|
||||
public constructor(
|
||||
private readonly logger: Logger,
|
||||
private readonly database: Database,
|
||||
private readonly settings: Settings,
|
||||
private readonly syncService: SyncService,
|
||||
private readonly operations: FileOperations,
|
||||
private readonly history: SyncHistory,
|
||||
private readonly contentCache: FixedSizeDocumentCache,
|
||||
private readonly serverConfig: ServerConfig
|
||||
) {
|
||||
this.ignorePatterns = globsToRegexes(
|
||||
this.settings.getSettings().ignorePatterns,
|
||||
this.logger
|
||||
);
|
||||
|
||||
this.settings.onSettingsChanged.add((newSettings) => {
|
||||
this.ignorePatterns = globsToRegexes(
|
||||
newSettings.ignorePatterns,
|
||||
this.logger
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
public async unrestrictedSyncLocallyCreatedFile(
|
||||
document: DocumentRecord
|
||||
): Promise<void> {
|
||||
const updateDetails: SyncCreateDetails = {
|
||||
type: SyncType.CREATE,
|
||||
relativePath: document.relativePath
|
||||
};
|
||||
|
||||
return this.executeSync(updateDetails, async () => {
|
||||
const originalRelativePath = document.relativePath;
|
||||
if (document.isDeleted) {
|
||||
this.logger.debug(
|
||||
`Document ${originalRelativePath} has been already deleted, no need to create it`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const contentBytes =
|
||||
await this.operations.read(originalRelativePath); // this can throw FileNotFoundError
|
||||
const contentHash = hash(contentBytes);
|
||||
|
||||
const response = await this.syncService.create({
|
||||
documentId: document.documentId,
|
||||
relativePath: originalRelativePath,
|
||||
contentBytes
|
||||
});
|
||||
|
||||
// In case a document with the same name (but different ID) had existed remotely that we haven't known about
|
||||
if (response.relativePath != originalRelativePath) {
|
||||
this.logger.debug(
|
||||
`Document ${originalRelativePath} has been created remotely at a different path: ${response.relativePath}, moving it locally`
|
||||
);
|
||||
await this.operations.move(
|
||||
document.relativePath,
|
||||
response.relativePath
|
||||
); // this can throw FileNotFoundError
|
||||
}
|
||||
|
||||
this.database.updateDocumentMetadata(
|
||||
{
|
||||
parentVersionId: response.vaultUpdateId,
|
||||
hash: contentHash,
|
||||
remoteRelativePath: response.relativePath
|
||||
},
|
||||
document
|
||||
);
|
||||
|
||||
this.database.addSeenUpdateId(response.vaultUpdateId);
|
||||
await this.updateCache(
|
||||
response.vaultUpdateId,
|
||||
contentBytes,
|
||||
response.relativePath
|
||||
);
|
||||
|
||||
this.history.addHistoryEntry({
|
||||
status: SyncStatus.SUCCESS,
|
||||
details: updateDetails,
|
||||
message: `Successfully uploaded locally created file`
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
public async unrestrictedSyncLocallyDeletedFile(
|
||||
document: DocumentRecord
|
||||
): Promise<void> {
|
||||
const updateDetails: SyncDeleteDetails = {
|
||||
type: SyncType.DELETE,
|
||||
relativePath: document.relativePath
|
||||
};
|
||||
|
||||
await this.executeSync(updateDetails, async () => {
|
||||
const response = await this.syncService.delete({
|
||||
documentId: document.documentId,
|
||||
relativePath: document.relativePath
|
||||
});
|
||||
|
||||
this.database.updateDocumentMetadata(
|
||||
{
|
||||
parentVersionId: response.vaultUpdateId,
|
||||
hash: EMPTY_HASH,
|
||||
remoteRelativePath: document.relativePath
|
||||
},
|
||||
document
|
||||
);
|
||||
|
||||
this.database.addSeenUpdateId(response.vaultUpdateId);
|
||||
|
||||
this.history.addHistoryEntry({
|
||||
status: SyncStatus.SUCCESS,
|
||||
details: updateDetails,
|
||||
message: `Successfully deleted locally deleted file on the server`,
|
||||
author: response.userId
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
public async unrestrictedSyncLocallyUpdatedFile({
|
||||
oldPath,
|
||||
document,
|
||||
// We use the same code path for both local and remote updates. We need to force the update
|
||||
// if there are no local changes but we know that the remote version is newer.
|
||||
force = false
|
||||
}: {
|
||||
oldPath?: RelativePath;
|
||||
force?: boolean;
|
||||
document: DocumentRecord;
|
||||
}): Promise<void> {
|
||||
const updateDetails: SyncUpdateDetails | SyncMovedDetails =
|
||||
oldPath !== undefined
|
||||
? {
|
||||
type: SyncType.MOVE,
|
||||
relativePath: document.relativePath,
|
||||
movedFrom: oldPath
|
||||
}
|
||||
: {
|
||||
type: SyncType.UPDATE,
|
||||
relativePath: document.relativePath
|
||||
};
|
||||
|
||||
await this.executeSync(updateDetails, async () => {
|
||||
const originalRelativePath = document.relativePath;
|
||||
|
||||
if (document.isDeleted || document.metadata === undefined) {
|
||||
this.logger.debug(
|
||||
`Document ${document.relativePath} has been already deleted, no need to update it`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const contentBytes = await this.operations.read(
|
||||
document.relativePath
|
||||
); // this can throw FileNotFoundError
|
||||
let contentHash = hash(contentBytes);
|
||||
|
||||
const areThereLocalChanges = !(
|
||||
document.metadata.hash === contentHash && oldPath === undefined
|
||||
);
|
||||
|
||||
let response: DocumentVersion | DocumentUpdateResponse | undefined =
|
||||
undefined;
|
||||
|
||||
if (areThereLocalChanges) {
|
||||
const isText =
|
||||
!isBinary(contentBytes) &&
|
||||
isFileTypeMergable(
|
||||
document.relativePath,
|
||||
(await this.serverConfig.getConfig())
|
||||
.mergeableFileExtensions
|
||||
);
|
||||
const cachedVersion = this.contentCache.get(
|
||||
document.metadata.parentVersionId
|
||||
);
|
||||
|
||||
response =
|
||||
isText && cachedVersion !== undefined
|
||||
? await this.syncService.putText({
|
||||
documentId: document.documentId,
|
||||
parentVersionId:
|
||||
document.metadata.parentVersionId,
|
||||
relativePath: document.relativePath,
|
||||
content: diff(
|
||||
new TextDecoder().decode(cachedVersion),
|
||||
new TextDecoder().decode(contentBytes)
|
||||
)
|
||||
})
|
||||
: await this.syncService.putBinary({
|
||||
documentId: document.documentId,
|
||||
parentVersionId:
|
||||
document.metadata.parentVersionId,
|
||||
relativePath: document.relativePath,
|
||||
contentBytes
|
||||
});
|
||||
} else {
|
||||
if (!force) {
|
||||
this.logger.debug(
|
||||
`File hash of ${document.relativePath} matches with last synced version and the path hasn't changed; no need to sync`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
response = await this.syncService.get({
|
||||
documentId: document.documentId
|
||||
});
|
||||
}
|
||||
|
||||
// `document` is mutable and reflects the latest state in the local database
|
||||
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
|
||||
if (document.isDeleted) {
|
||||
this.logger.info(
|
||||
`Document ${document.relativePath} has been deleted before we could finish updating it`
|
||||
);
|
||||
this.database.addSeenUpdateId(response.vaultUpdateId);
|
||||
return;
|
||||
}
|
||||
|
||||
if (
|
||||
// `Syncer` creates fake local document metadata for all remote docs with invalid hashes. The parent IDs will likely match
|
||||
// the latest versions so we still need to update the local versions to turn the fakes into real metadata.
|
||||
document.metadata.parentVersionId > response.vaultUpdateId
|
||||
) {
|
||||
this.logger.debug(
|
||||
`Document ${document.relativePath} is already more up to date than the fetched version`
|
||||
);
|
||||
this.database.addSeenUpdateId(response.vaultUpdateId); // in case the previous `vaultUpdateId` update hasn't made it through
|
||||
return;
|
||||
}
|
||||
|
||||
if (response.isDeleted) {
|
||||
return this.applyRemoteDeleteLocally(document, response);
|
||||
}
|
||||
|
||||
let actualPath = document.relativePath;
|
||||
|
||||
if (response.relativePath != originalRelativePath) {
|
||||
actualPath = response.relativePath;
|
||||
// Make sure to update the remote relative path to avoid uploading
|
||||
// the file as a result of this filesystem event.
|
||||
document.metadata.remoteRelativePath = response.relativePath;
|
||||
await this.operations.move(
|
||||
document.relativePath,
|
||||
response.relativePath
|
||||
); // this can throw FileNotFoundError
|
||||
}
|
||||
|
||||
if (!("type" in response) || response.type === "MergingUpdate") {
|
||||
const responseBytes = base64ToBytes(response.contentBase64);
|
||||
contentHash = hash(responseBytes);
|
||||
|
||||
this.database.updateDocumentMetadata(
|
||||
{
|
||||
parentVersionId: response.vaultUpdateId,
|
||||
hash: contentHash,
|
||||
remoteRelativePath: response.relativePath
|
||||
},
|
||||
document
|
||||
);
|
||||
await this.operations.write(
|
||||
actualPath,
|
||||
contentBytes,
|
||||
responseBytes
|
||||
);
|
||||
await this.updateCache(
|
||||
response.vaultUpdateId,
|
||||
responseBytes,
|
||||
actualPath
|
||||
);
|
||||
|
||||
if (!force) {
|
||||
this.history.addHistoryEntry({
|
||||
status: SyncStatus.SUCCESS,
|
||||
details: updateDetails,
|
||||
message: `The file we updated had been updated remotely, so we downloaded the merged version`
|
||||
});
|
||||
}
|
||||
} else {
|
||||
this.database.updateDocumentMetadata(
|
||||
{
|
||||
parentVersionId: response.vaultUpdateId,
|
||||
hash: contentHash,
|
||||
remoteRelativePath: response.relativePath
|
||||
},
|
||||
document
|
||||
);
|
||||
await this.updateCache(
|
||||
response.vaultUpdateId,
|
||||
contentBytes,
|
||||
actualPath
|
||||
);
|
||||
}
|
||||
|
||||
this.database.addSeenUpdateId(response.vaultUpdateId);
|
||||
|
||||
const actualUpdateDetails: SyncUpdateDetails | SyncMovedDetails =
|
||||
oldPath !== undefined ||
|
||||
response.relativePath != originalRelativePath
|
||||
? {
|
||||
type: SyncType.MOVE,
|
||||
relativePath: response.relativePath,
|
||||
movedFrom: originalRelativePath
|
||||
}
|
||||
: {
|
||||
type: SyncType.UPDATE,
|
||||
relativePath: response.relativePath
|
||||
};
|
||||
|
||||
if (areThereLocalChanges) {
|
||||
this.history.addHistoryEntry({
|
||||
status: SyncStatus.SUCCESS,
|
||||
details: actualUpdateDetails,
|
||||
message: `Successfully uploaded locally updated file to the server`,
|
||||
author: response.userId
|
||||
});
|
||||
} else {
|
||||
this.history.addHistoryEntry({
|
||||
status: SyncStatus.SUCCESS,
|
||||
details: actualUpdateDetails,
|
||||
message: `Successfully downloaded remotely updated file from the server`,
|
||||
author: response.userId,
|
||||
timestamp: new Date(response.updatedDate)
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public async unrestrictedSyncRemotelyUpdatedFile(
|
||||
remoteVersion: DocumentVersionWithoutContent,
|
||||
document?: DocumentRecord
|
||||
): Promise<void> {
|
||||
const updateDetails: SyncCreateDetails = {
|
||||
type: SyncType.CREATE,
|
||||
relativePath: remoteVersion.relativePath
|
||||
};
|
||||
|
||||
await this.executeSync(updateDetails, async () => {
|
||||
if (document?.metadata !== undefined) {
|
||||
// If the file exists locally, let's pretend the user has updated it
|
||||
// and deal with remote update/deletion within `unrestrictedSyncLocallyUpdatedFile`
|
||||
if (
|
||||
document.metadata.parentVersionId >=
|
||||
remoteVersion.vaultUpdateId
|
||||
) {
|
||||
this.logger.debug(
|
||||
`Document ${remoteVersion.relativePath} is already at least as up to date as the fetched version`
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
return this.unrestrictedSyncLocallyUpdatedFile({
|
||||
document,
|
||||
force: true
|
||||
});
|
||||
} else if (remoteVersion.isDeleted) {
|
||||
// Either the document hasn't made it to us before and therefore we don't need to delete it,
|
||||
// or we already have it, in which case the preceeding if would've dealt with it
|
||||
this.logger.debug(
|
||||
`Document ${remoteVersion.relativePath} has been deleted remotely, no need to sync`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// Don't download oversized files
|
||||
const historyEntryForSkippedOversizedFile =
|
||||
this.getHistoryEntryForSkippedOversizedFile(
|
||||
remoteVersion.contentSize,
|
||||
remoteVersion.relativePath
|
||||
);
|
||||
if (historyEntryForSkippedOversizedFile !== undefined) {
|
||||
this.history.addHistoryEntry(
|
||||
historyEntryForSkippedOversizedFile
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const contentBytes =
|
||||
await this.syncService.getDocumentVersionContent({
|
||||
documentId: remoteVersion.documentId,
|
||||
vaultUpdateId: remoteVersion.vaultUpdateId
|
||||
});
|
||||
|
||||
// We're trying to create an entirely new document that didn't exist locally
|
||||
document = this.database.getDocumentByDocumentId(
|
||||
remoteVersion.documentId
|
||||
);
|
||||
// It can happen that a concurrent sync operation has already created the document, so we can bail here
|
||||
if (document !== undefined) {
|
||||
this.logger.debug(
|
||||
`Document ${remoteVersion.relativePath} has already been created locally, no need to create it again`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
await this.operations.ensureClearPath(remoteVersion.relativePath);
|
||||
|
||||
const [promise, resolve] = createPromise();
|
||||
this.database.updateDocumentMetadata(
|
||||
{
|
||||
parentVersionId: remoteVersion.vaultUpdateId,
|
||||
hash: hash(contentBytes),
|
||||
remoteRelativePath: remoteVersion.relativePath
|
||||
},
|
||||
this.database.createNewPendingDocument(
|
||||
remoteVersion.documentId,
|
||||
remoteVersion.relativePath,
|
||||
promise
|
||||
)
|
||||
);
|
||||
|
||||
await this.operations.create(
|
||||
remoteVersion.relativePath,
|
||||
contentBytes
|
||||
);
|
||||
await this.updateCache(
|
||||
remoteVersion.vaultUpdateId,
|
||||
contentBytes,
|
||||
remoteVersion.relativePath
|
||||
);
|
||||
|
||||
resolve();
|
||||
this.database.removeDocumentPromise(promise);
|
||||
|
||||
this.history.addHistoryEntry({
|
||||
status: SyncStatus.SUCCESS,
|
||||
details: updateDetails,
|
||||
message: `Successfully downloaded remote file which hadn't existed locally`,
|
||||
author: remoteVersion.userId,
|
||||
timestamp: new Date(remoteVersion.updatedDate)
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
public async executeSync<T>(
|
||||
details: SyncDetails,
|
||||
fn: () => Promise<T>
|
||||
): Promise<T | undefined> {
|
||||
for (const pattern of this.ignorePatterns) {
|
||||
if (pattern.test(details.relativePath)) {
|
||||
this.logger.debug(
|
||||
`File '${details.relativePath}' is ignored by the ignore pattern: ${pattern}`
|
||||
);
|
||||
return; // bail without SKIPPED status because we were told to ignore this file and we shouldn't clutter up the history
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
// Only check the size of files which already exist locally.
|
||||
if (await this.operations.exists(details.relativePath)) {
|
||||
const sizeInBytes = await this.operations.getFileSize(
|
||||
details.relativePath
|
||||
);
|
||||
const historyEntryForSkippedOversizedFile =
|
||||
this.getHistoryEntryForSkippedOversizedFile(
|
||||
sizeInBytes,
|
||||
details.relativePath
|
||||
);
|
||||
if (historyEntryForSkippedOversizedFile !== undefined) {
|
||||
this.history.addHistoryEntry(
|
||||
historyEntryForSkippedOversizedFile
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
return await fn();
|
||||
} catch (e) {
|
||||
if (e instanceof FileNotFoundError) {
|
||||
// A subsequent sync operation must have been creating to deal with this
|
||||
this.logger.info(
|
||||
`Skiping file '${details.relativePath}' because it no longer exists when trying to ${details.type.toLocaleLowerCase()} it`
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (e instanceof SyncResetError) {
|
||||
this.logger.info(
|
||||
`Interrupting sync operation because of a reset`
|
||||
);
|
||||
return;
|
||||
} else {
|
||||
this.history.addHistoryEntry({
|
||||
status: SyncStatus.ERROR,
|
||||
details,
|
||||
message: `Failed to sync file '${details.relativePath}' because of ${e} when trying to ${details.type.toLocaleLowerCase()} it`
|
||||
});
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private getHistoryEntryForSkippedOversizedFile(
|
||||
sizeInBytes: number,
|
||||
relativePath: RelativePath
|
||||
): CommonHistoryEntry | undefined {
|
||||
const sizeInMB = Math.round(sizeInBytes / 1024 / 1024);
|
||||
const { maxFileSizeMB } = this.settings.getSettings();
|
||||
if (sizeInMB > maxFileSizeMB) {
|
||||
return {
|
||||
status: SyncStatus.SKIPPED,
|
||||
details: {
|
||||
type: SyncType.SKIPPED,
|
||||
relativePath
|
||||
},
|
||||
message: `File size of ${sizeInMB} MB exceeds the maximum file size limit of ${
|
||||
maxFileSizeMB
|
||||
} MB`
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
private async updateCache(
|
||||
updateId: number,
|
||||
contentBytes: Uint8Array,
|
||||
filePath: RelativePath
|
||||
): Promise<void> {
|
||||
if (
|
||||
isFileTypeMergable(
|
||||
filePath,
|
||||
(await this.serverConfig.getConfig()).mergeableFileExtensions
|
||||
) &&
|
||||
!isBinary(contentBytes)
|
||||
) {
|
||||
this.contentCache.put(updateId, contentBytes);
|
||||
}
|
||||
}
|
||||
|
||||
private async applyRemoteDeleteLocally(
|
||||
document: DocumentRecord,
|
||||
response: DocumentVersion | DocumentUpdateResponse
|
||||
): Promise<void> {
|
||||
this.history.addHistoryEntry({
|
||||
status: SyncStatus.SUCCESS,
|
||||
details: {
|
||||
type: SyncType.DELETE,
|
||||
relativePath: document.relativePath
|
||||
},
|
||||
message: "File has been deleted remotely, so we deleted it locally",
|
||||
author: response.userId,
|
||||
timestamp: new Date(response.updatedDate)
|
||||
});
|
||||
|
||||
this.database.delete(document.relativePath);
|
||||
this.database.updateDocumentMetadata(
|
||||
{
|
||||
parentVersionId: response.vaultUpdateId,
|
||||
hash: EMPTY_HASH,
|
||||
remoteRelativePath: response.relativePath
|
||||
},
|
||||
document
|
||||
);
|
||||
|
||||
await this.operations.delete(document.relativePath);
|
||||
|
||||
this.database.addSeenUpdateId(response.vaultUpdateId);
|
||||
}
|
||||
}
|
||||
|
|
@ -2,7 +2,7 @@ import {
|
|||
MAX_HISTORY_ENTRY_COUNT,
|
||||
TIMEOUT_FOR_MERGING_HISTORY_ENTRIES_IN_SECONDS
|
||||
} from "../consts";
|
||||
import type { RelativePath } from "../persistence/database";
|
||||
import type { RelativePath } from "../sync-operations/types";
|
||||
import type { Logger } from "./logger";
|
||||
import { removeFromArray } from "../utils/remove-from-array";
|
||||
import { EventListeners } from "../utils/data-structures/event-listeners";
|
||||
|
|
@ -28,7 +28,7 @@ export interface SyncDeleteDetails {
|
|||
relativePath: RelativePath;
|
||||
}
|
||||
|
||||
export interface SyncSkippedDetails {
|
||||
interface SyncSkippedDetails {
|
||||
type: SyncType.SKIPPED;
|
||||
relativePath: RelativePath;
|
||||
}
|
||||
|
|
@ -40,12 +40,15 @@ export type SyncDetails =
|
|||
| SyncMovedDetails
|
||||
| SyncSkippedDetails;
|
||||
|
||||
export interface CommonHistoryEntry {
|
||||
export interface HistoryEntry {
|
||||
status: SyncStatus;
|
||||
message: string;
|
||||
details: SyncDetails;
|
||||
timestamp: Date;
|
||||
// `author` is the server-side user id and only exists for entries that
|
||||
// round-tripped through the server. Local-only entries (e.g. SKIPPED)
|
||||
// legitimately have no author.
|
||||
author?: string;
|
||||
timestamp?: Date;
|
||||
}
|
||||
|
||||
export enum SyncType {
|
||||
|
|
@ -62,8 +65,6 @@ export enum SyncStatus {
|
|||
SKIPPED = "SKIPPED"
|
||||
}
|
||||
|
||||
export type HistoryEntry = CommonHistoryEntry & { timestamp: Date };
|
||||
|
||||
export interface HistoryStats {
|
||||
success: number;
|
||||
error: number;
|
||||
|
|
@ -88,30 +89,25 @@ export class SyncHistory {
|
|||
}
|
||||
|
||||
/**
|
||||
* Insert the entry at the beginning of the history list. If the entry
|
||||
* already in the list, it will get moved to the beginning and updated.
|
||||
*
|
||||
* If the entry list is too long, the oldest entry will be removed.
|
||||
*/
|
||||
public addHistoryEntry(entry: CommonHistoryEntry): void {
|
||||
const historyEntry = {
|
||||
...entry,
|
||||
timestamp: entry.timestamp ?? new Date()
|
||||
};
|
||||
|
||||
const candidate = this.findSimilarRecentUpdateEntry(historyEntry);
|
||||
* Insert the entry at the beginning of the history list. If the entry
|
||||
* already in the list, it will get moved to the beginning and updated.
|
||||
*
|
||||
* If the entry list is too long, the oldest entry will be removed.
|
||||
*/
|
||||
public addHistoryEntry(entry: HistoryEntry): void {
|
||||
const candidate = this.findSimilarRecentUpdateEntry(entry);
|
||||
if (candidate !== undefined) {
|
||||
removeFromArray(this._entries, candidate);
|
||||
}
|
||||
|
||||
// Insert the entry at the beginning
|
||||
this._entries.unshift(historyEntry);
|
||||
this._entries.unshift(entry);
|
||||
|
||||
if (this._entries.length > MAX_HISTORY_ENTRY_COUNT) {
|
||||
this._entries.pop();
|
||||
}
|
||||
|
||||
this.updateSuccessCount(historyEntry);
|
||||
this.updateSuccessCount(entry);
|
||||
}
|
||||
|
||||
public reset(): void {
|
||||
|
|
|
|||
|
|
@ -12,7 +12,5 @@
|
|||
"declaration": true,
|
||||
"declarationDir": "./dist/types"
|
||||
},
|
||||
"exclude": [
|
||||
"./dist"
|
||||
]
|
||||
"exclude": ["./dist"]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -49,11 +49,6 @@ module.exports = [
|
|||
type: "umd"
|
||||
},
|
||||
globalObject: "this"
|
||||
},
|
||||
resolve: {
|
||||
fallback: {
|
||||
ws: false // Exclude `ws` from the browser bundle
|
||||
}
|
||||
}
|
||||
}),
|
||||
merge(common, {
|
||||
|
|
@ -62,10 +57,6 @@ module.exports = [
|
|||
path: path.resolve(__dirname, "dist"),
|
||||
filename: "sync-client.node.js",
|
||||
libraryTarget: "commonjs2"
|
||||
},
|
||||
externals: {
|
||||
bufferutil: "bufferutil",
|
||||
"utf-8-validate": "utf-8-validate" // required for ws: https://github.com/websockets/ws/issues/2245#issuecomment-2250318733
|
||||
}
|
||||
})
|
||||
];
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue