jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-02 07:22:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge duncan/nip-rs-unread-manager (e5bdaa980) into hayt/nip-rs-unread-ui

Browse Source 
* origin/duncan/nip-rs-unread-manager:
  fix(nip-rs): add fencedLoader integration and manager/storage fixes
  fix(nip-rs): correct fencedLoader lapse detection and manager semantics
  fix(read-state): close Thufir pass-1 findings in readStateManager/Storage

Signed-off-by: npub1g8493u0xfsjrvflg4n08ezd7vec99mnwzlv0qgwpr9d7gvjwhuzqx59rhw <41ea58f1e64c243627e8acde7c89be667052ee6e17d8f021c1195be4324ebf04@buzz.block.builderlab.xyz>
This commit is contained in:
npub1g8493u0xfsjrvflg4n08ezd7vec99mnwzlv0qgwpr9d7gvjwhuzqx59rhw
2026-07-31 18:30:31 -04:00
parent ad0705e711 e5bdaa9804
commit d473e742d4
6 changed files with 1287 additions and 443 deletions
Download Patch File Download Diff File Expand all files Collapse all files

216
desktop/src/features/channels/readState/readStateFencedLoader.ts Normal file
View File

@@ -0,0 +1,216 @@
/**
* NIP-RS full-state fenced loader.
*
* Implements the NIP-RS §Full-State Load procedure (NIP-RS.md:321-377):
* - Tag-free filter established on the fence subscription BEFORE the first
* enumeration query (NIP-RS.md:341-345).
* - Lapse detection via BOTH a reconnect listener AND connection-generation
* checks before and after every band query. If the generation changes at
* any point, the fence lapsed and the load is potentially incomplete.
* - Descending `until` cursor with pinned-window check (NIP-RS.md:350-352).
* - Delivery barrier: after the terminal empty continuation, we wait one
* EVENT_BATCH_MS tick so the relay client's 16 ms event-batch timer can
* flush any in-flight fence events, then drain `fenceEvents` before verdict
* (NIP-RS.md:343,353).
* - `T === 0` completes only after an explicitly empty continuation
* (NIP-RS.md:352).
* - Coordinate deduplicated by `d` tag (greatest `created_at`, lowest id)
* fed into a single structured merge (NIP-RS.md:348).
*/
import type { RelayClient } from "@/shared/api/relayClientSession";
import type { RelayEvent } from "@/shared/api/types";
import { KIND_READ_STATE } from "@/shared/constants/kinds";
import { mergeReadStateEventsStructured } from "@/features/channels/readState/readStateSnapshot";
import type { MergedReadState } from "@/features/channels/readState/readStateSnapshot";
import { isValidReadStateDTag } from "@/features/channels/readState/readStateFormat";
import { EVENT_BATCH_MS } from "@/shared/api/relayClientTimings";
/** Number of events per enumeration band. MUST be ≥ L=2 (NIP-RS.md:347). */
const BAND_LIMIT = 500;
/** NIP-RS §Floor: relay MUST deliver ≥ L events when ≥ L exist (NIP-RS.md:360). */
const L = 2;
export type FencedLoadResult =
| { complete: true; merged: MergedReadState }
| { complete: false; merged: MergedReadState };
/**
* Perform one full-state fenced enumeration for `pubkey`.
*
* Returns the merged read state from all collected events and whether the load
* was proven complete per the NIP-RS §Full-State Load procedure. A
* `complete: false` result carries a best-effort baseline for local display.
*/
export async function fencedEnumerationLoad(
relay: RelayClient,
pubkey: string,
): Promise<FencedLoadResult> {
const baseFilter = {
kinds: [KIND_READ_STATE],
authors: [pubkey],
limit: BAND_LIMIT,
};
// ── Step 1: Establish the fence BEFORE the first query ──────────────────
// The fence must be established on the SAME connection as every subsequent
// band query (NIP-RS.md:341-345). We track lapse via:
// (a) reconnect listener — fires on any connection reset
// (b) generation checks — before/after each fetchEvents call
// If either signals a lapse, the load is potentially incomplete.
const fenceEvents: RelayEvent[] = [];
let lapsed = false;
const startGeneration = relay.getConnectionGeneration();
let unsubscribeFence: (() => Promise<void>) | null = null;
const unsubscribeReconnect = relay.subscribeToReconnects(() => {
lapsed = true;
});
try {
unsubscribeFence = await relay.subscribeLive(baseFilter, (ev) => {
fenceEvents.push(ev);
});
// Verify the subscribe itself didn't trigger a reconnect.
if (relay.getConnectionGeneration() !== startGeneration) lapsed = true;
} catch {
unsubscribeReconnect();
return emptyIncomplete();
}
// ── Step 2: Descending enumeration ──────────────────────────────────────
let C = 0;
let until: number | undefined;
const bandEvents: RelayEvent[] = [];
let complete = false;
while (true) {
if (lapsed) break;
const filter = { ...baseFilter, ...(until !== undefined ? { until } : {}) };
const genBefore = relay.getConnectionGeneration();
let band: RelayEvent[];
try {
band = await relay.fetchEvents(filter);
} catch {
break;
}
if (relay.getConnectionGeneration() !== genBefore || lapsed) {
break;
}
if (band.length === 0) {
// Empty continuation — load proven complete.
complete = true;
break;
}
if (band.length > C) C = band.length;
for (const ev of band) bandEvents.push(ev);
// T = lowest created_at in this band (NIP-RS.md:350).
let T = band[0].created_at;
for (const ev of band) if (ev.created_at < T) T = ev.created_at;
// ── Pinned window (NIP-RS.md:350-351) ───────────────────────────────
if (lapsed) break;
const genPinned = relay.getConnectionGeneration();
let pinned: RelayEvent[];
try {
pinned = await relay.fetchEvents({ ...baseFilter, since: T, until: T });
} catch {
break;
}
if (relay.getConnectionGeneration() !== genPinned || lapsed) {
break;
}
for (const ev of pinned) bandEvents.push(ev);
if (pinned.length > C) C = pinned.length;
if (pinned.length >= Math.max(C, L)) {
// Pinned window not discharged — potentially incomplete (spec :351).
break;
}
if (T === 0) {
// T=0 exhausted: require a strictly-older empty continuation (spec :352).
const genCont = relay.getConnectionGeneration();
let cont: RelayEvent[];
try {
cont = await relay.fetchEvents({ ...baseFilter, until: 0 });
} catch {
break;
}
if (relay.getConnectionGeneration() !== genCont || lapsed) {
break;
}
if (cont.length === 0) {
complete = true;
} else {
for (const ev of cont) bandEvents.push(ev);
}
break;
}
until = T - 1;
}
// ── Delivery barrier (NIP-RS.md:343,353) ────────────────────────────────
// Wait one EVENT_BATCH_MS tick: any fence events buffered in the relay
// client's 16 ms dispatch batch will flush and invoke our callback before
// we call unsubscribe and stop collecting.
await new Promise<void>((r) => window.setTimeout(r, EVENT_BATCH_MS + 1));
await unsubscribeFence?.();
unsubscribeReconnect();
const allEvents = [...bandEvents, ...fenceEvents];
if (lapsed && !complete) {
return buildResult(false, allEvents, pubkey);
}
// ── Coordinate deduplicate before merge (NIP-RS.md:348) ─────────────────
const deduped = deduplicateByCoordinate(allEvents);
const merged = await mergeReadStateEventsStructured(deduped, pubkey);
return { complete, merged };
}
/**
* Deduplicate events by `d` tag: retain the event with the greatest
* `created_at`; break ties by the lexicographically lowest event id.
* Prevents superseded coordinate versions from being merged as concurrent
* replicas (NIP-RS.md:348).
*/
export function deduplicateByCoordinate(events: RelayEvent[]): RelayEvent[] {
const best = new Map<string, RelayEvent>();
for (const ev of events) {
const dTag = ev.tags.find((t) => t[0] === "d")?.[1];
if (!dTag || !isValidReadStateDTag(dTag)) continue;
const existing = best.get(dTag);
if (
!existing ||
ev.created_at > existing.created_at ||
(ev.created_at === existing.created_at && ev.id < existing.id)
) {
best.set(dTag, ev);
}
}
return [...best.values()];
}
function emptyIncomplete(): FencedLoadResult {
return {
complete: false,
merged: { frontiers: new Map(), overrides: new Map() },
};
}
async function buildResult(
complete: boolean,
events: RelayEvent[],
pubkey: string,
): Promise<FencedLoadResult> {
const deduped = deduplicateByCoordinate(events);
const merged = await mergeReadStateEventsStructured(deduped, pubkey);
return { complete, merged };
}

791
desktop/src/features/channels/readState/readStateManager.test.mjs
View File

