From 0f31e1c301b73d80ae5c6e1fce1cfcfbbb6bf856 Mon Sep 17 00:00:00 2001 From: Laan Tungir Date: Tue, 30 Jun 2026 10:21:55 -0400 Subject: [PATCH] Batched broadcast publishing: split 600+ relays into batches of 25, publish sequentially to avoid browser WebSocket connection limits. 10s per-relay timeout per batch. --- plans/broadcast-batching-fix.md | 122 ++++++++++++++++++++++++++++ www/js/version.json | 6 +- www/ndk-worker.js | 139 +++++++++++++------------------- 3 files changed, 183 insertions(+), 84 deletions(-) create mode 100644 plans/broadcast-batching-fix.md diff --git a/plans/broadcast-batching-fix.md b/plans/broadcast-batching-fix.md new file mode 100644 index 0000000..6e65f7c --- /dev/null +++ b/plans/broadcast-batching-fix.md @@ -0,0 +1,122 @@ +# Broadcast Relay Batching Fix + +## Problem + +With 600+ broadcast relays, NDK's `fromRelayUrls` creates 630 temporary relays +and adds them all to the pool simultaneously. `NDKRelaySet.publish()` then fires +all 630 relay publish operations in parallel via `Promise.all`. Each relay that +isn't connected calls `relay.connect()` which opens a WebSocket. + +Browsers limit simultaneous WebSocket connections (practically ~50-100 at a +time before connections queue up). With 630 simultaneous connection attempts, +most relays never get a connection slot before the per-relay timeout expires. +Result: only ~13/630 relays succeed. + +## Solution: Chunked publishing + +Instead of creating one giant `NDKRelaySet` with all 630 relays and calling +`publish()` once, we split the relay URLs into **batches of 50** and publish +to each batch sequentially. Each batch gets its own `NDKRelaySet.fromRelayUrls()` +call, its own `publish()`, and its own set of live progress listeners. + +```mermaid +flowchart TD + A[630 relay URLs] --> B[Split into batches of 50] + B --> C[Batch 1: 50 relays] + C --> D[publish to batch 1] + D --> E{Batch 1 done?} + E -->|yes| F[Batch 2: 50 relays] + F --> G[publish to batch 2] + G --> H{Batch 2 done?} + H -->|yes| I[... continue ...] + I --> J[Batch 13: 30 relays] + J --> K[publish to batch 13] + K --> L{All batches done?} + L -->|yes| M[Final done event with total count] +``` + +### Key design points + +1. **Batch size: 50 relays** — small enough to stay within browser WebSocket + limits, large enough to finish in a reasonable time. Each batch takes ~5-10s. + +2. **Sequential batches** — wait for each batch to complete before starting the + next. This keeps the total simultaneous connections at ~50, not 630. + +3. **Live progress across batches** — the `publishedSoFar` and `failedSoFar` + sets accumulate across all batches. The `broadcastProgress` events stream + continuously, so the footer shows the count climbing from 0 to 630 across + all batches. + +4. **Per-relay timeout: 10s per batch** — each relay in a batch gets 10s to + connect + publish. Since only 50 are competing for connections (not 630), + this is plenty of time. + +5. **Overall deadline: removed** — since batches are sequential and each has + its own per-relay timeout, the overall time is naturally bounded at + `numBatches × perBatchTimeout`. No need for a separate overall deadline. + +6. **Connection timeout: 10s** — set on each relay in the batch, matching the + publish timeout. + +## Implementation (in `www/ndk-worker.js` `handlePublish`) + +Replace the current single `NDKRelaySet.fromRelayUrls(allUrls, ndk)` + single +`publish()` with a batched loop: + +```js +const BATCH_SIZE = 50; +const PER_RELAY_TIMEOUT_MS = 10000; + +if (isBroadcast) { + const allUrls = [...new Set([...outboxWriteUrls, ...activeBroadcastUrls])]; + const batches = []; + for (let i = 0; i < allUrls.length; i += BATCH_SIZE) { + batches.push(allUrls.slice(i, i + BATCH_SIZE)); + } + console.log(`[Worker] Broadcasting to ${allUrls.length} relays in ${batches.length} batches of ${BATCH_SIZE}`); + + // Emit start event + broadcast({ type: 'broadcastProgress', sessionId, eventId, eventKind, phase: 'start', total: allUrls.length, successful: 0, failed: 0, latestRelay: null }); + + // Attach listeners once — they accumulate across all batches + ndkEvent.on('relay:published', (relay) => { ... publishedSoFar.add ... }); + ndkEvent.on('relay:publish:failed', (relay, err) => { ... failedSoFar.add ... }); + + // Publish to each batch sequentially + for (let i = 0; i < batches.length; i++) { + const batch = batches[i]; + const batchSet = NDKRelaySet.fromRelayUrls(batch, ndk); + // Set connection timeout on each relay in this batch + for (const relay of batchSet.relays) { + try { relay.connectionTimeout = PER_RELAY_TIMEOUT_MS; } catch (_) {} + } + try { + await ndkEvent.publish(batchSet, PER_RELAY_TIMEOUT_MS, 1); + } catch (e) { + // Expected — some relays in this batch may fail + } + console.log(`[Worker] Batch ${i+1}/${batches.length} done: ${publishedSoFar.size}/${allUrls.length} total succeeded`); + } + + // Emit final done event + broadcast({ type: 'broadcastProgress', sessionId, eventId, eventKind, phase: 'done', total: allUrls.length, successful: publishedSoFar.size, failed: failedSoFar.size + timedOutSoFar.size, relayUrls: Array.from(publishedSoFar) }); + // Clean up listeners + ndkEvent.removeAllListeners('relay:published'); + ndkEvent.removeAllListeners('relay:publish:failed'); +} +``` + +## Files touched + +| File | Change | +|------|--------| +| [`www/ndk-worker.js`](www/ndk-worker.js) | Replace single publish with batched loop in `handlePublish` | + +## Testing + +1. Post with a user that has 600+ broadcast relays +2. Watch the footer: count should climb steadily as batches complete +3. Final count should be much higher than 13 (ideally 500+) +4. Each batch of 50 should take ~5-10 seconds +5. Total time for 630 relays: ~13 batches × ~10s = ~130 seconds diff --git a/www/js/version.json b/www/js/version.json index df66c85..5151d6d 100644 --- a/www/js/version.json +++ b/www/js/version.json @@ -1,5 +1,5 @@ { - "VERSION": "v0.7.76", - "VERSION_NUMBER": "0.7.76", - "BUILD_DATE": "2026-06-30T14:13:46.694Z" + "VERSION": "v0.7.77", + "VERSION_NUMBER": "0.7.77", + "BUILD_DATE": "2026-06-30T14:21:55.493Z" } diff --git a/www/ndk-worker.js b/www/ndk-worker.js index f4678c2..d897e0a 100644 --- a/www/ndk-worker.js +++ b/www/ndk-worker.js @@ -6192,7 +6192,6 @@ async function handlePublish(requestId, event, port) { // temporary NDKRelaySet. NDK's fromRelayUrls uses pool.useTemporaryRelay // so hundreds of broadcast relays connect transiently without polluting // the read subscription pool. - let targetRelaySet = null; const activeBroadcastUrls = getActiveBroadcastUrls(); const isBroadcast = wantsBroadcastRelays(event) && activeBroadcastUrls.length > 0; let outboxWriteUrls = []; @@ -6201,45 +6200,24 @@ async function handlePublish(requestId, event, port) { outboxWriteUrls = Array.from(relayTypes.entries()) .filter(([, t]) => t === 'write' || t === 'both') .map(([url]) => url); - const allUrls = [...new Set([...outboxWriteUrls, ...activeBroadcastUrls])]; - if (NDKRelaySet?.fromRelayUrls) { - targetRelaySet = NDKRelaySet.fromRelayUrls(allUrls, ndk); - // Set the connection timeout on all relays in the set to match - // the per-relay publish timeout. NDK's default - // connectionTimeout is 4400ms — we increase it so relays have - // time to connect even when hundreds are competing for browser - // connection slots. - if (targetRelaySet?.relays) { - for (const relay of targetRelaySet.relays) { - try { - if (relay.connectionTimeout !== undefined) { - relay.connectionTimeout = 60000; - } - } catch (_) { /* relay may be read-only */ } - } - } - console.log(`[Worker] Broadcasting to ${allUrls.length} relays ` + - `(${activeBroadcastUrls.length} broadcast + ${outboxWriteUrls.length} outbox, ` + - `${skippedRelayUrls.size} skipped)`); - } else { - console.warn('[Worker] NDKRelaySet.fromRelayUrls unavailable, falling back to default outbox'); - } } + const allBroadcastUrls = [...new Set([...outboxWriteUrls, ...activeBroadcastUrls])]; // --- Live broadcast progress streaming ------------------------------------ // NDK's NDKEvent emits 'relay:published' and 'relay:publish:failed' per - // relay as each completes (see ndk-core.bundle.js:8530 and :8552). We - // attach listeners BEFORE publish() and stream progress to all pages via - // broadcast() so footers can show "📡 42/150 relays · relay.example.com". + // relay as each completes. We attach listeners BEFORE publish() and stream + // progress to all pages via broadcast() so footers can show the count. const publishSessionId = `pub_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; const publishedSoFar = new Set(); const failedSoFar = new Set(); // explicit rejections (auth-required, blocked, etc.) const timedOutSoFar = new Set(); // timeouts — NOT skip-marked (could be slow connect) - const totalTarget = targetRelaySet && targetRelaySet.relays - ? targetRelaySet.relays.size - : (targetRelaySet ? (targetRelaySet.size || 0) : 0); + const totalTarget = isBroadcast ? allBroadcastUrls.length : 0; if (isBroadcast && totalTarget > 0) { + console.log(`[Worker] Broadcasting to ${allBroadcastUrls.length} relays ` + + `(${activeBroadcastUrls.length} broadcast + ${outboxWriteUrls.length} outbox, ` + + `${skippedRelayUrls.size} skipped)`); + broadcast({ type: 'broadcastProgress', sessionId: publishSessionId, @@ -6263,7 +6241,7 @@ async function handlePublish(requestId, event, port) { phase: 'progress', total: totalTarget, successful: publishedSoFar.size, - failed: failedSoFar.size, + failed: failedSoFar.size + timedOutSoFar.size, latestRelay: url, }); }); @@ -6272,10 +6250,6 @@ async function handlePublish(requestId, event, port) { const url = relay?.url || String(relay || ''); if (!url) return; const errMsg = err?.message || String(err) || ''; - // Distinguish timeouts from explicit rejections. Timeouts with - // hundreds of relays often just mean the relay was slow to connect, - // not that it's broken — so we track them separately and do NOT - // skip-mark them. Only explicit rejections go into failedSoFar. const isTimeout = /timeout/i.test(errMsg); if (isTimeout) { timedOutSoFar.add(url); @@ -6295,55 +6269,49 @@ async function handlePublish(requestId, event, port) { }); } - // Publish to relays (with broadcast relay set if broadcasting, else default outbox). - // Per-relay timeout for broadcasts: 60 seconds. NDK may not be able to handle - // hundreds of simultaneous WebSocket connections in true parallel — the browser - // has connection limits and NDK's pool may serialize some operations. A 60s - // per-relay timeout gives each relay ample time to connect and publish even when - // queued behind hundreds of others. - // requiredRelayCount=1 so NDK doesn't throw if only 1 of 600 relays succeeds. - const BROADCAST_PER_RELAY_TIMEOUT_MS = 60000; - const publishTimeoutMs = isBroadcast ? BROADCAST_PER_RELAY_TIMEOUT_MS : undefined; - const publishRequiredCount = isBroadcast ? 1 : undefined; + // --- Batched publishing for broadcasts ------------------------------------ + // Browsers can only handle ~50-100 simultaneous WebSocket connections. + // Publishing to 600+ relays in one shot causes most connections to queue + // and time out before getting a slot. We split the relay URLs into batches + // of 25 and publish to each batch sequentially, keeping the number of + // simultaneous connections manageable. + const BATCH_SIZE = 25; + const PER_RELAY_TIMEOUT_MS = 10000; // 10s per relay within a batch - let relaySet; - if (isBroadcast) { - // For broadcasts, race the publish against an overall deadline. - // NDK's publish() uses Promise.all internally, so it waits for ALL - // relays to resolve. The overall deadline is totalRelays × perRelayTimeout - // as a safety cap. With 600 relays × 60s = 36000s (10 hours) — this is - // effectively "wait until all relays finish" since the deadline will never - // fire before the per-relay timeouts resolve all promises. - const BROADCAST_OVERALL_DEADLINE_MS = Math.max(30000, totalTarget * BROADCAST_PER_RELAY_TIMEOUT_MS); - const publishPromise = ndkEvent.publish(targetRelaySet, publishTimeoutMs, publishRequiredCount) - .catch((err) => { - // NDKPublishError is expected when not all relays succeed — - // the live listeners already captured the successes. - console.warn('[Worker] Broadcast publish() resolved with error (expected):', err?.message || err); - return null; - }); - const deadlinePromise = new Promise((resolve) => { - setTimeout(() => resolve('__DEADLINE__'), BROADCAST_OVERALL_DEADLINE_MS); - }); - const result = await Promise.race([publishPromise, deadlinePromise]); - if (result === '__DEADLINE__') { - console.log(`[Worker] Broadcast overall deadline (${BROADCAST_OVERALL_DEADLINE_MS}ms) reached — ` + - `${publishedSoFar.size} succeeded, ${failedSoFar.size + timedOutSoFar.size} failed/timeout, ` + - `${totalTarget - publishedSoFar.size - failedSoFar.size - timedOutSoFar.size} still pending`); + let relaySet = null; + if (isBroadcast && totalTarget > 0 && NDKRelaySet?.fromRelayUrls) { + // Split all URLs into batches. + const batches = []; + for (let i = 0; i < allBroadcastUrls.length; i += BATCH_SIZE) { + batches.push(allBroadcastUrls.slice(i, i + BATCH_SIZE)); } - relaySet = result && result !== '__DEADLINE__' ? result : null; - } else { - // Normal publish — await the full result as before. - try { - relaySet = await ndkEvent.publish(targetRelaySet, publishTimeoutMs, publishRequiredCount); - } catch (publishError) { - console.warn('[Worker] publish() threw:', publishError?.message || publishError); - relaySet = null; - } - } + console.log(`[Worker] Publishing in ${batches.length} batches of ${BATCH_SIZE} (per-relay timeout: ${PER_RELAY_TIMEOUT_MS}ms)`); - // Final broadcast progress message with complete results. - if (isBroadcast && totalTarget > 0) { + for (let bi = 0; bi < batches.length; bi++) { + const batch = batches[bi]; + const batchSet = NDKRelaySet.fromRelayUrls(batch, ndk); + // Set connection timeout on each relay in this batch. + if (batchSet?.relays) { + for (const relay of batchSet.relays) { + try { + if (relay.connectionTimeout !== undefined) { + relay.connectionTimeout = PER_RELAY_TIMEOUT_MS; + } + } catch (_) { /* relay may be read-only */ } + } + } + try { + await ndkEvent.publish(batchSet, PER_RELAY_TIMEOUT_MS, 1); + } catch (batchErr) { + // NDKPublishError is expected when not all relays in the batch succeed. + // The live listeners already captured successes/failures. + } + console.log(`[Worker] Batch ${bi + 1}/${batches.length} complete — ` + + `${publishedSoFar.size}/${totalTarget} succeeded, ` + + `${failedSoFar.size + timedOutSoFar.size} failed/timeout so far`); + } + + // Final broadcast progress message with complete results. const totalFailed = failedSoFar.size + timedOutSoFar.size; broadcast({ type: 'broadcastProgress', @@ -6364,6 +6332,15 @@ async function handlePublish(requestId, event, port) { } catch (_cleanupErr) { // removeAllListeners may not exist on all EventEmitter impls; ignore. } + // relaySet stays null for broadcasts — we use publishedSoFar for results. + } else { + // Normal (non-broadcast) publish — await the full result as before. + try { + relaySet = await ndkEvent.publish(null, undefined, undefined); + } catch (publishError) { + console.warn('[Worker] publish() threw:', publishError?.message || publishError); + relaySet = null; + } } // For non-broadcast publishes, emit a simplified broadcastProgress 'done'