nip71: make ingest observable + throttle its feed refreshes

Two stabilization fixes after live testing showed ingest was invisible
and could amplify feed load:

1. Diagnostics were using devLogger.log, which is gated by IS_DEV_MODE
   (false in all current builds) — so the [nip71Ingest] subscribe/flush
   lines never printed and the feature was impossible to observe. Switched
   the subscribe + per-flush lines to userLogger.info so we can confirm in
   the console whether it subscribed (author count) and whether events are
   arriving/injecting. (Temporary while stabilizing.)

2. Coalesce ingest-driven "videos:updated" signals behind an 8s throttle.
   A burst of incoming events previously emitted a refresh per flush; each
   refresh re-runs the feed and re-subscribes moderation/social context for
   every video, which can contribute to a REQ storm. Injected videos are
   already in the active store, so throttling only delays the visual
   refresh, not the data.

Tests: throttle coalescing test (one immediate refresh per window, both
videos still stored); mutation-verified. 1234/1235 unit (known flake).

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
thePR0M3TH3ANandClaude Opus 4.8 committed 2026-06-22 16:05:01 -04:00
1 parent dcce28cc97
commit c4df08d5a1
2 files changed
+76 -22

No files matched your search

+57 -22
View File
@@ -20,9 +20,13 @@ import {
buildVideoFromNip71Event,
NIP71_KINDS,
} from "../nostr/nip71IngestAdapter.js";
import { devLogger } from "../utils/logger.js";
import { devLogger, userLogger } from "../utils/logger.js";
const FLUSH_DEBOUNCE_MS = 200;
// Coalesce ingest-driven feed refreshes so a burst of incoming events can't
// trigger a re-render/re-subscription storm. The injected videos are already in
// the active store; this only controls how often we *signal* the feed.
const REFRESH_THROTTLE_MS = 8000;
const DEFAULT_LIMIT = 200;
const SUBSCRIPTION_KEY = "nip71-ingest";
// The admin whitelist hydrates asynchronously (remote fetch, then a second hex
@@ -43,6 +47,7 @@ export function createNip71IngestService({
limit = DEFAULT_LIMIT,
openRetryDelayMs = OPEN_RETRY_DELAY_MS,
maxOpenAttempts = MAX_OPEN_ATTEMPTS,
refreshThrottleMs = REFRESH_THROTTLE_MS,
} = {}) {
let subscription = null;
let buffer = [];
@@ -51,6 +56,8 @@ export function createNip71IngestService({
let attemptsLeft = maxOpenAttempts;
let offWhitelistChange = null;
let started = false;
let lastRefreshAt = 0;
let pendingRefreshTimer = null;
function isAvailable() {
return (
@@ -148,35 +155,59 @@ export function createNip71IngestService({
}
}
logger?.log?.(
// Visible (not dev-gated) so ingest is observable in any build while we
// stabilize the feature.
userLogger.info(
`[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
// re-reads getFilteredActiveVideos itself, so this just signals "refresh".
let videos = [];
try {
videos =
typeof nostrService?.getFilteredActiveVideos === "function"
? nostrService.getFilteredActiveVideos()
: [];
} catch (error) {
logger?.warn?.("[nip71Ingest] getFilteredActiveVideos failed", error);
}
if (typeof nostrService?.emit === "function") {
nostrService.emit("videos:updated", {
videos,
deleted: [],
reason: "nip71-ingest",
});
}
requestRefresh();
}
return injected;
}
// Signal the feed to re-render, throttled so a burst of ingested events can't
// drive a re-render/re-subscription storm. The injected videos already live in
// the active store; the feed re-reads getFilteredActiveVideos itself, so this
// is purely a "refresh now" nudge.
function requestRefresh() {
if (pendingRefreshTimer) {
return;
}
const now = Date.now();
const elapsed = now - lastRefreshAt;
if (elapsed >= refreshThrottleMs) {
emitRefresh();
return;
}
pendingRefreshTimer = setTimeout(() => {
pendingRefreshTimer = null;
emitRefresh();
}, refreshThrottleMs - elapsed);
}
function emitRefresh() {
lastRefreshAt = Date.now();
let videos = [];
try {
videos =
typeof nostrService?.getFilteredActiveVideos === "function"
? nostrService.getFilteredActiveVideos()
: [];
} catch (error) {
logger?.warn?.("[nip71Ingest] getFilteredActiveVideos failed", error);
}
if (typeof nostrService?.emit === "function") {
nostrService.emit("videos:updated", {
videos,
deleted: [],
reason: "nip71-ingest",
});
}
}
function start() {
if (started || !isAvailable()) {
return false;
@@ -236,7 +267,7 @@ export function createNip71IngestService({
const authorCount = Array.isArray(filters[0]?.authors)
? filters[0].authors.length
: "all";
logger?.log?.(
userLogger.info(
`[nip71Ingest] subscribed (authors: ${authorCount}, kinds: ${filters[0]?.kinds?.join(",")})`,
);
return true;
@@ -256,6 +287,10 @@ export function createNip71IngestService({
clearTimeout(retryTimer);
retryTimer = null;
}
if (pendingRefreshTimer) {
clearTimeout(pendingRefreshTimer);
pendingRefreshTimer = null;
}
buffer = [];
if (subscription && typeof subscription.close === "function") {
try {
+19
View File
@@ -102,6 +102,8 @@ function makeService(harness, extra = {}) {
// 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,
// Emit refreshes synchronously (no throttle timer) for deterministic tests.
refreshThrottleMs: 0,
...extra,
});
}
@@ -216,6 +218,23 @@ test("re-subscribes when the whitelist changes", () => {
assert.deepEqual(h.capturedFilters[0].authors, ["aa", "cc"], "subscription re-scoped after change");
});
test("coalesces refresh signals so an event burst can't storm the feed", () => {
const h = makeHarness();
const svc = makeService(h, { refreshThrottleMs: 10000 });
svc.start();
h.fireEvent(foreignEvent({ id: "a", kind: 34235, tags: [["d", "ra"], ["title", "A"], ["imeta", "url https://e/a.mp4", "m video/mp4"]] }));
svc.flush();
h.fireEvent(foreignEvent({ id: "b", kind: 34235, tags: [["d", "rb"], ["title", "B"], ["imeta", "url https://e/b.mp4", "m video/mp4"]] }));
svc.flush();
const refreshes = h.emitted.filter((e) => e.name === "videos:updated");
assert.equal(refreshes.length, 1, "only one immediate refresh within the throttle window");
// Both videos still landed in the store regardless of the throttled signal.
assert.ok(h.activeMap.has("ROOT:ra") && h.activeMap.has("ROOT:rb"));
svc.stop();
});
test("FEATURE_NIP71_INGEST off makes the service completely inert", () => {
const h = makeHarness();
const svc = createNip71IngestService({