nip71: fix ingest never subscribing due to whitelist-hydration race

The admin whitelist hydrates asynchronously (remote fetch, then a second
hex-pubkey rebuild once NostrTools loads) and does not always emit a
change we can hook — e.g. loaded-from-cache-unchanged, or the async hex
rebuild that emits nothing. So at startup getWhitelistPubkeys() was empty,
resolveFilters() returned null, no subscription opened, and the
onWhitelistChange restart never fired: ingest stayed silent even when a
whitelisted author had foreign NIP-71 videos on the relays.

Fix: bounded retry (every 2s, ~60s) that re-attempts opening the
subscription until the author scope hydrates, in addition to the
onWhitelistChange listener. Author-scoping is kept deliberately — a broad
kinds:[21,22,...] subscription would be a REQ firehose that violates
bitvid's relay fan-out invariants.

Also adds dev-mode logging (subscribe author/kind count; per-flush events
in / injected) so ingest is observable in the console.

Tests: new race test (empty whitelist at start -> authors hydrate without
an emit -> retry subscribes); mutation-verified.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
thePR0M3TH3ANandClaude Opus 4.8 committed 2026-06-22 15:46:28 -04:00
1 parent c849969312
commit dcce28cc97
2 files changed
+73 -3

No files matched your search

+51 -3
View File
@@ -25,6 +25,13 @@ import { devLogger } from "../utils/logger.js";
const FLUSH_DEBOUNCE_MS = 200;
const DEFAULT_LIMIT = 200;
const SUBSCRIPTION_KEY = "nip71-ingest";
// The admin whitelist hydrates asynchronously (remote fetch, then a second hex
// rebuild once NostrTools loads) and does not always emit a change we can hook
// (e.g. loaded-from-cache unchanged). So if the author scope isn't ready at
// start, poll a bounded number of times until it is, instead of silently never
// subscribing.
const OPEN_RETRY_DELAY_MS = 2000;
const MAX_OPEN_ATTEMPTS = 30; // ~60s of retries
export function createNip71IngestService({
nostrClient,
@@ -34,10 +41,14 @@ export function createNip71IngestService({
logger = devLogger,
flushDelayMs = FLUSH_DEBOUNCE_MS,
limit = DEFAULT_LIMIT,
openRetryDelayMs = OPEN_RETRY_DELAY_MS,
maxOpenAttempts = MAX_OPEN_ATTEMPTS,
} = {}) {
let subscription = null;
let buffer = [];
let flushTimer = null;
let retryTimer = null;
let attemptsLeft = maxOpenAttempts;
let offWhitelistChange = null;
let started = false;
@@ -137,6 +148,10 @@ export function createNip71IngestService({
}
}
logger?.log?.(
`[nip71Ingest] flush: ${events.length} event(s) in, ${injected} injected`,
);
if (injected > 0) {
// Re-render through the existing reactive path. The payload is the
// post-filter active list (whitelist/NSFW/etc. already applied); the feed
@@ -167,6 +182,7 @@ export function createNip71IngestService({
return false;
}
started = true;
attemptsLeft = maxOpenAttempts;
// Re-subscribe when the whitelist changes so the author scope stays current.
if (typeof accessControl?.onWhitelistChange === "function") {
@@ -175,13 +191,34 @@ export function createNip71IngestService({
});
}
return openSubscription();
return attemptOpen();
}
// Open the subscription; if the author scope isn't ready yet (whitelist still
// hydrating), retry on a timer rather than giving up silently.
function attemptOpen() {
if (retryTimer) {
clearTimeout(retryTimer);
retryTimer = null;
}
if (subscription) {
return true;
}
const opened = openSubscription();
if (opened) {
return true;
}
if (started && attemptsLeft > 0) {
attemptsLeft -= 1;
retryTimer = setTimeout(attemptOpen, openRetryDelayMs);
}
return false;
}
function openSubscription() {
const filters = resolveFilters();
if (!filters) {
return false; // nothing to ingest (e.g. empty whitelist) — stay armed for re-subscribe
return false; // nothing to ingest yet (e.g. whitelist not hydrated) — retry
}
try {
const manager = nostrClient.getSubscriptionManager();
@@ -196,6 +233,12 @@ export function createNip71IngestService({
}
},
});
const authorCount = Array.isArray(filters[0]?.authors)
? filters[0].authors.length
: "all";
logger?.log?.(
`[nip71Ingest] subscribed (authors: ${authorCount}, kinds: ${filters[0]?.kinds?.join(",")})`,
);
return true;
} catch (error) {
logger?.warn?.("[nip71Ingest] subscribe failed", error);
@@ -209,6 +252,10 @@ export function createNip71IngestService({
clearTimeout(flushTimer);
flushTimer = null;
}
if (retryTimer) {
clearTimeout(retryTimer);
retryTimer = null;
}
buffer = [];
if (subscription && typeof subscription.close === "function") {
try {
@@ -225,7 +272,8 @@ export function createNip71IngestService({
return false;
}
closeSubscription();
return openSubscription();
attemptsLeft = maxOpenAttempts;
return attemptOpen();
}
function stop() {
+22
View File
@@ -99,6 +99,9 @@ function makeService(harness, extra = {}) {
accessControl: harness.accessControl,
featureEnabled: true,
flushDelayMs: 0,
// Disable the hydration retry by default so each test is deterministic and
// leaves no dangling timers; the retry behavior is covered explicitly below.
maxOpenAttempts: 0,
...extra,
});
}
@@ -182,6 +185,25 @@ test("opens an unscoped (capped) subscription when whitelist mode is off", () =>
assert.ok(Number.isFinite(filters[0].limit), "still capped by a limit");
});
test("retries until the whitelist hydrates, then subscribes (no missed-emit race)", async () => {
// Simulates the real startup race: whitelist mode is on but the hex author
// set isn't populated yet at start(), and no change event is emitted.
const h = makeHarness({ whitelistOn: true, whitelist: [] });
const svc = makeService(h, { maxOpenAttempts: 5, openRetryDelayMs: 5 });
assert.equal(svc.start(), false, "cannot open yet — authors not hydrated");
assert.equal(h.capturedFilters, null);
// Authors hydrate a moment later, WITHOUT firing onWhitelistChange.
h.accessControl.getWhitelistPubkeys = () => ["aa", "bb"];
// Wait past a retry tick.
await new Promise((resolve) => setTimeout(resolve, 20));
assert.ok(h.capturedFilters, "retry opened the subscription after hydration");
assert.deepEqual(h.capturedFilters[0].authors, ["aa", "bb"]);
svc.stop();
});
test("re-subscribes when the whitelist changes", () => {
const h = makeHarness({ whitelistOn: true, whitelist: ["aa"] });
const svc = makeService(h);