Fix concurrent-processing on Windows (per-thread SQLite connection) + Windows portability

concurrent-work/harness/REPRODUCE-WINDOWS.md

@@ -0,0 +1,47 @@
+# Reproducing the concurrency benchmarks on Windows
+
+The suite was authored on macOS/Linux (see [`REPRODUCE.md`](REPRODUCE.md)). This note covers the
+Windows-specific setup. Everything below uses the pinned engine (`server-v3.2.1`) and pinned LangChain
+(`langchain-core 0.3.86`), exactly like the macOS/Linux runs.
+
+## Setup (git-bash / MSYS)
+
+```bash
+cd rocketride-bench
+bash scripts/provision.sh --competitors   # Windows: pulls the win64 .zip engine + builds .venv
+                                           # (Scripts/ layout) + real LangChain
+ROCKETRIDE_PORT=5565 bash scripts/start_engine.sh    # engine.exe ai/eaas.py --port=5565
+```
+
+Notes specific to Windows:
+- The pinned release ships a **`win64.zip`** engine; `provision.sh` unzips it to `./engine` (the binary
+  is `engine.exe`). macOS/Linux still download the `.tar.gz`.
+- If the **RocketRide VS Code extension** is installed, its own engine may already hold `:5565`. Start the
+  benchmark engine on a **free port** and point the harness at it:
+  ```bash
+  ROCKETRIDE_PORT=5566 bash scripts/start_engine.sh
+  export ROCKETRIDE_URI="ws://localhost:5566"
+  ```
+- The runners force UTF-8 stdout (`harness/__init__.py`), so unicode output no longer crashes the
+  legacy-codepage (cp1252) Windows console.
+- Temp paths default to the OS temp dir (`%LOCALAPPDATA%\Temp`) instead of `/tmp`. Override the shared
+  params/db locations with `$ROCKETRIDE_BENCH_PARAMS` / `$BENCH_DB_DIR` if the engine and harness resolve
+  different temp dirs (e.g. run under different accounts).
+
+## Run
+
+```bash
+# from concurrent-work/harness/ :
+python run_isolated_windows.py
+```
+
+## Note on the workload SQLite connection
+
+The `concurrent-processing` workload node uses a **per-thread cached SQLite connection**
+(`conn="thread_local"` in `nodes/workload/IInstance.py`). SQLite connections are thread-affine, and an
+engine may dispatch a pipe's *sequential* documents across more than one OS worker thread on some
+platforms (observed on the `win64` `server-v3.2.1` build even at `threads=1`; macOS/Linux use one thread
+per pipe). A per-thread connection is safe under either topology, so the headline RocketRide cell runs
+clean regardless of OS. The `rr_appendix_threads4` honesty cell deliberately keeps the naive single
+shared connection (`conn="module"`) — it reproduces the `sqlite3.ProgrammingError` trap on purpose, to
+show what a naive shared handle does when a pipe genuinely uses multiple threads.

concurrent-work/harness/rocketride-bench/groups/scale-and-concurrency/concurrent-processing/run.py

@@ -4,7 +4,8 @@
 AST-asserted below) over N=64 docs:
 
   RocketRide:  M warm resident pipes (each its own runtime process; "each pipe is its own
-               data" — the node's module-level connection is per-process by construction).
+               data" — the node uses a per-thread cached connection, so it is robust to the
+               engine's per-pipe thread model on every OS).
   LangChain:   ONE chain object, the three idiomatic ways to run it over 64 inputs:
                  .batch   (shared conn)        -> sqlite3.ProgrammingError   (CRASH)
                  .abatch  (blocking sync work) -> event loop serializes      (SEQUENTIAL)
@@ -39,7 +40,9 @@
 N_DOCS = 64
 IO_S = 0.100
 MS = [int(x) for x in os.environ.get("BENCH_MS", "8,16,64").split(",")]  # env-configurable; default = upstream
-DB_DIR = "/tmp/rr_bench_sqlite"
+# OS temp dir (no /tmp on Windows); override with $BENCH_DB_DIR. The harness and the node both
+# read/write these sqlite files, so the path must be valid on the host running both.
+DB_DIR = os.environ.get("BENCH_DB_DIR", os.path.join(tempfile.gettempdir(), "rr_bench_sqlite"))
 PIPE = os.path.join(HERE, "pipeline.pipe")
 TRACE = os.path.join(HERE, "trace")
 
@@ -87,7 +90,7 @@ async def rr_cell(m):
     """N_DOCS through M warm resident pipes; evidence = RRBENCH markers + sqlite rows."""
     _clean_dbs()
     pipes.set_params(mode="sqlite", label="workload", io_ms=IO_S * 1000.0,
-                     db=DB_DIR + "/%d.db", conn="module")
+                     db=DB_DIR + "/%d.db", conn="thread_local")
     sampler = measure.RSSSampler(measure.task_kids, interval=0.05)
     async with TraceSink() as sink:
         sampler.start()
