mirror of
https://github.com/PR0M3TH3AN/bitvid.git
synced 2026-10-27 16:02:26 +00:00
refactor(moderation): batch kind-1984 report subs via SubscriptionManager (P2)
Replaces the per-video live report subscription (one pool.sub per visible video × ~20 relays, re-fired on every reconnect — the ~1000 REQ/s storm) with ONE batched SubscriptionManager subscription over the whole active id set, updated in place as the set changes. Backfill fetches history only for newly-added ids (one batched list()). Unchanged sets are a no-op (dedup guard). MODERATION_REPORT_SUBS_ENABLED flipped back on. Trusted-mute subs were already batched-by-author, so the relay cap + contact-loop fix already tamed them. Legacy per-id path retained only for SubscriptionManager-less environments (unit tests / mock clients). All 39 moderation tests pass; new cheat-resistant batching test added; nip07-heat shows reports no longer storm (steady idle). Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
1 parent
9f0541737f
commit
25d3e83c46
2 files changed
+201
-13
No files matched your search
@@ -58,10 +58,40 @@ import {
|
||||
* trusted social graph (follows).
|
||||
*/
|
||||
|
||||
// Temporary kill-switch for per-video live report subscriptions. Real-env logs
|
||||
// showed these fan out O(videos × relays) and storm relays (~1000 REQ/s),
|
||||
// starving list decryption. Re-enable once batched (single REQ for all ids).
|
||||
const MODERATION_REPORT_SUBS_ENABLED = false;
|
||||
// Live kind-1984 report subscriptions. Previously one sub PER VIDEO fanned
|
||||
// O(videos × relays) and stormed relays (~1000 REQ/s). Now batched into a single
|
||||
// SubscriptionManager subscription over the whole active id set, so this is back
|
||||
// on. (Set false to fall back to the disabled state for debugging.)
|
||||
const MODERATION_REPORT_SUBS_ENABLED = true;
|
||||
|
||||
// Max event ids per kind-1984 filter chunk. nostr REQ allows multiple filters in
|
||||
// one subscription, so a large active set becomes a few chunks in ONE sub —
|
||||
// still O(1) subscriptions, not O(ids).
|
||||
const REPORT_BATCH_SIZE = 200;
|
||||
|
||||
function setsEqual(a, b) {
|
||||
if (a === b) return true;
|
||||
if (!(a instanceof Set) || !(b instanceof Set) || a.size !== b.size) return false;
|
||||
for (const v of a) {
|
||||
if (!b.has(v)) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
// Build kind-1984 report filters for a set of event ids, chunked so no single
|
||||
// filter carries an unbounded "#e" array.
|
||||
function buildReportFilters(ids) {
|
||||
const list = Array.isArray(ids) ? ids : Array.from(ids || []);
|
||||
const filters = [];
|
||||
for (let i = 0; i < list.length; i += REPORT_BATCH_SIZE) {
|
||||
filters.push({
|
||||
kinds: [1984],
|
||||
"#e": list.slice(i, i + REPORT_BATCH_SIZE),
|
||||
limit: 500,
|
||||
});
|
||||
}
|
||||
return filters;
|
||||
}
|
||||
|
||||
export class ModerationService {
|
||||
/**
|
||||
@@ -1596,14 +1626,10 @@ export class ModerationService {
|
||||
}
|
||||
|
||||
async setActiveEventIds(ids = []) {
|
||||
// CIRCUIT-BREAKER (confirmed against real env): one live kind-1984 report
|
||||
// subscription PER VIDEO, fanned across ~20 relays and re-issued on every
|
||||
// flaky-relay reconnect, produced ~1000 REQ/s and starved list decryption.
|
||||
// Stays disabled until replaced by a single batched subscription (one REQ
|
||||
// with "#e":[...all ids]) targeting only healthy relays. See refactor plan.
|
||||
if (!MODERATION_REPORT_SUBS_ENABLED) {
|
||||
return;
|
||||
}
|
||||
|
||||
const nextIds = new Set();
|
||||
if (Array.isArray(ids) || ids instanceof Set) {
|
||||
for (const value of ids) {
|
||||
@@ -1614,6 +1640,78 @@ export class ModerationService {
|
||||
}
|
||||
}
|
||||
|
||||
// Skip redundant work when the visible set is unchanged (the feed stage
|
||||
// calls this on every run).
|
||||
const prevIds = this.activeEventIds instanceof Set ? this.activeEventIds : new Set();
|
||||
if (setsEqual(prevIds, nextIds)) {
|
||||
return;
|
||||
}
|
||||
|
||||
const manager =
|
||||
typeof this.nostrClient?.getSubscriptionManager === "function"
|
||||
? this.nostrClient.getSubscriptionManager()
|
||||
: null;
|
||||
|
||||
// Fallback for environments without the SubscriptionManager (tests/mocks):
|
||||
// keep the legacy per-id behavior so those paths are unaffected.
|
||||
if (!manager) {
|
||||
return this._setActiveEventIdsLegacy(nextIds, prevIds);
|
||||
}
|
||||
|
||||
this.activeEventIds = nextIds;
|
||||
|
||||
if (nextIds.size === 0) {
|
||||
if (this._reportSubHandle) {
|
||||
this._reportSubHandle.close();
|
||||
this._reportSubHandle = null;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const idArray = Array.from(nextIds);
|
||||
const filters = buildReportFilters(idArray);
|
||||
|
||||
// ONE live batched subscription over the whole active set, updated in place
|
||||
// as the set changes (no teardown/resubscribe churn).
|
||||
if (this._reportSubHandle) {
|
||||
this._reportSubHandle.update({ filters });
|
||||
} else {
|
||||
this._reportSubHandle = manager.subscribe({
|
||||
key: "moderation:reports",
|
||||
label: "reports",
|
||||
filters,
|
||||
onEvent: (event) => this.ingestReportEvent(event),
|
||||
});
|
||||
}
|
||||
|
||||
// Backfill history for only the newly-added ids — one batched list().
|
||||
const newIds = idArray.filter((id) => !prevIds.has(id));
|
||||
if (newIds.length) {
|
||||
try {
|
||||
const events = await manager.list({ filters: buildReportFilters(newIds) });
|
||||
if (Array.isArray(events)) {
|
||||
for (const event of events) {
|
||||
this.ingestReportEvent(event);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
this.log("[moderationService] batched report backfill failed", error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Legacy per-id path, retained only for environments without a
|
||||
// SubscriptionManager (unit tests / mock clients). The real app uses the
|
||||
// batched path above.
|
||||
async _setActiveEventIdsLegacy(nextIds, prevIds = null) {
|
||||
const previous =
|
||||
prevIds instanceof Set
|
||||
? prevIds
|
||||
: this.activeEventIds instanceof Set
|
||||
? this.activeEventIds
|
||||
: new Set();
|
||||
void previous;
|
||||
|
||||
for (const existing of Array.from(this.activeSubscriptions.keys())) {
|
||||
if (!nextIds.has(existing)) {
|
||||
this.teardownReportSubscription(existing);
|
||||
@@ -1630,9 +1728,11 @@ export class ModerationService {
|
||||
continue;
|
||||
}
|
||||
const promise = this.subscribeToReports(id);
|
||||
tasks.push(promise.catch((error) => {
|
||||
this.log(`(moderationService) failed to subscribe to reports for ${id}`, error);
|
||||
}));
|
||||
tasks.push(
|
||||
promise.catch((error) => {
|
||||
this.log(`(moderationService) failed to subscribe to reports for ${id}`, error);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
this.activeEventIds = nextIds;
|
||||
@@ -1647,11 +1747,22 @@ export class ModerationService {
|
||||
return;
|
||||
}
|
||||
|
||||
// With the batched subscription the report *data* is unchanged by a trust
|
||||
// graph update (same event ids); callers already recompute summaries. Just
|
||||
// ensure the single live subscription is (re)issued for its current set.
|
||||
if (this._reportSubHandle) {
|
||||
this._reportSubHandle.update({ filters: buildReportFilters(Array.from(this.activeEventIds)) });
|
||||
return;
|
||||
}
|
||||
|
||||
// Legacy path (no SubscriptionManager): tear down + re-subscribe via the
|
||||
// public (overridable) method. Clear activeEventIds first so the
|
||||
// unchanged-set guard in setActiveEventIds doesn't short-circuit the rebuild.
|
||||
const ids = Array.from(this.activeEventIds);
|
||||
for (const id of ids) {
|
||||
this.teardownReportSubscription(id);
|
||||
}
|
||||
|
||||
this.activeEventIds = new Set();
|
||||
await this.setActiveEventIds(ids);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
// Scenario (SCN-report-subs-batched):
|
||||
// Given the moderation service has a SubscriptionManager available,
|
||||
// When the feed sets the active visible event-id set,
|
||||
// Then it opens exactly ONE batched kind-1984 subscription (not one per id),
|
||||
// updates that same subscription in place as the set changes,
|
||||
// backfills history for only the newly-added ids, and does nothing when the
|
||||
// set is unchanged.
|
||||
//
|
||||
// This is the P2 fix for the ~1000 REQ/s report storm (one sub per video × ~20
|
||||
// relays, re-fired on every reconnect). See docs/architecture-refactor.md.
|
||||
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import {
|
||||
withMockedNostrTools,
|
||||
createModerationServiceHarness,
|
||||
} from "../helpers/moderation-test-helpers.mjs";
|
||||
|
||||
const A = "a".repeat(64);
|
||||
const B = "b".repeat(64);
|
||||
const C = "c".repeat(64);
|
||||
const f = (...ids) => [{ kinds: [1984], "#e": ids, limit: 500 }];
|
||||
|
||||
test("setActiveEventIds uses one batched report sub, updated in place", async (t) => {
|
||||
withMockedNostrTools(t);
|
||||
const { service } = createModerationServiceHarness(t);
|
||||
|
||||
const calls = { subscribe: 0, update: 0, list: 0, lastFilters: null, lastList: null };
|
||||
const handle = {
|
||||
update({ filters }) {
|
||||
calls.update += 1;
|
||||
calls.lastFilters = filters;
|
||||
},
|
||||
close() {},
|
||||
};
|
||||
const fakeManager = {
|
||||
subscribe({ filters }) {
|
||||
calls.subscribe += 1;
|
||||
calls.lastFilters = filters;
|
||||
return handle;
|
||||
},
|
||||
list({ filters }) {
|
||||
calls.list += 1;
|
||||
calls.lastList = filters;
|
||||
return Promise.resolve([]);
|
||||
},
|
||||
};
|
||||
service.nostrClient = { getSubscriptionManager: () => fakeManager };
|
||||
|
||||
// Initial set of two ids => ONE subscription, ONE backfill covering both.
|
||||
await service.setActiveEventIds([A, B]);
|
||||
assert.equal(calls.subscribe, 1, "exactly one batched subscription (not per-id)");
|
||||
assert.deepEqual(calls.lastFilters, f(A, B), "live sub covers the whole set in one filter");
|
||||
assert.equal(calls.list, 1, "one backfill list");
|
||||
assert.deepEqual(calls.lastList, f(A, B), "backfill covers the two new ids");
|
||||
|
||||
// Add a third id => same sub updated in place; backfill ONLY the new id.
|
||||
await service.setActiveEventIds([A, B, C]);
|
||||
assert.equal(calls.subscribe, 1, "no new subscription is opened");
|
||||
assert.equal(calls.update, 1, "existing subscription updated in place");
|
||||
assert.deepEqual(calls.lastFilters, f(A, B, C), "updated filter covers the full set");
|
||||
assert.equal(calls.list, 2);
|
||||
assert.deepEqual(calls.lastList, f(C), "backfill only the newly-added id");
|
||||
|
||||
// Unchanged set => no work at all.
|
||||
await service.setActiveEventIds([A, B, C]);
|
||||
assert.equal(calls.update, 1, "unchanged set must not re-issue");
|
||||
assert.equal(calls.list, 2, "unchanged set must not re-backfill");
|
||||
|
||||
// Empty set => close the subscription.
|
||||
let closed = false;
|
||||
handle.close = () => {
|
||||
closed = true;
|
||||
};
|
||||
await service.setActiveEventIds([]);
|
||||
assert.ok(closed, "emptying the set closes the live subscription");
|
||||
});
|
||||
Reference in new issue
Block a user