Simplify syncing logic
This commit is contained in:
parent
e8c57b3a37
commit
4493365076
48 changed files with 1054 additions and 918 deletions
|
|
@ -4,17 +4,14 @@ import type {
|
|||
DocumentRecord,
|
||||
RelativePath
|
||||
} from "../persistence/database";
|
||||
import type { SyncService } from "../services/sync-service";
|
||||
import type { Logger } from "../tracing/logger";
|
||||
import PQueue from "p-queue";
|
||||
import { hash } from "../utils/hash";
|
||||
import { v4 as uuidv4 } from "uuid";
|
||||
import type { Settings } from "../persistence/settings";
|
||||
import type { FileOperations } from "../file-operations/file-operations";
|
||||
import { findMatchingFile } from "../utils/find-matching-file";
|
||||
import type { UnrestrictedSyncer } from "./unrestricted-syncer";
|
||||
import { createPromise } from "../utils/create-promise";
|
||||
import { SyncResetError } from "../services/sync-reset-error";
|
||||
import { SyncResetError } from "../errors/sync-reset-error";
|
||||
import { Locks } from "../utils/data-structures/locks";
|
||||
import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent";
|
||||
import type { WebSocketVaultUpdate } from "../services/types/WebSocketVaultUpdate";
|
||||
|
|
@ -28,7 +25,7 @@ export class Syncer {
|
|||
(remainingOperations: number) => unknown
|
||||
>();
|
||||
|
||||
private readonly remoteDocumentsLock: Locks<DocumentId>;
|
||||
public readonly updatedDocumentsByPathAndKeysLocks: Locks<string>; // can be DocumentId or RelativePath
|
||||
|
||||
// FIFO to limit the number of concurrent sync operations
|
||||
private readonly syncQueue: PQueue;
|
||||
|
|
@ -42,16 +39,18 @@ export class Syncer {
|
|||
private readonly logger: Logger,
|
||||
private readonly database: Database,
|
||||
private readonly settings: Settings,
|
||||
private readonly syncService: SyncService,
|
||||
private readonly webSocketManager: WebSocketManager,
|
||||
private readonly operations: FileOperations,
|
||||
private readonly internalSyncer: UnrestrictedSyncer
|
||||
private readonly unrestrictedSyncer: UnrestrictedSyncer
|
||||
) {
|
||||
this.syncQueue = new PQueue({
|
||||
concurrency: settings.getSettings().syncConcurrency
|
||||
});
|
||||
|
||||
this.remoteDocumentsLock = new Locks<DocumentId>(this.logger);
|
||||
this.updatedDocumentsByPathAndKeysLocks = new Locks<DocumentId>(
|
||||
Syncer.name,
|
||||
this.logger
|
||||
);
|
||||
|
||||
settings.onSettingsChanged.add((newSettings, oldSettings) => {
|
||||
if (newSettings.syncConcurrency !== oldSettings.syncConcurrency) {
|
||||
|
|
@ -83,52 +82,50 @@ export class Syncer {
|
|||
return this._isFirstSyncComplete;
|
||||
}
|
||||
|
||||
public hasPendingOperationsForDocument(relativePath: string): boolean {
|
||||
return this.updatedDocumentsByPathAndKeysLocks.isLocked(relativePath);
|
||||
}
|
||||
|
||||
public async syncLocallyCreatedFile(
|
||||
relativePath: RelativePath
|
||||
): Promise<void> {
|
||||
// check whether someone else has already created the document in the database
|
||||
if (
|
||||
this.database.getLatestDocumentByRelativePath(relativePath)
|
||||
?.isDeleted === false
|
||||
) {
|
||||
// This is likely a consequence of us creating a file because of a remote update
|
||||
// which triggered a local create, so we don't need to do anything here.
|
||||
this.logger.debug(
|
||||
`Document ${relativePath} already exists in the database, skipping`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
const document = this.database.createNewPendingDocument(relativePath);
|
||||
|
||||
const id = uuidv4();
|
||||
const document = this.database.createNewPendingDocument(
|
||||
id,
|
||||
relativePath,
|
||||
promise
|
||||
await this.enqueueSyncOperation(
|
||||
async () =>
|
||||
this.unrestrictedSyncer.unrestrictedSyncLocallyCreatedOrUpdatedFile(
|
||||
{
|
||||
document
|
||||
}
|
||||
),
|
||||
[relativePath]
|
||||
);
|
||||
|
||||
try {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncLocallyCreatedFile(document)
|
||||
);
|
||||
|
||||
resolve();
|
||||
} catch (e) {
|
||||
reject(e);
|
||||
} finally {
|
||||
this.database.removeDocumentPromise(promise);
|
||||
}
|
||||
}
|
||||
|
||||
public async syncLocallyDeletedFile(
|
||||
relativePath: RelativePath
|
||||
): Promise<void> {
|
||||
if (
|
||||
this.database.getLatestDocumentByRelativePath(relativePath)
|
||||
?.isDeleted === true
|
||||
) {
|
||||
let document =
|
||||
this.database.getLatestDocumentByRelativePath(relativePath);
|
||||
|
||||
if (document == null || document.isDeleted === true) {
|
||||
// This is must be a consequence of us deleting a file because of a remote update
|
||||
// which triggered a local delete, so we don't need to do anything here.
|
||||
this.logger.debug(
|
||||
`Document ${relativePath} has already been markes as deleted, skipping`
|
||||
`Document ${relativePath} has already been marked as deleted, skipping`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
|
@ -137,26 +134,13 @@ export class Syncer {
|
|||
// document which finishes after the delete has succeeded and would introduce a phantom metadata record.
|
||||
this.database.delete(relativePath);
|
||||
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
|
||||
const document = await this.database.getResolvedDocumentByRelativePath(
|
||||
relativePath,
|
||||
promise
|
||||
);
|
||||
|
||||
try {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncLocallyDeletedFile(document)
|
||||
await this.enqueueSyncOperation(async () => {
|
||||
await this.unrestrictedSyncer.unrestrictedSyncLocallyDeletedFile(
|
||||
document
|
||||
);
|
||||
|
||||
resolve();
|
||||
|
||||
this.database.removeDocument(document);
|
||||
} catch (e) {
|
||||
reject(e);
|
||||
} finally {
|
||||
this.database.removeDocumentPromise(promise);
|
||||
}
|
||||
}, [document?.metadata?.documentId, relativePath]);
|
||||
}
|
||||
|
||||
public async syncLocallyUpdatedFile({
|
||||
|
|
@ -166,38 +150,10 @@ export class Syncer {
|
|||
oldPath?: RelativePath;
|
||||
relativePath: RelativePath;
|
||||
}): Promise<void> {
|
||||
if (oldPath !== undefined) {
|
||||
// We might have moved the document in the database before calling this method,
|
||||
// in that case, we mustn't move it again.
|
||||
if (
|
||||
this.database.getLatestDocumentByRelativePath(relativePath) ===
|
||||
undefined ||
|
||||
this.database.getLatestDocumentByRelativePath(relativePath)
|
||||
?.isDeleted === true
|
||||
) {
|
||||
if (oldPath === relativePath) {
|
||||
throw new Error(
|
||||
`Old path and new path are the same: ${oldPath}`
|
||||
);
|
||||
}
|
||||
|
||||
this.database.move(oldPath, relativePath);
|
||||
}
|
||||
}
|
||||
|
||||
let document =
|
||||
this.database.getLatestDocumentByRelativePath(relativePath);
|
||||
|
||||
if (
|
||||
oldPath !== undefined &&
|
||||
document?.metadata?.remoteRelativePath === relativePath
|
||||
) {
|
||||
this.logger.debug(
|
||||
`Document ${relativePath} has been moved as a result of a remote update, skipping sync`
|
||||
);
|
||||
return;
|
||||
}
|
||||
const document =
|
||||
this.database.getLatestDocumentByRelativePath(oldPath ?? relativePath);
|
||||
|
||||
// must have been removed after a successful delete
|
||||
if (document === undefined) {
|
||||
this.logger.debug(
|
||||
`Cannot find document ${relativePath} in the database, skipping`
|
||||
|
|
@ -212,27 +168,47 @@ export class Syncer {
|
|||
return;
|
||||
}
|
||||
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
const documentAtNewPath =
|
||||
this.database.getLatestDocumentByRelativePath(relativePath);
|
||||
|
||||
document = await this.database.getResolvedDocumentByRelativePath(
|
||||
relativePath,
|
||||
promise
|
||||
);
|
||||
if (oldPath !== undefined) {
|
||||
// We might have moved the document in the database before calling this method,
|
||||
// in that case, we mustn't move it again.
|
||||
if (
|
||||
documentAtNewPath === undefined ||
|
||||
documentAtNewPath.isDeleted
|
||||
) {
|
||||
if (oldPath === relativePath) {
|
||||
throw new Error(
|
||||
`Old path and new path are the same: ${oldPath}`
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncLocallyUpdatedFile({
|
||||
oldPath,
|
||||
document
|
||||
})
|
||||
);
|
||||
|
||||
resolve();
|
||||
} catch (e) {
|
||||
reject(e);
|
||||
} finally {
|
||||
this.database.removeDocumentPromise(promise);
|
||||
this.database.move(oldPath, relativePath);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
if (
|
||||
oldPath !== undefined &&
|
||||
document?.metadata?.remoteRelativePath === relativePath
|
||||
) {
|
||||
this.logger.debug(
|
||||
`Document ${relativePath} has been moved as a result of a remote update, skipping sync`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
await this.enqueueSyncOperation(
|
||||
async () =>
|
||||
this.unrestrictedSyncer.unrestrictedSyncLocallyCreatedOrUpdatedFile(
|
||||
{
|
||||
oldPath,
|
||||
document
|
||||
}
|
||||
),
|
||||
[document.metadata?.documentId, relativePath, oldPath]
|
||||
);
|
||||
}
|
||||
|
||||
public async scheduleSyncForOfflineChanges(): Promise<void> {
|
||||
|
|
@ -257,8 +233,6 @@ export class Syncer {
|
|||
`Not all local changes have been applied remotely: ${e}`
|
||||
);
|
||||
throw e;
|
||||
} finally {
|
||||
this.runningScheduleSyncForOfflineChanges = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -271,6 +245,8 @@ export class Syncer {
|
|||
message: WebSocketVaultUpdate
|
||||
): Promise<void> {
|
||||
try {
|
||||
await this.scheduleSyncForOfflineChanges();
|
||||
|
||||
const handlerPromise = awaitAll(
|
||||
message.documents.map(async (document) =>
|
||||
this.internalSyncRemotelyUpdatedFile(document)
|
||||
|
|
@ -296,7 +272,7 @@ export class Syncer {
|
|||
public reset(): void {
|
||||
this._isFirstSyncComplete = false;
|
||||
this.syncQueue.clear();
|
||||
this.remoteDocumentsLock.reset();
|
||||
this.updatedDocumentsByPathAndKeysLocks.reset();
|
||||
this.runningScheduleSyncForOfflineChanges = undefined;
|
||||
}
|
||||
|
||||
|
|
@ -313,86 +289,26 @@ export class Syncer {
|
|||
private async internalSyncRemotelyUpdatedFile(
|
||||
remoteVersion: DocumentVersionWithoutContent
|
||||
): Promise<void> {
|
||||
let document = this.database.getDocumentByDocumentId(
|
||||
const document = this.database.getDocumentByDocumentId(
|
||||
remoteVersion.documentId
|
||||
);
|
||||
|
||||
if (document === undefined) {
|
||||
// Let's avoid the same documents getting created in parallel multiple times.
|
||||
// There might be multiple tasks waiting for the lock
|
||||
return this.remoteDocumentsLock.withLock(
|
||||
remoteVersion.documentId,
|
||||
async () => {
|
||||
document = this.database.getDocumentByDocumentId(
|
||||
remoteVersion.documentId
|
||||
);
|
||||
|
||||
// We're either the first one to get the lock, so we have to create the document in `unrestrictedSyncRemotelyUpdatedFile`
|
||||
if (document === undefined) {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncRemotelyUpdatedFile(
|
||||
remoteVersion
|
||||
)
|
||||
);
|
||||
} else {
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
|
||||
document =
|
||||
await this.database.getResolvedDocumentByRelativePath(
|
||||
document.relativePath,
|
||||
promise
|
||||
);
|
||||
|
||||
try {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncRemotelyUpdatedFile(
|
||||
remoteVersion,
|
||||
document
|
||||
)
|
||||
);
|
||||
|
||||
resolve();
|
||||
} catch (e) {
|
||||
reject(e);
|
||||
} finally {
|
||||
this.database.removeDocumentPromise(promise);
|
||||
}
|
||||
}
|
||||
|
||||
this.database.addSeenUpdateId(remoteVersion.vaultUpdateId);
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
// We're either the first one to get the lock, so we have to create the document in `unrestrictedSyncRemotelyUpdatedFile`
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
|
||||
document = await this.database.getResolvedDocumentByRelativePath(
|
||||
document.relativePath,
|
||||
promise
|
||||
);
|
||||
|
||||
try {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncRemotelyUpdatedFile(
|
||||
await this.enqueueSyncOperation(
|
||||
async () =>
|
||||
this.unrestrictedSyncer.unrestrictedSyncRemotelyUpdatedFile(
|
||||
remoteVersion,
|
||||
document
|
||||
)
|
||||
);
|
||||
|
||||
resolve();
|
||||
} catch (e) {
|
||||
reject(e);
|
||||
} finally {
|
||||
this.database.removeDocumentPromise(promise);
|
||||
}
|
||||
),
|
||||
[
|
||||
document?.relativePath,
|
||||
remoteVersion.relativePath,
|
||||
remoteVersion.documentId
|
||||
]
|
||||
);
|
||||
|
||||
this.database.addSeenUpdateId(remoteVersion.vaultUpdateId);
|
||||
}
|
||||
|
||||
private async internalScheduleSyncForOfflineChanges(): Promise<void> {
|
||||
await this.createFakeDocumentsFromRemoteState();
|
||||
|
||||
const allLocalFiles = await this.operations.listFilesRecursively();
|
||||
this.logger.info(
|
||||
`Scheduling sync for ${allLocalFiles.length} local files`
|
||||
|
|
@ -409,7 +325,12 @@ export class Syncer {
|
|||
}
|
||||
}
|
||||
|
||||
await awaitAll(
|
||||
interface Instruction {
|
||||
type: "update" | "create";
|
||||
relativePath: string;
|
||||
oldPath?: string;
|
||||
}
|
||||
const instructions: (Instruction | undefined)[] = await awaitAll(
|
||||
allLocalFiles.map(async (relativePath) => {
|
||||
if (
|
||||
this.database.getLatestDocumentByRelativePath(relativePath)
|
||||
|
|
@ -419,16 +340,24 @@ export class Syncer {
|
|||
`Document ${relativePath} might have been updated locally, scheduling sync to validate and update it`
|
||||
);
|
||||
|
||||
return this.syncLocallyUpdatedFile({
|
||||
relativePath
|
||||
});
|
||||
return { type: "update", relativePath } as Instruction;
|
||||
}
|
||||
|
||||
// Perhaps the file has been moved; let's check by looking at the deleted files
|
||||
const contentHash = await this.syncQueue.add(async () => {
|
||||
const contentBytes =
|
||||
await this.operations.read(relativePath); // this can throw FileNotFoundError
|
||||
return hash(contentBytes);
|
||||
try {
|
||||
const contentBytes =
|
||||
await this.operations.read(relativePath); // this can throw FileNotFoundError
|
||||
return hash(contentBytes);
|
||||
} catch (e) {
|
||||
if (
|
||||
e instanceof Error &&
|
||||
e.name === "FileNotFoundError"
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
});
|
||||
|
||||
if (contentHash == undefined) {
|
||||
|
|
@ -454,18 +383,21 @@ export class Syncer {
|
|||
`Document '${originalFile.relativePath}' was not found under its current path in the database but was found under a different path (${relativePath}), scheduling sync to move it`
|
||||
);
|
||||
|
||||
// We're outside of the pqueue, so we need to call the public wrapper
|
||||
return this.syncLocallyUpdatedFile({
|
||||
return {
|
||||
type: "update",
|
||||
oldPath: originalFile.relativePath,
|
||||
relativePath
|
||||
});
|
||||
} as Instruction;
|
||||
}
|
||||
|
||||
this.logger.debug(
|
||||
`Document ${relativePath} not found in database, scheduling sync to create it`
|
||||
);
|
||||
// We're outside of the pqueue, so we need to call the public wrapper
|
||||
return this.syncLocallyCreatedFile(relativePath);
|
||||
|
||||
return {
|
||||
type: "create",
|
||||
relativePath
|
||||
} as Instruction;
|
||||
})
|
||||
);
|
||||
|
||||
|
|
@ -481,42 +413,49 @@ export class Syncer {
|
|||
return this.syncLocallyDeletedFile(relativePath);
|
||||
})
|
||||
);
|
||||
|
||||
await awaitAll(
|
||||
instructions.map(async (instruction) => {
|
||||
if (instruction === undefined) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (instruction.type === "update") {
|
||||
// We're outside of the pqueue, so we need to call the public wrapper
|
||||
await this.syncLocallyUpdatedFile({
|
||||
oldPath: instruction.oldPath,
|
||||
relativePath: instruction.relativePath
|
||||
});
|
||||
return;
|
||||
}
|
||||
})
|
||||
);
|
||||
|
||||
// we have to ensure the deletes & updates have finished before starting creates,
|
||||
// otherwise the server might return an existing document (that we're about to delete)
|
||||
// instead of actually creating a new one
|
||||
await awaitAll(
|
||||
instructions.map(async (instruction) => {
|
||||
if (instruction === undefined) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (instruction.type === "create") {
|
||||
// We're outside of the pqueue, so we need to call the public wrapper
|
||||
await this.syncLocallyCreatedFile(instruction.relativePath);
|
||||
return;
|
||||
}
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create fake documents in the database for all files that are present locally
|
||||
* and also exist remotely. This will stop the subequent syncs from duplicating
|
||||
* the documents by creating the same documents from multiple clients.
|
||||
*/
|
||||
private async createFakeDocumentsFromRemoteState(): Promise<void> {
|
||||
if (this.database.getHasInitialSyncCompleted()) {
|
||||
return;
|
||||
}
|
||||
|
||||
const [allLocalFiles, remote] = await awaitAll([
|
||||
this.operations.listFilesRecursively(),
|
||||
this.syncQueue.add(async () => this.syncService.getAll())
|
||||
]);
|
||||
|
||||
if (remote !== undefined) {
|
||||
remote.latestDocuments
|
||||
.filter(
|
||||
(remoteDocument) =>
|
||||
allLocalFiles.includes(remoteDocument.relativePath) &&
|
||||
!remoteDocument.isDeleted &&
|
||||
this.database.getDocumentByDocumentId(
|
||||
remoteDocument.documentId
|
||||
) === undefined
|
||||
)
|
||||
.forEach((remoteDocument) => {
|
||||
this.database.createNewEmptyDocument(
|
||||
remoteDocument.documentId,
|
||||
remoteDocument.vaultUpdateId,
|
||||
remoteDocument.relativePath
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
this.database.setHasInitialSyncCompleted(true);
|
||||
private async enqueueSyncOperation<T>(
|
||||
operation: () => Promise<T>,
|
||||
keys: (string | undefined | null)[]
|
||||
): Promise<T> {
|
||||
return this.updatedDocumentsByPathAndKeysLocks.withLock(
|
||||
keys.filter((k) => k !== undefined && k !== null),
|
||||
async () => this.syncQueue.add(operation)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue