feat: show P2P transfer and comparison progress

Show per-peer document revision counts and brief directional comparison
labels in both P2P panes and the editor status. Use the P2P projection for
the main status bar when P2P is the selected remote.

Retain labelled sequence progress for peers without statistics, and keep
adjunct P2P counters separate from the selected remote.

Requires the additive P2P statistics API in
vrtmrz/livesync-commonlib@be7bd0988e.
Consumer checks passed against a packed candidate; pin the published package
before integration.

src/modules/features/ModuleLog.ts

@@ -10,6 +10,7 @@ import {
     LOG_LEVEL_INFO,
     LOG_LEVEL_VERBOSE,
     PREFIXMD_LOGFILE,
+    REMOTE_P2P,
     type LOG_LEVEL,
 } from "@vrtmrz/livesync-commonlib/compat/common/types";
 import { scheduleTask } from "octagonal-wheels/concurrency/task";
@@ -37,6 +38,7 @@ import { LogPaneView, VIEW_TYPE_LOG } from "./Log/LogPaneView.ts";
 import { withLock } from "octagonal-wheels/concurrency/lock";
 import { $msg } from "@/common/translation";
 import { P2PLogCollector } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/P2PLogCollector";
+import { EVENT_SETTING_SAVED } from "@vrtmrz/livesync-commonlib/compat/events/coreEvents";
 import { REMOTE_REQUEST_ACTIVITY_MINIMUM_VISIBLE_MS, getTrackedRequestCount } from "./RemoteActivityStatus.ts";
 import {
     createChunkFetchCounterLabel,
@@ -44,6 +46,7 @@ import {
     createPaddedCounterLabel,
 } from "./StatusBarDisplay.ts";
 import { createFastForwardCheckedChangesDisplay, projectReplicationStatusLabel } from "./ReplicationStatusLabel.ts";
+import { createP2PProgressDisplay, selectReplicationStatus } from "./P2PStatusLabel.ts";
 import { createStatusBarRenderer, type StatusBarPart, type StatusBarRenderer } from "./StatusBarRenderer.ts";
 import type { LiveSyncCore } from "@/main.ts";
 import { LiveSyncError } from "@vrtmrz/livesync-commonlib/compat/common/LSError";
@@ -204,6 +207,14 @@ export class ModuleLog extends AbstractObsidianModule {
         const fastForwardCount = registerDisplay(
             createFastForwardCheckedChangesDisplay(this.services.replicator.replicationStatics)
         );
+        const p2pProgress = registerDisplay(createP2PProgressDisplay(this.p2pLogCollector.p2pReplicationStats));
+        const p2pIsPrimary = reactiveSource(this.settings.remoteType === REMOTE_P2P);
+        this.plugin.register(
+            this.services.context.events.onEvent(EVENT_SETTING_SAVED, (settings) => {
+                p2pIsPrimary.value = settings.remoteType === REMOTE_P2P;
+            })
+        );
+        this.plugin.register(() => this.p2pLogCollector.dispose());
         const replicationStatLabel = computed(() =>
             projectReplicationStatusLabel(
                 this.services.replicator.replicationStatics.value,
@@ -220,10 +231,13 @@ export class ModuleLog extends AbstractObsidianModule {
         const counterPart = (key: string, mark: string, value: number | undefined): StatusBarPart[] =>
             value === undefined ? [] : [{ key, text: " " + mark, value }];
         const statusLineLabel = computed(() => {
-            const replication = replicationStatLabel();
+            const { primary: replication, auxiliary: p2p } = selectReplicationStatus(
+                replicationStatLabel(),
+                p2pProgress.value,
+                p2pIsPrimary.value
+            );
             const queued = queueCountLabel();
             const waiting = waitingLabel();
-            const p2p = this.p2pLogCollector.p2pReplicationLine.value;
             const firstLineParts = [
                 ...replication.parts,
                 ...counterPart("file-processing", "⏳", labelProc.count.value),
@@ -239,10 +253,11 @@ export class ModuleLog extends AbstractObsidianModule {
                 ...counterPart("queue-conflicts", "🔩", labelConflictProcessCount.count.value),
             ];
             return {
-                message: replication.message + waiting + queued + (p2p === "" ? "" : "\n" + p2p),
-                parts: p2p === "" ? firstLineParts : [...firstLineParts, { key: "p2p", text: "\n" + p2p }],
+                message: replication.message + waiting + queued + (p2p.message === "" ? "" : "\n" + p2p.message),
+                parts:
+                    p2p.message === "" ? firstLineParts : [...firstLineParts, { key: "p2p", text: "\n" }, ...p2p.parts],
                 statusBarParts: firstLineParts,
-                hasP2P: p2p !== "",
+                hasP2P: p2p.message !== "",
             };
         });
 

src/modules/features/P2PStatusLabel.ts

@@ -0,0 +1,178 @@
+import { reactiveSource, type ReactiveValue } from "octagonal-wheels/dataobject/reactive";
+import type { P2PReplicationProgress } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/TrysteroReplicator";
+import type { P2PDiagnostics } from "@vrtmrz/livesync-commonlib/p2p";
+import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions";
+import { FAST_FORWARD_COUNT_LINGER_MS } from "./ReplicationStatusLabel.ts";
+import type { DisposableReactiveValue } from "./StatusBarDisplay.ts";
+import type { StatusBarPart } from "./StatusBarRenderer.ts";
+
+type PeerStatusProjection = { message: string; parts: StatusBarPart[] };
+export type P2PStatusProjection = PeerStatusProjection & {
+    peers: ReadonlyMap<string, PeerStatusProjection>;
+};
+
+export function selectReplicationStatus(
+    primary: PeerStatusProjection,
+    p2p: P2PStatusProjection,
+    p2pIsPrimary: boolean
+): { primary: PeerStatusProjection; auxiliary: PeerStatusProjection } {
+    if (!p2pIsPrimary) return { primary, auxiliary: p2p };
+    const parts: StatusBarPart[] = p2p.parts.length
+        ? [{ key: "sync", text: "Sync: " }, ...p2p.parts]
+        : [{ key: "sync", text: "Sync: 💤 P2P" }];
+    return { primary: { message: partsMessage(parts), parts }, auxiliary: { message: "", parts: [] } };
+}
+
+const directions = ["sending", "fetching"] as const;
+type Direction = (typeof directions)[number];
+
+function directionParts(
+    peer: P2PReplicationProgress,
+    direction: Direction,
+    checkedChanges: number | undefined
+): StatusBarPart[] {
+    const progress = peer[direction];
+    const key = `p2p-${peer.peerId}-${direction}`;
+    const arrow = direction === "sending" ? "↑" : "↓";
+    const parts: StatusBarPart[] = progress.statistics
+        ? [
+              {
+                  key,
+                  text: ` ${arrow} `,
+                  value: progress.statistics.documents,
+                  title: direction === "sending" ? "Sent document revisions" : "Received document revisions",
+              },
+          ]
+        : [
+              {
+                  key,
+                  text: ` ${arrow} seq `,
+                  value: progress.current,
+                  suffix: progress.current === progress.max ? "" : ` (${progress.max})`,
+              },
+          ];
+    if (checkedChanges !== undefined) {
+        parts.push({
+            key: `${key}-ff`,
+            text: ` ⏩${arrow} `,
+            value: checkedChanges,
+            title: "Compared changes-feed entries",
+        });
+    }
+    return parts;
+}
+
+function partsMessage(parts: readonly StatusBarPart[]): string {
+    return parts.map((part) => part.text + (part.value === undefined ? "" : part.value + (part.suffix ?? ""))).join("");
+}
+
+/** Retains transfer counts while independently expiring each peer's directional comparison labels. */
+export function createP2PProgressDisplay(
+    source: ReactiveValue<readonly P2PReplicationProgress[]>,
+    lingerMs = FAST_FORWARD_COUNT_LINGER_MS
+): DisposableReactiveValue<P2PStatusProjection> {
+    const displayed = reactiveSource<P2PStatusProjection>({ message: "", parts: [], peers: new Map() });
+    const observed = new Map<string, number>();
+    const visible = new Map<string, number>();
+    const timers = new Map<string, number>();
+    let disposed = false;
+
+    const cancelTimer = (key: string) => {
+        const timer = timers.get(key);
+        if (timer !== undefined) compatGlobal.clearTimeout(timer);
+        timers.delete(key);
+    };
+    const render = () => {
+        const peers = new Map<string, PeerStatusProjection>();
+        const parts: StatusBarPart[] = [];
+        for (const peer of source.value) {
+            const active = peer.fetching.isActive || peer.sending.isActive;
+            const peerParts = directions.flatMap((direction) =>
+                directionParts(peer, direction, visible.get(`${peer.peerId}:${direction}`))
+            );
+            const namedParts = [
+                { key: `p2p-${peer.peerId}`, text: `${active ? "⚡" : "💤"} ${peer.peerName}` },
+                ...peerParts,
+            ];
+            peers.set(peer.peerId, { message: partsMessage(peerParts).trim(), parts: peerParts });
+            if (parts.length) parts.push({ key: `p2p-${peer.peerId}-separator`, text: " / " });
+            parts.push(...namedParts);
+        }
+        displayed.value = { message: partsMessage(parts), parts, peers };
+    };
+    const update = () => {
+        if (disposed) return;
+        const present = new Set<string>();
+        for (const peer of source.value) {
+            for (const direction of directions) {
+                const key = `${peer.peerId}:${direction}`;
+                present.add(key);
+                const count = peer[direction].statistics?.checkedChanges;
+                if (count === observed.get(key)) continue;
+                cancelTimer(key);
+                visible.delete(key);
+                if (count === undefined) observed.delete(key);
+                else observed.set(key, count);
+                if (count === undefined || !Number.isFinite(count) || count <= 0) continue;
+                visible.set(key, count);
+                timers.set(
+                    key,
+                    compatGlobal.setTimeout(
+                        () => {
+                            timers.delete(key);
+                            if (disposed) return;
+                            visible.delete(key);
+                            render();
+                        },
+                        Math.max(0, lingerMs)
+                    )
+                );
+            }
+        }
+        for (const key of observed.keys()) {
+            if (present.has(key)) continue;
+            cancelTimer(key);
+            observed.delete(key);
+            visible.delete(key);
+        }
+        render();
+    };
+    source.onChanged(update);
+    update();
+    return {
+        get value() {
+            return displayed.value;
+        },
+        onChanged: displayed.onChanged,
+        offChanged: displayed.offChanged,
+        dispose() {
+            if (disposed) return;
+            disposed = true;
+            source.offChanged(update);
+            for (const key of timers.keys()) cancelTimer(key);
+        },
+    };
+}
+
+/** Bridges diagnostics snapshots to the same display used by the editor and status bar. */
+export function observeP2PDiagnosticsProgress(
+    diagnostics: P2PDiagnostics,
+    listener: (snapshot: P2PStatusProjection) => void
+): () => void {
+    const source = reactiveSource<readonly P2PReplicationProgress[]>(diagnostics.getReplicationProgress?.() ?? []);
+    const display = createP2PProgressDisplay(source);
+    const update = () => listener(display.value);
+    display.onChanged(update);
+    const unsubscribe = diagnostics.onReplicationProgress?.((snapshot) => {
+        source.value = snapshot;
+    });
+    update();
+    let disposed = false;
+    return () => {
+        if (disposed) return;
+        disposed = true;
+        unsubscribe?.();
+        display.offChanged(update);
+        display.dispose();
+    };
+}

src/modules/features/P2PStatusLabel.unit.spec.ts

@@ -0,0 +1,148 @@
+import { reactiveSource } from "octagonal-wheels/dataobject/reactive";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { P2PReplicationProgress } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/TrysteroReplicator";
+import type { P2PDiagnostics } from "@vrtmrz/livesync-commonlib/p2p";
+import {
+    observeP2PDiagnosticsProgress,
+    createP2PProgressDisplay,
+    selectReplicationStatus,
+    type P2PStatusProjection,
+} from "./P2PStatusLabel.ts";
+
+function peer(peerId = "peer-a"): P2PReplicationProgress {
+    return {
+        peerId,
+        peerName: peerId,
+        sending: { current: 800, max: 900, isActive: true, statistics: { documents: 2, checkedChanges: 40 } },
+        fetching: { current: 600, max: 700, isActive: true, statistics: { documents: 0, checkedChanges: 33 } },
+    };
+}
+
+describe("P2P progress display", () => {
+    const disposables: { dispose(): void }[] = [];
+    beforeEach(() => vi.useFakeTimers());
+    afterEach(() => {
+        for (const display of disposables.splice(0)) display.dispose();
+        vi.useRealTimers();
+    });
+    function displayFor(initial: P2PReplicationProgress[] = [peer()]) {
+        const source = reactiveSource<readonly P2PReplicationProgress[]>(initial);
+        const display = createP2PProgressDisplay(source);
+        disposables.push(display);
+        return { source, display };
+    }
+
+    it("shows transferred revisions independently from seq and zero-transfer comparisons", () => {
+        const { display, source } = displayFor();
+        expect(display.value.message).toBe("⚡ peer-a ↑ 2 ⏩↑ 40 ↓ 0 ⏩↓ 33");
+        expect(display.value.parts.find((part) => part.key === "p2p-peer-a-fetching")).toMatchObject({ value: 0 });
+        vi.advanceTimersByTime(3_000);
+        expect(display.value.message).toBe("⚡ peer-a ↑ 2 ↓ 0");
+        expect(source.value[0].fetching.statistics).toEqual({ documents: 0, checkedChanges: 33 });
+    });
+
+    it("labels legacy seq honestly and distinguishes unavailable counts from measured zero", () => {
+        const legacy = peer();
+        delete legacy.sending.statistics;
+        const { display } = displayFor([legacy]);
+        expect(display.value.message).toContain("↑ seq 800 (900)");
+        expect(display.value.message).toContain("↓ 0 ⏩↓ 33");
+        expect(display.value.message).not.toContain("⏩↑");
+    });
+
+    it("expires independently by direction and peer without extending on unrelated activity", () => {
+        const { display, source } = displayFor([peer(), peer("peer-b")]);
+        vi.advanceTimersByTime(2_000);
+        const next = peer();
+        next.fetching.statistics = { documents: 0, checkedChanges: 66 };
+        source.value = [next, peer("peer-b")];
+        vi.advanceTimersByTime(1_000);
+        expect(display.value.peers.get("peer-a")?.message).toBe("↑ 2 ↓ 0 ⏩↓ 66");
+        expect(display.value.peers.get("peer-b")?.message).toBe("↑ 2 ↓ 0");
+        source.value = [{ ...next, peerName: "renamed" }, peer("peer-b")];
+        vi.advanceTimersByTime(2_000);
+        expect(display.value.message).not.toContain("⏩");
+        source.value = [next, peer("peer-b")];
+        expect(display.value.message).not.toContain("⏩");
+    });
+
+    it("starts a fresh label lifetime after a counter reset and retains terminal transfer totals", () => {
+        const { display, source } = displayFor();
+        const reset = peer();
+        reset.fetching.statistics = { documents: 0, checkedChanges: 0 };
+        source.value = [reset];
+        expect(display.value.message).not.toContain("⏩↓");
+        vi.advanceTimersByTime(3_000);
+        const completed = peer();
+        completed.sending.isActive = false;
+        completed.fetching.isActive = false;
+        source.value = [completed];
+        expect(display.value.message).toBe("💤 peer-a ↑ 2 ↓ 0 ⏩↓ 33");
+        vi.advanceTimersByTime(3_000);
+        expect(display.value.message).toBe("💤 peer-a ↑ 2 ↓ 0");
+    });
+
+    it("removes labels and pending timers on disconnect, and detaches on disposal", () => {
+        const { display, source } = displayFor();
+        source.value = [];
+        expect(display.value.message).toBe("");
+        expect(vi.getTimerCount()).toBe(0);
+        source.value = [peer()];
+        expect(display.value.message).toContain("⏩↓ 33");
+        display.dispose();
+        const last = display.value;
+        source.value = [];
+        vi.advanceTimersByTime(5_000);
+        expect(display.value).toBe(last);
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it("takes a late diagnostics snapshot, follows updates, and unsubscribes for either pane", () => {
+        let listener: ((snapshot: readonly P2PReplicationProgress[]) => void) | undefined;
+        const unsubscribe = vi.fn();
+        const diagnostics: P2PDiagnostics = {
+            requestStatus: vi.fn(),
+            getPeerConnectionMetrics: vi.fn(),
+            getReplicationProgress: () => [peer()],
+            onReplicationProgress: (handler) => {
+                listener = handler;
+                return unsubscribe;
+            },
+        };
+        let snapshot: P2PStatusProjection | undefined;
+        const dispose = observeP2PDiagnosticsProgress(diagnostics, (value) => {
+            snapshot = value;
+        });
+        disposables.push({ dispose });
+        expect(snapshot?.peers.get("peer-a")?.message).toContain("↓ 0 ⏩↓ 33");
+        listener?.([]);
+        expect(snapshot?.peers.size).toBe(0);
+        dispose();
+        expect(unsubscribe).toHaveBeenCalledOnce();
+    });
+
+    it("supports diagnostics from an older service without progress methods", () => {
+        const listener = vi.fn();
+        const dispose = observeP2PDiagnosticsProgress(
+            { requestStatus: vi.fn(), getPeerConnectionMetrics: vi.fn() },
+            listener
+        );
+        disposables.push({ dispose });
+        expect(listener).toHaveBeenLastCalledWith({ message: "", parts: [], peers: new Map() });
+    });
+
+    it("uses P2P for the Only P2P primary display and preserves the CouchDB display when auxiliary", () => {
+        const { display } = displayFor();
+        const couchDB = { message: "Sync: ⚡ ↑ 12 ↓ 34", parts: [{ key: "sent", text: " ↑ ", value: 12 }] };
+        const auxiliary = selectReplicationStatus(couchDB, display.value, false);
+        expect(auxiliary.primary).toBe(couchDB);
+        expect(auxiliary.auxiliary.message).toContain("↑ 2");
+        const onlyP2P = selectReplicationStatus(couchDB, display.value, true);
+        expect(onlyP2P.primary.message).toBe("Sync: ⚡ peer-a ↑ 2 ⏩↑ 40 ↓ 0 ⏩↓ 33");
+        expect(onlyP2P.auxiliary.message).toBe("");
+        expect(couchDB.parts[0].value).toBe(12);
+        expect(
+            selectReplicationStatus(couchDB, { message: "", parts: [], peers: new Map() }, true).primary.message
+        ).toBe("Sync: 💤 P2P");
+    });
+});