mirror of
https://github.com/block/buzz.git
synced 2026-11-01 06:51:17 +00:00
fix(nip-rs): close pass-3 findings — projection-based observer, loadComplete ready gate, evidence retention
CRITICAL 1 (stale-stored + fetched tombstone resurrection): Replace communityUnreadObserver's stored+fetched componentwise max join with getProjection(). When loadComplete, projection.overrides is the sole authoritative override source — no per-field merge with fetched registers that can resurrect a superseded tombstone. effectiveFrontier() = max(fetched, projection) per channel. CRITICAL 2 (register evaluated against F=0 instead of real frontier): effectiveFrontierMap now merges projection.frontiers with fetched frontiers using max() per channel, so isUnreadExternalEvent always evaluates against the true committed frontier, not a zero default. IMPORTANT 1 (isReadStateReady = initialized, not loadComplete): Add isLoadComplete return from useReadState (manager.getProjection().loadComplete). Wire overrideApis.isReadStateReady to isLoadComplete — null liveness after an incomplete load is ambiguous, not known absence. readStateOverride.ts JSDoc updated. IMPORTANT 2 (markAllChannelsRead unconditionally wipes evidence): latestByChannelRef.delete and observedUnreadEventsByChannelRef.delete now inside the overrideCleared branch only — refused marks preserve the channel's observed evidence. IMPORTANT 3 (adversarial witnesses): Four tests added: - adversarial-1: stale live register + fetched tombstone → dark (no resurrection) - adversarial-2: projection frontier deactivates register → dark - adversarial-3: incomplete-load null liveness ≠ known absent → fail-closed - adversarial-4: mark-all refusal keeps per-channel observed evidence Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
parent
0ab6e129d5
commit
58f5feb289
@@ -508,3 +508,123 @@ test("applyOverrideRead_partial_refusal_pattern_clears_only_inactive", () => {
|
||||
);
|
||||
assert.equal(clearCallCount, 2, "C-bump attempted for both channels");
|
||||
});
|
||||
|
||||
// ── Adversarial witnesses (pass-3 closures) ───────────────────────────────────
|
||||
|
||||
test("adversarial-3: incomplete_load_null_liveness_not_known_absent_overrideStillActive", () => {
|
||||
// IMPORTANT: isReadStateReady = false (load incomplete) + null getOverrideLiveness
|
||||
// must NOT be treated as known register absence. The old bug: isReadStateReady was
|
||||
// set on initialization (not load-complete), so a partially-loaded manager reported
|
||||
// null liveness which was silently treated as "no register" and the local hint was
|
||||
// cleared. With isReadStateReady now meaning loadComplete, this path is fail-closed.
|
||||
//
|
||||
// This test drives applyOverrideRead with isReadStateReady=false (semantically:
|
||||
// load is not complete) and null liveness — must return overrideStillActive.
|
||||
let markCalled = false;
|
||||
const result = applyOverrideRead(
|
||||
"ch-1",
|
||||
makeApis({
|
||||
isReadStateReady: false, // load not complete
|
||||
markChannelRead: () => {
|
||||
markCalled = true;
|
||||
return { success: false, reason: "load_incomplete" };
|
||||
},
|
||||
getOverrideLiveness: () => null, // null from incomplete load ≠ known absent
|
||||
}),
|
||||
);
|
||||
assert.equal(
|
||||
result,
|
||||
"overrideStillActive",
|
||||
"incomplete load must fail closed",
|
||||
);
|
||||
assert.equal(
|
||||
markCalled,
|
||||
false,
|
||||
"must not call manager when load not complete",
|
||||
);
|
||||
});
|
||||
|
||||
test("adversarial-4: markAllChannelsRead_confirmed_active_refusal_keeps_evidence", () => {
|
||||
// IMPORTANT: markAllChannelsRead must NOT clear latestByChannelRef or
|
||||
// observedUnreadEventsByChannelRef when the override read is refused and
|
||||
// liveness remains active. Evidence must be preserved so the UI can still
|
||||
// represent the failed channel's unread state.
|
||||
//
|
||||
// This test uses applyOverrideRead directly to simulate the per-channel loop
|
||||
// in markAllChannelsRead and verifies that refused channels keep their evidence
|
||||
// while cleared channels discard theirs.
|
||||
const livenessMap = {
|
||||
"ch-cleared": { active: true, frontier: 100 },
|
||||
"ch-refused": { active: true, frontier: 100 }, // uint32_overflow refusal
|
||||
};
|
||||
|
||||
const apis = {
|
||||
isReadStateReady: true,
|
||||
markChannelUnread: () => ({ success: true }),
|
||||
markChannelRead: (id) => {
|
||||
if (id === "ch-cleared") {
|
||||
livenessMap["ch-cleared"] = { active: false, frontier: 101 };
|
||||
return { success: true };
|
||||
}
|
||||
// ch-refused: refuse; liveness stays active
|
||||
return { success: false, reason: "uint32_overflow" };
|
||||
},
|
||||
getOverrideLiveness: (id) => livenessMap[id] ?? null,
|
||||
};
|
||||
|
||||
// latestByChannel and observedEvents are the evidence maps.
|
||||
const latestByChannel = new Map([
|
||||
["ch-cleared", 200],
|
||||
["ch-refused", 200],
|
||||
]);
|
||||
const observedEvents = new Map([
|
||||
["ch-cleared", new Map([["ev1", { kind: 9, createdAt: 200 }]])],
|
||||
["ch-refused", new Map([["ev2", { kind: 9, createdAt: 200 }]])],
|
||||
]);
|
||||
const forcedMap = { "ch-cleared": 99, "ch-refused": 99 };
|
||||
|
||||
// Simulate the production per-channel loop in markAllChannelsRead.
|
||||
for (const channelId of ["ch-cleared", "ch-refused"]) {
|
||||
const outcome = applyOverrideRead(channelId, apis);
|
||||
if (outcome === "overrideCleared") {
|
||||
delete forcedMap[channelId];
|
||||
latestByChannel.delete(channelId);
|
||||
observedEvents.delete(channelId);
|
||||
}
|
||||
// On overrideStillActive: do NOT touch latestByChannel or observedEvents.
|
||||
}
|
||||
|
||||
// ch-cleared: override confirmed inactive → evidence discarded.
|
||||
assert.equal(
|
||||
latestByChannel.has("ch-cleared"),
|
||||
false,
|
||||
"cleared channel: latestByChannel removed",
|
||||
);
|
||||
assert.equal(
|
||||
observedEvents.has("ch-cleared"),
|
||||
false,
|
||||
"cleared channel: observedEvents removed",
|
||||
);
|
||||
assert.equal(
|
||||
Object.hasOwn(forcedMap, "ch-cleared"),
|
||||
false,
|
||||
"cleared channel: forcedMap entry removed",
|
||||
);
|
||||
|
||||
// ch-refused: override still active → evidence preserved.
|
||||
assert.equal(
|
||||
latestByChannel.has("ch-refused"),
|
||||
true,
|
||||
"refused channel: latestByChannel preserved",
|
||||
);
|
||||
assert.equal(
|
||||
observedEvents.has("ch-refused"),
|
||||
true,
|
||||
"refused channel: observedEvents preserved",
|
||||
);
|
||||
assert.equal(
|
||||
Object.hasOwn(forcedMap, "ch-refused"),
|
||||
true,
|
||||
"refused channel: forcedMap entry preserved",
|
||||
);
|
||||
});
|
||||
|
||||
@@ -3,6 +3,7 @@ import {
|
||||
ReadStateManager,
|
||||
type ContextParentResolver,
|
||||
type MarkResult,
|
||||
type ReadStateProjection,
|
||||
} from "@/features/channels/readState/readStateManager";
|
||||
import type { OverrideLiveness } from "@/features/channels/readState/readStateFormat";
|
||||
import type { RelayClient } from "@/shared/api/relayClientSession";
|
||||
@@ -16,6 +17,7 @@ const noopMarkOverride = (): MarkResult => ({
|
||||
reason: "load_incomplete",
|
||||
});
|
||||
const noopGetOverrideLiveness = (): OverrideLiveness | null => null;
|
||||
const noopGetProjection = (): ReadStateProjection | null => null;
|
||||
|
||||
/**
|
||||
* React hook that creates and manages a ReadStateManager instance.
|
||||
@@ -129,14 +131,25 @@ export function useReadState(
|
||||
[],
|
||||
);
|
||||
|
||||
const getProjection = React.useCallback((): ReadStateProjection | null => {
|
||||
return managerRef.current?.getProjection() ?? null;
|
||||
}, []);
|
||||
|
||||
const isReady = Boolean(
|
||||
pubkey && relayClient && initializedPubkey === pubkey,
|
||||
);
|
||||
|
||||
// loadComplete is true only when the manager's full-state enumeration has
|
||||
// terminated with a complete verdict. Distinct from isReady (initialized):
|
||||
// a manager can be initialized with an incomplete load.
|
||||
const isLoadComplete =
|
||||
isReady && (managerRef.current?.getProjection().loadComplete ?? false);
|
||||
|
||||
if (!pubkey || !relayClient) {
|
||||
return {
|
||||
getEffectiveTimestamp: noopGetTimestamp,
|
||||
isReady: false,
|
||||
isLoadComplete: false,
|
||||
markContextRead: noopMarkRead,
|
||||
seedContextRead: noopMarkRead,
|
||||
drainSyncedAdvances: noopDrainAdvances,
|
||||
@@ -146,12 +159,14 @@ export function useReadState(
|
||||
markChannelUnread: noopMarkOverride,
|
||||
markChannelRead: noopMarkOverride,
|
||||
getOverrideLiveness: noopGetOverrideLiveness,
|
||||
getProjection: noopGetProjection,
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
getEffectiveTimestamp,
|
||||
isReady,
|
||||
isLoadComplete,
|
||||
markContextRead,
|
||||
seedContextRead,
|
||||
drainSyncedAdvances,
|
||||
@@ -161,5 +176,6 @@ export function useReadState(
|
||||
markChannelUnread,
|
||||
markChannelRead,
|
||||
getOverrideLiveness,
|
||||
getProjection,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -42,6 +42,10 @@ export function overrideErrorMessage(
|
||||
|
||||
/** Slice-2 override APIs from ReadStateManager, proxied through useReadState. */
|
||||
export type OverrideAPIs = {
|
||||
/** True only when the manager's full-state load is complete (loadComplete).
|
||||
* A manager that finished initializing with an incomplete load must NOT be
|
||||
* treated as ready — null liveness after an incomplete load means "not found
|
||||
* in a partial view", not "known absent register". */
|
||||
isReadStateReady: boolean;
|
||||
markChannelUnread: (channelId: string) => MarkResult;
|
||||
markChannelRead: (channelId: string) => MarkResult;
|
||||
@@ -84,10 +88,11 @@ export type OverrideReadOutcome = "overrideCleared" | "overrideStillActive";
|
||||
* Attempt to clear the NIP-RS override for `channelId` as part of an explicit
|
||||
* mark-read transition. Models the transition as a single outcome:
|
||||
*
|
||||
* 1. Manager unavailable (!isReadStateReady): fail-closed —
|
||||
* return overrideStillActive (manager unavailable ≠ no register).
|
||||
* 2. liveness === null (ready but no register): return overrideCleared
|
||||
* (known absence; no C-bump needed).
|
||||
* 1. Load not complete (!isReadStateReady): fail-closed —
|
||||
* return overrideStillActive (null liveness after an incomplete load is
|
||||
* ambiguous, not known absence).
|
||||
* 2. liveness === null (load complete, no register): return overrideCleared
|
||||
* (known absence after a complete load; no C-bump needed).
|
||||
* 3. liveness exists (active or inactive): always call markChannelRead —
|
||||
* spec NIP-RS:537-539 requires the C-bump even when the frontier already
|
||||
* deactivated the override. Then re-read liveness.
|
||||
@@ -106,7 +111,7 @@ export function applyOverrideRead(
|
||||
|
||||
const liveness = apis.getOverrideLiveness(channelId);
|
||||
|
||||
// Step 2: ready but no register at all → known absence, cleared.
|
||||
// Step 2: load complete but no register at all → known absence, cleared.
|
||||
if (liveness === null) return "overrideCleared";
|
||||
|
||||
// Step 3: register exists — always attempt the C-bump (NIP-RS:537-539).
|
||||
|
||||
@@ -160,10 +160,10 @@ export function useUnreadChannels(
|
||||
normalizedRelayUrl,
|
||||
);
|
||||
|
||||
// Slice-2 override APIs — now first-class on useReadState's return.
|
||||
const {
|
||||
getEffectiveTimestamp,
|
||||
isReady: isReadStateReady,
|
||||
isLoadComplete,
|
||||
markContextRead,
|
||||
drainSyncedAdvances,
|
||||
setContextParentResolver,
|
||||
@@ -175,14 +175,14 @@ export function useUnreadChannels(
|
||||
} = useReadState(pubkey, relayClient);
|
||||
const overrideApis = React.useMemo<OverrideAPIs>(
|
||||
() => ({
|
||||
isReadStateReady,
|
||||
isReadStateReady: isLoadComplete, // loadComplete, not initialized — null after incomplete load ≠ known absent
|
||||
markChannelUnread: markChannelOverrideUnread,
|
||||
markChannelRead: markChannelOverrideRead,
|
||||
getOverrideLiveness,
|
||||
}),
|
||||
[
|
||||
getOverrideLiveness,
|
||||
isReadStateReady,
|
||||
isLoadComplete,
|
||||
markChannelOverrideRead,
|
||||
markChannelOverrideUnread,
|
||||
],
|
||||
@@ -962,9 +962,9 @@ export function useUnreadChannels(
|
||||
}
|
||||
if (applyOverrideRead(channelId, overrideApis) === "overrideCleared") {
|
||||
delete forcedUnreadRef.current[channelId];
|
||||
latestByChannelRef.current.delete(channelId);
|
||||
observedUnreadEventsByChannelRef.current.delete(channelId);
|
||||
}
|
||||
latestByChannelRef.current.delete(channelId);
|
||||
observedUnreadEventsByChannelRef.current.delete(channelId);
|
||||
}
|
||||
if (pubkey) {
|
||||
forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
|
||||
|
||||
@@ -728,11 +728,11 @@ function quietRelayWithReadState(readAtSeconds) {
|
||||
}
|
||||
|
||||
test("fetchCommunityUnread stored active register lights rail (hasUnread:true)", async () => {
|
||||
// Authoritative source (a): stored register is active (S=1,C=0,B=0 with F=0).
|
||||
// The fetch returns no override events — stored register is the authority.
|
||||
// Authoritative source: complete projection with active register (S=1,C=0,B=0 with F=0).
|
||||
// The fetch returns no override events — projection is the authority.
|
||||
const relay = quietRelay();
|
||||
// Active register: S=1, C=0, B=0 (baseline), frontier=0 → override_active = true
|
||||
const storedRegisters = new Map([[CHANNEL_ID, { s: 1, c: 0, b: 0 }]]);
|
||||
const overrides = new Map([[CHANNEL_ID, { s: 1, c: 0, b: 0 }]]);
|
||||
|
||||
const result = await fetchCommunityUnread({
|
||||
client: relay,
|
||||
@@ -741,18 +741,22 @@ test("fetchCommunityUnread stored active register lights rail (hasUnread:true)",
|
||||
decryptReadState: async (v) => v,
|
||||
decryptMutes: async (v) => v,
|
||||
readThreadRelationships: readRelationships(),
|
||||
readStoredRegisters: () => storedRegisters,
|
||||
getProjection: () => ({
|
||||
loadComplete: true,
|
||||
frontiers: new Map(),
|
||||
overrides,
|
||||
}),
|
||||
});
|
||||
|
||||
assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
|
||||
});
|
||||
|
||||
test("fetchCommunityUnread stored tombstoned register does NOT light rail (hasUnread:false)", async () => {
|
||||
// Authoritative source (a): stored register is tombstoned (S=1,C=1 → S=C, inactive).
|
||||
// Local cache cannot contradict this — must not light the dot.
|
||||
// Authoritative source: complete projection with tombstoned register (S=1,C=1 → inactive).
|
||||
// Must not light the dot.
|
||||
const relay = quietRelay();
|
||||
// Tombstoned: S=1, C=1, B=0 → override_active = false (S > C fails: 1 > 1 = false)
|
||||
const storedRegisters = new Map([[CHANNEL_ID, { s: 1, c: 1, b: 0 }]]);
|
||||
const overrides = new Map([[CHANNEL_ID, { s: 1, c: 1, b: 0 }]]);
|
||||
|
||||
const result = await fetchCommunityUnread({
|
||||
client: relay,
|
||||
@@ -761,7 +765,11 @@ test("fetchCommunityUnread stored tombstoned register does NOT light rail (hasUn
|
||||
decryptReadState: async (v) => v,
|
||||
decryptMutes: async (v) => v,
|
||||
readThreadRelationships: readRelationships(),
|
||||
readStoredRegisters: () => storedRegisters,
|
||||
getProjection: () => ({
|
||||
loadComplete: true,
|
||||
frontiers: new Map(),
|
||||
overrides,
|
||||
}),
|
||||
});
|
||||
|
||||
assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
|
||||
@@ -770,11 +778,11 @@ test("fetchCommunityUnread stored tombstoned register does NOT light rail (hasUn
|
||||
test("fetchCommunityUnread old active register beyond horizon still lights rail", async () => {
|
||||
// An active register that is older than the 7-day horizon must still light
|
||||
// the rail. The fetch is tag-free and has no `since`, so it returns no
|
||||
// events here, but the stored projection (authoritative source a) is used.
|
||||
// events here, but the complete projection is the authority.
|
||||
const relay = quietRelay();
|
||||
const VERY_OLD_FRONTIER = 1; // far beyond any 7-day horizon
|
||||
// S=1, C=0, B=VERY_OLD_FRONTIER, and frontier=VERY_OLD_FRONTIER → active
|
||||
const storedRegisters = new Map([
|
||||
const overrides = new Map([
|
||||
[CHANNEL_ID, { s: 1, c: 0, b: VERY_OLD_FRONTIER }],
|
||||
]);
|
||||
|
||||
@@ -785,7 +793,11 @@ test("fetchCommunityUnread old active register beyond horizon still lights rail"
|
||||
decryptReadState: async (v) => v,
|
||||
decryptMutes: async (v) => v,
|
||||
readThreadRelationships: readRelationships(),
|
||||
readStoredRegisters: () => storedRegisters,
|
||||
getProjection: () => ({
|
||||
loadComplete: true,
|
||||
frontiers: new Map(),
|
||||
overrides,
|
||||
}),
|
||||
});
|
||||
|
||||
assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
|
||||
@@ -837,7 +849,7 @@ test("fetchCommunityUnread no stored register and empty fetch → falls through
|
||||
decryptReadState: async (v) => v,
|
||||
decryptMutes: async (v) => v,
|
||||
readThreadRelationships: readRelationships(),
|
||||
readStoredRegisters: () => new Map(), // no stored registers
|
||||
// No projection — no override registers; falls through to relay evidence gate
|
||||
});
|
||||
|
||||
assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
|
||||
@@ -856,7 +868,11 @@ test("fetchCommunityUnread channel not in member list → hasUnread:false", asyn
|
||||
decryptReadState: async (v) => v,
|
||||
decryptMutes: async (v) => v,
|
||||
readThreadRelationships: readRelationships(),
|
||||
readStoredRegisters: () => new Map([[CHANNEL_ID, { s: 1, c: 0, b: 0 }]]),
|
||||
getProjection: () => ({
|
||||
loadComplete: true,
|
||||
frontiers: new Map(),
|
||||
overrides: new Map([[CHANNEL_ID, { s: 1, c: 0, b: 0 }]]),
|
||||
}),
|
||||
});
|
||||
|
||||
assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
|
||||
@@ -905,7 +921,11 @@ test("fetchCommunityUnread active register muted channel → hasUnread:false", a
|
||||
decryptMutes: async (v) => v,
|
||||
readThreadRelationships: readRelationships(),
|
||||
// Active register, but channel is muted — mute wins
|
||||
readStoredRegisters: () => new Map([[CHANNEL_ID, { s: 1, c: 0, b: 0 }]]),
|
||||
getProjection: () => ({
|
||||
loadComplete: true,
|
||||
frontiers: new Map(),
|
||||
overrides: new Map([[CHANNEL_ID, { s: 1, c: 0, b: 0 }]]),
|
||||
}),
|
||||
});
|
||||
|
||||
assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
|
||||
@@ -916,7 +936,7 @@ test("fetchCommunityUnread frontier advance deactivates stored register → hasU
|
||||
// Fetched frontier: 100 > 50 → override_active = false.
|
||||
const relay = quietRelayWithReadState(100);
|
||||
// B=50 means active only when F<=50; fetched frontier is 100 → inactive
|
||||
const storedRegisters = new Map([[CHANNEL_ID, { s: 1, c: 0, b: 50 }]]);
|
||||
const overrides = new Map([[CHANNEL_ID, { s: 1, c: 0, b: 50 }]]);
|
||||
|
||||
const result = await fetchCommunityUnread({
|
||||
client: relay,
|
||||
@@ -925,7 +945,117 @@ test("fetchCommunityUnread frontier advance deactivates stored register → hasU
|
||||
decryptReadState: async (v) => v,
|
||||
decryptMutes: async (v) => v,
|
||||
readThreadRelationships: readRelationships(),
|
||||
readStoredRegisters: () => storedRegisters,
|
||||
getProjection: () => ({
|
||||
loadComplete: true,
|
||||
frontiers: new Map(),
|
||||
overrides,
|
||||
}),
|
||||
});
|
||||
|
||||
assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
|
||||
});
|
||||
|
||||
// ── Adversarial witness tests ─────────────────────────────────────────────────
|
||||
//
|
||||
// These witnesses verify that the rail's override evaluation cannot resurrect
|
||||
// a superseded register or produce a false verdict from a torn/partial state.
|
||||
|
||||
test("fetchCommunityUnread adversarial-1: stale live register + fetched tombstone → dark when no complete projection", async () => {
|
||||
// The OLD per-field max join would: max(stale_live, tombstone) = live (resurrection).
|
||||
// Without a complete projection, the fetched coordinate-deduped tombstone is
|
||||
// authoritative; the register is inactive and the rail must remain dark.
|
||||
//
|
||||
// Setup: fetched event carries tombstone (S=0,C=4,B=0) for CHANNEL_ID.
|
||||
const SLOT = "a".repeat(32); // valid 32-hex slot
|
||||
const relay = relayFor([
|
||||
// 1. member events
|
||||
() => [
|
||||
event({
|
||||
tags: [
|
||||
["d", CHANNEL_ID],
|
||||
["p", PUBKEY],
|
||||
],
|
||||
}),
|
||||
],
|
||||
// 2. metadata
|
||||
() => [
|
||||
event({
|
||||
tags: [
|
||||
["d", CHANNEL_ID],
|
||||
["t", "stream"],
|
||||
],
|
||||
}),
|
||||
],
|
||||
// 3. visibility
|
||||
() => [],
|
||||
// 4. read-state: tombstone event (S=0, C=4, B=0 → inactive, S not > C)
|
||||
() => [
|
||||
event({
|
||||
pubkey: PUBKEY,
|
||||
created_at: 300,
|
||||
tags: [
|
||||
["d", `read-state:${SLOT}`],
|
||||
["t", "read-state"],
|
||||
],
|
||||
// Contexts blob: channel has tombstone register (ov_s=0 absent means S=0, ov_c=4)
|
||||
content: JSON.stringify({
|
||||
v: 1,
|
||||
client_id: "client",
|
||||
contexts: {
|
||||
[CHANNEL_ID]: 50,
|
||||
[`ov_c:${CHANNEL_ID}`]: 4,
|
||||
},
|
||||
}),
|
||||
}),
|
||||
],
|
||||
// 5. mutes
|
||||
() => [],
|
||||
// 6. unread events — none (the override is inactive, no message events either)
|
||||
() => [],
|
||||
// 7. mention events
|
||||
() => [],
|
||||
]);
|
||||
|
||||
// No complete projection — use fetched-only (no stored merge).
|
||||
// Previously this would be called with readStoredRegisters returning a stale
|
||||
// live register (S=5,C=0,B=100), which when max-joined with fetched tombstone
|
||||
// (S=0,C=4,B=0) produced (S=5,C=4,B=100) → active (resurrection).
|
||||
// New behavior: fetched state alone → (S=0,C=4,B=0) → inactive → dark.
|
||||
const result = await fetchCommunityUnread({
|
||||
client: relay,
|
||||
pubkey: PUBKEY,
|
||||
nowSeconds: 400,
|
||||
decryptReadState: async (v) => v,
|
||||
decryptMutes: async (v) => v,
|
||||
readThreadRelationships: readRelationships(),
|
||||
// No getProjection: fetched-deduped is the sole authority.
|
||||
});
|
||||
|
||||
assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
|
||||
});
|
||||
|
||||
test("fetchCommunityUnread adversarial-2: projection with frontier deactivates register → dark", async () => {
|
||||
// CRITICAL 2 fix: the rail now reads projection.frontiers, not only fetched frontiers.
|
||||
// A register (S=5,C=0,B=100) with projection frontier F=101 must evaluate as
|
||||
// inactive (101 > 100) — previously the frontier was not in the stored projection,
|
||||
// so it defaulted to F=0 and the register appeared active.
|
||||
//
|
||||
// The fetched relay state returns empty read-state (frontier not in fetched events).
|
||||
// The projection has the committed frontier F=101 from a successful mark-read.
|
||||
const relay = quietRelay(); // no read-state events fetched
|
||||
|
||||
const overrides = new Map([[CHANNEL_ID, { s: 5, c: 0, b: 100 }]]);
|
||||
// projection.frontiers has F=101 for CHANNEL_ID → deactivates the register
|
||||
const frontiers = new Map([[CHANNEL_ID, 101]]);
|
||||
|
||||
const result = await fetchCommunityUnread({
|
||||
client: relay,
|
||||
pubkey: PUBKEY,
|
||||
nowSeconds: 200,
|
||||
decryptReadState: async (v) => v,
|
||||
decryptMutes: async (v) => v,
|
||||
readThreadRelationships: readRelationships(),
|
||||
getProjection: () => ({ loadComplete: true, frontiers, overrides }),
|
||||
});
|
||||
|
||||
assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
|
||||
|
||||
@@ -7,8 +7,8 @@ import {
|
||||
msgContextKey,
|
||||
type OverrideRegister,
|
||||
} from "@/features/channels/readState/readStateFormat";
|
||||
import { readStoredReadState } from "@/features/channels/readState/readStateStorage";
|
||||
import { deduplicateByCoordinate } from "@/features/channels/readState/readStateFencedLoader";
|
||||
import type { ReadStateProjection } from "@/features/channels/readState/readStateManager";
|
||||
import {
|
||||
getThreadReference,
|
||||
isBroadcastReply,
|
||||
@@ -158,26 +158,22 @@ export async function fetchCommunityUnread(args: {
|
||||
decryptReadState?: (ciphertext: string) => Promise<string>;
|
||||
decryptMutes?: (ciphertext: string) => Promise<string>;
|
||||
readThreadRelationships?: (pubkey: string) => ThreadRelationships;
|
||||
/** Authoritative source (a): persisted override register projection. Defaults
|
||||
* to `readStoredReadState(pubkey).overrideRegisters`. Inject for tests. */
|
||||
readStoredRegisters?: (
|
||||
pubkey: string,
|
||||
) => ReadonlyMap<string, OverrideRegister>;
|
||||
/** Coherent manager projection for override evaluation. When provided and
|
||||
* `loadComplete` is true, the projection's `overrides` and `frontiers` are
|
||||
* the authoritative source — the fetched registers are still used for
|
||||
* frontier-only readAt, but override liveness is decided solely from the
|
||||
* projection (no per-field max join with fetched registers).
|
||||
*
|
||||
* When null or `loadComplete` is false, use fetched coordinate-deduped
|
||||
* registers only — do not fall back to a partial stored view whose
|
||||
* per-field merge can resurrect a superseded register. */
|
||||
getProjection?: () => ReadStateProjection | null;
|
||||
}): Promise<CommunityUnreadObserverResult> {
|
||||
const { client, pubkey } = args;
|
||||
const normalizedPubkey = pubkey.toLowerCase();
|
||||
const decryptMutes = args.decryptMutes ?? nip44DecryptFromSelf;
|
||||
const readRelationships =
|
||||
args.readThreadRelationships ?? defaultReadThreadRelationships;
|
||||
const getStoredRegisters =
|
||||
args.readStoredRegisters ??
|
||||
((pk: string) => {
|
||||
try {
|
||||
return readStoredReadState(pk).overrideRegisters;
|
||||
} catch {
|
||||
return new Map<string, OverrideRegister>();
|
||||
}
|
||||
});
|
||||
|
||||
const channels = await fetchObservedChannels(client, pubkey);
|
||||
if (channels.length === 0) {
|
||||
@@ -208,25 +204,48 @@ export async function fetchCommunityUnread(args: {
|
||||
args.decryptReadState,
|
||||
);
|
||||
|
||||
// Authoritative source (a): persisted full-state projection from the manager's
|
||||
// last complete load. Merge with source (b): max() per field, so a register
|
||||
// present in only one source is retained.
|
||||
const storedRegisters = getStoredRegisters(normalizedPubkey);
|
||||
const mergedOverrides = new Map<string, OverrideRegister>(
|
||||
readState.overrides,
|
||||
);
|
||||
for (const [rawCtx, stored] of storedRegisters) {
|
||||
const fetched = mergedOverrides.get(rawCtx);
|
||||
if (!fetched) {
|
||||
mergedOverrides.set(rawCtx, stored);
|
||||
} else {
|
||||
mergedOverrides.set(rawCtx, {
|
||||
s: Math.max(fetched.s, stored.s),
|
||||
c: Math.max(fetched.c, stored.c),
|
||||
b: Math.max(fetched.b, stored.b),
|
||||
});
|
||||
// Override liveness source: the manager projection when complete, otherwise
|
||||
// the fetched state only. Never per-field max the two — a projection register
|
||||
// and a fetched register for the same coordinate are NOT concurrent CRDT
|
||||
// replicas; the fetched register may be a newer tombstone superseding a stale
|
||||
// projection entry, and joining them per-field would resurrect the dead one.
|
||||
//
|
||||
// When the projection is complete (loadComplete=true) it is the most
|
||||
// authoritative source because it includes committed local override actions.
|
||||
// Use projection.overrides as the sole override verdict source. For frontiers,
|
||||
// take max(projection, fetched) per channel so a locally-committed frontier
|
||||
// advance is honoured even if the relay hasn't seen it yet.
|
||||
//
|
||||
// When no complete projection is available, use fetched-deduped state alone.
|
||||
const projection = args.getProjection?.() ?? null;
|
||||
const completeProjection = projection?.loadComplete ? projection : null;
|
||||
|
||||
const authoritative: ReadonlyMap<string, OverrideRegister> =
|
||||
completeProjection !== null
|
||||
? completeProjection.overrides
|
||||
: readState.overrides;
|
||||
|
||||
// Effective frontier: max of fetched and (when available) projection frontier.
|
||||
const effectiveFrontier = (channelId: string): number | null => {
|
||||
const fetched = readState.frontiers.get(channelId) ?? null;
|
||||
if (completeProjection === null) return fetched;
|
||||
const projected = completeProjection.frontiers.get(channelId) ?? null;
|
||||
if (fetched === null) return projected;
|
||||
if (projected === null) return fetched;
|
||||
return Math.max(fetched, projected);
|
||||
};
|
||||
|
||||
// Build a unified frontier map (fetched + projection max) for per-event
|
||||
// msg/thread context lookups inside isUnreadExternalEvent.
|
||||
const effectiveFrontierMap: ReadonlyMap<string, number> = (() => {
|
||||
if (completeProjection === null) return readState.frontiers;
|
||||
const merged = new Map<string, number>(readState.frontiers);
|
||||
for (const [ctx, pf] of completeProjection.frontiers) {
|
||||
const existing = merged.get(ctx);
|
||||
merged.set(ctx, existing !== undefined ? Math.max(existing, pf) : pf);
|
||||
}
|
||||
}
|
||||
return merged;
|
||||
})();
|
||||
|
||||
let mutedIds = new Set<string>();
|
||||
if (mutesEvents.length > 0) {
|
||||
@@ -254,14 +273,14 @@ export async function fetchCommunityUnread(args: {
|
||||
for (const channel of channels) {
|
||||
if (mutedIds.has(channel.id)) continue;
|
||||
|
||||
// Compute readAt from the merged frontiers.
|
||||
const readAt = readState.frontiers.get(channel.id) ?? null;
|
||||
// Compute readAt from effective frontiers (max of fetched + projection).
|
||||
const readAt = effectiveFrontier(channel.id);
|
||||
|
||||
// Override liveness: evaluate the merged authoritative register.
|
||||
// "No register in fetched events" is ambiguity, not absence — the local
|
||||
// cache is NOT a fallback verdict source here.
|
||||
// Override liveness: evaluate the authoritative register against the
|
||||
// effective frontier. Uses projection.overrides when complete (no
|
||||
// per-field max with fetched); fetched-deduped otherwise.
|
||||
if (!hasUnread) {
|
||||
const reg = mergedOverrides.get(channel.id);
|
||||
const reg = authoritative.get(channel.id);
|
||||
if (reg !== undefined && isOverrideActive(reg, readAt ?? 0)) {
|
||||
hasUnread = true;
|
||||
}
|
||||
@@ -296,7 +315,7 @@ export async function fetchCommunityUnread(args: {
|
||||
(event) =>
|
||||
isUnreadExternalEvent(
|
||||
event,
|
||||
readState.frontiers,
|
||||
effectiveFrontierMap,
|
||||
readAt,
|
||||
normalizedPubkey,
|
||||
) &&
|
||||
@@ -314,7 +333,7 @@ export async function fetchCommunityUnread(args: {
|
||||
mentionCount += mentionEvents.filter((event) =>
|
||||
isUnreadExternalEvent(
|
||||
event,
|
||||
readState.frontiers,
|
||||
effectiveFrontierMap,
|
||||
readAt,
|
||||
normalizedPubkey,
|
||||
),
|
||||
|
||||
Reference in New Issue
Block a user