This commit is contained in:
Andras Schmelczer 2025-03-02 10:25:38 +00:00
commit a2522ca44a
No known key found for this signature in database
GPG key ID: FC8F2C3D3D1A718C

View file

@ -1,19 +1,23 @@
import type { Database, RelativePath } from "../persistence/database";
import type {
Database,
DocumentMetadata,
RelativePath
} from "../persistence/database";
import type { SyncService } from "src/services/sync-service";
import type { Logger } from "src/tracing/logger";
import type { SyncHistory } from "src/tracing/sync-history";
import { SyncSource, SyncStatus, SyncType } from "src/tracing/sync-history";
import { hash } from "src/utils/hash";
import { EMPTY_HASH, hash } from "src/utils/hash";
import type { components } from "src/services/types";
import { deserialize } from "src/utils/deserialize";
import type { Settings } from "src/persistence/settings";
import type { FileOperations } from "src/file-operations/file-operations";
import { FileNotFoundError } from "src/file-operations/safe-filesystem-operations";
import { DocumentLocks } from "./document-locks";
import { DocumentLocks } from "../file-operations/document-locks";
export class UnrestrictedSyncer {
private readonly locks = new DocumentLocks();
private readonly locks: DocumentLocks;
public constructor(
private readonly logger: Logger,
@ -22,7 +26,9 @@ export class UnrestrictedSyncer {
private readonly syncService: SyncService,
private readonly operations: FileOperations,
private readonly history: SyncHistory
) {}
) {
this.locks = new DocumentLocks(logger);
}
public async unrestrictedSyncLocallyCreatedFile(
relativePath: RelativePath,
@ -31,96 +37,132 @@ export class UnrestrictedSyncer {
contentBytes?: Uint8Array;
contentHash?: string;
}
): Promise<void> {
await this.executeWhileHoldingFileLock(
): Promise<DocumentMetadata | undefined> {
return this.executeSync(
[relativePath],
SyncType.CREATE,
SyncSource.PUSH,
async () => {
const localMetadata = this.database.getDocument(relativePath);
if (
(await this.operations.getFileSize(relativePath)) / // this can throw FileNotFoundError
1024 /
1024 >
this.settings.getSettings().maxFileSizeMB
!(localMetadata instanceof Promise) &&
localMetadata &&
!localMetadata.isDeleted
) {
this.history.addHistoryEntry({
status: SyncStatus.ERROR,
relativePath,
message: `File size exceeds the maximum file size limit of ${
this.settings.getSettings().maxFileSizeMB
}MB`,
type: SyncType.CREATE
});
this.logger.debug(
`Document metadata already exists for ${relativePath}, it must have been downloaded from the server`
);
return;
}
const contentBytes =
optimisations?.contentBytes ??
(await this.operations.read(relativePath)); // this can throw FileNotFoundError
let contentHash =
const contentHash =
optimisations?.contentHash ?? hash(contentBytes);
const localMetadata = this.database.getDocument(relativePath);
if (localMetadata) {
this.logger.debug(
`Document metadata already exists for ${relativePath}, it must have been downloaded from the server`
);
if (localMetadata.hash === contentHash) {
this.history.addHistoryEntry({
status: SyncStatus.NO_OP,
relativePath,
message: `File hash matches with last synced version, no need to sync`,
type: SyncType.UPDATE
});
return;
}
}
const response = await this.syncService.create({
relativePath,
contentBytes,
createdDate: updateTime
});
const currentMetadata =
this.database.getDocumentByIdentity(localMetadata);
if (!currentMetadata) {
throw new Error(
`Document metadata for ${relativePath} not found after creation`
);
}
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PUSH,
relativePath,
relativePath: currentMetadata[0],
message: `Successfully uploaded locally created file`,
type: SyncType.CREATE
});
// The response can't have a different relative path than the one we sent
// because the relative path is the key when finding existing documents
// when a create request is sent.
if (response.type === "MergingUpdate") {
const responseBytes = deserialize(response.contentBase64);
contentHash = hash(responseBytes);
await this.operations.write(
relativePath,
contentBytes,
responseBytes
);
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PULL,
relativePath,
message: `The file we created locally has already existed remotely, so we have merged them`,
type: SyncType.UPDATE
});
}
const newMetadata = {
documentId: response.documentId,
parentVersionId: response.vaultUpdateId,
hash: contentHash,
isDeleted: false
};
await this.database.setDocument({
documentId: response.documentId,
relativePath: response.relativePath,
parentVersionId: response.vaultUpdateId,
hash: contentHash
relativePath: currentMetadata[0],
...newMetadata
});
await this.tryIncrementVaultUpdateId(response.vaultUpdateId);
return newMetadata;
}
);
}
public async unrestrictedSyncLocallyDeletedFile(
relativePath: RelativePath,
metadata: Promise<DocumentMetadata | undefined> | undefined
): Promise<void> {
await this.executeSync(
[relativePath],
SyncType.DELETE,
SyncSource.PUSH,
async () => {
const localMetadata =
metadata !== undefined
? await metadata
: this.database.getResolvedDocument(relativePath);
if (!localMetadata || localMetadata.isDeleted) {
this.logger.info(
`Locally deleted file hasn't been uploaded yet, so there's no need to delete it on the remote server`
);
return;
}
const response = await this.syncService.delete({
documentId: localMetadata.documentId,
relativePath,
createdDate: new Date() // We got the event now, so it must have been deleted just now
});
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PUSH,
relativePath,
message: `Successfully deleted locally deleted file on the remote server`,
type: SyncType.DELETE
});
const currentMetadata = this.database.getDocumentByDocumentId(
localMetadata.documentId
);
if (!currentMetadata || currentMetadata[1].isDeleted) {
this.logger.info(
`No metadata found for deleted file, '${relativePath}' must have been deleted by another operation`
);
return;
}
await this.operations.delete(currentMetadata[0]);
// We have to have a record of the delete in case there's an in-flight update for the same
// document which finishes after the delete has succeeded and would introduce a phantom metadata record.
await this.database.setDocument({
relativePath: currentMetadata[0],
documentId: response.documentId,
parentVersionId: response.vaultUpdateId,
hash: EMPTY_HASH,
isDeleted: true
});
}
);
}
@ -128,53 +170,34 @@ export class UnrestrictedSyncer {
public async unrestrictedSyncLocallyUpdatedFile({
oldPath,
relativePath,
metadata,
updateTime,
optimisations
}: {
oldPath?: RelativePath;
relativePath: RelativePath;
metadata: Promise<DocumentMetadata | undefined> | undefined;
updateTime: Date;
optimisations?: {
contentBytes?: Uint8Array;
contentHash?: string;
};
}): Promise<void> {
await this.executeWhileHoldingFileLock(
await this.executeSync(
[oldPath, relativePath].filter((path) => path !== undefined),
SyncType.UPDATE,
SyncSource.PUSH,
async () => {
// Check the new path first in case the metadata has been already moved
let localMetadata = this.database.getDocument(relativePath);
let metadataPath = relativePath;
const localMetadata =
metadata !== undefined
? await metadata
: this.database.getResolvedDocument(relativePath);
if (localMetadata === undefined && oldPath !== undefined) {
localMetadata = this.database.getDocument(oldPath);
metadataPath = oldPath;
}
if (!localMetadata) {
if (!localMetadata || localMetadata.isDeleted) {
// It's fine, a subsequent sync operation must have dealt with this
return;
}
if (
(await this.operations.getFileSize(relativePath)) / // this can throw FileNotFoundError
1024 /
1024 >
this.settings.getSettings().maxFileSizeMB
) {
this.history.addHistoryEntry({
status: SyncStatus.ERROR,
relativePath,
message: `File size exceeds the maximum file size limit of ${
this.settings.getSettings().maxFileSizeMB
}MB`,
type: SyncType.CREATE
});
return;
}
const contentBytes =
optimisations?.contentBytes ??
(await this.operations.read(relativePath)); // this can throw FileNotFoundError
@ -186,23 +209,48 @@ export class UnrestrictedSyncer {
localMetadata.hash === contentHash &&
oldPath === undefined
) {
this.history.addHistoryEntry({
status: SyncStatus.NO_OP,
relativePath,
message: `File hash matches with last synced version, no need to sync`,
type: SyncType.UPDATE
});
this.logger.debug(
`File hash of ${relativePath} matches with last synced version and the path hasn't changed; no need to sync`
);
return;
}
// Re-fetch based on the documentId instead of the relativePath because
// the relativePath might have changed since this operation was scheduled
let latestMetadata = this.database.getDocumentByDocumentId(
localMetadata.documentId
);
if (!latestMetadata || latestMetadata[1].isDeleted) {
// It's fine, a subsequent sync operation must have dealt with this
return;
}
const response = await this.syncService.put({
documentId: localMetadata.documentId,
parentVersionId: localMetadata.parentVersionId,
relativePath,
documentId: latestMetadata[1].documentId,
parentVersionId: latestMetadata[1].parentVersionId,
relativePath: latestMetadata[0],
contentBytes,
createdDate: updateTime
});
latestMetadata = this.database.getDocumentByDocumentId(
response.documentId
);
if (!latestMetadata || latestMetadata[1].isDeleted) {
// The document has been deleted since this operation was scheduled
return;
}
if (
latestMetadata[1].parentVersionId >= response.vaultUpdateId
) {
this.logger.debug(
`Document ${relativePath} is already more up to date than the fetched version`
);
return;
}
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PUSH,
@ -212,11 +260,7 @@ export class UnrestrictedSyncer {
});
if (response.isDeleted) {
await this.operations.remove(oldPath ?? relativePath);
await this.database.removeDocument(oldPath ?? relativePath);
await this.tryIncrementVaultUpdateId(
response.vaultUpdateId
);
await this.operations.delete(relativePath);
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
@ -227,110 +271,69 @@ export class UnrestrictedSyncer {
type: SyncType.DELETE
});
return;
}
if (
response.relativePath != relativePath &&
response.relativePath != oldPath
) {
await this.locks.waitForDocumentLock(response.relativePath);
}
try {
if (response.relativePath != relativePath) {
// TODO: this can fail, that's bad
await this.operations.move(
// this can throw FileNotFoundError
relativePath,
response.relativePath,
response.documentId
);
}
if (response.type === "MergingUpdate") {
const responseBytes = deserialize(
response.contentBase64
);
contentHash = hash(responseBytes);
await this.operations.write(
response.relativePath,
contentBytes,
responseBytes
);
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PULL,
relativePath,
message: `The file we updated had been updated remotely, so we downloaded the merged version`,
type: SyncType.UPDATE
});
}
if (metadataPath !== response.relativePath) {
await this.database.updatePath(
metadataPath,
response.relativePath
);
}
await this.database.setDocument({
documentId: localMetadata.documentId,
relativePath: response.relativePath,
documentId: response.documentId,
relativePath: latestMetadata[0],
parentVersionId: response.vaultUpdateId,
hash: contentHash
hash: EMPTY_HASH,
isDeleted: true
});
await this.tryIncrementVaultUpdateId(
response.vaultUpdateId
);
} finally {
if (
response.relativePath != relativePath &&
response.relativePath != oldPath
) {
this.locks.unlockDocument(response.relativePath);
}
}
}
);
}
public async unrestrictedSyncLocallyDeletedFile(
relativePath: RelativePath
): Promise<void> {
await this.executeWhileHoldingFileLock(
[relativePath],
SyncType.DELETE,
SyncSource.PUSH,
async () => {
const localMetadata = this.database.getDocument(relativePath);
if (!localMetadata) {
this.history.addHistoryEntry({
status: SyncStatus.NO_OP,
relativePath,
message: `Locally deleted file hasn't been uploaded yet, so there's no need to delete it on the remote server`,
type: SyncType.DELETE
});
return;
}
await this.syncService.delete({
documentId: localMetadata.documentId,
relativePath,
createdDate: new Date() // We got the event now, so it must have been deleted just now
if (
latestMetadata[1].parentVersionId >= response.vaultUpdateId
) {
this.logger.debug(
`Document ${relativePath} is already more up to date than the fetched version`
);
return;
}
if (response.relativePath != relativePath) {
await this.operations.move(
latestMetadata[0],
response.relativePath,
response.documentId
); // this can throw FileNotFoundError
}
if (response.type === "MergingUpdate") {
const responseBytes = deserialize(response.contentBase64);
contentHash = hash(responseBytes);
await this.operations.write(
response.relativePath,
contentBytes,
responseBytes
);
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PULL,
relativePath,
message: `The file we updated had been updated remotely, so we downloaded the merged version`,
type: SyncType.UPDATE
});
}
await this.database.setDocument({
documentId: response.documentId,
relativePath:
response.relativePath != relativePath
? response.relativePath
: latestMetadata[0],
parentVersionId: response.vaultUpdateId,
hash: contentHash,
isDeleted: response.isDeleted
});
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PUSH,
relativePath,
message: `Successfully deleted locally deleted file on the remote server`,
type: SyncType.DELETE
});
await this.database.removeDocument(relativePath);
await this.tryIncrementVaultUpdateId(response.vaultUpdateId);
}
);
}
@ -338,56 +341,68 @@ export class UnrestrictedSyncer {
public async unrestrictedSyncRemotelyUpdatedFile(
remoteVersion: components["schemas"]["DocumentVersionWithoutContent"]
): Promise<void> {
await this.executeWhileHoldingFileLock(
await this.executeSync(
[remoteVersion.relativePath],
SyncType.UPDATE,
SyncSource.PULL,
async () => {
let localMetadata = this.database.getDocumentByDocumentId(
const content = (
await this.syncService.get({
documentId: remoteVersion.documentId
})
).contentBase64;
const contentBytes = deserialize(content);
const contentHash = hash(contentBytes);
const localMetadata = this.database.getDocumentByDocumentId(
remoteVersion.documentId
);
if (
localMetadata &&
localMetadata[0] !== remoteVersion.relativePath
localMetadata?.[1].documentId ===
remoteVersion.documentId &&
localMetadata[1].parentVersionId >
remoteVersion.vaultUpdateId
) {
await this.locks.waitForDocumentLock(localMetadata[0]);
this.logger.info(
`Document ${remoteVersion.relativePath} is already up to date`
);
return;
}
// Waiting for the new lock might take a while so we need to fetch the database
// entry again in case it's changed.
localMetadata = this.database.getDocumentByDocumentId(
remoteVersion.documentId
);
if (!localMetadata) {
const localBytes = await this.operations.read(
remoteVersion.relativePath
); // this can throw FileNotFoundError
const localHash = hash(localBytes);
if (localHash !== localMetadata?.[1].hash) {
this.logger.info(
`Document ${remoteVersion.relativePath} has pending local changes, so we shouldn't update it here`
);
return;
}
if (!localMetadata || localMetadata[1].isDeleted) {
if (remoteVersion.isDeleted) {
this.history.addHistoryEntry({
status: SyncStatus.NO_OP,
source: SyncSource.PULL,
relativePath: remoteVersion.relativePath,
message: `Remotely deleted file hasn't been synced yet, so there's no need to delete it locally`,
type: SyncType.DELETE
});
this.logger.info(
`Remotely deleted file hasn't been synced yet, so there's no need to delete it locally`
);
return;
}
const content = (
await this.syncService.get({
documentId: remoteVersion.documentId
})
).contentBase64;
const contentBytes = deserialize(content);
await this.operations.create(
remoteVersion.relativePath,
contentBytes
contentBytes,
remoteVersion.documentId
);
await this.database.setDocument({
documentId: remoteVersion.documentId,
relativePath: remoteVersion.relativePath,
parentVersionId: remoteVersion.vaultUpdateId,
hash: hash(contentBytes)
hash: hash(contentBytes),
isDeleted: remoteVersion.isDeleted
});
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PULL,
@ -399,7 +414,6 @@ export class UnrestrictedSyncer {
}
const [relativePath, metadata] = localMetadata;
if (remoteVersion.vaultUpdateId <= metadata.parentVersionId) {
this.logger.debug(
`Document ${relativePath} is already up to date`
@ -407,89 +421,70 @@ export class UnrestrictedSyncer {
return;
}
try {
if (remoteVersion.isDeleted) {
await this.operations.remove(relativePath);
await this.database.removeDocument(relativePath);
if (remoteVersion.isDeleted) {
await this.operations.delete(relativePath);
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PULL,
relativePath: remoteVersion.relativePath,
message: `Successfully deleted remotely deleted file locally`,
type: SyncType.DELETE
});
} else {
// TODO: this can fail, that's bad
const currentContent =
await this.operations.read(relativePath); // this can throw FileNotFoundError
const currentHash = hash(currentContent);
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PULL,
relativePath: remoteVersion.relativePath,
message: `Successfully deleted remotely deleted file locally`,
type: SyncType.DELETE
});
if (currentHash !== metadata.hash) {
this.logger.info(
`Document ${relativePath} has been updated both remotely and locally, letting the local file update event handle it`
);
return;
}
await this.database.setDocument({
documentId: remoteVersion.documentId,
relativePath: relativePath,
parentVersionId: remoteVersion.vaultUpdateId,
hash: EMPTY_HASH,
isDeleted: true
});
const content = (
await this.syncService.get({
documentId: remoteVersion.documentId
})
).contentBase64;
const contentBytes = deserialize(content);
const contentHash = hash(contentBytes);
if (relativePath !== remoteVersion.relativePath) {
// TODO: this can fail, that's bad
await this.operations.move(
// this can throw FileNotFoundError
relativePath,
remoteVersion.relativePath,
remoteVersion.documentId
);
await this.database.updatePath(
relativePath,
remoteVersion.relativePath
);
}
await this.operations.write(
remoteVersion.relativePath,
currentContent,
contentBytes
);
await this.database.setDocument({
documentId: remoteVersion.documentId,
relativePath: remoteVersion.relativePath,
parentVersionId: remoteVersion.vaultUpdateId,
hash: contentHash
});
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PULL,
relativePath: remoteVersion.relativePath,
message: `Successfully updated remotely updated file locally`,
type: SyncType.UPDATE
});
}
} finally {
if (relativePath !== remoteVersion.relativePath) {
this.locks.unlockDocument(relativePath);
}
return;
}
if (relativePath !== remoteVersion.relativePath) {
// TODO: this can fail, that's bad
await this.operations.move(
// this can throw FileNotFoundError
relativePath,
remoteVersion.relativePath,
remoteVersion.documentId
);
}
// todo: why
await this.operations.create(
remoteVersion.relativePath,
contentBytes,
remoteVersion.documentId
);
await this.database.setDocument({
documentId: remoteVersion.documentId,
relativePath: remoteVersion.relativePath,
parentVersionId: remoteVersion.vaultUpdateId,
hash: contentHash,
isDeleted: remoteVersion.isDeleted
});
this.history.addHistoryEntry({
status: SyncStatus.SUCCESS,
source: SyncSource.PULL,
relativePath: remoteVersion.relativePath,
message: `Successfully updated remotely updated file locally`,
type: SyncType.UPDATE
});
}
);
}
public async executeWhileHoldingFileLock(
public async executeSync<T>(
lockedPaths: RelativePath[],
syncType: SyncType,
syncSource: SyncSource,
fn: () => Promise<void>
): Promise<void> {
fn: () => Promise<T>
): Promise<T | undefined> {
const relativePath = lockedPaths[lockedPaths.length - 1];
if (!this.settings.getSettings().isSyncEnabled) {
@ -498,31 +493,47 @@ export class UnrestrictedSyncer {
);
return;
}
if (!this.operations.isFileEligibleForSync(relativePath)) {
this.logger.info(
`File ${relativePath} is not eligible for syncing`
);
this.history.addHistoryEntry({
status: SyncStatus.ERROR,
relativePath,
message: `File ${relativePath} is not eligible for syncing`,
type: syncType
});
return;
}
this.logger.debug(
`Syncing ${relativePath} (${syncSource} - ${syncType})`
);
await Promise.all(
lockedPaths.map(this.locks.waitForDocumentLock.bind(this.locks))
);
try {
await fn();
if (
(await this.operations.exists(relativePath)) &&
(await this.operations.getFileSize(relativePath)) / // this can throw FileNotFoundError
1024 /
1024 >
this.settings.getSettings().maxFileSizeMB
) {
this.history.addHistoryEntry({
status: SyncStatus.ERROR,
relativePath,
message: `File size exceeds the maximum file size limit of ${
this.settings.getSettings().maxFileSizeMB
}MB`,
type: syncType
});
return;
}
return await fn();
} catch (e) {
if (e instanceof FileNotFoundError) {
// A subsequent sync operation must have been creating to deal with this
this.history.addHistoryEntry({
status: SyncStatus.NO_OP,
relativePath,
message: `Skip ${syncSource.toLocaleLowerCase()} file because it no longer exists when trying to ${syncType.toLocaleLowerCase()} it`,
type: syncType,
source: syncSource
});
this.logger.info(
`Skip ${syncSource.toLocaleLowerCase()} file because it no longer exists when trying to ${syncType.toLocaleLowerCase()} it`
);
} else {
this.history.addHistoryEntry({
status: SyncStatus.ERROR,
@ -533,8 +544,6 @@ export class UnrestrictedSyncer {
});
throw e;
}
} finally {
lockedPaths.forEach(this.locks.unlockDocument.bind(this.locks));
}
}