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

Defer replication runtime database acquisition

Browse Source
This commit is contained in:
vorotamoroz
2026-08-30 16:39:15 +00:00
parent 6364988453
commit 23dd4ab87e
6 changed files with 100 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/serviceFeatures/replication/ReplicateResultProcessor.ts
+12 -7
View File
@@ -44,11 +44,16 @@ type ReplicateResultProcessorServices = Pick<
* 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 keyValueDB: LiveSyncBaseCore["kvDB"];
readonly localDatabase: LiveSyncBaseCore["localDatabase"];
readonly getKeyValueDB: () => LiveSyncBaseCore["kvDB"];
readonly getLocalDatabase: () => LiveSyncBaseCore["localDatabase"];
readonly requestActiveReplicatorRetirement: () => void;
readonly runLocalApplicationActivity: <T>(
task: () => T | PromiseLike<T>,
@@ -77,7 +82,7 @@ export class ReplicateResultProcessor {
constructor(private readonly context: ReplicateResultProcessorContext) {}
private get localDatabase() {
return this.context.localDatabase;
return this.context.getLocalDatabase();
}
private get services() {
return this.context.services;
@@ -119,7 +124,7 @@ export class ReplicateResultProcessor {
queued: this._queuedChanges.slice(),
processing: this._processingChanges.slice(),
} satisfies ReplicateResultProcessorState;
await this.context.keyValueDB.set(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT, snapshot);
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
@@ -141,9 +146,9 @@ export class ReplicateResultProcessor {
* Restore from snapshot.
*/
public async restoreFromSnapshot() {
const snapshot = await this.context.keyValueDB.get<ReplicateResultProcessorState>(
KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT
);
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];
src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts
+5 -3
View File
@@ -56,8 +56,8 @@ function setup(options: SetupOptions = {}) {
};
const processor = new ReplicateResultProcessor({
currentSettings: () => ({ maxMTimeForReflectEvents: 0, suspendParseReplicationResult: false }),
keyValueDB: core.kvDB,
localDatabase: core.localDatabase,
getKeyValueDB: () => core.kvDB,
getLocalDatabase: () => core.localDatabase,
requestActiveReplicatorRetirement: () => {
void onCloseActiveReplication();
},
@@ -95,14 +95,16 @@ describe("ReplicateResultProcessor", () => {
const findAllNormalDocs = vi.fn(async function* () {
yield* documents;
});
const getLocalDatabase = vi.fn(() => ({ findAllNormalDocs }));
const processor = new ReplicateResultProcessor({
localDatabase: { findAllNormalDocs },
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);
});
src/serviceFeatures/replication/centralCompatibilityRecovery.ts
+10 -7
View File
@@ -17,9 +17,11 @@ type CentralCompatibilityRecoveryServices = Pick<
"API" | "appLifecycle" | "replicator" | "tweakValue"
>;
/** Collaborators for applying a compatibility decision to its failed publication. */
interface CentralCompatibilityRecoveryContext {
readonly confirm: LiveSyncBaseCore["confirm"];
readonly localDatabase: LiveSyncBaseCore["localDatabase"];
/** Obtain the database only when recovery runs, after initialisation or reset. */
readonly getLocalDatabase: () => LiveSyncBaseCore["localDatabase"];
readonly rebuilder: LiveSyncBaseCore["rebuilder"];
readonly services: CentralCompatibilityRecoveryServices;
}
@@ -56,7 +58,7 @@ export function createCentralCompatibilityRecovery(context: CentralCompatibility
) {
Logger("The remote database has been cleaned.", showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO);
await skipIfDuplicated("cleanup", async () => {
const count = await purgeUnreferencedChunks(context.localDatabase.localDatabase, true);
const 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.
@@ -83,6 +85,7 @@ Even if you choose to clean up, you will see this option again if you exit Obsid
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(),
@@ -94,16 +97,16 @@ Even if you choose to clean up, you will see this option again if you exit Obsid
}
try {
await purgeUnreferencedChunks(context.localDatabase.localDatabase, false);
context.localDatabase.clearCaches();
await purgeUnreferencedChunks(localDatabase.localDatabase, false);
localDatabase.clearCaches();
const replicated = await context.services.replicator.runFiniteReplicationActivity(
() => replicator.openOneShotReplication(setting, showMessage, false, "sync", true),
{ label: "replication" }
);
if (replicated) {
await balanceChunkPurgedDBs(context.localDatabase.localDatabase, remoteDatabase.db);
await purgeUnreferencedChunks(context.localDatabase.localDatabase, false);
context.localDatabase.clearCaches();
await balanceChunkPurgedDBs(localDatabase.localDatabase, remoteDatabase.db);
await purgeUnreferencedChunks(localDatabase.localDatabase, false);
localDatabase.clearCaches();
await replicator.markRemoteResolved(setting);
Logger(
"The local database has been cleaned up.",
src/serviceFeatures/replication/centralCompatibilityRecovery.unit.spec.ts
+7 -5
View File
@@ -28,7 +28,7 @@ describe("central compatibility recovery", () => {
try {
const recovery = createCentralCompatibilityRecovery({
confirm: {},
localDatabase: {},
getLocalDatabase: () => ({}),
rebuilder: {},
services: {
appLifecycle: {},
@@ -75,7 +75,7 @@ describe("central compatibility recovery", () => {
});
const recovery = createCentralCompatibilityRecovery({
confirm: {},
localDatabase: {},
getLocalDatabase: () => ({}),
rebuilder: {},
services: {
appLifecycle: {},
@@ -137,7 +137,7 @@ describe("central compatibility recovery", () => {
);
const recovery = createCentralCompatibilityRecovery({
confirm: {},
localDatabase: {},
getLocalDatabase: () => ({}),
rebuilder: {},
services: {
appLifecycle: {},
@@ -180,7 +180,7 @@ describe("central compatibility recovery", () => {
confirm: {
askSelectStringDialogue: vi.fn(async (_message: string, choices: string[]) => choices[1]),
},
localDatabase: {},
getLocalDatabase: () => ({}),
rebuilder: {},
services: {
appLifecycle: { scheduleRestart: vi.fn() },
@@ -228,9 +228,10 @@ describe("central compatibility recovery", () => {
task(expectedContext)
);
const localDatabase = { localDatabase: {}, clearCaches: vi.fn() };
const getLocalDatabase = vi.fn(() => localDatabase);
const recovery = createCentralCompatibilityRecovery({
confirm: { confirmWithMessage: vi.fn(async () => "Cleanup") },
localDatabase,
getLocalDatabase,
rebuilder: {},
services: {
appLifecycle: {},
@@ -258,6 +259,7 @@ describe("central compatibility recovery", () => {
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]);
});
src/serviceFeatures/replication/index.ts
+3 -3
View File
@@ -39,8 +39,8 @@ export function useReplicationFeature<TContext extends ServiceContext, TCommands
: undefined;
const resultProcessor = new ReplicateResultProcessor({
currentSettings: () => services.setting.currentSettings(),
keyValueDB: services.keyValueDB.kvDB,
localDatabase: core.localDatabase,
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.
@@ -81,7 +81,7 @@ export function useReplicationFeature<TContext extends ServiceContext, TCommands
const securitySeedPreflight = createSecuritySeedPreflight(unresolvedErrorManager, preflightContext);
const centralCompatibilityRecovery = createCentralCompatibilityRecovery({
confirm: core.confirm,
localDatabase: core.localDatabase,
getLocalDatabase: () => core.localDatabase,
rebuilder: core.rebuilder,
services: {
API: services.API,
src/serviceFeatures/replication/replicationFeature.unit.spec.ts
+63 -9
View File
@@ -6,8 +6,29 @@ 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(onCloseActiveReplication = vi.fn(async () => true)) {
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 }));
@@ -24,12 +45,7 @@ function setup(onCloseActiveReplication = vi.fn(async () => true)) {
},
context: createServiceContext(),
databaseEvents: { onDatabaseInitialised: { addHandler: vi.fn() } },
keyValueDB: {
kvDB: {
get: vi.fn(async () => undefined),
set: vi.fn(async () => undefined),
},
},
keyValueDB,
path: { getPath: vi.fn((entry: { path: string }) => entry.path) },
replication: {
onBeforeReplicate: {
@@ -59,7 +75,9 @@ function setup(onCloseActiveReplication = vi.fn(async () => true)) {
};
const core = {
confirm: {},
localDatabase: {},
get localDatabase() {
return getLocalDatabase();
},
rebuilder: {},
services,
};
@@ -80,6 +98,42 @@ function setup(onCloseActiveReplication = vi.fn(async () => true)) {
}
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();
@@ -109,7 +163,7 @@ describe("replication serviceFeature composition", () => {
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 harness = setup({ onCloseActiveReplication });
const versionInfo = {
_id: "versioninfo",
_rev: "1-test",