feat: show directional CouchDB comparison progress

src/modules/features/ReplicationStatusLabel.ts

@@ -13,13 +13,25 @@ import {
 
 export const FAST_FORWARD_COUNT_LINGER_MS = 3_000;
 
-function readPositiveCheckedChanges(source: ReactiveValue<ReplicationStatics>): number | undefined {
-    const count = source.value.couchDB?.pull?.checkedChanges;
+export type FastForwardCheckedChanges = {
+    push?: number;
+    pull?: number;
+};
+
+function readPositiveCheckedChanges(
+    source: ReactiveValue<ReplicationStatics>,
+    direction: keyof FastForwardCheckedChanges
+): number | undefined {
+    const count = source.value.couchDB?.[direction]?.checkedChanges;
     return typeof count === "number" && Number.isFinite(count) && count > 0 ? count : undefined;
 }
 
+function copyCheckedChanges(counts: FastForwardCheckedChanges): FastForwardCheckedChanges {
+    return { ...counts };
+}
+
 /**
- * Briefly displays changing CouchDB pull-feed comparison counts.
+ * Briefly displays changing CouchDB push- and pull-feed comparison counts.
  *
  * Unrelated replication-stat changes do not extend the linger or redisplay an
  * expired count. A positive initial snapshot is displayed to late subscribers.
@@ -27,46 +39,61 @@ function readPositiveCheckedChanges(source: ReactiveValue<ReplicationStatics>):
 export function createFastForwardCheckedChangesDisplay(
     source: ReactiveValue<ReplicationStatics>,
     lingerMs = FAST_FORWARD_COUNT_LINGER_MS
-): DisposableReactiveValue<number | undefined> {
-    const displayed = reactiveSource<number | undefined>(undefined);
+): DisposableReactiveValue<FastForwardCheckedChanges> {
+    const displayed = reactiveSource<FastForwardCheckedChanges>({});
     const linger = Math.max(0, lingerMs);
-    let lastObserved = readPositiveCheckedChanges(source);
-    let clearTimer: number | undefined;
+    const directions: (keyof FastForwardCheckedChanges)[] = ["push", "pull"];
+    const lastObserved: FastForwardCheckedChanges = {};
+    const clearTimers = new Map<keyof FastForwardCheckedChanges, number>();
     let disposed = false;
 
-    const cancelClear = () => {
-        if (clearTimer === undefined) return;
-        compatGlobal.clearTimeout(clearTimer);
-        clearTimer = undefined;
+    const cancelClear = (direction: keyof FastForwardCheckedChanges) => {
+        const timer = clearTimers.get(direction);
+        if (timer === undefined) return;
+        compatGlobal.clearTimeout(timer);
+        clearTimers.delete(direction);
     };
-    const scheduleClear = () => {
-        clearTimer = compatGlobal.setTimeout(() => {
-            clearTimer = undefined;
+    const scheduleClear = (direction: keyof FastForwardCheckedChanges) => {
+        const timer = compatGlobal.setTimeout(() => {
+            clearTimers.delete(direction);
             if (disposed) return;
-            if (readPositiveCheckedChanges(source) !== lastObserved) {
+            if (readPositiveCheckedChanges(source, direction) !== lastObserved[direction]) {
                 update();
                 return;
             }
-            displayed.value = undefined;
+            const nextDisplayed = copyCheckedChanges(displayed.value);
+            delete nextDisplayed[direction];
+            displayed.value = nextDisplayed;
         }, linger);
+        clearTimers.set(direction, timer);
     };
     const update = () => {
         if (disposed) return;
-        const nextCount = readPositiveCheckedChanges(source);
-        if (nextCount === lastObserved) return;
-        lastObserved = nextCount;
-        cancelClear();
-        if (nextCount === undefined) {
-            displayed.value = undefined;
-            return;
+        const nextDisplayed = copyCheckedChanges(displayed.value);
+        let changed = false;
+        for (const direction of directions) {
+            const nextCount = readPositiveCheckedChanges(source, direction);
+            if (nextCount === lastObserved[direction]) continue;
+            lastObserved[direction] = nextCount;
+            cancelClear(direction);
+            if (nextCount === undefined) {
+                delete nextDisplayed[direction];
+                changed = true;
+                continue;
+            }
+            nextDisplayed[direction] = nextCount;
+            scheduleClear(direction);
+            changed = true;
         }
-        displayed.value = nextCount;
-        scheduleClear();
+        if (changed) displayed.value = nextDisplayed;
     };
 
-    if (lastObserved !== undefined) {
-        displayed.value = lastObserved;
-        scheduleClear();
+    for (const direction of directions) {
+        const count = readPositiveCheckedChanges(source, direction);
+        if (count === undefined) continue;
+        lastObserved[direction] = count;
+        displayed.value = { ...displayed.value, [direction]: count };
+        scheduleClear(direction);
     }
     source.onChanged(update);
 
@@ -83,7 +110,7 @@ export function createFastForwardCheckedChangesDisplay(
         dispose() {
             if (disposed) return;
             disposed = true;
-            cancelClear();
+            for (const direction of directions) cancelClear(direction);
             source.offChanged(update);
         },
     };
@@ -113,7 +140,7 @@ function remainingSequenceLabel(
 export function projectReplicationStatusLabel(
     stats: ReplicationStatics,
     networkStatus: RemoteActivityStatus,
-    fastForwardCount?: number
+    fastForwardCounts?: FastForwardCheckedChanges
 ): ReplicationStatusProjection {
     const { sent, arrived, maxPullSeq, maxPushSeq, lastSyncPullSeq, lastSyncPushSeq } = stats;
     const networkActivity = formatRemoteActivityStatusLabel(networkStatus);
@@ -154,14 +181,19 @@ export function projectReplicationStatusLabel(
         (couchDB.pull?.checking === true || couchDB.push?.checking === true);
     if (isFastForwardChecking) icon = "⏩";
 
-    const visibleFastForwardCount =
-        couchDB !== undefined &&
-        typeof fastForwardCount === "number" &&
-        Number.isFinite(fastForwardCount) &&
-        fastForwardCount > 0
-            ? fastForwardCount
-            : undefined;
-    const fastForwardLabel = visibleFastForwardCount === undefined ? "" : " (⏩ " + visibleFastForwardCount + ")";
+    const visibleFastForwardCounts: FastForwardCheckedChanges = {};
+    if (couchDB !== undefined) {
+        for (const direction of ["push", "pull"] as const) {
+            const count = fastForwardCounts?.[direction];
+            if (typeof count === "number" && Number.isFinite(count) && count > 0) {
+                visibleFastForwardCounts[direction] = count;
+            }
+        }
+    }
+    const fastForwardLabels = (["push", "pull"] as const)
+        .filter((direction) => visibleFastForwardCounts[direction] !== undefined)
+        .map((direction) => ` (⏩${direction === "push" ? "↑" : "↓"} ${visibleFastForwardCounts[direction]})`);
+    const fastForwardLabel = fastForwardLabels.join("");
 
     const parts: StatusBarPart[] = [];
     if (networkStatus.trackedRequestCount > 0) {
@@ -177,13 +209,15 @@ export function projectReplicationStatusLabel(
         parts.push({ key: "network", text: networkActivity });
     }
     parts.push({ key: "sync", text: "Sync: " + icon });
-    if (visibleFastForwardCount !== undefined) {
+    for (const direction of ["push", "pull"] as const) {
+        const count = visibleFastForwardCounts[direction];
+        if (count === undefined) continue;
         parts.push({
-            key: "ff",
-            text: " (⏩ ",
-            value: visibleFastForwardCount,
+            key: `ff-${direction}`,
+            text: ` (⏩${direction === "push" ? "↑" : "↓"} `,
+            value: count,
             suffix: ")",
-            title: "Compared CouchDB feed entries",
+            title: `Compared CouchDB ${direction} feed entries`,
         });
     }
     parts.push({ key: "sent", text: " ↑ ", value: sent });

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

@@ -90,18 +90,22 @@ describe("projectReplicationStatusLabel", () => {
             const idleWithCount = projectReplicationStatusLabel(
                 stats({ syncStatus, couchDB: couchDBProgress("idle", 19, true) }),
                 noNetworkActivity,
-                19
+                { push: 19, pull: 19 }
             );
             expect(idleWithCount.icon).toBe(icon);
-            expect(idleWithCount.parts.some((part) => part.key === "ff")).toBe(true);
+            expect(idleWithCount.parts.some((part) => part.key === "ff-push")).toBe(true);
+            expect(idleWithCount.parts.some((part) => part.key === "ff-pull")).toBe(true);
 
-            const withoutCouchDB = projectReplicationStatusLabel(stats({ syncStatus }), noNetworkActivity, 19);
+            const withoutCouchDB = projectReplicationStatusLabel(stats({ syncStatus }), noNetworkActivity, {
+                push: 19,
+                pull: 19,
+            });
             expect(withoutCouchDB.icon).toBe(icon);
-            expect(withoutCouchDB.parts.some((part) => part.key === "ff")).toBe(false);
+            expect(withoutCouchDB.parts.some((part) => part.key.startsWith("ff-"))).toBe(false);
         }
 
         const noFastForward = projectReplicationStatusLabel(stats(), noNetworkActivity);
-        expect(noFastForward.parts.some((part) => part.key === "ff")).toBe(false);
+        expect(noFastForward.parts.some((part) => part.key.startsWith("ff-"))).toBe(false);
         expect(noFastForward.message).toBe("Sync: ⚡ ↑ 0 ↓ 0");
     });
 
@@ -123,18 +127,40 @@ describe("projectReplicationStatusLabel", () => {
                 couchDB: couchDBProgress("continuous", 1200, false),
             }),
             noNetworkActivity,
-            1200
+            { pull: 1200 }
         );
         expect(transferred.icon).toBe("⚡");
-        expect(transferred.message).toBe("Sync: ⚡ (⏩ 1200) ↑ 0 ↓ 7");
-        expect(transferred.parts.map((part) => part.key)).toEqual(["sync", "ff", "sent", "arrived"]);
+        expect(transferred.message).toBe("Sync: ⚡ (⏩↓ 1200) ↑ 0 ↓ 7");
+        expect(transferred.parts.map((part) => part.key)).toEqual(["sync", "ff-pull", "sent", "arrived"]);
         expect(transferred.parts[1]).toMatchObject({
-            key: "ff",
-            text: " (⏩ ",
+            key: "ff-pull",
+            text: " (⏩↓ ",
             value: 1200,
             suffix: ")",
+            title: "Compared CouchDB pull feed entries",
         });
 
+        const pushChecking = projectReplicationStatusLabel(
+            stats({
+                couchDB: {
+                    phase: "continuous",
+                    push: { checkedChanges: 71975, completedBatches: 2878, checking: true },
+                },
+            }),
+            noNetworkActivity,
+            { push: 71975 }
+        );
+        expect(pushChecking.icon).toBe("⏩");
+        expect(pushChecking.message).toBe("Sync: ⏩ (⏩↑ 71975) ↑ 0 ↓ 0");
+        expect(pushChecking.parts[1]).toMatchObject({ key: "ff-push", value: 71975 });
+
+        const bothChecking = projectReplicationStatusLabel(
+            stats({ couchDB: couchDBProgress("continuous", 10, true) }),
+            noNetworkActivity,
+            { push: 71975, pull: 10 }
+        );
+        expect(bothChecking.message).toBe("Sync: ⏩ (⏩↑ 71975) (⏩↓ 10) ↑ 0 ↓ 0");
+
         const idle = projectReplicationStatusLabel(
             stats({
                 syncStatus: "CONNECTED",
@@ -142,10 +168,10 @@ describe("projectReplicationStatusLabel", () => {
                 couchDB: couchDBProgress("idle", 1200, false),
             }),
             noNetworkActivity,
-            1200
+            { pull: 1200 }
         );
         expect(idle.icon).toBe("⚡");
-        expect(idle.message).toBe("Sync: ⚡ (⏩ 1200) ↑ 0 ↓ 7");
+        expect(idle.message).toBe("Sync: ⚡ (⏩↓ 1200) ↑ 0 ↓ 7");
     });
 
     it("keeps the error icon ahead of an active CouchDB checking indicator", () => {
@@ -154,7 +180,7 @@ describe("projectReplicationStatusLabel", () => {
                 syncStatus: "ERRORED",
                 couchDB: {
                     ...couchDBProgress("one-shot", 2, true),
-                    push: { checkedChanges: 0, completedBatches: 0, checking: true },
+                    push: { checkedChanges: 2, completedBatches: 1, checking: true },
                 },
             }),
             noNetworkActivity
@@ -181,41 +207,41 @@ describe("createFastForwardCheckedChangesDisplay", () => {
         );
         const display = createFastForwardCheckedChangesDisplay(source);
 
-        expect(display.value).toBe(1200);
+        expect(display.value).toEqual({ pull: 1200 });
         vi.advanceTimersByTime(1000);
         source.value = stats({
             sent: 3,
             couchDB: couchDBProgress("idle", 1200, false),
         });
-        expect(display.value).toBe(1200);
+        expect(display.value).toEqual({ pull: 1200 });
         vi.advanceTimersByTime(FAST_FORWARD_COUNT_LINGER_MS - 1001);
-        expect(display.value).toBe(1200);
+        expect(display.value).toEqual({ pull: 1200 });
 
         source.value = stats({
             sent: 4,
             couchDB: couchDBProgress("idle", 1200, false),
         });
         vi.advanceTimersByTime(1);
-        expect(display.value).toBeUndefined();
+        expect(display.value).toEqual({});
 
         source.value = stats({
             syncStatus: "PAUSED",
             couchDB: couchDBProgress("continuous", 1200, true),
         });
-        expect(display.value).toBeUndefined();
+        expect(display.value).toEqual({});
 
         source.value = stats({
             couchDB: couchDBProgress("continuous", 1201, true),
         });
-        expect(display.value).toBe(1201);
+        expect(display.value).toEqual({ pull: 1201 });
 
         source.value = stats({
             couchDB: couchDBProgress("one-shot", 0, true),
         });
-        expect(display.value).toBeUndefined();
+        expect(display.value).toEqual({});
 
         source.value = stats();
-        expect(display.value).toBeUndefined();
+        expect(display.value).toEqual({});
         display.dispose();
     });
 
@@ -227,16 +253,46 @@ describe("createFastForwardCheckedChangesDisplay", () => {
         vi.advanceTimersByTime(FAST_FORWARD_COUNT_LINGER_MS);
         source.value = stats({ couchDB: couchDBProgress("continuous", 6, true) });
 
-        expect(display.value).toBe(5);
+        expect(display.value).toEqual({ pull: 5 });
     });
 
     it("does not display progress for providers without CouchDB replication state", () => {
         const source = reactiveSource(stats());
         const display = createFastForwardCheckedChangesDisplay(source);
 
-        expect(display.value).toBeUndefined();
+        expect(display.value).toEqual({});
         source.value = stats({ couchDB: { phase: "continuous" } });
-        expect(display.value).toBeUndefined();
+        expect(display.value).toEqual({});
+        display.dispose();
+    });
+
+    it("tracks push and pull independently and expires only the direction that stopped changing", () => {
+        const source = reactiveSource(
+            stats({
+                couchDB: {
+                    phase: "catch-up",
+                    pull: { checkedChanges: 4, completedBatches: 1, checking: true },
+                    push: { checkedChanges: 40, completedBatches: 2, checking: true },
+                },
+            })
+        );
+        const display = createFastForwardCheckedChangesDisplay(source, 100);
+
+        expect(display.value).toEqual({ push: 40, pull: 4 });
+        vi.advanceTimersByTime(50);
+        source.value = stats({
+            couchDB: {
+                phase: "catch-up",
+                pull: { checkedChanges: 4, completedBatches: 1, checking: true },
+                push: { checkedChanges: 41, completedBatches: 2, checking: true },
+            },
+        });
+        expect(display.value).toEqual({ push: 41, pull: 4 });
+
+        vi.advanceTimersByTime(50);
+        expect(display.value).toEqual({ push: 41 });
+        vi.advanceTimersByTime(50);
+        expect(display.value).toEqual({});
         display.dispose();
     });
 });