public/obsidian-livesync
1
0 Fork 0
mirror of https://github.com/vrtmrz/obsidian-livesync.git synced 2026-10-31 21:38:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(syncguard): isolate CouchDB catch-up from P2P activity

Browse Source
This commit is contained in:
Artem Ovchinnikov committed 2026-10-21 21:44:17 +03:00
1 parent ee0272a65f
commit bc185d8bd0
2 files changed
+39 -23
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/serviceFeatures/syncGuard/index.ts
+4 -4
View File
@@ -73,14 +73,14 @@ export function useSyncGuardFeature(host: LiveSyncCore, createOverlay: SyncGuard
return; return;
} }
if (guard.presentation.phase === "observing" && if (guard.presentation.phase === "observing" &&
services.replicator.finiteReplicationActivityCount.value > 0) { services.replicator.couchDbCatchUpActivityCount.value > 0) {
return; return;
} }
guard.resume(Boolean(isEnabled()), settings.syncGuardMinimumFiles); guard.resume(Boolean(isEnabled()), settings.syncGuardMinimumFiles);
guard.finiteActivityChanged(services.replicator.finiteReplicationActivityCount.value > 0, readPendingCount(), Date.now()); guard.finiteActivityChanged(services.replicator.couchDbCatchUpActivityCount.value > 0, readPendingCount(), Date.now());
}; };
const onFiniteChanged = () => { const onFiniteChanged = () => {
const active = services.replicator.finiteReplicationActivityCount.value > 0; const active = services.replicator.couchDbCatchUpActivityCount.value > 0;
// Continuous replication can restart its finite pull following an internal reconnect, // Continuous replication can restart its finite pull following an internal reconnect,
// without an Obsidian onResumed notification. A new bounded transfer gets its own window. // without an Obsidian onResumed notification. A new bounded transfer gets its own window.
if (active && guard.presentation.phase === "idle" && !guard.wasBypassed && isEnabled()) beginCatchUp(); if (active && guard.presentation.phase === "idle" && !guard.wasBypassed && isEnabled()) beginCatchUp();
@@ -101,7 +101,7 @@ export function useSyncGuardFeature(host: LiveSyncCore, createOverlay: SyncGuard
onSettingsChanged(); onSettingsChanged();
}); });
const finite = services.replicator.finiteReplicationActivityCount; const finite = services.replicator.couchDbCatchUpActivityCount;
const replication = services.replicator.replicationStatics; const replication = services.replicator.replicationStatics;
const pendingSources = [ const pendingSources = [
services.replication.replicationResultCount, services.replication.replicationResultCount,
src/serviceFeatures/syncGuard/syncGuardFeature.unit.spec.ts
+35 -19
View File
@@ -28,6 +28,7 @@ function setup(overrides: { enabled?: boolean; minimumFiles?: number } = {}) {
const onSettingRealised: Array<() => Promise<boolean>> = []; const onSettingRealised: Array<() => Promise<boolean>> = [];
const parseHandlers: Array<(docs: PouchDB.Core.ExistingDocument<EntryDoc>[]) => Promise<boolean>> = []; const parseHandlers: Array<(docs: PouchDB.Core.ExistingDocument<EntryDoc>[]) => Promise<boolean>> = [];
const priority = vi.fn(); const priority = vi.fn();
const couchDbCatchUpActivityCount = reactiveSource(0);
const finiteReplicationActivityCount = reactiveSource(0); const finiteReplicationActivityCount = reactiveSource(0);
const replicationStatics = reactiveSource({ syncStatus: "NOT_CONNECTED" }); const replicationStatics = reactiveSource({ syncStatus: "NOT_CONNECTED" });
const replicationResultCount = reactiveSource(0); const replicationResultCount = reactiveSource(0);
@@ -61,7 +62,7 @@ function setup(overrides: { enabled?: boolean; minimumFiles?: number } = {}) {
}, },
}, },
}, },
replicator: { finiteReplicationActivityCount, replicationStatics }, replicator: { couchDbCatchUpActivityCount, finiteReplicationActivityCount, replicationStatics },
fileProcessing: { processing, totalQueued, batched }, fileProcessing: { processing, totalQueued, batched },
conflict: { conflictProcessQueueCount }, conflict: { conflictProcessQueueCount },
appLifecycle: { appLifecycle: {
@@ -92,6 +93,7 @@ function setup(overrides: { enabled?: boolean; minimumFiles?: number } = {}) {
onSettingRealised, onSettingRealised,
parseHandlers, parseHandlers,
priority, priority,
couchDbCatchUpActivityCount,
finiteReplicationActivityCount, finiteReplicationActivityCount,
replicationStatics, replicationStatics,
replicationResultCount, replicationResultCount,
@@ -113,12 +115,12 @@ describe("useSyncGuardFeature", () => {
const h = setup(); const h = setup();
expect(h.priority).toHaveBeenCalledWith(-100); expect(h.priority).toHaveBeenCalledWith(-100);
await h.onResumed[0](); await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("first"), receivedNote("second")]); await h.parseHandlers[0]([receivedNote("first"), receivedNote("second")]);
expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring"); expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring");
h.databaseQueueCount.value = 1; h.databaseQueueCount.value = 1;
h.replicationStatics.value = { syncStatus: "COMPLETED" }; h.replicationStatics.value = { syncStatus: "COMPLETED" };
h.finiteReplicationActivityCount.value = 0; h.couchDbCatchUpActivityCount.value = 0;
expect(h.updates[h.updates.length - 1]?.phase).toBe("applying"); expect(h.updates[h.updates.length - 1]?.phase).toBe("applying");
vi.advanceTimersByTime(5_000); vi.advanceTimersByTime(5_000);
expect(h.updates[h.updates.length - 1]?.phase).toBe("applying"); expect(h.updates[h.updates.length - 1]?.phase).toBe("applying");
@@ -129,10 +131,24 @@ describe("useSyncGuardFeature", () => {
expect(h.dispose).toHaveBeenCalledOnce(); expect(h.dispose).toHaveBeenCalledOnce();
}); });
it("ignores large P2P finite transfers when CouchDB catch-up is not running", async () => {
const h = setup({ minimumFiles: 2 });
await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; // P2P uses only the shared finite counter.
await h.parseHandlers[0]([receivedNote("peer-first"), receivedNote("peer-second")]);
h.finiteReplicationActivityCount.value = 0;
expect(h.updates.some((p) => p.phase === "transferring" || p.phase === "interrupted")).toBe(false);
// A real CouchDB catch-up must still arm normally in the same session.
h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("db-first"), receivedNote("db-second")]);
expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring");
await h.onUnload[0]();
});
it("never blocks normal transfers and ignores chunks and duplicate note revisions", async () => { it("never blocks normal transfers and ignores chunks and duplicate note revisions", async () => {
const h = setup({ minimumFiles: 3 }); const h = setup({ minimumFiles: 3 });
await h.onResumed[0](); await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("first"), receivedNote("first"), { _id: "h:chunk", _rev: "1-a", type: "leaf" } as never]); await h.parseHandlers[0]([receivedNote("first"), receivedNote("first"), { _id: "h:chunk", _rev: "1-a", type: "leaf" } as never]);
expect(h.updates.some((p) => p.phase === "transferring")).toBe(false); expect(h.updates.some((p) => p.phase === "transferring")).toBe(false);
await h.onUnload[0](); await h.onUnload[0]();
@@ -141,9 +157,9 @@ describe("useSyncGuardFeature", () => {
it("preserves offline access through manual bypass and never re-locks that catch-up", async () => { it("preserves offline access through manual bypass and never re-locks that catch-up", async () => {
const h = setup({ minimumFiles: 1 }); const h = setup({ minimumFiles: 1 });
await h.onResumed[0](); await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("first")]); await h.parseHandlers[0]([receivedNote("first")]);
h.finiteReplicationActivityCount.value = 0; h.couchDbCatchUpActivityCount.value = 0;
expect(h.updates[h.updates.length - 1]?.phase).toBe("interrupted"); expect(h.updates[h.updates.length - 1]?.phase).toBe("interrupted");
h.bypass(); h.bypass();
expect(h.updates[h.updates.length - 1]?.phase).toBe("idle"); expect(h.updates[h.updates.length - 1]?.phase).toBe("idle");
@@ -155,14 +171,14 @@ describe("useSyncGuardFeature", () => {
it("never shows a guard when disabled and releases an active guard when the setting is disabled", async () => { it("never shows a guard when disabled and releases an active guard when the setting is disabled", async () => {
const h = setup({ enabled: false, minimumFiles: 1 }); const h = setup({ enabled: false, minimumFiles: 1 });
await h.onResumed[0](); await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("first")]); await h.parseHandlers[0]([receivedNote("first")]);
expect(h.updates.some((p) => p.phase === "transferring")).toBe(false); expect(h.updates.some((p) => p.phase === "transferring")).toBe(false);
await h.onUnload[0](); await h.onUnload[0]();
const enabled = setup({ minimumFiles: 1 }); const enabled = setup({ minimumFiles: 1 });
await enabled.onResumed[0](); await enabled.onResumed[0]();
enabled.finiteReplicationActivityCount.value = 1; enabled.couchDbCatchUpActivityCount.value = 1;
await enabled.parseHandlers[0]([receivedNote("first")]); await enabled.parseHandlers[0]([receivedNote("first")]);
expect(enabled.updates[enabled.updates.length - 1]?.phase).toBe("transferring"); expect(enabled.updates[enabled.updates.length - 1]?.phase).toBe("transferring");
enabled.settings.syncGuardEnabled = false; enabled.settings.syncGuardEnabled = false;
@@ -173,11 +189,11 @@ describe("useSyncGuardFeature", () => {
it("observes an internal LiveSync reconnect even without an application resume notification", async () => { it("observes an internal LiveSync reconnect even without an application resume notification", async () => {
const h = setup({ minimumFiles: 1 }); const h = setup({ minimumFiles: 1 });
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("after-reconnect")]); await h.parseHandlers[0]([receivedNote("after-reconnect")]);
expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring"); expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring");
h.replicationStatics.value = { syncStatus: "COMPLETED" }; h.replicationStatics.value = { syncStatus: "COMPLETED" };
h.finiteReplicationActivityCount.value = 0; h.couchDbCatchUpActivityCount.value = 0;
expect(h.updates[h.updates.length - 1]?.phase).toBe("applying"); expect(h.updates[h.updates.length - 1]?.phase).toBe("applying");
await h.onUnload[0](); await h.onUnload[0]();
}); });
@@ -185,13 +201,13 @@ describe("useSyncGuardFeature", () => {
it("preserves an active guard across duplicate resume notifications", async () => { it("preserves an active guard across duplicate resume notifications", async () => {
const h = setup({ minimumFiles: 2 }); const h = setup({ minimumFiles: 2 });
await h.onResumed[0](); await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("first"), receivedNote("second")]); await h.parseHandlers[0]([receivedNote("first"), receivedNote("second")]);
expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring"); expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring");
await h.onResumed[0](); await h.onResumed[0]();
expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring"); expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring");
h.replicationStatics.value = { syncStatus: "COMPLETED" }; h.replicationStatics.value = { syncStatus: "COMPLETED" };
h.finiteReplicationActivityCount.value = 0; h.couchDbCatchUpActivityCount.value = 0;
expect(h.updates[h.updates.length - 1]?.phase).toBe("applying"); expect(h.updates[h.updates.length - 1]?.phase).toBe("applying");
await h.onUnload[0](); await h.onUnload[0]();
}); });
@@ -199,7 +215,7 @@ describe("useSyncGuardFeature", () => {
it("preserves accumulated files before the threshold when resume fires twice during one pull", async () => { it("preserves accumulated files before the threshold when resume fires twice during one pull", async () => {
const h = setup({ minimumFiles: 3 }); const h = setup({ minimumFiles: 3 });
await h.onResumed[0](); await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("one"), receivedNote("two")]); await h.parseHandlers[0]([receivedNote("one"), receivedNote("two")]);
await h.onResumed[0](); await h.onResumed[0]();
await h.parseHandlers[0]([receivedNote("three")]); await h.parseHandlers[0]([receivedNote("three")]);
@@ -213,11 +229,11 @@ describe("useSyncGuardFeature", () => {
it("keeps the guard after a received note could not be written to the Vault", async () => { it("keeps the guard after a received note could not be written to the Vault", async () => {
const h = setup({ minimumFiles: 1 }); const h = setup({ minimumFiles: 1 });
await h.onResumed[0](); await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("failed")]); await h.parseHandlers[0]([receivedNote("failed")]);
h.events.emitEvent(EVENT_VAULT_REFLECTION_FAILED); h.events.emitEvent(EVENT_VAULT_REFLECTION_FAILED);
h.replicationStatics.value = { syncStatus: "COMPLETED" }; h.replicationStatics.value = { syncStatus: "COMPLETED" };
h.finiteReplicationActivityCount.value = 0; h.couchDbCatchUpActivityCount.value = 0;
expect(h.updates[h.updates.length - 1]?.phase).toBe("interrupted"); expect(h.updates[h.updates.length - 1]?.phase).toBe("interrupted");
await h.onUnload[0](); await h.onUnload[0]();
}); });
@@ -225,7 +241,7 @@ describe("useSyncGuardFeature", () => {
it("responds immediately to settings saved by the UI without setting-realised", async () => { it("responds immediately to settings saved by the UI without setting-realised", async () => {
const h = setup({ minimumFiles: 20 }); const h = setup({ minimumFiles: 20 });
await h.onResumed[0](); await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("one"), receivedNote("two")]); await h.parseHandlers[0]([receivedNote("one"), receivedNote("two")]);
expect(h.updates[h.updates.length - 1]?.phase).toBe("observing"); expect(h.updates[h.updates.length - 1]?.phase).toBe("observing");
h.settings.syncGuardMinimumFiles = 2; h.settings.syncGuardMinimumFiles = 2;
@@ -240,12 +256,12 @@ describe("useSyncGuardFeature", () => {
it("does not re-lock after manual bypass when an internal reconnect starts", async () => { it("does not re-lock after manual bypass when an internal reconnect starts", async () => {
const h = setup({ minimumFiles: 1 }); const h = setup({ minimumFiles: 1 });
await h.onResumed[0](); await h.onResumed[0]();
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("first")]); await h.parseHandlers[0]([receivedNote("first")]);
expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring"); expect(h.updates[h.updates.length - 1]?.phase).toBe("transferring");
h.bypass(); h.bypass();
h.finiteReplicationActivityCount.value = 0; h.couchDbCatchUpActivityCount.value = 0;
h.finiteReplicationActivityCount.value = 1; h.couchDbCatchUpActivityCount.value = 1;
await h.parseHandlers[0]([receivedNote("second")]); await h.parseHandlers[0]([receivedNote("second")]);
expect(h.updates[h.updates.length - 1]?.phase).toBe("idle"); expect(h.updates[h.updates.length - 1]?.phase).toBe("idle");
await h.onUnload[0](); await h.onUnload[0]();