From f66124b0bcc2aa1b1d2c7eb62743498606c70dc7 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Mon, 15 Jun 2026 20:19:39 -0400 Subject: [PATCH] streaming: wall-clock flush so low-rate emissions appear immediately MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Worker batching used to wait for 64KB / 4096 newlines before flushing. A program like (let loop ((n 0)) (when (zero? (modulo n 10000)) (display "Emitted at iteration: ") (display n) (newline)) (loop (+ n 1))) emits ~25 bytes every 10K iterations, so the user saw nothing for several seconds — fox reported "this used to work and doesn't seem to anymore" because the first emission was buried in the batch buffer. Added a 100ms wall-clock check alongside the existing size / newline thresholds. The first chunk to arrive triggers a flush (lastFlushTime starts at 0), and low-rate streams cap at ~10 flushes/sec. High-rate streams still hit the byte / newline ceilings first so chunk batching for tight (display) loops is unaffected. Verified headlessly: the infinite-emitter program now shows the first emission within a frame, subsequent ones streaming live as the loop iterates. --- wasm/app/worker.mjs | 12 +++++++++++- wasm/dist-repl/worker.mjs | 12 +++++++++++- www/playground/worker.mjs | 12 +++++++++++- www/repl/worker.mjs | 12 +++++++++++- 4 files changed, 44 insertions(+), 4 deletions(-) diff --git a/wasm/app/worker.mjs b/wasm/app/worker.mjs index a5f39b7..cae8b53 100644 --- a/wasm/app/worker.mjs +++ b/wasm/app/worker.mjs @@ -55,13 +55,22 @@ self.onmessage = async (e) => { // calls. const CHUNK_FLUSH_BYTES = 65536; const CHUNK_FLUSH_NEWLINES = 4096; + // Wall-clock flush — guarantees the first emission lands on the UI + // immediately and caps low-rate streams at ~10 flushes/sec. Without + // this, a (let loop ...) that emits every 10K iterations would + // accumulate 30 bytes per emit, take 2000+ emits to hit the 64KB + // threshold, and the user sees nothing for seconds. lastFlushTime + // starts at 0 so the first chunk to land always triggers a flush. + const FLUSH_INTERVAL_MS = 100; let chunkBuffer = ""; let chunkNewlines = 0; + let lastFlushTime = 0; function flushChunkBuffer() { if (!chunkBuffer) return; self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer }); chunkBuffer = ""; chunkNewlines = 0; + lastFlushTime = Date.now(); } try { const output = await evalOnTier( @@ -77,7 +86,8 @@ self.onmessage = async (e) => { if (chunk.charCodeAt(i) === 10) chunkNewlines++; } if (chunkBuffer.length >= CHUNK_FLUSH_BYTES - || chunkNewlines >= CHUNK_FLUSH_NEWLINES) { + || chunkNewlines >= CHUNK_FLUSH_NEWLINES + || (Date.now() - lastFlushTime) >= FLUSH_INTERVAL_MS) { flushChunkBuffer(); } }, diff --git a/wasm/dist-repl/worker.mjs b/wasm/dist-repl/worker.mjs index a5f39b7..cae8b53 100644 --- a/wasm/dist-repl/worker.mjs +++ b/wasm/dist-repl/worker.mjs @@ -55,13 +55,22 @@ self.onmessage = async (e) => { // calls. const CHUNK_FLUSH_BYTES = 65536; const CHUNK_FLUSH_NEWLINES = 4096; + // Wall-clock flush — guarantees the first emission lands on the UI + // immediately and caps low-rate streams at ~10 flushes/sec. Without + // this, a (let loop ...) that emits every 10K iterations would + // accumulate 30 bytes per emit, take 2000+ emits to hit the 64KB + // threshold, and the user sees nothing for seconds. lastFlushTime + // starts at 0 so the first chunk to land always triggers a flush. + const FLUSH_INTERVAL_MS = 100; let chunkBuffer = ""; let chunkNewlines = 0; + let lastFlushTime = 0; function flushChunkBuffer() { if (!chunkBuffer) return; self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer }); chunkBuffer = ""; chunkNewlines = 0; + lastFlushTime = Date.now(); } try { const output = await evalOnTier( @@ -77,7 +86,8 @@ self.onmessage = async (e) => { if (chunk.charCodeAt(i) === 10) chunkNewlines++; } if (chunkBuffer.length >= CHUNK_FLUSH_BYTES - || chunkNewlines >= CHUNK_FLUSH_NEWLINES) { + || chunkNewlines >= CHUNK_FLUSH_NEWLINES + || (Date.now() - lastFlushTime) >= FLUSH_INTERVAL_MS) { flushChunkBuffer(); } }, diff --git a/www/playground/worker.mjs b/www/playground/worker.mjs index a5f39b7..cae8b53 100644 --- a/www/playground/worker.mjs +++ b/www/playground/worker.mjs @@ -55,13 +55,22 @@ self.onmessage = async (e) => { // calls. const CHUNK_FLUSH_BYTES = 65536; const CHUNK_FLUSH_NEWLINES = 4096; + // Wall-clock flush — guarantees the first emission lands on the UI + // immediately and caps low-rate streams at ~10 flushes/sec. Without + // this, a (let loop ...) that emits every 10K iterations would + // accumulate 30 bytes per emit, take 2000+ emits to hit the 64KB + // threshold, and the user sees nothing for seconds. lastFlushTime + // starts at 0 so the first chunk to land always triggers a flush. + const FLUSH_INTERVAL_MS = 100; let chunkBuffer = ""; let chunkNewlines = 0; + let lastFlushTime = 0; function flushChunkBuffer() { if (!chunkBuffer) return; self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer }); chunkBuffer = ""; chunkNewlines = 0; + lastFlushTime = Date.now(); } try { const output = await evalOnTier( @@ -77,7 +86,8 @@ self.onmessage = async (e) => { if (chunk.charCodeAt(i) === 10) chunkNewlines++; } if (chunkBuffer.length >= CHUNK_FLUSH_BYTES - || chunkNewlines >= CHUNK_FLUSH_NEWLINES) { + || chunkNewlines >= CHUNK_FLUSH_NEWLINES + || (Date.now() - lastFlushTime) >= FLUSH_INTERVAL_MS) { flushChunkBuffer(); } }, diff --git a/www/repl/worker.mjs b/www/repl/worker.mjs index a5f39b7..cae8b53 100644 --- a/www/repl/worker.mjs +++ b/www/repl/worker.mjs @@ -55,13 +55,22 @@ self.onmessage = async (e) => { // calls. const CHUNK_FLUSH_BYTES = 65536; const CHUNK_FLUSH_NEWLINES = 4096; + // Wall-clock flush — guarantees the first emission lands on the UI + // immediately and caps low-rate streams at ~10 flushes/sec. Without + // this, a (let loop ...) that emits every 10K iterations would + // accumulate 30 bytes per emit, take 2000+ emits to hit the 64KB + // threshold, and the user sees nothing for seconds. lastFlushTime + // starts at 0 so the first chunk to land always triggers a flush. + const FLUSH_INTERVAL_MS = 100; let chunkBuffer = ""; let chunkNewlines = 0; + let lastFlushTime = 0; function flushChunkBuffer() { if (!chunkBuffer) return; self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer }); chunkBuffer = ""; chunkNewlines = 0; + lastFlushTime = Date.now(); } try { const output = await evalOnTier( @@ -77,7 +86,8 @@ self.onmessage = async (e) => { if (chunk.charCodeAt(i) === 10) chunkNewlines++; } if (chunkBuffer.length >= CHUNK_FLUSH_BYTES - || chunkNewlines >= CHUNK_FLUSH_NEWLINES) { + || chunkNewlines >= CHUNK_FLUSH_NEWLINES + || (Date.now() - lastFlushTime) >= FLUSH_INTERVAL_MS) { flushChunkBuffer(); } },