diff --git a/desktop/src/features/channels/readState/readStateManager.test.mjs b/desktop/src/features/channels/readState/readStateManager.test.mjs index dfeddb6661..80f36a7ee7 100644 --- a/desktop/src/features/channels/readState/readStateManager.test.mjs +++ b/desktop/src/features/channels/readState/readStateManager.test.mjs @@ -47,6 +47,48 @@ function makeLocalStorage() { }); } +/** + * Build a FenceHandle-shaped fake for use in test relay objects. + * + * @param {object} opts + * @param {boolean} opts.eose — if true, `established` resolves immediately (EOSE received). + * @param {boolean} opts.lapsesAfterEose — if true, `lapsed` becomes true after established resolves. + * @param {boolean} opts.lapseBeforeEose — if true, sets lapsed=true and resolves established together. + * @param {(ev: object) => void} [opts.captureHandler] — called with the onEvent handler so tests can deliver events. + */ +function makeFenceHandle({ + eose = true, + lapsesAfterEose = false, + lapseBeforeEose = false, +} = {}) { + let lapsed = lapseBeforeEose; + let resolveEstablished; + const established = new Promise((r) => { + resolveEstablished = r; + }); + + if (lapseBeforeEose) { + // Lapsed before EOSE: resolve immediately with lapsed=true. + resolveEstablished(); + } else if (eose) { + resolveEstablished(); + if (lapsesAfterEose) lapsed = true; + } + // If neither, `established` never resolves (hung fence — caller will lapse via reconnect). + + return { + established, + get lapsed() { + return lapsed; + }, + unsubscribe: async () => {}, + /** For tests that need to trigger a lapse mid-enumeration. */ + _lapse() { + lapsed = true; + }, + }; +} + const threadKey = `thread:${"a".repeat(64)}`; const channelKey = "channel-1"; const channelResolver = (ctx) => @@ -528,7 +570,11 @@ test("publishSplitSlots_noopSuppression_skipsWhenUnchanged", async () => { const fakeRelay = { fetchEvents: async () => [], publishEvent: async () => {}, - subscribeLive: () => () => {}, + subscribeFenced: async (_filter, _onEvent) => + makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, }; const pubkey = "b".repeat(64); @@ -585,12 +631,18 @@ test("publishSplitSlots_noopSuppression_skipsWhenUnchanged", async () => { // ── NIP-RS override layer: mandatory acceptance tests ───────────────────────── // Helper: build a ReadStateManager with mocked relay and localStorage. +// subscribeFenced returns an immediately-established fence (happy path). +// subscribeLive is still wired for the live subscription path. function makeManager(pubkey = "a".repeat(64)) { globalThis.window.localStorage = makeLocalStorage(); const fakeRelay = { fetchEvents: async () => [], publishEvent: async () => {}, - subscribeLive: () => () => {}, + subscribeFenced: async (_filter, _onEvent) => + makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, }; return new ReadStateManager(pubkey, fakeRelay); } @@ -734,8 +786,9 @@ test("splitContextsIntoBudgetedSlots_escapedFrontierKeyStaysWithItsOverrideGroup } }); -// ── Test 2: NIP-RS fenced enumeration — complete, continuation, pinned window, -// short-cap, fence-lapse witnesses ────────────────────────────────────────── +// ── Test 2: NIP-RS fenced enumeration — EOSE, lapse, epoch-zero, retry ──────── +// All fetchAndMerge tests use subscribeFenced returning a proper FenceHandle so +// the loader only declares complete after an EOSE-established fence. // Helper to build a minimal valid-looking relay event for the pubkey. function makeFakeEvent(pubkey, createdAt) { @@ -750,19 +803,20 @@ function makeFakeEvent(pubkey, createdAt) { }; } +// ── 2a: empty relay + EOSE → complete ──────────────────────────────────────── test("fetchAndMerge_emptyRelay_setsLoadComplete", async () => { - // A relay with no events should produce an empty first band → complete. + // A relay with no events + EOSE-established fence → empty first band → complete. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "a".repeat(64); - let subscribeCallCount = 0; + let subscribeFencedCallCount = 0; const fakeRelay = { fetchEvents: async () => [], publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, - subscribeLive: async (_filter, _handler) => { - subscribeCallCount++; - return () => {}; + subscribeFenced: async (_filter, _onEvent) => { + subscribeFencedCallCount++; + return makeFenceHandle({ eose: true }); }, }; const mgr = new ReadStateManager(pubkey, fakeRelay); @@ -770,41 +824,110 @@ test("fetchAndMerge_emptyRelay_setsLoadComplete", async () => { assert.equal( mgr.isLoadComplete, true, - "empty relay must produce complete load", + "empty relay with EOSE-established fence must produce complete load", ); - // fence subscription must have been established (and then unsubscribed by fetchAndMerge). assert.equal( - subscribeCallCount, + subscribeFencedCallCount, 1, - "fence subscription must be set up exactly once", + "subscribeFenced must be called exactly once for the fence", ); mgr.destroy(); }); +// ── 2b: lapse before EOSE → incomplete (250 ms fallback does NOT count) ─────── +test("fetchAndMerge_lapseBeforeEose_setsLoadIncomplete", async () => { + // The fence lapses (lapsed=true) before EOSE resolves — this is the case + // the old subscribeLive 250 ms fallback would have falsely treated as complete. + // With a proper fence, lapse before EOSE must force complete:false. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "a1".repeat(32); + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_filter, _onEvent) => + makeFenceHandle({ eose: false, lapseBeforeEose: true }), + }; + const mgr = new ReadStateManager(pubkey, fakeRelay); + await mgr.fetchAndMerge(); + assert.equal( + mgr.isLoadComplete, + false, + "lapse before EOSE (e.g. 250 ms fallback path) must produce incomplete load", + ); + mgr.destroy(); +}); + +// ── 2c: terminal-CLOSED → lapse → incomplete ───────────────────────────────── +test("fetchAndMerge_terminalClosed_setsLoadIncomplete", async () => { + // Relay sends CLOSED before EOSE: fence.lapsed=true, established resolves. + // CLOSED does NOT count as EOSE — load must be incomplete. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "a2".repeat(32); + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_filter, _onEvent) => + makeFenceHandle({ eose: false, lapseBeforeEose: true }), + }; + const mgr = new ReadStateManager(pubkey, fakeRelay); + await mgr.fetchAndMerge(); + assert.equal( + mgr.isLoadComplete, + false, + "terminal CLOSED (fence lapse before EOSE) must produce incomplete load", + ); + mgr.destroy(); +}); + +// ── 2d: reconnect during post-empty barrier → lapse after tentative complete ── +test("fetchAndMerge_lapseAfterEmptyBand_forcesIncomplete", async () => { + // Empty first band → loader sets complete=true tentatively; fence lapses + // after EOSE (simulates mid-load reconnect). Final fence.lapsed check must + // override and force complete:false. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "a3".repeat(32); + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_filter, _onEvent) => + makeFenceHandle({ eose: true, lapsesAfterEose: true }), + }; + const mgr = new ReadStateManager(pubkey, fakeRelay); + await mgr.fetchAndMerge(); + assert.equal( + mgr.isLoadComplete, + false, + "lapse after tentative complete must override and force incomplete", + ); + mgr.destroy(); +}); + +// ── 2e: single event completes after pinned-window discharge ────────────────── test("fetchAndMerge_singleEvent_completesAfterPinnedWindowDischarge", async () => { - // Single event at T=1000: band delivers 1 event, C=1, L=2 → max(C,L)=2. - // Pinned window {since:1000, until:1000} returns 1 event → 1 < max(1,2)=2 → discharged. + // Single event at T=1000: band=1, C=1, L=2 → max(C,L)=2. + // Pinned window {since:1000, until:1000} returns 1 event → 1 < 2 → discharged. // Continuation {until:999} returns 0 → complete. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "b".repeat(64); const event = makeFakeEvent(pubkey, 1000); const fakeRelay = { fetchEvents: async (filter) => { - if (filter.since !== undefined && filter.until !== undefined) { - // Pinned window query — return the same event. - return [event]; - } - if (filter.until !== undefined && filter.until < 1000) { - // Continuation below T — empty. - return []; - } - // Initial band. - return [event]; + if (filter.since !== undefined && filter.until !== undefined) + return [event]; // pinned + if (filter.until !== undefined && filter.until < 1000) return []; // continuation + return [event]; // initial band }, publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, - subscribeLive: async (_filter, _handler) => () => {}, + subscribeFenced: async (_filter, _onEvent) => + makeFenceHandle({ eose: true }), }; const mgr = new ReadStateManager(pubkey, fakeRelay); await mgr.fetchAndMerge(); @@ -816,11 +939,10 @@ test("fetchAndMerge_singleEvent_completesAfterPinnedWindowDischarge", async () = mgr.destroy(); }); +// ── 2f: pinned-window-at-cap → incomplete ──────────────────────────────────── test("fetchAndMerge_pinnedWindowAtCap_setsLoadIncomplete", async () => { - // Pinned window returns max(C, L) events → potentially incomplete. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "c".repeat(64); - // Three events all at the same second T=2000. const events = [ makeFakeEvent(pubkey, 2000), makeFakeEvent(pubkey, 2000), @@ -828,17 +950,14 @@ test("fetchAndMerge_pinnedWindowAtCap_setsLoadIncomplete", async () => { ]; const fakeRelay = { fetchEvents: async (filter) => { - if (filter.since !== undefined && filter.until !== undefined) { - // Pinned window: return 3 events; C=3, max(C,L)=3 → incomplete. - return events; - } - // Initial band: 3 events, C=3. + if (filter.since !== undefined) return events; // pinned window returns cap-many return events; }, publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, - subscribeLive: async (_filter, _handler) => () => {}, + subscribeFenced: async (_filter, _handler) => + makeFenceHandle({ eose: true }), }; const mgr = new ReadStateManager(pubkey, fakeRelay); await mgr.fetchAndMerge(); @@ -850,8 +969,37 @@ test("fetchAndMerge_pinnedWindowAtCap_setsLoadIncomplete", async () => { mgr.destroy(); }); +// ── 2g: epoch-zero termination ─────────────────────────────────────────────── +test("fetchAndMerge_epochZero_completesAfterEmptyContinuation", async () => { + // T=0: an event at created_at=0 → T=0. The continuation query `until:0` + // with no `since` returns empty → history exhausted → complete. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "a4".repeat(32); + const event = makeFakeEvent(pubkey, 0); // created_at=0 + const fakeRelay = { + fetchEvents: async (filter) => { + if (filter.since !== undefined) return [event]; // pinned window: 1 < max(1,2)=2 → discharged + if (filter.until === 0 && filter.since === undefined) return []; // T=0 continuation → complete + return [event]; // initial band + }, + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_filter, _onEvent) => + makeFenceHandle({ eose: true }), + }; + const mgr = new ReadStateManager(pubkey, fakeRelay); + await mgr.fetchAndMerge(); + assert.equal( + mgr.isLoadComplete, + true, + "T=0 with empty continuation must produce complete load", + ); + mgr.destroy(); +}); + +// ── 2h: subscribeFenced throws → fence fails → incomplete ──────────────────── test("fetchAndMerge_fenceFails_setsLoadIncomplete", async () => { - // subscribeLive throws → fence cannot be established → load is incomplete. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "d".repeat(64); const fakeRelay = { @@ -859,7 +1007,7 @@ test("fetchAndMerge_fenceFails_setsLoadIncomplete", async () => { publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, - subscribeLive: async () => { + subscribeFenced: async () => { throw new Error("connection refused"); }, }; @@ -868,13 +1016,13 @@ test("fetchAndMerge_fenceFails_setsLoadIncomplete", async () => { assert.equal( mgr.isLoadComplete, false, - "fence failure must produce incomplete load", + "subscribeFenced failure must produce incomplete load", ); mgr.destroy(); }); +// ── 2i: incomplete load blocks gated operations ─────────────────────────────── test("fetchAndMerge_incompleteLoad_blocksGatedOperations", async () => { - // Pinned window returns ≥ max(C,L) → incomplete → four gated ops refuse. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "e".repeat(64); const events = [ @@ -893,6 +1041,7 @@ test("fetchAndMerge_incompleteLoad_blocksGatedOperations", async () => { }, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), subscribeLive: async (_f, _h) => () => {}, }; const mgr = new ReadStateManager(pubkey, fakeRelay); @@ -901,23 +1050,18 @@ test("fetchAndMerge_incompleteLoad_blocksGatedOperations", async () => { assert.equal(mgr.isLoadComplete, false, "precondition: load is incomplete"); - // 1. publish() must be blocked. - // Replace fetchOwnBlobBeforePublish to avoid second relay call. mgr.fetchOwnBlobBeforePublish = async () => true; await mgr.publish(); assert.equal(publishCalls, 0, "publish must be blocked when load incomplete"); - // 2. markChannelUnread must return load_incomplete. const ur = mgr.markChannelUnread("ch"); assert.equal(ur.success, false); assert.equal(ur.reason, "load_incomplete"); - // 3. markChannelRead must return load_incomplete. const rr = mgr.markChannelRead("ch"); assert.equal(rr.success, false); assert.equal(rr.reason, "load_incomplete"); - // 4. deleteExtraSlots must be blocked. mgr.extraSlotIds = ["fakeextraslot0000000000000000000"]; await mgr.deleteExtraSlots(); assert.equal( @@ -929,7 +1073,67 @@ test("fetchAndMerge_incompleteLoad_blocksGatedOperations", async () => { mgr.destroy(); }); -// ── Test 2b: retry path — a second fetchAndMerge can clear incomplete ───────── +// ── 2j: production retry path fires on reconnect ───────────────────────────── +test("retryLoad_firesOnReconnect_andClearsIncomplete", async () => { + // startLiveSubscription wires retryLoad() via subscribeToReconnects. + // (1) reconnect listener registered; (2) firing it re-runs fetchAndMerge. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "a5".repeat(32); + let reconnectCb = null; + let callRound = 0; + const fakeRelay = { + fetchEvents: async (filter) => { + callRound++; + if (callRound <= 3) { + if (filter.since !== undefined) + return [ + makeFakeEvent(pubkey, 1000), + makeFakeEvent(pubkey, 1000), + makeFakeEvent(pubkey, 1000), + ]; + return [ + makeFakeEvent(pubkey, 1000), + makeFakeEvent(pubkey, 1000), + makeFakeEvent(pubkey, 1000), + ]; + } + return []; + }, + publishEvent: async () => {}, + subscribeToReconnects: (cb) => { + reconnectCb = cb; + return () => { + reconnectCb = null; + }; + }, + getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + }; + const mgr = new ReadStateManager(pubkey, fakeRelay); + await mgr.initialize(); + assert.equal( + mgr.isLoadComplete, + false, + "precondition: first load is incomplete", + ); + assert.ok( + reconnectCb !== null, + "subscribeToReconnects listener must be registered", + ); + + callRound = 999; + reconnectCb(); + await new Promise((r) => setTimeout(r, 50)); + assert.equal( + mgr.isLoadComplete, + true, + "reconnect-triggered retry must clear incomplete when relay is now empty", + ); + mgr.destroy(); +}); + +// ── 2k: direct retry clears incomplete ─────────────────────────────────────── test("fetchAndMerge_retryClears_incomplete", async () => { globalThis.window.localStorage = makeLocalStorage(); const pubkey = "f".repeat(64); @@ -938,33 +1142,31 @@ test("fetchAndMerge_retryClears_incomplete", async () => { fetchEvents: async (filter) => { callRound++; if (callRound <= 3) { - // First attempt: pinned window fires at round 2, returns cap-many events. - if (filter.since !== undefined) { + if (filter.since !== undefined) return [ makeFakeEvent(pubkey, 1000), makeFakeEvent(pubkey, 1000), makeFakeEvent(pubkey, 1000), ]; - } return [ makeFakeEvent(pubkey, 1000), makeFakeEvent(pubkey, 1000), makeFakeEvent(pubkey, 1000), ]; } - // Retry: empty relay → complete. return []; }, publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), subscribeLive: async (_f, _h) => () => {}, }; const mgr = new ReadStateManager(pubkey, fakeRelay); await mgr.fetchAndMerge(); assert.equal(mgr.isLoadComplete, false, "first load must be incomplete"); - callRound = 999; // reset to "retry" leg + callRound = 999; await mgr.fetchAndMerge(); assert.equal( mgr.isLoadComplete, @@ -983,13 +1185,15 @@ test("handleIncomingEvent_liveOverride_updatesRegisterViaIngest", async () => { const pubkey = "aa".repeat(32); const rawCtx = `live-channel-${"x".repeat(51)}`; const channelFrontier = 100; - // Build a blob where the register (S=3, C=1, B=50) is ACTIVE: S > C, S > frontier. - // encode: ov_s:=3, ov_c:=1, ov_b:=50, = + // Build a blob where the register (S=3, C=1, B=50) is ACTIVE: + // isOverrideActive = S>0 && F<=B && S>C. B=50, frontier=100 → F>B → INACTIVE. + // Corrected: to make this active we need B >= F, e.g. B=100 with F=100 (boundary). + // Using B=100 so the register is genuinely active: 3>0, 100<=100, 3>1 → active. const blobContexts = { [rawCtx]: channelFrontier, [`ov_s:${rawCtx}`]: 3, [`ov_c:${rawCtx}`]: 1, - [`ov_b:${rawCtx}`]: 50, + [`ov_b:${rawCtx}`]: 100, // B=100 >= F=100 → active }; const plaintext = JSON.stringify({ v: 1, @@ -1014,6 +1218,8 @@ test("handleIncomingEvent_liveOverride_updatesRegisterViaIngest", async () => { publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, + subscribeFenced: async (_filter, _onEvent) => + makeFenceHandle({ eose: true }), subscribeLive: async (_filter, handler) => { liveHandler = handler; return () => {}; @@ -1038,16 +1244,40 @@ test("handleIncomingEvent_liveOverride_updatesRegisterViaIngest", async () => { }; // Deliver via the live subscription callback (same path as relay push). + // Track decrypt count: each live event must decrypt exactly once. + let decryptCount = 0; + const origInvoke = globalThis.window.__TAURI_INTERNALS__.invoke; + globalThis.window.__TAURI_INTERNALS__.invoke = async (command, args) => { + if (command === "nip44_decrypt_from_self") decryptCount++; + return origInvoke(command, args); + }; assert.ok(liveHandler !== null, "live subscription must be established"); liveHandler(fakeEvent); // void-wrapped; wait for async completion await new Promise((r) => setTimeout(r, 50)); + assert.equal( + decryptCount, + 1, + "live event must decrypt exactly once (no double-parse)", + ); // The override register must now reflect the ingested remote values. const reg = mgr.overrideRegisters.get(rawCtx); assert.ok(reg, "override register must exist after live delivery"); assert.equal(reg.s, 3, "S must be 3 from live event"); assert.equal(reg.c, 1, "C must be 1 from live event"); - assert.equal(reg.b, 50, "B must be 50 from live event"); + assert.equal(reg.b, 100, "B must be 100 from live event"); + + // Verify the register is genuinely active: isOverrideActive(S=3,C=1,B=100,F=100). + const livenessActive = mgr.getOverrideLiveness(rawCtx); + assert.ok( + livenessActive !== null, + "liveness must be available after live delivery", + ); + assert.equal( + livenessActive.active, + true, + "register must be active (B=100 >= F=100, S>C)", + ); // ── existing-key live clear (higher C defeating S) ─────────────────── // A follow-up event with S=3, C=4 (C > S → inactive/tombstone). @@ -1085,7 +1315,7 @@ test("handleIncomingEvent_liveOverride_updatesRegisterViaIngest", async () => { const regAfterClear = mgr.overrideRegisters.get(rawCtx); assert.ok(regAfterClear, "register must still exist after clear event"); // tombstone floor: only ov_c:ctx=4 is present → S=0, C=4, B=0 merged via componentwise max - // after merge with prior (S=3, C=1, B=50): S=3, C=4, B=50 — override_active = S <= C = inactive + // after merge with prior (S=3, C=1, B=100): S=3, C=4, B=100 — override_active = S <= C = inactive assert.equal( regAfterClear.c, 4, @@ -1121,6 +1351,7 @@ test("publish_fetchOwnBlobFails_doesNotPublish", async () => { }, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), subscribeLive: async (_f, _h) => () => {}, }; const mgr = new ReadStateManager(pubkey, fakeRelay); @@ -1147,6 +1378,7 @@ test("overrideRegister_survivesRestartBeforeDebounce", () => { publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), subscribeLive: async (_f, _h) => () => {}, }; const mgr1 = new ReadStateManager(pubkey, fakeRelay); @@ -1181,6 +1413,7 @@ test("overrideRegister_tombstoneFloorSurvivesRestartWithFetchFailure", async () publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), subscribeLive: async (_f, _h) => () => {}, }; // Establish a mark-read tombstone floor: S=1, C=max(S,C)+1=2 (clear-wins → inactive). @@ -1206,6 +1439,9 @@ test("overrideRegister_tombstoneFloorSurvivesRestartWithFetchFailure", async () publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, + subscribeFenced: async () => { + throw new Error("network unavailable"); + }, subscribeLive: async () => { throw new Error("network unavailable"); }, @@ -1308,7 +1544,7 @@ test("markChannelUnread_visibleRefusal_atBudgetExhaustionAndUint32Max", () => { mgr.destroy(); }); -// ── Test 8: persistence failure fails the mark ─────────────────────────────── +// ── Test 8: persistence failure — storage_failed + rollback + coherent restart ─ test("markChannelUnread_storageFailure_returnsStorageFailed", () => { // Use a throwing localStorage to simulate quota failure. const throwingLS = makeLocalStorage(); @@ -1324,12 +1560,19 @@ test("markChannelUnread_storageFailure_returnsStorageFailed", () => { const fakeRelay = { fetchEvents: async () => [], publishEvent: async () => {}, - subscribeLive: () => () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, }; const mgr = new ReadStateManager("st".repeat(32), fakeRelay); mgr.isLoadComplete = true; mgr.effectiveState.set("storage-test-ch", 1000); + // Snapshot pre-mutation state for rollback verification. + const regBefore = mgr.overrideRegisters.get("storage-test-ch"); + const wasPublishableBefore = mgr.publishableContextIds.has("storage-test-ch"); + const result = mgr.markChannelUnread("storage-test-ch"); assert.equal( result.success, @@ -1337,7 +1580,32 @@ test("markChannelUnread_storageFailure_returnsStorageFailed", () => { "markChannelUnread must fail when localStorage throws", ); assert.equal(result.reason, "storage_failed"); + + // Contract 3 rollback: manager state must be unchanged after storage_failed. + const regAfter = mgr.overrideRegisters.get("storage-test-ch"); + assert.deepEqual( + regAfter, + regBefore, + "overrideRegisters must be rolled back after storage failure", + ); + assert.equal( + mgr.publishableContextIds.has("storage-test-ch"), + wasPublishableBefore, + "publishableContextIds must be rolled back after storage failure", + ); + + // Coherent restart: a new manager on the same (throwing) storage must see no + // orphaned register — the failed write must not have partially persisted. + const mgr2 = new ReadStateManager("st".repeat(32), fakeRelay); + mgr2.hydrateFromLocalStorage(); + const regOnRestart = mgr2.overrideRegisters.get("storage-test-ch"); + assert.equal( + regOnRestart, + undefined, + "a failed mark must not persist any register (coherent restart)", + ); mgr.destroy(); + mgr2.destroy(); }); // ── Test 9: inactive existing register still gets C-bump on markChannelRead ─── @@ -1414,30 +1682,27 @@ test("deduplicateByCoordinate_newerVersionWins_olderDropped", async () => { // ── Test 11: lapse mid-enumeration → incomplete ─────────────────────────────── test("fetchAndMerge_lapseMidEnumeration_setsLoadIncomplete", async () => { - // Simulate a reconnect (generation change) after the first band is fetched. + // Simulate a connection lapse after the first band is fetched but before + // the pinned window returns. The fence.lapsed flag is set mid-enumeration. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "ef".repeat(32); const event = makeFakeEvent(pubkey, 1000); - let generation = 0; - let reconnectCb = null; + + // Build a fence that starts unlapsed but lapses when the pinned window fires. + const fence = makeFenceHandle({ eose: true }); const fakeRelay = { fetchEvents: async (filter) => { if (filter.since !== undefined) { - // Pinned window: simulate a reconnect BEFORE returning. - generation++; - reconnectCb?.(); // fire reconnect listener + // Pinned window: trigger lapse mid-enumeration. + fence._lapse(); return [event]; } return [event]; // initial band }, publishEvent: async () => {}, - subscribeToReconnects: (cb) => { - reconnectCb = cb; - return () => { - reconnectCb = null; - }; - }, - getConnectionGeneration: () => generation, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_filter, _onEvent) => fence, subscribeLive: async (_f, _h) => () => {}, }; const mgr = new ReadStateManager(pubkey, fakeRelay); @@ -1449,3 +1714,169 @@ test("fetchAndMerge_lapseMidEnumeration_setsLoadIncomplete", async () => { ); mgr.destroy(); }); + +// ── Test 12: foreign client_id at initial load triggers slot rotation ───────── +test("fetchAndMerge_foreignClientId_rotatesSlotAndUpdatesMetadata", async () => { + // If a fetched event carries our slot coordinate but a different client_id, + // the manager must rotate slotId and record maxFetchedCreatedAt. + // Also validates read-before-write path: fetchOwnBlobBeforePublish runs the + // same parsed-record metadata path (Contract 2). + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "a6".repeat(32); + + // Install Tauri IPC mock. + const slotId = "deadbeefdeadbeef0123456789abcdef"; // will be the initial slotId + const foreignClientId = "other-client-uuid-9999"; + const blob = JSON.stringify({ + v: 1, + client_id: foreignClientId, + contexts: { "ch-conflict": 500 }, + }); + globalThis.window.__TAURI_INTERNALS__ = { + invoke: async (command, args) => { + if (command === "nip44_decrypt_from_self") { + if (args.ciphertext === "CONFLICT_CIPHER") return blob; + return JSON.stringify({ + v: 1, + client_id: foreignClientId, + contexts: {}, + }); + } + throw new Error(`Unexpected: ${command}`); + }, + }; + + // Build a fetched event at our slot coordinate but with foreign client_id. + const conflictEvent = { + id: "f0".repeat(32), + pubkey, + created_at: 12345, + kind: 30078, + tags: [ + ["d", `read-state:${slotId}`], + ["t", "read-state"], + ], + content: "CONFLICT_CIPHER", + sig: "s".repeat(128), + }; + + const fakeRelay = { + // Respect `until` so the loader terminates: once `until` drops below the + // event's created_at the relay returns empty, ending the enumeration. + fetchEvents: async (filter) => { + if (filter.until !== undefined && conflictEvent.created_at > filter.until) + return []; + return [conflictEvent]; + }, + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + }; + + try { + const mgr = new ReadStateManager(pubkey, fakeRelay); + // Pre-seed the manager's slotId to match the conflict event's d-tag. + mgr.slotId = slotId; + await mgr.fetchAndMerge(); + + // Slot rotation: slotId must differ from the conflicting coordinate. + assert.notEqual( + mgr.slotId, + slotId, + "slotId must rotate when fetched event carries foreign client_id at our coordinate", + ); + + // maxFetchedCreatedAt must reflect the fetched event's created_at. + assert.equal( + mgr.maxFetchedCreatedAt, + 12345, + "maxFetchedCreatedAt must be updated from fetched event created_at", + ); + } finally { + delete globalThis.window.__TAURI_INTERNALS__; + } +}); + +// ── Test 13: frontier-only advance schedules canonical convergence ───────────── +test("ingest_frontierAdvanceFlipsRegister_schedulesCanonicalConvergence", async () => { + // A frontier advance that flips an override register from live→tombstone must + // set canonicalChanged=true and trigger schedulePublish (debounce timer set). + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "a7".repeat(32); + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + }; + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.isLoadComplete = true; + + // Seed an active register: S=5, C=0, B=10, F=5. + // isOverrideActive(S=5,C=0,B=10,F=5) = 5>0 && 5<=10 && 5>0 → active. + const ctx = "convergence-ch"; + mgr.overrideRegisters.set(ctx, { s: 5, c: 0, b: 10 }); + mgr.effectiveState.set(ctx, 5); + mgr.publishableContextIds.add(ctx); + + // Verify it's active before the frontier advance. + const before = mgr.getOverrideLiveness(ctx); + assert.ok(before?.active, "register must be active before frontier advance"); + + // Deliver a frontier advance F=11 (> B=10) → register becomes dead. + // Simulate via a live event that carries only the frontier for this ctx. + // Build a minimal fake event that encodes just the frontier key. + const frontierBlob = JSON.stringify({ + v: 1, + client_id: "peer-device", + contexts: { [ctx]: 11 }, // frontier advance → F=11 > B=10 → inactive + }); + globalThis.window.__TAURI_INTERNALS__ = { + invoke: async (command, _args) => { + if (command === "nip44_decrypt_from_self") return frontierBlob; + throw new Error(`Unexpected: ${command}`); + }, + }; + + try { + // Fire the live subscription callback directly. + const fakeEvent = { + id: "cc".repeat(32), + pubkey, + created_at: 9999, + kind: 30078, + tags: [ + ["d", "read-state:cc99aa00112233445566778899aabbcc"], + ["t", "read-state"], + ], + content: "FRONTIER_CIPHER", + sig: "s".repeat(128), + }; + + // Access the private handleIncomingEvent via the test helper path. + await mgr.handleIncomingEvent(fakeEvent); + + // Register must now be inactive. + const after = mgr.getOverrideLiveness(ctx); + assert.ok(after !== null, "liveness must still be available"); + assert.equal( + after.active, + false, + "frontier advance past B must flip register to inactive", + ); + + // debounceTimer must be set (schedulePublish was called for convergence). + assert.ok( + mgr.debounceTimer !== null, + "frontier-only canonical deactivation must schedule a convergence publish", + ); + } finally { + delete globalThis.window.__TAURI_INTERNALS__; + mgr.destroy(); + } +}); diff --git a/desktop/src/features/channels/readState/readStateManager.ts b/desktop/src/features/channels/readState/readStateManager.ts index c807795851..fc8f5d7cb7 100644 --- a/desktop/src/features/channels/readState/readStateManager.ts +++ b/desktop/src/features/channels/readState/readStateManager.ts @@ -21,7 +21,7 @@ import { } from "@/features/channels/readState/readStateFormat"; import { parseReadStateEvent, - mergeReadStateEventsStructured, + mergeOverrideRegisterMaps, type MergedReadState, type ParsedReadStateEvent, } from "@/features/channels/readState/readStateSnapshot"; @@ -99,6 +99,13 @@ export type ApplyRemoteContextResult = "unchanged" | "advanced"; export type ContextParentResolver = (contextId: string) => string | null; +/** Coherent projection (completeness + frontiers + overrides) for UI consumption via `getProjection()`. */ +export type ReadStateProjection = { + loadComplete: boolean; + frontiers: ReadonlyMap; + overrides: ReadonlyMap; +}; + /** NIP-RS Hierarchical Frontier: `effective(ctx) = max(merged[ctx], effective(parent(ctx)))`. */ export function resolveEffectiveTimestamp(args: { effectiveState: Map; @@ -381,12 +388,13 @@ export class ReadStateManager { const result = await fencedEnumerationLoad(this.relayClient, this.pubkey); if (this.destroyed) return; this.isLoadComplete = result.complete; - if (!result.complete) { + if (!result.complete) console.warn( "[ReadStateManager] fetchAndMerge: load incomplete — gated ops blocked", ); - } - await this.ingest(result.merged); + this.processLoadedParsedEvents(result.events); + const merged = mergeParsedEvents(result.events); + await this.ingest(merged); this.persistLocalState(); this.notifyListeners(); } @@ -398,12 +406,7 @@ export class ReadStateManager { await this.fetchAndMerge(); } - /** - * Shared ingest: accept a pre-merged MergedReadState (from fencedEnumerationLoad, - * live delivery, or read-before-write). Returns an explicit semantic delta. - * Persist/notify on every register change; schedule convergence only on - * canonical-to-canonical difference. - */ + /** Shared ingest: accept pre-merged state and return semantic delta (changed contexts, canonical flip). */ private async ingest(merged: MergedReadState): Promise { const delta: IngestDelta = { changedContexts: new Set(), @@ -411,6 +414,18 @@ export class ReadStateManager { }; // ── Frontier merge ──────────────────────────────────────────────────── + // Snapshot canonical forms for any register whose frontier might change: + // a frontier advance alone can flip live→tombstone without touching (S,C,B). + const prevCanonicalByCtx = new Map(); + for (const rawCtx of merged.frontiers.keys()) { + const reg = this.overrideRegisters.get(rawCtx); + if (reg !== undefined) + prevCanonicalByCtx.set( + rawCtx, + canonicalKey(reg, this.channelFrontier(rawCtx)), + ); + } + for (const [rawCtx, ts] of merged.frontiers) { const result = applyRemoteContextTimestamp({ effectiveState: this.effectiveState, @@ -426,48 +441,44 @@ export class ReadStateManager { } } + // Check whether any frontier advance changed an affected register's canonical form. + for (const [rawCtx, prevCanon] of prevCanonicalByCtx) { + const reg = this.overrideRegisters.get(rawCtx); + if (reg !== undefined) { + const newCanon = canonicalKey(reg, this.channelFrontier(rawCtx)); + if (newCanon !== prevCanon) delta.canonicalChanged = true; + } + } + // ── Override register merge — compare component-by-component ───────── for (const [rawCtx, reg] of merged.overrides) { const ex = this.overrideRegisters.get(rawCtx); - const merged2 = ex + const m: OverrideRegister = ex ? { s: Math.max(ex.s, reg.s), c: Math.max(ex.c, reg.c), b: Math.max(ex.b, reg.b), } : reg; - const changed = - !ex || merged2.s !== ex.s || merged2.c !== ex.c || merged2.b !== ex.b; + const changed = !ex || m.s !== ex.s || m.c !== ex.c || m.b !== ex.b; if (changed) { - this.overrideRegisters.set(rawCtx, merged2); + this.overrideRegisters.set(rawCtx, m); delta.changedContexts.add(rawCtx); this.publishableContextIds.add(rawCtx); - // Canonical change: existing register had a different canonical form. const prevCanon = ex ? canonicalKey(ex, this.channelFrontier(rawCtx)) : null; - const newCanon = canonicalKey(merged2, this.channelFrontier(rawCtx)); - if (prevCanon !== newCanon) delta.canonicalChanged = true; + if (prevCanon !== canonicalKey(m, this.channelFrontier(rawCtx))) + delta.canonicalChanged = true; } } return delta; } - /** - * Ingest from raw relay events (live delivery path). Deduplicates by - * coordinate, then delegates to the MergedReadState ingest path. - * Returns an explicit delta for change detection. - */ - private async ingestEvents(events: RelayEvent[]): Promise { - const deduped = deduplicateByCoordinate(events); - // Track maxFetchedCreatedAt and own-blob metadata from the raw events. - for (const event of events) { - const parsed: ParsedReadStateEvent | null = await parseReadStateEvent( - event, - this.pubkey, - ); - if (!parsed) continue; + /** Process already-parsed events for metadata, conflict rotation, and maxFetchedCreatedAt. */ + private processLoadedParsedEvents(events: ParsedReadStateEvent[]): void { + for (const parsed of events) { this.maxFetchedCreatedAt = Math.max( this.maxFetchedCreatedAt, parsed.createdAt, @@ -477,6 +488,8 @@ export class ReadStateManager { if (parsed.createdAt > src) this.contextSourceCreatedAt.set(rawCtx, parsed.createdAt); } + // Slot-conflict rotation: if our slot coordinate carries another client's + // client_id, rotate to a new slot ID (spec MUST NOT, NIP-RS.md:55-64,402-406). if ( parsed.dTag === `read-state:${this.slotId}` && parsed.blob.client_id !== this.clientId @@ -487,6 +500,7 @@ export class ReadStateManager { this.slotId, ); } + // Own-blob metadata: union lastPublishedContexts and mark publishable. if (parsed.blob.client_id === this.clientId) { const unionContexts: Record = { ...this.lastPublishedContexts, @@ -500,11 +514,23 @@ export class ReadStateManager { this.lastPublishedContexts = unionContexts; } } - const merged = await mergeReadStateEventsStructured(deduped, this.pubkey); + } + + /** Ingest parsed events: process metadata then merge into frontier/override maps. */ + private async ingestParsedEvents( + events: ParsedReadStateEvent[], + ): Promise { + this.processLoadedParsedEvents(events); + const merged = mergeParsedEvents(events); return this.ingest(merged); } private async startLiveSubscription(): Promise { + // Wire retryLoad on reconnects so incomplete init doesn't brick the manager. + const unsubReconnect = this.relayClient.subscribeToReconnects(() => { + if (this.destroyed) return; + void this.retryLoad(); + }); try { const unsub = await this.relayClient.subscribeLive( { @@ -518,22 +544,29 @@ export class ReadStateManager { ); if (this.destroyed) { unsub(); + unsubReconnect(); return; } - this.unsubscribeLive = unsub; + this.unsubscribeLive = () => { + unsub(); + unsubReconnect(); + }; } catch { + unsubReconnect(); // Live subscription is best-effort; missed events will be caught on reconnect. } } private async handleIncomingEvent(event: RelayEvent): Promise { if (event.pubkey !== this.pubkey || this.destroyed) return; - const delta = await this.ingestEvents([event]); + // Parse once: use the result for both ingest and the convergence client_id check. + const parsed = await parseReadStateEvent(event, this.pubkey); + if (!parsed) return; + const delta = await this.ingestParsedEvents([parsed]); if (delta.changedContexts.size > 0 || this.pendingSyncedAdvances.size > 0) { this.persistLocalState(); this.notifyListeners(); - const parsed = await parseReadStateEvent(event, this.pubkey); - if (parsed?.blob.client_id !== this.clientId && delta.canonicalChanged) { + if (delta.canonicalChanged && parsed.blob.client_id !== this.clientId) { this.schedulePublish(); } } @@ -659,7 +692,14 @@ export class ReadStateManager { limit: READ_STATE_FETCH_LIMIT, }); const deduped = deduplicateByCoordinate(events); - const merged = await mergeReadStateEventsStructured(deduped, this.pubkey); + // Parse once: use same path as full-state load (Contract 2 — metadata/conflict). + const parsed: ParsedReadStateEvent[] = []; + for (const ev of deduped) { + const p = await parseReadStateEvent(ev, this.pubkey); + if (p) parsed.push(p); + } + this.processLoadedParsedEvents(parsed); + const merged = mergeParsedEvents(parsed); await this.ingest(merged); this.persistLocalState(); return true; @@ -779,12 +819,16 @@ export class ReadStateManager { }; } - /** - * Pure candidate planner: trials the candidate register against BOTH - * single-slot and multi-slot paths. Refuses only when the override-bearing - * primary cannot fit even after splitting to maxSlots. Side-effect-free - * (restores state on return). Used by both mark trials and actual publish. - */ + /** Return a coherent snapshot (completeness + frontiers + overrides). UI consumes via getProjection(). */ + getProjection(): ReadStateProjection { + return { + loadComplete: this.isLoadComplete, + frontiers: this.effectiveState, + overrides: this.overrideRegisters, + }; + } + + /** Pure candidate planner: trials candidate against single- and multi-slot paths. Side-effect-free. */ private tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean { const prev = this.overrideRegisters.get(rawCtxId); const wasPublishable = this.publishableContextIds.has(rawCtxId); @@ -822,10 +866,18 @@ export class ReadStateManager { if (!this.tryCandidatePlan(channelId, newReg)) { return { success: false, reason: "budget_exhausted" }; } + // Snapshot before mutation for Contract 3 rollback. + const prevReg = this.overrideRegisters.get(channelId); + const wasPublishable = this.publishableContextIds.has(channelId); this.overrideRegisters.set(channelId, newReg); this.publishableContextIds.add(channelId); - if (!this.persistLocalState()) + if (!this.persistLocalState()) { + // Rollback: restore pre-mutation state. + if (prevReg === undefined) this.overrideRegisters.delete(channelId); + else this.overrideRegisters.set(channelId, prevReg); + if (!wasPublishable) this.publishableContextIds.delete(channelId); return { success: false, reason: "storage_failed" }; + } this.notifyListeners(); this.schedulePublish(); return { success: true }; @@ -850,10 +902,16 @@ export class ReadStateManager { ); return { success: false, reason: "already_inactive" }; } + // Snapshot before mutation for Contract 3 rollback. + const wasPublishable = this.publishableContextIds.has(channelId); this.overrideRegisters.set(channelId, newReg); this.publishableContextIds.add(channelId); - if (!this.persistLocalState()) + if (!this.persistLocalState()) { + // Rollback: restore pre-mutation state. + this.overrideRegisters.set(channelId, reg); + if (!wasPublishable) this.publishableContextIds.delete(channelId); return { success: false, reason: "storage_failed" }; + } this.notifyListeners(); this.schedulePublish(); return { success: true }; @@ -867,30 +925,36 @@ export class ReadStateManager { this.publishableContextIds.add(contextId); for (const [contextId, createdAt] of stored.contextSourceCreatedAt) this.contextSourceCreatedAt.set(contextId, createdAt); - for (const [rawCtx, reg] of stored.overrideRegisters) { - const ex = this.overrideRegisters.get(rawCtx); - this.overrideRegisters.set( - rawCtx, - ex - ? { - s: Math.max(ex.s, reg.s), - c: Math.max(ex.c, reg.c), - b: Math.max(ex.b, reg.b), - } - : reg, - ); + for (const [ctx, e] of stored.overrideRegisters) { + const ex = this.overrideRegisters.get(ctx); + const merged = ex + ? { + s: Math.max(ex.s, e.s), + c: Math.max(ex.c, e.c), + b: Math.max(ex.b, e.b), + } + : { s: e.s, c: e.c, b: e.b }; + this.overrideRegisters.set(ctx, merged); + if (e.f > 0 && !this.effectiveState.has(ctx)) + this.effectiveState.set(ctx, e.f); } this.persistLocalState(); } /** Persist local state. Returns false if any write failed (mark-action must fail). */ private persistLocalState(): boolean { + const entries = new Map( + [...this.overrideRegisters].map(([ctx, r]) => [ + ctx, + { s: r.s, c: r.c, b: r.b, f: this.channelFrontier(ctx) }, + ]), + ); return writeStoredReadState( this.pubkey, this.effectiveState, this.publishableContextIds, this.contextSourceCreatedAt, - this.overrideRegisters, + entries, ); } @@ -918,3 +982,17 @@ function canonicalKey(reg: OverrideRegister, frontier: number): string { const floor = Math.max(reg.s, reg.c); return floor > 0 ? `dead:${floor}` : "virgin"; } + +/** Build MergedReadState from pre-parsed events — no second decrypt. */ +function mergeParsedEvents(events: ParsedReadStateEvent[]): MergedReadState { + const frontiers = new Map(); + const overrideMaps: Array> = []; + for (const p of events) { + for (const [rawCtx, ts] of p.contexts.frontiers) { + const current = frontiers.get(rawCtx) ?? 0; + if (ts > current) frontiers.set(rawCtx, ts); + } + overrideMaps.push(p.contexts.overrides); + } + return { frontiers, overrides: mergeOverrideRegisterMaps(...overrideMaps) }; +} diff --git a/desktop/src/features/channels/readState/readStateStorage.test.mjs b/desktop/src/features/channels/readState/readStateStorage.test.mjs index 51b1752222..49d6107ee6 100644 --- a/desktop/src/features/channels/readState/readStateStorage.test.mjs +++ b/desktop/src/features/channels/readState/readStateStorage.test.mjs @@ -129,7 +129,7 @@ test("writeStoredReadState round-trips through readStoredReadState", () => { new Map([["channel-9", nowSeconds]]), new Set(["channel-9"]), new Map([["channel-9", nowSeconds]]), - new Map([["channel-9", { s: 3, c: 1, b: nowSeconds }]]), + new Map([["channel-9", { s: 3, c: 1, b: nowSeconds, f: nowSeconds }]]), ); const stored = readStoredReadState(pubkey);