laantungir/client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Batched broadcast publishing: split 600+ relays into batches of 25, publish sequentially to avoid browser WebSocket connection limits. 10s per-relay timeout per batch.

Browse Source
This commit is contained in:
Laan Tungir
2026-06-30 10:21:55 -04:00
parent c9a20e63b5
commit 0f31e1c301
3 changed files with 183 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
plans/broadcast-batching-fix.md Normal file
View File

@@ -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

6
www/js/version.json
View File

@@ -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"
}

139
www/ndk-worker.js
View File

@@ -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'