mirror of
https://github.com/PR0M3TH3AN/bitvid.git
synced 2026-10-27 07:52:24 +00:00
Refactor fetchListIncrementally into RelayBatchFetcher and genericize timeout logic
Extracted `fetchListIncrementally` from `js/nostr/client.js` into a new `RelayBatchFetcher` class in `js/nostr/relayBatchFetcher.js` to reduce complexity in `NostrClient`. Moved `withRequestTimeout` to `js/utils/asyncUtils.js` to make it reusable and remove duplication. Updated `js/nostr/client.js` to use these new components while preserving existing behavior and specific logging for COUNT diagnostics. - Created `js/nostr/relayBatchFetcher.js` - Updated `js/utils/asyncUtils.js` - Refactored `js/nostr/client.js` - Verified with existing tests Co-authored-by: PR0M3TH3AN <[email protected]>
This commit is contained in:
google-labs-jules[bot]andPR0M3TH3AN committed
1 parent
c785894943
commit
6a74b3ddfc
3 files changed
+304
-269
No files matched your search
+17
-269
@@ -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`
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user