@@ -205,10 +208,14 @@ async def main():
     subprocess.run([sys.executable, os.path.join(REPO, "scripts", "make_diagrams.py"),
                     PIPE, os.path.join(HERE, "canvas")])
     vm = out["verdict_metrics"]
-    print("\nVERDICT @ top M: RR(top M) %.2fs ok | LC .batch(shared) %s (%s) | "
+    # Reflect the ACTUAL RR status (results.json already records rr_topM_ok) instead of a
+    # hardcoded "ok" — on Windows the RR cell can be `check` (see rr64["status"]), and the
+    # console line was the only place that masked it.
+    rr_status = "ok" if vm["rr_topM_ok"] else "CHECK (node errors — see rocketride[].status)"
+    print("\nVERDICT @ top M: RR(top M) %.2fs %s | LC .batch(shared) %s (%s) | "
           "LC .abatch(blocking) %.1fs | LC seq %.1fs"
-          % (vm["rr_topM_wall_s"], vm["lc_batch_shared_status"], vm["lc_batch_shared_error"],
-             vm["lc_abatch_blocking_wall_s"], vm["lc_seq_wall_s"]))
+          % (vm["rr_topM_wall_s"], rr_status, vm["lc_batch_shared_status"],
+             vm["lc_batch_shared_error"], vm["lc_abatch_blocking_wall_s"], vm["lc_seq_wall_s"]))
 
 
 if __name__ == "__main__":

concurrent-work/harness/rocketride-bench/harness/__init__.py