@@ -606,7 +606,7 @@ test("splitContextsIntoBudgetedSlots_noOverrideKeyInNonPrimarySlot", () => {
[`ov_s:${ctx1}`, 1],
[`ov_c:${ctx1}`, 0],
[`ov_b:${ctx1}`, 100],
[ctx1, 100], // frontier for ctx1
[ctx1, 100], // frontier for ctx1 (normal, no escape needed)
[`ov_s:${ctx2}`, 2],
[`ov_c:${ctx2}`, 1],
[`ov_b:${ctx2}`, 200],
@@ -670,92 +670,562 @@ test("splitContextsIntoBudgetedSlots_noOverrideKeyInNonPrimarySlot", () => {
assert.ok(ctx2 in slot0, "frontier for ctx2 must be in slot 0");
});
// ── Test 2: full-page fetch blocks four gated operations ─────────────────────
test("fetchAndMerge_fullPage_blocksGatedOperations", async () => {
globalThis.window.localStorage = makeLocalStorage();
// ── Test 1b: reserved esc: raw ID — unescape-before-group rule ───────────────
test("splitContextsIntoBudgetedSlots_escapedFrontierKeyStaysWithItsOverrideGroup", () => {
// A context whose raw ID starts with "ov_" must be escaped to "esc:ov_s:evil"
// as a frontier wire key. The ov_* siblings are keyed by the RAW suffix
// "ov_s:evil". The splitter must unescape "esc:ov_s:evil" → "ov_s:evil"
// and recognise it as belonging to the same group as ov_s:ov_s:evil etc.
const rawCtx = "ov_s:evil";
const wireKey = `esc:${rawCtx}`; // what currentContexts() emits
// Relay returns exactly READ_STATE_FULL_FETCH_LIMIT events → truncation guard fires.
// We simulate this by making fetchEvents return an array of `limit` length.
// The actual events don't need to be valid — mergeEvents handles parse failures.
const FULL_FETCH_LIMIT = 5_000;
const fakeEvents = new Array(FULL_FETCH_LIMIT).fill({
id: "x".repeat(64),
pubkey: "a".repeat(64),
const channelEntries = [
[`ov_s:${rawCtx}`, 1], // ov_s:ov_s:evil
[`ov_c:${rawCtx}`, 0], // ov_c:ov_s:evil
[`ov_b:${rawCtx}`, 50], // ov_b:ov_s:evil
[wireKey, 50], // esc:ov_s:evil (escaped frontier)
];
// Add plain entries to force multi-slot so the splitter actually partitions.
const plain = [];
for (let i = 0; i < 20; i++) plain.push([makeChannelKey(i), i + 1]);
const allEntries = [...channelEntries, ...plain];
const encoder = new TextEncoder();
// Budget: fits the override group + 5 plain entries, not all 20.
const groupOnly = Object.fromEntries(channelEntries);
const fivePlain = Object.fromEntries(plain.slice(0, 5));
const budget =
encoder.encode(
JSON.stringify({ v: 1, client_id: CLIENT_ID, contexts: groupOnly }),
).length +
encoder.encode(
JSON.stringify({ v: 1, client_id: CLIENT_ID, contexts: fivePlain }),
).length;
const result = splitContextsIntoBudgetedSlots({
channelEntries: allEntries,
threadMsgEntries: [],
clientId: CLIENT_ID,
initialSlotCount: 1,
maxSlots: 8,
maxBytes: budget,
slotIdGenerator: deterministicSlotId,
});
assert.ok(result !== null, "should succeed");
assert.ok(result.slots.length >= 2, "should split");
const slot0 = result.slots[0];
// The escaped frontier key and all three ov_* siblings must be in slot 0.
assert.ok(
wireKey in slot0,
`${wireKey} (escaped frontier) must be in slot 0`,
);
assert.ok(`ov_s:${rawCtx}` in slot0, "ov_s: sibling must be in slot 0");
assert.ok(`ov_c:${rawCtx}` in slot0, "ov_c: sibling must be in slot 0");
assert.ok(`ov_b:${rawCtx}` in slot0, "ov_b: sibling must be in slot 0");
// No ov_* or esc: entry in non-primary slots.
for (let i = 1; i < result.slots.length; i++) {
for (const key of Object.keys(result.slots[i])) {
assert.ok(
!key.startsWith("ov_") && !key.startsWith("esc:"),
`reserved key "${key}" must not appear in slot ${i}`,
);
}
}
});
// ── Test 2: NIP-RS fenced enumeration — complete, continuation, pinned window,
// short-cap, fence-lapse witnesses ──────────────────────────────────────────
// Helper to build a minimal valid-looking relay event for the pubkey.
function makeFakeEvent(pubkey, createdAt) {
return {
id: `${createdAt.toString(16).padStart(8, "0")}${"0".repeat(56)}`,
pubkey,
kind: 30078,
content: "",
tags: [],
created_at: 1000,
created_at: createdAt,
sig: "s".repeat(128),
});
};
}
test("fetchAndMerge_emptyRelay_setsLoadComplete", async () => {
// A relay with no events should produce an empty first band → complete.
globalThis.window.localStorage = makeLocalStorage();
const pubkey = "a".repeat(64);
let subscribeCallCount = 0;
const fakeRelay = {
fetchEvents: async () => [],
publishEvent: async () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async (_filter, _handler) => {
subscribeCallCount++;
return () => {};
},
};
const mgr = new ReadStateManager(pubkey, fakeRelay);
await mgr.fetchAndMerge();
assert.equal(
mgr.isLoadComplete,
true,
"empty relay must produce complete load",
);
// fence subscription must have been established (and then unsubscribed by fetchAndMerge).
assert.equal(
subscribeCallCount,
1,
"fence subscription must be set up exactly once",
);
mgr.destroy();
});
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.
// 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];
},
publishEvent: async () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async (_filter, _handler) => () => {},
};
const mgr = new ReadStateManager(pubkey, fakeRelay);
await mgr.fetchAndMerge();
assert.equal(
mgr.isLoadComplete,
true,
"single-event relay must produce complete load after pinned-window discharge",
);
mgr.destroy();
});
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),
makeFakeEvent(pubkey, 2000),
];
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.
return events;
},
publishEvent: async () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async (_filter, _handler) => () => {},
};
const mgr = new ReadStateManager(pubkey, fakeRelay);
await mgr.fetchAndMerge();
assert.equal(
mgr.isLoadComplete,
false,
"pinned window returning ≥ max(C,L) must produce incomplete load",
);
mgr.destroy();
});
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 = {
fetchEvents: async () => [],
publishEvent: async () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async () => {
throw new Error("connection refused");
},
};
const mgr = new ReadStateManager(pubkey, fakeRelay);
await mgr.fetchAndMerge();
assert.equal(
mgr.isLoadComplete,
false,
"fence failure must produce incomplete load",
);
mgr.destroy();
});
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 = [
makeFakeEvent(pubkey, 1000),
makeFakeEvent(pubkey, 1000),
makeFakeEvent(pubkey, 1000),
];
let publishCalls = 0;
const fakeRelay = {
fetchEvents: async () => fakeEvents,
fetchEvents: async (filter) => {
if (filter.since !== undefined) return events; // pinned window
return events; // band
},
publishEvent: async () => {
publishCalls++;
},
subscribeLive: async () => () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async (_f, _h) => () => {},
};
const pubkey = "a".repeat(64);
const mgr = new ReadStateManager(pubkey, fakeRelay);
// Seed some context reads so there's something to publish.
mgr.markContextRead("channel-test", 1000);
// Run fetchAndMerge (internals) via initialize() — the relay returns a full
// page, so isLoadComplete must remain false.
// We override subscribeLive to avoid hanging on a real subscription.
mgr.markContextRead("ch", 1000);
await mgr.fetchAndMerge();
// 1. publish() must be blocked (isLoadComplete=false).
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 isLoadComplete=false",
);
assert.equal(publishCalls, 0, "publish must be blocked when load incomplete");
// 2. markChannelUnread must return load_incomplete.
const unreadResult = mgr.markChannelUnread("channel-test");
assert.equal(unreadResult.success, false);
assert.equal(
unreadResult.reason,
"load_incomplete",
"markChannelUnread must refuse with 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 readResult = mgr.markChannelRead("channel-test");
assert.equal(readResult.success, false);
assert.equal(
readResult.reason,
"load_incomplete",
"markChannelRead must refuse with load_incomplete",
);
const rr = mgr.markChannelRead("ch");
assert.equal(rr.success, false);
assert.equal(rr.reason, "load_incomplete");
// 4. deleteExtraSlots must be blocked (isLoadComplete=false).
// Inject a fake extraSlotId to verify deletion is suppressed.
mgr.extraSlotIds = ["fakeextraslot000"];
// 4. deleteExtraSlots must be blocked.
mgr.extraSlotIds = ["fakeextraslot0000000000000000000"];
await mgr.deleteExtraSlots();
assert.equal(
publishCalls,
0,
"deleteExtraSlots must not call publishEvent when isLoadComplete=false",
"deleteExtraSlots must not publish when load incomplete",
);
mgr.destroy();
});
// ── Test 3: visible refusal at budget exhaustion and uint32 max ───────────────
// ── Test 2b: retry path — a second fetchAndMerge can clear incomplete ─────────
test("fetchAndMerge_retryClears_incomplete", async () => {
globalThis.window.localStorage = makeLocalStorage();
const pubkey = "f".repeat(64);
let callRound = 0;
const fakeRelay = {
fetchEvents: async (filter) => {
callRound++;
if (callRound <= 3) {
// First attempt: pinned window fires at round 2, returns cap-many events.
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,
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
await mgr.fetchAndMerge();
assert.equal(
mgr.isLoadComplete,
true,
"retry with empty relay must set complete",
);
mgr.destroy();
});
// ── Test 3: live events go through structured ingest ──────────────────────────
test("handleIncomingEvent_liveOverride_updatesRegisterViaIngest", async () => {
// Construct a fake event whose content decrypts to a NIP-RS blob with an
// override register. Use a __TAURI_INTERNALS__ mock so parseReadStateEvent
// can decrypt without a real NIP-44 key.
globalThis.window.localStorage = makeLocalStorage();
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:<ctx>=3, ov_c:<ctx>=1, ov_b:<ctx>=50, <ctx>=<frontier>
const blobContexts = {
[rawCtx]: channelFrontier,
[`ov_s:${rawCtx}`]: 3,
[`ov_c:${rawCtx}`]: 1,
[`ov_b:${rawCtx}`]: 50,
};
const plaintext = JSON.stringify({
v: 1,
client_id: "other-device",
contexts: blobContexts,
});
// Install Tauri IPC mock so nip44_decrypt_from_self returns our plaintext.
globalThis.window.__TAURI_INTERNALS__ = {
invoke: async (command, args) => {
if (command === "nip44_decrypt_from_self") {
if (args.ciphertext === "FAKE_CIPHER") return plaintext;
throw new Error("unknown ciphertext");
}
throw new Error(`Unexpected Tauri command: ${command}`);
},
};
let liveHandler = null;
const fakeRelay = {
fetchEvents: async () => [],
publishEvent: async () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async (_filter, handler) => {
liveHandler = handler;
return () => {};
},
};
try {
const mgr = new ReadStateManager(pubkey, fakeRelay);
await mgr.initialize();
// Build fake NIP-RS event.
const fakeEvent = {
id: "b".repeat(64),
pubkey,
created_at: 2_000_000,
kind: 30078,
tags: [
["d", "read-state:a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6"],
["t", "read-state"],
],
content: "FAKE_CIPHER",
sig: "s".repeat(128),
};
// Deliver via the live subscription callback (same path as relay push).
assert.ok(liveHandler !== null, "live subscription must be established");
liveHandler(fakeEvent); // void-wrapped; wait for async completion
await new Promise((r) => setTimeout(r, 50));
// 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");
// ── existing-key live clear (higher C defeating S) ───────────────────
// A follow-up event with S=3, C=4 (C > S → inactive/tombstone).
const blobClear = {
[rawCtx]: channelFrontier,
[`ov_c:${rawCtx}`]: 4, // tombstone floor: max(S=3,C=1)+1=4
};
const ptClear = JSON.stringify({
v: 1,
client_id: "other-device-2",
contexts: blobClear,
});
const fakeEventClear = {
id: "c".repeat(64),
pubkey,
created_at: 2_000_001,
kind: 30078,
tags: [
["d", "read-state:b1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6"],
["t", "read-state"],
],
content: "FAKE_CIPHER_2",
sig: "s".repeat(128),
};
globalThis.window.__TAURI_INTERNALS__.invoke = async (command, args) => {
if (command === "nip44_decrypt_from_self") {
if (args.ciphertext === "FAKE_CIPHER_2") return ptClear;
if (args.ciphertext === "FAKE_CIPHER") return plaintext;
throw new Error("unknown ciphertext");
}
throw new Error(`Unexpected Tauri command: ${command}`);
};
liveHandler(fakeEventClear); // void-wrapped; wait for async completion
await new Promise((r) => setTimeout(r, 50));
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
assert.equal(
regAfterClear.c,
4,
"C must be 4 (tombstone floor from clear event)",
);
const liveness = mgr.getOverrideLiveness(rawCtx);
assert.ok(liveness !== null, "liveness must be available");
assert.equal(
liveness.active,
false,
"override must be inactive after clear",
);
mgr.destroy();
} finally {
delete globalThis.window.__TAURI_INTERNALS__;
}
});
// ── Test 4: fetch-before-write failure → zero publishes ──────────────────────
test("publish_fetchOwnBlobFails_doesNotPublish", async () => {
globalThis.window.localStorage = makeLocalStorage();
const pubkey = "bb".repeat(32);
let publishCalls = 0;
const fakeRelay = {
fetchEvents: async (filter) => {
// Own-blob fetch (has #d filter) → fail.
if (filter["#d"]) throw new Error("relay unreachable");
return [];
},
publishEvent: async () => {
publishCalls++;
},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async (_f, _h) => () => {},
};
const mgr = new ReadStateManager(pubkey, fakeRelay);
mgr.isLoadComplete = true;
mgr.markContextRead("ch", 1000);
await mgr.publish();
assert.equal(
publishCalls,
0,
"publish must not call publishEvent when fetchOwnBlobBeforePublish fails",
);
mgr.destroy();
});
// ── Test 5: durability — register survives restart before debounce ────────────
test("overrideRegister_survivesRestartBeforeDebounce", () => {
// markChannelUnread persists the register via persistLocalState.
// A new ReadStateManager constructed from the same localStorage must see it.
const ls = makeLocalStorage();
globalThis.window.localStorage = ls;
const pubkey = "cc".repeat(32);
const fakeRelay = {
fetchEvents: async () => [],
publishEvent: async () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async (_f, _h) => () => {},
};
const mgr1 = new ReadStateManager(pubkey, fakeRelay);
mgr1.isLoadComplete = true;
mgr1.effectiveState.set("restart-ch", 1000);
mgr1.publishableContextIds.add("restart-ch");
const result = mgr1.markChannelUnread("restart-ch");
assert.equal(result.success, true, "mark-unread must succeed");
mgr1.destroy();
// Construct a new manager on the same localStorage — no relay fetch yet.
globalThis.window.localStorage = ls;
const mgr2 = new ReadStateManager(pubkey, fakeRelay);
mgr2.hydrateFromLocalStorage();
const liveness = mgr2.getOverrideLiveness("restart-ch");
assert.ok(liveness !== null, "register must be hydrated after restart");
assert.equal(
liveness.active,
true,
"override must still be active after restart",
);
mgr2.destroy();
});
// ── Test 6: durability — tombstone floor survives restart with fetch failure ──
test("overrideRegister_tombstoneFloorSurvivesRestartWithFetchFailure", async () => {
const ls = makeLocalStorage();
globalThis.window.localStorage = ls;
const pubkey = "dd".repeat(32);
const goodRelay = {
fetchEvents: async () => [],
publishEvent: async () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async (_f, _h) => () => {},
};
// Establish a mark-read tombstone floor: S=1, C=max(S,C)+1=2 (clear-wins → inactive).
const mgr1 = new ReadStateManager(pubkey, goodRelay);
mgr1.isLoadComplete = true;
mgr1.effectiveState.set("tombstone-ch", 500);
mgr1.publishableContextIds.add("tombstone-ch");
mgr1.markChannelUnread("tombstone-ch"); // S→max(0,0)+1=1, C=0, B→frontier
mgr1.markChannelRead("tombstone-ch"); // C→max(1,0)+1=2 (clear-wins)
const tomb = mgr1.overrideRegisters.get("tombstone-ch");
assert.ok(tomb, "tombstone register must exist");
assert.equal(tomb.s, 1);
assert.equal(tomb.c, 2); // max(S=1,C=0)+1 = 2
mgr1.destroy();
// New manager: initialize() with a relay that throws on fetchEvents.
// Tombstone must still be present even after failed fetch.
globalThis.window.localStorage = ls;
const failRelay = {
fetchEvents: async () => {
throw new Error("network unavailable");
},
publishEvent: async () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
subscribeLive: async () => {
throw new Error("network unavailable");
},
};
const mgr2 = new ReadStateManager(pubkey, failRelay);
await mgr2.initialize(); // fetch fails, but hydration must have run first
const reg = mgr2.overrideRegisters.get("tombstone-ch");
assert.ok(reg, "tombstone register must survive restart with fetch failure");
assert.equal(reg.s, 1, "S must be preserved");
assert.equal(reg.c, 2, "C must be preserved (tombstone floor: max(1,0)+1=2)");
mgr2.destroy();
});
// ── Test 7: budget planner ────────────────────────────────────────────────────
test("markChannelUnread_visibleRefusal_atBudgetExhaustionAndUint32Max", () => {
const mgr = makeManager();
// Simulate completed load so mark operations are not blocked by load_incomplete.
mgr.isLoadComplete = true;
// ── uint32 max refusal ────────────────────────────────────────────────────
// Inject a register where S is already at uint32 max and S == C (so
// max(S,C)+1 would overflow).
const UINT32_MAX = 0xffffffff;
const overflowCtx = "overflow-channel";
mgr.overrideRegisters.set(overflowCtx, {
@@ -774,41 +1244,208 @@ test("markChannelUnread_visibleRefusal_atBudgetExhaustionAndUint32Max", () => {
"markChannelUnread must refuse with uint32_overflow when S is at max",
);
// ── budget exhaustion refusal ─────────────────────────────────────────────
// Fill effectiveState with enough channel keys to consume the entire 32 KiB
// budget, then attempt to add a new override group. With no prunable entries
// (no msg:/thread: keys), the budget check must fail.
//
// Each channel key is ~70 bytes. 32768 / 70 ≈ 468 keys to fill the budget.
// Use 500 keys to ensure we are well over budget on the channel-only side.
const budgetCtx = "budget-channel-00000000000000000000000000000000";
for (let i = 0; i < 500; i++) {
const ch = `ch-${i.toString().padStart(64, "0")}`;
mgr.effectiveState.set(ch, i + 1);
mgr.publishableContextIds.add(ch);
// ── multi-slot success: 700 frontier-only channels → split planner must succeed ─
// Thufir's deterministic witness: 700 plain frontier channels overflow a single slot
// but the splitter can spread them across ≤ 8 slots. A new override group in the
// primary must succeed because the pure candidate planner tries the split path.
const splitMgr = makeManager("f".repeat(64));
splitMgr.isLoadComplete = true;
for (let i = 0; i < 700; i++) {
const ctx = `frontier-ch-${i.toString().padStart(60, "0")}`;
splitMgr.effectiveState.set(ctx, 1000 + i);
splitMgr.publishableContextIds.add(ctx);
}
// The new context has no existing override register — mark it unread.
mgr.effectiveState.set(budgetCtx, 999);
mgr.publishableContextIds.add(budgetCtx);
const splitCtx = `new-override-ctx-${"n".repeat(48)}`;
const splitResult = splitMgr.markChannelUnread(splitCtx);
assert.equal(
splitResult.success,
true,
"700 frontier-only channels must allow a new override via multi-slot split",
);
assert.ok(
splitMgr.overrideRegisters.has(splitCtx),
"override register must be committed on split success",
);
splitMgr.destroy();
const budgetResult = mgr.markChannelUnread(budgetCtx);
// 500 channel keys × ~70 bytes ≈ 35 KB > 32 KiB; with no prunable entries
// (no msg:/thread: keys), the budget check must fail.
// ── near-limit refusal: all 8 slots insufficient → budget_exhausted, no mutation ─
// Pack so many non-evictable override groups that even 8 slots cannot accommodate
// the new entry. Each override group contributes ov_s+ov_c+ov_b+frontier ≈ 4 keys
// × ~75 bytes each + JSON overhead. 200 groups × 4 keys ≈ 300 bytes each ≈ 60 KB
// per slot if split across 8 → ~7.5 KB per slot, which fits. We need them to NOT fit.
// Easiest: fill primary to near-capacity with 250 big-key override groups so even
// splitting all 8 slots still cannot carry the new group in primary (which must
// hold ALL override groups per spec, so they all go in slot 0).
const fullMgr = makeManager("aa".repeat(32));
fullMgr.isLoadComplete = true;
// 250 override groups with long context IDs ≈ 250 × (4 keys × ~100 bytes) = 100 KB
// → exceeds READ_STATE_MAX_PLAINTEXT_BYTES (32 KB) even in slot 0 alone.
for (let i = 0; i < 250; i++) {
const ctx = `ov-ch-${i.toString().padStart(60, "0")}`;
fullMgr.overrideRegisters.set(ctx, { s: 1, c: 0, b: 0 });
fullMgr.effectiveState.set(ctx, 1000 + i);
fullMgr.publishableContextIds.add(ctx);
}
const nearCtx = `near-limit-ctx-${"z".repeat(49)}`;
const nearResult = fullMgr.markChannelUnread(nearCtx);
assert.equal(
budgetResult.success,
nearResult.success,
false,
"markChannelUnread must refuse when channel-only keys exhaust the budget",
"near-limit manager with 250 non-evictable override groups must refuse",
);
assert.equal(
budgetResult.reason,
"budget_exhausted",
"refusal must name budget_exhausted (not a silent drop or panic)",
assert.equal(nearResult.reason, "budget_exhausted");
assert.ok(
!fullMgr.overrideRegisters.has(nearCtx),
"budget_exhausted must not mutate overrideRegisters",
);
// Whether it succeeds or fails, the manager must not have corrupted state.
// Verify the uint32_overflow register was not mutated.
fullMgr.destroy();
// Verify the uint32_overflow register was not mutated in the original mgr.
const reg = mgr.overrideRegisters.get(overflowCtx);
assert.ok(reg, "overflow channel register must still exist");
assert.equal(reg.s, UINT32_MAX, "overflow register S must be unchanged");
mgr.destroy();
});
// ── Test 8: persistence failure fails the mark ───────────────────────────────
test("markChannelUnread_storageFailure_returnsStorageFailed", () => {
// Use a throwing localStorage to simulate quota failure.
const throwingLS = makeLocalStorage();
const originalSetItem = throwingLS.setItem.bind(throwingLS);
// Allow initial writes (ClientId, slotId), then fail on override writes.
let writeCount = 0;
throwingLS.setItem = (key, value) => {
writeCount++;
if (writeCount > 2) throw new Error("QuotaExceededError");
originalSetItem(key, value);
};
globalThis.window.localStorage = throwingLS;
const fakeRelay = {
fetchEvents: async () => [],
publishEvent: async () => {},
subscribeLive: () => () => {},
};
const mgr = new ReadStateManager("st".repeat(32), fakeRelay);
mgr.isLoadComplete = true;
mgr.effectiveState.set("storage-test-ch", 1000);
const result = mgr.markChannelUnread("storage-test-ch");
assert.equal(
result.success,
false,
"markChannelUnread must fail when localStorage throws",
);
assert.equal(result.reason, "storage_failed");
mgr.destroy();
});
// ── Test 9: inactive existing register still gets C-bump on markChannelRead ───
test("markChannelRead_inactiveExistingRegister_performsCBump", () => {
const mgr = makeManager();
mgr.isLoadComplete = true;
// Inject an inactive register: S=1, C=2 (clear-wins: C > S → inactive).
const ctx = "inactive-ch";
mgr.overrideRegisters.set(ctx, { s: 1, c: 2, b: 0 });
mgr.publishableContextIds.add(ctx);
mgr.effectiveState.set(ctx, 100);
// Verify it is indeed inactive.
const livenessBefore = mgr.getOverrideLiveness(ctx);
assert.ok(livenessBefore !== null, "register must exist");
assert.equal(livenessBefore.active, false, "register must be inactive");
// markChannelRead must still perform the C-bump (spec NIP-RS.md:537-539).
const result = mgr.markChannelRead(ctx);
assert.equal(
result.success,
true,
"markChannelRead must succeed on inactive register",
);
const reg = mgr.overrideRegisters.get(ctx);
assert.ok(reg, "register must still exist after markChannelRead");
// newC = max(S=1, C=2) + 1 = 3
assert.equal(
reg.c,
3,
"C must be bumped to max(S,C)+1=3 even when already inactive",
);
assert.equal(reg.s, 1, "S must be unchanged");
mgr.destroy();
});
// ── Test 10: coordinate dedupe — newer version wins, older version dropped ────
test("deduplicateByCoordinate_newerVersionWins_olderDropped", async () => {
const { deduplicateByCoordinate } = await import(
"./readStateFencedLoader.ts"
);
const pubkey = "de".repeat(32);
const dTag = "read-state:a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6";
const older = {
id: "b".repeat(64),
pubkey,
created_at: 1_000,
kind: 30078,
tags: [["d", dTag]],
content: "older",
sig: "s".repeat(128),
};
const newer = {
id: "a".repeat(64), // lower id — for tie-break test below
pubkey,
created_at: 2_000,
kind: 30078,
tags: [["d", dTag]],
content: "newer",
sig: "s".repeat(128),
};
const deduped = deduplicateByCoordinate([older, newer]);
assert.equal(deduped.length, 1, "dedup must yield one event");
assert.equal(deduped[0].content, "newer", "newer created_at must win");
// Tie-break: same created_at, lower id wins.
const tie1 = { ...older, created_at: 3_000, id: "c".repeat(64) };
const tie2 = { ...newer, created_at: 3_000, id: "a".repeat(64) };
const tieDuped = deduplicateByCoordinate([tie1, tie2]);
assert.equal(tieDuped.length, 1, "tie-break dedup must yield one event");
assert.equal(tieDuped[0].id, "a".repeat(64), "lower id must win on tie");
});
// ── Test 11: lapse mid-enumeration → incomplete ───────────────────────────────
test("fetchAndMerge_lapseMidEnumeration_setsLoadIncomplete", async () => {
// Simulate a reconnect (generation change) after the first band is fetched.
globalThis.window.localStorage = makeLocalStorage();
const pubkey = "ef".repeat(32);
const event = makeFakeEvent(pubkey, 1000);
let generation = 0;
let reconnectCb = null;
const fakeRelay = {
fetchEvents: async (filter) => {
if (filter.since !== undefined) {
// Pinned window: simulate a reconnect BEFORE returning.
generation++;
reconnectCb?.(); // fire reconnect listener
return [event];
}
return [event]; // initial band
},
publishEvent: async () => {},
subscribeToReconnects: (cb) => {
reconnectCb = cb;
return () => {
reconnectCb = null;
};
},
getConnectionGeneration: () => generation,
subscribeLive: async (_f, _h) => () => {},
};
const mgr = new ReadStateManager(pubkey, fakeRelay);
await mgr.fetchAndMerge();
assert.equal(
mgr.isLoadComplete,
false,
"lapse during enumeration must produce incomplete load",
);
mgr.destroy();
});

636
desktop/src/features/channels/readState/readStateManager.ts
View File

@@ -12,6 +12,8 @@ import {
isOverrideKey,
encodeOverrideGroup,
isOverrideActive,
escapeFrontierKey,
unescapeFrontierKey,
localExtraSlotIdsKey,
type ReadStateBlob,
type OverrideRegister,
@@ -21,20 +23,21 @@ import {
parseReadStateEvent,
mergeReadStateEventsStructured,
type MergedReadState,
type ParsedReadStateEvent,
} from "@/features/channels/readState/readStateSnapshot";
import {
readStoredReadState,
writeStoredReadState,
} from "@/features/channels/readState/readStateStorage";
import {
fencedEnumerationLoad,
deduplicateByCoordinate,
} from "@/features/channels/readState/readStateFencedLoader";
import { setLocalStorageItemWithRecovery } from "@/shared/lib/localStorageQuota";
import { truncatePubkey } from "@/shared/lib/pubkey";
const CLIENT_ID_KEY_PREFIX = "buzz.nip-rs.client-id";
const SLOT_ID_KEY_PREFIX = "buzz.nip-rs.slot-id";
const DEBOUNCE_MS = 5_000;
// Full-state fetch limit; sub-limit response confirms completeness (no truncation).
const READ_STATE_FULL_FETCH_LIMIT = 5_000;
type OwnBlobEntry = { blob: ReadStateBlob; createdAt: number };
export type MarkResult =
| { success: true }
@@ -44,9 +47,18 @@ export type MarkResult =
| "uint32_overflow"
| "budget_exhausted"
| "load_incomplete"
| "already_inactive";
| "already_inactive"
| "storage_failed";
};
/** Explicit delta returned by `ingest()`. */
export type IngestDelta = {
/** Contexts whose frontier or override register changed. */
changedContexts: Set<string>;
/** True if any canonical form changed (triggers convergence publish). */
canonicalChanged: boolean;
};
function generateHex(bytes: number): string {
const arr = new Uint8Array(bytes);
crypto.getRandomValues(arr);
@@ -62,14 +74,6 @@ function getOrCreatePersisted(key: string, generator: () => string): string {
return value;
}
function clientIdKey(pubkey: string): string {
return `${CLIENT_ID_KEY_PREFIX}:${pubkey}`;
}
function slotIdKey(pubkey: string): string {
return `${SLOT_ID_KEY_PREFIX}:${pubkey}`;
}
function loadExtraSlotIds(pubkey: string): string[] {
try {
const raw = localStorage.getItem(localExtraSlotIdsKey(pubkey));
@@ -95,10 +99,7 @@ export type ApplyRemoteContextResult = "unchanged" | "advanced";
export type ContextParentResolver = (contextId: string) => string | null;
/**
* NIP-RS Hierarchical Frontier: `effective(ctx) = max(merged[ctx], effective(parent(ctx)))`.
* Thread→channel relationship derived from event graph via `parentResolver`.
*/
/** NIP-RS Hierarchical Frontier: `effective(ctx) = max(merged[ctx], effective(parent(ctx)))`. */
export function resolveEffectiveTimestamp(args: {
effectiveState: Map<string, number>;
contextId: string;
@@ -139,22 +140,13 @@ export function applyRemoteContextTimestamp(args: {
return result;
}
/**
* Result of a `splitContextsIntoBudgetedSlots` call.
*/
/** Result of `splitContextsIntoBudgetedSlots`. */
export interface SlotSplitResult {
/** Contexts record for each slot (primary slot first). */
slots: Array<Record<string, number>>;
/** Extra slot IDs beyond the first. Length is `slots.length - 1`. */
extraSlotIds: string[];
}
/**
* Partition `channelEntries` across slots so each slot fits within `maxBytes`.
* Override-bearing groups are pinned to slot 0; remaining keys distributed round-robin.
* Returns `{ slots, extraSlotIds }` or null when maxSlots is insufficient.
* Exported for unit testing; callers should prefer `splitContextsIntoSlots()`.
*/
/** Partition `channelEntries` across slots, override groups pinned to slot 0. Exported for testing. */
export function splitContextsIntoBudgetedSlots(args: {
channelEntries: [string, number][];
threadMsgEntries: [string, number][];
@@ -178,14 +170,14 @@ export function splitContextsIntoBudgetedSlots(args: {
const blobFor = (c: Record<string, number>) =>
JSON.stringify({ v: 1, client_id: clientId, contexts: c });
const overrideSuffixes = new Set<string>();
const overrideRawIds = new Set<string>();
for (const [key] of channelEntries) {
if (isOverrideKey(key)) overrideSuffixes.add(key.slice(5)); // all ov_?: prefixes are 5 chars
if (isOverrideKey(key)) overrideRawIds.add(key.slice(5));
}
const pinnedEntries: [string, number][] = [];
const roundRobinEntries: [string, number][] = [];
for (const [key, ts] of channelEntries) {
if (isOverrideKey(key) || overrideSuffixes.has(key)) {
if (isOverrideKey(key) || overrideRawIds.has(unescapeFrontierKey(key))) {
pinnedEntries.push([key, ts]);
} else {
roundRobinEntries.push([key, ts]);
@@ -280,18 +272,17 @@ export class ReadStateManager {
private pendingSyncedAdvances = new Set<string>();
private destroyed = false;
private parentResolver: ContextParentResolver | null = null;
/** Override registers keyed by raw context ID. */
private overrideRegisters = new Map<string, OverrideRegister>();
/** False until initial full-state fetch returns sub-limit (no truncation). Gated ops blocked while false. */
private isLoadComplete = false;
constructor(pubkey: string, relayClient: RelayClient) {
this.pubkey = pubkey;
this.relayClient = relayClient;
this.clientId = getOrCreatePersisted(clientIdKey(pubkey), () =>
crypto.randomUUID(),
this.clientId = getOrCreatePersisted(
`${CLIENT_ID_KEY_PREFIX}:${pubkey}`,
() => crypto.randomUUID(),
);
this.slotId = getOrCreatePersisted(slotIdKey(pubkey), () =>
this.slotId = getOrCreatePersisted(`${SLOT_ID_KEY_PREFIX}:${pubkey}`, () =>
generateHex(16),
);
this.extraSlotIds = loadExtraSlotIds(pubkey);
@@ -299,10 +290,6 @@ export class ReadStateManager {
async initialize(): Promise<void> {
if (this.initialized || this.destroyed) return;
console.debug(
`[ReadStateManager] initialize pubkey=${truncatePubkey(this.pubkey)} clientId=${this.clientId.substring(0, 8)}… slotId=${this.slotId}`,
);
this.hydrateFromLocalStorage();
await this.fetchAndMerge();
if (this.destroyed) return;
@@ -315,11 +302,7 @@ export class ReadStateManager {
) {
this.schedulePublish();
}
this.initialized = true;
console.debug(
`[ReadStateManager] initialize complete maxFetchedCreatedAt=${this.maxFetchedCreatedAt} contexts=${this.effectiveState.size}`,
);
this.notifyListeners();
}
@@ -364,19 +347,10 @@ export class ReadStateManager {
});
}
/**
* The context's OWN merged read marker, WITHOUT the hierarchical parent term.
* Use this for background-channel threads (getEffectiveTimestamp folds in
* parentResolver which maps threads to the active channel).
*/
getOwnTimestamp(contextId: string): number | null {
return this.effectiveState.get(contextId) ?? null;
}
/**
* Inject the thread→channel parent resolver (NIP-RS.md:136-139).
* The hierarchical max in getEffectiveTimestamp is a no-op until this is set.
*/
setContextParentResolver(resolver: ContextParentResolver | null): void {
this.parentResolver = resolver;
}
@@ -403,70 +377,97 @@ export class ReadStateManager {
}
private async fetchAndMerge(): Promise<void> {
let events: RelayEvent[];
try {
events = await this.relayClient.fetchEvents({
kinds: [KIND_READ_STATE],
authors: [this.pubkey],
"#t": ["read-state"],
// No `since` — a finite window can exclude the sole tombstone carrier.
limit: READ_STATE_FULL_FETCH_LIMIT,
});
} catch (error) {
console.debug("[ReadStateManager] fetchAndMerge failed:", error);
return;
}
// Truncation guard: at-limit response may be incomplete; block gated ops.
this.isLoadComplete = events.length < READ_STATE_FULL_FETCH_LIMIT;
if (!this.isLoadComplete) {
if (this.destroyed) return;
const result = await fencedEnumerationLoad(this.relayClient, this.pubkey);
if (this.destroyed) return;
this.isLoadComplete = result.complete;
if (!result.complete) {
console.warn(
`[ReadStateManager] fetchAndMerge: ${events.length} events (== limit) — gated ops blocked`,
"[ReadStateManager] fetchAndMerge: load incomplete — gated ops blocked",
);
}
await this.mergeEvents(events);
await this.ingest(result.merged);
this.persistLocalState();
this.notifyListeners();
}
private async mergeEvents(events: RelayEvent[]): Promise<void> {
// Structured merge via slice-1 protocol layer.
const merged: MergedReadState = await mergeReadStateEventsStructured(
events,
this.pubkey,
);
/** Retry a failed load: resets isLoadComplete, re-runs the full enumeration. */
async retryLoad(): Promise<void> {
if (this.destroyed) return;
this.isLoadComplete = false;
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.
*/
private async ingest(merged: MergedReadState): Promise<IngestDelta> {
const delta: IngestDelta = {
changedContexts: new Set(),
canonicalChanged: false,
};
// ── Frontier merge ────────────────────────────────────────────────────
for (const [rawCtx, ts] of merged.frontiers) {
const result = applyRemoteContextTimestamp({
effectiveState: this.effectiveState,
contextSourceCreatedAt: this.contextSourceCreatedAt,
contextId: rawCtx,
eventCreatedAt: 0, // per-event timestamps updated in the loop below
eventCreatedAt: 0,
timestamp: ts,
});
if (result !== "unchanged") {
delta.changedContexts.add(rawCtx);
this.pendingSyncedAdvances.add(rawCtx);
this.publishableContextIds.add(rawCtx);
}
}
// ── Override register merge — compare component-by-component ─────────
for (const [rawCtx, reg] of merged.overrides) {
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,
);
this.publishableContextIds.add(rawCtx);
const merged2 = 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;
if (changed) {
this.overrideRegisters.set(rawCtx, merged2);
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;
}
}
const ownBlobsBySlot = new Map<string, OwnBlobEntry>();
for (const event of events) {
const parsed = await parseReadStateEvent(event, this.pubkey);
if (!parsed) continue;
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<IngestDelta> {
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;
this.maxFetchedCreatedAt = Math.max(
this.maxFetchedCreatedAt,
parsed.createdAt,
@@ -476,37 +477,31 @@ export class ReadStateManager {
if (parsed.createdAt > src)
this.contextSourceCreatedAt.set(rawCtx, parsed.createdAt);
}
// Rotate slotId if another client_id squats on our coord.
if (
parsed.dTag === `read-state:${this.slotId}` &&
parsed.blob.client_id !== this.clientId
) {
this.slotId = generateHex(16);
setLocalStorageItemWithRecovery(slotIdKey(this.pubkey), this.slotId);
setLocalStorageItemWithRecovery(
`${SLOT_ID_KEY_PREFIX}:${this.pubkey}`,
this.slotId,
);
}
if (parsed.blob.client_id === this.clientId) {
const existing = ownBlobsBySlot.get(parsed.dTag);
if (!existing || parsed.createdAt > existing.createdAt) {
ownBlobsBySlot.set(parsed.dTag, {
blob: parsed.blob,
createdAt: parsed.createdAt,
});
}
}
}
if (ownBlobsBySlot.size > 0) {
const unionContexts: Record<string, number> = {};
for (const { blob } of ownBlobsBySlot.values()) {
for (const [key, ts] of Object.entries(blob.contexts)) {
const unionContexts: Record<string, number> = {
...this.lastPublishedContexts,
};
for (const [key, ts] of Object.entries(parsed.blob.contexts)) {
const ex = unionContexts[key];
if (ex === undefined || ts > ex) unionContexts[key] = ts;
}
for (const contextId of Object.keys(blob.contexts))
for (const contextId of Object.keys(parsed.blob.contexts))
this.publishableContextIds.add(contextId);
this.lastPublishedContexts = unionContexts;
}
this.lastPublishedContexts = unionContexts;
}
const merged = await mergeReadStateEventsStructured(deduped, this.pubkey);
return this.ingest(merged);
}
private async startLiveSubscription(): Promise<void> {
@@ -515,7 +510,6 @@ export class ReadStateManager {
{
kinds: [KIND_READ_STATE],
authors: [this.pubkey],
"#t": ["read-state"],
limit: READ_STATE_FETCH_LIMIT,
},
(event: RelayEvent) => {
@@ -527,60 +521,26 @@ export class ReadStateManager {
return;
}
this.unsubscribeLive = unsub;
console.debug("[ReadStateManager] live subscription established");
} catch (error) {
console.debug("[ReadStateManager] live subscription FAILED:", error);
} catch {
// Live subscription is best-effort; missed events will be caught on reconnect.
}
}
private async handleIncomingEvent(event: RelayEvent): Promise<void> {
if (event.pubkey !== this.pubkey || this.destroyed) return;
console.debug(
`[ReadStateManager] incoming event=${event.id.substring(0, 8)}… created_at=${event.created_at}`,
);
const parsed = await parseReadStateEvent(event, this.pubkey);
if (!parsed) return;
this.maxFetchedCreatedAt = Math.max(
this.maxFetchedCreatedAt,
parsed.createdAt,
);
const { blob } = parsed;
let anyAdvanced = false;
for (const [ctx, ts] of Object.entries(blob.contexts)) {
const result = applyRemoteContextTimestamp({
effectiveState: this.effectiveState,
contextSourceCreatedAt: this.contextSourceCreatedAt,
contextId: ctx,
timestamp: ts,
eventCreatedAt: parsed.createdAt,
});
if (result === "advanced") {
this.pendingSyncedAdvances.add(ctx);
anyAdvanced = true;
}
if (!this.publishableContextIds.has(ctx)) {
this.publishableContextIds.add(ctx);
anyAdvanced = true;
}
}
console.debug(
`[ReadStateManager] incoming result anyAdvanced=${anyAdvanced} clientId=${blob.client_id.substring(0, 8)}…`,
);
if (anyAdvanced) {
const delta = await this.ingestEvents([event]);
if (delta.changedContexts.size > 0 || this.pendingSyncedAdvances.size > 0) {
this.persistLocalState();
this.notifyListeners();
// Another client instance: schedule re-publish so our blob converges.
if (blob.client_id !== this.clientId) this.schedulePublish();
const parsed = await parseReadStateEvent(event, this.pubkey);
if (parsed?.blob.client_id !== this.clientId && delta.canonicalChanged) {
this.schedulePublish();
}
}
}
private schedulePublish(): void {
if (this.debounceTimer !== null) {
window.clearTimeout(this.debounceTimer);
}
if (this.debounceTimer !== null) window.clearTimeout(this.debounceTimer);
this.debounceTimer = window.setTimeout(() => {
this.debounceTimer = null;
void this.publish();
@@ -588,30 +548,26 @@ export class ReadStateManager {
}
private async publish(): Promise<void> {
console.debug(`[ReadStateManager] publish starting slotId=${this.slotId}`);
if (!this.isLoadComplete) {
console.debug("[ReadStateManager] publish blocked: isLoadComplete=false");
if (!this.isLoadComplete) return;
if (!(await this.fetchOwnBlobBeforePublish())) {
console.warn(
"[ReadStateManager] publish aborted: read-before-write failed",
);
return;
}
await this.fetchOwnBlobBeforePublish();
const contexts = this.currentContexts();
if (contexts === null) {
await this.publishSplitSlots();
return;
}
// Transitioning from split to single: delete stale extra-slot blobs.
if (this.extraSlotIds.length > 0) {
await this.deleteExtraSlots();
this.lastPublishedContexts = {};
}
if (this.isIdenticalToLastPublished(contexts)) return;
await this.publishOneSlot(this.slotId, contexts);
}
/** Publish a single slot's blob. Updates lastPublishedContexts and maxFetchedCreatedAt on success. */
private async publishOneSlot(
slotId: string,
contexts: Record<string, number>,
@@ -619,10 +575,6 @@ export class ReadStateManager {
const blob: ReadStateBlob = { v: 1, client_id: this.clientId, contexts };
try {
const ciphertext = await nip44EncryptToSelf(JSON.stringify(blob));
const tags: string[][] = [
["d", `read-state:${slotId}`],
["t", "read-state"],
];
const createdAt = Math.max(
Math.floor(Date.now() / 1_000),
this.maxFetchedCreatedAt + 1,
@@ -631,16 +583,16 @@ export class ReadStateManager {
kind: KIND_READ_STATE,
content: ciphertext,
createdAt,
tags,
tags: [
["d", `read-state:${slotId}`],
["t", "read-state"],
],
});
await this.relayClient.publishEvent(
event,
"Timed out publishing read state.",
"Failed to publish read state.",
);
console.debug(
`[ReadStateManager] publish accepted slotId=${slotId} createdAt=${createdAt}`,
);
for (const key of Object.keys(contexts)) {
if (this.lastPublishedContexts[key] !== contexts[key])
this.contextSourceCreatedAt.set(key, createdAt);
@@ -656,15 +608,9 @@ export class ReadStateManager {
}
}
/**
* Multi-slot publish path. Partitions channel keys across slots and publishes
* each independently. Skips if nothing changed since last publish.
*/
private async publishSplitSlots(): Promise<void> {
const slots = this.splitContextsIntoSlots();
if (slots === null) return;
// No-op suppression: skip if nothing changed.
const unionContexts: Record<string, number> = {};
for (const { contexts } of slots) {
for (const [key, ts] of Object.entries(contexts)) {
@@ -673,25 +619,13 @@ export class ReadStateManager {
}
}
if (this.isIdenticalToLastPublished(unionContexts)) return;
// Reset before multi-slot publish to rebuild as union of all slots.
this.lastPublishedContexts = {};
for (const { slotId, contexts } of slots) {
for (const { slotId, contexts } of slots)
await this.publishOneSlot(slotId, contexts);
}
}
/**
* Publish NIP-09 kind:5 delete events for all extra slot blobs, then clear
* extraSlotIds. Gated on isLoadComplete.
*/
private async deleteExtraSlots(): Promise<void> {
if (!this.isLoadComplete) {
console.debug(
"[ReadStateManager] deleteExtraSlots blocked: isLoadComplete=false",
);
return;
}
if (!this.isLoadComplete) return;
for (const slotId of this.extraSlotIds) {
try {
const aTagValue = `${KIND_READ_STATE}:${this.pubkey}:${READ_STATE_D_TAG_PREFIX}${slotId}`;
@@ -705,22 +639,18 @@ export class ReadStateManager {
"Timed out deleting extra read-state slot.",
"Failed to delete extra read-state slot.",
);
console.debug(`[ReadStateManager] deleted extra slot slotId=${slotId}`);
} catch (error) {
console.debug(
`[ReadStateManager] deleteExtraSlots failed for slotId=${slotId}:`,
error,
);
// Non-fatal: stale blob will expire from relay within the horizon window.
} catch {
/* Non-fatal */
}
}
this.extraSlotIds = [];
saveExtraSlotIds(this.pubkey, []);
}
private async fetchOwnBlobBeforePublish(): Promise<void> {
const allSlotIds = [this.slotId, ...this.extraSlotIds];
const dTags = allSlotIds.map((id) => `${READ_STATE_D_TAG_PREFIX}${id}`);
private async fetchOwnBlobBeforePublish(): Promise<boolean> {
const dTags = [this.slotId, ...this.extraSlotIds].map(
(id) => `${READ_STATE_D_TAG_PREFIX}${id}`,
);
try {
const events = await this.relayClient.fetchEvents({
kinds: [KIND_READ_STATE],
@@ -728,13 +658,13 @@ export class ReadStateManager {
"#d": dTags,
limit: READ_STATE_FETCH_LIMIT,
});
await this.mergeEvents(events);
const deduped = deduplicateByCoordinate(events);
const merged = await mergeReadStateEventsStructured(deduped, this.pubkey);
await this.ingest(merged);
this.persistLocalState();
} catch (error) {
console.debug(
"[ReadStateManager] fetchOwnBlobBeforePublish failed:",
error,
);
return true;
} catch {
return false;
}
}
@@ -750,113 +680,8 @@ export class ReadStateManager {
return true;
}
private currentContexts(): Record<string, number> | null {
const contexts: Record<string, number> = {};
for (const [ctx, ts] of this.effectiveState) {
if (this.publishableContextIds.has(ctx)) contexts[ctx] = ts;
}
// Include ov_s/c/b wire entries for any context with an active or tombstoned register.
for (const [rawCtx, reg] of this.overrideRegisters) {
if (!this.publishableContextIds.has(rawCtx)) continue;
const effectiveFrontier = this.resolveOwnEffectiveFrontier(rawCtx);
const wireEntries = encodeOverrideGroup(rawCtx, reg, effectiveFrontier);
for (const [key, val] of Object.entries(wireEntries)) contexts[key] = val;
}
// Evict oldest msg: then thread: entries until blob fits 32 KB.
// Channel keys and ov_* entries are never evicted.
const { evicted, fitsAfterTrim } = trimContextsToBudget(
contexts,
this.clientId,
READ_STATE_MAX_PLAINTEXT_BYTES,
);
if (evicted > 0) {
console.warn(
`[ReadStateManager] currentContexts trimmed ${evicted} entries`,
);
}
if (!fitsAfterTrim) {
console.warn(
"[ReadStateManager] currentContexts: budget exceeded — will split",
);
return null;
}
return contexts;
}
/**
* Partition the full publishable contexts across multiple slots when channel
* keys alone exceed READ_STATE_MAX_PLAINTEXT_BYTES. Override-bearing context
* groups are pinned to slot 0. Returns null when even READ_STATE_MAX_SLOTS
* slots can't accommodate all channel keys.
*/
private splitContextsIntoSlots(): Array<{
slotId: string;
contexts: Record<string, number>;
}> | null {
// Separate channel keys from thread/msg entries. Override wire entries
// (ov_s:, ov_c:, ov_b:) go in channelEntries for pinning to slot 0.
const channelEntries: [string, number][] = [];
const threadMsgEntries: [string, number][] = [];
for (const [ctx, ts] of this.effectiveState) {
if (!this.publishableContextIds.has(ctx)) continue;
if (ctx.startsWith(MSG_PREFIX) || ctx.startsWith(THREAD_PREFIX)) {
threadMsgEntries.push([ctx, ts]);
} else {
channelEntries.push([ctx, ts]);
}
}
// Append override wire entries for publishable contexts.
for (const [rawCtx, reg] of this.overrideRegisters) {
if (!this.publishableContextIds.has(rawCtx)) continue;
const effectiveFrontier = this.resolveOwnEffectiveFrontier(rawCtx);
for (const [key, val] of Object.entries(
encodeOverrideGroup(rawCtx, reg, effectiveFrontier),
)) {
channelEntries.push([key, val]);
}
}
const allSlotIds = [this.slotId, ...this.extraSlotIds];
const result = splitContextsIntoBudgetedSlots({
channelEntries,
threadMsgEntries,
clientId: this.clientId,
initialSlotCount: allSlotIds.length,
maxSlots: READ_STATE_MAX_SLOTS,
maxBytes: READ_STATE_MAX_PLAINTEXT_BYTES,
slotIdGenerator: () => generateHex(16),
});
if (result === null) {
console.error(
`[ReadStateManager] splitContextsIntoSlots: ${channelEntries.length} channel keys exceed ${READ_STATE_MAX_SLOTS}-slot budget`,
);
return null;
}
// Persist any newly allocated extra slot IDs.
const newExtraSlotIds = [...allSlotIds.slice(1), ...result.extraSlotIds];
if (newExtraSlotIds.length !== this.extraSlotIds.length) {
this.extraSlotIds = newExtraSlotIds;
saveExtraSlotIds(this.pubkey, this.extraSlotIds);
}
const finalSlotIds = [...allSlotIds, ...result.extraSlotIds];
return finalSlotIds.map((slotId, i) => ({
slotId,
contexts: result.slots[i],
}));
}
/** Resolve the effective frontier for `rawCtxId` from own state only (no parentResolver). */
private resolveOwnEffectiveFrontier(rawCtxId: string): number {
return this.effectiveState.get(rawCtxId) ?? 0;
}
/** Resolve the effective frontier for `channelId` including the parent resolver. */
private resolveChannelFrontier(channelId: string): number {
/** Effective frontier including parent resolver. */
private channelFrontier(channelId: string): number {
return (
resolveEffectiveTimestamp({
effectiveState: this.effectiveState,
@@ -867,13 +692,87 @@ export class ReadStateManager {
}
/**
* Return the liveness status of the manual-unread override for `channelId`,
* or `null` when no override register exists.
* Build the single-slot contexts record. Uses `channelFrontier()` for
* canonical serialization — same resolver as `getOverrideLiveness()`.
* Returns null when the record doesn't fit in one slot.
*/
private currentContexts(): Record<string, number> | null {
const contexts: Record<string, number> = {};
for (const [ctx, ts] of this.effectiveState) {
if (this.publishableContextIds.has(ctx))
contexts[escapeFrontierKey(ctx)] = ts;
}
for (const [rawCtx, reg] of this.overrideRegisters) {
if (!this.publishableContextIds.has(rawCtx)) continue;
// Use channelFrontier — same resolver as getOverrideLiveness.
const frontier = this.channelFrontier(rawCtx);
const wireEntries = encodeOverrideGroup(rawCtx, reg, frontier);
for (const [key, val] of Object.entries(wireEntries)) contexts[key] = val;
}
const { evicted, fitsAfterTrim } = trimContextsToBudget(
contexts,
this.clientId,
READ_STATE_MAX_PLAINTEXT_BYTES,
);
if (evicted > 0)
console.warn(
`[ReadStateManager] currentContexts trimmed ${evicted} entries`,
);
if (!fitsAfterTrim) return null;
return contexts;
}
private splitContextsIntoSlots(): Array<{
slotId: string;
contexts: Record<string, number>;
}> | null {
const channelEntries: [string, number][] = [];
const threadMsgEntries: [string, number][] = [];
for (const [ctx, ts] of this.effectiveState) {
if (!this.publishableContextIds.has(ctx)) continue;
if (ctx.startsWith(MSG_PREFIX) || ctx.startsWith(THREAD_PREFIX)) {
threadMsgEntries.push([ctx, ts]);
} else {
channelEntries.push([escapeFrontierKey(ctx), ts]);
}
}
for (const [rawCtx, reg] of this.overrideRegisters) {
if (!this.publishableContextIds.has(rawCtx)) continue;
// Use channelFrontier — same resolver as getOverrideLiveness.
const frontier = this.channelFrontier(rawCtx);
for (const [key, val] of Object.entries(
encodeOverrideGroup(rawCtx, reg, frontier),
)) {
channelEntries.push([key, val]);
}
}
const allSlotIds = [this.slotId, ...this.extraSlotIds];
const result = splitContextsIntoBudgetedSlots({
channelEntries,
threadMsgEntries,
clientId: this.clientId,
initialSlotCount: allSlotIds.length,
maxSlots: READ_STATE_MAX_SLOTS,
maxBytes: READ_STATE_MAX_PLAINTEXT_BYTES,
slotIdGenerator: () => generateHex(16),
});
if (result === null) return null;
const newExtraSlotIds = [...allSlotIds.slice(1), ...result.extraSlotIds];
if (newExtraSlotIds.length !== this.extraSlotIds.length) {
this.extraSlotIds = newExtraSlotIds;
saveExtraSlotIds(this.pubkey, this.extraSlotIds);
}
const finalSlotIds = [...allSlotIds, ...result.extraSlotIds];
return finalSlotIds.map((slotId, i) => ({
slotId,
contexts: result.slots[i],
}));
}
getOverrideLiveness(channelId: string): OverrideLiveness | null {
const reg = this.overrideRegisters.get(channelId);
if (!reg) return null;
const effectiveFrontier = this.resolveChannelFrontier(channelId);
const effectiveFrontier = this.channelFrontier(channelId);
return {
active: isOverrideActive(reg, effectiveFrontier),
frontier: effectiveFrontier,
@@ -881,10 +780,31 @@ export class ReadStateManager {
}
/**
* Mark a channel as manually unread. Bumps S to `max(S,C)+1`, sets B to the
* effective frontier. Refuses with `load_incomplete`, `uint32_overflow`, or
* `budget_exhausted` when applicable.
* 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.
*/
private tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean {
const prev = this.overrideRegisters.get(rawCtxId);
const wasPublishable = this.publishableContextIds.has(rawCtxId);
this.overrideRegisters.set(rawCtxId, reg);
this.publishableContextIds.add(rawCtxId);
// Try single-slot first; fall back to split planner.
const single = this.currentContexts();
const fits = single !== null || this.splitContextsIntoSlots() !== null;
// Restore state.
if (prev === undefined) {
this.overrideRegisters.delete(rawCtxId);
} else {
this.overrideRegisters.set(rawCtxId, prev);
}
if (!wasPublishable) this.publishableContextIds.delete(rawCtxId);
return fits;
}
markChannelUnread(channelId: string): MarkResult {
if (!this.isLoadComplete)
return { success: false, reason: "load_incomplete" };
@@ -894,39 +814,36 @@ export class ReadStateManager {
const b = existing?.b ?? 0;
const newS = Math.max(s, c) + 1;
if (newS > 0xffffffff) return { success: false, reason: "uint32_overflow" };
const effectiveFrontier = this.resolveChannelFrontier(channelId);
const newReg: OverrideRegister = {
s: newS,
c,
b: Math.max(b, effectiveFrontier),
b: Math.max(b, this.channelFrontier(channelId)),
};
if (this.currentContextsWithOverride(channelId, newReg) === null) {
if (!this.tryCandidatePlan(channelId, newReg)) {
return { success: false, reason: "budget_exhausted" };
}
this.overrideRegisters.set(channelId, newReg);
this.publishableContextIds.add(channelId);
this.persistLocalState();
if (!this.persistLocalState())
return { success: false, reason: "storage_failed" };
this.notifyListeners();
this.schedulePublish();
return { success: true };
}
/**
* Mark a channel as read. Bumps C to `max(S,C)+1` (clear-wins).
* Refuses with `already_inactive` when no active override exists.
*/
markChannelRead(channelId: string): MarkResult {
if (!this.isLoadComplete)
return { success: false, reason: "load_incomplete" };
const reg = this.overrideRegisters.get(channelId);
const effectiveFrontier = this.resolveChannelFrontier(channelId);
if (!reg || !isOverrideActive(reg, effectiveFrontier)) {
return { success: false, reason: "already_inactive" };
}
const effectiveFrontier = this.channelFrontier(channelId);
// No register at all — nothing to clear.
if (!reg) return { success: false, reason: "already_inactive" };
// Register exists: always attempt C-bump (spec NIP-RS.md:537-539 — explicit
// read advances monotone frontier AND increments C; frontier-only deactivation
// is the success fallback when increment would overflow, not a skip condition).
const newC = Math.max(reg.s, reg.c) + 1;
if (newC > 0xffffffff) return { success: false, reason: "uint32_overflow" };
const newReg: OverrideRegister = { s: reg.s, c: newC, b: reg.b };
// Unreachable: clear-wins means newC > reg.s always. Defensive guard.
if (isOverrideActive(newReg, effectiveFrontier)) {
console.error(
"[ReadStateManager] markChannelRead: override still active after bump",
@@ -935,31 +852,13 @@ export class ReadStateManager {
}
this.overrideRegisters.set(channelId, newReg);
this.publishableContextIds.add(channelId);
this.persistLocalState();
if (!this.persistLocalState())
return { success: false, reason: "storage_failed" };
this.notifyListeners();
this.schedulePublish();
return { success: true };
}
/**
* Trial budget check: temporarily apply `reg` for `rawCtxId` and call
* `currentContexts()`. Returns null if budget is exhausted after trim.
*/
private currentContextsWithOverride(
rawCtxId: string,
reg: OverrideRegister,
): Record<string, number> | null {
const prev = this.overrideRegisters.get(rawCtxId);
this.overrideRegisters.set(rawCtxId, reg);
const result = this.currentContexts();
if (prev === undefined) {
this.overrideRegisters.delete(rawCtxId);
} else {
this.overrideRegisters.set(rawCtxId, prev);
}
return result;
}
private hydrateFromLocalStorage(): void {
const stored = readStoredReadState(this.pubkey);
for (const [contextId, timestamp] of stored.contexts)
@@ -968,15 +867,30 @@ 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,
);
}
this.persistLocalState();
}
private persistLocalState(): void {
writeStoredReadState(
/** Persist local state. Returns false if any write failed (mark-action must fail). */
private persistLocalState(): boolean {
return writeStoredReadState(
this.pubkey,
this.effectiveState,
this.publishableContextIds,
this.contextSourceCreatedAt,
this.overrideRegisters,
);
}
@@ -996,3 +910,11 @@ export class ReadStateManager {
}
}
}
/** Canonical key for an override register: encodes liveness + component values. */
function canonicalKey(reg: OverrideRegister, frontier: number): string {
const active = isOverrideActive(reg, frontier);
if (active) return `live:${reg.s},${reg.c},${reg.b}`;
const floor = Math.max(reg.s, reg.c);
return floor > 0 ? `dead:${floor}` : "virgin";
}

18
desktop/src/features/channels/readState/readStateStorage.test.mjs
View File

@@ -97,6 +97,7 @@ test("writeStoredReadState prunes all three keys consistently", () => {
[staleThread, stale],
[freshThread, nowSeconds],
]),
new Map(),
);
const state = JSON.parse(
@@ -128,15 +129,21 @@ 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 }]]),
);
const stored = readStoredReadState(pubkey);
assert.equal(stored.contexts.get("channel-9"), nowSeconds);
assert.equal(stored.publishableContextIds.has("channel-9"), true);
assert.equal(stored.contextSourceCreatedAt.get("channel-9"), nowSeconds);
const reg = stored.overrideRegisters.get("channel-9");
assert.ok(reg, "overrideRegisters must round-trip");
assert.equal(reg.s, 3, "register S must round-trip");
assert.equal(reg.c, 1, "register C must round-trip");
assert.equal(reg.b, nowSeconds, "register B must round-trip");
});
test("writeStoredReadState survives a throwing localStorage.setItem", () => {
test("writeStoredReadState_survives_throwing_localStorage_and_returns_false", () => {
const ls = installLocalStorage();
ls.setItem = () => {
throw new Error("QuotaExceededError");
@@ -144,12 +151,19 @@ test("writeStoredReadState survives a throwing localStorage.setItem", () => {
const pubkey = "d".repeat(64);
const nowSeconds = Math.floor(Date.now() / 1_000);
let result;
assert.doesNotThrow(() => {
writeStoredReadState(
result = writeStoredReadState(
pubkey,
new Map([["channel-1", nowSeconds]]),
new Set(["channel-1"]),
new Map([["channel-1", nowSeconds]]),
new Map(),
);
});
assert.equal(
result,
false,
"writeStoredReadState must return false on quota failure",
);
});

63
desktop/src/features/channels/readState/readStateStorage.ts
View File

@@ -8,6 +8,7 @@ import {
MSG_PREFIX,
READ_STATE_HORIZON_SECONDS,
THREAD_PREFIX,
type OverrideRegister,
} from "@/features/channels/readState/readStateFormat";
import { setLocalStorageItemWithRecovery } from "@/shared/lib/localStorageQuota";
@@ -15,6 +16,7 @@ export type StoredReadState = {
contexts: Map<string, number>;
publishableContextIds: Set<string>;
contextSourceCreatedAt: Map<string, number>;
overrideRegisters: Map<string, OverrideRegister>;
};
function mergeLocalStorageKey(
@@ -100,6 +102,7 @@ export function readStoredReadState(pubkey: string): StoredReadState {
contexts,
publishableContextIds: readPublishableContextIds(pubkey),
contextSourceCreatedAt: readContextSourceCreatedAt(pubkey),
overrideRegisters: readOverrideRegisters(pubkey),
};
}
@@ -109,6 +112,45 @@ function isPrunableContextKey(contextId: string): boolean {
);
}
// Key for persisted override registers (raw context ID → OverrideRegister triple).
function localOverrideRegistersKey(pubkey: string): string {
return `buzz.nip-rs.override-registers.v1:${pubkey}`;
}
function readOverrideRegisters(pubkey: string): Map<string, OverrideRegister> {
const result = new Map<string, OverrideRegister>();
const raw = localStorage.getItem(localOverrideRegistersKey(pubkey));
if (!raw) return result;
try {
const parsed = JSON.parse(raw);
if (!isPlainRecord(parsed)) return result;
for (const [rawCtx, value] of Object.entries(parsed)) {
if (!isPlainRecord(value)) continue;
const s = value.s;
const c = value.c;
const b = value.b;
if (
typeof s === "number" &&
Number.isInteger(s) &&
s >= 0 &&
s <= 0xffffffff &&
typeof c === "number" &&
Number.isInteger(c) &&
c >= 0 &&
c <= 0xffffffff &&
typeof b === "number" &&
Number.isInteger(b) &&
b >= 0
) {
result.set(rawCtx, { s, c, b });
}
}
} catch {
// Corrupt storage — return empty map; will be repopulated on next ingest.
}
return result;
}
/**
* Drops msg:/thread: markers older than the relay's 7-day horizon, then caps
* the survivors at LOCAL_MAX_PRUNABLE_CONTEXTS (oldest first). Channel keys
@@ -147,7 +189,8 @@ export function writeStoredReadState(
contexts: ReadonlyMap<string, number>,
publishableContextIds: ReadonlySet<string>,
contextSourceCreatedAt: ReadonlyMap<string, number>,
): void {
overrideRegisters: ReadonlyMap<string, OverrideRegister>,
): boolean {
const pruned = pruneStaleContexts(contexts, Math.floor(Date.now() / 1_000));
const state: Record<string, string> = {};
@@ -155,11 +198,11 @@ export function writeStoredReadState(
state[contextId] = new Date(timestamp * 1_000).toISOString();
}
setLocalStorageItemWithRecovery(
const ok1 = setLocalStorageItemWithRecovery(
localReadStateKey(pubkey),
JSON.stringify(state),
);
setLocalStorageItemWithRecovery(
const ok2 = setLocalStorageItemWithRecovery(
localPublishableContextKey(pubkey),
JSON.stringify([...publishableContextIds].filter((id) => pruned.has(id))),
);
@@ -170,8 +213,20 @@ export function writeStoredReadState(
sourceState[contextId] = createdAt;
}
}
setLocalStorageItemWithRecovery(
const ok3 = setLocalStorageItemWithRecovery(
localSourceCreatedAtKey(pubkey),
JSON.stringify(sourceState),
);
// Persist override registers atomically with frontier state.
const regState: Record<string, { s: number; c: number; b: number }> = {};
for (const [rawCtx, reg] of overrideRegisters) {
regState[rawCtx] = { s: reg.s, c: reg.c, b: reg.b };
}
const ok4 = setLocalStorageItemWithRecovery(
localOverrideRegistersKey(pubkey),
JSON.stringify(regState),
);
return ok1 && ok2 && ok3 && ok4;
}

6
desktop/src/shared/api/relayClientSession.ts
View File

@@ -448,17 +448,17 @@ export class RelayClient {
subscribeToReconnects(listener: () => void) {
this.reconnectListeners.add(listener);
return () => {
this.reconnectListeners.delete(listener);
};
}
getConnectionGeneration(): number {
return this.connectionGeneration;
}
/** Current connection state — synchronous read. */
getConnectionState(): ConnectionState {
return this.connectionStateEmitter.get();
}
/**
* Subscribe to connection-state transitions. The listener is invoked
* immediately with the current state so callers don't need a separate