fix(worker:): make the in-browser SAB transport reliable + prove parallel serve

Root-cause + fix for the flaky hangs in the `worker:` SAB transport, plus the
multithreaded-execution proof and documentation.

Primary bug (stale pthread heap views): the JS ring stubs read the channel through
emscripten's cached Module.HEAPU8 view, which goes stale on a pool pthread that passed
through a memory-growth event — its Atomics.load then never observes another thread's
stores into the live buffer, so a ring read sees avail=0 forever and hangs (native C++
pointers alias live memory and see the data, so the two sides diverge). Flaky because it
depended on which pthread DuckDB scheduled the scan onto; never reproduced in the
single-threaded native harness. Fixed host-side (haybarn-wasm) by reading wasmMemory.buffer.

Transport-side fixes here:
- SabInputStream::Read / SabOutputStream::Write re-publish the channel offset onto the
  executing pthread's realm (VgiSabEnsureChannelOnRealm) — vgiSab.base is per-realm and a
  scan can run on a different pthread than bound the connection.
- Unique claim-id STATE handoff (HDR_CLAIM_SEQ): slot_open writes a globally-unique id, the
  worker dispatcher waits for STATE to leave its served id — kills the ABA where
  release+immediate-reclaim (1->0->1) read as "not released" and wedged await_release.
- Proactive slot-free on the error path (fixes "channel exhausted" under reuse pressure).

Multithreading: parallel_probe (max_workers=4) fixture proves a single scan fans out to 4
concurrent worker: connections/slots/serve pthreads (browser E2E: maxConcurrency=4).

Native [sab-conn]/[sab-e2e] unit tests pass (57 assertions). Browser E2E reliably green
across fresh loads. Docs: sab_transport_abi.md (unique-claim-id + the two load-bearing
browser footguns) + new wasm-worker-transport.md.

Known limitation (documented, NOT a transport bug): a worker throw under threads>1 amid heavy
concurrent load occasionally leaves DuckDB-WASM's async query promise unsettled (Asyncify +
pthread + C++ exception race) — transport tears down cleanly; error test runs threads=1.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01S2jkoYMNsNYkcpSh6Fwk2c

Author: rustyconover

CLAUDE.md

