mirror of
https://github.com/vrtmrz/obsidian-livesync.git
synced 2026-09-21 18:17:05 +00:00
Integrate E2EE rebuild preservation with current main
This commit is contained in:
vorotamoroz
@@ -0,0 +1,579 @@
|
||||
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.
|
||||
*
|
||||
* Runtime databases are deliberately obtained through operation-time
|
||||
* accessors. Feature composition precedes their initialisation, and database
|
||||
* reset may replace their backing instances, so retaining an earlier concrete
|
||||
* database would be invalid.
|
||||
*/
|
||||
interface ReplicateResultProcessorContext {
|
||||
readonly currentSettings: () => ReplicateResultProcessorSettings;
|
||||
readonly getKeyValueDB: () => LiveSyncBaseCore["kvDB"];
|
||||
readonly getLocalDatabase: () => 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.getLocalDatabase();
|
||||
}
|
||||
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.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.
|
||||
*/
|
||||
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
|
||||
.getKeyValueDB()
|
||||
.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,184 @@
|
||||
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 = {
|
||||
applicationReady?: boolean;
|
||||
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 isReady = vi.fn(() => options.applicationReady ?? true);
|
||||
const core = {
|
||||
services: {
|
||||
appLifecycle: { isReady, 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 }),
|
||||
getKeyValueDB: () => core.kvDB,
|
||||
getLocalDatabase: () => core.localDatabase,
|
||||
requestActiveReplicatorRetirement: () => {
|
||||
void onCloseActiveReplication();
|
||||
},
|
||||
runLocalApplicationActivity: runBoundedLocalApplicationActivity,
|
||||
services: core.services,
|
||||
} as never);
|
||||
return {
|
||||
isReady,
|
||||
onCloseActiveReplication,
|
||||
processor,
|
||||
processSynchroniseResult,
|
||||
runBoundedLocalApplicationActivity,
|
||||
};
|
||||
}
|
||||
|
||||
describe("ReplicateResultProcessor", () => {
|
||||
it("suspends result application while the application is not ready", () => {
|
||||
const { isReady, processor } = setup({ applicationReady: false });
|
||||
|
||||
expect(processor.isSuspended).toBe(true);
|
||||
expect(isReady).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
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 getLocalDatabase = vi.fn(() => ({ findAllNormalDocs }));
|
||||
const processor = new ReplicateResultProcessor({
|
||||
getLocalDatabase,
|
||||
} as never);
|
||||
const enqueueAll = vi.spyOn(processor, "enqueueAll").mockImplementation(() => undefined);
|
||||
|
||||
await expect(processor.reprocessStoredDocuments()).resolves.toBe(2);
|
||||
|
||||
expect(findAllNormalDocs).toHaveBeenCalledOnce();
|
||||
expect(getLocalDatabase).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,213 @@
|
||||
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,
|
||||
REPLICATION_PROGRESS_PRESENTATIONS,
|
||||
type ReplicatorInstance,
|
||||
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"
|
||||
>;
|
||||
|
||||
/** Collaborators for applying a compatibility decision to its failed publication. */
|
||||
interface CentralCompatibilityRecoveryContext {
|
||||
readonly confirm: LiveSyncBaseCore["confirm"];
|
||||
/** Obtain the database only when recovery runs, after initialisation or reset. */
|
||||
readonly getLocalDatabase: () => LiveSyncBaseCore["localDatabase"];
|
||||
readonly rebuilder: LiveSyncBaseCore["rebuilder"];
|
||||
readonly services: CentralCompatibilityRecoveryServices;
|
||||
}
|
||||
|
||||
interface PreferredRemoteTweakWriter extends ReplicatorInstance {
|
||||
setPreferredRemoteTweakSettings(setting: ObsidianLiveSyncSettings): Promise<void>;
|
||||
}
|
||||
|
||||
interface ResolvedRemoteWriter extends ReplicatorInstance {
|
||||
markRemoteResolved(setting: ObsidianLiveSyncSettings): Promise<void>;
|
||||
}
|
||||
|
||||
function canSetPreferredRemoteTweakSettings(replicator: ReplicatorInstance): replicator is PreferredRemoteTweakWriter {
|
||||
return (
|
||||
"setPreferredRemoteTweakSettings" in replicator &&
|
||||
typeof replicator.setPreferredRemoteTweakSettings === "function"
|
||||
);
|
||||
}
|
||||
|
||||
function canMarkRemoteResolved(replicator: ReplicatorInstance): replicator is ResolvedRemoteWriter {
|
||||
return "markRemoteResolved" in replicator && typeof replicator.markRemoteResolved === "function";
|
||||
}
|
||||
|
||||
/**
|
||||
* 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(
|
||||
showProgress: boolean,
|
||||
setting: ObsidianLiveSyncSettings,
|
||||
expectedContext: ReplicationFailureRequest["context"]
|
||||
) {
|
||||
Logger("The remote database has been cleaned.", showProgress ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO);
|
||||
await skipIfDuplicated("cleanup", async () => {
|
||||
const count = await purgeUnreferencedChunks(context.getLocalDatabase().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 localDatabase = context.getLocalDatabase();
|
||||
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(localDatabase.localDatabase, false);
|
||||
localDatabase.clearCaches();
|
||||
const replicated = await context.services.replicator.runFiniteReplicationActivity(
|
||||
() => replicator.openOneShotReplication(setting, showProgress, false, "sync", true),
|
||||
{ label: "replication" }
|
||||
);
|
||||
if (replicated) {
|
||||
await balanceChunkPurgedDBs(localDatabase.localDatabase, remoteDatabase.db);
|
||||
await purgeUnreferencedChunks(localDatabase.localDatabase, false);
|
||||
localDatabase.clearCaches();
|
||||
await replicator.markRemoteResolved(setting);
|
||||
Logger(
|
||||
"The local database has been cleaned up.",
|
||||
showProgress ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO
|
||||
);
|
||||
} else {
|
||||
Logger(
|
||||
"Replication has been cancelled. Please try it again.",
|
||||
showProgress ? 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, progressPresentation, setting } = request;
|
||||
const showProgress = progressPresentation === REPLICATION_PROGRESS_PRESENTATIONS.NOTICE;
|
||||
if (interaction.kind === "forbidden") {
|
||||
// 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.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;
|
||||
if (!canSetPreferredRemoteTweakSettings(activeContext.replicator)) return;
|
||||
await activeContext.replicator.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(showProgress, 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;
|
||||
if (!canMarkRemoteResolved(activeContext.replicator)) return;
|
||||
await activeContext.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,267 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import { defaultLogger, LOG_LEVEL_INFO, LOG_LEVEL_NOTICE, setGlobalLogFunction } from "octagonal-wheels/common/logger";
|
||||
import {
|
||||
CENTRAL_COMPATIBILITY_REJECTION_REASONS,
|
||||
NO_INTERACTION,
|
||||
REPLICATION_PROGRESS_PRESENTATIONS,
|
||||
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("characterises unattended central failure handling as one INFO log without a NOTICE", async () => {
|
||||
const log = vi.fn((_message: unknown, _level?: number, _key?: string) => undefined);
|
||||
setGlobalLogFunction(log);
|
||||
try {
|
||||
const recovery = createCentralCompatibilityRecovery({
|
||||
confirm: {},
|
||||
getLocalDatabase: () => ({}),
|
||||
rebuilder: {},
|
||||
services: {
|
||||
appLifecycle: {},
|
||||
API: {},
|
||||
replicator: {},
|
||||
tweakValue: {},
|
||||
},
|
||||
} as never);
|
||||
|
||||
await expect(
|
||||
recovery.handleReplicationFailure({
|
||||
context: { provider: {}, replicator: {} },
|
||||
setting: {},
|
||||
outcome: replicationFailed(new Error("provider failed")),
|
||||
progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.QUIET,
|
||||
interaction: NO_INTERACTION,
|
||||
} as never)
|
||||
).resolves.toBe(false);
|
||||
|
||||
expect(log).toHaveBeenCalledOnce();
|
||||
expect(log).toHaveBeenCalledWith("Replication failed on an unattended path.", LOG_LEVEL_INFO, undefined);
|
||||
expect(log.mock.calls.map(([, level]) => level)).not.toContain(LOG_LEVEL_NOTICE);
|
||||
} finally {
|
||||
setGlobalLogFunction(defaultLogger);
|
||||
}
|
||||
});
|
||||
|
||||
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: {},
|
||||
getLocalDatabase: () => ({}),
|
||||
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,
|
||||
progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.QUIET,
|
||||
interaction: NO_INTERACTION,
|
||||
} as never);
|
||||
expect(askResolvingMismatched).not.toHaveBeenCalled();
|
||||
|
||||
await recovery.handleReplicationFailure({
|
||||
context: failedContext,
|
||||
setting: {},
|
||||
outcome,
|
||||
progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.QUIET,
|
||||
interaction: {
|
||||
kind: "permitted",
|
||||
permissions: { ...USER_INITIATED_REPLICATION_AUTHORITY.permissions, failureRecovery: false },
|
||||
},
|
||||
} as never);
|
||||
expect(askResolvingMismatched).not.toHaveBeenCalled();
|
||||
|
||||
await recovery.handleReplicationFailure({
|
||||
context: failedContext,
|
||||
setting: {},
|
||||
outcome,
|
||||
progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.QUIET,
|
||||
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: {},
|
||||
getLocalDatabase: () => ({}),
|
||||
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 },
|
||||
}),
|
||||
progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.NOTICE,
|
||||
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]),
|
||||
},
|
||||
getLocalDatabase: () => ({}),
|
||||
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,
|
||||
}),
|
||||
progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.NOTICE,
|
||||
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 getLocalDatabase = vi.fn(() => localDatabase);
|
||||
const recovery = createCentralCompatibilityRecovery({
|
||||
confirm: { confirmWithMessage: vi.fn(async () => "Cleanup") },
|
||||
getLocalDatabase,
|
||||
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(getLocalDatabase).toHaveBeenCalledTimes(2);
|
||||
expect(close).toHaveBeenCalledOnce();
|
||||
expect(close.mock.invocationCallOrder[0]).toBeLessThan(activityFinished.mock.invocationCallOrder[0]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,116 @@
|
||||
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>;
|
||||
};
|
||||
|
||||
function ownsLocalApplicationActivity(value: object): value is LocalApplicationActivityOwner {
|
||||
return (
|
||||
"runBoundedLocalApplicationActivity" in value && typeof value.runBoundedLocalApplicationActivity === "function"
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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, so lifecycle handlers observe the required order.
|
||||
*/
|
||||
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 = ownsLocalApplicationActivity(services.replicator)
|
||||
? services.replicator
|
||||
: undefined;
|
||||
const resultProcessor = new ReplicateResultProcessor({
|
||||
currentSettings: () => services.setting.currentSettings(),
|
||||
getKeyValueDB: () => services.keyValueDB.kvDB,
|
||||
getLocalDatabase: () => 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
|
||||
? 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,
|
||||
getLocalDatabase: () => 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,180 @@
|
||||
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>;
|
||||
type KeyValueDBFixture = {
|
||||
readonly kvDB: {
|
||||
get: (key: string) => Promise<unknown>;
|
||||
set: (key: string, value: unknown) => Promise<unknown>;
|
||||
};
|
||||
};
|
||||
type SetupOptions = {
|
||||
readonly getLocalDatabase?: () => object;
|
||||
readonly keyValueDB?: KeyValueDBFixture;
|
||||
readonly onCloseActiveReplication?: () => Promise<boolean>;
|
||||
};
|
||||
|
||||
function setup(options: SetupOptions = {}) {
|
||||
const {
|
||||
getLocalDatabase = () => ({}),
|
||||
keyValueDB = {
|
||||
kvDB: {
|
||||
get: vi.fn(async () => undefined),
|
||||
set: vi.fn(async () => undefined),
|
||||
},
|
||||
},
|
||||
onCloseActiveReplication = vi.fn(async () => true),
|
||||
} = options;
|
||||
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,
|
||||
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: {},
|
||||
get localDatabase() {
|
||||
return getLocalDatabase();
|
||||
},
|
||||
rebuilder: {},
|
||||
services,
|
||||
};
|
||||
|
||||
useReplicationFeature(core as never);
|
||||
|
||||
return {
|
||||
beforeReplicateHandlers,
|
||||
centralRemoteHandlers,
|
||||
createRemoteResource,
|
||||
dispose,
|
||||
get parseHandler() {
|
||||
return parseHandler;
|
||||
},
|
||||
onCloseActiveReplication,
|
||||
read,
|
||||
};
|
||||
}
|
||||
|
||||
describe("replication serviceFeature composition", () => {
|
||||
it("does not acquire the local database while composing result handlers", () => {
|
||||
const acquireLocalDatabase = vi.fn(() => {
|
||||
throw new Error("Local database is not ready yet");
|
||||
});
|
||||
|
||||
expect(() => setup({ getLocalDatabase: acquireLocalDatabase })).not.toThrow();
|
||||
expect(acquireLocalDatabase).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("acquires the current key-value database only when snapshot recovery starts", async () => {
|
||||
const backingDatabase = {
|
||||
get: vi.fn(async () => undefined),
|
||||
set: vi.fn(async () => undefined),
|
||||
};
|
||||
let isReady = false;
|
||||
const acquireKeyValueDB = vi.fn(() => {
|
||||
if (!isReady) throw new Error("KeyValueDB is not initialized yet");
|
||||
return backingDatabase;
|
||||
});
|
||||
const keyValueDB = {
|
||||
get kvDB() {
|
||||
return acquireKeyValueDB();
|
||||
},
|
||||
};
|
||||
|
||||
const { beforeReplicateHandlers } = setup({ keyValueDB });
|
||||
|
||||
expect(acquireKeyValueDB).not.toHaveBeenCalled();
|
||||
isReady = true;
|
||||
const restoreSnapshot = beforeReplicateHandlers.get(100);
|
||||
expect(restoreSnapshot).toBeDefined();
|
||||
await expect(restoreSnapshot!(false)).resolves.toBe(true);
|
||||
expect(acquireKeyValueDB).toHaveBeenCalledOnce();
|
||||
expect(backingDatabase.get).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,341 @@
|
||||
import { LOG_LEVEL_VERBOSE } from "octagonal-wheels/common/logger";
|
||||
import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import type { NecessaryServices } from "@vrtmrz/livesync-commonlib/compat/interfaces/ServiceModule";
|
||||
import { createInstanceLogFunction } from "@vrtmrz/livesync-commonlib/compat/services/lib/logUtils";
|
||||
import {
|
||||
CAPABILITY_UNAVAILABLE_REASONS,
|
||||
isReplicationCompleted,
|
||||
NO_INTERACTION,
|
||||
type ContinuousReplicationRequest,
|
||||
type ReplicationOutcome,
|
||||
type UnattendedOneShotRequest,
|
||||
} from "@vrtmrz/livesync-commonlib/replication";
|
||||
import { PeriodicProcessor } from "@/common/PeriodicProcessor";
|
||||
|
||||
type ReplicationSchedulingSettings = Pick<
|
||||
ObsidianLiveSyncSettings,
|
||||
"isConfigured" | "liveSync" | "syncOnStart" | "periodicReplication" | "periodicReplicationInterval"
|
||||
>;
|
||||
|
||||
/** Timer operations required by the scheduling state owner. */
|
||||
export interface ReplicationSchedulingTimer {
|
||||
enable(intervalMs: number): void;
|
||||
disable(): void;
|
||||
}
|
||||
|
||||
/** Daemon-only controls which do not expose mutable scheduling state. */
|
||||
export interface ReplicationSchedulingControl {
|
||||
/** Let an external daemon poller become, or cease to be, the recurring-work owner. */
|
||||
setExternalPollingMode(enabled: boolean): void;
|
||||
/** Consume the next resume-triggered OneShot because the daemon has already converged once. */
|
||||
markInitialOneShotSatisfied(): void;
|
||||
}
|
||||
|
||||
interface ReplicationSchedulingDependencies {
|
||||
isReady(): boolean;
|
||||
isSuspended(): boolean;
|
||||
currentSettings(): ReplicationSchedulingSettings;
|
||||
replicateUnattended(request: UnattendedOneShotRequest): Promise<ReplicationOutcome>;
|
||||
startContinuous(request: ContinuousReplicationRequest): Promise<ReplicationOutcome>;
|
||||
timer: ReplicationSchedulingTimer;
|
||||
log(error: unknown): void;
|
||||
}
|
||||
|
||||
interface ReplicationSchedulingState {
|
||||
externalPolling: boolean;
|
||||
continuousOwnsRecurring: boolean;
|
||||
initialOneShotSatisfied: boolean;
|
||||
lifecycleAllowsScheduling: boolean;
|
||||
lifecycleGeneration: number;
|
||||
resumeOperation: Promise<void> | undefined;
|
||||
runningResumeGeneration: number | undefined;
|
||||
queuedResumeGeneration: number | undefined;
|
||||
}
|
||||
|
||||
/** Private state and collaborators owned by the replication scheduling serviceFeature. */
|
||||
interface ReplicationSchedulingContext {
|
||||
readonly dependencies: ReplicationSchedulingDependencies;
|
||||
readonly state: ReplicationSchedulingState;
|
||||
}
|
||||
|
||||
function isCapabilityUnavailable(result: ReplicationOutcome): boolean {
|
||||
return (
|
||||
result.status === "blocked" &&
|
||||
(result.reason === CAPABILITY_UNAVAILABLE_REASONS.NOT_APPLICABLE ||
|
||||
result.reason === CAPABILITY_UNAVAILABLE_REASONS.NOT_IMPLEMENTED)
|
||||
);
|
||||
}
|
||||
|
||||
/** Construct the independently testable context owned by the serviceFeature. */
|
||||
export function createReplicationSchedulingContext(
|
||||
dependencies: ReplicationSchedulingDependencies
|
||||
): ReplicationSchedulingContext {
|
||||
return {
|
||||
dependencies,
|
||||
state: {
|
||||
externalPolling: false,
|
||||
continuousOwnsRecurring: false,
|
||||
initialOneShotSatisfied: false,
|
||||
// AppLifecycleService does not expose physical visibility as
|
||||
// isSuspended(). Keep the observed state in this private context.
|
||||
lifecycleAllowsScheduling: false,
|
||||
lifecycleGeneration: 0,
|
||||
resumeOperation: undefined,
|
||||
runningResumeGeneration: undefined,
|
||||
queuedResumeGeneration: undefined,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function canRunPeriodic(context: ReplicationSchedulingContext, settings: ReplicationSchedulingSettings): boolean {
|
||||
const { dependencies, state } = context;
|
||||
return (
|
||||
state.lifecycleAllowsScheduling &&
|
||||
!state.externalPolling &&
|
||||
!state.continuousOwnsRecurring &&
|
||||
dependencies.isReady() &&
|
||||
!dependencies.isSuspended() &&
|
||||
settings.isConfigured === true &&
|
||||
settings.periodicReplication === true
|
||||
);
|
||||
}
|
||||
|
||||
function reconcilePeriodic(context: ReplicationSchedulingContext): void {
|
||||
const { dependencies } = context;
|
||||
const settings = dependencies.currentSettings();
|
||||
if (canRunPeriodic(context, settings)) {
|
||||
dependencies.timer.enable(settings.periodicReplicationInterval * 1000);
|
||||
} else {
|
||||
dependencies.timer.disable();
|
||||
}
|
||||
}
|
||||
|
||||
function setContinuousOwnership(context: ReplicationSchedulingContext, ownsRecurring: boolean): void {
|
||||
const { state } = context;
|
||||
if (state.continuousOwnsRecurring === ownsRecurring) return;
|
||||
state.continuousOwnsRecurring = ownsRecurring;
|
||||
reconcilePeriodic(context);
|
||||
}
|
||||
|
||||
function isCurrentLifecycleGeneration(context: ReplicationSchedulingContext, generation: number): boolean {
|
||||
return generation === context.state.lifecycleGeneration;
|
||||
}
|
||||
|
||||
function canRunResume(context: ReplicationSchedulingContext, generation: number): boolean {
|
||||
const { dependencies, state } = context;
|
||||
return (
|
||||
isCurrentLifecycleGeneration(context, generation) &&
|
||||
state.lifecycleAllowsScheduling &&
|
||||
!state.externalPolling &&
|
||||
dependencies.isReady() &&
|
||||
!dependencies.isSuspended()
|
||||
);
|
||||
}
|
||||
|
||||
async function runAfterResume(context: ReplicationSchedulingContext, generation: number): Promise<void> {
|
||||
if (!canRunResume(context, generation)) return;
|
||||
|
||||
const { dependencies, state } = context;
|
||||
const settings = dependencies.currentSettings();
|
||||
if (!settings.isConfigured) {
|
||||
setContinuousOwnership(context, false);
|
||||
return;
|
||||
}
|
||||
|
||||
const skipOneShot = state.initialOneShotSatisfied;
|
||||
// This marker belongs to one resume attempt. Consume it before any network
|
||||
// await so an exceptional Continuous start cannot suppress a later retry.
|
||||
state.initialOneShotSatisfied = false;
|
||||
if (settings.liveSync) {
|
||||
setContinuousOwnership(context, true);
|
||||
let result: ReplicationOutcome;
|
||||
try {
|
||||
result = await dependencies.startContinuous({
|
||||
trigger: "resume",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
} catch (error) {
|
||||
if (isCurrentLifecycleGeneration(context, generation)) {
|
||||
setContinuousOwnership(context, false);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
if (!isReplicationCompleted(result) && isCurrentLifecycleGeneration(context, generation)) {
|
||||
setContinuousOwnership(context, false);
|
||||
}
|
||||
// A suspend/resume may have started a new lifecycle generation while
|
||||
// Continuous was settling. Do not let the obsolete result schedule a
|
||||
// finite fallback for the new generation.
|
||||
if (isCapabilityUnavailable(result) && canRunResume(context, generation)) {
|
||||
const currentSettings = dependencies.currentSettings();
|
||||
if (
|
||||
currentSettings.isConfigured &&
|
||||
currentSettings.liveSync &&
|
||||
currentSettings.syncOnStart &&
|
||||
!skipOneShot
|
||||
) {
|
||||
await dependencies.replicateUnattended({
|
||||
trigger: "resume",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
setContinuousOwnership(context, false);
|
||||
if (settings.syncOnStart && !skipOneShot) {
|
||||
await dependencies.replicateUnattended({
|
||||
trigger: "resume",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function scheduleAfterResume(context: ReplicationSchedulingContext): void {
|
||||
const { dependencies, state } = context;
|
||||
const requestedGeneration = state.lifecycleGeneration;
|
||||
if (state.resumeOperation) {
|
||||
// Duplicate notifications within one generation share the current
|
||||
// operation. A later lifecycle generation must run after it.
|
||||
if (state.runningResumeGeneration !== requestedGeneration) {
|
||||
state.queuedResumeGeneration = requestedGeneration;
|
||||
}
|
||||
return;
|
||||
}
|
||||
state.runningResumeGeneration = requestedGeneration;
|
||||
state.resumeOperation = runAfterResume(context, requestedGeneration)
|
||||
.catch((error: unknown) => {
|
||||
dependencies.log(error);
|
||||
})
|
||||
.finally(() => {
|
||||
state.resumeOperation = undefined;
|
||||
state.runningResumeGeneration = undefined;
|
||||
const queuedGeneration = state.queuedResumeGeneration;
|
||||
state.queuedResumeGeneration = undefined;
|
||||
if (queuedGeneration === state.lifecycleGeneration && state.lifecycleAllowsScheduling) {
|
||||
scheduleAfterResume(context);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** Schedule eligible work after the application has resumed. */
|
||||
export function resumeReplicationScheduling(context: ReplicationSchedulingContext): void {
|
||||
const { state } = context;
|
||||
if (!state.lifecycleAllowsScheduling) {
|
||||
state.lifecycleGeneration += 1;
|
||||
}
|
||||
state.lifecycleAllowsScheduling = true;
|
||||
// runAfterResume executes synchronously until its first await. A Continuous
|
||||
// request therefore reserves ownership before Periodic is reconciled.
|
||||
scheduleAfterResume(context);
|
||||
reconcilePeriodic(context);
|
||||
}
|
||||
|
||||
/** Stop generic Periodic scheduling before the application suspends. */
|
||||
export function suspendReplicationScheduling(context: ReplicationSchedulingContext): void {
|
||||
context.state.lifecycleAllowsScheduling = false;
|
||||
context.state.queuedResumeGeneration = undefined;
|
||||
context.dependencies.timer.disable();
|
||||
}
|
||||
|
||||
/** Stop generic Periodic scheduling while settings and provider bindings change. */
|
||||
export function prepareReplicationSchedulingForSettings(context: ReplicationSchedulingContext): void {
|
||||
context.dependencies.timer.disable();
|
||||
}
|
||||
|
||||
/** Reconcile generic Periodic scheduling after settings have settled. */
|
||||
export function realiseReplicationScheduling(context: ReplicationSchedulingContext): void {
|
||||
reconcilePeriodic(context);
|
||||
}
|
||||
|
||||
/** Prevent later timer callbacks from scheduling new work during unload. */
|
||||
export function unloadReplicationScheduling(context: ReplicationSchedulingContext): void {
|
||||
context.state.lifecycleAllowsScheduling = false;
|
||||
context.state.queuedResumeGeneration = undefined;
|
||||
context.dependencies.timer.disable();
|
||||
}
|
||||
|
||||
/** Execute one timer callback if Periodic still owns recurring work. */
|
||||
export async function runPeriodicReplication(context: ReplicationSchedulingContext): Promise<void> {
|
||||
const { dependencies } = context;
|
||||
// Clearing an interval does not retract a callback which is already queued.
|
||||
// Recheck ownership and lifecycle state at execution time.
|
||||
if (!canRunPeriodic(context, dependencies.currentSettings())) return;
|
||||
await dependencies.replicateUnattended({
|
||||
trigger: "periodic",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
}
|
||||
|
||||
/** Declare that an external poller has become, or ceased to be, the recurring-work owner. */
|
||||
export function setExternalPollingMode(context: ReplicationSchedulingContext, enabled: boolean): void {
|
||||
if (context.state.externalPolling === enabled) return;
|
||||
context.state.externalPolling = enabled;
|
||||
reconcilePeriodic(context);
|
||||
}
|
||||
|
||||
/** Consume the next resume-triggered OneShot because the daemon has already converged once. */
|
||||
export function markInitialOneShotSatisfied(context: ReplicationSchedulingContext): void {
|
||||
context.state.initialOneShotSatisfied = true;
|
||||
}
|
||||
|
||||
type ReplicationSchedulingHost = NecessaryServices<
|
||||
"API" | "appLifecycle" | "control" | "replication" | "setting",
|
||||
never
|
||||
>;
|
||||
|
||||
type ReplicationSchedulingTimerFactory = (process: () => Promise<void>) => ReplicationSchedulingTimer;
|
||||
|
||||
/**
|
||||
* Compose host lifecycle bindings around one private scheduling context.
|
||||
*
|
||||
* The returned view is intentionally limited to daemon scheduling controls.
|
||||
* @param host Narrow service container used to bind scheduling to the host lifecycle.
|
||||
* @param createTimer Timer adapter factory, replaceable by focused tests.
|
||||
* @returns Commands which let the CLI daemon declare its scheduling ownership.
|
||||
*/
|
||||
export function useReplicationScheduling(
|
||||
host: ReplicationSchedulingHost,
|
||||
createTimer: ReplicationSchedulingTimerFactory = (process) => new PeriodicProcessor(host, process)
|
||||
): ReplicationSchedulingControl {
|
||||
const services = host.services;
|
||||
const log = createInstanceLogFunction("SF:ReplicationScheduling", services.API);
|
||||
let context!: ReplicationSchedulingContext;
|
||||
const timer = createTimer(async () => await runPeriodicReplication(context));
|
||||
context = createReplicationSchedulingContext({
|
||||
isReady: () => services.appLifecycle.isReady(),
|
||||
isSuspended: () => services.appLifecycle.isSuspended(),
|
||||
currentSettings: () => services.setting.currentSettings(),
|
||||
replicateUnattended: (request) => services.replication.replicateUnattended(request),
|
||||
startContinuous: (request) => services.replication.startContinuous(request),
|
||||
timer,
|
||||
log: (error) => log(error, LOG_LEVEL_VERBOSE),
|
||||
});
|
||||
|
||||
services.appLifecycle.onUnload.addHandler(() => {
|
||||
unloadReplicationScheduling(context);
|
||||
return Promise.resolve(true);
|
||||
});
|
||||
services.setting.onBeforeRealiseSetting.addHandler(() => {
|
||||
prepareReplicationSchedulingForSettings(context);
|
||||
return Promise.resolve(true);
|
||||
});
|
||||
services.setting.onSettingRealised.addHandler(() => {
|
||||
realiseReplicationScheduling(context);
|
||||
return Promise.resolve(true);
|
||||
});
|
||||
services.appLifecycle.onSuspending.addHandler(() => {
|
||||
suspendReplicationScheduling(context);
|
||||
return Promise.resolve(true);
|
||||
});
|
||||
services.appLifecycle.onResumed.addHandler(() => {
|
||||
resumeReplicationScheduling(context);
|
||||
return Promise.resolve(true);
|
||||
});
|
||||
|
||||
return Object.freeze({
|
||||
setExternalPollingMode: (enabled: boolean) => setExternalPollingMode(context, enabled),
|
||||
markInitialOneShotSatisfied: () => markInitialOneShotSatisfied(context),
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,370 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { DEFAULT_SETTINGS, REMOTE_MINIO } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||
import { NO_INTERACTION, type ReplicationOutcome } from "@vrtmrz/livesync-commonlib/replication";
|
||||
import {
|
||||
createReplicationSchedulingContext,
|
||||
markInitialOneShotSatisfied,
|
||||
resumeReplicationScheduling,
|
||||
runPeriodicReplication,
|
||||
setExternalPollingMode,
|
||||
suspendReplicationScheduling,
|
||||
useReplicationScheduling,
|
||||
type ReplicationSchedulingTimer,
|
||||
} from "./replicationScheduling";
|
||||
|
||||
function createDeferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((res) => {
|
||||
resolve = res;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
function createControllerHarness(
|
||||
overrides: Partial<{
|
||||
remoteType: typeof DEFAULT_SETTINGS.remoteType;
|
||||
liveSync: boolean;
|
||||
syncOnStart: boolean;
|
||||
periodicReplication: boolean;
|
||||
periodicReplicationInterval: number;
|
||||
}> = {}
|
||||
) {
|
||||
const settings = {
|
||||
...DEFAULT_SETTINGS,
|
||||
isConfigured: true,
|
||||
liveSync: false,
|
||||
syncOnStart: true,
|
||||
periodicReplication: false,
|
||||
periodicReplicationInterval: 60,
|
||||
...overrides,
|
||||
};
|
||||
const timer: ReplicationSchedulingTimer = {
|
||||
enable: vi.fn(),
|
||||
disable: vi.fn(),
|
||||
};
|
||||
const replicateUnattended = vi.fn(async (): Promise<ReplicationOutcome> => ({ status: "completed" }));
|
||||
const startContinuous = vi.fn(async (): Promise<ReplicationOutcome> => ({ status: "completed" }));
|
||||
const log = vi.fn();
|
||||
const context = createReplicationSchedulingContext({
|
||||
isReady: vi.fn(() => true),
|
||||
isSuspended: vi.fn(() => false),
|
||||
currentSettings: vi.fn(() => settings),
|
||||
replicateUnattended,
|
||||
startContinuous,
|
||||
timer,
|
||||
log,
|
||||
});
|
||||
return { context, log, replicateUnattended, settings, startContinuous, timer };
|
||||
}
|
||||
|
||||
describe("replication scheduling context", () => {
|
||||
it("starts a configured unattended OneShot without exposing the operation to the lifecycle handler", async () => {
|
||||
const { context, replicateUnattended } = createControllerHarness();
|
||||
|
||||
expect(resumeReplicationScheduling(context)).toBeUndefined();
|
||||
|
||||
await vi.waitFor(() => expect(replicateUnattended).toHaveBeenCalledOnce());
|
||||
expect(replicateUnattended).toHaveBeenCalledWith({
|
||||
trigger: "resume",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
});
|
||||
|
||||
it("reserves Continuous ownership before reconciling the periodic timer", async () => {
|
||||
const timeline: string[] = [];
|
||||
const continuous = createDeferred<ReplicationOutcome>();
|
||||
const { context, startContinuous, timer } = createControllerHarness({
|
||||
liveSync: true,
|
||||
periodicReplication: true,
|
||||
});
|
||||
vi.mocked(timer.disable).mockImplementation(() => {
|
||||
timeline.push("timer-disabled");
|
||||
});
|
||||
startContinuous.mockImplementation(() => {
|
||||
timeline.push("continuous-started");
|
||||
return continuous.promise;
|
||||
});
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
expect(timeline[0]).toBe("timer-disabled");
|
||||
expect(timeline).toContain("continuous-started");
|
||||
expect(timer.enable).not.toHaveBeenCalled();
|
||||
|
||||
continuous.resolve({ status: "completed" });
|
||||
await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledOnce());
|
||||
});
|
||||
|
||||
it("restores Periodic and falls back to OneShot when Continuous is not applicable", async () => {
|
||||
const { context, replicateUnattended, startContinuous, timer } = createControllerHarness({
|
||||
liveSync: true,
|
||||
periodicReplication: true,
|
||||
periodicReplicationInterval: 45,
|
||||
});
|
||||
startContinuous.mockResolvedValue({
|
||||
status: "blocked",
|
||||
reason: "capability-not-applicable",
|
||||
});
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
await vi.waitFor(() => expect(replicateUnattended).toHaveBeenCalledOnce());
|
||||
expect(timer.enable).toHaveBeenCalledWith(45_000);
|
||||
});
|
||||
|
||||
it("schedules migrated Object Storage syncOnStart through unattended OneShot", async () => {
|
||||
const { context, replicateUnattended, startContinuous } = createControllerHarness({
|
||||
remoteType: REMOTE_MINIO,
|
||||
liveSync: true,
|
||||
syncOnStart: true,
|
||||
});
|
||||
startContinuous.mockResolvedValue({
|
||||
status: "blocked",
|
||||
reason: "capability-not-applicable",
|
||||
});
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
await vi.waitFor(() => expect(replicateUnattended).toHaveBeenCalledOnce());
|
||||
expect(replicateUnattended).toHaveBeenCalledWith({
|
||||
trigger: "resume",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
});
|
||||
|
||||
it("does not run a finite fallback after Continuous starts successfully", async () => {
|
||||
const { context, replicateUnattended, startContinuous } = createControllerHarness({ liveSync: true });
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledOnce());
|
||||
expect(replicateUnattended).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not run a finite fallback after an actual Continuous failure", async () => {
|
||||
const { context, replicateUnattended, startContinuous } = createControllerHarness({ liveSync: true });
|
||||
startContinuous.mockResolvedValue({
|
||||
status: "failed",
|
||||
error: new Error("connection failed"),
|
||||
});
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledOnce());
|
||||
expect(replicateUnattended).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("coalesces concurrent resume notifications", async () => {
|
||||
const replication = createDeferred<ReplicationOutcome>();
|
||||
const { context, replicateUnattended } = createControllerHarness();
|
||||
replicateUnattended.mockImplementation(() => replication.promise);
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
expect(replicateUnattended).toHaveBeenCalledOnce();
|
||||
replication.resolve({ status: "completed" });
|
||||
await vi.waitFor(() => expect(replicateUnattended).toHaveBeenCalledOnce());
|
||||
});
|
||||
|
||||
it("runs a fresh lifecycle generation instead of applying a stale Continuous fallback", async () => {
|
||||
const firstContinuous = createDeferred<ReplicationOutcome>();
|
||||
const { context, replicateUnattended, startContinuous, timer } = createControllerHarness({
|
||||
liveSync: true,
|
||||
periodicReplication: true,
|
||||
});
|
||||
startContinuous
|
||||
.mockImplementationOnce(() => firstContinuous.promise)
|
||||
.mockResolvedValueOnce({ status: "completed" });
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
suspendReplicationScheduling(context);
|
||||
resumeReplicationScheduling(context);
|
||||
firstContinuous.resolve({
|
||||
status: "blocked",
|
||||
reason: "capability-not-applicable",
|
||||
});
|
||||
|
||||
await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledTimes(2));
|
||||
expect(replicateUnattended).not.toHaveBeenCalled();
|
||||
expect(timer.enable).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("lets the daemon suppress scheduling through the focused control view", async () => {
|
||||
const { context, replicateUnattended, startContinuous, timer } = createControllerHarness({
|
||||
liveSync: true,
|
||||
periodicReplication: true,
|
||||
});
|
||||
|
||||
setExternalPollingMode(context, true);
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
expect(timer.disable).toHaveBeenCalled();
|
||||
expect(startContinuous).not.toHaveBeenCalled();
|
||||
expect(replicateUnattended).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("consumes the daemon's initial OneShot marker without suppressing a Continuous attempt", async () => {
|
||||
const { context, replicateUnattended, startContinuous } = createControllerHarness({ liveSync: true });
|
||||
startContinuous.mockResolvedValue({
|
||||
status: "blocked",
|
||||
reason: "capability-not-applicable",
|
||||
});
|
||||
|
||||
markInitialOneShotSatisfied(context);
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledOnce());
|
||||
expect(replicateUnattended).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("consumes the daemon marker even when the first Continuous attempt throws", async () => {
|
||||
const { context, log, replicateUnattended, startContinuous } = createControllerHarness({ liveSync: true });
|
||||
startContinuous.mockRejectedValueOnce(new Error("start failed")).mockResolvedValueOnce({
|
||||
status: "blocked",
|
||||
reason: "capability-not-applicable",
|
||||
});
|
||||
|
||||
markInitialOneShotSatisfied(context);
|
||||
resumeReplicationScheduling(context);
|
||||
await vi.waitFor(() => expect(log).toHaveBeenCalledOnce());
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledTimes(2));
|
||||
expect(replicateUnattended).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("runs the periodic callback through the unattended replication boundary", async () => {
|
||||
const { context, replicateUnattended } = createControllerHarness({
|
||||
syncOnStart: false,
|
||||
periodicReplication: true,
|
||||
});
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
await runPeriodicReplication(context);
|
||||
|
||||
expect(replicateUnattended).toHaveBeenCalledWith({
|
||||
trigger: "periodic",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
});
|
||||
|
||||
it("ignores a queued periodic callback before resume and after suspension", async () => {
|
||||
const { context, replicateUnattended, timer } = createControllerHarness({
|
||||
syncOnStart: false,
|
||||
periodicReplication: true,
|
||||
});
|
||||
|
||||
await runPeriodicReplication(context);
|
||||
expect(replicateUnattended).not.toHaveBeenCalled();
|
||||
|
||||
resumeReplicationScheduling(context);
|
||||
expect(timer.enable).toHaveBeenCalledWith(60_000);
|
||||
|
||||
suspendReplicationScheduling(context);
|
||||
await runPeriodicReplication(context);
|
||||
expect(replicateUnattended).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("ignores a queued periodic callback while external polling owns recurring work", async () => {
|
||||
const { context, replicateUnattended } = createControllerHarness({
|
||||
syncOnStart: false,
|
||||
periodicReplication: true,
|
||||
});
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
setExternalPollingMode(context, true);
|
||||
await runPeriodicReplication(context);
|
||||
|
||||
expect(replicateUnattended).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("ignores a queued periodic callback while Continuous owns recurring work", async () => {
|
||||
const continuous = createDeferred<ReplicationOutcome>();
|
||||
const { context, replicateUnattended, startContinuous } = createControllerHarness({
|
||||
liveSync: true,
|
||||
syncOnStart: false,
|
||||
periodicReplication: true,
|
||||
});
|
||||
startContinuous.mockImplementation(() => continuous.promise);
|
||||
resumeReplicationScheduling(context);
|
||||
|
||||
await runPeriodicReplication(context);
|
||||
|
||||
expect(replicateUnattended).not.toHaveBeenCalled();
|
||||
continuous.resolve({ status: "completed" });
|
||||
});
|
||||
});
|
||||
|
||||
describe("replication scheduling serviceFeature", () => {
|
||||
it("binds lifecycle handlers and exposes only daemon scheduling controls", async () => {
|
||||
const handlers: Record<string, () => Promise<boolean>> = {};
|
||||
const timer: ReplicationSchedulingTimer = {
|
||||
enable: vi.fn(),
|
||||
disable: vi.fn(),
|
||||
};
|
||||
let periodicProcess!: () => Promise<void>;
|
||||
const replicateUnattended = vi.fn(async (): Promise<ReplicationOutcome> => ({ status: "completed" }));
|
||||
const addHandler = (name: string) =>
|
||||
vi.fn((handler: () => Promise<boolean>) => {
|
||||
handlers[name] = handler;
|
||||
return () => undefined;
|
||||
});
|
||||
const services = {
|
||||
context: {},
|
||||
API: { addLog: vi.fn() },
|
||||
appLifecycle: {
|
||||
isReady: vi.fn(() => true),
|
||||
isSuspended: vi.fn(() => false),
|
||||
onResumed: { addHandler: addHandler("resumed") },
|
||||
onSuspending: { addHandler: addHandler("suspending") },
|
||||
onUnload: { addHandler: addHandler("unload") },
|
||||
},
|
||||
control: { hasUnloaded: vi.fn(() => false) },
|
||||
replication: {
|
||||
replicateUnattended,
|
||||
startContinuous: vi.fn(async (): Promise<ReplicationOutcome> => ({ status: "completed" })),
|
||||
},
|
||||
setting: {
|
||||
currentSettings: vi.fn(() => ({
|
||||
...DEFAULT_SETTINGS,
|
||||
isConfigured: true,
|
||||
liveSync: false,
|
||||
syncOnStart: true,
|
||||
periodicReplication: true,
|
||||
})),
|
||||
onBeforeRealiseSetting: { addHandler: addHandler("before-setting") },
|
||||
onSettingRealised: { addHandler: addHandler("setting-realised") },
|
||||
},
|
||||
};
|
||||
|
||||
const control = useReplicationScheduling({ services, serviceModules: {} } as never, (process) => {
|
||||
periodicProcess = process;
|
||||
return timer;
|
||||
});
|
||||
|
||||
expect(Object.keys(control).sort()).toEqual(["markInitialOneShotSatisfied", "setExternalPollingMode"]);
|
||||
expect(Object.keys(handlers).sort()).toEqual([
|
||||
"before-setting",
|
||||
"resumed",
|
||||
"setting-realised",
|
||||
"suspending",
|
||||
"unload",
|
||||
]);
|
||||
|
||||
await expect(handlers.resumed()).resolves.toBe(true);
|
||||
await vi.waitFor(() => expect(replicateUnattended).toHaveBeenCalledOnce());
|
||||
|
||||
replicateUnattended.mockClear();
|
||||
await periodicProcess();
|
||||
expect(replicateUnattended).toHaveBeenCalledWith({
|
||||
trigger: "periodic",
|
||||
interaction: NO_INTERACTION,
|
||||
});
|
||||
|
||||
control.setExternalPollingMode(true);
|
||||
expect(timer.disable).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -1,8 +1,6 @@
|
||||
import { eventHub, EVENT_REQUEST_OPEN_P2P } from "@/common/events";
|
||||
import { reactiveSource } from "octagonal-wheels/dataobject/reactive_v2";
|
||||
import type { NecessaryServices } from "@vrtmrz/livesync-commonlib/compat/interfaces/ServiceModule";
|
||||
import { type UseP2PReplicatorResult } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/UseP2PReplicatorResult";
|
||||
import { P2PLogCollector } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/P2PLogCollector";
|
||||
import {
|
||||
P2PServerStatusPaneView,
|
||||
VIEW_TYPE_P2P_SERVER_STATUS,
|
||||
@@ -21,6 +19,9 @@ class LegacyP2PStatusPaneView extends P2PServerStatusPaneView {
|
||||
}
|
||||
}
|
||||
|
||||
/** Host-owned peer-selection entry used by the two adjunct P2P commands. */
|
||||
export type OpenInteractiveP2PReplication = (showResult: boolean) => Promise<boolean | void>;
|
||||
|
||||
export function hasP2PConfiguration(settings: Partial<ObsidianLiveSyncSettings>): boolean {
|
||||
if (
|
||||
settings.remoteType === REMOTE_P2P ||
|
||||
@@ -61,7 +62,8 @@ export function useP2PReplicatorUI(
|
||||
never
|
||||
>,
|
||||
core: LiveSyncCore,
|
||||
replicator: UseP2PReplicatorResult
|
||||
replicator: UseP2PReplicatorResult,
|
||||
openInteractiveReplication: OpenInteractiveP2PReplication
|
||||
) {
|
||||
const api = host.services.API as {
|
||||
showWindow: (type: string) => Promise<void>;
|
||||
@@ -80,26 +82,11 @@ export function useP2PReplicatorUI(
|
||||
) => { addClass?: (name: string) => unknown; remove?: () => void } | undefined;
|
||||
};
|
||||
|
||||
// const env: LiveSyncTrysteroReplicatorEnv = { services: host.services as any };
|
||||
const getReplicator = () => replicator.replicator;
|
||||
const p2pLogCollector = new P2PLogCollector(host.services.context.events);
|
||||
const storeP2PStatusLine = reactiveSource("");
|
||||
p2pLogCollector.p2pReplicationLine.onChanged((line) => {
|
||||
storeP2PStatusLine.value = line.value;
|
||||
});
|
||||
const p2pParams = {
|
||||
get replicator() {
|
||||
return getReplicator();
|
||||
},
|
||||
p2pLogCollector,
|
||||
storeP2PStatusLine,
|
||||
};
|
||||
|
||||
const statusFactory = (leaf: WorkspaceLeaf) => {
|
||||
return new P2PServerStatusPaneView(leaf, core, p2pParams);
|
||||
return new P2PServerStatusPaneView(leaf, core, replicator);
|
||||
};
|
||||
const legacyStatusFactory = (leaf: WorkspaceLeaf) => {
|
||||
return new LegacyP2PStatusPaneView(leaf, core, p2pParams);
|
||||
return new LegacyP2PStatusPaneView(leaf, core, replicator);
|
||||
};
|
||||
const openStatusPane = () => {
|
||||
if (api.showWindowOnRight) {
|
||||
@@ -108,13 +95,11 @@ export function useP2PReplicatorUI(
|
||||
return api.showWindow(VIEW_TYPE_P2P_SERVER_STATUS);
|
||||
};
|
||||
const runOpenReplication = () => {
|
||||
const activeReplicator = replicator.replicator;
|
||||
if (!activeReplicator) return;
|
||||
const settings = host.services.setting.currentSettings();
|
||||
void host.services.replicator.runFiniteReplicationActivity(
|
||||
() => activeReplicator.openReplication(settings, false, true, false),
|
||||
{ label: "replication" }
|
||||
);
|
||||
// Peer selection is a host UI concern. The injected operation opens
|
||||
// the dialogue, while its transfer callbacks use focused P2P views.
|
||||
void host.services.replicator.runFiniteReplicationActivity(() => openInteractiveReplication(true), {
|
||||
label: "replication",
|
||||
});
|
||||
};
|
||||
// Keep the retired view type registered only long enough to restore an
|
||||
// existing workspace leaf with the current status UI. Layout-ready
|
||||
@@ -167,7 +152,7 @@ export function useP2PReplicatorUI(
|
||||
const isAvailable =
|
||||
hasP2PConfiguration(settings) &&
|
||||
settings.remoteType !== REMOTE_P2P &&
|
||||
(replicator.replicator?.server?.isServing ?? false);
|
||||
replicator.transportLifecycle.isConnected;
|
||||
if (!isAvailable) return false;
|
||||
if (!isChecking) {
|
||||
runOpenReplication();
|
||||
@@ -183,7 +168,7 @@ export function useP2PReplicatorUI(
|
||||
const isAvailable =
|
||||
hasP2PConfiguration(settings) &&
|
||||
settings.remoteType !== REMOTE_P2P &&
|
||||
(replicator.replicator?.server?.isServing ?? false);
|
||||
replicator.transportLifecycle.isConnected;
|
||||
if (!isAvailable) return false;
|
||||
if (!isChecking) {
|
||||
runOpenReplication();
|
||||
@@ -198,10 +183,13 @@ export function useP2PReplicatorUI(
|
||||
checkCallback: (isChecking: boolean) => {
|
||||
const isAvailable =
|
||||
hasP2PConfiguration(host.services.setting.currentSettings()) &&
|
||||
(replicator.replicator?.server?.isServing ?? false);
|
||||
replicator.transportLifecycle.isConnected;
|
||||
if (!isAvailable) return false;
|
||||
if (!isChecking) {
|
||||
void replicator.replicator?.replicateFromCommand(true);
|
||||
void host.services.replicator.runFiniteReplicationActivity(
|
||||
() => replicator.targetedTransfer.synchroniseConfiguredTargets(),
|
||||
{ label: "replication" }
|
||||
);
|
||||
}
|
||||
return true;
|
||||
},
|
||||
@@ -239,5 +227,4 @@ export function useP2PReplicatorUI(
|
||||
);
|
||||
return true;
|
||||
});
|
||||
return p2pParams;
|
||||
}
|
||||
|
||||
@@ -13,6 +13,8 @@ vi.mock("@/features/P2PSync/P2PReplicator/P2PServerStatusPaneView", () => ({
|
||||
|
||||
import { useP2PReplicatorUI } from "./useP2PReplicatorUI";
|
||||
|
||||
const noopOpenInteractiveReplication = () => Promise.resolve(false);
|
||||
|
||||
describe("useP2PReplicatorUI commands", () => {
|
||||
it("waits for settings to load before deciding whether to show the P2P ribbon", async () => {
|
||||
let initialise: (() => Promise<unknown>) | undefined;
|
||||
@@ -49,7 +51,7 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
},
|
||||
} as any;
|
||||
|
||||
useP2PReplicatorUI(host, {} as any, { replicator: undefined } as any);
|
||||
useP2PReplicatorUI(host, {} as any, {} as any, noopOpenInteractiveReplication);
|
||||
|
||||
await expect(initialise?.()).resolves.toBe(true);
|
||||
expect(currentSettings).not.toHaveBeenCalled();
|
||||
@@ -64,7 +66,7 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
it("exposes a direct modal P2P replication command as finite replication activity", async () => {
|
||||
const commands: Array<{ id: string; checkCallback?: (isChecking: boolean) => unknown }> = [];
|
||||
let initialise: (() => Promise<unknown>) | undefined;
|
||||
const openReplication = vi.fn(async () => true);
|
||||
const openInteractiveReplication = vi.fn(async () => true);
|
||||
const runFiniteReplicationActivity = vi.fn(async (task: () => unknown) => await task());
|
||||
const host = {
|
||||
services: {
|
||||
@@ -95,58 +97,19 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
},
|
||||
} as any;
|
||||
const p2p = {
|
||||
replicator: {
|
||||
server: { isServing: true },
|
||||
openReplication,
|
||||
replicateFromCommand: vi.fn(),
|
||||
},
|
||||
transportLifecycle: { isConnected: true },
|
||||
} as any;
|
||||
|
||||
useP2PReplicatorUI(host, {} as any, p2p);
|
||||
useP2PReplicatorUI(host, {} as any, p2p, openInteractiveReplication);
|
||||
await initialise?.();
|
||||
commands.find((command) => command.id === "replicate-now-by-p2p")?.checkCallback?.(false);
|
||||
|
||||
await vi.waitFor(() => expect(openReplication).toHaveBeenCalledOnce());
|
||||
await vi.waitFor(() => expect(openInteractiveReplication).toHaveBeenCalledWith(true));
|
||||
expect(runFiniteReplicationActivity).toHaveBeenCalledWith(expect.any(Function), {
|
||||
label: "replication",
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps the current replicator in the pane parameters after replacement", () => {
|
||||
const first = { id: "first" };
|
||||
const second = { id: "second" };
|
||||
let current = first;
|
||||
const p2p = {
|
||||
get replicator() {
|
||||
return current;
|
||||
},
|
||||
} as any;
|
||||
const host = {
|
||||
services: {
|
||||
context: createServiceContext(),
|
||||
API: {
|
||||
showWindow: vi.fn(async () => undefined),
|
||||
registerWindow: vi.fn(),
|
||||
addCommand: vi.fn(),
|
||||
addRibbonIcon: vi.fn(),
|
||||
getPlatform: vi.fn(() => "obsidian"),
|
||||
},
|
||||
appLifecycle: {
|
||||
onInitialise: { addHandler: vi.fn() },
|
||||
onSettingLoaded: { addHandler: vi.fn() },
|
||||
onLayoutReady: { addHandler: vi.fn() },
|
||||
},
|
||||
setting: { currentSettings: vi.fn(() => ({ remoteType: "COUCHDB" })) },
|
||||
replicator: { runFiniteReplicationActivity: vi.fn() },
|
||||
},
|
||||
} as any;
|
||||
|
||||
const paneParams = useP2PReplicatorUI(host, {} as any, p2p);
|
||||
current = second;
|
||||
|
||||
expect(paneParams.replicator).toBe(second);
|
||||
});
|
||||
|
||||
it("retains only the current P2P status command and routes existing open requests to it", async () => {
|
||||
const commands: Array<{
|
||||
id: string;
|
||||
@@ -185,9 +148,9 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
replicator: { runFiniteReplicationActivity: vi.fn() },
|
||||
},
|
||||
} as any;
|
||||
const p2p = { replicator: undefined } as any;
|
||||
const p2p = {} as any;
|
||||
|
||||
useP2PReplicatorUI(host, {} as any, p2p);
|
||||
useP2PReplicatorUI(host, {} as any, p2p, noopOpenInteractiveReplication);
|
||||
await initialise?.();
|
||||
|
||||
expect(commands.map((command) => command.id)).not.toContain("open-p2p-replicator");
|
||||
@@ -209,6 +172,7 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
remoteType: "COUCHDB",
|
||||
remoteConfigurations: {},
|
||||
};
|
||||
const runFiniteReplicationActivity = vi.fn(async (task: () => unknown) => await task());
|
||||
const host = {
|
||||
services: {
|
||||
context: createServiceContext(),
|
||||
@@ -232,18 +196,16 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
currentSettings: vi.fn(() => settings),
|
||||
onSettingSaved: { addHandler: vi.fn() },
|
||||
},
|
||||
replicator: { runFiniteReplicationActivity: vi.fn() },
|
||||
replicator: { runFiniteReplicationActivity },
|
||||
},
|
||||
} as any;
|
||||
const synchroniseConfiguredTargets = vi.fn(async () => ({ status: "completed" }));
|
||||
const p2p = {
|
||||
replicator: {
|
||||
server: { isServing: true },
|
||||
openReplication: vi.fn(),
|
||||
replicateFromCommand: vi.fn(),
|
||||
},
|
||||
transportLifecycle: { isConnected: true },
|
||||
targetedTransfer: { synchroniseConfiguredTargets },
|
||||
} as any;
|
||||
|
||||
useP2PReplicatorUI(host, {} as any, p2p);
|
||||
useP2PReplicatorUI(host, {} as any, p2p, noopOpenInteractiveReplication);
|
||||
await initialise?.();
|
||||
|
||||
for (const commandId of [
|
||||
@@ -274,6 +236,12 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
]) {
|
||||
expect(commands.find(({ id }) => id === commandId)?.checkCallback?.(true)).toBe(true);
|
||||
}
|
||||
|
||||
commands.find(({ id }) => id === "p2p-sync-targets")?.checkCallback?.(false);
|
||||
await vi.waitFor(() => expect(synchroniseConfiguredTargets).toHaveBeenCalledOnce());
|
||||
expect(runFiniteReplicationActivity).toHaveBeenCalledWith(expect.any(Function), {
|
||||
label: "replication",
|
||||
});
|
||||
});
|
||||
|
||||
it("does not open the P2P status pane automatically when the workspace becomes ready", async () => {
|
||||
@@ -310,7 +278,7 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
},
|
||||
} as any;
|
||||
|
||||
useP2PReplicatorUI(host, {} as any, { replicator: undefined } as any);
|
||||
useP2PReplicatorUI(host, {} as any, {} as any, noopOpenInteractiveReplication);
|
||||
await layoutReady?.();
|
||||
|
||||
expect(showWindow).not.toHaveBeenCalled();
|
||||
@@ -366,7 +334,7 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
},
|
||||
} as any;
|
||||
|
||||
useP2PReplicatorUI(host, {} as any, { replicator: undefined } as any);
|
||||
useP2PReplicatorUI(host, {} as any, {} as any, noopOpenInteractiveReplication);
|
||||
await initialise?.();
|
||||
await settingLoaded?.();
|
||||
expect(addRibbonIcon).not.toHaveBeenCalled();
|
||||
@@ -440,7 +408,7 @@ describe("useP2PReplicatorUI commands", () => {
|
||||
},
|
||||
} as any;
|
||||
|
||||
useP2PReplicatorUI(host, {} as any, { replicator: undefined } as any);
|
||||
useP2PReplicatorUI(host, {} as any, {} as any, noopOpenInteractiveReplication);
|
||||
await layoutReady?.();
|
||||
|
||||
expect(legacyLeaf.setViewState).toHaveBeenCalledWith({
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import { NEW_VAULT_SETTINGS } from "@vrtmrz/livesync-commonlib/settings";
|
||||
import { LOG_LEVEL_NOTICE } from "octagonal-wheels/common/logger";
|
||||
import type { UseP2PReplicatorResult } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/UseP2PReplicatorResult";
|
||||
import type ObsidianLiveSyncPlugin from "@/main";
|
||||
import type { LiveSyncCore } from "@/main";
|
||||
import type { WorkspaceLeaf } from "@/deps";
|
||||
@@ -47,7 +46,6 @@ async function runVaultRoundTrip(plugin: ObsidianLiveSyncPlugin): Promise<Review
|
||||
export function useReviewHarness(
|
||||
core: LiveSyncCore,
|
||||
plugin: ObsidianLiveSyncPlugin,
|
||||
p2p: UseP2PReplicatorResult,
|
||||
compatibilityReview: CompatibilityReviewController
|
||||
): ReviewHarnessController {
|
||||
const services = core.services;
|
||||
@@ -59,11 +57,6 @@ export function useReviewHarness(
|
||||
isCompatibilityReviewInitialised: () => compatibilityReview.initialised,
|
||||
getCompatibilityPause: () => compatibilityReview.pendingPause,
|
||||
openCompatibilityReview: () => compatibilityReview.openReview(),
|
||||
getP2PComposition: () => ({
|
||||
first: p2p.replicator,
|
||||
second: p2p.replicator,
|
||||
expectedServices: services,
|
||||
}),
|
||||
runVaultRoundTrip: () => runVaultRoundTrip(plugin),
|
||||
readContinuation: () => services.setting.getSmallConfig(REVIEW_HARNESS_STATE_KEY),
|
||||
writeContinuation: (value) => services.setting.setSmallConfig(REVIEW_HARNESS_STATE_KEY, value),
|
||||
|
||||
@@ -42,7 +42,6 @@ function createFixture(options: { enableDebugTools?: boolean; continuation?: str
|
||||
periodicReplication: true,
|
||||
};
|
||||
const services = {} as Record<string, unknown>;
|
||||
const replicator = { env: { services } };
|
||||
const api = {
|
||||
registerWindow: vi.fn(),
|
||||
addCommand: vi.fn(),
|
||||
@@ -101,7 +100,7 @@ function createFixture(options: { enableDebugTools?: boolean; continuation?: str
|
||||
},
|
||||
};
|
||||
|
||||
const controller = useReviewHarness(core as never, plugin as never, { replicator } as never, compatibilityReview as never);
|
||||
const controller = useReviewHarness(core as never, plugin as never, compatibilityReview as never);
|
||||
return {
|
||||
controller,
|
||||
api,
|
||||
|
||||
Reference in New Issue
Block a user