diff --git a/js/nostr/client.js b/js/nostr/client.js index 6230bb25..ea488e7c 100644 --- a/js/nostr/client.js +++ b/js/nostr/client.js @@ -135,6 +135,7 @@ import { summarizeDmEventForLog, } from "./dmDecryptDiagnostics.js"; import { devLogger, userLogger } from "../utils/logger.js"; +import { withRequestTimeout } from "../utils/asyncUtils.js"; import { LRUCache } from "../utils/lruCache.js"; import { updateConversationFromMessage, writeMessages } from "../storage/dmDb.js"; import { @@ -212,6 +213,7 @@ import { createNip46Adapter } from "./adapters/nip46Adapter.js"; import { queueSignEvent } from "./signRequestQueue.js"; import { EventsMap } from "./eventsMap.js"; import { Nip46Connector } from "./nip46Connector.js"; +import { RelayBatchFetcher } from "./relayBatchFetcher.js"; function normalizeProfileFromEvent(event) { if (!event || !event.content) return null; @@ -893,55 +895,6 @@ export const __testExports = { attemptDecryptNip46HandshakePayload, }; -function withRequestTimeout(promise, timeoutMs, onTimeout, message = "Request timed out") { - const resolvedTimeout = Number(timeoutMs); - const effectiveTimeout = - Number.isFinite(resolvedTimeout) && resolvedTimeout > 0 - ? Math.floor(resolvedTimeout) - : 4000; - - let timeoutId = null; - let settled = false; - - return new Promise((resolve, reject) => { - timeoutId = setTimeout(() => { - if (settled) { - return; - } - settled = true; - if (typeof onTimeout === "function") { - try { - onTimeout(); - } catch (cleanupError) { - logCountTimeoutCleanupFailure(cleanupError); - } - } - reject(new Error(message)); - }, effectiveTimeout); - - Promise.resolve(promise) - .then((value) => { - if (settled) { - return; - } - settled = true; - if (timeoutId) { - clearTimeout(timeoutId); - } - resolve(value); - }) - .catch((error) => { - if (settled) { - return; - } - settled = true; - if (timeoutId) { - clearTimeout(timeoutId); - } - reject(error); - }); - }); -} function extractVideoPublishPayload(rawPayload) { let videoData = rawPayload; @@ -1406,6 +1359,7 @@ export class NostrClient { this.remoteSignerListeners = new Set(); this.sessionActorListeners = new Set(); this.nip46Connector = new Nip46Connector(this); + this.relayBatchFetcher = new RelayBatchFetcher(this); const storedRemoteSigner = this.getStoredNip46Metadata(); this.remoteSignerStatus = { state: storedRemoteSigner.hasSession ? "stored" : "idle", @@ -3160,225 +3114,15 @@ export class NostrClient { since, timeoutMs = 10000, } = {}) { - // Explanation: - // Optimizes network usage by only fetching "new" events from each relay. - // It uses `SyncMetadataStore` to track the `created_at` of the last seen event per relay. - // - If we have a `lastSeen` timestamp, we request `since: lastSeen + 1`. - // - If that returns nothing (or fails), we might fall back to a full fetch depending on context. - // - Updates the `lastSeen` timestamp upon success. - if (!kind || !pubkey) { - throw new Error("fetchListIncrementally requires kind and pubkey"); - } - - const normalizedPubkey = normalizeNostrPubkey(pubkey); - if (!normalizedPubkey) { - throw new Error("Invalid pubkey for fetchListIncrementally"); - } - - const relaysToUse = Array.isArray(relayUrls) && relayUrls.length - ? relayUrls - : this.relays; - - const healthyCandidates = this.getHealthyRelays(relaysToUse); - let relayCandidates = healthyCandidates; - if (!healthyCandidates.length && relaysToUse.length) { - const sanitizedFallback = sanitizeRelayList(relaysToUse); - const defaultFallback = sanitizeRelayList(Array.from(DEFAULT_RELAY_URLS)); - relayCandidates = sanitizedFallback.length - ? sanitizedFallback - : defaultFallback.length - ? defaultFallback - : sanitizeRelayList(Array.from(RELAY_URLS)); - devLogger.warn( - "[fetchListIncrementally] Healthy relays exhausted; using fallback relay list for one-off fetch.", - { - requested: relaysToUse, - fallback: relayCandidates, - }, - ); - } - const readPreferences = new Set(Array.isArray(this.readRelays) ? this.readRelays : []); - - // Sort relays: prefer user's read relays first - const sortedRelays = relayCandidates.sort((a, b) => { - const aPreferred = readPreferences.has(a); - const bPreferred = readPreferences.has(b); - if (aPreferred && !bPreferred) return -1; - if (!aPreferred && bPreferred) return 1; - return 0; + return this.relayBatchFetcher.fetchListIncrementally({ + kind, + pubkey, + dTag, + relayUrls, + fetchFn, + since, + timeoutMs, }); - - // Cap the number of relays to avoid excessive requests - const cappedRelays = sortedRelays.slice(0, 8); - const normalizedRelays = sanitizeRelayList(cappedRelays); - - if (isDevMode) { - devLogger.log("[fetchListIncrementally] Selected relays:", { - count: normalizedRelays.length, - relays: normalizedRelays, - }); - } - - const results = []; - const concurrencyLimit = 8; - - // We'll use the pool's list method if fetchFn isn't provided, - // but we need to call it per-relay. - const pool = await this.ensurePool(); - const actualFetchFn = typeof fetchFn === "function" - ? fetchFn - : (r, f, timeout) => pool.list([r], [f], { timeout }); - - const chunks = []; - for (let i = 0; i < normalizedRelays.length; i += concurrencyLimit) { - chunks.push(normalizedRelays.slice(i, i + concurrencyLimit)); - } - - let anySuccess = false; - const failures = []; - - for (const chunk of chunks) { - const promises = chunk.map(async (relayUrl) => { - let lastSeen = this.getSyncLastSeen( - kind, - normalizedPubkey, - dTag, - relayUrl, - ); - - // If an explicit 'since' override is provided, use it instead of storage. - // This allows services to anchor fetching to their own known state. - if (since !== undefined) { - const overrideSince = Number(since); - lastSeen = - Number.isFinite(overrideSince) && overrideSince >= 0 - ? Math.floor(overrideSince) - : 0; - } - - const filter = { - kinds: [kind], - authors: [normalizedPubkey], - }; - - if (dTag) { - filter["#d"] = [dTag]; - } - - // Strategy: - // If we have a lastSeen, ask for since = lastSeen + 1. - // If that fails or returns nothing, we don't necessarily fall back - // unless there's an error. - // Wait, requirements say: - // "If metadata is missing for a relay → do a full fetch" - // "If a relay returns an error ... fall back to full fetch" - - let doFullFetch = true; - if (lastSeen > 0) { - filter.since = lastSeen + 1; - doFullFetch = false; - } - - // Allow callers to override list fetch timeouts without changing non-list query behavior. - const listTimeoutMs = Number.isFinite(timeoutMs) && timeoutMs > 0 ? timeoutMs : 10000; - - try { - // Wrap fetch with a timeout to prevent hanging on slow relays - let events = await withRequestTimeout( - actualFetchFn(relayUrl, filter, listTimeoutMs), - listTimeoutMs, - null, - `Fetch from ${relayUrl} timed out` - ); - - // If incremental returned empty, we assume no updates (success). - // But if we did a full fetch (no since), empty means empty. - - if (!events || !Array.isArray(events)) { - events = []; - } - - // On success, update lastSeen with max created_at - let maxCreated = 0; - for (const ev of events) { - if (ev.created_at > maxCreated) { - maxCreated = ev.created_at; - } - } - - if (maxCreated > 0) { - // Only update if we found something newer or if we did a full fetch - // Actually, if we did incremental and found new stuff, maxCreated > lastSeen. - // If we did full fetch, maxCreated is the latest. - this.updateSyncLastSeen(kind, normalizedPubkey, dTag, relayUrl, maxCreated); - } - - return { events, ok: true }; - } catch (error) { - // Fallback to full fetch if incremental failed or if relay error - if (!doFullFetch) { - try { - delete filter.since; - const events = await withRequestTimeout( - actualFetchFn(relayUrl, filter, listTimeoutMs), - listTimeoutMs, - null, - `Full fetch fallback from ${relayUrl} timed out` - ); - // Update lastSeen on success of fallback - let maxCreated = 0; - if (Array.isArray(events)) { - for (const ev of events) { - if (ev.created_at > maxCreated) { - maxCreated = ev.created_at; - } - } - } - if (maxCreated > 0) { - this.updateSyncLastSeen(kind, normalizedPubkey, dTag, relayUrl, maxCreated); - } - return { events: events || [], ok: true }; - } catch (fallbackError) { - devLogger.warn(`[fetchListIncrementally] Full fetch fallback failed for ${relayUrl}`, fallbackError); - return { error: fallbackError, ok: false }; - } - } else { - devLogger.warn(`[fetchListIncrementally] Fetch failed for ${relayUrl}`, error); - this.markRelayUnreachable(relayUrl); - return { error, ok: false }; - } - } - }); - - const chunkResults = await Promise.all(promises); - for (const res of chunkResults) { - if (res.ok) { - anySuccess = true; - if (Array.isArray(res.events)) { - results.push(...res.events); - } - } else { - failures.push(res.error); - } - } - } - - if (!anySuccess && normalizedRelays.length > 0) { - const error = new Error("All relays failed to fetch list."); - error.code = "fetch-failed"; - error.failures = failures; - throw error; - } - - // Deduplicate by ID - const uniqueEvents = new Map(); - for (const ev of results) { - if (ev && ev.id) { - uniqueEvents.set(ev.id, ev); - } - } - - return Array.from(uniqueEvents.values()); } applyRelayPreferences(preferences = {}) { @@ -7478,8 +7222,12 @@ export class NostrClient { countPromise, timeoutMs, () => { - if (relay?.openCountRequests instanceof Map) { - relay.openCountRequests.delete(requestId); + try { + if (relay?.openCountRequests instanceof Map) { + relay.openCountRequests.delete(requestId); + } + } catch (cleanupError) { + logCountTimeoutCleanupFailure(cleanupError); } }, `COUNT request timed out after ${timeoutMs}ms` diff --git a/js/nostr/relayBatchFetcher.js b/js/nostr/relayBatchFetcher.js new file mode 100644 index 00000000..7afe346e --- /dev/null +++ b/js/nostr/relayBatchFetcher.js @@ -0,0 +1,235 @@ +import { isDevMode } from "../config.js"; +import { devLogger } from "../utils/logger.js"; +import { withRequestTimeout } from "../utils/asyncUtils.js"; +import { normalizeNostrPubkey, sanitizeRelayList } from "./nip46Client.js"; +import { DEFAULT_RELAY_URLS, RELAY_URLS } from "./toolkit.js"; + +export class RelayBatchFetcher { + constructor(client) { + this.client = client; + } + + async fetchListIncrementally({ + kind, + pubkey, + dTag, + relayUrls, + fetchFn, + since, + timeoutMs = 10000, + } = {}) { + if (!kind || !pubkey) { + throw new Error("fetchListIncrementally requires kind and pubkey"); + } + + const normalizedPubkey = normalizeNostrPubkey(pubkey); + if (!normalizedPubkey) { + throw new Error("Invalid pubkey for fetchListIncrementally"); + } + + const relaysToUse = Array.isArray(relayUrls) && relayUrls.length + ? relayUrls + : this.client.relays; + + const healthyCandidates = this.client.getHealthyRelays(relaysToUse); + let relayCandidates = healthyCandidates; + if (!healthyCandidates.length && relaysToUse.length) { + const sanitizedFallback = sanitizeRelayList(relaysToUse); + const defaultFallback = sanitizeRelayList(Array.from(DEFAULT_RELAY_URLS)); + relayCandidates = sanitizedFallback.length + ? sanitizedFallback + : defaultFallback.length + ? defaultFallback + : sanitizeRelayList(Array.from(RELAY_URLS)); + devLogger.warn( + "[fetchListIncrementally] Healthy relays exhausted; using fallback relay list for one-off fetch.", + { + requested: relaysToUse, + fallback: relayCandidates, + }, + ); + } + const readPreferences = new Set(Array.isArray(this.client.readRelays) ? this.client.readRelays : []); + + // Sort relays: prefer user's read relays first + const sortedRelays = relayCandidates.sort((a, b) => { + const aPreferred = readPreferences.has(a); + const bPreferred = readPreferences.has(b); + if (aPreferred && !bPreferred) return -1; + if (!aPreferred && bPreferred) return 1; + return 0; + }); + + // Cap the number of relays to avoid excessive requests + const cappedRelays = sortedRelays.slice(0, 8); + const normalizedRelays = sanitizeRelayList(cappedRelays); + + if (isDevMode) { + devLogger.log("[fetchListIncrementally] Selected relays:", { + count: normalizedRelays.length, + relays: normalizedRelays, + }); + } + + const results = []; + const concurrencyLimit = 8; + + // We'll use the pool's list method if fetchFn isn't provided, + // but we need to call it per-relay. + const pool = await this.client.ensurePool(); + const actualFetchFn = typeof fetchFn === "function" + ? fetchFn + : (r, f, timeout) => pool.list([r], [f], { timeout }); + + const chunks = []; + for (let i = 0; i < normalizedRelays.length; i += concurrencyLimit) { + chunks.push(normalizedRelays.slice(i, i + concurrencyLimit)); + } + + let anySuccess = false; + const failures = []; + + for (const chunk of chunks) { + const promises = chunk.map(async (relayUrl) => { + let lastSeen = this.client.getSyncLastSeen( + kind, + normalizedPubkey, + dTag, + relayUrl, + ); + + // If an explicit 'since' override is provided, use it instead of storage. + // This allows services to anchor fetching to their own known state. + if (since !== undefined) { + const overrideSince = Number(since); + lastSeen = + Number.isFinite(overrideSince) && overrideSince >= 0 + ? Math.floor(overrideSince) + : 0; + } + + const filter = { + kinds: [kind], + authors: [normalizedPubkey], + }; + + if (dTag) { + filter["#d"] = [dTag]; + } + + // Strategy: + // If we have a lastSeen, ask for since = lastSeen + 1. + // If that fails or returns nothing, we don't necessarily fall back + // unless there's an error. + // Wait, requirements say: + // "If metadata is missing for a relay → do a full fetch" + // "If a relay returns an error ... fall back to full fetch" + + let doFullFetch = true; + if (lastSeen > 0) { + filter.since = lastSeen + 1; + doFullFetch = false; + } + + // Allow callers to override list fetch timeouts without changing non-list query behavior. + const listTimeoutMs = Number.isFinite(timeoutMs) && timeoutMs > 0 ? timeoutMs : 10000; + + try { + // Wrap fetch with a timeout to prevent hanging on slow relays + let events = await withRequestTimeout( + actualFetchFn(relayUrl, filter, listTimeoutMs), + listTimeoutMs, + null, + `Fetch from ${relayUrl} timed out` + ); + + // If incremental returned empty, we assume no updates (success). + // But if we did a full fetch (no since), empty means empty. + + if (!events || !Array.isArray(events)) { + events = []; + } + + // On success, update lastSeen with max created_at + let maxCreated = 0; + for (const ev of events) { + if (ev.created_at > maxCreated) { + maxCreated = ev.created_at; + } + } + + if (maxCreated > 0) { + // Only update if we found something newer or if we did a full fetch + // Actually, if we did incremental and found new stuff, maxCreated > lastSeen. + // If we did full fetch, maxCreated is the latest. + this.client.updateSyncLastSeen(kind, normalizedPubkey, dTag, relayUrl, maxCreated); + } + + return { events, ok: true }; + } catch (error) { + // Fallback to full fetch if incremental failed or if relay error + if (!doFullFetch) { + try { + delete filter.since; + const events = await withRequestTimeout( + actualFetchFn(relayUrl, filter, listTimeoutMs), + listTimeoutMs, + null, + `Full fetch fallback from ${relayUrl} timed out` + ); + // Update lastSeen on success of fallback + let maxCreated = 0; + if (Array.isArray(events)) { + for (const ev of events) { + if (ev.created_at > maxCreated) { + maxCreated = ev.created_at; + } + } + } + if (maxCreated > 0) { + this.client.updateSyncLastSeen(kind, normalizedPubkey, dTag, relayUrl, maxCreated); + } + return { events: events || [], ok: true }; + } catch (fallbackError) { + devLogger.warn(`[fetchListIncrementally] Full fetch fallback failed for ${relayUrl}`, fallbackError); + return { error: fallbackError, ok: false }; + } + } else { + devLogger.warn(`[fetchListIncrementally] Fetch failed for ${relayUrl}`, error); + this.client.markRelayUnreachable(relayUrl); + return { error, ok: false }; + } + } + }); + + const chunkResults = await Promise.all(promises); + for (const res of chunkResults) { + if (res.ok) { + anySuccess = true; + if (Array.isArray(res.events)) { + results.push(...res.events); + } + } else { + failures.push(res.error); + } + } + } + + if (!anySuccess && normalizedRelays.length > 0) { + const error = new Error("All relays failed to fetch list."); + error.code = "fetch-failed"; + error.failures = failures; + throw error; + } + + // Deduplicate by ID + const uniqueEvents = new Map(); + for (const ev of results) { + if (ev && ev.id) { + uniqueEvents.set(ev.id, ev); + } + } + + return Array.from(uniqueEvents.values()); + } +} diff --git a/js/utils/asyncUtils.js b/js/utils/asyncUtils.js index f6ceb271..2cb12df0 100644 --- a/js/utils/asyncUtils.js +++ b/js/utils/asyncUtils.js @@ -1,3 +1,5 @@ +import { devLogger } from "./logger.js"; + /** * Maps over an iterable with a concurrency limit. * @@ -51,3 +53,53 @@ export async function pMap(iterable, mapper, { concurrency = Infinity } = {}) { next(); }); } + +export function withRequestTimeout(promise, timeoutMs, onTimeout, message = "Request timed out") { + const resolvedTimeout = Number(timeoutMs); + const effectiveTimeout = + Number.isFinite(resolvedTimeout) && resolvedTimeout > 0 + ? Math.floor(resolvedTimeout) + : 4000; + + let timeoutId = null; + let settled = false; + + return new Promise((resolve, reject) => { + timeoutId = setTimeout(() => { + if (settled) { + return; + } + settled = true; + if (typeof onTimeout === "function") { + try { + onTimeout(); + } catch (cleanupError) { + devLogger.warn("[asyncUtils] Timeout cleanup failed:", cleanupError); + } + } + reject(new Error(message)); + }, effectiveTimeout); + + Promise.resolve(promise) + .then((value) => { + if (settled) { + return; + } + settled = true; + if (timeoutId) { + clearTimeout(timeoutId); + } + resolve(value); + }) + .catch((error) => { + if (settled) { + return; + } + settled = true; + if (timeoutId) { + clearTimeout(timeoutId); + } + reject(error); + }); + }); +}