mirror of
https://github.com/PR0M3TH3AN/bitvid.git
synced 2026-10-27 07:52:24 +00:00
Harden Nostr startup performance
This commit is contained in:
1 parent
4525c1c657
commit
fe976b3ab7
17 files changed
+828
-93
No files matched your search
@@ -176,6 +176,22 @@ export function createAuthSessionCoordinator(deps) {
|
||||
|
||||
return {
|
||||
async handleAuthLogin(detail = {}) {
|
||||
const requestedPubkey =
|
||||
this.normalizeHexPubkey?.(detail?.pubkey || this.pubkey) ||
|
||||
detail?.pubkey ||
|
||||
this.pubkey ||
|
||||
"";
|
||||
const inFlightLogin = this._authLoginSync;
|
||||
if (
|
||||
requestedPubkey &&
|
||||
inFlightLogin?.pubkey === requestedPubkey &&
|
||||
inFlightLogin?.promise
|
||||
) {
|
||||
userLogger.info("[auth-login-duplicate-ignored]", {
|
||||
pubkey: requestedPubkey,
|
||||
});
|
||||
return inFlightLogin.promise;
|
||||
}
|
||||
const authLoginStart = now();
|
||||
userLogger.info("[auth-login-start]", {
|
||||
pubkey: detail?.pubkey || this.pubkey || null,
|
||||
@@ -721,6 +737,20 @@ export function createAuthSessionCoordinator(deps) {
|
||||
return listSyncDetail;
|
||||
});
|
||||
|
||||
// A provider/widget can repeat its success event while the first login is
|
||||
// still hydrating lists. The identity is already active, so replaying the
|
||||
// full lifecycle only doubles relay work and grid refreshes. Join the
|
||||
// original lifecycle until its required list sync settles instead.
|
||||
const authLoginSync = Promise.resolve(listStatePromise).finally(() => {
|
||||
if (this._authLoginSync?.promise === authLoginSync) {
|
||||
this._authLoginSync = null;
|
||||
}
|
||||
});
|
||||
this._authLoginSync = {
|
||||
pubkey: activePubkey || requestedPubkey,
|
||||
promise: authLoginSync,
|
||||
};
|
||||
|
||||
// DMs are intentionally NOT eagerly loaded at login. A cold DM load
|
||||
// (limit: 50) is a burst of nip-07 decrypts that competes with the
|
||||
// feed-driving lists (blocks/subscriptions/hashtags) for the SAME
|
||||
|
||||
+79
-44
@@ -1,69 +1,104 @@
|
||||
// Dev-only relay REQ-rate monitor.
|
||||
// Local-only relay REQ-rate monitor.
|
||||
//
|
||||
// Patches WebSocket.send to count outgoing Nostr REQ frames and, every 2s,
|
||||
// console.warns when the rate exceeds a threshold — naming the storming filter
|
||||
// kinds. This runs inside the real browser session against real relays/lists,
|
||||
// which the headless harness cannot fully reproduce. It is purely observational
|
||||
// and only activates on localhost (or when `localStorage.__bitvidReqMonitor__`
|
||||
// is set), so production is unaffected.
|
||||
//
|
||||
// Remove this import from index.js once the login-perf work is done.
|
||||
// It patches WebSocket.send for a real-session relay-storm investigation, so
|
||||
// normal localhost development must opt in explicitly. Public deployments can
|
||||
// never enable it, even if a storage key is present.
|
||||
|
||||
(function installReqMonitor() {
|
||||
if (typeof window === "undefined" || typeof WebSocket === "undefined") return;
|
||||
export const REQUEST_MONITOR_STORAGE_KEY = "__bitvidReqMonitor__";
|
||||
|
||||
const enabled =
|
||||
location.hostname === "localhost" ||
|
||||
location.hostname === "127.0.0.1" ||
|
||||
(() => {
|
||||
try {
|
||||
return localStorage.getItem("__bitvidReqMonitor__") === "1";
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
})();
|
||||
if (!enabled || window.__bitvidReqMonitorInstalled) return;
|
||||
window.__bitvidReqMonitorInstalled = true;
|
||||
const LOCAL_HOSTS = new Set(["localhost", "127.0.0.1", "::1"]);
|
||||
const THRESHOLD = 15;
|
||||
const WINDOW_MS = 2000;
|
||||
|
||||
function readStorage(storage) {
|
||||
try {
|
||||
return storage?.getItem?.(REQUEST_MONITOR_STORAGE_KEY) ?? null;
|
||||
} catch (_) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export function isRequestMonitorEnabled({ locationRef, storage } = {}) {
|
||||
const hostname = typeof locationRef?.hostname === "string"
|
||||
? locationRef.hostname.toLowerCase()
|
||||
: "";
|
||||
return LOCAL_HOSTS.has(hostname) && readStorage(storage) === "1";
|
||||
}
|
||||
|
||||
export function installRequestMonitor({
|
||||
scope = typeof window !== "undefined" ? window : null,
|
||||
locationRef = scope?.location,
|
||||
storage = scope?.localStorage,
|
||||
WebSocketClass = scope?.WebSocket,
|
||||
setIntervalFn = scope?.setInterval?.bind(scope),
|
||||
clearIntervalFn = scope?.clearInterval?.bind(scope),
|
||||
logger = scope?.console || console,
|
||||
} = {}) {
|
||||
if (
|
||||
!scope ||
|
||||
!WebSocketClass?.prototype ||
|
||||
typeof WebSocketClass.prototype.send !== "function" ||
|
||||
typeof setIntervalFn !== "function" ||
|
||||
!isRequestMonitorEnabled({ locationRef, storage })
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
if (scope.__bitvidReqMonitorInstalled) {
|
||||
return scope.__bitvidReqMonitorInstalled;
|
||||
}
|
||||
|
||||
const THRESHOLD = 15; // REQ/s above this is a storm worth reporting
|
||||
const WINDOW_MS = 2000;
|
||||
let windowReqs = 0;
|
||||
const byKind = new Map();
|
||||
const originalSend = WebSocketClass.prototype.send;
|
||||
|
||||
const origSend = WebSocket.prototype.send;
|
||||
WebSocket.prototype.send = function (data) {
|
||||
WebSocketClass.prototype.send = function monitoredSend(data) {
|
||||
try {
|
||||
if (typeof data === "string" && data.startsWith('["REQ"')) {
|
||||
windowReqs += 1;
|
||||
const msg = JSON.parse(data);
|
||||
for (const f of msg.slice(2)) {
|
||||
const k = Array.isArray(f?.kinds) ? `kind ${f.kinds.join("/")}` : "kind ?";
|
||||
byKind.set(k, (byKind.get(k) || 0) + 1);
|
||||
const message = JSON.parse(data);
|
||||
for (const filter of message.slice(2)) {
|
||||
const kind = Array.isArray(filter?.kinds)
|
||||
? `kind ${filter.kinds.join("/")}`
|
||||
: "kind ?";
|
||||
byKind.set(kind, (byKind.get(kind) || 0) + 1);
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
/* ignore */
|
||||
} catch (_) {
|
||||
// Diagnostics must never affect a WebSocket send.
|
||||
}
|
||||
return origSend.apply(this, arguments);
|
||||
return originalSend.apply(this, arguments);
|
||||
};
|
||||
|
||||
setInterval(() => {
|
||||
const perSec = (windowReqs / WINDOW_MS) * 1000;
|
||||
if (perSec >= THRESHOLD) {
|
||||
const intervalId = setIntervalFn(() => {
|
||||
const perSecond = (windowReqs / WINDOW_MS) * 1000;
|
||||
if (perSecond >= THRESHOLD) {
|
||||
const top = [...byKind.entries()]
|
||||
.sort((a, b) => b[1] - a[1])
|
||||
.slice(0, 6)
|
||||
.map(([k, n]) => `${k}=${n}`)
|
||||
.map(([kind, count]) => `${kind}=${count}`)
|
||||
.join(", ");
|
||||
// eslint-disable-next-line no-console
|
||||
console.warn(
|
||||
`[req-monitor] RELAY STORM ${perSec.toFixed(0)} REQ/s over ${WINDOW_MS / 1000}s → ${top}`,
|
||||
logger?.warn?.(
|
||||
`[req-monitor] RELAY STORM ${perSecond.toFixed(0)} REQ/s over ${WINDOW_MS / 1000}s → ${top}`,
|
||||
);
|
||||
}
|
||||
windowReqs = 0;
|
||||
byKind.clear();
|
||||
}, WINDOW_MS);
|
||||
|
||||
// eslint-disable-next-line no-console
|
||||
console.info("[req-monitor] active — will warn if relay REQ rate spikes.");
|
||||
})();
|
||||
const monitor = {
|
||||
stop() {
|
||||
WebSocketClass.prototype.send = originalSend;
|
||||
if (typeof clearIntervalFn === "function") {
|
||||
clearIntervalFn(intervalId);
|
||||
}
|
||||
if (scope.__bitvidReqMonitorInstalled === monitor) {
|
||||
delete scope.__bitvidReqMonitorInstalled;
|
||||
}
|
||||
},
|
||||
};
|
||||
scope.__bitvidReqMonitorInstalled = monitor;
|
||||
logger?.info?.("[req-monitor] active — will warn if relay REQ rate spikes.");
|
||||
return monitor;
|
||||
}
|
||||
|
||||
installRequestMonitor();
|
||||
@@ -3,6 +3,7 @@
|
||||
import { validateInstanceConfig } from "../config/validate-config.js";
|
||||
import "./bufferPolyfill.js";
|
||||
import "./devReqMonitor.js"; // dev-only relay REQ-rate monitor (localhost only)
|
||||
import { initPerformanceHarness } from "./performanceHarness.js";
|
||||
import {
|
||||
getApplication,
|
||||
setApplication,
|
||||
@@ -59,6 +60,10 @@ import { installTestHarness } from "./testHarness.js";
|
||||
|
||||
validateInstanceConfig();
|
||||
|
||||
// Local-only, explicit opt-in measurement. This is a no-op in production and
|
||||
// does not patch application APIs or change scheduling.
|
||||
initPerformanceHarness();
|
||||
|
||||
// Install test harness early so relay overrides apply before connections.
|
||||
// Only activates when ?__test__=1 or localStorage __bitvidTestMode__ is set.
|
||||
installTestHarness();
|
||||
|
||||
@@ -0,0 +1,236 @@
|
||||
// Local-only performance measurement harness.
|
||||
//
|
||||
// This intentionally observes rather than changes application behavior. It is
|
||||
// available only on a loopback host *and* after an explicit localStorage opt-in,
|
||||
// so a production visitor cannot accidentally enable it.
|
||||
|
||||
const STORAGE_KEY = "__bitvidPerformanceHarness__";
|
||||
const GLOBAL_KEY = "__bitvidPerformance";
|
||||
const LOCAL_HOSTS = new Set(["localhost", "127.0.0.1", "::1"]);
|
||||
const MAX_LONG_TASKS = 200;
|
||||
const MAX_MARKS = 200;
|
||||
|
||||
function readStorage(storage, key) {
|
||||
try {
|
||||
return storage?.getItem?.(key) ?? null;
|
||||
} catch (_) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function finiteNumber(value, fallback = 0) {
|
||||
return Number.isFinite(value) ? Number(value) : fallback;
|
||||
}
|
||||
|
||||
function normalizeMarkName(name) {
|
||||
return typeof name === "string" ? name.trim().slice(0, 120) : "";
|
||||
}
|
||||
|
||||
export function isLocalPerformanceHarnessEnabled({
|
||||
locationRef = typeof location !== "undefined" ? location : null,
|
||||
storage = typeof localStorage !== "undefined" ? localStorage : null,
|
||||
} = {}) {
|
||||
const hostname = typeof locationRef?.hostname === "string"
|
||||
? locationRef.hostname.toLowerCase()
|
||||
: "";
|
||||
return LOCAL_HOSTS.has(hostname) && readStorage(storage, STORAGE_KEY) === "1";
|
||||
}
|
||||
|
||||
export class PerformanceHarness {
|
||||
constructor({ scope, performanceRef, PerformanceObserverClass } = {}) {
|
||||
this.scope = scope || null;
|
||||
this.performance = performanceRef || this.scope?.performance || null;
|
||||
this.PerformanceObserverClass =
|
||||
PerformanceObserverClass || this.scope?.PerformanceObserver || null;
|
||||
this.startedAt = 0;
|
||||
this.longTasks = [];
|
||||
this.marks = [];
|
||||
this.gauges = new Map();
|
||||
this.longTaskObserver = null;
|
||||
this.started = false;
|
||||
}
|
||||
|
||||
start() {
|
||||
if (this.started) {
|
||||
return this;
|
||||
}
|
||||
this.started = true;
|
||||
this.startedAt = this.now();
|
||||
this.observeLongTasks();
|
||||
this.mark("harness-start");
|
||||
if (this.scope) {
|
||||
this.scope[GLOBAL_KEY] = this.publicApi();
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
stop() {
|
||||
try {
|
||||
this.longTaskObserver?.disconnect?.();
|
||||
} catch (_) {
|
||||
// Diagnostics must never affect application teardown.
|
||||
}
|
||||
this.longTaskObserver = null;
|
||||
this.started = false;
|
||||
if (this.scope?.[GLOBAL_KEY]) {
|
||||
delete this.scope[GLOBAL_KEY];
|
||||
}
|
||||
}
|
||||
|
||||
now() {
|
||||
return finiteNumber(this.performance?.now?.(), Date.now());
|
||||
}
|
||||
|
||||
observeLongTasks() {
|
||||
if (typeof this.PerformanceObserverClass !== "function") {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const observer = new this.PerformanceObserverClass((list) => {
|
||||
const entries = list?.getEntries?.() || [];
|
||||
for (const entry of entries) {
|
||||
this.recordLongTask(entry);
|
||||
}
|
||||
});
|
||||
observer.observe({ type: "longtask", buffered: true });
|
||||
this.longTaskObserver = observer;
|
||||
} catch (_) {
|
||||
// Long Task timing is optional and unsupported in several browsers.
|
||||
}
|
||||
}
|
||||
|
||||
recordLongTask(entry) {
|
||||
const task = {
|
||||
startTime: finiteNumber(entry?.startTime),
|
||||
duration: finiteNumber(entry?.duration),
|
||||
name: typeof entry?.name === "string" ? entry.name : "longtask",
|
||||
};
|
||||
this.longTasks.push(task);
|
||||
if (this.longTasks.length > MAX_LONG_TASKS) {
|
||||
this.longTasks.splice(0, this.longTasks.length - MAX_LONG_TASKS);
|
||||
}
|
||||
}
|
||||
|
||||
mark(name, detail = {}) {
|
||||
const normalizedName = normalizeMarkName(name);
|
||||
if (!normalizedName) {
|
||||
return false;
|
||||
}
|
||||
const entry = {
|
||||
name: normalizedName,
|
||||
at: this.now(),
|
||||
detail: detail && typeof detail === "object" ? { ...detail } : {},
|
||||
};
|
||||
this.marks.push(entry);
|
||||
if (this.marks.length > MAX_MARKS) {
|
||||
this.marks.splice(0, this.marks.length - MAX_MARKS);
|
||||
}
|
||||
try {
|
||||
this.performance?.mark?.(`bitvid:${normalizedName}`);
|
||||
} catch (_) {
|
||||
// User Timing is also optional; retain the local mark either way.
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
setGauge(name, value) {
|
||||
const normalizedName = normalizeMarkName(name);
|
||||
if (!normalizedName || !Number.isFinite(value)) {
|
||||
return false;
|
||||
}
|
||||
this.gauges.set(normalizedName, Number(value));
|
||||
return true;
|
||||
}
|
||||
|
||||
increment(name, amount = 1) {
|
||||
const normalizedName = normalizeMarkName(name);
|
||||
if (!normalizedName || !Number.isFinite(amount)) {
|
||||
return false;
|
||||
}
|
||||
return this.setGauge(normalizedName, (this.gauges.get(normalizedName) || 0) + amount);
|
||||
}
|
||||
|
||||
snapshot() {
|
||||
const resources = this.performance?.getEntriesByType?.("resource") || [];
|
||||
const navigation = this.performance?.getEntriesByType?.("navigation")?.[0] || null;
|
||||
const memory = this.performance?.memory;
|
||||
const longTaskDurationMs = this.longTasks.reduce(
|
||||
(total, task) => total + finiteNumber(task.duration),
|
||||
0,
|
||||
);
|
||||
return {
|
||||
enabled: this.started,
|
||||
startedAt: this.startedAt,
|
||||
elapsedMs: Math.max(0, this.now() - this.startedAt),
|
||||
longTasks: {
|
||||
count: this.longTasks.length,
|
||||
totalDurationMs: longTaskDurationMs,
|
||||
recent: this.longTasks.slice(-20),
|
||||
},
|
||||
marks: this.marks.slice(-50),
|
||||
gauges: Object.fromEntries(this.gauges.entries()),
|
||||
resources: {
|
||||
count: resources.length,
|
||||
transferSize: resources.reduce(
|
||||
(total, entry) => total + finiteNumber(entry?.transferSize),
|
||||
0,
|
||||
),
|
||||
},
|
||||
navigation: navigation
|
||||
? {
|
||||
domContentLoadedMs: finiteNumber(navigation.domContentLoadedEventEnd),
|
||||
loadEventMs: finiteNumber(navigation.loadEventEnd),
|
||||
}
|
||||
: null,
|
||||
memory: memory
|
||||
? {
|
||||
usedJsHeapSize: finiteNumber(memory.usedJSHeapSize),
|
||||
totalJsHeapSize: finiteNumber(memory.totalJSHeapSize),
|
||||
jsHeapSizeLimit: finiteNumber(memory.jsHeapSizeLimit),
|
||||
}
|
||||
: null,
|
||||
};
|
||||
}
|
||||
|
||||
reset() {
|
||||
this.longTasks.length = 0;
|
||||
this.marks.length = 0;
|
||||
this.gauges.clear();
|
||||
this.mark("harness-reset");
|
||||
}
|
||||
|
||||
publicApi() {
|
||||
return Object.freeze({
|
||||
snapshot: () => this.snapshot(),
|
||||
mark: (name, detail) => this.mark(name, detail),
|
||||
setGauge: (name, value) => this.setGauge(name, value),
|
||||
increment: (name, amount) => this.increment(name, amount),
|
||||
reset: () => this.reset(),
|
||||
stop: () => this.stop(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export function initPerformanceHarness({
|
||||
scope = typeof window !== "undefined" ? window : null,
|
||||
locationRef = scope?.location,
|
||||
storage = scope?.localStorage,
|
||||
performanceRef = scope?.performance,
|
||||
PerformanceObserverClass = scope?.PerformanceObserver,
|
||||
} = {}) {
|
||||
if (!scope || !isLocalPerformanceHarnessEnabled({ locationRef, storage })) {
|
||||
return null;
|
||||
}
|
||||
if (scope[GLOBAL_KEY]) {
|
||||
return scope[GLOBAL_KEY];
|
||||
}
|
||||
const harness = new PerformanceHarness({
|
||||
scope,
|
||||
performanceRef,
|
||||
PerformanceObserverClass,
|
||||
});
|
||||
harness.start();
|
||||
return scope[GLOBAL_KEY];
|
||||
}
|
||||
|
||||
export const PERFORMANCE_HARNESS_STORAGE_KEY = STORAGE_KEY;
|
||||
@@ -41,6 +41,9 @@ const MAX_OPEN_ATTEMPTS = 30; // ~60s of retries
|
||||
// (kind 30000), which during cold-start competes with the initial feed load; a
|
||||
// fallback covers the empty-feed case where no videos:updated ever fires.
|
||||
const FEED_READY_FALLBACK_MS = 12000;
|
||||
// Some relays do not reliably send EOSE. Do not hold the first visible update
|
||||
// forever, but give the initial relay burst time to settle before painting it.
|
||||
const INITIAL_SYNC_FALLBACK_MS = 5000;
|
||||
|
||||
export function createNip71IngestService({
|
||||
nostrClient,
|
||||
@@ -54,6 +57,7 @@ export function createNip71IngestService({
|
||||
maxOpenAttempts = MAX_OPEN_ATTEMPTS,
|
||||
refreshThrottleMs = REFRESH_THROTTLE_MS,
|
||||
feedReadyFallbackMs = FEED_READY_FALLBACK_MS,
|
||||
initialSyncFallbackMs = INITIAL_SYNC_FALLBACK_MS,
|
||||
} = {}) {
|
||||
let subscription = null;
|
||||
let buffer = [];
|
||||
@@ -67,6 +71,9 @@ export function createNip71IngestService({
|
||||
let deferArmed = false;
|
||||
let deferTimer = null;
|
||||
let offFeedReady = null;
|
||||
let initialSyncing = false;
|
||||
let initialRefreshPending = false;
|
||||
let initialSyncTimer = null;
|
||||
|
||||
function isAvailable() {
|
||||
return (
|
||||
@@ -183,7 +190,13 @@ export function createNip71IngestService({
|
||||
);
|
||||
|
||||
if (injected > 0) {
|
||||
requestRefresh();
|
||||
if (initialSyncing) {
|
||||
// Keep native BitVid, NIP-71 videos, and future NIP-71 media types in
|
||||
// the store immediately; defer only the expensive full-feed repaint.
|
||||
initialRefreshPending = true;
|
||||
} else {
|
||||
requestRefresh();
|
||||
}
|
||||
}
|
||||
|
||||
return injected;
|
||||
@@ -211,6 +224,10 @@ export function createNip71IngestService({
|
||||
|
||||
function emitRefresh() {
|
||||
lastRefreshAt = Date.now();
|
||||
const performanceHarness = globalThis?.__bitvidPerformance;
|
||||
const performanceRef = globalThis?.performance;
|
||||
const startedAt = typeof performanceRef?.now === "function" ? performanceRef.now() : 0;
|
||||
performanceHarness?.mark?.("nip71-refresh-filter-start");
|
||||
let videos = [];
|
||||
try {
|
||||
videos =
|
||||
@@ -220,6 +237,11 @@ export function createNip71IngestService({
|
||||
} catch (error) {
|
||||
logger?.warn?.("[nip71Ingest] getFilteredActiveVideos failed", error);
|
||||
}
|
||||
const filteredAt = typeof performanceRef?.now === "function" ? performanceRef.now() : startedAt;
|
||||
performanceHarness?.mark?.("nip71-refresh-filter-end", {
|
||||
durationMs: Math.max(0, filteredAt - startedAt),
|
||||
videos: videos.length,
|
||||
});
|
||||
// Confirm ingested videos survive the feed filter (whitelist/NSFW/etc.).
|
||||
const foreignCount = Array.isArray(videos)
|
||||
? videos.filter((v) => v?.source === "nip71-ingest").length
|
||||
@@ -234,6 +256,37 @@ export function createNip71IngestService({
|
||||
reason: "nip71-ingest",
|
||||
});
|
||||
}
|
||||
const emittedAt = typeof performanceRef?.now === "function" ? performanceRef.now() : filteredAt;
|
||||
performanceHarness?.mark?.("nip71-refresh-dispatch-end", {
|
||||
filterMs: Math.max(0, filteredAt - startedAt),
|
||||
dispatchMs: Math.max(0, emittedAt - filteredAt),
|
||||
videos: videos.length,
|
||||
});
|
||||
if (performanceHarness && (filteredAt - startedAt > 50 || emittedAt - filteredAt > 50)) {
|
||||
userLogger.info(
|
||||
`[nip71Ingest] refresh timing: filter ${Math.round(filteredAt - startedAt)}ms, dispatch ${Math.round(emittedAt - filteredAt)}ms, videos ${videos.length}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function finishInitialSync() {
|
||||
if (!initialSyncing) {
|
||||
return;
|
||||
}
|
||||
if (initialSyncTimer) {
|
||||
clearTimeout(initialSyncTimer);
|
||||
initialSyncTimer = null;
|
||||
}
|
||||
if (flushTimer) {
|
||||
clearTimeout(flushTimer);
|
||||
flushTimer = null;
|
||||
flush();
|
||||
}
|
||||
initialSyncing = false;
|
||||
if (initialRefreshPending) {
|
||||
initialRefreshPending = false;
|
||||
requestRefresh();
|
||||
}
|
||||
}
|
||||
|
||||
// Defer the first start until the native feed has rendered (first
|
||||
@@ -255,7 +308,7 @@ export function createNip71IngestService({
|
||||
deferTimer = null;
|
||||
}
|
||||
deferArmed = false;
|
||||
start();
|
||||
start({ coalesceInitialSync: true });
|
||||
};
|
||||
|
||||
if (typeof nostrService?.on === "function") {
|
||||
@@ -265,12 +318,17 @@ export function createNip71IngestService({
|
||||
return true;
|
||||
}
|
||||
|
||||
function start() {
|
||||
function start({ coalesceInitialSync = false } = {}) {
|
||||
if (started || !isAvailable()) {
|
||||
return false;
|
||||
}
|
||||
started = true;
|
||||
attemptsLeft = maxOpenAttempts;
|
||||
initialSyncing = coalesceInitialSync;
|
||||
initialRefreshPending = false;
|
||||
if (initialSyncing) {
|
||||
initialSyncTimer = setTimeout(finishInitialSync, initialSyncFallbackMs);
|
||||
}
|
||||
|
||||
// Re-subscribe when the whitelist changes so the author scope stays current.
|
||||
if (typeof accessControl?.onWhitelistChange === "function") {
|
||||
@@ -320,6 +378,7 @@ export function createNip71IngestService({
|
||||
scheduleFlush();
|
||||
}
|
||||
},
|
||||
onEose: finishInitialSync,
|
||||
});
|
||||
const authorCount = Array.isArray(filters[0]?.authors)
|
||||
? filters[0].authors.length
|
||||
@@ -348,6 +407,12 @@ export function createNip71IngestService({
|
||||
clearTimeout(pendingRefreshTimer);
|
||||
pendingRefreshTimer = null;
|
||||
}
|
||||
if (initialSyncTimer) {
|
||||
clearTimeout(initialSyncTimer);
|
||||
initialSyncTimer = null;
|
||||
}
|
||||
initialSyncing = false;
|
||||
initialRefreshPending = false;
|
||||
buffer = [];
|
||||
if (subscription && typeof subscription.close === "function") {
|
||||
try {
|
||||
|
||||
+49
-29
@@ -1071,6 +1071,12 @@ export class NostrService {
|
||||
|
||||
this.dmActorPubkey = normalizedActor;
|
||||
|
||||
const relaySelection = await this.resolveDirectMessageRelaySelection(
|
||||
normalizedActor,
|
||||
relays,
|
||||
);
|
||||
const resolvedRelays = relaySelection.relays;
|
||||
|
||||
let resolvedSince =
|
||||
Number.isFinite(sinceCandidate) && sinceCandidate >= 0
|
||||
? Math.floor(sinceCandidate)
|
||||
@@ -1095,7 +1101,7 @@ export class NostrService {
|
||||
let messages = [];
|
||||
try {
|
||||
messages = await this.nostrClient.listDirectMessages(normalizedActor, {
|
||||
relays,
|
||||
relays: resolvedRelays,
|
||||
...options,
|
||||
limit: resolvedLimit,
|
||||
decryptLimit: resolvedDecryptLimit,
|
||||
@@ -1133,6 +1139,14 @@ export class NostrService {
|
||||
messages: snapshot,
|
||||
actorPubkey: normalizedActor,
|
||||
});
|
||||
if (relaySelection.warning === DM_RELAY_WARNING_FALLBACK) {
|
||||
this.emit("directMessages:relayWarning", {
|
||||
actorPubkey: normalizedActor,
|
||||
warning: relaySelection.warning,
|
||||
relays: resolvedRelays,
|
||||
source: relaySelection.source,
|
||||
});
|
||||
}
|
||||
this.emit("directMessages:updated", {
|
||||
messages: snapshot,
|
||||
reason: "load",
|
||||
@@ -1156,6 +1170,35 @@ export class NostrService {
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
async resolveDirectMessageRelaySelection(actorPubkey, relays = null) {
|
||||
if (Array.isArray(relays)) {
|
||||
return { relays, source: "explicit", warning: null };
|
||||
}
|
||||
|
||||
const discoveryRelays = Array.isArray(this.nostrClient?.readRelays)
|
||||
? this.nostrClient.readRelays
|
||||
: Array.isArray(this.nostrClient?.relays)
|
||||
? this.nostrClient.relays
|
||||
: [];
|
||||
|
||||
let pool = this.nostrClient?.pool || null;
|
||||
if (!pool && typeof this.nostrClient?.ensurePool === "function") {
|
||||
try {
|
||||
pool = await this.nostrClient.ensurePool();
|
||||
} catch (error) {
|
||||
devLogger.warn("[nostrService] Failed to initialize DM relay pool.", error);
|
||||
}
|
||||
}
|
||||
|
||||
return resolveDmRelaySelection({
|
||||
pubkey: actorPubkey,
|
||||
discoveryRelays,
|
||||
fallbackRelays: discoveryRelays,
|
||||
pool,
|
||||
log: { dev: devLogger, user: userLogger },
|
||||
});
|
||||
}
|
||||
|
||||
async ensureDirectMessageSubscription({ actorPubkey, relays, ...handlers } = {}) {
|
||||
const activeActor = actorPubkey || this.resolveActiveDmActor();
|
||||
const normalizedActor = normalizeHexPubkey(activeActor);
|
||||
@@ -1189,34 +1232,11 @@ export class NostrService {
|
||||
this.stopDirectMessageSubscription();
|
||||
}
|
||||
|
||||
let resolvedRelays = Array.isArray(relays) ? relays : null;
|
||||
let relaySelection = null;
|
||||
|
||||
if (!resolvedRelays) {
|
||||
const discoveryRelays = Array.isArray(this.nostrClient?.readRelays)
|
||||
? this.nostrClient.readRelays
|
||||
: Array.isArray(this.nostrClient?.relays)
|
||||
? this.nostrClient.relays
|
||||
: [];
|
||||
|
||||
let pool = this.nostrClient?.pool || null;
|
||||
if (!pool && typeof this.nostrClient?.ensurePool === "function") {
|
||||
try {
|
||||
pool = await this.nostrClient.ensurePool();
|
||||
} catch (error) {
|
||||
devLogger.warn("[nostrService] Failed to initialize DM relay pool.", error);
|
||||
}
|
||||
}
|
||||
|
||||
relaySelection = await resolveDmRelaySelection({
|
||||
pubkey: normalizedActor,
|
||||
discoveryRelays,
|
||||
fallbackRelays: discoveryRelays,
|
||||
pool,
|
||||
log: { dev: devLogger, user: userLogger },
|
||||
});
|
||||
resolvedRelays = relaySelection.relays;
|
||||
}
|
||||
const relaySelection = await this.resolveDirectMessageRelaySelection(
|
||||
normalizedActor,
|
||||
relays,
|
||||
);
|
||||
const resolvedRelays = relaySelection.relays;
|
||||
|
||||
try {
|
||||
const subscription = this.nostrClient.subscribeDirectMessages(
|
||||
|
||||
+27
-9
@@ -138,6 +138,29 @@ export class TorrentClient {
|
||||
let timeoutId = null;
|
||||
let pollId = null;
|
||||
|
||||
// A successful probe commonly settles from WebTorrent's `wire` event.
|
||||
// Do not destroy the temporary torrent while that event is still being
|
||||
// dispatched: WebTorrent continues its handshake bookkeeping after
|
||||
// listeners return, and synchronous destruction can leave its bitfield
|
||||
// null (`recalculate(...).fill` in the bundled client). The next task
|
||||
// guarantees the callback has fully unwound before cleanup.
|
||||
const disposeProbeTorrent = (target, onDisposed = () => {}) => {
|
||||
if (!target) {
|
||||
onDisposed();
|
||||
return;
|
||||
}
|
||||
const dispose = () => {
|
||||
try {
|
||||
target.destroy({ destroyStore: true });
|
||||
} catch (err) {
|
||||
// A completed or already-destroyed probe is harmless.
|
||||
} finally {
|
||||
onDisposed();
|
||||
}
|
||||
};
|
||||
setTimeout(dispose, 0);
|
||||
};
|
||||
|
||||
const finalize = (overrides = {}) => {
|
||||
if (settled) {
|
||||
return;
|
||||
@@ -151,14 +174,6 @@ export class TorrentClient {
|
||||
clearInterval(pollId);
|
||||
pollId = null;
|
||||
}
|
||||
if (torrent) {
|
||||
try {
|
||||
torrent.destroy({ destroyStore: true });
|
||||
} catch (err) {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
|
||||
const endedAt =
|
||||
typeof performance !== "undefined" && performance?.now
|
||||
? performance.now()
|
||||
@@ -178,7 +193,10 @@ export class TorrentClient {
|
||||
...overrides,
|
||||
};
|
||||
emit("torrent-probe-result", result);
|
||||
resolve(result);
|
||||
// Keep the grid-health queue slot occupied until the delayed teardown
|
||||
// completes. Otherwise the next probe can begin before this probe has
|
||||
// detached its tracker listeners, briefly exceeding listener limits.
|
||||
disposeProbeTorrent(torrent, () => resolve(result));
|
||||
};
|
||||
|
||||
try {
|
||||
|
||||
Reference in new issue
Block a user