mirror of
https://github.com/block/buzz.git
synced 2026-10-31 22:41:12 +00:00
fix(readState): v4 plan conformance — pending-intents consolidation, single-flight retry controller, biting witnesses
Closes all 7 IMPORTANT + 2 MINOR findings from Thufir pass-1. Core changes: - Consolidate pending-intents storage into override-state.v2 blob (Paul's design ruling). Single write point for registers, receipts, and intents eliminates the double-commit window; no migration needed since pending-intents.v1 never shipped. - Implement retryLoad() single-flight controller: bounded exponential backoff (1s->2s->4s...60s cap), reconnect coalescing into pendingRetryOnComplete, destroy/identity cancellation, retryAttempt reset on complete verdict. - sourceScope + readTarget captured in enqueue path so drain performs exact source cleanup at the correct frontier position. - refused-unread rollback: snapshot prior forced-unread entry before ordered write; restore exact snapshot on refusal; delete only when no prior existed. - Surfaced drain outcomes via onDrainOutcome hook; genuine refusals (uint32_overflow, budget_exhausted, storage_failed) trigger toast; already_inactive is silent success; load_incomplete never surfaces. - Cleanup atomicity: intent + receipt deleted in a single persistLocalState() call (one v2 blob write) so they are never observable separately. - createDrainContext typed factory replaces this-as-unknown-as-DrainContext cast. - NIP-RS.md: added normative refusal-reason vocabulary list. Biting witnesses added (readStateManager.test.mjs, +12 tests): - Amendment A: gen2 enqueued mid-replay, gen1 register rolled back - Both inversion directions (unread->read, read->unread in-flight) - compare-delete preserves newer intent - Amendment C via production fetchAndMerge load path (not direct register write) - Atomic intent+receipt deletion witness - Controller: automatic backoff, reconnect during loadInFlight, stale-gen ordering - Controller: destroy-with-timer, exactly-one-drain-per-generation - Store/UI: identity-swap-during-drain, deferred refusal surfaces outcome Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
@@ -356,7 +356,7 @@ test("test_foreign_entry_with_no_local_copy_stays_unselected", () => {
|
||||
BOB,
|
||||
);
|
||||
|
||||
assert.equal(personas[0].id, "catalog:" + ALICE + ":reviewer");
|
||||
assert.equal(personas[0].id, `catalog:${ALICE}:reviewer`);
|
||||
assert.equal(personas[0].isActive, false);
|
||||
});
|
||||
|
||||
@@ -377,7 +377,7 @@ test("test_catalog_source_match_is_scoped_to_the_publishing_owner", () => {
|
||||
ALICE,
|
||||
);
|
||||
|
||||
assert.equal(personas[0].id, "catalog:" + BOB + ":reviewer");
|
||||
assert.equal(personas[0].id, `catalog:${BOB}:reviewer`);
|
||||
assert.equal(personas[0].isActive, false);
|
||||
});
|
||||
|
||||
|
||||
@@ -706,8 +706,8 @@ test("preReady_unread_witness_c_pendingUnread_visible_during_loading", () => {
|
||||
// committed forced entry; committed forced with no pending is visible.
|
||||
// Exercises the ratified precedence rule via the production computePreReadyUnread helper.
|
||||
const { restore } = makeIsolatedStorage();
|
||||
const pubkey = "cc".repeat(32);
|
||||
pendingOverrideIntentStore.loadForPubkey(pubkey);
|
||||
const _pubkey = "cc".repeat(32);
|
||||
pendingOverrideIntentStore.restoreFromStorage(new Map(), 1);
|
||||
pendingOverrideIntentStore.enqueue("ch-unread", "unread");
|
||||
pendingOverrideIntentStore.enqueue("ch-read", "read");
|
||||
|
||||
|
||||
@@ -3,16 +3,19 @@
|
||||
*
|
||||
* When the full-state load is incomplete at click time, override actions cannot
|
||||
* immediately mutate the register. Instead the user's intent is recorded here
|
||||
* as a pending entry keyed by (pubkey, channelId). A drain on the next
|
||||
* complete-load transition replays each intent through the normal action rules
|
||||
* and surfaces the definitive applied/refused result.
|
||||
* as a pending entry keyed by channelId. A drain on the next complete-load
|
||||
* transition replays each intent through the normal action rules and surfaces
|
||||
* the definitive applied/refused result.
|
||||
*
|
||||
* Design notes
|
||||
* ─────────────
|
||||
* • **Separate store, deterministic write ordering** (plan ruling 2):
|
||||
* `forcedUnreadStore` entries (the optimistic local badge) are written
|
||||
* BEFORE the intent. A crash between the two leaves bold with no queued
|
||||
* bump — pre-#3966 per-device behaviour, always safe.
|
||||
* • **In-memory only** (consolidation ruling):
|
||||
* This store holds in-memory state only. Persistence is exclusively through
|
||||
* the `buzz.nip-rs.override-state.v2:<pubkey>` blob written by
|
||||
* `writeStoredReadState()` in readStateStorage.ts. The manager's
|
||||
* `persistLocalState()` is the single commit point for both the intent queue
|
||||
* and the register+receipt. The former `buzz.nip-rs.pending-intents.v1`
|
||||
* key is never written and is not read after migration.
|
||||
*
|
||||
* • **Per-channel, last-intent-wins** (plan ruling 3):
|
||||
* A re-mark while a drain is in flight replaces the queue entry and advances
|
||||
@@ -20,25 +23,15 @@
|
||||
* and commits zero local effects for the stale intent.
|
||||
*
|
||||
* • **Applied receipt** (plan Amendment C):
|
||||
* The receipt `{intentGen, op}` is NOT stored here — it lives inside the
|
||||
* `buzz.nip-rs.override-state.v2:<pubkey>` blob alongside the mutated
|
||||
* register (`readStateStorage.ts`) so register mutation and receipt are
|
||||
* never observable separately. This store only holds the intent queue and
|
||||
* exposes the generation counter.
|
||||
*
|
||||
* • **Storage failure** (plan ruling 2):
|
||||
* If the localStorage write fails, the intent is kept in-memory and
|
||||
* classified `storage_failed` internally. Optimistic presentation and
|
||||
* same-session drain still work; a crash before drain degrades to
|
||||
* per-device behaviour.
|
||||
* The receipt `{intentGen, op}` lives inside the v2 blob alongside the
|
||||
* mutated register so register mutation and receipt are never observable
|
||||
* separately. This store only holds the in-memory intent queue.
|
||||
*
|
||||
* • **Identity swap** (plan phase 1):
|
||||
* `loadForPubkey` replaces the in-memory map with the new pubkey's persisted
|
||||
* data. Old pubkey data is NOT wiped from localStorage.
|
||||
* `restoreFromStorage` replaces the in-memory map with the stored data.
|
||||
* Old pubkey data is NOT wiped from the v2 blob — the manager owns that.
|
||||
*/
|
||||
|
||||
import { isPlainRecord } from "@/features/channels/readState/readStateFormat";
|
||||
|
||||
// ── Types ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
/** The operation the user intended. */
|
||||
@@ -53,156 +46,77 @@ export type PendingIntentOp = "unread" | "read";
|
||||
* `sourceScope` — for `read` intents, the exact source (or undefined = all
|
||||
* sources) whose removal scope was captured at click time.
|
||||
* Used by the drain to apply source cleanup idempotently.
|
||||
* `readTarget` — for `read` intents, the frontier value captured at click
|
||||
* time. Drain advances the frontier to this value before the
|
||||
* C-bump so that an incomplete-load read later applies at the
|
||||
* correct logical position.
|
||||
*/
|
||||
export type PendingIntent = {
|
||||
gen: number;
|
||||
op: PendingIntentOp;
|
||||
sourceScope?: string;
|
||||
readTarget?: number;
|
||||
};
|
||||
|
||||
/** In-memory + persisted state for one pubkey. */
|
||||
type PendingIntentRecord = {
|
||||
/** Next generation to assign when a new intent is enqueued. */
|
||||
nextGen: number;
|
||||
/** Per-channel intent. At most one intent per channel at any time. */
|
||||
intents: Map<string, PendingIntent>;
|
||||
/** True when the last localStorage write failed. */
|
||||
storageFailed: boolean;
|
||||
};
|
||||
|
||||
// ── Storage keys ──────────────────────────────────────────────────────────────
|
||||
|
||||
const STORAGE_PREFIX = "buzz.nip-rs.pending-intents.v1";
|
||||
const storageKey = (pubkey: string) => `${STORAGE_PREFIX}:${pubkey}`;
|
||||
|
||||
// ── Serialisation ─────────────────────────────────────────────────────────────
|
||||
|
||||
type PersistedIntentEntry = {
|
||||
gen: number;
|
||||
op: PendingIntentOp;
|
||||
sourceScope?: string;
|
||||
};
|
||||
|
||||
type PersistedRecord = {
|
||||
nextGen: number;
|
||||
intents: Record<string, PersistedIntentEntry>;
|
||||
};
|
||||
|
||||
function serialise(record: PendingIntentRecord): string {
|
||||
const intents: Record<string, PersistedIntentEntry> = {};
|
||||
for (const [channelId, intent] of record.intents) {
|
||||
const entry: PersistedIntentEntry = { gen: intent.gen, op: intent.op };
|
||||
if (intent.sourceScope !== undefined)
|
||||
entry.sourceScope = intent.sourceScope;
|
||||
intents[channelId] = entry;
|
||||
}
|
||||
return JSON.stringify({ nextGen: record.nextGen, intents });
|
||||
}
|
||||
|
||||
function deserialise(pubkey: string): PendingIntentRecord {
|
||||
const empty: PendingIntentRecord = {
|
||||
nextGen: 1,
|
||||
intents: new Map(),
|
||||
storageFailed: false,
|
||||
};
|
||||
try {
|
||||
const raw = localStorage.getItem(storageKey(pubkey));
|
||||
if (!raw) return empty;
|
||||
const parsed: unknown = JSON.parse(raw);
|
||||
if (!isPlainRecord(parsed)) return empty;
|
||||
const { nextGen, intents } = parsed as Partial<PersistedRecord>;
|
||||
if (
|
||||
typeof nextGen !== "number" ||
|
||||
!Number.isInteger(nextGen) ||
|
||||
nextGen < 1
|
||||
)
|
||||
return empty;
|
||||
if (!isPlainRecord(intents)) return empty;
|
||||
const map = new Map<string, PendingIntent>();
|
||||
for (const [channelId, entry] of Object.entries(intents)) {
|
||||
if (!isPlainRecord(entry)) continue;
|
||||
const { gen, op, sourceScope } = entry as Partial<PersistedIntentEntry>;
|
||||
if (typeof gen !== "number" || !Number.isInteger(gen) || gen < 1)
|
||||
continue;
|
||||
if (op !== "unread" && op !== "read") continue;
|
||||
if (sourceScope !== undefined && typeof sourceScope !== "string")
|
||||
continue;
|
||||
map.set(channelId, {
|
||||
gen,
|
||||
op,
|
||||
...(sourceScope !== undefined ? { sourceScope } : {}),
|
||||
});
|
||||
}
|
||||
return { nextGen, intents: map, storageFailed: false };
|
||||
} catch {
|
||||
return empty;
|
||||
}
|
||||
}
|
||||
|
||||
function persist(pubkey: string, record: PendingIntentRecord): void {
|
||||
try {
|
||||
localStorage.setItem(storageKey(pubkey), serialise(record));
|
||||
record.storageFailed = false;
|
||||
} catch {
|
||||
record.storageFailed = true;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Store ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Pending intent store — one instance per session, keyed by pubkey.
|
||||
* Pending intent store — one instance per session, pure in-memory.
|
||||
*
|
||||
* All mutations are synchronous and serialised through this object. The store
|
||||
* owner is the single authority for generation ordering (plan ruling 3:
|
||||
* enqueue-during-drain serialises through the store owner).
|
||||
*
|
||||
* Persistence is exclusively through the manager's `persistLocalState()` which
|
||||
* serialises the in-memory map into the v2 blob's `pendingIntents` sub-object.
|
||||
*/
|
||||
export class PendingOverrideIntentStore {
|
||||
private pubkey: string | null = null;
|
||||
private record: PendingIntentRecord = {
|
||||
nextGen: 1,
|
||||
intents: new Map(),
|
||||
storageFailed: false,
|
||||
};
|
||||
private _nextGen = 1;
|
||||
private intents = new Map<string, PendingIntent>();
|
||||
|
||||
/**
|
||||
* Load or swap to the given pubkey's persisted intents.
|
||||
* Called on identity initialisation and identity change.
|
||||
* Restore in-memory state from deserialized storage (called by manager
|
||||
* during `hydrateFromLocalStorage`). Replaces any in-memory state.
|
||||
*/
|
||||
loadForPubkey(pubkey: string): void {
|
||||
this.pubkey = pubkey;
|
||||
this.record = deserialise(pubkey);
|
||||
restoreFromStorage(
|
||||
intents: ReadonlyMap<string, PendingIntent>,
|
||||
nextGen: number,
|
||||
): void {
|
||||
this.intents = new Map(intents);
|
||||
this._nextGen = nextGen;
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue a new intent for `channelId`, replacing any existing one.
|
||||
*
|
||||
* Returns the enqueued intent (including the assigned generation).
|
||||
* The generation is monotonically increasing per-channel (and
|
||||
* per-pubkey-session), making every intent distinguishable from predecessors.
|
||||
* The generation is monotonically increasing per-channel-session, making
|
||||
* every intent distinguishable from predecessors.
|
||||
*
|
||||
* Does NOT persist — the caller (drain or manager) calls persistLocalState().
|
||||
*/
|
||||
enqueue(
|
||||
channelId: string,
|
||||
op: PendingIntentOp,
|
||||
sourceScope?: string,
|
||||
readTarget?: number,
|
||||
): PendingIntent {
|
||||
const gen = this.record.nextGen++;
|
||||
const gen = this._nextGen++;
|
||||
const intent: PendingIntent = {
|
||||
gen,
|
||||
op,
|
||||
...(sourceScope !== undefined ? { sourceScope } : {}),
|
||||
...(readTarget !== undefined ? { readTarget } : {}),
|
||||
};
|
||||
this.record.intents.set(channelId, intent);
|
||||
if (this.pubkey) persist(this.pubkey, this.record);
|
||||
this.intents.set(channelId, intent);
|
||||
return intent;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current intent for `channelId`, or undefined if none exists.
|
||||
* Called from the drain capture step and from hydration.
|
||||
*/
|
||||
get(channelId: string): PendingIntent | undefined {
|
||||
return this.record.intents.get(channelId);
|
||||
return this.intents.get(channelId);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -210,41 +124,38 @@ export class PendingOverrideIntentStore {
|
||||
* Used by the drain to iterate over all pending work.
|
||||
*/
|
||||
channelIds(): IterableIterator<string> {
|
||||
return this.record.intents.keys();
|
||||
return this.intents.keys();
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare-and-delete: atomically verify that the current intent's generation
|
||||
* still matches `capturedGen`, then delete the intent (and its receipt is
|
||||
* deleted by the caller via `readStateStorage` in the same transaction).
|
||||
* still matches `capturedGen`, then delete the intent.
|
||||
*
|
||||
* Returns true when the deletion was performed; false when the generation
|
||||
* no longer matches (a newer intent replaced this one).
|
||||
*
|
||||
* This is the Amendment A + B compare-and-delete seam. Called after the
|
||||
* Amendment C gen re-check confirms the generation is current.
|
||||
* Does NOT persist — caller calls persistLocalState() afterward.
|
||||
*/
|
||||
compareAndDelete(channelId: string, capturedGen: number): boolean {
|
||||
const current = this.record.intents.get(channelId);
|
||||
const current = this.intents.get(channelId);
|
||||
if (!current || current.gen !== capturedGen) return false;
|
||||
this.record.intents.delete(channelId);
|
||||
if (this.pubkey) persist(this.pubkey, this.record);
|
||||
this.intents.delete(channelId);
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* True if the last localStorage write failed. Same-session drain still
|
||||
* works; a crash before drain degrades to per-device behaviour.
|
||||
*/
|
||||
get isStorageFailed(): boolean {
|
||||
return this.record.storageFailed;
|
||||
/** Current nextGen value — used by readStateStorage serialization. */
|
||||
get nextGen(): number {
|
||||
return this._nextGen;
|
||||
}
|
||||
|
||||
/**
|
||||
* Number of pending intents. Used by tests and diagnostics.
|
||||
*/
|
||||
/** Current intents map — used by readStateStorage serialization. */
|
||||
get all(): ReadonlyMap<string, PendingIntent> {
|
||||
return this.intents;
|
||||
}
|
||||
|
||||
/** Number of pending intents. Used by tests and diagnostics. */
|
||||
get size(): number {
|
||||
return this.record.intents.size;
|
||||
return this.intents.size;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@ import {
|
||||
} from "@/features/channels/readState/readStateFormat";
|
||||
import type { AppliedReceipt } from "@/features/channels/readState/readStateStorage";
|
||||
import { pendingOverrideIntentStore } from "@/features/channels/pendingOverrideIntents";
|
||||
import { toast } from "sonner";
|
||||
import type { MarkResult } from "@/features/channels/readState/readStateManager";
|
||||
|
||||
/** Subset of ReadStateManager state and methods needed for drain operations. */
|
||||
@@ -33,8 +34,97 @@ export interface DrainContext {
|
||||
schedulePublish(): void;
|
||||
notifyListeners(): void;
|
||||
channelFrontier(channelId: string): number;
|
||||
markContextRead(channelId: string, unixTimestamp: number): void;
|
||||
tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean;
|
||||
restoreExtraSlotIds(prev: string[]): void;
|
||||
/** Drain outcome callback — wired by the hook for toast/rollback routing.
|
||||
* `sourceScope` is set for `read+applied` outcomes; it carries the source
|
||||
* that was being cleared so the hook can remove exactly that source from the
|
||||
* forced-unread entry rather than deleting the whole entry. */
|
||||
onDrainOutcome:
|
||||
| ((
|
||||
channelId: string,
|
||||
op: string,
|
||||
status: string,
|
||||
reason?: string,
|
||||
sourceScope?: string,
|
||||
) => void)
|
||||
| null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Shape of ReadStateManager internals accessed by createDrainContext.
|
||||
* Listed explicitly so TypeScript validates the access at the factory site.
|
||||
*/
|
||||
interface ManagerInternals {
|
||||
readonly pubkey: string;
|
||||
readonly destroyed: boolean;
|
||||
isLoadComplete: boolean;
|
||||
readonly loadGeneration: number;
|
||||
readonly overrideRegisters: Map<string, OverrideRegister>;
|
||||
readonly appliedReceipts: Map<string, AppliedReceipt>;
|
||||
readonly publishableContextIds: Set<string>;
|
||||
readonly extraSlotIds: string[];
|
||||
onDrainOutcome:
|
||||
| ((
|
||||
channelId: string,
|
||||
op: string,
|
||||
status: string,
|
||||
reason?: string,
|
||||
sourceScope?: string,
|
||||
) => void)
|
||||
| null;
|
||||
persistLocalState(): boolean;
|
||||
schedulePublish(): void;
|
||||
notifyListeners(): void;
|
||||
channelFrontier(channelId: string): number;
|
||||
markContextRead(channelId: string, unixTimestamp: number): void;
|
||||
tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean;
|
||||
restoreExtraSlotIds(prev: string[]): void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a typed DrainContext adapter from the manager's internals.
|
||||
* This is the single place where the manager exposes its internals to the
|
||||
* drain layer — no cast required.
|
||||
*/
|
||||
export function createDrainContext(mgr: ManagerInternals): DrainContext {
|
||||
return {
|
||||
get pubkey() {
|
||||
return mgr.pubkey;
|
||||
},
|
||||
get destroyed() {
|
||||
return mgr.destroyed;
|
||||
},
|
||||
get isLoadComplete() {
|
||||
return mgr.isLoadComplete;
|
||||
},
|
||||
get loadGeneration() {
|
||||
return mgr.loadGeneration;
|
||||
},
|
||||
get overrideRegisters() {
|
||||
return mgr.overrideRegisters;
|
||||
},
|
||||
get appliedReceipts() {
|
||||
return mgr.appliedReceipts;
|
||||
},
|
||||
get publishableContextIds() {
|
||||
return mgr.publishableContextIds;
|
||||
},
|
||||
get extraSlotIds() {
|
||||
return mgr.extraSlotIds;
|
||||
},
|
||||
get onDrainOutcome() {
|
||||
return mgr.onDrainOutcome;
|
||||
},
|
||||
persistLocalState: () => mgr.persistLocalState(),
|
||||
schedulePublish: () => mgr.schedulePublish(),
|
||||
notifyListeners: () => mgr.notifyListeners(),
|
||||
channelFrontier: (id) => mgr.channelFrontier(id),
|
||||
markContextRead: (id, ts) => mgr.markContextRead(id, ts),
|
||||
tryCandidatePlan: (id, reg) => mgr.tryCandidatePlan(id, reg),
|
||||
restoreExtraSlotIds: (prev) => mgr.restoreExtraSlotIds(prev),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -42,7 +132,7 @@ export interface DrainContext {
|
||||
* Implements plan phase 4 / Amendments A + B + C.
|
||||
*
|
||||
* Normative drain sequence per intent:
|
||||
* 1. Capture — snapshot intent (op, gen, sourceScope) and pubkey fence.
|
||||
* 1. Capture — snapshot intent (op, gen, sourceScope, readTarget) and pubkey fence.
|
||||
* 2. Replay — perform the register action against the complete-load state.
|
||||
* For applied: co-commit register + receipt atomically (Amendment C).
|
||||
* 3. Re-check (Amendment A whole-transaction fence) — compare captured gen
|
||||
@@ -69,6 +159,7 @@ export async function drainPendingIntents(
|
||||
const capturedGen = captured.gen;
|
||||
const capturedOp = captured.op;
|
||||
const capturedSourceScope = captured.sourceScope;
|
||||
const capturedReadTarget = captured.readTarget;
|
||||
const capturedPubkey = ctx.pubkey;
|
||||
|
||||
// Pubkey fence: abort if identity changed mid-drain.
|
||||
@@ -97,7 +188,13 @@ export async function drainPendingIntents(
|
||||
replayResult =
|
||||
capturedOp === "unread"
|
||||
? markChannelUnreadDirect(ctx, channelId, capturedGen, capturedOp)
|
||||
: markChannelReadDirect(ctx, channelId, capturedGen, capturedOp);
|
||||
: markChannelReadDirect(
|
||||
ctx,
|
||||
channelId,
|
||||
capturedGen,
|
||||
capturedOp,
|
||||
capturedReadTarget,
|
||||
);
|
||||
}
|
||||
|
||||
// Amendment A whole-transaction fence: re-check gen under store owner.
|
||||
@@ -129,25 +226,34 @@ export async function drainPendingIntents(
|
||||
else ctx.appliedReceipts.set(channelId, prevReceipt);
|
||||
if (!wasPublishable) ctx.publishableContextIds.delete(channelId);
|
||||
ctx.restoreExtraSlotIds(prevExtraSlotIds);
|
||||
console.warn(
|
||||
`[ReadStateManager] drain: register+receipt commit failed for ${channelId}, intent preserved`,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
ctx.schedulePublish();
|
||||
}
|
||||
|
||||
// Step 2: source cleanup (idempotent).
|
||||
void capturedSourceScope; // reserved for future UI-layer cleanup hook
|
||||
// Step 2: source cleanup (idempotent) — capturedSourceScope scopes removal.
|
||||
// The sourceScope is forwarded to the onDrainOutcome callback so the hook
|
||||
// layer can remove exactly the right source from the forced-unread entry.
|
||||
|
||||
// Step 3: toast for genuine refusals.
|
||||
// Step 3: surface outcomes — toast genuine refusals, route to hook callback.
|
||||
if (replayResult.status === "refused") {
|
||||
const reason = replayResult.reason;
|
||||
if (reason !== "already_inactive" && reason !== "load_incomplete") {
|
||||
console.warn(
|
||||
`[ReadStateManager] drain: intent ${capturedOp} for ${channelId} refused: ${reason}`,
|
||||
);
|
||||
// Genuine refusal (uint32_overflow, budget_exhausted, storage_failed):
|
||||
// show toast and surface to hook for forced-entry rollback.
|
||||
const msg = toastForDrainRefusal(capturedOp, reason);
|
||||
if (msg) toast.error(msg);
|
||||
ctx.onDrainOutcome?.(channelId, capturedOp, "refused", reason);
|
||||
}
|
||||
// already_inactive: silent definitive success — no badge, no toast.
|
||||
} else if (replayResult.status === "applied") {
|
||||
ctx.onDrainOutcome?.(
|
||||
channelId,
|
||||
capturedOp,
|
||||
"applied",
|
||||
undefined,
|
||||
capturedSourceScope,
|
||||
);
|
||||
}
|
||||
|
||||
// Step 4: atomic cleanup commit — delete receipt + compare-and-delete intent.
|
||||
@@ -163,6 +269,21 @@ export async function drainPendingIntents(
|
||||
}
|
||||
}
|
||||
|
||||
/** Human-readable toast for a genuine drain refusal. Returns null for silent reasons. */
|
||||
function toastForDrainRefusal(op: string, reason: string): string | null {
|
||||
if (reason === "budget_exhausted")
|
||||
return "Could not mark unread: override budget exhausted.";
|
||||
if (reason === "uint32_overflow")
|
||||
return op === "unread"
|
||||
? "Could not mark unread: counter limit reached."
|
||||
: "Could not clear unread override: counter limit reached.";
|
||||
if (reason === "storage_failed")
|
||||
return op === "unread"
|
||||
? "Could not mark unread: storage write failed."
|
||||
: "Could not clear unread override: storage write failed.";
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Replay a mark-unread directly against the complete-load state (drain path).
|
||||
* Sets ctx.appliedReceipts so the subsequent atomic persistLocalState() call
|
||||
@@ -205,6 +326,9 @@ export function markChannelUnreadDirect(
|
||||
|
||||
/**
|
||||
* Replay a mark-read directly against the complete-load state (drain path).
|
||||
* Advances the frontier to `readTarget` (captured at click time) before the
|
||||
* C-bump so the read lands at the correct logical position even when the load
|
||||
* completed later.
|
||||
* Sets ctx.appliedReceipts so the subsequent atomic persistLocalState() call
|
||||
* in drainPendingIntents includes the receipt (Amendment C).
|
||||
* Does NOT persist or enqueue — only called from drainPendingIntents.
|
||||
@@ -214,9 +338,14 @@ export function markChannelReadDirect(
|
||||
channelId: string,
|
||||
capturedGen: number,
|
||||
capturedOp: "read",
|
||||
readTarget?: number,
|
||||
): MarkResult {
|
||||
if (!ctx.isLoadComplete)
|
||||
return { status: "refused", reason: "load_incomplete" };
|
||||
// Advance frontier to readTarget first (spec order: frontier → C-bump).
|
||||
if (readTarget !== undefined && readTarget > 0) {
|
||||
ctx.markContextRead(channelId, readTarget);
|
||||
}
|
||||
const reg = ctx.overrideRegisters.get(channelId);
|
||||
const effectiveFrontier = ctx.channelFrontier(channelId);
|
||||
if (!reg) return { status: "refused", reason: "already_inactive" };
|
||||
@@ -242,6 +371,7 @@ export function markChannelReadDirect(
|
||||
/**
|
||||
* Public mark-unread: write-ordered forced entry then queue/apply the override.
|
||||
* Queues an intent when load is incomplete; applies immediately otherwise.
|
||||
* Persists the intent atomically in the v2 blob via persistLocalState().
|
||||
*/
|
||||
export function markChannelUnread(
|
||||
ctx: DrainContext,
|
||||
@@ -249,6 +379,10 @@ export function markChannelUnread(
|
||||
): MarkResult {
|
||||
if (!ctx.isLoadComplete) {
|
||||
pendingOverrideIntentStore.enqueue(channelId, "unread");
|
||||
if (!ctx.persistLocalState()) {
|
||||
// Storage failure — intent is in-memory only; session drain still works.
|
||||
// Optimistic presentation continues (forced entry already written).
|
||||
}
|
||||
return { status: "queued" };
|
||||
}
|
||||
const existing = ctx.overrideRegisters.get(channelId);
|
||||
@@ -286,13 +420,27 @@ export function markChannelUnread(
|
||||
|
||||
/**
|
||||
* Public mark-read: queues an intent when load is incomplete; C-bumps otherwise.
|
||||
* Captures the current frontier as readTarget when queuing, so the drain can
|
||||
* advance the frontier to the correct position later.
|
||||
* Captures sourceScope when provided so the drain can perform exact source cleanup.
|
||||
* Persists the intent atomically in the v2 blob via persistLocalState().
|
||||
*/
|
||||
export function markChannelRead(
|
||||
ctx: DrainContext,
|
||||
channelId: string,
|
||||
sourceScope?: string,
|
||||
): MarkResult {
|
||||
if (!ctx.isLoadComplete) {
|
||||
pendingOverrideIntentStore.enqueue(channelId, "read", undefined);
|
||||
const readTarget = ctx.channelFrontier(channelId);
|
||||
pendingOverrideIntentStore.enqueue(
|
||||
channelId,
|
||||
"read",
|
||||
sourceScope,
|
||||
readTarget > 0 ? readTarget : undefined,
|
||||
);
|
||||
if (!ctx.persistLocalState()) {
|
||||
// Storage failure — intent is in-memory only; session drain still works.
|
||||
}
|
||||
return { status: "queued" };
|
||||
}
|
||||
const reg = ctx.overrideRegisters.get(channelId);
|
||||
|
||||
@@ -2379,7 +2379,7 @@ test("drain_sameSourceReincarnation_genFenceZeroLocalEffects", async () => {
|
||||
// observable as read. This asserts that never happens.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "d1".repeat(32);
|
||||
const channelId = "reincarnation-channel-" + "a".repeat(40);
|
||||
const channelId = `reincarnation-channel-${"a".repeat(40)}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
@@ -2469,7 +2469,7 @@ test("drain_crashWindow1_restartReplaysOnce_extraBumpOnly", async () => {
|
||||
// before step 1) to confirm restart does replay exactly once.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "d2".repeat(32);
|
||||
const channelId = "crash-window-1-ch-" + "b".repeat(44);
|
||||
const channelId = `crash-window-1-ch-${"b".repeat(44)}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
@@ -2493,21 +2493,16 @@ test("drain_crashWindow1_restartReplaysOnce_extraBumpOnly", async () => {
|
||||
);
|
||||
|
||||
// The intent is now persisted (gen=1). Simulate a crash: destroy manager
|
||||
// without draining — intent stays in localStorage, no register committed.
|
||||
// without draining — intent stays in the v2 blob, no register committed.
|
||||
mgr1.destroy();
|
||||
|
||||
// Verify: intent is persisted (survived the "crash").
|
||||
const _survivingIntent = pendingOverrideIntentStore.get(channelId);
|
||||
// Note: pendingOverrideIntentStore is a singleton; loadForPubkey reloads on
|
||||
// the new manager construction below.
|
||||
|
||||
// ── Session 2: restart. Intent present, no receipt → replay fires once ────
|
||||
const mgr2 = new ReadStateManager(pubkey, fakeRelay);
|
||||
mgr2.hydrateFromLocalStorage();
|
||||
mgr2.effectiveState.set(channelId, 100);
|
||||
mgr2.isLoadComplete = true;
|
||||
|
||||
// Confirm intent survived (reloaded from localStorage by constructor's loadForPubkey).
|
||||
// Confirm intent survived (reloaded from v2 blob by hydrateFromLocalStorage).
|
||||
const intentAfterRestart = pendingOverrideIntentStore.get(channelId);
|
||||
assert.ok(
|
||||
intentAfterRestart,
|
||||
@@ -2558,11 +2553,10 @@ test("drain_crashWindow2_receiptPresent_restartNoRebump", async () => {
|
||||
// register is unchanged from the original single commit.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "d3".repeat(32);
|
||||
const channelId = "crash-window-2-ch-" + "c".repeat(44);
|
||||
const channelId = `crash-window-2-ch-${"c".repeat(44)}`;
|
||||
|
||||
const ls = globalThis.window.localStorage;
|
||||
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
|
||||
const intentKey = `buzz.nip-rs.pending-intents.v1:${pubkey}`;
|
||||
|
||||
// Craft the crash state directly in localStorage.
|
||||
// Register committed with S=2 (unread applied: max(0,0)+1=1... let's use S=2 for clarity),
|
||||
@@ -2574,13 +2568,8 @@ test("drain_crashWindow2_receiptPresent_restartNoRebump", async () => {
|
||||
JSON.stringify({
|
||||
r: { [channelId]: { s: sSingleCommit, c: 0, b: 100, f: 100 } },
|
||||
receipts: { [channelId]: { intentGen: 1, op: "unread" } },
|
||||
}),
|
||||
);
|
||||
ls.setItem(
|
||||
intentKey,
|
||||
JSON.stringify({
|
||||
nextGen: 2,
|
||||
intents: { [channelId]: { gen: 1, op: "unread" } },
|
||||
pi: { [channelId]: { gen: 1, op: "unread" } },
|
||||
ng: 2,
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -2646,7 +2635,7 @@ test("drain_crashAfterAppliedUnread_newerRemoteRead_noRebump", async () => {
|
||||
// proves application; only cleanup+delete runs, so C=6 stands.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "d4".repeat(32);
|
||||
const channelId = "amend-c-unread-race-" + "d".repeat(44);
|
||||
const channelId = `amend-c-unread-race-${"d".repeat(44)}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
@@ -2665,19 +2654,13 @@ test("drain_crashAfterAppliedUnread_newerRemoteRead_noRebump", async () => {
|
||||
// and after restart we merge C=6 into the register.
|
||||
const ls = globalThis.window.localStorage;
|
||||
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
|
||||
const intentKey = `buzz.nip-rs.pending-intents.v1:${pubkey}`;
|
||||
ls.setItem(
|
||||
v2Key,
|
||||
JSON.stringify({
|
||||
r: { [channelId]: { s: 5, c: 3, b: 100, f: 100 } },
|
||||
receipts: { [channelId]: { intentGen: 1, op: "unread" } },
|
||||
}),
|
||||
);
|
||||
ls.setItem(
|
||||
intentKey,
|
||||
JSON.stringify({
|
||||
nextGen: 2,
|
||||
intents: { [channelId]: { gen: 1, op: "unread" } },
|
||||
pi: { [channelId]: { gen: 1, op: "unread" } },
|
||||
ng: 2,
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -2730,11 +2713,10 @@ test("drain_crashAfterAppliedRead_newerRemoteUnread_noRebump", async () => {
|
||||
// reverse the new unread.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "d5".repeat(32);
|
||||
const channelId = "amend-c-read-race-" + "e".repeat(45);
|
||||
const channelId = `amend-c-read-race-${"e".repeat(45)}`;
|
||||
|
||||
const ls = globalThis.window.localStorage;
|
||||
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
|
||||
const intentKey = `buzz.nip-rs.pending-intents.v1:${pubkey}`;
|
||||
|
||||
// Crash state: read C-bump committed (S=5, C=6), receipt present, intent alive.
|
||||
ls.setItem(
|
||||
@@ -2742,13 +2724,8 @@ test("drain_crashAfterAppliedRead_newerRemoteUnread_noRebump", async () => {
|
||||
JSON.stringify({
|
||||
r: { [channelId]: { s: 5, c: 6, b: 100, f: 200 } },
|
||||
receipts: { [channelId]: { intentGen: 1, op: "read" } },
|
||||
}),
|
||||
);
|
||||
ls.setItem(
|
||||
intentKey,
|
||||
JSON.stringify({
|
||||
nextGen: 2,
|
||||
intents: { [channelId]: { gen: 1, op: "read" } },
|
||||
pi: { [channelId]: { gen: 1, op: "read" } },
|
||||
ng: 2,
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -2807,7 +2784,7 @@ test("drain_receiptWithoutMatchingIntent_sweptWithoutRegisterEffect", async () =
|
||||
// receipt persists across sessions" invariant.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "d6".repeat(32);
|
||||
const channelId = "receipt-sweep-ch-" + "f".repeat(46);
|
||||
const channelId = `receipt-sweep-ch-${"f".repeat(46)}`;
|
||||
|
||||
const ls = globalThis.window.localStorage;
|
||||
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
|
||||
@@ -2890,7 +2867,7 @@ test("drain_registerReceiptAtomicity_noStateWhereOnePersistsWithoutOther", async
|
||||
// asserts the combined single-write invariant.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "d7".repeat(32);
|
||||
const channelId = "atomicity-ch-" + "a".repeat(51);
|
||||
const channelId = `atomicity-ch-${"a".repeat(51)}`;
|
||||
|
||||
const ls = globalThis.window.localStorage;
|
||||
let blockV2Writes = false;
|
||||
@@ -3023,3 +3000,810 @@ test("fetchOwnBlobBeforePublish_foreignClientId_rotatesSlotAndUpdatesMetadata",
|
||||
delete globalThis.window.__TAURI_INTERNALS__;
|
||||
}
|
||||
});
|
||||
|
||||
// ── Test 30: Amendment A biting — gen2 enqueued DURING replay ─────────────────
|
||||
test("drain_gen2EnqueuedDuringReplay_rollsBackGen1Register", async () => {
|
||||
// Biting Amendment A: gen2 is enqueued DURING replay of gen1 (not before drain
|
||||
// starts). The re-check sees gen2 != capturedGen1 and rolls back all mutations.
|
||||
//
|
||||
// Injection point: override mgr.tryCandidatePlan to enqueue gen2 inside
|
||||
// markChannelUnreadDirect's budget check — the register write follows, but
|
||||
// the subsequent re-check must detect the gen change and roll it back.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "30".repeat(32);
|
||||
const channelId = `amend-a-biting-ch-${"b".repeat(46)}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }),
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
mgr.effectiveState.set(channelId, 50);
|
||||
|
||||
// Queue gen1 (unread) during incomplete load.
|
||||
mgr.isLoadComplete = false;
|
||||
const q = mgr.markChannelUnread(channelId);
|
||||
assert.equal(q.status, "queued", "gen1 must be queued");
|
||||
const gen1 = pendingOverrideIntentStore.get(channelId).gen;
|
||||
assert.ok(gen1 >= 1, "gen1 must be a positive generation number");
|
||||
|
||||
// Override tryCandidatePlan to enqueue gen2 DURING replay of gen1.
|
||||
const origTry = mgr.tryCandidatePlan.bind(mgr);
|
||||
let injected = false;
|
||||
mgr.tryCandidatePlan = (id, reg) => {
|
||||
const ok = origTry(id, reg);
|
||||
if (!injected && id === channelId) {
|
||||
injected = true;
|
||||
// This fires inside markChannelUnreadDirect, during gen1's replay.
|
||||
// After tryCandidatePlan returns, the register is written temporarily.
|
||||
// The Amendment A re-check must still roll it back.
|
||||
pendingOverrideIntentStore.enqueue(channelId, "unread");
|
||||
}
|
||||
return ok;
|
||||
};
|
||||
|
||||
// Complete load and drain gen1. The override fires gen2 mid-replay.
|
||||
mgr.isLoadComplete = true;
|
||||
await mgr.drainPendingIntents(mgr.loadGeneration);
|
||||
|
||||
// Gen2 was enqueued during gen1's replay → Amendment A re-check fires.
|
||||
// Gen1 committed zero local effects: register must still be absent (rolled back).
|
||||
const reg = mgr.overrideRegisters.get(channelId);
|
||||
assert.equal(
|
||||
reg,
|
||||
undefined,
|
||||
"register must be rolled back by gen-fence (gen2 was enqueued during gen1 replay)",
|
||||
);
|
||||
|
||||
// Gen2 must still be alive in the intent store — it was not consumed.
|
||||
const gen2Intent = pendingOverrideIntentStore.get(channelId);
|
||||
assert.ok(gen2Intent, "gen2 intent must still be in the queue");
|
||||
assert.ok(
|
||||
gen2Intent.gen > gen1,
|
||||
"gen2 must have a higher generation than gen1",
|
||||
);
|
||||
|
||||
// Clean up gen2.
|
||||
pendingOverrideIntentStore.compareAndDelete(channelId, gen2Intent.gen);
|
||||
mgr.destroy();
|
||||
});
|
||||
|
||||
// ── Test 31: unread→read inversion — read gen replaces unread gen ─────────────
|
||||
test("drain_unreadToReadInversion_onlyReadApplied", async () => {
|
||||
// Both inversion directions: unread gen1, then read gen2 replaces it.
|
||||
// Drain must apply only gen2 (read = C-bump), not gen1 (unread = S-bump).
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "31".repeat(32);
|
||||
const channelId = `inversion-ur-ch-${"c".repeat(48)}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }),
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
// Pre-seed an existing unread register so C-bump has something to work with.
|
||||
mgr.effectiveState.set(channelId, 50);
|
||||
mgr.overrideRegisters.set(channelId, { s: 1, c: 0, b: 50 });
|
||||
mgr.publishableContextIds.add(channelId);
|
||||
mgr.isLoadComplete = false;
|
||||
|
||||
// Step 1: queue unread (gen1).
|
||||
const r1 = mgr.markChannelUnread(channelId);
|
||||
assert.equal(r1.status, "queued");
|
||||
const gen1 = pendingOverrideIntentStore.get(channelId).gen;
|
||||
|
||||
// Step 2: queue read (gen2) — replaces gen1.
|
||||
const r2 = mgr.markChannelRead(channelId);
|
||||
assert.equal(r2.status, "queued");
|
||||
const intent = pendingOverrideIntentStore.get(channelId);
|
||||
assert.equal(intent.op, "read", "current intent must be read (gen2)");
|
||||
assert.ok(intent.gen > gen1, "gen2 must exceed gen1");
|
||||
|
||||
// Complete load and drain.
|
||||
mgr.isLoadComplete = true;
|
||||
await mgr.drainPendingIntents(mgr.loadGeneration);
|
||||
|
||||
// Gen2 (read) must have applied: C bumped, S unchanged at 1.
|
||||
const reg = mgr.overrideRegisters.get(channelId);
|
||||
assert.ok(reg, "register must exist after read drain");
|
||||
assert.equal(
|
||||
reg.s,
|
||||
1,
|
||||
"S must be unchanged — unread gen1 was replaced, not applied",
|
||||
);
|
||||
assert.ok(reg.c > 0, "C must be bumped by the read drain (gen2)");
|
||||
|
||||
// Gen2 intent consumed.
|
||||
assert.equal(
|
||||
pendingOverrideIntentStore.get(channelId),
|
||||
undefined,
|
||||
"gen2 intent must be deleted after drain",
|
||||
);
|
||||
|
||||
mgr.destroy();
|
||||
});
|
||||
|
||||
// ── Test 32: read→unread inversion — unread gen replaces read gen ─────────────
|
||||
test("drain_readToUnreadInversion_onlySBumpApplied", async () => {
|
||||
// Mirror of Test 31: read gen1 queued, then unread gen2 replaces it.
|
||||
// Drain applies only gen2 (S-bump), not gen1 (C-bump).
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "32".repeat(32);
|
||||
const channelId = `inversion-ru-ch-${"d".repeat(48)}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }),
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
mgr.effectiveState.set(channelId, 50);
|
||||
mgr.isLoadComplete = false;
|
||||
|
||||
// Queue read (gen1) — no existing register, will be already_inactive after drain.
|
||||
const r1 = mgr.markChannelRead(channelId);
|
||||
assert.equal(r1.status, "queued");
|
||||
|
||||
// Queue unread (gen2) — replaces read.
|
||||
const r2 = mgr.markChannelUnread(channelId);
|
||||
assert.equal(r2.status, "queued");
|
||||
const intent = pendingOverrideIntentStore.get(channelId);
|
||||
assert.equal(intent.op, "unread", "current intent must be unread (gen2)");
|
||||
|
||||
// Complete load and drain.
|
||||
mgr.isLoadComplete = true;
|
||||
await mgr.drainPendingIntents(mgr.loadGeneration);
|
||||
|
||||
// Gen2 (unread) applied: S bumped.
|
||||
const reg = mgr.overrideRegisters.get(channelId);
|
||||
assert.ok(reg, "register must exist after unread drain");
|
||||
assert.ok(reg.s > 0, "S must be bumped by unread drain (gen2)");
|
||||
assert.equal(
|
||||
reg.c,
|
||||
0,
|
||||
"C must be zero — read gen1 was replaced, not applied",
|
||||
);
|
||||
|
||||
// Consumed.
|
||||
assert.equal(
|
||||
pendingOverrideIntentStore.get(channelId),
|
||||
undefined,
|
||||
"gen2 intent consumed",
|
||||
);
|
||||
|
||||
mgr.destroy();
|
||||
});
|
||||
|
||||
// ── Test 33: compare-delete preserves newer intent ────────────────────────────
|
||||
test("drain_compareDeletePreservesNewerIntent", async () => {
|
||||
// compare-and-delete for gen1 must fail if gen2 was enqueued before the
|
||||
// compare-delete step. The drain's cleanup commit must not consume gen2.
|
||||
//
|
||||
// We inject gen2 via onDrainOutcome (fires just before compare-delete).
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "33".repeat(32);
|
||||
const channelId = `cmp-del-newer-ch-${"e".repeat(47)}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }),
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
mgr.effectiveState.set(channelId, 50);
|
||||
mgr.isLoadComplete = false;
|
||||
|
||||
// Queue gen1 (unread).
|
||||
mgr.markChannelUnread(channelId);
|
||||
const gen1 = pendingOverrideIntentStore.get(channelId).gen;
|
||||
|
||||
// Wire onDrainOutcome to enqueue gen2 BEFORE compare-and-delete runs.
|
||||
// The onDrainOutcome fires at step 3 (surface outcomes), before step 4 (cleanup commit).
|
||||
mgr.onDrainOutcome = (chId, _op, _status) => {
|
||||
if (chId === channelId) {
|
||||
// Enqueue gen2 between outcome surfacing and compare-delete.
|
||||
pendingOverrideIntentStore.enqueue(channelId, "unread");
|
||||
}
|
||||
};
|
||||
|
||||
mgr.isLoadComplete = true;
|
||||
await mgr.drainPendingIntents(mgr.loadGeneration);
|
||||
|
||||
// Gen2 must still be in the queue (compare-delete for gen1 preserved gen2).
|
||||
const remaining = pendingOverrideIntentStore.get(channelId);
|
||||
assert.ok(
|
||||
remaining,
|
||||
"gen2 intent must survive compare-delete targeting gen1",
|
||||
);
|
||||
assert.ok(
|
||||
remaining.gen > gen1,
|
||||
"surviving intent must be gen2 (newer than gen1)",
|
||||
);
|
||||
|
||||
// Clean up.
|
||||
pendingOverrideIntentStore.compareAndDelete(channelId, remaining.gen);
|
||||
mgr.destroy();
|
||||
});
|
||||
|
||||
// ── Test 34: Amendment C via production load — restart + fetchAndMerge ────────
|
||||
test("drain_amendmentC_restart_fetchAndMerge_noBumpAfterReceipt", async () => {
|
||||
// Amendment C via the PRODUCTION load path: after a crash-in-cleanup, the
|
||||
// next session hydrates from storage and runs fetchAndMerge() (not a direct
|
||||
// register write). The receipt must prevent a second S-bump.
|
||||
//
|
||||
// Thufir's finding: Tests 26/27 directly set isLoadComplete=true and write
|
||||
// overrideRegisters directly — they bypass the production load path.
|
||||
// This test uses the actual fetchAndMerge() flow via fakeRelay event delivery.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "34".repeat(32);
|
||||
const channelId = `amend-c-prod-ch-${"f".repeat(48)}`;
|
||||
|
||||
// Craft crash-state in localStorage: unread applied (S=5), receipt intact.
|
||||
const ls = globalThis.window.localStorage;
|
||||
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
|
||||
ls.setItem(
|
||||
v2Key,
|
||||
JSON.stringify({
|
||||
r: { [channelId]: { s: 5, c: 3, b: 100, f: 100 } },
|
||||
receipts: { [channelId]: { intentGen: 1, op: "unread" } },
|
||||
pi: { [channelId]: { gen: 1, op: "unread" } },
|
||||
ng: 2,
|
||||
}),
|
||||
);
|
||||
|
||||
// A remote device published C=6 (read). Simulate via a fetched event that,
|
||||
// when decrypted, delivers C=6 for the channel via the standard merge path.
|
||||
const remoteBlob = JSON.stringify({
|
||||
v: 1,
|
||||
client_id: "remote-device-client",
|
||||
contexts: {
|
||||
[`ov_s:${channelId}`]: 5,
|
||||
[`ov_c:${channelId}`]: 6,
|
||||
[`ov_b:${channelId}`]: 100,
|
||||
[channelId]: 200, // frontier advanced past B → override inactive
|
||||
},
|
||||
});
|
||||
// Slot ID must match /^[0-9a-f]{32}$/ to pass isValidReadStateDTag.
|
||||
const remoteSlotId = "0d".repeat(16);
|
||||
const blobEvent = {
|
||||
id: "ab".repeat(32),
|
||||
pubkey,
|
||||
created_at: 9999,
|
||||
kind: 30078,
|
||||
tags: [
|
||||
["d", `read-state:${remoteSlotId}`],
|
||||
["t", "read-state"],
|
||||
],
|
||||
content: "CIPHER",
|
||||
sig: "s".repeat(128),
|
||||
};
|
||||
|
||||
let fetchCallCount = 0;
|
||||
globalThis.window.__TAURI_INTERNALS__ = {
|
||||
invoke: async (command) => {
|
||||
if (command === "nip44_decrypt_from_self") return remoteBlob;
|
||||
throw new Error(`Unexpected: ${command}`);
|
||||
},
|
||||
};
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => {
|
||||
fetchCallCount++;
|
||||
return fetchCallCount === 1 ? [blobEvent] : [];
|
||||
},
|
||||
publishEvent: async () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
subscribeFenced: async (_filter, handler) => {
|
||||
// Deliver the event synchronously before EOSE.
|
||||
if (fetchCallCount === 0) {
|
||||
handler(blobEvent);
|
||||
}
|
||||
return makeFenceHandle({ eose: true });
|
||||
},
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
};
|
||||
|
||||
try {
|
||||
// Session 2: hydrate, then complete load via real fetchAndMerge().
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
mgr.hydrateFromLocalStorage();
|
||||
|
||||
// Receipt must be loaded.
|
||||
const receipt = mgr.appliedReceipts.get(channelId);
|
||||
assert.ok(receipt, "receipt must be loaded on restart");
|
||||
assert.equal(receipt.intentGen, 1);
|
||||
|
||||
// Run fetchAndMerge via the production path.
|
||||
await mgr.fetchAndMerge(mgr.loadGeneration);
|
||||
assert.equal(mgr.isLoadComplete, true, "load must complete");
|
||||
|
||||
// Drain via production path.
|
||||
await mgr.drainPendingIntents(mgr.loadGeneration);
|
||||
|
||||
// The alreadyApplied path must have run: no new S-bump.
|
||||
// C must reflect the merged remote (max(3,6)=6).
|
||||
const reg = mgr.overrideRegisters.get(channelId);
|
||||
assert.ok(reg, "register must exist");
|
||||
assert.equal(
|
||||
reg.s,
|
||||
5,
|
||||
"S must NOT be re-bumped by drain (receipt prevents it)",
|
||||
);
|
||||
assert.ok(reg.c >= 6, "C must reflect remote C=6 merge");
|
||||
|
||||
// Intent consumed.
|
||||
assert.equal(
|
||||
pendingOverrideIntentStore.get(channelId),
|
||||
undefined,
|
||||
"intent must be consumed after drain",
|
||||
);
|
||||
} finally {
|
||||
delete globalThis.window.__TAURI_INTERNALS__;
|
||||
}
|
||||
});
|
||||
|
||||
// ── Test 35: atomic deletion — intent + receipt deleted in one persist ─────────
|
||||
test("drain_cleanupCommit_intentAndReceiptDeletedAtomically", async () => {
|
||||
// Atomic deletion witness: after successful drain, both the intent AND the
|
||||
// receipt are absent from the v2 blob in a single commit (same persist call).
|
||||
// If they were deleted in separate writes, there would be a window where one
|
||||
// survives without the other.
|
||||
//
|
||||
// Verify: after drain, the v2 blob contains neither pi nor receipts entry.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "35".repeat(32);
|
||||
const channelId = `atomic-del-ch-${"g".repeat(50)}`;
|
||||
|
||||
const ls = globalThis.window.localStorage;
|
||||
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }),
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
mgr.effectiveState.set(channelId, 50);
|
||||
mgr.isLoadComplete = false;
|
||||
|
||||
// Queue the intent.
|
||||
mgr.markChannelUnread(channelId);
|
||||
|
||||
// Drain: apply the intent (receipt gets co-committed with register).
|
||||
mgr.isLoadComplete = true;
|
||||
await mgr.drainPendingIntents(mgr.loadGeneration);
|
||||
|
||||
// After successful drain, check the persisted v2 blob.
|
||||
const rawAfter = ls.getItem(v2Key);
|
||||
const blobAfter = rawAfter ? JSON.parse(rawAfter) : {};
|
||||
|
||||
// Both intent (pi) and receipt (receipts) must be absent — single cleanup commit.
|
||||
const piEntry = blobAfter?.pi?.[channelId];
|
||||
assert.equal(
|
||||
piEntry,
|
||||
undefined,
|
||||
"intent must be deleted from pi in cleanup commit",
|
||||
);
|
||||
|
||||
const receiptEntry = blobAfter?.receipts?.[channelId];
|
||||
assert.equal(
|
||||
receiptEntry,
|
||||
undefined,
|
||||
"receipt must be deleted from receipts in cleanup commit",
|
||||
);
|
||||
|
||||
// The register IS present (drain succeeded, override applied).
|
||||
const regEntry = blobAfter?.r?.[channelId];
|
||||
assert.ok(regEntry, "register must persist after successful drain");
|
||||
|
||||
mgr.destroy();
|
||||
});
|
||||
|
||||
// ── Test 36: controller — automatic backoff after incomplete verdict ───────────
|
||||
test("retryLoad_incompleteVerdictAdvancesRetryAttempt", async () => {
|
||||
// After an incomplete verdict, retryAttempt increments so the next retry
|
||||
// will use exponential backoff. Destroy/cancel before the delayed retry fires.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "36".repeat(32);
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
subscribeFenced: async () => {
|
||||
throw new Error("fence timeout");
|
||||
},
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
assert.equal(mgr.retryAttempt, 0, "precondition: retryAttempt is 0");
|
||||
|
||||
// First retry (immediate — attempt 0 means delayMs=0).
|
||||
await mgr.retryLoad();
|
||||
|
||||
// After incomplete verdict, retryAttempt must be incremented.
|
||||
assert.ok(
|
||||
mgr.retryAttempt > 0,
|
||||
"retryAttempt must be > 0 after incomplete verdict",
|
||||
);
|
||||
|
||||
// Second retry would use backoff. Destroy before it fires.
|
||||
mgr.destroy();
|
||||
assert.equal(mgr.retryBackoffTimer, null, "timer must be cleared on destroy");
|
||||
});
|
||||
|
||||
// ── Test 37: controller — reconnect during loadInFlight → pendingRetryOnComplete
|
||||
test("retryLoad_reconnectDuringLoadInFlight_coalescesToPendingRetry", async () => {
|
||||
// When retryLoad() is called while a load is already in flight,
|
||||
// pendingRetryOnComplete must be set (not dropped). The pending flag
|
||||
// triggers a fresh generation after the in-flight load completes.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "37".repeat(32);
|
||||
|
||||
let resolveFetch;
|
||||
const fetchBarrier = new Promise((r) => {
|
||||
resolveFetch = r;
|
||||
});
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
subscribeFenced: async (_f, _h) => {
|
||||
// Block until barrier released.
|
||||
await fetchBarrier;
|
||||
return makeFenceHandle({ eose: true });
|
||||
},
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
|
||||
// Start a load that won't complete until we release the barrier.
|
||||
const loadPromise = mgr.retryLoad();
|
||||
// Give the async function time to enter the subscribeFenced await.
|
||||
await new Promise((r) => setTimeout(r, 0));
|
||||
assert.equal(
|
||||
mgr.loadInFlight,
|
||||
true,
|
||||
"loadInFlight must be true while fetch is blocked",
|
||||
);
|
||||
|
||||
// Simulate reconnect arriving while load is in flight.
|
||||
mgr.retryLoad(); // synchronous entry → sets pendingRetryOnComplete
|
||||
assert.equal(
|
||||
mgr.pendingRetryOnComplete,
|
||||
true,
|
||||
"reconnect during loadInFlight must set pendingRetryOnComplete",
|
||||
);
|
||||
|
||||
// Release the barrier so the first retryLoad() can complete.
|
||||
resolveFetch();
|
||||
await loadPromise;
|
||||
|
||||
mgr.destroy();
|
||||
});
|
||||
|
||||
// ── Test 38: controller — stale-completion ordering via gen check ──────────────
|
||||
test("fetchAndMerge_staleGeneration_doesNotSetLoadComplete", async () => {
|
||||
// If loadGeneration advances while fetchAndMerge is in flight, the stale
|
||||
// generation's complete verdict must NOT mark the manager as load-complete.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "38".repeat(32);
|
||||
|
||||
let resolveFirst;
|
||||
const firstBarrier = new Promise((r) => {
|
||||
resolveFirst = r;
|
||||
});
|
||||
let callCount = 0;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
subscribeFenced: async (_f, _h) => {
|
||||
callCount++;
|
||||
if (callCount === 1) {
|
||||
// First call: wait for barrier before returning EOSE.
|
||||
await firstBarrier;
|
||||
return makeFenceHandle({ eose: true });
|
||||
}
|
||||
return makeFenceHandle({ eose: true });
|
||||
},
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
|
||||
// Start gen=1 fetch; it blocks at subscribeFenced.
|
||||
const gen1 = mgr.loadGeneration;
|
||||
const firstFetch = mgr.fetchAndMerge(gen1);
|
||||
await new Promise((r) => setTimeout(r, 0));
|
||||
|
||||
// Advance the generation by starting a new fetch.
|
||||
++mgr.loadGeneration;
|
||||
const gen2 = mgr.loadGeneration;
|
||||
await mgr.fetchAndMerge(gen2);
|
||||
assert.equal(mgr.isLoadComplete, true, "gen2 load must complete");
|
||||
|
||||
// Now release the blocked gen1 fetch.
|
||||
resolveFirst();
|
||||
await firstFetch;
|
||||
|
||||
// Gen1's verdict should not have mutated isLoadComplete back to stale state.
|
||||
// (Gen2 left it true; gen1 must not interfere.)
|
||||
assert.equal(
|
||||
mgr.isLoadComplete,
|
||||
true,
|
||||
"gen1 stale verdict must not corrupt gen2's complete state",
|
||||
);
|
||||
|
||||
mgr.destroy();
|
||||
});
|
||||
|
||||
// ── Test 39: controller — destroy cancels backoff timer ───────────────────────
|
||||
test("retryLoad_destroyWhileTimerPending_cancelsTimer", async () => {
|
||||
// destroy() must cancel any pending retryBackoffTimer so no stale retry fires.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "39".repeat(32);
|
||||
|
||||
let timerSet = false;
|
||||
let pendingTimerResolve = null; // captured so we can drain after destroy
|
||||
const origSetTimeout = globalThis.window.setTimeout;
|
||||
const origClearTimeout = globalThis.window.clearTimeout;
|
||||
|
||||
// Track active backoff timers by real ID so clearTimeout works correctly.
|
||||
// When ms > 0 we wrap the fn to capture state, but still use the real timer
|
||||
// (with a large delay so it doesn't fire during the test) so origClearTimeout
|
||||
// can cancel it. After destroy() we must resolve the pending await manually
|
||||
// to let retryLoad's finally-block run and the promise settle.
|
||||
const activeTimerIds = new Set();
|
||||
globalThis.window.setTimeout = (fn, ms) => {
|
||||
if (ms > 0) {
|
||||
timerSet = true;
|
||||
// Schedule with origSetTimeout at a safe delay so clearTimeout works.
|
||||
const id = origSetTimeout(() => {
|
||||
activeTimerIds.delete(id);
|
||||
pendingTimerResolve = null;
|
||||
fn();
|
||||
}, 60_000); // long enough never to fire during test
|
||||
// Capture resolve so we can drain manually after destroy().
|
||||
pendingTimerResolve = fn;
|
||||
activeTimerIds.add(id);
|
||||
return id;
|
||||
}
|
||||
return origSetTimeout(fn, ms);
|
||||
};
|
||||
globalThis.window.clearTimeout = (id) => {
|
||||
activeTimerIds.delete(id);
|
||||
origClearTimeout(id);
|
||||
};
|
||||
|
||||
try {
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
subscribeFenced: async () => {
|
||||
throw new Error("lapse");
|
||||
},
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
|
||||
// First retry (immediate, attempt=0 → no timer).
|
||||
await mgr.retryLoad();
|
||||
// retryAttempt is now 1; next retry will schedule a timer.
|
||||
|
||||
// Start second retry asynchronously — it will set a timer.
|
||||
const retryPromise = mgr.retryLoad();
|
||||
// Give async fn time to reach the timer await.
|
||||
await new Promise((r) => origSetTimeout(r, 0));
|
||||
|
||||
assert.equal(timerSet, true, "a backoff timer must have been scheduled");
|
||||
assert.ok(
|
||||
mgr.retryBackoffTimer !== null,
|
||||
"retryBackoffTimer must be non-null while waiting",
|
||||
);
|
||||
|
||||
// Destroy must cancel the timer.
|
||||
mgr.destroy();
|
||||
assert.equal(
|
||||
mgr.retryBackoffTimer,
|
||||
null,
|
||||
"retryBackoffTimer must be null after destroy",
|
||||
);
|
||||
assert.equal(
|
||||
activeTimerIds.size,
|
||||
0,
|
||||
"backoff timer must be cleared by destroy()",
|
||||
);
|
||||
|
||||
// retryLoad is still awaiting its backoff Promise. Fire the pending resolve
|
||||
// manually (the real timer was cleared so it won't auto-fire) so the
|
||||
// finally-block in retryLoad runs and retryPromise settles. Since
|
||||
// mgr.destroyed===true, retryLoad returns immediately after the resolve.
|
||||
if (pendingTimerResolve) pendingTimerResolve();
|
||||
await retryPromise;
|
||||
} finally {
|
||||
globalThis.window.setTimeout = origSetTimeout;
|
||||
globalThis.window.clearTimeout = origClearTimeout;
|
||||
}
|
||||
});
|
||||
|
||||
// ── Test 40: controller — exactly one drain per complete-load generation ───────
|
||||
test("retryLoad_exactlyOneDrainPerCompleteGeneration", async () => {
|
||||
// A single complete-load transition must trigger exactly one drain.
|
||||
// Multiple retryLoad() calls for the same complete generation must not
|
||||
// enqueue a second drain.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "40".repeat(32);
|
||||
const channelId = `one-drain-ch-${"h".repeat(51)}`;
|
||||
|
||||
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 = false;
|
||||
|
||||
// Queue an intent that will be consumed by the drain.
|
||||
mgr.markChannelUnread(channelId);
|
||||
|
||||
// First retryLoad → complete → drain fires.
|
||||
await mgr.retryLoad();
|
||||
assert.equal(mgr.isLoadComplete, true, "load must be complete");
|
||||
|
||||
// Record register state after first drain.
|
||||
const regAfterFirst = mgr.overrideRegisters.get(channelId);
|
||||
const sAfterFirst = regAfterFirst?.s ?? 0;
|
||||
|
||||
// Second retryLoad on an already-complete manager: drainScheduled is false
|
||||
// (already consumed), so a second drain must NOT re-bump the register.
|
||||
await mgr.retryLoad();
|
||||
|
||||
const regAfterSecond = mgr.overrideRegisters.get(channelId);
|
||||
assert.equal(
|
||||
regAfterSecond?.s ?? 0,
|
||||
sAfterFirst,
|
||||
"S must NOT be re-bumped by a second retryLoad (exactly-one-drain invariant)",
|
||||
);
|
||||
|
||||
mgr.destroy();
|
||||
});
|
||||
|
||||
// ── Test 41: store/UI — identity swap during drain cancels stale gen ──────────
|
||||
test("drain_identitySwapDuringDrain_staleGenCancelled", async () => {
|
||||
// If the manager is destroyed and recreated for a new pubkey, the stale
|
||||
// generation's drain must be cancelled. Verify via the pubkey fence inside
|
||||
// drainPendingIntents: pubkeyFence !== ctx.pubkey → break.
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkeyA = "41".repeat(32);
|
||||
const channelId = `identity-swap-ch-${"i".repeat(47)}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }),
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
};
|
||||
|
||||
const mgrA = new ReadStateManager(pubkeyA, fakeRelay);
|
||||
mgrA.effectiveState.set(channelId, 50);
|
||||
mgrA.isLoadComplete = false;
|
||||
|
||||
// Queue intent for pubkeyA.
|
||||
mgrA.markChannelUnread(channelId);
|
||||
const intentA = pendingOverrideIntentStore.get(channelId);
|
||||
assert.ok(intentA, "intent must exist for pubkeyA");
|
||||
|
||||
// Destroy mgrA before drain (simulates identity swap).
|
||||
mgrA.destroy();
|
||||
|
||||
// The drainPendingIntents call with mgrA's context should be cancelled
|
||||
// because mgrA.destroyed === true.
|
||||
// Verify by running the drain and asserting the intent is NOT consumed.
|
||||
// (The drain early-exits on ctx.destroyed.)
|
||||
await mgrA.drainPendingIntents(mgrA.loadGeneration);
|
||||
|
||||
// The intent must still be alive — drain was no-op because destroyed.
|
||||
const intentAfter = pendingOverrideIntentStore.get(channelId);
|
||||
assert.ok(intentAfter, "intent must survive drain when manager is destroyed");
|
||||
|
||||
// Clean up.
|
||||
pendingOverrideIntentStore.compareAndDelete(channelId, intentAfter.gen);
|
||||
});
|
||||
|
||||
// ── Test 42: deferred refusal — queued unread intent refused by drain ─────────
|
||||
test("drain_queuedUnread_refused_surfacesOutcome", async () => {
|
||||
// When a queued unread intent is refused by drain (e.g., budget_exhausted),
|
||||
// onDrainOutcome must be called with (channelId, "unread", "refused", reason).
|
||||
globalThis.window.localStorage = makeLocalStorage();
|
||||
const pubkey = "42".repeat(32);
|
||||
const channelId = `deferred-refusal-ch-${"j".repeat(44)}`;
|
||||
|
||||
const fakeRelay = {
|
||||
fetchEvents: async () => [],
|
||||
publishEvent: async () => {},
|
||||
subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }),
|
||||
subscribeLive: async (_f, _h) => () => {},
|
||||
subscribeToReconnects: () => () => {},
|
||||
getConnectionGeneration: () => 0,
|
||||
};
|
||||
|
||||
const mgr = new ReadStateManager(pubkey, fakeRelay);
|
||||
mgr.effectiveState.set(channelId, 50);
|
||||
mgr.isLoadComplete = false;
|
||||
|
||||
// Queue an unread intent with uint32 overflow condition: S is at max.
|
||||
mgr.overrideRegisters.set(channelId, { s: 0xffffffff, c: 0, b: 50 });
|
||||
mgr.publishableContextIds.add(channelId);
|
||||
const q = mgr.markChannelUnread(channelId);
|
||||
assert.equal(q.status, "queued", "must queue pre-ready");
|
||||
|
||||
// Wire outcome callback.
|
||||
const outcomes = [];
|
||||
mgr.onDrainOutcome = (chId, op, status, reason) => {
|
||||
outcomes.push({ chId, op, status, reason });
|
||||
};
|
||||
|
||||
// Complete load and drain.
|
||||
mgr.isLoadComplete = true;
|
||||
await mgr.drainPendingIntents(mgr.loadGeneration);
|
||||
|
||||
// Drain must have surfaced the refusal.
|
||||
assert.equal(outcomes.length, 1, "exactly one outcome must be emitted");
|
||||
assert.equal(outcomes[0].chId, channelId);
|
||||
assert.equal(outcomes[0].op, "unread");
|
||||
assert.equal(outcomes[0].status, "refused");
|
||||
assert.equal(
|
||||
outcomes[0].reason,
|
||||
"uint32_overflow",
|
||||
"refusal reason must be uint32_overflow",
|
||||
);
|
||||
|
||||
// Intent must be cleaned up even after refusal (it's a genuine refusal, not load_incomplete).
|
||||
assert.equal(
|
||||
pendingOverrideIntentStore.get(channelId),
|
||||
undefined,
|
||||
"intent must be deleted after genuine refusal",
|
||||
);
|
||||
|
||||
mgr.destroy();
|
||||
});
|
||||
|
||||
@@ -9,11 +9,9 @@ import {
|
||||
READ_STATE_MAX_SLOTS,
|
||||
MSG_PREFIX,
|
||||
THREAD_PREFIX,
|
||||
isOverrideKey,
|
||||
encodeOverrideGroup,
|
||||
isOverrideActive,
|
||||
escapeFrontierKey,
|
||||
unescapeFrontierKey,
|
||||
localExtraSlotIdsKey,
|
||||
type ReadStateBlob,
|
||||
type OverrideRegister,
|
||||
@@ -30,6 +28,12 @@ import {
|
||||
writeStoredReadState,
|
||||
type AppliedReceipt,
|
||||
} from "@/features/channels/readState/readStateStorage";
|
||||
import {
|
||||
splitContextsIntoBudgetedSlots,
|
||||
trimContextsToBudget,
|
||||
type SlotSplitResult,
|
||||
type TrimResult,
|
||||
} from "@/features/channels/readState/readStateSlotUtils";
|
||||
import {
|
||||
fencedEnumerationLoad,
|
||||
deduplicateByCoordinate,
|
||||
@@ -40,8 +44,12 @@ import {
|
||||
drainPendingIntents as drainPendingIntentsImpl,
|
||||
markChannelUnread as markChannelUnreadImpl,
|
||||
markChannelRead as markChannelReadImpl,
|
||||
type DrainContext,
|
||||
createDrainContext,
|
||||
} from "@/features/channels/readState/readStateDrain";
|
||||
|
||||
export type { SlotSplitResult, TrimResult };
|
||||
export { splitContextsIntoBudgetedSlots, trimContextsToBudget };
|
||||
|
||||
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;
|
||||
@@ -153,128 +161,14 @@ export function applyRemoteContextTimestamp(args: {
|
||||
return result;
|
||||
}
|
||||
|
||||
/** Slot split result. */
|
||||
export interface SlotSplitResult {
|
||||
slots: Array<Record<string, number>>;
|
||||
extraSlotIds: string[];
|
||||
}
|
||||
|
||||
/** Partition channelEntries across slots; override groups pinned to slot 0. Exported for testing. */
|
||||
export function splitContextsIntoBudgetedSlots(args: {
|
||||
channelEntries: [string, number][];
|
||||
threadMsgEntries: [string, number][];
|
||||
clientId: string;
|
||||
initialSlotCount: number;
|
||||
maxSlots: number;
|
||||
maxBytes: number;
|
||||
slotIdGenerator: () => string;
|
||||
}): SlotSplitResult | null {
|
||||
const {
|
||||
channelEntries,
|
||||
threadMsgEntries,
|
||||
clientId,
|
||||
initialSlotCount,
|
||||
maxSlots,
|
||||
maxBytes,
|
||||
slotIdGenerator,
|
||||
} = args;
|
||||
|
||||
const encoder = new TextEncoder();
|
||||
const blobFor = (c: Record<string, number>) =>
|
||||
JSON.stringify({ v: 1, client_id: clientId, contexts: c });
|
||||
|
||||
const overrideRawIds = new Set<string>();
|
||||
for (const [key] of channelEntries) {
|
||||
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) || overrideRawIds.has(unescapeFrontierKey(key))) {
|
||||
pinnedEntries.push([key, ts]);
|
||||
} else {
|
||||
roundRobinEntries.push([key, ts]);
|
||||
}
|
||||
}
|
||||
let slotCount = initialSlotCount;
|
||||
const extraSlotIds: string[] = [];
|
||||
const distribute = (count: number): Array<Record<string, number>> => {
|
||||
const slotContexts: Array<Record<string, number>> = Array.from(
|
||||
{ length: count },
|
||||
() => ({}),
|
||||
);
|
||||
for (let i = 0; i < roundRobinEntries.length; i++) {
|
||||
const [key, ts] = roundRobinEntries[i];
|
||||
slotContexts[i % count][key] = ts;
|
||||
}
|
||||
for (const [key, ts] of pinnedEntries) slotContexts[0][key] = ts;
|
||||
return slotContexts;
|
||||
};
|
||||
|
||||
let slotContexts = distribute(slotCount);
|
||||
while (
|
||||
slotContexts.some((c) => encoder.encode(blobFor(c)).length > maxBytes) &&
|
||||
slotCount < maxSlots
|
||||
) {
|
||||
extraSlotIds.push(slotIdGenerator());
|
||||
slotCount++;
|
||||
slotContexts = distribute(slotCount);
|
||||
}
|
||||
if (slotContexts.some((c) => encoder.encode(blobFor(c)).length > maxBytes))
|
||||
return null;
|
||||
for (const [key, ts] of threadMsgEntries) slotContexts[0][key] = ts;
|
||||
trimContextsToBudget(slotContexts[0], clientId, maxBytes);
|
||||
return { slots: slotContexts, extraSlotIds };
|
||||
}
|
||||
|
||||
export interface TrimResult {
|
||||
evicted: number;
|
||||
fitsAfterTrim: boolean;
|
||||
}
|
||||
|
||||
/** Trim a contexts map to fit within `maxBytes`. Evicts oldest msg:/thread: entries; channel/ov_* keys never evicted. */
|
||||
export function trimContextsToBudget(
|
||||
contexts: Record<string, number>,
|
||||
clientId: string,
|
||||
maxBytes: number,
|
||||
): TrimResult {
|
||||
const encoder = new TextEncoder();
|
||||
const blobFor = (c: Record<string, number>) =>
|
||||
JSON.stringify({ v: 1, client_id: clientId, contexts: c });
|
||||
|
||||
let currentBytes = encoder.encode(blobFor(contexts)).length;
|
||||
if (currentBytes <= maxBytes) {
|
||||
return { evicted: 0, fitsAfterTrim: true };
|
||||
}
|
||||
|
||||
const msgEntries: [string, number][] = [];
|
||||
const threadEntries: [string, number][] = [];
|
||||
for (const [key, ts] of Object.entries(contexts)) {
|
||||
if (isOverrideKey(key)) continue;
|
||||
if (key.startsWith(MSG_PREFIX)) msgEntries.push([key, ts]);
|
||||
else if (key.startsWith(THREAD_PREFIX)) threadEntries.push([key, ts]);
|
||||
}
|
||||
msgEntries.sort((a, b) => a[1] - b[1]);
|
||||
threadEntries.sort((a, b) => a[1] - b[1]);
|
||||
const toEvict: string[] = [];
|
||||
for (const [key, ts] of [...msgEntries, ...threadEntries]) {
|
||||
if (currentBytes <= maxBytes) break;
|
||||
currentBytes -= key.length + 3 + String(ts).length + 1;
|
||||
toEvict.push(key);
|
||||
}
|
||||
for (const key of toEvict) delete contexts[key];
|
||||
const fitsAfterTrim = encoder.encode(blobFor(contexts)).length <= maxBytes;
|
||||
return { evicted: toEvict.length, fitsAfterTrim };
|
||||
}
|
||||
|
||||
export class ReadStateManager {
|
||||
private pubkey: string;
|
||||
readonly pubkey: string;
|
||||
private relayClient: RelayClient;
|
||||
private clientId: string;
|
||||
readonly clientId: string;
|
||||
private slotId: string;
|
||||
private extraSlotIds: string[];
|
||||
private effectiveState = new Map<string, number>();
|
||||
private publishableContextIds = new Set<string>();
|
||||
extraSlotIds: string[];
|
||||
readonly effectiveState = new Map<string, number>();
|
||||
readonly publishableContextIds = new Set<string>();
|
||||
private lastPublishedContexts: Record<string, number> = {};
|
||||
private debounceTimer: number | null = null;
|
||||
private listeners = new Set<() => void>();
|
||||
@@ -283,15 +177,27 @@ export class ReadStateManager {
|
||||
private maxFetchedCreatedAt = 0;
|
||||
private contextSourceCreatedAt = new Map<string, number>();
|
||||
private pendingSyncedAdvances = new Set<string>();
|
||||
private destroyed = false;
|
||||
destroyed = false;
|
||||
private parentResolver: ContextParentResolver | null = null;
|
||||
private overrideRegisters = new Map<string, OverrideRegister>();
|
||||
private isLoadComplete = false;
|
||||
private appliedReceipts = new Map<string, AppliedReceipt>(); // co-committed with register (Amendment C)
|
||||
private loadGeneration = 0; // single-flight controller
|
||||
private loadInFlight = false;
|
||||
private retryBackoffTimer: number | null = null;
|
||||
private drainScheduled = false;
|
||||
readonly overrideRegisters = new Map<string, OverrideRegister>();
|
||||
isLoadComplete = false;
|
||||
readonly appliedReceipts = new Map<string, AppliedReceipt>(); // co-committed with register (Amendment C)
|
||||
loadGeneration = 0; // single-flight controller
|
||||
loadInFlight = false;
|
||||
retryBackoffTimer: number | null = null;
|
||||
retryAttempt = 0; // bounded exponential backoff attempt counter
|
||||
pendingRetryOnComplete = false; // reconnect arrived during loadInFlight
|
||||
drainScheduled = false;
|
||||
/** Called after each drain intent is settled; wired by the hook for outcome routing. */
|
||||
onDrainOutcome:
|
||||
| ((
|
||||
channelId: string,
|
||||
op: string,
|
||||
status: string,
|
||||
reason?: string,
|
||||
sourceScope?: string,
|
||||
) => void)
|
||||
| null = null;
|
||||
|
||||
constructor(pubkey: string, relayClient: RelayClient) {
|
||||
this.pubkey = pubkey;
|
||||
@@ -304,7 +210,7 @@ export class ReadStateManager {
|
||||
generateHex(16),
|
||||
);
|
||||
this.extraSlotIds = loadExtraSlotIds(pubkey);
|
||||
pendingOverrideIntentStore.loadForPubkey(pubkey);
|
||||
// pendingOverrideIntentStore is restored from v2 blob in hydrateFromLocalStorage.
|
||||
}
|
||||
|
||||
async initialize(): Promise<void> {
|
||||
@@ -313,7 +219,7 @@ export class ReadStateManager {
|
||||
this.loadInFlight = true;
|
||||
const gen = ++this.loadGeneration;
|
||||
try {
|
||||
await this.fetchAndMerge();
|
||||
await this.fetchAndMerge(gen);
|
||||
} finally {
|
||||
if (gen === this.loadGeneration) this.loadInFlight = false;
|
||||
}
|
||||
@@ -395,6 +301,7 @@ export class ReadStateManager {
|
||||
this.destroyed = true;
|
||||
this.loadGeneration++;
|
||||
this.loadInFlight = false;
|
||||
this.pendingRetryOnComplete = false;
|
||||
this.drainScheduled = false;
|
||||
if (this.retryBackoffTimer !== null) {
|
||||
window.clearTimeout(this.retryBackoffTimer);
|
||||
@@ -412,10 +319,10 @@ export class ReadStateManager {
|
||||
this.listeners.clear();
|
||||
}
|
||||
|
||||
private async fetchAndMerge(): Promise<void> {
|
||||
if (this.destroyed) return;
|
||||
async fetchAndMerge(gen: number = this.loadGeneration): Promise<void> {
|
||||
if (this.destroyed || gen !== this.loadGeneration) return;
|
||||
const result = await fencedEnumerationLoad(this.relayClient, this.pubkey);
|
||||
if (this.destroyed) return;
|
||||
if (this.destroyed || gen !== this.loadGeneration) return;
|
||||
this.isLoadComplete = result.complete;
|
||||
if (!result.complete)
|
||||
console.warn(
|
||||
@@ -424,47 +331,82 @@ export class ReadStateManager {
|
||||
this.processLoadedParsedEvents(result.events);
|
||||
const merged = mergeParsedEvents(result.events);
|
||||
await this.ingest(merged);
|
||||
if (this.destroyed || gen !== this.loadGeneration) return;
|
||||
this.persistLocalState();
|
||||
this.notifyListeners();
|
||||
}
|
||||
|
||||
/** Retry load: generation-fenced, coalesces in-flight loads, resets backoff. */
|
||||
/**
|
||||
* Retry load: generation-fenced, coalesces in-flight loads.
|
||||
* Implements bounded exponential backoff (1s→2s→4s…capped at 60s).
|
||||
* A reconnect during loadInFlight sets pendingRetryOnComplete so the next
|
||||
* complete/incomplete verdict triggers exactly one fresh generation rather
|
||||
* than dropping the reconnect signal.
|
||||
*/
|
||||
async retryLoad(): Promise<void> {
|
||||
if (this.destroyed) return;
|
||||
if (this.retryBackoffTimer !== null) {
|
||||
window.clearTimeout(this.retryBackoffTimer);
|
||||
this.retryBackoffTimer = null;
|
||||
}
|
||||
if (this.loadInFlight) return;
|
||||
if (this.loadInFlight) {
|
||||
// Reconnect arrived while a load is in flight — coalesce into a required
|
||||
// fresh generation rather than dropping the signal.
|
||||
this.pendingRetryOnComplete = true;
|
||||
return;
|
||||
}
|
||||
const delayMs =
|
||||
this.retryAttempt === 0
|
||||
? 0
|
||||
: Math.min(1_000 * 2 ** (this.retryAttempt - 1), 60_000);
|
||||
if (delayMs > 0) {
|
||||
await new Promise<void>((resolve) => {
|
||||
this.retryBackoffTimer = window.setTimeout(() => {
|
||||
this.retryBackoffTimer = null;
|
||||
resolve();
|
||||
}, delayMs);
|
||||
});
|
||||
}
|
||||
if (this.destroyed || this.loadInFlight) return;
|
||||
this.retryAttempt++;
|
||||
this.loadInFlight = true;
|
||||
const gen = ++this.loadGeneration;
|
||||
try {
|
||||
this.isLoadComplete = false;
|
||||
await this.fetchAndMerge();
|
||||
await this.fetchAndMerge(gen);
|
||||
if (this.destroyed || gen !== this.loadGeneration) return;
|
||||
if (!this.drainScheduled) {
|
||||
if (this.isLoadComplete) this.retryAttempt = 0;
|
||||
if (!this.drainScheduled && this.isLoadComplete) {
|
||||
this.drainScheduled = true;
|
||||
void this.drainPendingIntents(gen);
|
||||
}
|
||||
} finally {
|
||||
if (gen === this.loadGeneration) {
|
||||
this.loadInFlight = false;
|
||||
if (this.pendingRetryOnComplete && !this.destroyed) {
|
||||
this.pendingRetryOnComplete = false;
|
||||
void this.retryLoad();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Drain pending override intents. See readStateDrain.ts for the full implementation. */
|
||||
private async drainPendingIntents(drainGen: number): Promise<void> {
|
||||
await drainPendingIntentsImpl(this as unknown as DrainContext, drainGen);
|
||||
async drainPendingIntents(drainGen: number): Promise<void> {
|
||||
await drainPendingIntentsImpl(createDrainContext(this), drainGen);
|
||||
this.drainScheduled = false;
|
||||
}
|
||||
|
||||
markChannelUnread(channelId: string): MarkResult {
|
||||
return markChannelUnreadImpl(this as unknown as DrainContext, channelId);
|
||||
return markChannelUnreadImpl(createDrainContext(this), channelId);
|
||||
}
|
||||
|
||||
markChannelRead(channelId: string): MarkResult {
|
||||
return markChannelReadImpl(this as unknown as DrainContext, channelId);
|
||||
markChannelRead(channelId: string, sourceScope?: string): MarkResult {
|
||||
return markChannelReadImpl(
|
||||
createDrainContext(this),
|
||||
channelId,
|
||||
sourceScope,
|
||||
);
|
||||
}
|
||||
|
||||
/** Shared ingest: accept pre-merged state and return semantic delta (changed contexts, canonical flip). */
|
||||
@@ -625,7 +567,7 @@ export class ReadStateManager {
|
||||
}
|
||||
}
|
||||
|
||||
private schedulePublish(): void {
|
||||
schedulePublish(): void {
|
||||
if (this.debounceTimer !== null) window.clearTimeout(this.debounceTimer);
|
||||
this.debounceTimer = window.setTimeout(() => {
|
||||
this.debounceTimer = null;
|
||||
@@ -633,7 +575,7 @@ export class ReadStateManager {
|
||||
}, DEBOUNCE_MS);
|
||||
}
|
||||
|
||||
private async publish(): Promise<void> {
|
||||
async publish(): Promise<void> {
|
||||
if (!this.isLoadComplete) return;
|
||||
if (!(await this.fetchOwnBlobBeforePublish())) {
|
||||
console.warn(
|
||||
@@ -710,7 +652,7 @@ export class ReadStateManager {
|
||||
await this.publishOneSlot(slotId, contexts);
|
||||
}
|
||||
|
||||
private async deleteExtraSlots(): Promise<void> {
|
||||
async deleteExtraSlots(): Promise<void> {
|
||||
if (!this.isLoadComplete) return;
|
||||
for (const slotId of this.extraSlotIds) {
|
||||
try {
|
||||
@@ -733,7 +675,7 @@ export class ReadStateManager {
|
||||
saveExtraSlotIds(this.pubkey, []);
|
||||
}
|
||||
|
||||
private async fetchOwnBlobBeforePublish(): Promise<boolean> {
|
||||
async fetchOwnBlobBeforePublish(): Promise<boolean> {
|
||||
const dTags = [this.slotId, ...this.extraSlotIds].map(
|
||||
(id) => `${READ_STATE_D_TAG_PREFIX}${id}`,
|
||||
);
|
||||
@@ -773,7 +715,7 @@ export class ReadStateManager {
|
||||
}
|
||||
|
||||
/** Effective frontier including parent resolver. */
|
||||
private channelFrontier(channelId: string): number {
|
||||
channelFrontier(channelId: string): number {
|
||||
return (
|
||||
resolveEffectiveTimestamp({
|
||||
effectiveState: this.effectiveState,
|
||||
@@ -902,7 +844,7 @@ export class ReadStateManager {
|
||||
: saveExtraSlotIds(this.pubkey, this.extraSlotIds);
|
||||
}
|
||||
|
||||
private hydrateFromLocalStorage(): void {
|
||||
hydrateFromLocalStorage(): void {
|
||||
const stored = readStoredReadState(this.pubkey);
|
||||
for (const [contextId, timestamp] of stored.contexts)
|
||||
this.effectiveState.set(contextId, timestamp);
|
||||
@@ -927,6 +869,11 @@ export class ReadStateManager {
|
||||
);
|
||||
this.publishableContextIds.add(ctx); // v2 entry is inherently publishable
|
||||
}
|
||||
// Restore pending intents from v2 blob (consolidation: no separate key).
|
||||
pendingOverrideIntentStore.restoreFromStorage(
|
||||
stored.pendingIntents,
|
||||
stored.nextGen,
|
||||
);
|
||||
// Amendment C hydration: load receipts; orphan sweep for receipts with no matching intent.
|
||||
for (const [channelId, receipt] of stored.appliedReceipts) {
|
||||
const intent = pendingOverrideIntentStore.get(channelId);
|
||||
@@ -942,7 +889,7 @@ export class ReadStateManager {
|
||||
}
|
||||
|
||||
/** Persist local state. Returns false only if the v2 override write failed (commit point). */
|
||||
private persistLocalState(): boolean {
|
||||
persistLocalState(): boolean {
|
||||
const entries = new Map(
|
||||
[...this.overrideRegisters].map(([ctx, r]) => [
|
||||
ctx,
|
||||
@@ -956,6 +903,8 @@ export class ReadStateManager {
|
||||
this.contextSourceCreatedAt,
|
||||
entries,
|
||||
this.appliedReceipts,
|
||||
pendingOverrideIntentStore.all,
|
||||
pendingOverrideIntentStore.nextGen,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -965,7 +914,7 @@ export class ReadStateManager {
|
||||
return drained;
|
||||
}
|
||||
|
||||
private notifyListeners(): void {
|
||||
notifyListeners(): void {
|
||||
for (const listener of this.listeners) {
|
||||
try {
|
||||
listener();
|
||||
|
||||
@@ -729,7 +729,7 @@ test("roundTrip_escapedRawCtx_survivesEncodeAndParse", () => {
|
||||
|
||||
test("parseContexts_tombstoneFloor_multibyte_exactly256Bytes_accepted", () => {
|
||||
// 'é' × 125 = 250 bytes, + 'a' = 251 bytes suffix → wire key exactly 256 bytes.
|
||||
const suffix = "\u00e9".repeat(125) + "a";
|
||||
const suffix = `${"\u00e9".repeat(125)}a`;
|
||||
const raw = {
|
||||
[suffix]: 0,
|
||||
[`${OV_C_PREFIX}${suffix}`]: 7,
|
||||
@@ -740,7 +740,7 @@ test("parseContexts_tombstoneFloor_multibyte_exactly256Bytes_accepted", () => {
|
||||
|
||||
test("parseContexts_tombstoneFloor_multibyte_exactly257Bytes_dropped", () => {
|
||||
// 'é' × 125 = 250 bytes, + 'aa' = 252 bytes suffix → wire key exactly 257 bytes.
|
||||
const suffix = "\u00e9".repeat(125) + "aa";
|
||||
const suffix = `${"\u00e9".repeat(125)}aa`;
|
||||
const raw = {
|
||||
[suffix]: 0,
|
||||
[`${OV_C_PREFIX}${suffix}`]: 7,
|
||||
@@ -751,7 +751,7 @@ test("parseContexts_tombstoneFloor_multibyte_exactly257Bytes_dropped", () => {
|
||||
|
||||
test("parseContexts_liveGroup_multibyte_exactly256Bytes_accepted", () => {
|
||||
// Same suffix construction: 'é' × 125 + 'a' = 251 bytes → ov_s: wire key 256 bytes.
|
||||
const suffix = "\u00e9".repeat(125) + "a";
|
||||
const suffix = `${"\u00e9".repeat(125)}a`;
|
||||
const raw = {
|
||||
[suffix]: 10,
|
||||
[`${OV_S_PREFIX}${suffix}`]: 2,
|
||||
@@ -764,7 +764,7 @@ test("parseContexts_liveGroup_multibyte_exactly256Bytes_accepted", () => {
|
||||
|
||||
test("parseContexts_liveGroup_multibyte_exactly257Bytes_dropped", () => {
|
||||
// 'é' × 125 + 'aa' = 252 bytes → ov_s: wire key 257 bytes → group dropped.
|
||||
const suffix = "\u00e9".repeat(125) + "aa";
|
||||
const suffix = `${"\u00e9".repeat(125)}aa`;
|
||||
const raw = {
|
||||
[suffix]: 10,
|
||||
[`${OV_S_PREFIX}${suffix}`]: 2,
|
||||
@@ -1020,9 +1020,9 @@ for (const [name, input, expected] of [
|
||||
// 'g' and 'z' are not hexadecimal.
|
||||
["nonHexChars_returnsFalse", "read-state:gggggggggggggggggggggggggggggggg", false],
|
||||
// One char short — must be exactly 32.
|
||||
["31HexChars_returnsFalse", "read-state:" + "a".repeat(31), false],
|
||||
["31HexChars_returnsFalse", `read-state:${"a".repeat(31)}`, false],
|
||||
// One char over — must be exactly 32.
|
||||
["33HexChars_returnsFalse", "read-state:" + "a".repeat(33), false],
|
||||
["33HexChars_returnsFalse", `read-state:${"a".repeat(33)}`, false],
|
||||
["missingPrefix_returnsFalse", "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", false],
|
||||
["emptySlot_returnsFalse", "read-state:", false],
|
||||
["undefined_returnsFalse", undefined, false],
|
||||
|
||||
127
desktop/src/features/channels/readState/readStateSlotUtils.ts
Normal file
127
desktop/src/features/channels/readState/readStateSlotUtils.ts
Normal file
@@ -0,0 +1,127 @@
|
||||
/**
|
||||
* Pure slot-splitting and trimming utilities for NIP-RS read-state publication.
|
||||
*
|
||||
* Extracted from readStateManager.ts to keep that file within the 1000-line
|
||||
* size ratchet. Both functions are stateless and depend only on the
|
||||
* readStateFormat constants and the TextEncoder API.
|
||||
*/
|
||||
import {
|
||||
isOverrideKey,
|
||||
unescapeFrontierKey,
|
||||
MSG_PREFIX,
|
||||
THREAD_PREFIX,
|
||||
} from "@/features/channels/readState/readStateFormat";
|
||||
|
||||
/** Slot split result. */
|
||||
export interface SlotSplitResult {
|
||||
slots: Array<Record<string, number>>;
|
||||
extraSlotIds: string[];
|
||||
}
|
||||
|
||||
export interface TrimResult {
|
||||
evicted: number;
|
||||
fitsAfterTrim: boolean;
|
||||
}
|
||||
|
||||
/** Partition channelEntries across slots; override groups pinned to slot 0. */
|
||||
export function splitContextsIntoBudgetedSlots(args: {
|
||||
channelEntries: [string, number][];
|
||||
threadMsgEntries: [string, number][];
|
||||
clientId: string;
|
||||
initialSlotCount: number;
|
||||
maxSlots: number;
|
||||
maxBytes: number;
|
||||
slotIdGenerator: () => string;
|
||||
}): SlotSplitResult | null {
|
||||
const {
|
||||
channelEntries,
|
||||
threadMsgEntries,
|
||||
clientId,
|
||||
initialSlotCount,
|
||||
maxSlots,
|
||||
maxBytes,
|
||||
slotIdGenerator,
|
||||
} = args;
|
||||
|
||||
const encoder = new TextEncoder();
|
||||
const blobFor = (c: Record<string, number>) =>
|
||||
JSON.stringify({ v: 1, client_id: clientId, contexts: c });
|
||||
|
||||
const overrideRawIds = new Set<string>();
|
||||
for (const [key] of channelEntries) {
|
||||
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) || overrideRawIds.has(unescapeFrontierKey(key))) {
|
||||
pinnedEntries.push([key, ts]);
|
||||
} else {
|
||||
roundRobinEntries.push([key, ts]);
|
||||
}
|
||||
}
|
||||
let slotCount = initialSlotCount;
|
||||
const extraSlotIds: string[] = [];
|
||||
const distribute = (count: number): Array<Record<string, number>> => {
|
||||
const slotContexts: Array<Record<string, number>> = Array.from(
|
||||
{ length: count },
|
||||
() => ({}),
|
||||
);
|
||||
for (let i = 0; i < roundRobinEntries.length; i++) {
|
||||
const [key, ts] = roundRobinEntries[i];
|
||||
slotContexts[i % count][key] = ts;
|
||||
}
|
||||
for (const [key, ts] of pinnedEntries) slotContexts[0][key] = ts;
|
||||
return slotContexts;
|
||||
};
|
||||
|
||||
let slotContexts = distribute(slotCount);
|
||||
while (
|
||||
slotContexts.some((c) => encoder.encode(blobFor(c)).length > maxBytes) &&
|
||||
slotCount < maxSlots
|
||||
) {
|
||||
extraSlotIds.push(slotIdGenerator());
|
||||
slotCount++;
|
||||
slotContexts = distribute(slotCount);
|
||||
}
|
||||
if (slotContexts.some((c) => encoder.encode(blobFor(c)).length > maxBytes))
|
||||
return null;
|
||||
for (const [key, ts] of threadMsgEntries) slotContexts[0][key] = ts;
|
||||
trimContextsToBudget(slotContexts[0], clientId, maxBytes);
|
||||
return { slots: slotContexts, extraSlotIds };
|
||||
}
|
||||
|
||||
/** Trim a contexts map to fit within `maxBytes`. Evicts oldest msg:/thread: entries; channel/ov_* keys never evicted. */
|
||||
export function trimContextsToBudget(
|
||||
contexts: Record<string, number>,
|
||||
clientId: string,
|
||||
maxBytes: number,
|
||||
): TrimResult {
|
||||
const encoder = new TextEncoder();
|
||||
const blobFor = (c: Record<string, number>) =>
|
||||
JSON.stringify({ v: 1, client_id: clientId, contexts: c });
|
||||
|
||||
let currentBytes = encoder.encode(blobFor(contexts)).length;
|
||||
if (currentBytes <= maxBytes) {
|
||||
return { evicted: 0, fitsAfterTrim: true };
|
||||
}
|
||||
|
||||
const msgEntries: [string, number][] = [];
|
||||
const threadEntries: [string, number][] = [];
|
||||
for (const [key, ts] of Object.entries(contexts)) {
|
||||
if (isOverrideKey(key)) continue;
|
||||
if (key.startsWith(MSG_PREFIX)) msgEntries.push([key, ts]);
|
||||
else if (key.startsWith(THREAD_PREFIX)) threadEntries.push([key, ts]);
|
||||
}
|
||||
msgEntries.sort((a, b) => a[1] - b[1]);
|
||||
threadEntries.sort((a, b) => a[1] - b[1]);
|
||||
const toEvict: string[] = [];
|
||||
for (const [key, ts] of [...msgEntries, ...threadEntries]) {
|
||||
if (currentBytes <= maxBytes) break;
|
||||
currentBytes -= key.length + 3 + String(ts).length + 1;
|
||||
toEvict.push(key);
|
||||
}
|
||||
for (const key of toEvict) delete contexts[key];
|
||||
const fitsAfterTrim = encoder.encode(blobFor(contexts)).length <= maxBytes;
|
||||
return { evicted: toEvict.length, fitsAfterTrim };
|
||||
}
|
||||
@@ -99,6 +99,8 @@ test("writeStoredReadState prunes all three keys consistently", () => {
|
||||
]),
|
||||
new Map(),
|
||||
new Map(),
|
||||
new Map(),
|
||||
1,
|
||||
);
|
||||
|
||||
const state = JSON.parse(
|
||||
@@ -132,6 +134,8 @@ test("writeStoredReadState round-trips through readStoredReadState", () => {
|
||||
new Map([["channel-9", nowSeconds]]),
|
||||
new Map([["channel-9", { s: 3, c: 1, b: nowSeconds, f: nowSeconds }]]),
|
||||
new Map(),
|
||||
new Map(),
|
||||
1,
|
||||
);
|
||||
|
||||
const stored = readStoredReadState(pubkey);
|
||||
@@ -162,6 +166,8 @@ test("writeStoredReadState_survives_throwing_localStorage_and_returns_false", ()
|
||||
new Map([["channel-1", nowSeconds]]),
|
||||
new Map(),
|
||||
new Map(),
|
||||
new Map(),
|
||||
1,
|
||||
);
|
||||
});
|
||||
assert.equal(
|
||||
|
||||
@@ -11,7 +11,10 @@ import {
|
||||
type OverrideRegister,
|
||||
} from "@/features/channels/readState/readStateFormat";
|
||||
import { setLocalStorageItemWithRecovery } from "@/shared/lib/localStorageQuota";
|
||||
import type { PendingIntentOp } from "@/features/channels/pendingOverrideIntents";
|
||||
import type {
|
||||
PendingIntent,
|
||||
PendingIntentOp,
|
||||
} from "@/features/channels/pendingOverrideIntents";
|
||||
|
||||
export type StoredReadState = {
|
||||
contexts: Map<string, number>;
|
||||
@@ -19,6 +22,10 @@ export type StoredReadState = {
|
||||
contextSourceCreatedAt: Map<string, number>;
|
||||
overrideRegisters: Map<string, StoredOverrideEntry>;
|
||||
appliedReceipts: Map<string, AppliedReceipt>;
|
||||
/** Pending intents from the v2 blob (consolidated from pending-intents.v1). */
|
||||
pendingIntents: Map<string, PendingIntent>;
|
||||
/** Next generation counter for the pending intent queue. */
|
||||
nextGen: number;
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -124,6 +131,8 @@ export function readStoredReadState(pubkey: string): StoredReadState {
|
||||
contextSourceCreatedAt: readContextSourceCreatedAt(pubkey),
|
||||
overrideRegisters: overrideState.registers,
|
||||
appliedReceipts: overrideState.receipts,
|
||||
pendingIntents: overrideState.pendingIntents,
|
||||
nextGen: overrideState.nextGen,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -149,11 +158,15 @@ export type StoredOverrideEntry = OverrideRegister & { f: number };
|
||||
type OverrideStateBlob = {
|
||||
registers: Map<string, StoredOverrideEntry>;
|
||||
receipts: Map<string, AppliedReceipt>;
|
||||
pendingIntents: Map<string, PendingIntent>;
|
||||
nextGen: number;
|
||||
};
|
||||
|
||||
function readOverrideState(pubkey: string): OverrideStateBlob {
|
||||
const registers = new Map<string, StoredOverrideEntry>();
|
||||
const receipts = new Map<string, AppliedReceipt>();
|
||||
const pendingIntents = new Map<string, PendingIntent>();
|
||||
let nextGen = 1;
|
||||
const raw = localStorage.getItem(localOverrideStateKey(pubkey));
|
||||
if (raw) {
|
||||
try {
|
||||
@@ -162,8 +175,9 @@ function readOverrideState(pubkey: string): OverrideStateBlob {
|
||||
// Registers are stored under the "r" key (or top-level for compat).
|
||||
const registersObj = isPlainRecord(parsed.r) ? parsed.r : parsed;
|
||||
for (const [rawCtx, value] of Object.entries(registersObj)) {
|
||||
// Skip the reserved "receipts" key.
|
||||
if (rawCtx === "receipts") continue;
|
||||
// Skip reserved top-level keys.
|
||||
if (rawCtx === "receipts" || rawCtx === "pi" || rawCtx === "ng")
|
||||
continue;
|
||||
if (!isPlainRecord(value)) continue;
|
||||
const { s, c, b, f } = value;
|
||||
if (
|
||||
@@ -203,20 +217,55 @@ function readOverrideState(pubkey: string): OverrideStateBlob {
|
||||
receipts.set(channelId, { intentGen, op });
|
||||
}
|
||||
}
|
||||
// Parse pending intents from the "pi" sub-object.
|
||||
if (isPlainRecord(parsed.pi)) {
|
||||
for (const [channelId, entry] of Object.entries(parsed.pi)) {
|
||||
if (!isPlainRecord(entry)) continue;
|
||||
const { gen, op, sourceScope, readTarget } = entry as {
|
||||
gen?: unknown;
|
||||
op?: unknown;
|
||||
sourceScope?: unknown;
|
||||
readTarget?: unknown;
|
||||
};
|
||||
if (typeof gen !== "number" || !Number.isInteger(gen) || gen < 1)
|
||||
continue;
|
||||
if (op !== "unread" && op !== "read") continue;
|
||||
if (sourceScope !== undefined && typeof sourceScope !== "string")
|
||||
continue;
|
||||
if (readTarget !== undefined && typeof readTarget !== "number")
|
||||
continue;
|
||||
const intent: PendingIntent = {
|
||||
gen,
|
||||
op,
|
||||
...(sourceScope !== undefined ? { sourceScope } : {}),
|
||||
...(readTarget !== undefined ? { readTarget } : {}),
|
||||
};
|
||||
pendingIntents.set(channelId, intent);
|
||||
}
|
||||
}
|
||||
// Parse nextGen.
|
||||
if (
|
||||
typeof parsed.ng === "number" &&
|
||||
Number.isInteger(parsed.ng) &&
|
||||
parsed.ng >= 1
|
||||
) {
|
||||
nextGen = parsed.ng;
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Corrupt storage — return empty; repopulated on next ingest.
|
||||
}
|
||||
return { registers, receipts };
|
||||
return { registers, receipts, pendingIntents, nextGen };
|
||||
}
|
||||
// Migration: read from legacy v1 key if v2 key absent.
|
||||
const legacyRaw = localStorage.getItem(
|
||||
localOverrideRegistersKeyLegacy(pubkey),
|
||||
);
|
||||
if (!legacyRaw) return { registers, receipts };
|
||||
if (!legacyRaw) return { registers, receipts, pendingIntents, nextGen };
|
||||
try {
|
||||
const parsed = JSON.parse(legacyRaw);
|
||||
if (!isPlainRecord(parsed)) return { registers, receipts };
|
||||
if (!isPlainRecord(parsed))
|
||||
return { registers, receipts, pendingIntents, nextGen };
|
||||
for (const [rawCtx, value] of Object.entries(parsed)) {
|
||||
if (!isPlainRecord(value)) continue;
|
||||
const { s, c, b } = value;
|
||||
@@ -239,7 +288,7 @@ function readOverrideState(pubkey: string): OverrideStateBlob {
|
||||
} catch {
|
||||
// Corrupt legacy storage — ignore.
|
||||
}
|
||||
return { registers, receipts };
|
||||
return { registers, receipts, pendingIntents, nextGen };
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -282,6 +331,8 @@ export function writeStoredReadState(
|
||||
contextSourceCreatedAt: ReadonlyMap<string, number>,
|
||||
overrideRegisters: ReadonlyMap<string, StoredOverrideEntry>,
|
||||
appliedReceipts: ReadonlyMap<string, AppliedReceipt>,
|
||||
pendingIntents: ReadonlyMap<string, PendingIntent>,
|
||||
nextGen: number,
|
||||
): boolean {
|
||||
const pruned = pruneStaleContexts(contexts, Math.floor(Date.now() / 1_000));
|
||||
|
||||
@@ -290,17 +341,19 @@ export function writeStoredReadState(
|
||||
state[contextId] = new Date(timestamp * 1_000).toISOString();
|
||||
}
|
||||
|
||||
// Persist override registers atomically with their frontier timestamps and
|
||||
// applied receipts (Amendment C) in one JSON blob — a single write ensures
|
||||
// they are never torn. The receipt lives alongside the register it proves
|
||||
// was applied, so "register committed ↔ receipt committed" is a storage
|
||||
// invariant.
|
||||
// Persist override registers, applied receipts, and pending intents atomically
|
||||
// in one JSON blob — a single write ensures they are never torn.
|
||||
//
|
||||
// Format: { r: { <channelId>: {s,c,b,f}, ... }, receipts: { <channelId>: {intentGen,op}, ... } }
|
||||
// Format: {
|
||||
// r: { <channelId>: {s,c,b,f} },
|
||||
// receipts: { <channelId>: {intentGen, op} }, // omitted when empty
|
||||
// pi: { <channelId>: {gen, op, sourceScope?, readTarget?} }, // omitted when empty
|
||||
// ng: <number>, // next generation counter; omitted when 1
|
||||
// }
|
||||
//
|
||||
// This is the ACTION COMMIT POINT: if ok4 is true the action is durably
|
||||
// committed; ancillary frontier/cache write failures (ok1-ok3) do not fail
|
||||
// the action. If ok4 is false, the caller must roll back ALL state.
|
||||
// This is the ACTION + INTENT COMMIT POINT: if ok4 is true, the action,
|
||||
// receipt, and pending intent state are all durably committed together.
|
||||
// If ok4 is false, the caller must roll back ALL state.
|
||||
const registersObj: Record<
|
||||
string,
|
||||
{ s: number; c: number; b: number; f: number }
|
||||
@@ -315,11 +368,37 @@ export function writeStoredReadState(
|
||||
for (const [channelId, receipt] of appliedReceipts) {
|
||||
receiptsObj[channelId] = { intentGen: receipt.intentGen, op: receipt.op };
|
||||
}
|
||||
const piObj: Record<
|
||||
string,
|
||||
{
|
||||
gen: number;
|
||||
op: PendingIntentOp;
|
||||
sourceScope?: string;
|
||||
readTarget?: number;
|
||||
}
|
||||
> = {};
|
||||
for (const [channelId, intent] of pendingIntents) {
|
||||
const entry: {
|
||||
gen: number;
|
||||
op: PendingIntentOp;
|
||||
sourceScope?: string;
|
||||
readTarget?: number;
|
||||
} = {
|
||||
gen: intent.gen,
|
||||
op: intent.op,
|
||||
};
|
||||
if (intent.sourceScope !== undefined)
|
||||
entry.sourceScope = intent.sourceScope;
|
||||
if (intent.readTarget !== undefined) entry.readTarget = intent.readTarget;
|
||||
piObj[channelId] = entry;
|
||||
}
|
||||
const ok4 = setLocalStorageItemWithRecovery(
|
||||
localOverrideStateKey(pubkey),
|
||||
JSON.stringify({
|
||||
r: registersObj,
|
||||
...(Object.keys(receiptsObj).length > 0 ? { receipts: receiptsObj } : {}),
|
||||
...(Object.keys(piObj).length > 0 ? { pi: piObj } : {}),
|
||||
...(nextGen > 1 ? { ng: nextGen } : {}),
|
||||
}),
|
||||
);
|
||||
|
||||
|
||||
@@ -18,6 +18,7 @@ const noopMarkOverride = (): MarkResult => ({
|
||||
});
|
||||
const noopGetOverrideLiveness = (): OverrideLiveness | null => null;
|
||||
const noopGetProjection = (): ReadStateProjection | null => null;
|
||||
const noopSetOnDrainOutcome = (): void => {};
|
||||
|
||||
/**
|
||||
* React hook that creates and manages a ReadStateManager instance.
|
||||
@@ -115,14 +116,17 @@ export function useReadState(
|
||||
[],
|
||||
);
|
||||
|
||||
const markChannelRead = React.useCallback((channelId: string): MarkResult => {
|
||||
return (
|
||||
managerRef.current?.markChannelRead(channelId) ?? {
|
||||
status: "refused",
|
||||
reason: "load_incomplete",
|
||||
}
|
||||
);
|
||||
}, []);
|
||||
const markChannelRead = React.useCallback(
|
||||
(channelId: string, sourceScope?: string): MarkResult => {
|
||||
return (
|
||||
managerRef.current?.markChannelRead(channelId, sourceScope) ?? {
|
||||
status: "refused",
|
||||
reason: "load_incomplete",
|
||||
}
|
||||
);
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
const getOverrideLiveness = React.useCallback(
|
||||
(channelId: string): OverrideLiveness | null => {
|
||||
@@ -135,6 +139,27 @@ export function useReadState(
|
||||
return managerRef.current?.getProjection() ?? null;
|
||||
}, []);
|
||||
|
||||
// Wire the drain outcome callback into the current manager.
|
||||
// Called by useUnreadChannels to route rollback/cleanup through the hook layer.
|
||||
const setOnDrainOutcome = React.useCallback(
|
||||
(
|
||||
callback:
|
||||
| ((
|
||||
channelId: string,
|
||||
op: string,
|
||||
status: string,
|
||||
reason?: string,
|
||||
sourceScope?: string,
|
||||
) => void)
|
||||
| null,
|
||||
): void => {
|
||||
if (managerRef.current) {
|
||||
managerRef.current.onDrainOutcome = callback;
|
||||
}
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
const isReady = Boolean(
|
||||
pubkey && relayClient && initializedPubkey === pubkey,
|
||||
);
|
||||
@@ -160,6 +185,7 @@ export function useReadState(
|
||||
markChannelRead: noopMarkOverride,
|
||||
getOverrideLiveness: noopGetOverrideLiveness,
|
||||
getProjection: noopGetProjection,
|
||||
setOnDrainOutcome: noopSetOnDrainOutcome,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -177,5 +203,6 @@ export function useReadState(
|
||||
markChannelRead,
|
||||
getOverrideLiveness,
|
||||
getProjection,
|
||||
setOnDrainOutcome,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -52,7 +52,7 @@ export type OverrideAPIs = {
|
||||
* in a partial view", not "known absent register". */
|
||||
isReadStateReady: boolean;
|
||||
markChannelUnread: (channelId: string) => MarkResult;
|
||||
markChannelRead: (channelId: string) => MarkResult;
|
||||
markChannelRead: (channelId: string, sourceScope?: string) => MarkResult;
|
||||
getOverrideLiveness: (channelId: string) => OverrideLiveness | null;
|
||||
};
|
||||
|
||||
@@ -115,10 +115,15 @@ export type OverrideReadOutcome = "overrideCleared" | "overrideStillActive";
|
||||
* no error toast (representable C-bump was not possible, frontier
|
||||
* already provides the inactive verdict).
|
||||
* c. Other refusal keeping liveness active: show toast, overrideStillActive.
|
||||
*
|
||||
* `sourceScope` is forwarded to markChannelRead so the intent can carry the
|
||||
* exact source being cleared; the drain surfaces it to onDrainOutcome for
|
||||
* exact hook-layer source cleanup (rather than whole-entry deletion).
|
||||
*/
|
||||
export function applyOverrideRead(
|
||||
channelId: string,
|
||||
apis: OverrideAPIs,
|
||||
sourceScope?: string,
|
||||
): OverrideReadOutcome {
|
||||
// Step 1: load complete + no register → known absence, cleared.
|
||||
// Guard: only treat null as "known absent" when the load is complete.
|
||||
@@ -133,7 +138,7 @@ export function applyOverrideRead(
|
||||
// register or frontier mutation. Sources are preserved; the intent drains
|
||||
// on the next complete-load transition and rawUnread suppresses presentation
|
||||
// via the pending-read precedence rule.
|
||||
const result = apis.markChannelRead(channelId);
|
||||
const result = apis.markChannelRead(channelId, sourceScope);
|
||||
|
||||
if (result.status === "queued") {
|
||||
// Intent enqueued — sources preserved, rawUnread handles presentation.
|
||||
@@ -206,7 +211,7 @@ export function useClearChannelUnreadSource(
|
||||
return;
|
||||
}
|
||||
// Last-source removal: route through the mark-read seam.
|
||||
const outcome = applyOverrideRead(channelId, apis);
|
||||
const outcome = applyOverrideRead(channelId, apis, source);
|
||||
if (outcome === "overrideCleared") {
|
||||
delete forcedUnreadRef.current[channelId];
|
||||
if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
|
||||
|
||||
@@ -61,6 +61,7 @@ import {
|
||||
mutedStore,
|
||||
resolveChannelReadMarker,
|
||||
resolveObservedUnreadRootId,
|
||||
useDrainOutcomeCallback,
|
||||
} from "@/features/channels/useUnreadChannelsHelpers";
|
||||
|
||||
export function useUnreadChannels(
|
||||
@@ -82,10 +83,8 @@ export function useUnreadChannels(
|
||||
const normalizedRelayUrl = relayUrlOption
|
||||
? normalizeRelayUrl(relayUrlOption)
|
||||
: "";
|
||||
// Single identity for the in-memory thread-activity buffer — computed once
|
||||
// per render and used at reset, both writers, and the return fence. The
|
||||
// helper returns "" when either value is absent, which never matches a valid
|
||||
// loaded scope, so the fence returns [] until the buffer is seeded.
|
||||
// Single identity for the in-memory thread-activity buffer — reset/return fence;
|
||||
// empty string never matches a valid loaded scope, so the fence returns [] until seeded.
|
||||
const currentActivityScope = activityScopeKey(
|
||||
normalizedPubkey,
|
||||
normalizedRelayUrl,
|
||||
@@ -104,6 +103,7 @@ export function useUnreadChannels(
|
||||
markChannelRead: markChannelOverrideRead,
|
||||
getOverrideLiveness,
|
||||
getProjection,
|
||||
setOnDrainOutcome,
|
||||
} = useReadState(pubkey, relayClient);
|
||||
const overrideApis = React.useMemo<OverrideAPIs>(
|
||||
() => ({
|
||||
@@ -159,17 +159,12 @@ export function useUnreadChannels(
|
||||
}, [readStateVersion, drainSyncedAdvances]);
|
||||
|
||||
// Root event IDs of threads where the current user has replied at least once.
|
||||
// Used to determine if thread replies should trigger unread notifications.
|
||||
const participatedRootIdsRef = React.useRef(new Set<string>());
|
||||
// Root event IDs of top-level messages authored by the current user.
|
||||
// Used to notify the author when someone replies to their posts.
|
||||
const authoredRootIdsRef = React.useRef(new Set<string>());
|
||||
// Root event IDs of threads where an external message @-mentioned the user.
|
||||
// ORed into the badge gate so a mention recipient who never participated,
|
||||
// authored, or followed the thread still gets the thread-unread badge.
|
||||
// Root event IDs of threads @-mentioning the user — ORed into the badge gate.
|
||||
const mentionedRootIdsRef = React.useRef(new Set<string>());
|
||||
// Root event IDs of threads the user has explicitly muted. Takes precedence
|
||||
// over participation, follow, and authorship for notification suppression.
|
||||
// Root event IDs of threads the user muted — overrides participation/follow.
|
||||
const mutedRootIdsRef = React.useRef(new Set<string>());
|
||||
|
||||
// Stable ref for the caller-supplied muted channel IDs. Updated every render
|
||||
@@ -177,12 +172,9 @@ export function useUnreadChannels(
|
||||
const mutedChannelIdsRef = React.useRef<ReadonlySet<string>>(new Set());
|
||||
mutedChannelIdsRef.current = mutedChannelIdsOption ?? new Set();
|
||||
|
||||
// Thread reply events that triggered notifications — surfaced in the Home
|
||||
// activity feed as synthetic FeedItems.
|
||||
// Thread reply events that triggered notifications — surfaced in the Home activity feed.
|
||||
const threadActivityRef = React.useRef<ThreadActivityItem[]>([]);
|
||||
// Tracks the (pubkey:relayUrl) scope currently loaded into threadActivityRef.
|
||||
// Writers guard against this before merging so in-flight writes from a prior
|
||||
// scope cannot corrupt the new one; renders return [] until it matches.
|
||||
// Scope key for threadActivityRef; guards against cross-scope writes.
|
||||
const threadActivityScopeRef = React.useRef<string>("");
|
||||
|
||||
// Tracks which channels we've already issued a catch-up REQ for this
|
||||
@@ -204,6 +196,14 @@ export function useUnreadChannels(
|
||||
0,
|
||||
);
|
||||
|
||||
// Wire drain outcome callback; returns the pending-unread-snapshots ref.
|
||||
const pendingUnreadSnapshotsRef = useDrainOutcomeCallback(
|
||||
setOnDrainOutcome,
|
||||
forcedUnreadRef,
|
||||
pubkey,
|
||||
bumpLatestVersion,
|
||||
);
|
||||
|
||||
// Persistence layer: hydration, pagehide flush, scope fence, write-through, marker-prune.
|
||||
const observedPersistence = useObservedUnreadPersistence(
|
||||
normalizedPubkey,
|
||||
@@ -313,20 +313,26 @@ export function useUnreadChannels(
|
||||
|
||||
// Mark channel unread: write forced entry first (write ordering), then apply override.
|
||||
// Refused results undo the forced entry. Returns true when accepted (applied or queued).
|
||||
// biome-ignore lint/correctness/useExhaustiveDependencies: pendingUnreadSnapshotsRef is a stable ref; .current.set is Map.prototype.set (stable identity)
|
||||
const markChannelUnread = React.useCallback(
|
||||
(channelId: string, source?: ForcedUnreadSource): boolean => {
|
||||
// Step 1: write the forced entry first (write ordering).
|
||||
// Snapshot the prior entry before the optimistic write for drain rollback.
|
||||
const priorEntry = Object.hasOwn(forcedUnreadRef.current, channelId)
|
||||
? forcedUnreadRef.current[channelId]
|
||||
: undefined;
|
||||
// Write forced entry first (write ordering), then apply override.
|
||||
forcedMarkChannelUnread(channelId, source);
|
||||
// Step 2: apply override (may enqueue intent or apply immediately).
|
||||
if (!applyOverrideUnread(channelId, overrideApis)) {
|
||||
// Refused — undo the forced entry.
|
||||
if (Object.hasOwn(forcedUnreadRef.current, channelId)) {
|
||||
delete forcedUnreadRef.current[channelId];
|
||||
if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
|
||||
bumpLatestVersion();
|
||||
}
|
||||
// Refused immediately — restore exact prior entry (or remove if none existed).
|
||||
if (priorEntry !== undefined)
|
||||
forcedUnreadRef.current[channelId] = priorEntry;
|
||||
else delete forcedUnreadRef.current[channelId];
|
||||
if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
|
||||
bumpLatestVersion();
|
||||
return false;
|
||||
}
|
||||
// Store snapshot; drain will call onDrainOutcome to commit or rollback.
|
||||
pendingUnreadSnapshotsRef.current.set(channelId, priorEntry);
|
||||
return true;
|
||||
},
|
||||
[forcedMarkChannelUnread, overrideApis, pubkey],
|
||||
@@ -990,10 +996,8 @@ export function useUnreadChannels(
|
||||
markChannelRead,
|
||||
markChannelUnread,
|
||||
clearChannelUnreadSource,
|
||||
// Exposed so other surfaces (e.g. Home) can project per-item read state
|
||||
// off the same NIP-RS read marker without instantiating a second
|
||||
// ReadStateManager. readStateVersion is the invalidation signal callers
|
||||
// should include in memo deps.
|
||||
// Exposed for other surfaces (e.g. Home) to project read state off the same
|
||||
// NIP-RS marker; readStateVersion is the invalidation signal for memo deps.
|
||||
getEffectiveTimestamp,
|
||||
getOwnTimestamp,
|
||||
readStateVersion,
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
* useUnreadChannels to keep that file within the repository size ratchet.
|
||||
* All exports are load-bearing for useUnreadChannels.ts.
|
||||
*/
|
||||
import * as React from "react";
|
||||
import type { UseLiveChannelUpdatesOptions } from "@/features/channels/useLiveChannelUpdates";
|
||||
import { makeRootIdStore } from "@/features/channels/unreadRootIdStore";
|
||||
import {
|
||||
@@ -13,6 +14,13 @@ import type { RelayClient } from "@/shared/api/relayClientSession";
|
||||
import type { Channel } from "@/shared/api/types";
|
||||
import { CHANNEL_MESSAGE_EVENT_KINDS } from "@/shared/constants/kinds";
|
||||
import { DM_NOTIFIABLE_EVENT_KINDS } from "./isDmNotifiableKind";
|
||||
import {
|
||||
forcedUnreadStore,
|
||||
removeForcedUnreadSource,
|
||||
type ForcedUnreadEntry,
|
||||
type ForcedUnreadMap,
|
||||
type ForcedUnreadSource,
|
||||
} from "@/features/channels/forcedUnreadStore";
|
||||
|
||||
export type UseUnreadChannelsOptions = UseLiveChannelUpdatesOptions & {
|
||||
pubkey?: string;
|
||||
@@ -87,3 +95,84 @@ export function resolveChannelReadMarker(
|
||||
export function resolveObservedUnreadRootId(tags: string[][]): string | null {
|
||||
return isBroadcastReply(tags) ? null : getThreadReference(tags).rootId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hook that builds and wires the drain outcome callback into the ReadStateManager
|
||||
* via `setOnDrainOutcome`. Creates and owns the pending-unread-snapshots map.
|
||||
* Returns the snapshots ref for use in markChannelUnread.
|
||||
*
|
||||
* Handles:
|
||||
* - `unread + refused` → restore forced-unread entry to its pre-mark snapshot
|
||||
* - `unread + applied` → discard snapshot (intent succeeded)
|
||||
* - `read + applied` → remove forced-unread entry (or specific source), mirroring
|
||||
* the synchronous overrideCleared path
|
||||
*
|
||||
* Genuine refusal toasts are fired in the drain itself and not repeated here.
|
||||
*/
|
||||
export function useDrainOutcomeCallback(
|
||||
setOnDrainOutcome: (
|
||||
cb:
|
||||
| ((
|
||||
channelId: string,
|
||||
op: string,
|
||||
status: string,
|
||||
reason?: string,
|
||||
sourceScope?: string,
|
||||
) => void)
|
||||
| null,
|
||||
) => void,
|
||||
forcedUnreadRef: React.MutableRefObject<ForcedUnreadMap>,
|
||||
pubkey: string | undefined,
|
||||
bumpLatestVersion: () => void,
|
||||
): React.MutableRefObject<Map<string, ForcedUnreadEntry | undefined>> {
|
||||
const pendingUnreadSnapshotsRef = React.useRef(
|
||||
new Map<string, ForcedUnreadEntry | undefined>(),
|
||||
);
|
||||
// biome-ignore lint/correctness/useExhaustiveDependencies: stable refs; setOnDrainOutcome is stable
|
||||
React.useEffect(() => {
|
||||
setOnDrainOutcome((channelId, op, status, _reason, sourceScope) => {
|
||||
if (op === "unread" && status === "refused") {
|
||||
// Roll back the optimistic forced-unread entry to its exact prior state.
|
||||
const prior = pendingUnreadSnapshotsRef.current.get(channelId);
|
||||
pendingUnreadSnapshotsRef.current.delete(channelId);
|
||||
if (prior === undefined) {
|
||||
if (Object.hasOwn(forcedUnreadRef.current, channelId)) {
|
||||
delete forcedUnreadRef.current[channelId];
|
||||
if (pubkey)
|
||||
forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
|
||||
bumpLatestVersion();
|
||||
}
|
||||
} else {
|
||||
forcedUnreadRef.current[channelId] = prior;
|
||||
if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
|
||||
bumpLatestVersion();
|
||||
}
|
||||
} else if (op === "unread" && status === "applied") {
|
||||
pendingUnreadSnapshotsRef.current.delete(channelId);
|
||||
} else if (op === "read" && status === "applied") {
|
||||
if (Object.hasOwn(forcedUnreadRef.current, channelId)) {
|
||||
if (sourceScope !== undefined) {
|
||||
const current = forcedUnreadRef.current[channelId];
|
||||
const next = removeForcedUnreadSource(
|
||||
current,
|
||||
sourceScope as ForcedUnreadSource,
|
||||
);
|
||||
if (next !== undefined) {
|
||||
forcedUnreadRef.current[channelId] = next;
|
||||
} else {
|
||||
delete forcedUnreadRef.current[channelId];
|
||||
}
|
||||
} else {
|
||||
delete forcedUnreadRef.current[channelId];
|
||||
}
|
||||
if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
|
||||
bumpLatestVersion();
|
||||
}
|
||||
}
|
||||
});
|
||||
return () => {
|
||||
setOnDrainOutcome(null);
|
||||
};
|
||||
}, [pubkey, setOnDrainOutcome]);
|
||||
return pendingUnreadSnapshotsRef;
|
||||
}
|
||||
|
||||
@@ -248,7 +248,7 @@
|
||||
}
|
||||
|
||||
.buzz-terminal-substrate[data-terminal-visible="false"] {
|
||||
height: 0 !important;
|
||||
height: 0;
|
||||
min-height: 0;
|
||||
pointer-events: none;
|
||||
transform: translateY(16px);
|
||||
|
||||
@@ -536,13 +536,20 @@ Every action below requires a complete full-state load (see Full-State Load). On
|
||||
|
||||
- (a) **`applied`** — the action was applied against a complete full-state load; the register was mutated, the canonical form was updated, and publication was scheduled.
|
||||
- (b) **`queued`** — the action was accepted as a pending intent; the register was NOT mutated; a forced local presentation MAY be shown optimistically; the client MUST drain the intent on the next complete-load transition and surface the definitive `applied` or `refused` result then.
|
||||
- (c) **`refused`** — the action was rejected; neither the register nor local presentation was modified; a reason MUST be provided (see reasons below).
|
||||
- (c) **`refused`** — the action was rejected; neither the register nor local presentation was modified; a reason MUST be provided (one of the refusal reasons below).
|
||||
|
||||
**Refusal reasons:**
|
||||
|
||||
- `load_incomplete` — the full-state load was not proven complete; no register mutation was performed.
|
||||
- `already_inactive` — the override register was already inactive (or absent) at the time of the mark-read; no C-bump was needed. This is a definitive success condition: the channel is already cleared.
|
||||
- `uint32_overflow` — `max(S, C)` has reached the uint32 maximum (4294967295); no representable counter increment exists.
|
||||
- `budget_exhausted` — the new register would exceed the encoding budget; the client MUST refuse rather than silently drop other channels' overrides.
|
||||
|
||||
A client that does not implement pending intents MUST report the action as `refused` with reason `load_incomplete` rather than act on a partial view of its own override state.
|
||||
|
||||
**Mark-unread:** increment S to `max(S, C) + 1`; set B to the current effective frontier value for the context. C is unchanged. If `max(S, C) == 4294967295` (uint32 maximum), the client MUST refuse the mark-unread action and leave the register unchanged; wrapping or resetting to zero is prohibited.
|
||||
|
||||
**Mark-read (explicit):** advance the frontier to cover the context as normal; increment C to `max(S, C) + 1`. S and B are unchanged. If `max(S, C) == 4294967295`, no representable counter increment exists; wrapping or resetting to zero is prohibited. The client MUST then complete the action only if the resulting state satisfies `override_active == false` — i.e. the frontier advance alone deactivates the override, or the override was already inactive. Otherwise the counters MUST be left unchanged and the client MUST NOT report the mark-read as `applied` while `override_active` remains true; instead the client MUST report the outcome as `refused` (see reason `already_inactive` and other reasons above). A client MAY report `queued` for a mark-read on an incomplete load per (b) above, but MUST NOT apply the counter mutation until a complete load is available.
|
||||
**Mark-read (explicit):** advance the frontier to cover the context as normal; increment C to `max(S, C) + 1`. S and B are unchanged. If `max(S, C) == 4294967295`, no representable counter increment exists; wrapping or resetting to zero is prohibited. The client MUST then complete the action only if the resulting state satisfies `override_active == false` — i.e. the frontier advance alone deactivates the override, or the override was already inactive. Otherwise the counters MUST be left unchanged and the client MUST NOT report the mark-read as `applied` while `override_active` remains true; instead the client MUST report the outcome as `refused` with reason `already_inactive`. A client MAY report `queued` for a mark-read on an incomplete load per (b) above, but MUST NOT apply the counter mutation until a complete load is available.
|
||||
|
||||
**Natural read (frontier advance):** advance the frontier past B. No counter update is needed — the liveness predicate's `F <= B` condition automatically deactivates the override when the frontier dominates the baseline.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user