@@ -4,3 +4,14 @@
 server + SDK + tracer). It never reimplements engine behavior. See the repo README and
 NOTICE for the "what's RocketRide vs what's ours" breakdown.
 """
+# Force UTF-8 on stdout/stderr so the runners' unicode (→, ×, …) prints on any console.
+# Windows defaults to a legacy codepage (cp1252), where a bare `print("… → …")` raises
+# UnicodeEncodeError and aborts a run mid-benchmark. Guarded + idempotent; no-op on POSIX,
+# which is already UTF-8. reconfigure() exists on TextIOWrapper (Py3.7+).
+import sys as _sys
+
+for _stream in (_sys.stdout, _sys.stderr):
+    try:
+        _stream.reconfigure(encoding="utf-8")
+    except (AttributeError, ValueError):
+        pass

concurrent-work/harness/rocketride-bench/harness/config.py

@@ -13,13 +13,16 @@
 import platform
 import subprocess
 import sys
+import tempfile
 
 HARNESS_DIR = os.path.dirname(os.path.abspath(__file__))
 REPO_DIR = os.path.dirname(HARNESS_DIR)
 
-# Engine: env-first, else the provisioned ./engine inside the repo.
+# Engine: env-first, else the provisioned ./engine inside the repo. The prebuilt binary is
+# `engine.exe` on Windows and `engine` elsewhere — pick the right name so provenance() and
+# engine_version() find it instead of silently recording "unknown".
 ENGINE_DIR = os.environ.get("ENGINE_DIR") or os.path.join(REPO_DIR, "engine")
-ENGINE = os.path.join(ENGINE_DIR, "engine")
+ENGINE = os.path.join(ENGINE_DIR, "engine.exe" if os.name == "nt" else "engine")
 
 # Direct-connect server (the headline product path).
 URI = os.environ.get("ROCKETRIDE_URI", "ws://localhost:5565")
@@ -30,7 +33,11 @@
 RESULTS_DIR = os.path.join(REPO_DIR, "results")
 DATA_DIR = os.path.join(REPO_DIR, "data")
 NODES_DIR = os.path.join(REPO_DIR, "nodes")
-BENCH_PARAMS = os.environ.get("ROCKETRIDE_BENCH_PARAMS", "/tmp/rr_bench_params.json")
+# Params file the harness writes and the workload node reads. Default to the OS temp dir so it
+# is writable on Windows too (there is no /tmp); keep the node's fallback (nodes/workload/
+# IInstance.py) in sync. Override with $ROCKETRIDE_BENCH_PARAMS to pin an explicit shared path.
+BENCH_PARAMS = os.environ.get("ROCKETRIDE_BENCH_PARAMS",
+                              os.path.join(tempfile.gettempdir(), "rr_bench_params.json"))
 
 
 def engine_present():

concurrent-work/harness/rocketride-bench/nodes/workload/IInstance.py

@@ -30,20 +30,26 @@
 import json
 import os
 import sys
+import tempfile
 import time
 import threading
 
 from rocketlib import IInstanceBase
 
 _PARAMS = None
+# Cross-platform fallbacks (no /tmp on Windows). The harness normally passes explicit paths via
+# $ROCKETRIDE_BENCH_PARAMS and the `db` param, so these only bite the local-binary floor path;
+# kept in sync with harness/config.py so both sides resolve the same file when unset.
+_DEFAULT_PARAMS = os.path.join(tempfile.gettempdir(), "rr_bench_params.json")
+_DEFAULT_DB = os.path.join(tempfile.gettempdir(), "rr_bench_sqlite", "%d.db")
 
 
 def _params():
     global _PARAMS
     if _PARAMS is not None:
         return _PARAMS
     path = os.environ.get("ROCKETRIDE_BENCH_PARAMS") or os.environ.get("BENCH_PARAMS") \
-        or "/tmp/rr_bench_params.json"
+        or _DEFAULT_PARAMS
     p = {}
     try:
         with open(path) as f:
@@ -57,7 +63,7 @@ def _params():
     p.setdefault("url", os.environ.get("BENCH_URL", "http://127.0.0.1:8799/"))
     p.setdefault("label", os.environ.get("BENCH_LABEL", ""))
     # scale-and-concurrency params
-    p.setdefault("db", os.environ.get("BENCH_DB", "/tmp/rr_bench_sqlite/%d.db"))
+    p.setdefault("db", os.environ.get("BENCH_DB", _DEFAULT_DB))
     p.setdefault("io_ms", float(os.environ.get("BENCH_IO_MS", "0") or 0))
     p.setdefault("pdf", os.environ.get("BENCH_PDF", ""))
     p.setdefault("conn", os.environ.get("BENCH_CONN", "module"))  # "module" | "per_call"
@@ -83,7 +89,8 @@ def _sqlite_doc_work(conn, label, io_s):
         time.sleep(io_s)
 
 
-_CONN = None  # the NAIVE module-level connection (conn="module") — one per task process
+_CONN = None  # the NAIVE single shared connection (conn="module") — the LC-trap mirror used by the
+# rr_appendix_threads4 honesty cell; unsafe once a pipe uses >1 thread. The headline uses "thread_local".
 
 
 def _sqlite_conn(p):
@@ -102,6 +109,29 @@ def _sqlite_conn(p):
     return _CONN
 
 
+_CONN_TL = threading.local()  # per-thread cached connection (conn="thread_local"). Each pipe is its
+# own OS process; within a process the connection is cached PER OS THREAD, so it is never used off its
+# creating thread even if the engine dispatches a pipe's (sequential) docs across >1 worker thread
+# (e.g. on Windows) at threads=1. Docs run sequentially per pipe, so there is no lock contention.
+
+
+def _sqlite_conn_tl(p):
+    conn = getattr(_CONN_TL, "conn", None)
+    if conn is None:
+        import sqlite3
+
+        db = p["db"]
+        if "%d" in db:
+            db = db % os.getpid()
+        d = os.path.dirname(db)
+        if d:
+            os.makedirs(d, exist_ok=True)
+        conn = sqlite3.connect(db)
+        conn.execute("CREATE TABLE IF NOT EXISTS docs (id INTEGER PRIMARY KEY, content TEXT)")
+        _CONN_TL.conn = conn
+    return conn
+
+
 _DOC = None  # the NAIVE module-level fitz.Document (conn="module") — one per task process
 
 
@@ -195,6 +225,9 @@ def open(self, obj):
                             _sqlite_doc_work(conn, label or "doc", float(p["io_ms"]) / 1000.0)
                         finally:
                             conn.close()
+                    elif p["conn"] == "thread_local":
+                        _sqlite_doc_work(_sqlite_conn_tl(p), label or "doc",
+                                         float(p["io_ms"]) / 1000.0)
                     else:  # "module": the naive shared-connection idiom (the LC-trap mirror)
                         _sqlite_doc_work(_sqlite_conn(p), label or "doc",
                                          float(p["io_ms"]) / 1000.0)

concurrent-work/harness/rocketride-bench/scripts/provision.sh

@@ -7,46 +7,60 @@ REPO_DIR="$(cd "$(dirname "$0")/.." && pwd)"
 RR_ENGINE_VERSION="${RR_ENGINE_VERSION:-3.2.1}"
 ENGINE_DIR="${ENGINE_DIR:-$REPO_DIR/engine}"
 
+# Engine binary name differs on Windows (git-bash/MSYS): engine.exe vs engine.
+case "$(uname -s)" in MINGW*|MSYS*|CYGWIN*) ENGINE_BIN=engine.exe ;; *) ENGINE_BIN=engine ;; esac
+
 # 1) Engine: use an existing one, else download the prebuilt for this OS/arch.
-if [ -x "$ENGINE_DIR/engine" ]; then
-  echo "engine present: $ENGINE_DIR/engine"
+if [ -x "$ENGINE_DIR/$ENGINE_BIN" ]; then
+  echo "engine present: $ENGINE_DIR/$ENGINE_BIN"
 else
   os="$(uname -s)"; arch="$(uname -m)"
+  ext=tar.gz
   case "$os/$arch" in
     Darwin/arm64) plat=darwin-arm64 ;;
     Darwin/x86_64) plat=darwin-x64 ;;
     Linux/*)      plat=linux-x64 ;;
+    # Windows via git-bash/MSYS/Cygwin: the release ships a .zip (not .tar.gz) that extracts
+    # engine.exe at the archive root (no leading component to strip).
+    MINGW*/*|MSYS*/*|CYGWIN*/*) plat=win64; ext=zip ;;
     *) echo "unsupported $os/$arch; use Docker (ghcr.io/rocketride-org/rocketride-engine)"; exit 1 ;;
   esac
-  asset="rocketride-server-v${RR_ENGINE_VERSION}-${plat}.tar.gz"
+  asset="rocketride-server-v${RR_ENGINE_VERSION}-${plat}.${ext}"
   url="https://github.com/rocketride-org/rocketride-server/releases/download/server-v${RR_ENGINE_VERSION}/${asset}"
   echo "downloading $url"
   mkdir -p "$ENGINE_DIR"
-  curl -fL "$url" -o "/tmp/$asset"
-  tar -xzf "/tmp/$asset" -C "$ENGINE_DIR" --strip-components=1
+  tmp_asset="${TMPDIR:-/tmp}/$asset"
+  curl -fL "$url" -o "$tmp_asset"
+  if [ "$ext" = "zip" ]; then
+    unzip -oq "$tmp_asset" -d "$ENGINE_DIR"        # win64.zip: engine.exe at the root
+  else
+    tar -xzf "$tmp_asset" -C "$ENGINE_DIR" --strip-components=1
+  fi
   echo "extracted engine -> $ENGINE_DIR"
 fi
-( cd "$ENGINE_DIR" && ./engine --version 2>&1 | head -1 ) || true
+( cd "$ENGINE_DIR" && "./$ENGINE_BIN" --version 2>&1 | head -1 ) || true
 
 # NOTE: native prebuilts exist for darwin-arm64/darwin-x64/linux-x64/win64. On Apple Silicon the
 # Docker image (linux-x64) runs under emulation — for fair Mac numbers use the darwin-arm64
 # prebuilt; for fair Docker numbers run on a linux-x64 host and containerize the competitors too.
 
-# 2) Harness venv.
-if [ ! -x "$REPO_DIR/.venv/bin/python" ]; then
+# 2) Harness venv. venv lays python under Scripts/ on Windows, bin/ elsewhere.
+case "$(uname -s)" in MINGW*|MSYS*|CYGWIN*) VENV_PY="$REPO_DIR/.venv/Scripts/python.exe" ;;
+                     *) VENV_PY="$REPO_DIR/.venv/bin/python" ;; esac
+if [ ! -x "$VENV_PY" ]; then
   echo "creating venv"
   python3 -m venv "$REPO_DIR/.venv"
 fi
-"$REPO_DIR/.venv/bin/python" -m pip install --quiet --upgrade pip
-"$REPO_DIR/.venv/bin/python" -m pip install --quiet -r "$REPO_DIR/requirements.txt"
-echo "venv ready: $REPO_DIR/.venv  ($("$REPO_DIR/.venv/bin/python" -c 'import rocketride; print("rocketride", rocketride.__version__)'))"
+"$VENV_PY" -m pip install --quiet --upgrade pip
+"$VENV_PY" -m pip install --quiet -r "$REPO_DIR/requirements.txt"
+echo "venv ready: $REPO_DIR/.venv  ($("$VENV_PY" -c 'import rocketride; print("rocketride", rocketride.__version__)'))"
 
 # 3) Optional: the REAL LangChain competitor baseline for the Tier-1 head-to-heads (no infra, no
 #    creds — the model is a fixed-latency mock). `make provision-competitors` does the same thing.
 if [ "${1:-}" = "--competitors" ]; then
   echo "installing competitor baselines (real LangChain)"
-  "$REPO_DIR/.venv/bin/python" -m pip install --quiet -r "$REPO_DIR/requirements-competitors.txt"
-  echo "competitors ready: $("$REPO_DIR/.venv/bin/python" -c 'import langchain_core; print("langchain-core", langchain_core.__version__)')"
+  "$VENV_PY" -m pip install --quiet -r "$REPO_DIR/requirements-competitors.txt"
+  echo "competitors ready: $("$VENV_PY" -c 'import langchain_core; print("langchain-core", langchain_core.__version__)')"
 fi
 
 echo

concurrent-work/harness/rocketride-bench/scripts/start_engine.sh

@@ -10,8 +10,11 @@ PORT="${ROCKETRIDE_PORT:-5565}"
 LOG="${ENGINE_LOG:-$REPO_DIR/results/engine.log}"
 PIDFILE="$REPO_DIR/results/engine.pid"
 
-if [ ! -x "$ENGINE_DIR/engine" ]; then
-  echo "RocketRide runtime not found at $ENGINE_DIR/engine" >&2
+# Engine binary name is engine.exe on Windows (git-bash/MSYS), engine elsewhere.
+case "$(uname -s)" in MINGW*|MSYS*|CYGWIN*) ENGINE_BIN=engine.exe ;; *) ENGINE_BIN=engine ;; esac
+
+if [ ! -x "$ENGINE_DIR/$ENGINE_BIN" ]; then
+  echo "RocketRide runtime not found at $ENGINE_DIR/$ENGINE_BIN" >&2
   echo "  set \$ENGINE_DIR or run scripts/provision.sh to download the pinned prebuilt." >&2
   exit 1
 fi
@@ -28,8 +31,8 @@ done
 
 mkdir -p "$(dirname "$LOG")"
 cd "$ENGINE_DIR"
-echo "starting: $ENGINE_DIR/engine ai/eaas.py --host=0.0.0.0  (port $PORT)"
-nohup ./engine ai/eaas.py --host=0.0.0.0 >"$LOG" 2>&1 &
+echo "starting: $ENGINE_DIR/$ENGINE_BIN ai/eaas.py --host=0.0.0.0 --port=$PORT"
+nohup "./$ENGINE_BIN" ai/eaas.py --host=0.0.0.0 --port="$PORT" >"$LOG" 2>&1 &
 echo $! > "$PIDFILE"
 
 # Readiness = the HTTP server answers at all. /ping returns 401 without auth (that's still a
@@ -38,7 +41,7 @@ for _ in $(seq 1 60); do
   code="$(curl -s -o /dev/null -w '%{http_code}' "http://localhost:$PORT/ping" 2>/dev/null || echo 000)"
   if [ "$code" != "000" ]; then
     echo "engine healthy on :$PORT (HTTP $code, pid $(cat "$PIDFILE"))"
-    ./engine --version 2>&1 | head -1
+    "./$ENGINE_BIN" --version 2>&1 | head -1
     exit 0
   fi
   sleep 0.5