public/obsidian-livesync
1
0
Fork 0
mirror of https://github.com/vrtmrz/obsidian-livesync.git synced 2026-10-09 02:42:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify remote feature checks and clarify manual rebuild guidance

Browse Source
This commit is contained in:
vorotamoroz
2026-09-27 11:08:53 +00:00
parent 9a047d44f3
commit 2c6cd56618
15 changed files with 251 additions and 684 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/features/SettingDialogue/PaneRemoteConfig.ts
+6 -28
View File
@@ -5,7 +5,6 @@ import {
DEFAULT_SETTINGS,
LOG_LEVEL_NOTICE,
type ObsidianLiveSyncSettings,
type EncryptionSettings,
LOG_LEVEL_VERBOSE,
} from "@vrtmrz/livesync-commonlib/compat/common/types";
import { Menu, type ButtonComponent } from "@/deps.ts";
@@ -32,13 +31,11 @@ import {
import { ConnectionStringParser } from "@vrtmrz/livesync-commonlib/compat/common/ConnectionString";
import type { RemoteConfigurationResult } from "@vrtmrz/livesync-commonlib/compat/common/ConnectionString";
import SetupRemote from "@/modules/features/SetupWizard/dialogs/SetupRemote.svelte";
import SetupRemoteE2EE from "@/modules/features/SetupWizard/dialogs/SetupRemoteE2EE.svelte";
import SetupRemoteCouchDB from "@/modules/features/SetupWizard/dialogs/SetupRemoteCouchDB.svelte";
import SetupRemoteBucket from "@/modules/features/SetupWizard/dialogs/SetupRemoteBucket.svelte";
import type {
SetupRemoteCouchDBInitialData,
SetupRemoteCouchDBResultType,
SetupRemoteE2EEResultType,
} from "@/modules/features/SetupWizard/dialogs/setupDialogTypes.ts";
import { syncActivatedRemoteSettings } from "./remoteConfigBuffer.ts";
@@ -119,34 +116,15 @@ export function paneRemoteConfig(
.onClick(async () => {
const setupManager = this.core.getModule(SetupManager);
const originalSettings = getSettingsFromEditingSettings(this.editingSettings);
const e2eeConf = await setupManager.dialogManager.openWithExplicitCancel<
SetupRemoteE2EEResultType,
EncryptionSettings
>(SetupRemoteE2EE, originalSettings);
if (e2eeConf === "cancelled") {
return;
}
const onlyInternalMetadataPreferenceChanged =
originalSettings.encryptInternalMetadata !== e2eeConf.encryptInternalMetadata &&
originalSettings.encrypt === e2eeConf.encrypt &&
originalSettings.passphrase === e2eeConf.passphrase &&
originalSettings.E2EEAlgorithm === e2eeConf.E2EEAlgorithm &&
originalSettings.usePathObfuscation === e2eeConf.usePathObfuscation;
if (onlyInternalMetadataPreferenceChanged) {
await this.services.setting.applyPartial(
{ encryptInternalMetadata: e2eeConf.encryptInternalMetadata },
true
);
this.editingSettings.encryptInternalMetadata = e2eeConf.encryptInternalMetadata;
const applied = await setupManager.onlyE2EEConfiguration(UserMode.Update, originalSettings);
if (applied) {
this.editingSettings.encryptInternalMetadata =
this.core.settings.encryptInternalMetadata;
if (this.initialSettings) {
this.initialSettings.encryptInternalMetadata = e2eeConf.encryptInternalMetadata;
this.initialSettings.encryptInternalMetadata =
this.core.settings.encryptInternalMetadata;
}
this.requestUpdate();
} else {
await setupManager.onConfirmApplySettingsFromWizard(
{ ...originalSettings, ...e2eeConf },
UserMode.Update
);
}
updateE2EESummary();
})
src/modules/features/SettingDialogue/PaneRemoteConfig.unit.spec.ts
+5 -15
View File
@@ -162,24 +162,15 @@ describe("paneRemoteConfig", () => {
encryptInternalMetadata: false,
remoteConfigurations: {},
};
const applyPartial = vi.fn(async () => {});
const onConfirmApplySettingsFromWizard = vi.fn(async () => {});
const setupManager = {
dialogManager: {
openWithExplicitCancel: vi.fn(async () => ({
encrypt: true,
passphrase: "passphrase",
E2EEAlgorithm: "v2",
usePathObfuscation: true,
encryptInternalMetadata: true,
})),
},
onConfirmApplySettingsFromWizard,
onlyE2EEConfiguration: vi.fn(async () => {
host.core.settings.encryptInternalMetadata = true;
return true;
}),
};
const host = {
editingSettings: { ...originalSettings },
initialSettings: { ...originalSettings },
services: { setting: { applyPartial } },
core: {
settings: { ...originalSettings },
getModule: vi.fn(() => setupManager),
@@ -198,8 +189,7 @@ describe("paneRemoteConfig", () => {
paneRemoteConfig.call(host as never, {} as HTMLElement, { addPanel } as never);
await runtime.clickHandlers[0]();
expect(applyPartial).toHaveBeenCalledWith({ encryptInternalMetadata: true }, true);
expect(onConfirmApplySettingsFromWizard).not.toHaveBeenCalled();
expect(setupManager.onlyE2EEConfiguration).toHaveBeenCalledOnce();
expect(host.editingSettings.encryptInternalMetadata).toBe(true);
expect(host.initialSettings.encryptInternalMetadata).toBe(true);
expect(host.requestUpdate).toHaveBeenCalledOnce();
src/modules/features/SetupManager.ts
+25
View File
@@ -336,6 +336,31 @@ export class SetupManager extends AbstractModule {
this._log("E2EE configuration cancelled.", LOG_LEVEL_NOTICE);
return false;
}
const onlyInternalMetadataPreferenceChanged =
currentSetting.encryptInternalMetadata !== e2eeConf.encryptInternalMetadata &&
currentSetting.encrypt === e2eeConf.encrypt &&
currentSetting.passphrase === e2eeConf.passphrase &&
currentSetting.E2EEAlgorithm === e2eeConf.E2EEAlgorithm &&
currentSetting.usePathObfuscation === e2eeConf.usePathObfuscation;
if (userMode === UserMode.Update && onlyInternalMetadataPreferenceChanged) {
if (e2eeConf.encryptInternalMetadata && currentSetting.remoteType === REMOTE_COUCHDB) {
const proceed = "Enable without rebuilding — update every other device first";
const choice = await this.core.confirm.askSelectStringDialogue(
"A manual remote Rebuild is strongly recommended to protect existing Metadata. " +
"Before continuing without rebuilding, update every other synchronising device to a version " +
"which supports this option, including devices currently running LiveSync. " +
"Existing Metadata remains unchanged until it is rewritten or rebuilt.",
[proceed, "Cancel"],
{ title: "Encrypt internal file Metadata", defaultAction: "Cancel" }
);
if (choice !== proceed) return false;
}
await this.services.setting.applyPartial(
{ encryptInternalMetadata: e2eeConf.encryptInternalMetadata },
true
);
return true;
}
const newSetting = {
...currentSetting,
...e2eeConf,
src/modules/features/SetupManager.unit.spec.ts
+20
View File
@@ -659,3 +659,23 @@ describe("SetupManager", () => {
expect(setting.currentSettings().P2P_ActiveRemoteConfigurationId).toBe("existing");
});
});
describe("internal Metadata configuration", () => {
it.each([true, false])(
"applies the preference only after accepting the no-Rebuild warning (%s)",
async (accept) => {
const { manager, setting, dialogManager, core } = createSetupManager();
const current = { ...setting.settings, encryptInternalMetadata: false, remoteType: REMOTE_COUCHDB };
dialogManager.openWithExplicitCancel.mockResolvedValue({ ...current, encryptInternalMetadata: true });
const ask = vi.fn(async (_message: string, choices: string[]) => (accept ? choices[0] : "Cancel"));
core.confirm = { askSelectStringDialogue: ask };
const apply = vi.spyOn(setting, "applyPartial").mockResolvedValue(undefined);
await expect(manager.onlyE2EEConfiguration(UserMode.Update, current)).resolves.toBe(accept);
expect(ask.mock.calls[0][1][0]).toContain("update every other device first");
expect(ask.mock.calls[0][0]).toContain("currently running LiveSync");
expect(apply).toHaveBeenCalledTimes(accept ? 1 : 0);
expect(core.rebuilder.scheduleRebuild).not.toHaveBeenCalled();
expect(core.rebuilder.scheduleFetch).not.toHaveBeenCalled();
}
);
});
src/modules/features/SetupWizard/dialogs/SetupRemoteE2EE.svelte
+2 -1
View File
@@ -105,7 +105,8 @@
(Obfuscate Properties). The remote type is selected later in this setup wizard.
<br />
It protects Metadata written after the option is enabled; existing Metadata is not rewritten. A manual remote
Rebuild is strongly recommended to protect existing Metadata.
Rebuild is strongly recommended to protect existing Metadata. Update every other synchronising device to a compatible
version before enabling this option, including devices currently running LiveSync.
</InfoNote>
<ExtraItems title={translateMessage("Advanced")}>
src/serviceFeatures/replication/ReplicateResultProcessor.ts
+33 -222
View File
@@ -1,7 +1,7 @@
import { assessRemoteFeatureDocument, describeRemoteFeatureRejection } from "@vrtmrz/livesync-commonlib/replication";
import {
SYNCINFO_ID,
VERSIONING_DOCID,
type EntryVersionInfo,
type AnyEntry,
type EntryDoc,
type EntryLeaf,
@@ -9,11 +9,6 @@ import {
type MetaEntry,
type ObsidianLiveSyncSettings,
} from "@vrtmrz/livesync-commonlib/compat/common/types";
import {
assessRemoteFeatureDocument,
describeRemoteFeatureRejection,
REMOTE_FEATURE_GENERATION,
} from "@vrtmrz/livesync-commonlib/replication";
import { isChunk } from "@vrtmrz/livesync-commonlib/compat/common/typeUtils";
import {
LOG_LEVEL_DEBUG,
@@ -69,10 +64,6 @@ interface ReplicateResultProcessorContext {
readonly services: ReplicateResultProcessorServices;
}
type ReplicateResultProcessorState = {
databaseId?: string;
observedFeatures?: string[];
highestObservedVersion?: number;
invalidControlObserved?: boolean;
queued: PouchDB.Core.ExistingDocument<EntryDoc>[];
processing: PouchDB.Core.ExistingDocument<EntryDoc>[];
};
@@ -83,10 +74,6 @@ function shortenRev(rev: string | undefined): string {
if (!rev) return "undefined";
return rev.length > 10 ? rev.substring(0, 10) : rev;
}
function getPhysicalDatabaseId(database: PouchDB.Database<EntryDoc>): Promise<string | undefined> {
const identified = database as PouchDB.Database<EntryDoc> & { id?: () => Promise<string> };
return typeof identified.id === "function" ? identified.id() : Promise.resolve(undefined);
}
export class ReplicateResultProcessor {
private log(message: string, level: LOG_LEVEL = LOG_LEVEL_INFO) {
Logger(`[ReplicateResultProcessor] ${message}`, level);
@@ -129,89 +116,6 @@ export class ReplicateResultProcessor {
// If true, the processing queue processor bails the loop.
private _suspended: boolean = false;
// A temporary lifecycle resume cannot make an unknown remote format safe to apply.
private _compatibilityBlocked = false;
private _assessingDatabase = false;
private _physicalDatabase?: PouchDB.Database<EntryDoc>;
private _observedFeatures = new Set<string>();
private _highestObservedVersion = 0;
private _invalidControlObserved = false;
public get isCompatibilityBlocked() {
return this._compatibilityBlocked;
}
private refreshPhysicalDatabase() {
const current = this.localDatabase.localDatabase;
if (current === this._physicalDatabase) return;
if (this._physicalDatabase) {
this._compatibilityBlocked = false;
this._assessingDatabase = true;
this._observedFeatures.clear();
this._highestObservedVersion = 0;
this._invalidControlObserved = false;
this._queuedChanges = [];
this._processingChanges = [];
this._restoreFromSnapshot = undefined;
this.updateProcessingActivity();
}
this._physicalDatabase = current;
}
private shouldStopApplication(sourceDatabase: PouchDB.Database<EntryDoc>) {
return (
this._compatibilityBlocked || this._assessingDatabase || sourceDatabase !== this.localDatabase.localDatabase
);
}
private blockForIncompatibleVersion(document: unknown, recordObservation = true) {
const assessment = assessRemoteFeatureDocument(document);
const hadObservedVersion = this._highestObservedVersion > 0;
let newFeaturesAdded = false;
if (recordObservation) {
let changed = false;
if (assessment.status === "invalid-control") {
changed = !this._invalidControlObserved;
this._invalidControlObserved = true;
} else {
const version = (document as EntryVersionInfo).version;
if (version > this._highestObservedVersion) {
this._highestObservedVersion = version;
changed = true;
}
if (assessment.status === "supported" || assessment.status === "unknown-features") {
for (const feature of assessment.status === "supported"
? assessment.usedFeatures
: ((document as EntryVersionInfo).used_features ?? [])) {
if (this._observedFeatures.has(feature)) continue;
this._observedFeatures.add(feature);
changed = true;
newFeaturesAdded = true;
}
}
}
if (changed) this.triggerTakeSnapshot();
}
if (assessment.status === "supported" || assessment.status === "older-generation") {
// A live writer must recheck the shared Tweak policy after another
// client starts using a newly declared representation.
if (
assessment.status === "supported" &&
hadObservedVersion &&
newFeaturesAdded &&
!this._assessingDatabase
) {
this.context.requestActiveReplicatorRetirement();
}
return;
}
if (this._compatibilityBlocked) return;
this._compatibilityBlocked = true;
this.updateProcessingActivity();
this.log(describeRemoteFeatureRejection(assessment), LOG_LEVEL_NOTICE);
this.context.requestActiveReplicatorRetirement();
}
/**
* Whether the application accepts replicated documents being applied.
*
@@ -232,8 +136,6 @@ export class ReplicateResultProcessor {
public get isSuspended() {
return (
this._suspended ||
this._compatibilityBlocked ||
this._assessingDatabase ||
!this.acceptsResultApplication ||
this.context.currentSettings().suspendParseReplicationResult ||
this.services.appLifecycle.isSuspended()
@@ -244,38 +146,17 @@ export class ReplicateResultProcessor {
* Take a snapshot of the current processing state.
* This snapshot is stored in the KV database for recovery on restart.
*/
private _snapshotWriter: Promise<void> = Promise.resolve();
protected _takeSnapshot(): Promise<void> {
// A blocked-batch flush must follow any earlier throttled write, or an
// older snapshot could replace the queue after the replication callback.
const write = this._snapshotWriter
.catch((): void => undefined)
.then(async () => {
const physicalDatabase = this.localDatabase.localDatabase;
const databaseId = await getPhysicalDatabaseId(physicalDatabase);
if (physicalDatabase !== this.localDatabase.localDatabase) return;
const snapshot = {
...(databaseId ? { databaseId } : {}),
observedFeatures: [...this._observedFeatures],
highestObservedVersion: this._highestObservedVersion,
invalidControlObserved: this._invalidControlObserved,
queued: this._queuedChanges.slice(),
processing: this._processingChanges.slice(),
} satisfies ReplicateResultProcessorState;
await this.context.getKeyValueDB().set(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT, snapshot);
this.log(
`Snapshot taken. Queued: ${snapshot.queued.length}, Processing: ${snapshot.processing.length}`,
LOG_LEVEL_DEBUG
);
this.reportStatus();
});
this._snapshotWriter = write;
return write;
}
public async persistBlockedSnapshot(): Promise<void> {
if (this._compatibilityBlocked) await this._takeSnapshot();
protected async _takeSnapshot() {
const snapshot = {
queued: this._queuedChanges.slice(),
processing: this._processingChanges.slice(),
} satisfies ReplicateResultProcessorState;
await this.context.getKeyValueDB().set(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT, snapshot);
this.log(
`Snapshot taken. Queued: ${snapshot.queued.length}, Processing: ${snapshot.processing.length}`,
LOG_LEVEL_DEBUG
);
this.reportStatus();
}
/**
* Trigger taking a snapshot.
@@ -292,42 +173,10 @@ export class ReplicateResultProcessor {
* Restore from snapshot.
*/
public async restoreFromSnapshot() {
const physicalDatabase = this.localDatabase.localDatabase;
// Replication may have checkpointed a version document before its accompanying
// file changes reached the Vault. Assess the persisted requirement first.
let versionInfo: unknown;
try {
versionInfo = await this.localDatabase.getRaw(VERSIONING_DOCID);
} catch (error) {
if (!isNotFoundError(error)) throw error;
}
if (physicalDatabase !== this.localDatabase.localDatabase) return;
const snapshot = await this.context
.getKeyValueDB()
.get<ReplicateResultProcessorState>(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT);
if (physicalDatabase !== this.localDatabase.localDatabase) return;
const databaseId = await getPhysicalDatabaseId(physicalDatabase);
if (snapshot && (!snapshot.databaseId || !databaseId || snapshot.databaseId === databaseId)) {
for (const feature of snapshot.observedFeatures ?? []) this._observedFeatures.add(feature);
this._highestObservedVersion = Math.max(this._highestObservedVersion, snapshot.highestObservedVersion ?? 0);
this._invalidControlObserved ||= snapshot.invalidControlObserved === true;
}
if (versionInfo !== undefined) this.blockForIncompatibleVersion(versionInfo);
if (this._invalidControlObserved) this.blockForIncompatibleVersion(null, false);
if (this._highestObservedVersion > 0) {
this.blockForIncompatibleVersion(
{
_id: VERSIONING_DOCID,
type: "versioninfo",
version: this._highestObservedVersion,
...(this._highestObservedVersion >= REMOTE_FEATURE_GENERATION
? { used_features: [...this._observedFeatures] }
: {}),
},
false
);
}
if (snapshot && (!snapshot.databaseId || !databaseId || snapshot.databaseId === databaseId)) {
if (snapshot) {
// Restoring the snapshot re-runs processing for both queued and processing items.
const newQueue = [...snapshot.processing, ...snapshot.queued, ...this._queuedChanges];
this._queuedChanges = [];
@@ -338,9 +187,6 @@ export class ReplicateResultProcessor {
);
// await this._takeSnapshot();
}
this._assessingDatabase = false;
this.updateProcessingActivity();
this.triggerProcessQueue();
}
private _restoreFromSnapshot: Promise<void> | undefined = undefined;
@@ -350,9 +196,7 @@ export class ReplicateResultProcessor {
* @returns Promise that resolves when restoration is complete.
*/
public restoreFromSnapshotOnce() {
this.refreshPhysicalDatabase();
if (!this._restoreFromSnapshot) {
this._assessingDatabase = true;
this._restoreFromSnapshot = this.restoreFromSnapshot();
}
return this._restoreFromSnapshot;
@@ -386,17 +230,7 @@ export class ReplicateResultProcessor {
* @param changes Changes to enqueue
*/
public enqueueAll(changes: PouchDB.Core.ExistingDocument<EntryDoc>[], sourceDatabase?: PouchDB.Database<EntryDoc>) {
if (sourceDatabase && sourceDatabase !== this.localDatabase.localDatabase) return;
const previousPhysicalDatabase = this._physicalDatabase;
this.refreshPhysicalDatabase();
if (previousPhysicalDatabase && previousPhysicalDatabase !== this._physicalDatabase) {
fireAndForget(() => this.restoreFromSnapshotOnce());
}
// Inspect every control document before a note in this batch can start applying.
for (const change of changes) {
if (change?._id === VERSIONING_DOCID) this.blockForIncompatibleVersion(change);
}
public enqueueAll(changes: PouchDB.Core.ExistingDocument<EntryDoc>[]) {
for (const change of changes) {
// Check if the change is not a document change (e.g., chunk, versioninfo, syncinfo), and processed it directly.
const isProcessed = this.processIfNonDocumentChange(change);
@@ -441,8 +275,17 @@ export class ReplicateResultProcessor {
this.log(`Processed chunk: ${shortenId(change._id)}`, LOG_LEVEL_DEBUG);
return true;
}
if (change._id === VERSIONING_DOCID) {
if (change._id === VERSIONING_DOCID || change.type === "versioninfo") {
this.log(`Version info document received: ${change._id}`, LOG_LEVEL_VERBOSE);
const assessment = assessRemoteFeatureDocument(change);
if (assessment.status !== "supported" && assessment.status !== "older-generation") {
// Fence and retire the active publication through its owner.
this.context.requestActiveReplicatorRetirement();
this.log(
`${describeRemoteFeatureRejection(assessment)} Update Self-hosted LiveSync before synchronising.`,
LOG_LEVEL_NOTICE
);
}
return true;
}
if (
@@ -568,12 +411,11 @@ export class ReplicateResultProcessor {
// (per-document serialisation caps concurrency).
const releaser = await this._semaphore.acquire();
releaser();
if (this.isSuspended) break;
// Dequeue the next change
const doc = this._queuedChanges.shift();
if (doc) {
this._processingChanges.push(doc);
void this.parseDocumentChange(doc, this.localDatabase.localDatabase);
void this.parseDocumentChange(doc);
}
// Take snapshot (to be restored on next startup if needed)
this.triggerTakeSnapshot();
@@ -589,12 +431,8 @@ export class ReplicateResultProcessor {
* @param change
* @returns
*/
async parseDocumentChange(
change: PouchDB.Core.ExistingDocument<EntryDoc>,
sourceDatabase: PouchDB.Database<EntryDoc> = this.localDatabase.localDatabase
) {
async parseDocumentChange(change: PouchDB.Core.ExistingDocument<EntryDoc>) {
try {
if (this.shouldStopApplication(sourceDatabase)) return;
if (isAnyNote(change)) {
const docMtime = change.mtime ?? 0;
const maxMTime = this.context.currentSettings().maxMTimeForReflectEvents;
@@ -611,7 +449,6 @@ export class ReplicateResultProcessor {
}
// If the document is a virtual document, process it in the virtual document processor.
if (await this.services.replication.processVirtualDocument(change)) return;
if (this.shouldStopApplication(sourceDatabase)) return;
// If the document is version info, check compatibility and return.
if (isAnyNote(change)) {
const docPath = this.getPath(change);
@@ -619,7 +456,6 @@ export class ReplicateResultProcessor {
this.log(`Skipped: ${docPath}`, LOG_LEVEL_VERBOSE);
return;
}
if (this.shouldStopApplication(sourceDatabase)) return;
const size = change.size;
// Note that this size check depends size that in metadata, not the actual content size.
if (this.services.vault.isFileSizeTooLarge(size)) {
@@ -629,20 +465,11 @@ export class ReplicateResultProcessor {
);
return;
}
return await this.applyToDatabase(change, sourceDatabase);
return await this.applyToDatabase(change);
}
this.log(`Skipped unexpected non-note document: ${change._id}`, LOG_LEVEL_INFO);
return;
} finally {
// An in-flight parse may have started before the control document arrived.
// Retain it even if a later asynchronous boundary stopped application.
if (
this._compatibilityBlocked &&
sourceDatabase === this.localDatabase.localDatabase &&
!this._queuedChanges.includes(change)
) {
this._queuedChanges.push(change);
}
// Remove from processing queue
this._processingChanges = this._processingChanges.filter((e) => e !== change);
try {
@@ -662,16 +489,12 @@ export class ReplicateResultProcessor {
}
// Phase 2: apply the document to database
protected applyToDatabase(
doc: PouchDB.Core.ExistingDocument<AnyEntry>,
sourceDatabase: PouchDB.Database<EntryDoc> = this.localDatabase.localDatabase
) {
protected applyToDatabase(doc: PouchDB.Core.ExistingDocument<AnyEntry>) {
return this.withCounting(async () => {
let releaser: Awaited<ReturnType<typeof this._semaphore.acquire>> | undefined = undefined;
try {
releaser = await this._semaphore.acquire();
if (this.shouldStopApplication(sourceDatabase)) return;
await this._applyToDatabase(doc, sourceDatabase);
await this._applyToDatabase(doc);
} catch (e) {
this.log(`Error while processing replication result`, LOG_LEVEL_NOTICE);
this.logError(e);
@@ -685,16 +508,12 @@ export class ReplicateResultProcessor {
}
// Phase 2.1: process the document and apply to storage
// This function is serialized per document to avoid race-condition for the same document.
private _applyToDatabase(
doc_: PouchDB.Core.ExistingDocument<AnyEntry>,
sourceDatabase: PouchDB.Database<EntryDoc>
) {
private _applyToDatabase(doc_: PouchDB.Core.ExistingDocument<AnyEntry>) {
const dbDoc = doc_ as LoadedEntry; // It has no `data`
const path = this.getPath(dbDoc);
return serialized(`replication-process:${dbDoc._id}`, async () => {
const docNote = `${path} (${shortenId(dbDoc._id)}, ${shortenRev(dbDoc._rev)})`;
const isRequired = await this.checkIsChangeRequiredForDatabaseProcessing(dbDoc);
if (this.shouldStopApplication(sourceDatabase)) return;
if (!isRequired) {
this.log(`Skipped (Not latest): ${docNote}`, LOG_LEVEL_VERBOSE);
return;
@@ -708,7 +527,6 @@ export class ReplicateResultProcessor {
const doc = isDeleted
? { ...dbDoc, data: "" }
: await this.localDatabase.getDBEntryFromMeta({ ...dbDoc }, false, true);
if (this.shouldStopApplication(sourceDatabase)) return;
if (!doc) {
// Failed to gather content
this.log(`Failed to gather content of ${docNote}`, LOG_LEVEL_NOTICE);
@@ -719,10 +537,9 @@ export class ReplicateResultProcessor {
// Already processed
this.log(`Processed by other processor: ${docNote}`, LOG_LEVEL_DEBUG);
} else if (this.services.vault.isValidPath(this.getPath(doc))) {
if (this.shouldStopApplication(sourceDatabase)) return;
// Apply to storage if the path is valid
try {
const reflected = await this.applyToStorage(doc as MetaEntry, sourceDatabase);
const reflected = await this.applyToStorage(doc as MetaEntry);
if (!reflected) {
this.reportVaultReflectionFailure(doc as MetaEntry);
return;
@@ -743,15 +560,9 @@ export class ReplicateResultProcessor {
* @param entry
* @returns
*/
protected applyToStorage(
entry: MetaEntry,
sourceDatabase: PouchDB.Database<EntryDoc> = this.localDatabase.localDatabase
) {
protected applyToStorage(entry: MetaEntry) {
return this.withCounting(
() =>
this.shouldStopApplication(sourceDatabase)
? Promise.resolve(false)
: this.services.replication.processSynchroniseResult(entry),
() => this.services.replication.processSynchroniseResult(entry),
this.services.replication.storageApplyingCount
);
}
src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts
+43 -202
View File
@@ -120,6 +120,34 @@ function setup(options: SetupOptions = {}) {
}
describe("ReplicateResultProcessor", () => {
it("does not add a permanent application block when snapshot recovery fails", async () => {
const { processor } = setup({
getSnapshot: async () => {
throw new Error("KV unavailable");
},
});
await expect(processor.restoreFromSnapshotOnce()).rejects.toThrow("KV unavailable");
expect(processor.isSuspended).toBe(false);
});
it("restores pending notes without retaining a past feature rejection in KV", async () => {
const { processor, processSynchroniseResult, onCloseActiveReplication } = setup({
databaseId: "same-database",
getSnapshot: async () => ({
databaseId: "same-database",
invalidControlObserved: true,
observedFeatures: ["future-format-v7"],
observedGeneration: 14,
queued: [note("recovered-note")],
processing: [],
}),
});
await processor.restoreFromSnapshotOnce();
expect(processor.isSuspended).toBe(false);
await vi.waitFor(() => expect(processSynchroniseResult).toHaveBeenCalledOnce());
expect(onCloseActiveReplication).not.toHaveBeenCalled();
});
it.each([
["Windows", isValidFilenameInWidows],
["Android", isValidFilenameInAndroid],
@@ -249,30 +277,6 @@ describe("ReplicateResultProcessor", () => {
expect(onCloseActiveReplication).not.toHaveBeenCalled();
});
it.each(["first", "last"] as const)(
"holds an entire received batch when an unknown feature is %s",
async (position) => {
const { onCloseActiveReplication, processor, processSynchroniseResult } = setup();
const versionInfo = {
_id: VERSIONING_DOCID,
_rev: "2-unknown",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
} as PouchDB.Core.ExistingDocument<EntryDoc>;
const document = note("pending-after-compatibility-change");
processor.enqueueAll(position === "first" ? [versionInfo, document] : [document, versionInfo]);
await vi.waitFor(() => expect(onCloseActiveReplication).toHaveBeenCalledOnce());
await Promise.resolve();
expect(processor.isSuspended).toBe(true);
expect(processSynchroniseResult).not.toHaveBeenCalled();
processor.resume();
await Promise.resolve();
expect(processSynchroniseResult).not.toHaveBeenCalled();
}
);
it("continues when a newly received feature is supported", async () => {
const { onCloseActiveReplication, processor, processSynchroniseResult } = setup();
const versionInfo = {
@@ -289,149 +293,23 @@ describe("ReplicateResultProcessor", () => {
expect(onCloseActiveReplication).not.toHaveBeenCalled();
});
it("rechecks shared writer settings when a supported feature is added to an active database", async () => {
const { onCloseActiveReplication, processor } = setup({
localVersionInfo: {
_id: VERSIONING_DOCID,
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: [],
},
});
await processor.restoreFromSnapshotOnce();
const versionInfo = {
_id: VERSIONING_DOCID,
_rev: "2-supported",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: [ENCRYPTED_INTERNAL_METADATA_FEATURE],
} as PouchDB.Core.ExistingDocument<EntryDoc>;
processor.enqueueAll([versionInfo]);
processor.enqueueAll([{ ...versionInfo, _rev: "3-same-features" }]);
expect(onCloseActiveReplication).toHaveBeenCalledOnce();
expect(processor.isSuspended).toBe(false);
});
it("retains an in-flight note when a feature change arrives during an asynchronous check", async () => {
const targetCheck = promiseWithResolvers<boolean>();
const { isTargetFile, onCloseActiveReplication, processor, processSynchroniseResult } = setup({
isTargetFile: async () => targetCheck.promise,
});
const pending = note("in-flight");
processor.enqueueAll([pending]);
await vi.waitFor(() => expect(isTargetFile).toHaveBeenCalledOnce());
processor.enqueueAll([
{
_id: VERSIONING_DOCID,
_rev: "2-unknown",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
} as PouchDB.Core.ExistingDocument<EntryDoc>,
]);
targetCheck.resolve(true);
await vi.waitFor(() => expect(processor["_processingChanges"]).toHaveLength(0));
expect(processor["_queuedChanges"]).toContain(pending);
expect(processSynchroniseResult).not.toHaveBeenCalled();
expect(onCloseActiveReplication).toHaveBeenCalledOnce();
});
it("restores a checkpointed note only after assessing the persisted version document", async () => {
const pending = note("checkpointed");
const { onCloseActiveReplication, processor, processSynchroniseResult } = setup({
localVersionInfo: {
_id: VERSIONING_DOCID,
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
},
getSnapshot: async () => ({ processing: [pending], queued: [] }),
});
await processor.restoreFromSnapshotOnce();
expect(processor.isSuspended).toBe(true);
expect(processor["_queuedChanges"]).toContain(pending);
expect(processSynchroniseResult).not.toHaveBeenCalled();
expect(onCloseActiveReplication).toHaveBeenCalledOnce();
});
it("retains an observed unknown feature when the local version list is later shortened", async () => {
let snapshot: unknown;
const first = setup({
databaseId: "same-database",
setSnapshot: async (_key, value) => {
snapshot = value;
},
});
const pending = note("checkpointed-after-list-change");
first.processor.enqueueAll([
{
_id: VERSIONING_DOCID,
_rev: "2-unknown",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
} as PouchDB.Core.ExistingDocument<EntryDoc>,
pending,
]);
await vi.waitFor(() => expect(snapshot).toBeDefined());
const resumed = setup({
databaseId: "same-database",
localVersionInfo: {
_id: VERSIONING_DOCID,
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: [],
},
getSnapshot: async () => snapshot,
});
await resumed.processor.restoreFromSnapshotOnce();
expect(resumed.processor.isSuspended).toBe(true);
expect(resumed.processor["_queuedChanges"]).toContain(pending);
expect(resumed.processSynchroniseResult).not.toHaveBeenCalled();
expect(resumed.onCloseActiveReplication).toHaveBeenCalledOnce();
});
it("does not restore pending work belonging to a replaced physical database", async () => {
const pending = note("old-database");
const { processor, processSynchroniseResult } = setup({
databaseId: "replacement-database",
getSnapshot: async () => ({ databaseId: "retired-database", processing: [pending], queued: [] }),
});
await processor.restoreFromSnapshotOnce();
expect(processor["_queuedChanges"]).toHaveLength(0);
expect(processSynchroniseResult).not.toHaveBeenCalled();
expect(processor.isSuspended).toBe(false);
});
it("reports unknown identifiers and retires only once for repeated notifications", () => {
const log = vi.fn((_message: unknown, _level?: number) => undefined);
setGlobalLogFunction(log);
it("reports unknown feature identifiers and requests Replicator retirement", () => {
const logger = vi.fn();
setGlobalLogFunction(logger);
try {
const { onCloseActiveReplication, processor } = setup();
const versionInfo = {
_id: VERSIONING_DOCID,
_rev: "2-unknown",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
} as PouchDB.Core.ExistingDocument<EntryDoc>;
processor.enqueueAll([versionInfo]);
processor.enqueueAll([versionInfo]);
const { processor, onCloseActiveReplication } = setup();
processor.enqueueAll([
{
_id: VERSIONING_DOCID,
_rev: "1-unknown",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
} as PouchDB.Core.ExistingDocument<EntryDoc>,
]);
expect(onCloseActiveReplication).toHaveBeenCalledOnce();
expect(log).toHaveBeenCalledWith(
"[ReplicateResultProcessor] Unknown features are in use: future-format-v7",
expect(logger).toHaveBeenCalledWith(
expect.stringContaining("future-format-v7"),
LOG_LEVEL_NOTICE,
undefined
);
@@ -440,43 +318,6 @@ describe("ReplicateResultProcessor", () => {
}
});
it("ignores a late feature notification from a replaced physical database", () => {
const { onCloseActiveReplication, processor } = setup();
const oldPhysicalDatabase = {} as PouchDB.Database<EntryDoc>;
const versionInfo = {
_id: VERSIONING_DOCID,
_rev: "2-unknown",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
} as PouchDB.Core.ExistingDocument<EntryDoc>;
processor.enqueueAll([versionInfo], oldPhysicalDatabase);
expect(processor.isSuspended).toBe(false);
expect(onCloseActiveReplication).not.toHaveBeenCalled();
});
it("reassesses compatibility for a replacement local database", async () => {
const { localDatabase, onCloseActiveReplication, processor } = setup();
processor.enqueueAll([
{
_id: VERSIONING_DOCID,
_rev: "2-unknown",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
} as PouchDB.Core.ExistingDocument<EntryDoc>,
]);
expect(processor.isSuspended).toBe(true);
localDatabase.localDatabase = {} as PouchDB.Database<EntryDoc>;
await processor.restoreFromSnapshotOnce();
expect(processor.isSuspended).toBe(false);
expect(onCloseActiveReplication).toHaveBeenCalledOnce();
});
it("scans normal-file metadata without loading chunk documents and requeues it", async () => {
const documents = [
{ _id: "first", _rev: "1-a", type: "plain", path: "first.md" },
src/serviceFeatures/replication/index.ts
+6 -8
View File
@@ -98,21 +98,19 @@ export function useReplicationFeature<TContext extends ServiceContext, TCommands
clearHandlers();
return Promise.resolve(true);
});
services.databaseEvents.onDatabaseInitialised.addHandler(async () => {
await resultProcessor.restoreFromSnapshotOnce();
return true;
services.databaseEvents.onDatabaseInitialised.addHandler(() => {
fireAndForget(() => resultProcessor.restoreFromSnapshotOnce());
return Promise.resolve(true);
});
services.appLifecycle.onSettingLoaded.addHandler(initialiseAutomaticReplicationTriggers);
services.replication.parseSynchroniseResult.addHandler(async (documents, sourceDatabase) => {
resultProcessor.enqueueAll(documents, sourceDatabase);
await resultProcessor.persistBlockedSnapshot();
return true;
services.replication.parseSynchroniseResult.addHandler((documents) => {
resultProcessor.enqueueAll(documents);
return Promise.resolve(true);
});
services.replication.onBeforeReplicate.addHandler(onlinePreflight, 10);
services.replication.onPrepareCentralRemoteReplication.addHandler(securitySeedPreflight);
services.replication.onBeforeReplicate.addHandler(async () => {
await resultProcessor.restoreFromSnapshotOnce();
if (resultProcessor.isCompatibilityBlocked) return false;
unresolvedErrorManager.clearErrors();
return true;
}, 100);
src/serviceFeatures/replication/replicationFeature.unit.spec.ts
+3 -56
View File
@@ -22,7 +22,9 @@ type SetupOptions = {
function setup(options: SetupOptions = {}) {
const defaultLocalDatabase = {
localDatabase: {},
getRaw: vi.fn(async () => { throw { status: 404 }; }),
getRaw: vi.fn(async () => {
throw { status: 404 };
}),
};
const {
getLocalDatabase = () => defaultLocalDatabase,
@@ -183,59 +185,4 @@ describe("replication serviceFeature composition", () => {
retirement.resolve(true);
});
it("persists a blocked batch before its replication callback settles", async () => {
const writeFinished = promiseWithResolvers<void>();
const writes: Array<{ queued: Array<{ _id: string }> }> = [];
const { parseHandler } = setup({
keyValueDB: {
kvDB: {
get: vi.fn(async () => undefined),
set: vi.fn(async (_key, value) => {
writes.push(value as { queued: Array<{ _id: string }> });
await writeFinished.promise;
}),
},
},
});
const pending = {
_id: "checkpointed-note",
_rev: "1-test",
type: "plain",
path: "checkpointed-note.md",
} as PouchDB.Core.ExistingDocument<EntryDoc>;
const unknownVersion = {
_id: VERSIONING_DOCID,
_rev: "2-unknown",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
} as PouchDB.Core.ExistingDocument<EntryDoc>;
let callbackSettled = false;
const callback = parseHandler!([pending, unknownVersion]).then((result) => {
callbackSettled = true;
return result;
});
await vi.waitFor(() => expect(writes.length).toBeGreaterThan(0));
expect(callbackSettled).toBe(false);
writeFinished.resolve();
await expect(callback).resolves.toBe(true);
expect(writes[writes.length - 1]?.queued.map((entry) => entry._id)).toContain(pending._id);
});
it("refuses another replication after observing an unknown feature locally", async () => {
const { beforeReplicateHandlers, parseHandler } = setup();
const unknownVersion = {
_id: VERSIONING_DOCID,
_rev: "2-unknown",
type: "versioninfo",
version: REMOTE_FEATURE_GENERATION,
used_features: ["future-format-v7"],
} as PouchDB.Core.ExistingDocument<EntryDoc>;
await parseHandler!([unknownVersion]);
await expect(beforeReplicateHandlers.get(100)!(false)).resolves.toBe(false);
});
});