mirror of
https://github.com/vrtmrz/obsidian-livesync.git
synced 2026-09-21 18:17:05 +00:00
Compose replication lifecycle without a legacy module
This commit is contained in:
vorotamoroz
@@ -0,0 +1,574 @@
|
||||
import {
|
||||
SYNCINFO_ID,
|
||||
VER,
|
||||
type AnyEntry,
|
||||
type EntryDoc,
|
||||
type EntryLeaf,
|
||||
type LoadedEntry,
|
||||
type MetaEntry,
|
||||
type ObsidianLiveSyncSettings,
|
||||
} from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import { isChunk } from "@vrtmrz/livesync-commonlib/compat/common/typeUtils";
|
||||
import {
|
||||
LOG_LEVEL_DEBUG,
|
||||
LOG_LEVEL_INFO,
|
||||
LOG_LEVEL_NOTICE,
|
||||
LOG_LEVEL_VERBOSE,
|
||||
Logger,
|
||||
type LOG_LEVEL,
|
||||
} from "@vrtmrz/livesync-commonlib/compat/common/logger";
|
||||
import { fireAndForget, isAnyNote, throttle } from "@vrtmrz/livesync-commonlib/compat/common/utils";
|
||||
import { Semaphore } from "octagonal-wheels/concurrency/semaphore_v2";
|
||||
import { serialized } from "octagonal-wheels/concurrency/lock";
|
||||
import type { ReactiveSource } from "octagonal-wheels/dataobject/reactive_v2";
|
||||
import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore";
|
||||
import { isNotFoundError } from "@vrtmrz/livesync-commonlib/compat/common/utils.doc";
|
||||
import type PouchDB from "pouchdb-core";
|
||||
import { promiseWithResolvers, type PromiseWithResolvers } from "octagonal-wheels/promises";
|
||||
|
||||
const KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT = "replicationResultProcessorSnapshot";
|
||||
const REPROCESS_BATCH_SIZE = 100;
|
||||
type ReplicateResultProcessorSettings = Pick<
|
||||
ObsidianLiveSyncSettings,
|
||||
"maxMTimeForReflectEvents" | "suspendParseReplicationResult"
|
||||
>;
|
||||
type ReplicateResultProcessorServices = Pick<
|
||||
LiveSyncBaseCore["services"],
|
||||
"appLifecycle" | "path" | "replication" | "vault"
|
||||
>;
|
||||
|
||||
/**
|
||||
* Narrow collaborators for applying replicated documents.
|
||||
*
|
||||
* `requestActiveReplicatorRetirement` starts the owner transition without
|
||||
* awaiting it. Result application can still be running inside work admitted by
|
||||
* that owner, so awaiting retirement here could make each side wait for the
|
||||
* other to finish.
|
||||
*/
|
||||
interface ReplicateResultProcessorContext {
|
||||
readonly currentSettings: () => ReplicateResultProcessorSettings;
|
||||
readonly keyValueDB: LiveSyncBaseCore["kvDB"];
|
||||
readonly localDatabase: LiveSyncBaseCore["localDatabase"];
|
||||
readonly requestActiveReplicatorRetirement: () => void;
|
||||
readonly runLocalApplicationActivity: <T>(
|
||||
task: () => T | PromiseLike<T>,
|
||||
options?: { label?: string }
|
||||
) => Promise<T>;
|
||||
readonly services: ReplicateResultProcessorServices;
|
||||
}
|
||||
type ReplicateResultProcessorState = {
|
||||
queued: PouchDB.Core.ExistingDocument<EntryDoc>[];
|
||||
processing: PouchDB.Core.ExistingDocument<EntryDoc>[];
|
||||
};
|
||||
function shortenId(id: string): string {
|
||||
return id.length > 10 ? id.substring(0, 10) : id;
|
||||
}
|
||||
function shortenRev(rev: string | undefined): string {
|
||||
if (!rev) return "undefined";
|
||||
return rev.length > 10 ? rev.substring(0, 10) : rev;
|
||||
}
|
||||
export class ReplicateResultProcessor {
|
||||
private log(message: string, level: LOG_LEVEL = LOG_LEVEL_INFO) {
|
||||
Logger(`[ReplicateResultProcessor] ${message}`, level);
|
||||
}
|
||||
private logError(e: unknown) {
|
||||
Logger(e, LOG_LEVEL_VERBOSE);
|
||||
}
|
||||
constructor(private readonly context: ReplicateResultProcessorContext) {}
|
||||
|
||||
private get localDatabase() {
|
||||
return this.context.localDatabase;
|
||||
}
|
||||
private get services() {
|
||||
return this.context.services;
|
||||
}
|
||||
|
||||
getPath(entry: AnyEntry): string {
|
||||
return this.services.path.getPath(entry);
|
||||
}
|
||||
|
||||
public suspend() {
|
||||
this._suspended = true;
|
||||
this.updateProcessingActivity();
|
||||
}
|
||||
public resume() {
|
||||
this._suspended = false;
|
||||
this.updateProcessingActivity();
|
||||
fireAndForget(() => this.runProcessQueue());
|
||||
}
|
||||
|
||||
// Whether the processing is suspended
|
||||
// If true, the processing queue processor bails the loop.
|
||||
private _suspended: boolean = false;
|
||||
|
||||
public get isSuspended() {
|
||||
return (
|
||||
this._suspended ||
|
||||
!this.services.appLifecycle.isReady ||
|
||||
this.context.currentSettings().suspendParseReplicationResult ||
|
||||
this.services.appLifecycle.isSuspended()
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Take a snapshot of the current processing state.
|
||||
* This snapshot is stored in the KV database for recovery on restart.
|
||||
*/
|
||||
protected async _takeSnapshot() {
|
||||
const snapshot = {
|
||||
queued: this._queuedChanges.slice(),
|
||||
processing: this._processingChanges.slice(),
|
||||
} satisfies ReplicateResultProcessorState;
|
||||
await this.context.keyValueDB.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.
|
||||
*/
|
||||
protected _triggerTakeSnapshot() {
|
||||
fireAndForget(() => this._takeSnapshot());
|
||||
}
|
||||
/**
|
||||
* Throttled version of triggerTakeSnapshot.
|
||||
*/
|
||||
protected triggerTakeSnapshot = throttle(() => this._triggerTakeSnapshot(), 50);
|
||||
|
||||
/**
|
||||
* Restore from snapshot.
|
||||
*/
|
||||
public async restoreFromSnapshot() {
|
||||
const snapshot = await this.context.keyValueDB.get<ReplicateResultProcessorState>(
|
||||
KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT
|
||||
);
|
||||
if (snapshot) {
|
||||
// Restoring the snapshot re-runs processing for both queued and processing items.
|
||||
const newQueue = [...snapshot.processing, ...snapshot.queued, ...this._queuedChanges];
|
||||
this._queuedChanges = [];
|
||||
this.enqueueAll(newQueue);
|
||||
this.log(
|
||||
`Restored from snapshot (${snapshot.processing.length + snapshot.queued.length} items)`,
|
||||
LOG_LEVEL_INFO
|
||||
);
|
||||
// await this._takeSnapshot();
|
||||
}
|
||||
}
|
||||
|
||||
private _restoreFromSnapshot: Promise<void> | undefined = undefined;
|
||||
|
||||
/**
|
||||
* Restore from snapshot only once.
|
||||
* @returns Promise that resolves when restoration is complete.
|
||||
*/
|
||||
public restoreFromSnapshotOnce() {
|
||||
if (!this._restoreFromSnapshot) {
|
||||
this._restoreFromSnapshot = this.restoreFromSnapshot();
|
||||
}
|
||||
return this._restoreFromSnapshot;
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform the given procedure while counting the concurrency.
|
||||
* @param proc async procedure to perform
|
||||
* @param countValue reactive source to count concurrency
|
||||
* @returns result of the procedure
|
||||
*/
|
||||
async withCounting<T>(proc: () => Promise<T>, countValue: ReactiveSource<number>) {
|
||||
countValue.value++;
|
||||
try {
|
||||
return await proc();
|
||||
} finally {
|
||||
countValue.value--;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Report the current status.
|
||||
*/
|
||||
protected reportStatus() {
|
||||
this.services.replication.replicationResultCount.value =
|
||||
this._queuedChanges.length + this._processingChanges.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue all the given changes for processing.
|
||||
* @param changes Changes to enqueue
|
||||
*/
|
||||
|
||||
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);
|
||||
if (!isProcessed) {
|
||||
this.enqueueChange(change);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Requeues stored normal-file metadata after its reflection filters change.
|
||||
* Replication checkpoints may already cover documents which were skipped
|
||||
* by the previous filter, so a later ordinary sync cannot emit them again.
|
||||
*/
|
||||
public async reprocessStoredDocuments(): Promise<number> {
|
||||
let count = 0;
|
||||
let batch: PouchDB.Core.ExistingDocument<EntryDoc>[] = [];
|
||||
for await (const document of this.localDatabase.findAllNormalDocs()) {
|
||||
batch.push(document);
|
||||
count++;
|
||||
if (batch.length < REPROCESS_BATCH_SIZE) continue;
|
||||
this.enqueueAll(batch);
|
||||
batch = [];
|
||||
}
|
||||
if (batch.length > 0) this.enqueueAll(batch);
|
||||
this.log(`Requeued ${count} stored document(s) after the reflection filters changed`, LOG_LEVEL_INFO);
|
||||
return count;
|
||||
}
|
||||
/**
|
||||
* Process the change if it is not a document change.
|
||||
* @param change Change to process
|
||||
* @returns True if the change was processed; false otherwise
|
||||
*/
|
||||
protected processIfNonDocumentChange(change: PouchDB.Core.ExistingDocument<EntryDoc>) {
|
||||
if (!change) {
|
||||
this.log(`Received empty change`, LOG_LEVEL_VERBOSE);
|
||||
return true;
|
||||
}
|
||||
if (isChunk(change._id)) {
|
||||
// Emit event for new chunk
|
||||
this.localDatabase.onNewLeaf(change as EntryLeaf);
|
||||
this.log(`Processed chunk: ${shortenId(change._id)}`, LOG_LEVEL_DEBUG);
|
||||
return true;
|
||||
}
|
||||
if (change.type == "versioninfo") {
|
||||
this.log(`Version info document received: ${change._id}`, LOG_LEVEL_VERBOSE);
|
||||
if (change.version > VER) {
|
||||
// Fence and retire the active publication through its owner.
|
||||
this.context.requestActiveReplicatorRetirement();
|
||||
this.log(
|
||||
`Remote database updated to incompatible version. update your Self-hosted LiveSync plugin.`,
|
||||
LOG_LEVEL_NOTICE
|
||||
);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
if (
|
||||
change._id == SYNCINFO_ID || // Synchronisation information data
|
||||
change._id.startsWith("_design") //design document
|
||||
) {
|
||||
this.log(`Skipped system document: ${change._id}`, LOG_LEVEL_VERBOSE);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Queue of changes to be processed.
|
||||
*/
|
||||
private _queuedChanges: PouchDB.Core.ExistingDocument<EntryDoc>[] = [];
|
||||
|
||||
/**
|
||||
* List of changes being processed.
|
||||
*/
|
||||
private _processingChanges: PouchDB.Core.ExistingDocument<EntryDoc>[] = [];
|
||||
|
||||
private _processingActivity?: Promise<void>;
|
||||
private _processingActivityDone?: PromiseWithResolvers<void>;
|
||||
|
||||
private updateProcessingActivity() {
|
||||
if (this.isSuspended) {
|
||||
this._processingActivityDone?.resolve();
|
||||
return;
|
||||
}
|
||||
const hasPendingDocuments = this._queuedChanges.length > 0 || this._processingChanges.length > 0;
|
||||
if (!hasPendingDocuments) {
|
||||
this._processingActivityDone?.resolve();
|
||||
return;
|
||||
}
|
||||
if (this._processingActivity) return;
|
||||
|
||||
const activityDone = promiseWithResolvers<void>();
|
||||
this._processingActivityDone = activityDone;
|
||||
this._processingActivity = this.context
|
||||
.runLocalApplicationActivity(() => activityDone.promise, {
|
||||
label: "replicated-document-application",
|
||||
})
|
||||
.catch((error) => this.logError(error))
|
||||
.finally(() => {
|
||||
if (this._processingActivityDone === activityDone) this._processingActivityDone = undefined;
|
||||
this._processingActivity = undefined;
|
||||
this.updateProcessingActivity();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue the given document change for processing.
|
||||
* @param doc Document change to enqueue
|
||||
* @returns
|
||||
*/
|
||||
protected enqueueChange(doc: PouchDB.Core.ExistingDocument<EntryDoc>) {
|
||||
const old = this._queuedChanges.find((e) => e._id == doc._id);
|
||||
const path = "path" in doc ? this.getPath(doc) : "<unknown>";
|
||||
const docNote = `${path} (${shortenId(doc._id)}, ${shortenRev(doc._rev)})`;
|
||||
if (old) {
|
||||
if (old._rev == doc._rev) {
|
||||
this.log(`[Enqueue] skipped (Already queued): ${docNote}`, LOG_LEVEL_VERBOSE);
|
||||
return;
|
||||
}
|
||||
|
||||
const oldRev = old._rev ?? "";
|
||||
const isDeletedBefore = old._deleted === true || ("deleted" in old && old.deleted === true);
|
||||
const isDeletedNow = doc._deleted === true || ("deleted" in doc && doc.deleted === true);
|
||||
|
||||
// Replace the old queued change (This may performed batched updates, actually process performed always with the latest version, hence we can simply replace it if the change is the same type).
|
||||
if (isDeletedBefore === isDeletedNow) {
|
||||
this._queuedChanges = this._queuedChanges.filter((e) => e._id != doc._id);
|
||||
this.log(`[Enqueue] requeued: ${docNote} (from rev: ${shortenRev(oldRev)})`, LOG_LEVEL_VERBOSE);
|
||||
}
|
||||
}
|
||||
// Enqueue the change
|
||||
this._queuedChanges.push(doc);
|
||||
this.updateProcessingActivity();
|
||||
this.triggerTakeSnapshot();
|
||||
this.triggerProcessQueue();
|
||||
}
|
||||
|
||||
/**
|
||||
* Trigger processing of the queued changes.
|
||||
*/
|
||||
protected triggerProcessQueue() {
|
||||
fireAndForget(() => this.runProcessQueue());
|
||||
}
|
||||
|
||||
/**
|
||||
* Semaphore to limit concurrent processing.
|
||||
* This is the per-id semaphore + concurrency-control (max 10 concurrent = 10 documents being processed at the same time).
|
||||
*/
|
||||
private _semaphore = Semaphore(10);
|
||||
|
||||
/**
|
||||
* Flag indicating whether the process queue is currently running.
|
||||
*/
|
||||
private _isRunningProcessQueue: boolean = false;
|
||||
|
||||
/**
|
||||
* Process the queued changes.
|
||||
*/
|
||||
private async runProcessQueue() {
|
||||
// Avoid re-entrance, suspend processing, or empty queue loop consumption.
|
||||
if (this._isRunningProcessQueue) return;
|
||||
if (this.isSuspended) return;
|
||||
if (this._queuedChanges.length == 0) return;
|
||||
try {
|
||||
this._isRunningProcessQueue = true;
|
||||
while (this._queuedChanges.length > 0) {
|
||||
// If getting suspended, bail the loop. Some concurrent tasks may still be running.
|
||||
if (this.isSuspended) {
|
||||
this.log(
|
||||
`Processing has got suspended. Remaining items in queue: ${this._queuedChanges.length}`,
|
||||
LOG_LEVEL_INFO
|
||||
);
|
||||
break;
|
||||
}
|
||||
|
||||
// Acquire semaphore for new processing slot
|
||||
// (per-document serialisation caps concurrency).
|
||||
const releaser = await this._semaphore.acquire();
|
||||
releaser();
|
||||
// Dequeue the next change
|
||||
const doc = this._queuedChanges.shift();
|
||||
if (doc) {
|
||||
this._processingChanges.push(doc);
|
||||
void this.parseDocumentChange(doc);
|
||||
}
|
||||
// Take snapshot (to be restored on next startup if needed)
|
||||
this.triggerTakeSnapshot();
|
||||
}
|
||||
} finally {
|
||||
this._isRunningProcessQueue = false;
|
||||
}
|
||||
}
|
||||
|
||||
// Phase 1: parse replication result
|
||||
/**
|
||||
* Parse the given document change.
|
||||
* @param change
|
||||
* @returns
|
||||
*/
|
||||
async parseDocumentChange(change: PouchDB.Core.ExistingDocument<EntryDoc>) {
|
||||
try {
|
||||
if (isAnyNote(change)) {
|
||||
const docMtime = change.mtime ?? 0;
|
||||
const maxMTime = this.context.currentSettings().maxMTimeForReflectEvents;
|
||||
if (maxMTime > 0 && docMtime > maxMTime) {
|
||||
const docPath = this.getPath(change);
|
||||
this.log(
|
||||
`Processing ${docPath} has been skipped due to modification time (${new Date(
|
||||
docMtime * 1000
|
||||
).toISOString()}) exceeding the limit`,
|
||||
LOG_LEVEL_INFO
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
// If the document is a virtual document, process it in the virtual document processor.
|
||||
if (await this.services.replication.processVirtualDocument(change)) return;
|
||||
// If the document is version info, check compatibility and return.
|
||||
if (isAnyNote(change)) {
|
||||
const docPath = this.getPath(change);
|
||||
if (!(await this.services.vault.isTargetFile(docPath))) {
|
||||
this.log(`Skipped: ${docPath}`, LOG_LEVEL_VERBOSE);
|
||||
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)) {
|
||||
this.log(
|
||||
`Processing ${docPath} has been skipped due to file size exceeding the limit`,
|
||||
LOG_LEVEL_NOTICE
|
||||
);
|
||||
return;
|
||||
}
|
||||
return await this.applyToDatabase(change);
|
||||
}
|
||||
this.log(`Skipped unexpected non-note document: ${change._id}`, LOG_LEVEL_INFO);
|
||||
return;
|
||||
} finally {
|
||||
// Remove from processing queue
|
||||
this._processingChanges = this._processingChanges.filter((e) => e !== change);
|
||||
try {
|
||||
if (this._queuedChanges.length === 0 && this._processingChanges.length === 0) {
|
||||
try {
|
||||
await this._takeSnapshot();
|
||||
} catch (error) {
|
||||
this.logError(error);
|
||||
}
|
||||
} else {
|
||||
this.triggerTakeSnapshot();
|
||||
}
|
||||
} finally {
|
||||
this.updateProcessingActivity();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Phase 2: apply the document to database
|
||||
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();
|
||||
await this._applyToDatabase(doc);
|
||||
} catch (e) {
|
||||
this.log(`Error while processing replication result`, LOG_LEVEL_NOTICE);
|
||||
this.logError(e);
|
||||
} finally {
|
||||
// Remove from processing queue (To remove from "in-progress" list, and snapshot will not include it)
|
||||
if (releaser) {
|
||||
releaser();
|
||||
}
|
||||
}
|
||||
}, this.services.replication.databaseQueueCount);
|
||||
}
|
||||
// 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>) {
|
||||
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 (!isRequired) {
|
||||
this.log(`Skipped (Not latest): ${docNote}`, LOG_LEVEL_VERBOSE);
|
||||
return;
|
||||
}
|
||||
// If `Read chunks online` is disabled, chunks should be transferred before here.
|
||||
// However, in some cases, chunks are after that. So, if missing chunks exist, we have to wait for them.
|
||||
// (If `Use Only Local Chunks` is enabled, we should not attempt to fetch chunks online automatically).
|
||||
|
||||
const isDeleted = dbDoc._deleted === true || ("deleted" in dbDoc && dbDoc.deleted === true);
|
||||
// Gather full document if not deleted
|
||||
const doc = isDeleted
|
||||
? { ...dbDoc, data: "" }
|
||||
: await this.localDatabase.getDBEntryFromMeta({ ...dbDoc }, false, true);
|
||||
if (!doc) {
|
||||
// Failed to gather content
|
||||
this.log(`Failed to gather content of ${docNote}`, LOG_LEVEL_NOTICE);
|
||||
return;
|
||||
}
|
||||
// Check if other processor wants to process this document, if so, skip processing here.
|
||||
if (await this.services.replication.processOptionalSynchroniseResult(dbDoc)) {
|
||||
// Already processed
|
||||
this.log(`Processed by other processor: ${docNote}`, LOG_LEVEL_DEBUG);
|
||||
} else if (this.services.vault.isValidPath(this.getPath(doc))) {
|
||||
// Apply to storage if the path is valid
|
||||
await this.applyToStorage(doc as MetaEntry);
|
||||
this.log(`Processed: ${docNote}`, LOG_LEVEL_DEBUG);
|
||||
} else {
|
||||
// Should process, but have an invalid path
|
||||
this.log(`Unprocessed (Invalid path): ${docNote}`, LOG_LEVEL_VERBOSE);
|
||||
}
|
||||
return;
|
||||
});
|
||||
}
|
||||
/**
|
||||
* Phase 3: Apply the given entry to storage.
|
||||
* @param entry
|
||||
* @returns
|
||||
*/
|
||||
protected applyToStorage(entry: MetaEntry) {
|
||||
return this.withCounting(async () => {
|
||||
await this.services.replication.processSynchroniseResult(entry);
|
||||
}, this.services.replication.storageApplyingCount);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check whether processing is required for the given document.
|
||||
* @param dbDoc Document to check
|
||||
* @returns True if processing is required; false otherwise
|
||||
*/
|
||||
protected async checkIsChangeRequiredForDatabaseProcessing(dbDoc: LoadedEntry): Promise<boolean> {
|
||||
const path = this.getPath(dbDoc);
|
||||
try {
|
||||
const savedDoc = await this.localDatabase.getRaw<LoadedEntry>(dbDoc._id, {
|
||||
conflicts: true,
|
||||
revs_info: true,
|
||||
});
|
||||
const newRev = dbDoc._rev ?? "";
|
||||
const latestRev = savedDoc._rev ?? "";
|
||||
const revisions = savedDoc._revs_info?.map((e) => e.rev) ?? [];
|
||||
if (savedDoc._conflicts && savedDoc._conflicts.length > 0) {
|
||||
// There are conflicts, so we have to process it.
|
||||
// (May auto-resolve or user intervention will be occurred).
|
||||
return true;
|
||||
}
|
||||
if (newRev == latestRev) {
|
||||
// The latest revision. Simply we can process it.
|
||||
return true;
|
||||
}
|
||||
const index = revisions.indexOf(newRev);
|
||||
if (index >= 0) {
|
||||
// The revision has been inserted before.
|
||||
return false; // This means that the document already processed (While no conflict existed).
|
||||
}
|
||||
return true; // This mostly should not happen, but we have to process it just in case.
|
||||
} catch (e) {
|
||||
if (isNotFoundError(e)) {
|
||||
// getRaw failed due to not existing, it may not be happened normally especially on replication.
|
||||
// If the process caused by some other reason, we **probably** have to process it.
|
||||
// Note that this is not a common case.
|
||||
return true;
|
||||
} else {
|
||||
this.log(
|
||||
`Failed to get existing document for ${path} (${shortenId(dbDoc._id)}, ${shortenRev(dbDoc._rev)}) `,
|
||||
LOG_LEVEL_NOTICE
|
||||
);
|
||||
this.logError(e);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
import { promiseWithResolvers } from "octagonal-wheels/promises";
|
||||
import { reactiveSource } from "octagonal-wheels/dataobject/reactive";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { VER, type EntryDoc } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import { ReplicateResultProcessor } from "./ReplicateResultProcessor";
|
||||
|
||||
function note(id: string): PouchDB.Core.ExistingDocument<EntryDoc> {
|
||||
return {
|
||||
_id: id,
|
||||
_rev: "1-test",
|
||||
path: `${id}.md`,
|
||||
ctime: 1,
|
||||
mtime: 2,
|
||||
size: 1,
|
||||
children: [],
|
||||
datatype: "plain",
|
||||
type: "plain",
|
||||
eden: {},
|
||||
} as unknown as PouchDB.Core.ExistingDocument<EntryDoc>;
|
||||
}
|
||||
|
||||
type SetupOptions = {
|
||||
processSynchroniseResult?: (entry: unknown) => Promise<void>;
|
||||
setSnapshot?: (key: string, value: unknown) => Promise<unknown>;
|
||||
};
|
||||
|
||||
function setup(options: SetupOptions = {}) {
|
||||
const processSynchroniseResult = vi.fn(options.processSynchroniseResult ?? (async () => undefined));
|
||||
const setSnapshot = vi.fn(options.setSnapshot ?? (async () => undefined));
|
||||
const runBoundedLocalApplicationActivity = vi.fn(async (task: () => Promise<void>) => await task());
|
||||
const onCloseActiveReplication = vi.fn(async () => true);
|
||||
const core = {
|
||||
services: {
|
||||
appLifecycle: { isReady: true, isSuspended: () => false },
|
||||
path: { getPath: (entry: { path: string }) => entry.path },
|
||||
replication: {
|
||||
databaseQueueCount: reactiveSource(0),
|
||||
storageApplyingCount: reactiveSource(0),
|
||||
replicationResultCount: reactiveSource(0),
|
||||
processVirtualDocument: vi.fn(async () => false),
|
||||
processOptionalSynchroniseResult: vi.fn(async () => false),
|
||||
processSynchroniseResult,
|
||||
},
|
||||
replicator: { onCloseActiveReplication, runBoundedLocalApplicationActivity },
|
||||
vault: {
|
||||
isTargetFile: vi.fn(async () => true),
|
||||
isFileSizeTooLarge: vi.fn(() => false),
|
||||
isValidPath: vi.fn(() => true),
|
||||
},
|
||||
},
|
||||
kvDB: { set: setSnapshot },
|
||||
localDatabase: {
|
||||
getRaw: vi.fn(async (id: string) => ({ _id: id, _rev: "1-test" })),
|
||||
getDBEntryFromMeta: vi.fn(async (entry: object) => ({ ...entry, data: "x" })),
|
||||
},
|
||||
};
|
||||
const processor = new ReplicateResultProcessor({
|
||||
currentSettings: () => ({ maxMTimeForReflectEvents: 0, suspendParseReplicationResult: false }),
|
||||
keyValueDB: core.kvDB,
|
||||
localDatabase: core.localDatabase,
|
||||
requestActiveReplicatorRetirement: () => {
|
||||
void onCloseActiveReplication();
|
||||
},
|
||||
runLocalApplicationActivity: runBoundedLocalApplicationActivity,
|
||||
services: core.services,
|
||||
} as never);
|
||||
return {
|
||||
onCloseActiveReplication,
|
||||
processor,
|
||||
processSynchroniseResult,
|
||||
runBoundedLocalApplicationActivity,
|
||||
};
|
||||
}
|
||||
|
||||
describe("ReplicateResultProcessor", () => {
|
||||
it("retires active ownership when a newer remote version is observed", async () => {
|
||||
const { onCloseActiveReplication, processor } = setup();
|
||||
const versionInfo = {
|
||||
_id: "versioninfo",
|
||||
_rev: "1-test",
|
||||
type: "versioninfo",
|
||||
version: VER + 1,
|
||||
} as unknown as PouchDB.Core.ExistingDocument<EntryDoc>;
|
||||
|
||||
processor.enqueueAll([versionInfo]);
|
||||
|
||||
await vi.waitFor(() => 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" },
|
||||
{ _id: "second", _rev: "1-b", type: "plain", path: "second.md" },
|
||||
] as unknown as PouchDB.Core.ExistingDocument<EntryDoc>[];
|
||||
const findAllNormalDocs = vi.fn(async function* () {
|
||||
yield* documents;
|
||||
});
|
||||
const processor = new ReplicateResultProcessor({
|
||||
localDatabase: { findAllNormalDocs },
|
||||
} as never);
|
||||
const enqueueAll = vi.spyOn(processor, "enqueueAll").mockImplementation(() => undefined);
|
||||
|
||||
await expect(processor.reprocessStoredDocuments()).resolves.toBe(2);
|
||||
|
||||
expect(findAllNormalDocs).toHaveBeenCalledOnce();
|
||||
expect(enqueueAll).toHaveBeenCalledOnce();
|
||||
expect(enqueueAll).toHaveBeenCalledWith(documents);
|
||||
});
|
||||
|
||||
it("keeps one local application activity until every replicated document has been applied", async () => {
|
||||
const applying = promiseWithResolvers<void>();
|
||||
let activityFinished = false;
|
||||
const { processor, processSynchroniseResult, runBoundedLocalApplicationActivity } = setup({
|
||||
processSynchroniseResult: async () => applying.promise,
|
||||
});
|
||||
runBoundedLocalApplicationActivity.mockImplementation(async (task: () => Promise<void>) => {
|
||||
await task();
|
||||
activityFinished = true;
|
||||
});
|
||||
|
||||
processor.enqueueAll([note("one"), note("two")]);
|
||||
|
||||
await vi.waitFor(() => expect(processSynchroniseResult).toHaveBeenCalledTimes(2));
|
||||
expect(runBoundedLocalApplicationActivity).toHaveBeenCalledTimes(1);
|
||||
expect(runBoundedLocalApplicationActivity).toHaveBeenCalledWith(expect.any(Function), {
|
||||
label: "replicated-document-application",
|
||||
});
|
||||
expect(activityFinished).toBe(false);
|
||||
|
||||
applying.resolve();
|
||||
|
||||
await vi.waitFor(() => expect(activityFinished).toBe(true));
|
||||
});
|
||||
|
||||
it("settles local application activity when the final recovery snapshot fails", async () => {
|
||||
let activityFinished = false;
|
||||
const { processor, runBoundedLocalApplicationActivity } = setup({
|
||||
setSnapshot: async () => Promise.reject(new Error("snapshot failed")),
|
||||
});
|
||||
runBoundedLocalApplicationActivity.mockImplementation(async (task: () => Promise<void>) => {
|
||||
await task();
|
||||
activityFinished = true;
|
||||
});
|
||||
|
||||
processor.enqueueAll([note("one")]);
|
||||
|
||||
await vi.waitFor(() => expect(activityFinished).toBe(true));
|
||||
});
|
||||
|
||||
it("releases and reacquires local application activity around processing suspension", async () => {
|
||||
const applying = promiseWithResolvers<void>();
|
||||
let completedActivities = 0;
|
||||
const { processor, processSynchroniseResult, runBoundedLocalApplicationActivity } = setup({
|
||||
processSynchroniseResult: async () => applying.promise,
|
||||
});
|
||||
runBoundedLocalApplicationActivity.mockImplementation(async (task: () => Promise<void>) => {
|
||||
await task();
|
||||
completedActivities++;
|
||||
});
|
||||
processor.enqueueAll([note("one")]);
|
||||
await vi.waitFor(() => expect(processSynchroniseResult).toHaveBeenCalledOnce());
|
||||
|
||||
processor.suspend();
|
||||
await vi.waitFor(() => expect(completedActivities).toBe(1));
|
||||
|
||||
processor.resume();
|
||||
await vi.waitFor(() => expect(runBoundedLocalApplicationActivity).toHaveBeenCalledTimes(2));
|
||||
|
||||
applying.resolve();
|
||||
await vi.waitFor(() => expect(completedActivities).toBe(2));
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,71 @@
|
||||
import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import { fireAndForget } from "octagonal-wheels/promises";
|
||||
import { scheduleTask } from "octagonal-wheels/concurrency/task";
|
||||
import { EVENT_FILE_SAVED, EVENT_SETTING_SAVED, eventHub } from "@/common/events";
|
||||
|
||||
type ReflectionFilterSettings = Pick<
|
||||
ObsidianLiveSyncSettings,
|
||||
| "handleFilenameCaseSensitive"
|
||||
| "ignoreFiles"
|
||||
| "maxMTimeForReflectEvents"
|
||||
| "syncIgnoreRegEx"
|
||||
| "syncInternalFiles"
|
||||
| "syncMaxSizeInMB"
|
||||
| "syncOnlyRegEx"
|
||||
| "useIgnoreFiles"
|
||||
>;
|
||||
|
||||
interface AutomaticReplicationTriggerContext {
|
||||
readonly currentSettings: () => ObsidianLiveSyncSettings;
|
||||
readonly isSuspended: () => boolean;
|
||||
readonly replicateDatabaseEvent: () => Promise<unknown>;
|
||||
readonly reprocessStoredDocuments: () => Promise<number>;
|
||||
readonly resumeResultApplication: () => void;
|
||||
readonly suspendResultApplication: () => void;
|
||||
}
|
||||
|
||||
function normalFileReflectionFilterSignature(settings: ReflectionFilterSettings): string {
|
||||
return JSON.stringify({
|
||||
handleFilenameCaseSensitive: settings.handleFilenameCaseSensitive ?? false,
|
||||
ignoreFiles: settings.ignoreFiles ?? "",
|
||||
maxMTimeForReflectEvents: settings.maxMTimeForReflectEvents ?? 0,
|
||||
syncIgnoreRegEx: settings.syncIgnoreRegEx ?? "",
|
||||
syncInternalFiles: settings.syncInternalFiles ?? false,
|
||||
syncMaxSizeInMB: settings.syncMaxSizeInMB ?? 0,
|
||||
syncOnlyRegEx: settings.syncOnlyRegEx ?? "",
|
||||
useIgnoreFiles: settings.useIgnoreFiles ?? false,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the settings-loaded handler which installs automatic replication and
|
||||
* result-application reactions. The returned closure owns the previous filter
|
||||
* signature; it is private composition state rather than a shared service.
|
||||
*/
|
||||
export function createAutomaticReplicationTriggers(context: AutomaticReplicationTriggerContext) {
|
||||
let reflectionFilterSignature: string | undefined;
|
||||
|
||||
return function initialiseAutomaticReplicationTriggers(): Promise<boolean> {
|
||||
reflectionFilterSignature = normalFileReflectionFilterSignature(context.currentSettings());
|
||||
eventHub.onEvent(EVENT_FILE_SAVED, () => {
|
||||
if (context.currentSettings().syncOnSave && !context.isSuspended()) {
|
||||
scheduleTask("perform-replicate-after-save", 250, () => context.replicateDatabaseEvent());
|
||||
}
|
||||
});
|
||||
eventHub.onEvent(EVENT_SETTING_SAVED, (settings) => {
|
||||
const previousReflectionFilter = reflectionFilterSignature;
|
||||
const nextReflectionFilter = normalFileReflectionFilterSignature(settings);
|
||||
reflectionFilterSignature = nextReflectionFilter;
|
||||
if (settings.suspendParseReplicationResult) {
|
||||
context.suspendResultApplication();
|
||||
} else {
|
||||
context.resumeResultApplication();
|
||||
}
|
||||
if (previousReflectionFilter !== undefined && previousReflectionFilter !== nextReflectionFilter) {
|
||||
fireAndForget(() => context.reprocessStoredDocuments());
|
||||
}
|
||||
});
|
||||
|
||||
return Promise.resolve(true);
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,328 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
AUTO_MERGED,
|
||||
DEFAULT_SETTINGS,
|
||||
REMOTE_P2P,
|
||||
type FilePathWithPrefix,
|
||||
type ObsidianLiveSyncSettings,
|
||||
} from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication";
|
||||
import { EVENT_FILE_SAVED, EVENT_SETTING_SAVED, eventHub } from "@/common/events";
|
||||
|
||||
const taskMocks = vi.hoisted(() => ({
|
||||
scheduleTask: vi.fn((_key: string, _delay: number, task: () => unknown) => task()),
|
||||
}));
|
||||
|
||||
vi.mock("octagonal-wheels/concurrency/task", () => taskMocks);
|
||||
|
||||
import { ModuleConflictResolver } from "@/modules/coreFeatures/ModuleConflictResolver";
|
||||
import { ModuleObsidianEvents } from "@/modules/essentialObsidian/ModuleObsidianEvents";
|
||||
import {
|
||||
createReplicationSchedulingContext,
|
||||
realiseReplicationScheduling,
|
||||
resumeReplicationScheduling,
|
||||
runPeriodicReplication,
|
||||
} from "@/serviceFeatures/replicationScheduling";
|
||||
import { createAutomaticReplicationTriggers } from "./automaticTriggers";
|
||||
|
||||
function createApi() {
|
||||
return {
|
||||
addLog: vi.fn(),
|
||||
addCommand: vi.fn(),
|
||||
registerWindow: vi.fn(),
|
||||
addRibbonIcon: vi.fn(),
|
||||
registerProtocolHandler: vi.fn(),
|
||||
setInterval: vi.fn(),
|
||||
clearInterval: vi.fn(),
|
||||
};
|
||||
}
|
||||
|
||||
function p2pSettings(overrides: Partial<typeof DEFAULT_SETTINGS> = {}) {
|
||||
return {
|
||||
...DEFAULT_SETTINGS,
|
||||
remoteType: REMOTE_P2P,
|
||||
isConfigured: true,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function createObsidianEventHarness(settings: Partial<typeof DEFAULT_SETTINGS>) {
|
||||
const save = vi.fn();
|
||||
const saveCommand = { callback: save };
|
||||
const replicateUnattendedByEvent = vi.fn(async () => ({ status: "completed" as const }));
|
||||
const queueCheckForIfOpen = vi.fn(async () => undefined);
|
||||
const services = {
|
||||
API: createApi(),
|
||||
appLifecycle: {
|
||||
isReady: vi.fn(() => true),
|
||||
isSuspended: vi.fn(() => false),
|
||||
},
|
||||
conflict: { queueCheckForIfOpen },
|
||||
control: { hasUnloaded: vi.fn(() => false) },
|
||||
fileProcessing: { commitPendingFileEvents: vi.fn(async () => true) },
|
||||
replication: { replicateUnattendedByEvent },
|
||||
};
|
||||
const core = {
|
||||
_services: services,
|
||||
services,
|
||||
settings: p2pSettings(settings),
|
||||
} as any;
|
||||
const plugin = {
|
||||
app: {
|
||||
commands: {
|
||||
commands: { "editor:save-file": saveCommand },
|
||||
executeCommandById: vi.fn(),
|
||||
},
|
||||
},
|
||||
} as any;
|
||||
|
||||
return {
|
||||
module: new ModuleObsidianEvents(plugin, core),
|
||||
queueCheckForIfOpen,
|
||||
replicateUnattendedByEvent,
|
||||
save,
|
||||
saveCommand,
|
||||
services,
|
||||
};
|
||||
}
|
||||
|
||||
describe("automatic replication triggers while P2P is active", () => {
|
||||
afterEach(() => {
|
||||
eventHub.offAll();
|
||||
taskMocks.scheduleTask.mockClear();
|
||||
});
|
||||
|
||||
it("keeps periodic synchronisation on the provider-independent replication boundary", async () => {
|
||||
const replicateUnattended = vi.fn(async () => ({ status: "completed" as const }));
|
||||
const services = {
|
||||
API: createApi(),
|
||||
control: { hasUnloaded: vi.fn(() => false) },
|
||||
replication: { replicateUnattended },
|
||||
};
|
||||
const core = {
|
||||
_services: services,
|
||||
services,
|
||||
settings: p2pSettings({ periodicReplication: true, syncOnStart: false }),
|
||||
} as any;
|
||||
const context = createReplicationSchedulingContext({
|
||||
isReady: vi.fn(() => true),
|
||||
isSuspended: vi.fn(() => false),
|
||||
currentSettings: vi.fn(() => core.settings),
|
||||
replicateUnattended,
|
||||
startContinuous: vi.fn(async () => ({ status: "completed" as const })),
|
||||
timer: { enable: vi.fn(), disable: vi.fn() },
|
||||
log: vi.fn(),
|
||||
});
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
await runPeriodicReplication(context);
|
||||
|
||||
expect(replicateUnattended).toHaveBeenCalledOnce();
|
||||
expect(replicateUnattended).toHaveBeenCalledWith({
|
||||
trigger: "periodic",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps database-save synchronisation on the event replication boundary", async () => {
|
||||
const replicateUnattendedByEvent = vi.fn(async (_request: unknown) => ({ status: "completed" as const }));
|
||||
const settings = p2pSettings({ syncOnSave: true });
|
||||
const initialise = createAutomaticReplicationTriggers({
|
||||
currentSettings: () => settings,
|
||||
isSuspended: vi.fn(() => false),
|
||||
replicateDatabaseEvent: () =>
|
||||
replicateUnattendedByEvent({
|
||||
trigger: "database-event",
|
||||
interaction: NO_INTERACTION,
|
||||
}),
|
||||
reprocessStoredDocuments: vi.fn(async () => 0),
|
||||
resumeResultApplication: vi.fn(),
|
||||
suspendResultApplication: vi.fn(),
|
||||
});
|
||||
|
||||
await initialise();
|
||||
eventHub.emitEvent(EVENT_FILE_SAVED);
|
||||
|
||||
await vi.waitFor(() => expect(replicateUnattendedByEvent).toHaveBeenCalledOnce());
|
||||
expect(replicateUnattendedByEvent).toHaveBeenCalledWith({
|
||||
trigger: "database-event",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
});
|
||||
|
||||
it("reprocesses stored documents when normal-file target filters change", async () => {
|
||||
const settings = {
|
||||
...DEFAULT_SETTINGS,
|
||||
ignoreFiles: ".gitignore",
|
||||
syncOnlyRegEx: "^E2E/allowed/.*",
|
||||
} as ObsidianLiveSyncSettings;
|
||||
const reprocessStoredDocuments = vi.fn(async () => 1);
|
||||
const resumeResultApplication = vi.fn();
|
||||
const suspendResultApplication = vi.fn();
|
||||
const initialise = createAutomaticReplicationTriggers({
|
||||
currentSettings: () => settings,
|
||||
isSuspended: vi.fn(() => false),
|
||||
replicateDatabaseEvent: vi.fn(async () => undefined),
|
||||
reprocessStoredDocuments,
|
||||
resumeResultApplication,
|
||||
suspendResultApplication,
|
||||
});
|
||||
|
||||
await initialise();
|
||||
eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings });
|
||||
await Promise.resolve();
|
||||
expect(reprocessStoredDocuments).not.toHaveBeenCalled();
|
||||
expect(resumeResultApplication).toHaveBeenCalledOnce();
|
||||
expect(suspendResultApplication).not.toHaveBeenCalled();
|
||||
|
||||
eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings, suspendParseReplicationResult: true });
|
||||
expect(suspendResultApplication).toHaveBeenCalledOnce();
|
||||
|
||||
Object.assign(settings, { syncOnlyRegEx: "" });
|
||||
eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings });
|
||||
await vi.waitFor(() => expect(reprocessStoredDocuments).toHaveBeenCalledOnce());
|
||||
|
||||
settings.syncMaxSizeInMB = 10;
|
||||
eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings });
|
||||
await vi.waitFor(() => expect(reprocessStoredDocuments).toHaveBeenCalledTimes(2));
|
||||
});
|
||||
|
||||
it("keeps editor-save synchronisation on the event replication boundary", async () => {
|
||||
const { module, replicateUnattendedByEvent, save, saveCommand } = createObsidianEventHarness({
|
||||
syncOnEditorSave: true,
|
||||
});
|
||||
|
||||
module.swapSaveCommand();
|
||||
saveCommand.callback();
|
||||
|
||||
expect(save).toHaveBeenCalledOnce();
|
||||
await vi.waitFor(() => expect(replicateUnattendedByEvent).toHaveBeenCalledOnce());
|
||||
expect(replicateUnattendedByEvent).toHaveBeenCalledWith({
|
||||
trigger: "editor-save",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps file-open synchronisation on the event replication boundary", async () => {
|
||||
const { module, queueCheckForIfOpen, replicateUnattendedByEvent, services } = createObsidianEventHarness({
|
||||
syncOnFileOpen: true,
|
||||
});
|
||||
const file = { path: "opened.md" } as never;
|
||||
|
||||
await module.watchWorkspaceOpenAsync(file);
|
||||
|
||||
expect(services.fileProcessing.commitPendingFileEvents).toHaveBeenCalledOnce();
|
||||
expect(replicateUnattendedByEvent).toHaveBeenCalledOnce();
|
||||
expect(replicateUnattendedByEvent).toHaveBeenCalledWith({
|
||||
trigger: "file-open",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
expect(queueCheckForIfOpen).toHaveBeenCalledWith("opened.md");
|
||||
});
|
||||
|
||||
it("keeps post-merge synchronisation on the event replication boundary", async () => {
|
||||
const replicateUnattendedByEvent = vi.fn(async () => ({ status: "completed" as const }));
|
||||
const queueCheckFor = vi.fn(async () => undefined);
|
||||
const path = "merged.md" as FilePathWithPrefix;
|
||||
const module = {
|
||||
settings: p2pSettings({ syncAfterMerge: true }),
|
||||
services: {
|
||||
appLifecycle: { isSuspended: vi.fn(() => false) },
|
||||
conflict: { queueCheckFor },
|
||||
replication: { replicateUnattendedByEvent },
|
||||
},
|
||||
checkConflictAndPerformAutoMerge: vi.fn(async () => AUTO_MERGED),
|
||||
_log: vi.fn(),
|
||||
};
|
||||
|
||||
await (ModuleConflictResolver.prototype as any)._resolveConflict.call(module, path);
|
||||
|
||||
expect(replicateUnattendedByEvent).toHaveBeenCalledOnce();
|
||||
expect(replicateUnattendedByEvent).toHaveBeenCalledWith({
|
||||
trigger: "merge",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
expect(queueCheckFor).toHaveBeenCalledWith(path);
|
||||
});
|
||||
});
|
||||
|
||||
describe("recurring replication scheduling precedence", () => {
|
||||
afterEach(() => {
|
||||
eventHub.offAll();
|
||||
});
|
||||
|
||||
function createRecurringSchedulingHarness() {
|
||||
let resolveContinuous!: (
|
||||
outcome: { status: "completed" } | { status: "blocked"; reason: "capability-not-applicable" }
|
||||
) => void;
|
||||
const startContinuous = vi.fn(
|
||||
() =>
|
||||
new Promise<{ status: "completed" } | { status: "blocked"; reason: "capability-not-applicable" }>(
|
||||
(resolve) => {
|
||||
resolveContinuous = resolve;
|
||||
}
|
||||
)
|
||||
);
|
||||
const API = createApi();
|
||||
const settings = {
|
||||
...DEFAULT_SETTINGS,
|
||||
isConfigured: true,
|
||||
liveSync: true,
|
||||
syncOnStart: true,
|
||||
periodicReplication: true,
|
||||
periodicReplicationInterval: 60,
|
||||
};
|
||||
const context = createReplicationSchedulingContext({
|
||||
isReady: vi.fn(() => true),
|
||||
isSuspended: vi.fn(() => false),
|
||||
currentSettings: vi.fn(() => settings),
|
||||
startContinuous,
|
||||
replicateUnattended: vi.fn(async () => ({ status: "completed" as const })),
|
||||
timer: {
|
||||
enable: (interval) => {
|
||||
API.setInterval(vi.fn(), interval);
|
||||
},
|
||||
disable: () => {
|
||||
API.clearInterval(0);
|
||||
},
|
||||
},
|
||||
log: vi.fn(),
|
||||
});
|
||||
|
||||
return {
|
||||
API,
|
||||
resolveContinuous: (
|
||||
outcome: { status: "completed" } | { status: "blocked"; reason: "capability-not-applicable" }
|
||||
) => resolveContinuous(outcome),
|
||||
resume: async () => {
|
||||
resumeReplicationScheduling(context);
|
||||
await Promise.resolve();
|
||||
},
|
||||
realiseSettings: async () => {
|
||||
realiseReplicationScheduling(context);
|
||||
await Promise.resolve();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
it("does not enable the generic periodic timer while Continuous owns recurring synchronisation", async () => {
|
||||
const harness = createRecurringSchedulingHarness();
|
||||
|
||||
await harness.resume();
|
||||
await harness.realiseSettings();
|
||||
|
||||
expect(harness.API.setInterval).not.toHaveBeenCalled();
|
||||
harness.resolveContinuous({ status: "completed" });
|
||||
await vi.waitFor(() => expect(harness.API.setInterval).not.toHaveBeenCalled());
|
||||
});
|
||||
|
||||
it("restores the generic periodic timer when Continuous is not applicable", async () => {
|
||||
const harness = createRecurringSchedulingHarness();
|
||||
|
||||
await harness.resume();
|
||||
await harness.realiseSettings();
|
||||
harness.resolveContinuous({ status: "blocked", reason: "capability-not-applicable" });
|
||||
|
||||
await vi.waitFor(() => expect(harness.API.setInterval).toHaveBeenCalledOnce());
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,194 @@
|
||||
import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import { LOG_LEVEL_INFO, LOG_LEVEL_NOTICE, Logger } from "octagonal-wheels/common/logger";
|
||||
import { skipIfDuplicated } from "octagonal-wheels/concurrency/lock";
|
||||
import { balanceChunkPurgedDBs, purgeUnreferencedChunks } from "@vrtmrz/livesync-commonlib/compat/pouchdb/chunks";
|
||||
import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator";
|
||||
import {
|
||||
CENTRAL_COMPATIBILITY_REJECTION_REASONS,
|
||||
type ReplicationFailureRequest,
|
||||
} from "@vrtmrz/livesync-commonlib/replication";
|
||||
import { $msg } from "@/common/translation";
|
||||
import { usesLegacyIndexedDBAdapter } from "@/common/compatibilitySettings";
|
||||
import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore";
|
||||
|
||||
type CentralCompatibilityRecoveryServices = Pick<
|
||||
LiveSyncBaseCore["services"],
|
||||
"API" | "appLifecycle" | "replicator" | "tweakValue"
|
||||
>;
|
||||
|
||||
interface CentralCompatibilityRecoveryContext {
|
||||
readonly confirm: LiveSyncBaseCore["confirm"];
|
||||
readonly localDatabase: LiveSyncBaseCore["localDatabase"];
|
||||
readonly rebuilder: LiveSyncBaseCore["rebuilder"];
|
||||
readonly services: CentralCompatibilityRecoveryServices;
|
||||
}
|
||||
|
||||
/**
|
||||
* Compose central compatibility recovery around the exact failed publication.
|
||||
* Remote mutations re-admit that publication and become no-ops after a
|
||||
* replacement; the failure result is never re-read from the current instance.
|
||||
*/
|
||||
export function createCentralCompatibilityRecovery(context: CentralCompatibilityRecoveryContext) {
|
||||
async function reconcileCleanedRemote(
|
||||
showMessage: boolean,
|
||||
setting: ObsidianLiveSyncSettings,
|
||||
expectedContext: ReplicationFailureRequest["context"]
|
||||
) {
|
||||
Logger("The remote database has been cleaned.", showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO);
|
||||
await skipIfDuplicated("cleanup", async () => {
|
||||
const count = await purgeUnreferencedChunks(context.localDatabase.localDatabase, true);
|
||||
const message = `The remote database has been cleaned up.
|
||||
To synchronize, this device must be also cleaned up. ${count} chunk(s) will be erased from this device.
|
||||
However, If there are many chunks to be deleted, maybe fetching again is faster.
|
||||
We will lose the history of this device if we fetch the remote database again.
|
||||
Even if you choose to clean up, you will see this option again if you exit Obsidian and then synchronise again.`;
|
||||
const CHOICE_FETCH = "Fetch again";
|
||||
const CHOICE_CLEAN = "Cleanup";
|
||||
const CHOICE_DISMISS = "Dismiss";
|
||||
const selected = await context.confirm.confirmWithMessage(
|
||||
"Cleaned",
|
||||
message,
|
||||
[CHOICE_FETCH, CHOICE_CLEAN, CHOICE_DISMISS],
|
||||
CHOICE_DISMISS,
|
||||
30
|
||||
);
|
||||
if (selected == CHOICE_FETCH) {
|
||||
await context.rebuilder.$performRebuildDB("localOnly");
|
||||
}
|
||||
if (selected != CHOICE_CLEAN) return;
|
||||
|
||||
await context.services.replicator.runBoundedRemoteActivity(
|
||||
() =>
|
||||
context.services.replicator.runWithActiveReplicatorContext(async (activeContext) => {
|
||||
if (activeContext !== expectedContext) return;
|
||||
const replicator = activeContext.replicator;
|
||||
if (!(replicator instanceof LiveSyncCouchDBReplicator)) return;
|
||||
const remoteDatabase = await replicator.connectRemoteCouchDBWithSetting(
|
||||
setting,
|
||||
context.services.API.isMobile(),
|
||||
true
|
||||
);
|
||||
if (typeof remoteDatabase == "string") {
|
||||
Logger(remoteDatabase, LOG_LEVEL_NOTICE);
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
await purgeUnreferencedChunks(context.localDatabase.localDatabase, false);
|
||||
context.localDatabase.clearCaches();
|
||||
const replicated = await context.services.replicator.runFiniteReplicationActivity(
|
||||
() => replicator.openOneShotReplication(setting, showMessage, false, "sync", true),
|
||||
{ label: "replication" }
|
||||
);
|
||||
if (replicated) {
|
||||
await balanceChunkPurgedDBs(context.localDatabase.localDatabase, remoteDatabase.db);
|
||||
await purgeUnreferencedChunks(context.localDatabase.localDatabase, false);
|
||||
context.localDatabase.clearCaches();
|
||||
await replicator.markRemoteResolved(setting);
|
||||
Logger(
|
||||
"The local database has been cleaned up.",
|
||||
showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO
|
||||
);
|
||||
} else {
|
||||
Logger(
|
||||
"Replication has been cancelled. Please try it again.",
|
||||
showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO
|
||||
);
|
||||
}
|
||||
} finally {
|
||||
await remoteDatabase.close();
|
||||
}
|
||||
}),
|
||||
{ label: "database-cleanup" }
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async function handleReplicationFailure(request: ReplicationFailureRequest): Promise<boolean> {
|
||||
const { context: failedContext, interaction, outcome, setting, showMessage } = request;
|
||||
if (!showMessage) {
|
||||
// Automatic requests may report the failure, but must not enter
|
||||
// tweak, lock, fetch, unlock, or cleanup dialogues.
|
||||
Logger("Replication failed on an unattended path.", LOG_LEVEL_INFO);
|
||||
return false;
|
||||
}
|
||||
if (interaction.kind !== "permitted" || !interaction.permissions.failureRecovery) return false;
|
||||
const recovery = outcome.recoveryHint;
|
||||
if (!recovery) return false;
|
||||
|
||||
if (
|
||||
recovery.reason === CENTRAL_COMPATIBILITY_REJECTION_REASONS.TWEAK_MISMATCH &&
|
||||
recovery.preferredTweakValue
|
||||
) {
|
||||
await context.services.tweakValue.askResolvingMismatched(
|
||||
recovery.preferredTweakValue,
|
||||
async (effectiveSetting) => {
|
||||
let updated = false;
|
||||
await context.services.replicator.runWithActiveReplicatorContext(async (activeContext) => {
|
||||
if (activeContext !== failedContext) return;
|
||||
const candidate = activeContext.replicator as typeof activeContext.replicator & {
|
||||
setPreferredRemoteTweakSettings?: (setting: ObsidianLiveSyncSettings) => Promise<void>;
|
||||
};
|
||||
if (typeof candidate.setPreferredRemoteTweakSettings !== "function") return;
|
||||
await candidate.setPreferredRemoteTweakSettings({ ...effectiveSetting });
|
||||
updated = true;
|
||||
});
|
||||
return updated;
|
||||
}
|
||||
);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (
|
||||
recovery.reason !== CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_LOCKED &&
|
||||
recovery.reason !== CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_CLEANED
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (
|
||||
recovery.reason === CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_CLEANED &&
|
||||
usesLegacyIndexedDBAdapter(setting)
|
||||
) {
|
||||
await reconcileCleanedRemote(showMessage, setting, failedContext);
|
||||
return false;
|
||||
}
|
||||
|
||||
const message = $msg("Replicator.Dialogue.Locked.Message");
|
||||
const CHOICE_FETCH = $msg("Replicator.Dialogue.Locked.Action.Fetch");
|
||||
const CHOICE_DISMISS = $msg("Replicator.Dialogue.Locked.Action.Dismiss");
|
||||
const CHOICE_UNLOCK = $msg("Replicator.Dialogue.Locked.Action.Unlock");
|
||||
const selected = await context.confirm.askSelectStringDialogue(
|
||||
message,
|
||||
[CHOICE_FETCH, CHOICE_UNLOCK, CHOICE_DISMISS],
|
||||
{
|
||||
title: $msg("Replicator.Dialogue.Locked.Title"),
|
||||
defaultAction: CHOICE_DISMISS,
|
||||
timeout: 60,
|
||||
}
|
||||
);
|
||||
if (selected == CHOICE_FETCH) {
|
||||
Logger($msg("Replicator.Dialogue.Locked.Message.Fetch"), LOG_LEVEL_NOTICE);
|
||||
await context.rebuilder.scheduleFetch();
|
||||
context.services.appLifecycle.scheduleRestart();
|
||||
return false;
|
||||
}
|
||||
if (selected != CHOICE_UNLOCK) return false;
|
||||
|
||||
let unlocked = false;
|
||||
await context.services.replicator.runWithActiveReplicatorContext(async (activeContext) => {
|
||||
if (activeContext !== failedContext) return;
|
||||
const replicator = activeContext.replicator as typeof activeContext.replicator & {
|
||||
markRemoteResolved(setting: ObsidianLiveSyncSettings): Promise<void>;
|
||||
};
|
||||
if (typeof replicator.markRemoteResolved !== "function") return;
|
||||
await replicator.markRemoteResolved(setting);
|
||||
unlocked = true;
|
||||
});
|
||||
if (unlocked) {
|
||||
Logger($msg("Replicator.Dialogue.Locked.Message.Unlocked"), LOG_LEVEL_NOTICE);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
return Object.freeze({ handleReplicationFailure, reconcileCleanedRemote });
|
||||
}
|
||||
@@ -0,0 +1,229 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import {
|
||||
CENTRAL_COMPATIBILITY_REJECTION_REASONS,
|
||||
NO_INTERACTION,
|
||||
USER_INITIATED_REPLICATION_AUTHORITY,
|
||||
replicationFailed,
|
||||
} from "@vrtmrz/livesync-commonlib/replication";
|
||||
|
||||
const chunkMocks = vi.hoisted(() => ({
|
||||
purgeUnreferencedChunks: vi.fn(async (_database: unknown, countOnly: boolean) => (countOnly ? 2 : 0)),
|
||||
balanceChunkPurgedDBs: vi.fn(async () => undefined),
|
||||
}));
|
||||
|
||||
vi.mock("@vrtmrz/livesync-commonlib/compat/pouchdb/chunks", () => chunkMocks);
|
||||
vi.mock("@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator", () => ({
|
||||
LiveSyncCouchDBReplicator: class {},
|
||||
}));
|
||||
|
||||
import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator";
|
||||
import { createCentralCompatibilityRecovery } from "./centralCompatibilityRecovery";
|
||||
|
||||
describe("central compatibility recovery", () => {
|
||||
it("uses the exact failed outcome and permits dialogue only with recovery authority", async () => {
|
||||
const askResolvingMismatched = vi.fn(async (..._arguments: unknown[]) => undefined);
|
||||
const failedSetPreferred = vi.fn(async (_setting: unknown) => undefined);
|
||||
const failedReplicator = { setPreferredRemoteTweakSettings: failedSetPreferred };
|
||||
const replacementSetPreferred = vi.fn(async (_setting: unknown) => undefined);
|
||||
const replacementReplicator = {
|
||||
tweakSettingsMismatched: true,
|
||||
preferredTweakValue: { customChunkSize: 99 },
|
||||
setPreferredRemoteTweakSettings: replacementSetPreferred,
|
||||
};
|
||||
const failedContext = { provider: {}, replicator: failedReplicator };
|
||||
const replacementContext = { provider: {}, replicator: replacementReplicator };
|
||||
const preferredTweakValue = { customChunkSize: 60 };
|
||||
const outcome = replicationFailed(new Error("mismatched"), {
|
||||
reason: CENTRAL_COMPATIBILITY_REJECTION_REASONS.TWEAK_MISMATCH,
|
||||
preferredTweakValue,
|
||||
});
|
||||
const recovery = createCentralCompatibilityRecovery({
|
||||
confirm: {},
|
||||
localDatabase: {},
|
||||
rebuilder: {},
|
||||
services: {
|
||||
appLifecycle: {},
|
||||
API: {},
|
||||
replicator: {
|
||||
runWithActiveReplicatorContext: vi.fn(async (task: (context: unknown) => unknown) =>
|
||||
task(replacementContext)
|
||||
),
|
||||
},
|
||||
tweakValue: { askResolvingMismatched },
|
||||
},
|
||||
} as never);
|
||||
|
||||
await recovery.handleReplicationFailure({
|
||||
context: failedContext,
|
||||
setting: {},
|
||||
outcome,
|
||||
showMessage: false,
|
||||
interaction: NO_INTERACTION,
|
||||
} as never);
|
||||
expect(askResolvingMismatched).not.toHaveBeenCalled();
|
||||
|
||||
await recovery.handleReplicationFailure({
|
||||
context: failedContext,
|
||||
setting: {},
|
||||
outcome,
|
||||
showMessage: false,
|
||||
interaction: {
|
||||
kind: "permitted",
|
||||
permissions: { ...USER_INITIATED_REPLICATION_AUTHORITY.permissions, failureRecovery: false },
|
||||
},
|
||||
} as never);
|
||||
expect(askResolvingMismatched).not.toHaveBeenCalled();
|
||||
|
||||
await recovery.handleReplicationFailure({
|
||||
context: failedContext,
|
||||
setting: {},
|
||||
outcome,
|
||||
showMessage: true,
|
||||
interaction: USER_INITIATED_REPLICATION_AUTHORITY,
|
||||
} as never);
|
||||
expect(askResolvingMismatched).toHaveBeenCalledWith(preferredTweakValue, expect.any(Function));
|
||||
const updatePreferredRemote = askResolvingMismatched.mock.calls[0][1] as (
|
||||
setting: Record<string, unknown>
|
||||
) => Promise<boolean>;
|
||||
await expect(updatePreferredRemote({ customChunkSize: 64 })).resolves.toBe(false);
|
||||
expect(failedSetPreferred).not.toHaveBeenCalled();
|
||||
expect(replacementSetPreferred).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("writes a mismatch decision only through the still-active failed publication", async () => {
|
||||
const setPreferredRemoteTweakSettings = vi.fn(async (_setting: unknown) => undefined);
|
||||
const failedContext = { provider: {}, replicator: { setPreferredRemoteTweakSettings } };
|
||||
let updatePreferredRemote: ((setting: Record<string, unknown>) => Promise<boolean>) | undefined;
|
||||
const askResolvingMismatched = vi.fn(
|
||||
async (_preferred: unknown, update: (setting: Record<string, unknown>) => Promise<boolean>) => {
|
||||
updatePreferredRemote = update;
|
||||
}
|
||||
);
|
||||
const recovery = createCentralCompatibilityRecovery({
|
||||
confirm: {},
|
||||
localDatabase: {},
|
||||
rebuilder: {},
|
||||
services: {
|
||||
appLifecycle: {},
|
||||
API: {},
|
||||
replicator: {
|
||||
runWithActiveReplicatorContext: vi.fn(async (task: (context: unknown) => unknown) =>
|
||||
task(failedContext)
|
||||
),
|
||||
},
|
||||
tweakValue: { askResolvingMismatched },
|
||||
},
|
||||
} as never);
|
||||
|
||||
await recovery.handleReplicationFailure({
|
||||
context: failedContext,
|
||||
setting: {},
|
||||
outcome: replicationFailed(new Error("mismatched"), {
|
||||
reason: CENTRAL_COMPATIBILITY_REJECTION_REASONS.TWEAK_MISMATCH,
|
||||
preferredTweakValue: { customChunkSize: 60 },
|
||||
}),
|
||||
showMessage: true,
|
||||
interaction: USER_INITIATED_REPLICATION_AUTHORITY,
|
||||
} as never);
|
||||
|
||||
const effectiveSetting = { customChunkSize: 64 };
|
||||
await expect(updatePreferredRemote?.(effectiveSetting)).resolves.toBe(true);
|
||||
expect(setPreferredRemoteTweakSettings).toHaveBeenCalledWith(effectiveSetting);
|
||||
expect(setPreferredRemoteTweakSettings.mock.calls[0][0]).not.toBe(effectiveSetting);
|
||||
});
|
||||
|
||||
it("does not apply an unlock selected for a replaced failed publication", async () => {
|
||||
const failedMarkResolved = vi.fn(async () => undefined);
|
||||
const replacementMarkResolved = vi.fn(async () => undefined);
|
||||
const failedContext = { provider: {}, replicator: { markRemoteResolved: failedMarkResolved } };
|
||||
const replacementContext = { provider: {}, replicator: { markRemoteResolved: replacementMarkResolved } };
|
||||
const runWithActiveReplicatorContext = vi.fn(async (task: (context: unknown) => unknown) =>
|
||||
task(replacementContext)
|
||||
);
|
||||
const recovery = createCentralCompatibilityRecovery({
|
||||
confirm: {
|
||||
askSelectStringDialogue: vi.fn(async (_message: string, choices: string[]) => choices[1]),
|
||||
},
|
||||
localDatabase: {},
|
||||
rebuilder: {},
|
||||
services: {
|
||||
appLifecycle: { scheduleRestart: vi.fn() },
|
||||
API: {},
|
||||
replicator: { runWithActiveReplicatorContext },
|
||||
tweakValue: {},
|
||||
},
|
||||
} as never);
|
||||
|
||||
await recovery.handleReplicationFailure({
|
||||
context: failedContext,
|
||||
setting: {},
|
||||
outcome: replicationFailed(new Error("locked"), {
|
||||
reason: CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_LOCKED,
|
||||
}),
|
||||
showMessage: true,
|
||||
interaction: USER_INITIATED_REPLICATION_AUTHORITY,
|
||||
} as never);
|
||||
|
||||
expect(runWithActiveReplicatorContext).toHaveBeenCalledOnce();
|
||||
expect(failedMarkResolved).not.toHaveBeenCalled();
|
||||
expect(replacementMarkResolved).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps cleaned-remote replication and balancing inside the shared activity boundary", async () => {
|
||||
const activityFinished = vi.fn();
|
||||
const runBoundedRemoteActivity = vi.fn(async (task: () => unknown) => {
|
||||
try {
|
||||
return await task();
|
||||
} finally {
|
||||
activityFinished();
|
||||
}
|
||||
});
|
||||
const runFiniteReplicationActivity = vi.fn(async (task: () => unknown) => await task());
|
||||
const openOneShotReplication = vi.fn(async () => true);
|
||||
const remoteDatabase = { close: vi.fn(async () => undefined) };
|
||||
const close = vi.fn(async () => undefined);
|
||||
const activeReplicator = Object.assign(new LiveSyncCouchDBReplicator({} as never), {
|
||||
connectRemoteCouchDBWithSetting: vi.fn(async () => ({ db: remoteDatabase, close })),
|
||||
openOneShotReplication,
|
||||
markRemoteResolved: vi.fn(async () => undefined),
|
||||
});
|
||||
const expectedContext = { provider: {}, replicator: activeReplicator };
|
||||
const runWithActiveReplicatorContext = vi.fn(async (task: (context: unknown) => unknown) =>
|
||||
task(expectedContext)
|
||||
);
|
||||
const localDatabase = { localDatabase: {}, clearCaches: vi.fn() };
|
||||
const recovery = createCentralCompatibilityRecovery({
|
||||
confirm: { confirmWithMessage: vi.fn(async () => "Cleanup") },
|
||||
localDatabase,
|
||||
rebuilder: {},
|
||||
services: {
|
||||
appLifecycle: {},
|
||||
API: { isMobile: vi.fn(() => false) },
|
||||
replicator: {
|
||||
runBoundedRemoteActivity,
|
||||
runFiniteReplicationActivity,
|
||||
runWithActiveReplicatorContext,
|
||||
},
|
||||
tweakValue: {},
|
||||
},
|
||||
} as never);
|
||||
|
||||
await recovery.reconcileCleanedRemote(true, {} as ObsidianLiveSyncSettings, expectedContext as never);
|
||||
|
||||
expect(runBoundedRemoteActivity).toHaveBeenCalledWith(expect.any(Function), {
|
||||
label: "database-cleanup",
|
||||
});
|
||||
expect(runFiniteReplicationActivity).toHaveBeenCalledWith(expect.any(Function), {
|
||||
label: "replication",
|
||||
});
|
||||
expect(runWithActiveReplicatorContext).toHaveBeenCalledOnce();
|
||||
expect(openOneShotReplication).toHaveBeenCalledOnce();
|
||||
expect(openOneShotReplication.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
activityFinished.mock.invocationCallOrder[0]
|
||||
);
|
||||
expect(chunkMocks.balanceChunkPurgedDBs).toHaveBeenCalledOnce();
|
||||
expect(close).toHaveBeenCalledOnce();
|
||||
expect(close.mock.invocationCallOrder[0]).toBeLessThan(activityFinished.mock.invocationCallOrder[0]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,110 @@
|
||||
import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication";
|
||||
import { clearHandlers } from "@vrtmrz/livesync-commonlib/compat/replication/SyncParamsHandler";
|
||||
import { UnresolvedErrorManager } from "@vrtmrz/livesync-commonlib/compat/services/base/UnresolvedErrorManager";
|
||||
import type { ServiceContext } from "@vrtmrz/livesync-commonlib/context";
|
||||
import { fireAndForget } from "octagonal-wheels/promises";
|
||||
import type { IMinimumLiveSyncCommands, LiveSyncBaseCore } from "@/LiveSyncBaseCore";
|
||||
import { createAutomaticReplicationTriggers } from "./automaticTriggers";
|
||||
import { createCentralCompatibilityRecovery } from "./centralCompatibilityRecovery";
|
||||
import { createOnlineReplicationPreflight, createSecuritySeedPreflight } from "./preflight";
|
||||
import { ReplicateResultProcessor } from "./ReplicateResultProcessor";
|
||||
|
||||
type LocalApplicationActivityOwner = {
|
||||
runBoundedLocalApplicationActivity<T>(task: () => T | PromiseLike<T>, options?: { label?: string }): Promise<T>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Compose result application, automatic triggers, preflight, and central
|
||||
* compatibility recovery around the existing typed Services.
|
||||
*
|
||||
* Registration order is observable for equal-priority handlers. The host must
|
||||
* call this after host serviceFeatures and add-ons are composed, but before
|
||||
* legacy modules are bound. This preserves the former ModuleReplicator
|
||||
* lifecycle-handler order without retaining a public module identity.
|
||||
*/
|
||||
export function useReplicationFeature<TContext extends ServiceContext, TCommands extends IMinimumLiveSyncCommands>(
|
||||
core: LiveSyncBaseCore<TContext, TCommands>
|
||||
): void {
|
||||
const { services } = core;
|
||||
// Obsidian adds an application-activity owner to its ReplicatorService.
|
||||
// Generic hosts retain the former direct-execution fallback.
|
||||
const localApplicationActivityOwner = services.replicator as typeof services.replicator &
|
||||
Partial<LocalApplicationActivityOwner>;
|
||||
const resultProcessor = new ReplicateResultProcessor({
|
||||
currentSettings: () => services.setting.currentSettings(),
|
||||
keyValueDB: services.keyValueDB.kvDB,
|
||||
localDatabase: core.localDatabase,
|
||||
requestActiveReplicatorRetirement: () => {
|
||||
// Do not await a retirement transition from result application: it
|
||||
// may be draining the replication work which delivered this item.
|
||||
fireAndForget(() => services.replicator.onCloseActiveReplication());
|
||||
},
|
||||
runLocalApplicationActivity: async (task, options) =>
|
||||
localApplicationActivityOwner.runBoundedLocalApplicationActivity
|
||||
? await localApplicationActivityOwner.runBoundedLocalApplicationActivity(task, options)
|
||||
: await task(),
|
||||
services: {
|
||||
appLifecycle: services.appLifecycle,
|
||||
path: services.path,
|
||||
replication: services.replication,
|
||||
vault: services.vault,
|
||||
},
|
||||
});
|
||||
const unresolvedErrorManager = new UnresolvedErrorManager(services.appLifecycle, services.context.events);
|
||||
const initialiseAutomaticReplicationTriggers = createAutomaticReplicationTriggers({
|
||||
currentSettings: () => services.setting.currentSettings(),
|
||||
isSuspended: () => services.appLifecycle.isSuspended(),
|
||||
replicateDatabaseEvent: () =>
|
||||
services.replication.replicateUnattendedByEvent({
|
||||
trigger: "database-event",
|
||||
interaction: NO_INTERACTION,
|
||||
}),
|
||||
reprocessStoredDocuments: () => resultProcessor.reprocessStoredDocuments(),
|
||||
resumeResultApplication: () => resultProcessor.resume(),
|
||||
suspendResultApplication: () => resultProcessor.suspend(),
|
||||
});
|
||||
const preflightContext = {
|
||||
services: {
|
||||
API: services.API,
|
||||
replicator: services.replicator,
|
||||
setting: services.setting,
|
||||
},
|
||||
};
|
||||
const onlinePreflight = createOnlineReplicationPreflight(unresolvedErrorManager, preflightContext);
|
||||
const securitySeedPreflight = createSecuritySeedPreflight(unresolvedErrorManager, preflightContext);
|
||||
const centralCompatibilityRecovery = createCentralCompatibilityRecovery({
|
||||
confirm: core.confirm,
|
||||
localDatabase: core.localDatabase,
|
||||
rebuilder: core.rebuilder,
|
||||
services: {
|
||||
API: services.API,
|
||||
appLifecycle: services.appLifecycle,
|
||||
replicator: services.replicator,
|
||||
tweakValue: services.tweakValue,
|
||||
},
|
||||
});
|
||||
|
||||
services.replicator.onBeforeReplicatorPublication.addHandler(() => {
|
||||
// Key-derivation handlers belong to the candidate which is about to
|
||||
// become active; discard callbacks retained by the previous owner.
|
||||
clearHandlers();
|
||||
return Promise.resolve(true);
|
||||
});
|
||||
services.databaseEvents.onDatabaseInitialised.addHandler(() => {
|
||||
fireAndForget(() => resultProcessor.restoreFromSnapshotOnce());
|
||||
return Promise.resolve(true);
|
||||
});
|
||||
services.appLifecycle.onSettingLoaded.addHandler(initialiseAutomaticReplicationTriggers);
|
||||
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();
|
||||
unresolvedErrorManager.clearErrors();
|
||||
return true;
|
||||
}, 100);
|
||||
services.replication.onReplicationFailed.addHandler(centralCompatibilityRecovery.handleReplicationFailure);
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
import { $msg } from "@/common/translation";
|
||||
import { withOwnedRemoteResource } from "@/common/ownedRemoteResource";
|
||||
import { LOG_LEVEL_INFO, LOG_LEVEL_NOTICE, LOG_LEVEL_VERBOSE, Logger } from "octagonal-wheels/common/logger";
|
||||
import { MARK_LOG_NETWORK_ERROR } from "@vrtmrz/livesync-commonlib/compat/services/lib/logUtils";
|
||||
import { UnresolvedErrorManager } from "@vrtmrz/livesync-commonlib/compat/services/base/UnresolvedErrorManager";
|
||||
import { REMOTE_RESOURCE_KINDS } from "@vrtmrz/livesync-commonlib/replication";
|
||||
import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore";
|
||||
|
||||
type ReplicationPreflightServices = Pick<LiveSyncBaseCore["services"], "API" | "replicator" | "setting">;
|
||||
|
||||
interface ReplicationPreflightContext {
|
||||
readonly services: ReplicationPreflightServices;
|
||||
}
|
||||
|
||||
/** Return the generic online preflight without inspecting a provider kind. */
|
||||
export function createOnlineReplicationPreflight(
|
||||
errorManager: UnresolvedErrorManager,
|
||||
context: ReplicationPreflightContext
|
||||
) {
|
||||
return function isOnlineAndCanReplicate(showMessage: boolean): Promise<boolean> {
|
||||
const errorMessage = "Network is offline";
|
||||
if (!context.services.API.isOnline) {
|
||||
errorManager.showError(errorMessage, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO);
|
||||
return Promise.resolve(false);
|
||||
}
|
||||
errorManager.clearError(errorMessage);
|
||||
return Promise.resolve(true);
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the central-remote Security Seed preflight. The acquired resource is
|
||||
* owned only for this read and is disposed before the handler settles.
|
||||
*/
|
||||
export function createSecuritySeedPreflight(
|
||||
errorManager: UnresolvedErrorManager,
|
||||
context: ReplicationPreflightContext
|
||||
) {
|
||||
return async function canReplicateWithSecuritySeed(showMessage: boolean): Promise<boolean> {
|
||||
const currentSettings = context.services.setting.currentSettings();
|
||||
const errorMessage = $msg("Replicator.Message.InitialiseFatalError");
|
||||
// This is a fatal preparation error, so the non-interactive path still
|
||||
// records it while choosing a quieter log level.
|
||||
const ensureMessage = `${MARK_LOG_NETWORK_ERROR}Failed to initialise the encryption key, preventing replication.`;
|
||||
try {
|
||||
const resource = await context.services.replicator.createRemoteResource(
|
||||
REMOTE_RESOURCE_KINDS.SECURITY_SEED,
|
||||
currentSettings
|
||||
);
|
||||
if (!resource) {
|
||||
errorManager.showError(errorMessage, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO);
|
||||
return false;
|
||||
}
|
||||
errorManager.clearError(errorMessage);
|
||||
const seed = await withOwnedRemoteResource(resource, (ownedResource) => ownedResource.read());
|
||||
if (seed.length == 0) throw new Error("PBKDF2 salt (Security Seed) is empty");
|
||||
} catch (error) {
|
||||
Logger(error, LOG_LEVEL_VERBOSE);
|
||||
errorManager.showError(ensureMessage, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO);
|
||||
return false;
|
||||
}
|
||||
errorManager.clearError(ensureMessage);
|
||||
return true;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { createServiceContext } from "@vrtmrz/livesync-commonlib/context";
|
||||
import { VER, type EntryDoc } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import { promiseWithResolvers } from "octagonal-wheels/promises";
|
||||
import { useReplicationFeature } from "./index";
|
||||
|
||||
type BooleanHandler = (showMessage: boolean) => Promise<boolean>;
|
||||
type ParseHandler = (documents: PouchDB.Core.ExistingDocument<EntryDoc>[]) => Promise<boolean>;
|
||||
|
||||
function setup(onCloseActiveReplication = vi.fn(async () => true)) {
|
||||
const read = vi.fn(async () => new Uint8Array([1]));
|
||||
const dispose = vi.fn(async () => undefined);
|
||||
const createRemoteResource = vi.fn(async () => ({ read, dispose }));
|
||||
const beforeReplicateHandlers = new Map<number, BooleanHandler>();
|
||||
const centralRemoteHandlers: BooleanHandler[] = [];
|
||||
let parseHandler: ParseHandler | undefined;
|
||||
const services = {
|
||||
API: { isMobile: vi.fn(() => false), isOnline: true },
|
||||
appLifecycle: {
|
||||
getUnresolvedMessages: { addHandler: vi.fn() },
|
||||
isReady: true,
|
||||
isSuspended: vi.fn(() => false),
|
||||
onSettingLoaded: { addHandler: vi.fn() },
|
||||
},
|
||||
context: createServiceContext(),
|
||||
databaseEvents: { onDatabaseInitialised: { addHandler: vi.fn() } },
|
||||
keyValueDB: {
|
||||
kvDB: {
|
||||
get: vi.fn(async () => undefined),
|
||||
set: vi.fn(async () => undefined),
|
||||
},
|
||||
},
|
||||
path: { getPath: vi.fn((entry: { path: string }) => entry.path) },
|
||||
replication: {
|
||||
onBeforeReplicate: {
|
||||
addHandler: vi.fn((handler: BooleanHandler, priority = 0) => {
|
||||
beforeReplicateHandlers.set(priority, handler);
|
||||
}),
|
||||
},
|
||||
onPrepareCentralRemoteReplication: {
|
||||
addHandler: vi.fn((handler: BooleanHandler) => centralRemoteHandlers.push(handler)),
|
||||
},
|
||||
onReplicationFailed: { addHandler: vi.fn() },
|
||||
parseSynchroniseResult: {
|
||||
addHandler: vi.fn((handler: ParseHandler) => {
|
||||
parseHandler = handler;
|
||||
}),
|
||||
},
|
||||
replicateUnattendedByEvent: vi.fn(async () => ({ status: "completed" as const })),
|
||||
},
|
||||
replicator: {
|
||||
createRemoteResource,
|
||||
onBeforeReplicatorPublication: { addHandler: vi.fn() },
|
||||
onCloseActiveReplication,
|
||||
},
|
||||
setting: { currentSettings: vi.fn(() => ({})) },
|
||||
tweakValue: {},
|
||||
vault: {},
|
||||
};
|
||||
const core = {
|
||||
confirm: {},
|
||||
localDatabase: {},
|
||||
rebuilder: {},
|
||||
services,
|
||||
};
|
||||
|
||||
useReplicationFeature(core as never);
|
||||
|
||||
return {
|
||||
beforeReplicateHandlers,
|
||||
centralRemoteHandlers,
|
||||
createRemoteResource,
|
||||
dispose,
|
||||
get parseHandler() {
|
||||
return parseHandler;
|
||||
},
|
||||
onCloseActiveReplication,
|
||||
read,
|
||||
};
|
||||
}
|
||||
|
||||
describe("replication serviceFeature composition", () => {
|
||||
it("refreshes and disposes the remote Security Seed before central replication", async () => {
|
||||
const { centralRemoteHandlers, createRemoteResource, dispose, read } = setup();
|
||||
|
||||
await expect(centralRemoteHandlers[0](false)).resolves.toBe(true);
|
||||
|
||||
expect(createRemoteResource).toHaveBeenCalledWith("security-seed", {});
|
||||
expect(read).toHaveBeenCalledOnce();
|
||||
expect(dispose).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("keeps generic preflight separate from central-remote preparation", async () => {
|
||||
const { beforeReplicateHandlers, centralRemoteHandlers, createRemoteResource } = setup();
|
||||
const online = beforeReplicateHandlers.get(10);
|
||||
const general = beforeReplicateHandlers.get(100);
|
||||
|
||||
expect(online).toBeDefined();
|
||||
expect(general).toBeDefined();
|
||||
expect(centralRemoteHandlers).toHaveLength(1);
|
||||
await expect(online!(false)).resolves.toBe(true);
|
||||
await expect(general!(false)).resolves.toBe(true);
|
||||
expect(createRemoteResource).not.toHaveBeenCalled();
|
||||
|
||||
await expect(centralRemoteHandlers[0](false)).resolves.toBe(true);
|
||||
expect(createRemoteResource).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("requests owner retirement without awaiting the transition from result application", async () => {
|
||||
const retirement = promiseWithResolvers<boolean>();
|
||||
const onCloseActiveReplication = vi.fn(() => retirement.promise);
|
||||
const harness = setup(onCloseActiveReplication);
|
||||
const versionInfo = {
|
||||
_id: "versioninfo",
|
||||
_rev: "1-test",
|
||||
type: "versioninfo",
|
||||
version: VER + 1,
|
||||
} as unknown as PouchDB.Core.ExistingDocument<EntryDoc>;
|
||||
|
||||
expect(harness.parseHandler).toBeDefined();
|
||||
await expect(harness.parseHandler!([versionInfo])).resolves.toBe(true);
|
||||
expect(onCloseActiveReplication).toHaveBeenCalledOnce();
|
||||
|
||||
retirement.resolve(true);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user