Raise transactor throughput and add a Cloudflare write benchmark

CONTRIBUTING.md

@@ -85,6 +85,57 @@ the stage. It requires:
 
 Set `KEEP_STAGE=1` to retain the temporary stage for investigation.
 
+## Operation benchmark
+
+```sh
+bun run bench:op [concurrency] [seconds] [transact|operation|both]
+```
+
+`bun run bench:op` drives the Cloudflare bench catalog through the
+in-process transactor over a local SQLite file, so the raw transaction path
+and the operation path can be compared and profiled without a deployment,
+for example with `bun --cpu-prof`. `BENCH_INDEX=1` keeps the production
+index thresholds instead of suppressing index runs.
+
+## Cloudflare benchmark
+
+```sh
+bun run bench:cf [lanes] [parallel] [seconds-per-phase]
+```
+
+`bun run bench:cf` deploys a throwaway benchmark stage with test hooks and a
+small operation catalog, drives the raw transaction path and the `/op` path,
+prints throughput and latency per phase with the transactor's batch
+statistics, and destroys the stage. It requires the same Cloudflare variables
+as the e2e suite.
+
+The load is generated inside Cloudflare so that the round trip from this
+machine does not bound the measurement. The stage exposes a lane route that
+calls the peer through a service binding to itself; `bench.ts` keeps `lanes`
+lane invocations running at once, each with `parallel` requests in flight
+(at most 6, the Workers simultaneous connection limit), so the transactor sees
+`lanes × parallel` concurrent writers. Each lane invocation returns after
+`BENCH_LANE_REQUESTS` requests (default 500) and is relaunched until the phase
+ends. `BENCH_PING=1` adds a phase that only reads transactor info, which
+measures the request ceiling of one Durable Object, and `BENCH_SOCKET=1`
+adds ping and write phases over a WebSocket into the same object. Each phase
+prints client-side throughput and latency, the peer's warn
+and error telemetry captured inside the stage, and the transactor's batch
+counters. Batch cadence is reported as wall-clock time per batch and per
+transaction because Workers do not advance timers during synchronous work.
+Past a few hundred concurrent writers the platform sheds load with
+"Durable Object is overloaded" errors, which the bench reports as server
+events, so raise `lanes` until that appears to find the ceiling of a single
+transactor.
+Transactor tuning variables such as `RAMOSE_MAX_BATCH`,
+`RAMOSE_BATCH_BUDGET_MS`, and `RAMOSE_OP_COALESCE_MS` (the window in which a
+Worker isolate groups concurrent operations into one transactor request,
+default 2, 0 to disable) are forwarded to the stage, and `BENCH_LABEL` tags
+the printed rows so runs can be compared. `KEEP_STAGE=1` retains the stage.
+`bun run bench:cf:compare` runs the same benchmark twice, first against the
+`master` package source and then against the current branch, so the two
+result tables can be compared directly.
+
 ## Documentation site
 
 ```sh

alchemy.run.ts

@@ -29,6 +29,8 @@ export const Server = Ramose.Server("Ramose", {
     RAMOSE_STAGE: stage,
     ...tuning(
       "RAMOSE_MAX_BATCH",
+      "RAMOSE_BATCH_BUDGET_MS",
+      "RAMOSE_OP_COALESCE_MS",
       "RAMOSE_QUERY_MAX_CELLS",
       "RAMOSE_LOG_LEVEL",
       "RAMOSE_INDEX_TX_THRESHOLD",

bench/cloudflare-compare.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+fail() { echo "error: $*" >&2; exit 1; }
+
+BASE="${BASE_REF:-master}"
+LANES="${1:-32}"
+PARALLEL="${2:-4}"
+SECONDS_PER_PHASE="${3:-15}"
+SRC="packages/ramose/src"
+
+[ -z "$(git status --porcelain -- "$SRC")" ] || fail "$SRC has uncommitted changes; commit or stash them first."
+git rev-parse --verify --quiet "$BASE" >/dev/null || fail "unknown base ref '$BASE' (set BASE_REF)."
+
+restore() { git checkout --quiet HEAD -- "$SRC"; }
+trap restore EXIT
+
+echo ">> Benchmarking $BASE package source with this branch's harness ..."
+git checkout --quiet "$BASE" -- "$SRC"
+BENCH_LABEL="$BASE" bash bench/cloudflare.sh "$LANES" "$PARALLEL" "$SECONDS_PER_PHASE"
+restore
+
+echo ">> Benchmarking $(git rev-parse --abbrev-ref HEAD) ..."
+BENCH_LABEL="$(git rev-parse --abbrev-ref HEAD)" bash bench/cloudflare.sh "$LANES" "$PARALLEL" "$SECONDS_PER_PHASE"

bench/cloudflare.sh

@@ -0,0 +1,90 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+fail() { echo "error: $*" >&2; exit 1; }
+
+if [ -z "${CLOUDFLARE_API_TOKEN:-}" ] && [ ! -f "${HOME}/.alchemy/profiles.json" ]; then
+  fail "set CLOUDFLARE_API_TOKEN and CLOUDFLARE_ACCOUNT_ID, or log in with 'bun alchemy login'."
+fi
+command -v bun >/dev/null 2>&1 || fail "bun is not on PATH."
+
+LANES="${1:-32}"
+PARALLEL="${2:-4}"
+SECONDS_PER_PHASE="${3:-15}"
+STACK="bench/cloudflare/alchemy.run.ts"
+
+echo ">> Building the package ..."
+bun run scripts/build-packages.ts
+
+STAGE="${ALCHEMY_STAGE:-bench-$(date +%s)-${RANDOM}}"
+DEPLOY_LOG="$(mktemp "${TMPDIR:-/tmp}/ramose-bench-deploy.XXXXXX.log")"
+
+export ALCHEMY_STATE=local
+export ALCHEMY_STAGE="$STAGE"
+export CI=1
+export RAMOSE_BENCH_CAPABILITY="${RAMOSE_BENCH_CAPABILITY:-$(bun -e 'console.log(crypto.randomUUID().replace(/-/g, "") + crypto.randomUUID().replace(/-/g, ""))')}"
+
+cleanup() {
+  local cleanup_status=0
+  if [ "${KEEP_STAGE:-0}" = "1" ]; then
+    echo ">> KEEP_STAGE=1 set; leaving stage '$STAGE' deployed. Destroy later with:" >&2
+    echo "   ALCHEMY_STATE=local CI=1 RAMOSE_BENCH_CAPABILITY=$RAMOSE_BENCH_CAPABILITY bun alchemy destroy $STACK --stage $STAGE --yes" >&2
+  else
+    echo ">> Destroying stage '$STAGE' ..." >&2
+    local ok=""
+    for attempt in $(seq 1 5); do
+      if bun alchemy destroy "$STACK" --stage "$STAGE" --yes; then ok=1; break; fi
+      echo ">> destroy attempt ${attempt} failed; retrying..." >&2
+      sleep 5
+    done
+    if [ -z "$ok" ]; then
+      echo "error: destroy failed for stage '$STAGE'; check the Cloudflare dashboard." >&2
+      cleanup_status=1
+    fi
+  fi
+  rm -f "$DEPLOY_LOG"
+  return "$cleanup_status"
+}
+
+on_exit() {
+  local status="$1"
+  trap - EXIT
+  if cleanup; then
+    exit "$status"
+  fi
+  [ "$status" -ne 0 ] && exit "$status"
+  exit 1
+}
+trap 'on_exit "$?"' EXIT
+
+echo ">> Deploying bench stage '$STAGE' to Cloudflare ..."
+bun alchemy deploy "$STACK" --stage "$STAGE" --yes 2>&1 | tee "$DEPLOY_LOG"
+
+extract_url() {
+  local key="$1"
+  grep -oE "${key}: \"https://[^\"]+\"" "$DEPLOY_LOG" | head -1 \
+    | sed -E "s/^${key}: \"//; s/\"$//"
+}
+
+URL="$(extract_url peerUrl)"
+if [ -z "$URL" ]; then
+  URL="$(grep -oE 'https://[a-zA-Z0-9._-]+\.workers\.dev' "$DEPLOY_LOG" | head -n1 || true)"
+fi
+[ -n "$URL" ] || fail "could not find a Worker URL in the deploy output."
+
+echo ">> Deployed bench peer: $URL"
+
+echo ">> Waiting for $URL/health ..."
+ok=""
+for _ in $(seq 1 30); do
+  if curl -fsS "$URL/health" >/dev/null 2>&1; then ok=1; break; fi
+  sleep 2
+done
+[ -n "$ok" ] || fail "peer did not become healthy at $URL/health within ~60s."
+echo ">> Peer is healthy."
+
+echo ">> Driving write benchmark from Cloudflare against $URL ..."
+RAMOSE_URL="$URL" bun run bench/cloudflare/bench.ts "$LANES" "$PARALLEL" "$SECONDS_PER_PHASE"

bench/cloudflare/alchemy.run.ts

@@ -0,0 +1,50 @@
+import * as Ramose from "ramose";
+import * as Alchemy from "alchemy";
+import * as Cloudflare from "alchemy/Cloudflare";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import { AUD, ISS, JWKS } from "../../test/local/auth-keys.ts";
+
+const stage = process.env.ALCHEMY_STAGE ?? process.env.STAGE ?? process.env.USER ?? "bench";
+const capability = process.env.RAMOSE_BENCH_CAPABILITY;
+if (capability === undefined || capability.length < 32) {
+  throw new Error("RAMOSE_BENCH_CAPABILITY must be set to at least 32 characters");
+}
+const tuning = (...names: string[]): Record<string, string> =>
+  Object.fromEntries(names.filter((n) => process.env[n] !== undefined).map((n) => [n, process.env[n]!]));
+
+export const Server = Ramose.Server("Bench", {
+  peer: "BenchPeer",
+  storage: Cloudflare.R2.Bucket("BenchStore", { forceDestroy: true }),
+  main: import.meta.resolve("./worker.ts"),
+  auth: { jwksJson: JWKS, issuers: ISS, aud: AUD },
+  env: {
+    BENCH_SELF: Cloudflare.Workers.Self,
+    RAMOSE_STAGE: stage,
+    RAMOSE_TEST_HOOKS: "1",
+    RAMOSE_TEST_CAPABILITY: capability,
+    ...tuning(
+      "RAMOSE_MAX_BATCH",
+      "RAMOSE_BATCH_BUDGET_MS",
+      "RAMOSE_OP_COALESCE_MS",
+      "RAMOSE_INDEX_TX_THRESHOLD",
+      "RAMOSE_INDEX_INTERVAL_MS",
+      "RAMOSE_INDEX_MAX_TXS_PER_RUN",
+      "RAMOSE_LOG_KEEP_TXS",
+      "RAMOSE_TIMING_YIELDS",
+      "RAMOSE_LOG_LEVEL",
+    ),
+  },
+});
+
+export default Alchemy.Stack(
+  "ramose-bench",
+  {
+    providers: Layer.mergeAll(Cloudflare.providers(), Ramose.providers()),
+    state: process.env.ALCHEMY_STATE === "local" ? Alchemy.localState() : Cloudflare.state(),
+  },
+  Effect.gen(function* () {
+    const server = yield* Server;
+    return { peerUrl: server.url };
+  }),
+);