@@ -836,6 +836,7 @@ Catalogs may register additional settings at `ATTACH` time (e.g., `greeting`, `m
 - `launch:<argv>` → **AF_UNIX** transport with the worker spawned via the launcher
 - `oci://<image>[:tag]` (or `docker://` alias) → **container** transport: the worker runs inside an OCI container via the host runtime (docker/podman/nerdctl/Apple `container`), wired over stdin/stdout like a subprocess and pooled the same way. See [docs/container-transport.md](docs/container-transport.md)
 - `github://owner/repo@tag/asset[#sha256=][#path=]` and `github-auto://owner/repo@tag[/prefix]` → **GitHub-release** transport (any subprocess-capable platform — POSIX **and** Windows; not Emscripten): download a worker executable from a GitHub release, SHA256-verify it, extract the full archive (`.tar.gz`/`.tar.zst`/`.zst`/`.gz`/bare via in-process Decompress; `.zip` via vendored miniz) into `${XDG_CACHE_HOME:-~/.cache}/vgi/releases`, and run it over the subprocess transport. All file I/O goes through DuckDB's cross-platform `FileSystem` (cross-process write-lock + atomic dir install); only the exec bit (`chmod`), tar symlinks, and macOS codesign stay POSIX. `github://` names the asset explicitly (+ optional `#sha256=` pin, enforced even on cache hit); `github-auto://` builds `{prefix=repo}-{tag}-{DuckDB-platform}.tar.gz` (`.zip` on Windows) and verifies against the published `.sha256` sidecar. Entrypoint = the single exec-bit member (tar) / single `.exe` (zip), or `#path=`. A DX convenience (overlaps `oci://`); SHA256-only (cosign provenance deferred); no allowlist; cwd inherited (publisher builds relocatable). See [docs/github-transport.md](docs/github-transport.md)
+- `worker:<url>` → **in-browser SAB** transport (Emscripten/DuckDB-WASM only): the worker runs client-side in a Web Worker, exchanging Arrow batches over a SharedArrayBuffer duplex-ring channel in DuckDB's linear memory — **no server**. Reuses the subprocess `FunctionConnection` streaming paths (only the byte transport changes: `Atomics`-blocked SAB rings vs an OS pipe). Parallel scans work (`max_workers>1` → one serve pthread per slot; browser E2E proves `maxConcurrency=4`). Two emscripten-pthread footguns are load-bearing — JS ring stubs must read `wasmMemory.buffer` (not the stale cached `Module.HEAPU8` view) and re-publish the per-realm channel offset before each ring op — else intermittent scheduler-dependent hangs. Files: `src/vgi_webworker_function_connection.cpp`, `src/vgi_sab_stream.cpp`, `src/include/vgi_sab_abi.hpp`; host glue in `haybarn-wasm` (`lib/js-stubs.js` + `vgi-webworker-bridge.ts`); worker via `vgi-rust`. See [docs/wasm-worker-transport.md](docs/wasm-worker-transport.md) + [docs/sab_transport_abi.md](docs/sab_transport_abi.md)
 
 The `launch:` and `unix://` paths share one warm worker process across every DuckDB instance that points at the same `(cmd, args, cwd, VGI_RPC_*-env)` tuple — coordinated system-wide via per-hash flock + AF_UNIX socket.  See `docs/launcher-protocol.md` for the wire-protocol contract (state-dir layout, hash inputs, lockfile semantics) shared with the Python reference launcher in `vgi-rpc/vgi_rpc/launcher.py`.
 

docs/sab_transport_abi.md

@@ -21,11 +21,14 @@ header (i32 lanes):
   [3] ring_cap       = bytes per ring (per direction)
   [4] slot_stride    = bytes per slot
   [5] slots_off      = byte offset of slot[0]  (= 64)
-  [6..15] reserved
+  [6] reserved
+  [7] ensure-worker ready flag (client-side dedup; 1 = worker booted)
+  [8] claim_seq      monotonic global claim-id counter (Atomics.add on slot_open)
+  [9..15] reserved
 
 slot (at slots_off + i*slot_stride):
   control (i32 lanes, first 64B, cache-line isolated):
-    [0] state          0 = free, 1 = claimed        (slot_open CAS target)
+    [0] state          0 = free, nonzero = unique claim id  (slot_open CAS target)
     [1] c2w_write_pos  monotonic bytes written (client → worker)
     [2] c2w_read_pos   monotonic bytes read
     [3] c2w_closed     1 = client finished writing (EOS on input)
@@ -39,6 +42,15 @@ slot (at slots_off + i*slot_stride):
 slot_stride = align_up(64 + 2*ring_cap, 64)
 ```
 
+**`state` carries a UNIQUE claim id, not a constant 1.** `slot_open` writes
+`Atomics.add(header[claim_seq], 1) + 1` (a globally-unique nonzero id) into the slot `state`,
+not `1`. The worker's per-slot dispatcher records the id it is serving and, after serving,
+waits for `state` to leave *that* id before accepting the next claim. A constant `1` would make
+a release+immediate-reclaim (`state 1 → 0 → 1`) an **ABA race** — the worker would read the new
+`1` as "my claim not released yet" and block forever. Unique ids make `state != served_id` true
+for both the free state and any fresh reclaim, so the handoff is unambiguous. (The native single-
+serve harness never reuses a slot mid-dispatch, so it may still store `1`.)
+
 **SPSC ring semantics** (per direction): positions are **monotonic** (never wrap; index =
 `pos % ring_cap`), so `avail = write_pos - read_pos`, `free = ring_cap - avail`. A writer
 that finds `free == 0` blocks on `read_pos` (waits for the reader to advance); a reader that
@@ -95,6 +107,31 @@ Errors surface **in-band** as Arrow error batches (reuse `ClassifyBatch`/
 `slot_read` block with a bounded timeout and re-check `context.client.interrupted`
 (adaptive backoff), mirroring `FdInputStream::Read`.
 
+## Browser implementation notes (load-bearing — non-obvious)
+
+Two emscripten-pthread footguns must be handled or the transport hangs **intermittently**
+(depending on which pool pthread the scheduler runs a scan on — so it passes in single-thread /
+native tests and flakes only in the real threaded browser):
+
+1. **Read the LIVE memory buffer, never the cached `Module.HEAPU8` view.** emscripten's cached
+   `Module.HEAP*` typed-array views go **stale** on a pthread that was mid-compute/blocked through
+   a memory-growth event — its `Atomics.load` then never observes another thread's stores into the
+   *current* buffer, so a ring read sees `avail == 0` forever even though the bytes are there (a
+   native C++ pointer, always aliasing live memory, sees them — so the two sides silently diverge).
+   The JS ring stubs must derive their views from `wasmMemory.buffer` (a live getter) on every op,
+   not `Module.HEAPU8.buffer`. This was the primary cause of the early flaky hangs.
+
+2. **Re-publish the channel offset onto the executing pthread's realm before each ring op.**
+   `vgiSab.base` (the offset the JS ring stubs index through) is per-realm JS state. A connection is
+   bound on one pthread, but DuckDB can schedule its ring I/O (tick/scan reads, input writes) onto a
+   different pool pthread whose realm never had the offset set → it indexes the wrong linear-memory
+   location. The C++ `SabInputStream::Read` / `SabOutputStream::Write` call a
+   `vgi_wasm_set_channel(EnsureVgiSabChannel())` shim first so the current realm is always correct.
+
+The native (POSIX-shm) harness has neither issue (native pointers + a single serve thread), so
+**both must be covered by the browser E2E** (`test/support/wasm-worker/browser-e2e/`), not the
+native `[sab-conn]` unit tests.
+
 ## Rust worker side
 
 `serve_sab` (in `vgi/src/transport.rs`) is thread-per-slot: each thread parks on its slot's

docs/wasm-worker-transport.md

@@ -0,0 +1,81 @@
+# In-browser `worker:` (SAB) transport
+
+Run a VGI worker **entirely in the browser** — in a Web Worker, exchanging Arrow batches with
+the DuckDB-WASM extension over a **SharedArrayBuffer duplex-ring channel** — with **no server**.
+This is the WASM-only sibling of the subprocess transport: same streaming `IFunctionConnection`
+paths, only the byte transport differs (SAB rings + `Atomics` instead of an OS pipe).
+
+```sql
+-- direct
+SELECT * FROM vgi_table_function('worker:/workers/example.js', 'count_to', [5]);
+-- catalog
+ATTACH 'worker:/workers/example.js' AS w (TYPE vgi);
+SELECT * FROM w.main.some_table;
+```
+
+`LOCATION 'worker:<url>'` is any URL `new Worker(url)` accepts (same-origin by default; the
+host bridge gates which URLs SQL may spawn). The worker's JS entry is either a Rust
+`-pthread`/emscripten worker or a plain JS/TS worker that speaks the slot protocol.
+
+## How it works (one paragraph)
+
+The channel is `malloc`'d **inside DuckDB's own wasm linear memory** (a `SharedArrayBuffer` under
+`-pthread`), so every DuckDB scan pthread reaches it natively and the VGI worker gets a view of it
+via one `postMessage` at spawn. It holds a header + N **slots**; each slot is a duplex pair of
+SPSC byte rings (client→worker, worker→client) with blocking `Atomics.wait`/`notify` flow control
+— the ring *is* the chunker, so batches larger than a ring stream through in bounded pieces. A
+connection claims a slot for its lifetime; the worker runs one serve thread per slot. Full wire
+contract: [sab_transport_abi.md](sab_transport_abi.md).
+
+## Multithreading
+
+Parallel table-function scans work: a scan whose worker advertises `max_workers > 1` fans out
+across DuckDB scan threads, each acquiring its own slot + its own worker serve pthread, so N
+serves run **concurrently**. The browser E2E proves this — a single `parallel_probe(max_workers=4)`
+scan under `SET threads=4` reaches an observed **peak concurrency of 4** (a process-global guard in
+the worker's shared memory counts simultaneously-active serves). Note: multiple *separate* queries
+serialize at DuckDB-WASM's single-threaded `AsyncDuckDB` worker boundary — parallelism comes from
+*one* fanned-out scan, not concurrent connections.
+
+## Build
+
+Three independently-built repos against the shared ABI:
+
+- **Extension** (`vgi`): built into the DuckDB-WASM COI build — `build-wasm-coi.sh` (SAB files are
+  `#if defined(__EMSCRIPTEN__)`). Native `[sab-conn]`/`[sab-e2e]` unit tests link the same code
+  with `-DVGI_SAB_NATIVE_TEST` against a POSIX-shm backend.
+- **Host glue** (`haybarn-wasm`): `lib/js-stubs.js` (the `--js-library` ring stubs) +
+  `packages/duckdb-wasm-app/src/lib/vgi-webworker-bridge.ts` (main-thread spawn bridge). Editing
+  `js-stubs.js` needs an engine re-link (`rm build/.../duckdb_wasm.{wasm,js}` — CMake misses
+  `--js-library` deps).
+- **Worker** (`vgi-rust` → `test/support/sabtable` + `test/support/wasm-worker/build.sh`): the Rust
+  serve framework built for `wasm32-unknown-emscripten -pthread`. `-Z build-std` needs
+  `RUSTFLAGS='-C target-feature=+atomics,+bulk-memory,+mutable-globals -C link-args=-pthread'` or
+  wasm-ld rejects `--shared-memory`.
+
+## Testing
+
+- **Native (the fast gate):** `[sab-conn]`/`[sab-e2e]` in `vgi_unit_tests` cover the transport
+  logic (bind→init→stream, multi-batch, worker error) over the in-process ring backend — but
+  single-threaded, so they cannot exercise the browser threading/heap-view paths.
+- **Browser E2E** (`test/support/wasm-worker/browser-e2e/`): the only test of the real stack
+  (engine + wasm extension + worker module + bridge + actual SAB/Atomics). Reliably green across
+  fresh loads: `LOAD` → direct scan → multi-batch → ATTACH+discovery → concurrent → worker-error →
+  parallel-serve proof (`maxConcurrency=4`). Headless Chromium under puppeteer can session-close
+  under nested-worker load; MCP Playwright / a real browser is stable. `probe-*.mjs` are isolation
+  harnesses (`VGI_ENTRY=probe-throw.mjs node serve.mjs`).
+
+## Reliability notes / known limitation
+
+The transport is reliable on the happy path (streaming, catalog, concurrent scans, parallel
+serve). Two emscripten-pthread footguns are handled and documented in
+[sab_transport_abi.md](sab_transport_abi.md) → *Browser implementation notes* (stale cached heap
+views → read `wasmMemory.buffer`; per-realm channel offset → re-publish before each ring op) —
+getting either wrong causes intermittent, scheduler-dependent hangs.
+
+**Known limitation:** when a worker **throws** under `threads > 1` amid the full suite's heavy
+concurrent load, DuckDB-WASM *occasionally* leaves the async query promise unsettled — a flaky
+DuckDB-WASM-side error-propagation race (Asyncify + pthread + C++ exception), **not** a transport
+bug: the transport tears down cleanly (all slots freed, rings closed, worker idle — verified), a
+worker throw propagates fine under `threads=4` in isolation, and a generic non-VGI parallel throw
+settles fine. The E2E runs the worker-error case under `threads=1` for determinism.

src/include/vgi_sab_abi.hpp

@@ -22,12 +22,14 @@ enum HeaderLane : int32_t {
 	HDR_RING_CAP = 3,     // bytes per ring (per direction)
 	HDR_SLOT_STRIDE = 4,  // bytes per slot
 	HDR_SLOTS_OFF = 5,    // byte offset of slot[0]
+	// lane 7 = ensure-worker ready flag (client-side, js-stubs.js)
+	HDR_CLAIM_SEQ = 8,    // monotonic global claim-id counter (Atomics.add on slot_open)
 };
 constexpr int32_t kHeaderBytes = 64;
 
 // ---- Per-slot control (i32 lane indices within the slot's control block) --
 enum SlotLane : int32_t {
-	SLOT_STATE = 0,         // 0 = free, 1 = claimed
+	SLOT_STATE = 0,         // 0 = free, nonzero = unique claim id (see below)
 	C2W_WRITE_POS = 1,      // client -> worker ring
 	C2W_READ_POS = 2,
 	C2W_CLOSED = 3,
@@ -37,7 +39,12 @@ enum SlotLane : int32_t {
 };
 constexpr int32_t kSlotControlBytes = 64; // control block, cache-line isolated
 
-// state values
+// state values. STATE=0 is free; a claim stores a UNIQUE nonzero id (from the
+// HDR_CLAIM_SEQ counter) rather than a constant 1, so the per-slot worker
+// dispatcher can tell "still my claim" from "already reclaimed by the next scan"
+// (STATE 1->0->1 would otherwise be an ABA race that wedges await_release). The
+// native harness uses a single serve per slot (await is a no-op there) and may
+// still store 1. kSlotClaimed is retained for that native path.
 constexpr int32_t kSlotFree = 0;
 constexpr int32_t kSlotClaimed = 1;
 // closed flag

src/vgi_sab_stream.cpp

@@ -7,6 +7,13 @@
 namespace duckdb {
 namespace vgi {
 
+// Publish the shared channel offset onto the calling pthread's JS realm before ring
+// I/O (defined in vgi_webworker_function_connection.cpp). A connection is bound on one
+// pthread but DuckDB may run its ring reads/writes on a different pool pthread whose
+// realm never had the offset set; without this the JS ring stubs read the wrong
+// linear-memory location and block. wasm-only; a no-op on the native test harness.
+void VgiSabEnsureChannelOnRealm();
+
 // ---- SabInputStream ---------------------------------------------------------
 
 SabInputStream::SabInputStream(int slot, ClientContext *context)
@@ -36,6 +43,7 @@ arrow::Result<int64_t> SabInputStream::Read(int64_t nbytes, void *out) {
 	if (!is_open_) {
 		return arrow::Status::IOError("SabInputStream: read on closed stream");
 	}
+	VgiSabEnsureChannelOnRealm(); // this pthread's realm must know the channel offset
 	auto *dst = static_cast<uint8_t *>(out);
 	int64_t total = 0;
 	while (total < nbytes) {
@@ -84,6 +92,7 @@ arrow::Status SabOutputStream::Write(const void *data, int64_t nbytes) {
 	if (!is_open_) {
 		return arrow::Status::IOError("SabOutputStream: write on closed stream");
 	}
+	VgiSabEnsureChannelOnRealm(); // this pthread's realm must know the channel offset
 	const auto *src = static_cast<const uint8_t *>(data);
 	int64_t total = 0;
 	while (total < nbytes) {

src/vgi_webworker_function_connection.cpp

@@ -243,6 +243,20 @@ namespace {
 // because WebWorkerInvokeUnary (below) calls it before that definition.
 int EnsureVgiSabChannel();
 } // namespace
+
+// Publish the shared channel's byte offset onto the CALLING pthread's JS realm.
+// `vgiSab.base` (the offset the JS ring stubs read through) is per-realm state, but a
+// connection is bound on one pthread while its ring I/O (ReadDataBatch tick/scan,
+// input writes) can be scheduled by DuckDB onto a DIFFERENT pool pthread whose realm
+// never had the offset set — its ring reads then look at the wrong linear-memory
+// location and block forever (an intermittent, thread-scheduling-dependent hang). The
+// SAB stream Read/Write call this first so the current realm is always correct. Cheap:
+// EnsureVgiSabChannel is a cached atomic load and set_channel is a JS field store.
+void VgiSabEnsureChannelOnRealm() {
+	vgi_wasm_set_channel(EnsureVgiSabChannel());
+}
+#else
+void VgiSabEnsureChannelOnRealm() {}
 #endif
 
 // Standalone unary RPC over the `worker:` SAB transport, for catalog RPCs (ATTACH
@@ -333,10 +347,12 @@ int EnsureVgiSabChannel() {
 	if (existing != 0) {
 		return existing;
 	}
-	// Concurrent slot count = the worker's serve-thread count. 4 covers typical
-	// concurrent scans (a bare Web Worker with many pthreads is heavy on the browser;
-	// AsyncDuckDB also serializes multi-connection queries, so deep concurrency is
-	// rarely realized). Raise if a parallel-scan workload needs more simultaneous slots.
+	// Concurrent slot count = the worker's serve-thread count, matched to the engine's
+	// max scan threads (4). Same-slot reuse is made safe by the worker-done-aware
+	// release (AwaitWorkerDoneThenRelease) + the unique claim-id STATE handoff, so no
+	// extra headroom is needed — a released slot is provably idle and cleanly reusable.
+	// (A bare Web Worker with many pthreads is heavy on the browser; keep this modest.
+	// PTHREAD_POOL_SIZE in the worker build must be >= this.)
 	const int n_slots = 4;
 	const int ring_cap = sab::kDefaultRingCap;
 	const int stride = sab::SlotStride(ring_cap);
@@ -772,6 +788,16 @@ std::shared_ptr<arrow::RecordBatch> WebWorkerFunctionConnection::ReadDataBatch()
 				// ignore secondary errors while draining post-error
 			}
 			data_finished_ = true;
+			// Free the slot NOW rather than waiting for the connection destructor:
+			// DuckDB(-wasm) may defer destroying an errored scan's local state well
+			// past the throw, so a parallel scan that follows would otherwise see this
+			// slot still claimed and fail "channel exhausted". The w2c drain above has
+			// already consumed the worker's output to EOS. Idempotent; the destructor's
+			// release then no-ops (slot_ < 0).
+			if (slot_ >= 0) {
+				vgi_wasm_slot_release(slot_);
+				slot_ = -1;
+			}
 			throw;
 		}
 

test/support/sabtable/Cargo.lock

@@ -20,12 +20,19 @@ path = "src/lib.rs"
 # (tokio); the worker falls back to the zero-dep FsStorage backend, so the native
 # staticlib builds with no heavy/C deps.
 vgi = { path = "../../../../vgi-rust/vgi", default-features = false }
-# Must match the exact vgi-rpc that `vgi` resolves to (crates.io 0.11.0) so the
-# `OutputCollector` / `Result` / `RpcError` types are one crate instance, not two.
-vgi-rpc = { version = "0.11.0", default-features = false, features = ["macros", "stream-codec"] }
+# Must match the exact vgi-rpc that `vgi` resolves to (local 0.12.0 via the
+# [patch.crates-io] below) so the `OutputCollector` / `Result` / `RpcError`
+# types are one crate instance, not two.
+vgi-rpc = { version = "0.12.0", default-features = false, features = ["macros", "stream-codec"] }
 arrow-array = "59"
 arrow-schema = "59"
 serde = { version = "1", features = ["derive"] }
 
 [profile.release]
 opt-level = 2
+
+# vgi (via ../../../../vgi-rust) resolves vgi-rpc to the local 0.12.0 checkout;
+# point this standalone crate at the same source so both unify to ONE vgi-rpc
+# instance. Remove once vgi-rpc 0.12.0 is on crates.io.
+[patch.crates-io]
+vgi-rpc = { path = "../../../../vgi-rpc-rust/vgi-rpc" }