diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 530365f18..c2eb41166 100644 --- a/CONTRIBUTING.md +++ b/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 diff --git a/alchemy.run.ts b/alchemy.run.ts index 71456122d..7a7bc4c38 100644 --- a/alchemy.run.ts +++ b/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", diff --git a/bench/cloudflare-compare.sh b/bench/cloudflare-compare.sh new file mode 100755 index 000000000..f5f0ea322 --- /dev/null +++ b/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" diff --git a/bench/cloudflare.sh b/bench/cloudflare.sh new file mode 100755 index 000000000..947dfc3dc --- /dev/null +++ b/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" diff --git a/bench/cloudflare/alchemy.run.ts b/bench/cloudflare/alchemy.run.ts new file mode 100644 index 000000000..a91df7085 --- /dev/null +++ b/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 => + 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 }; + }), +); diff --git a/bench/cloudflare/bench.ts b/bench/cloudflare/bench.ts new file mode 100644 index 000000000..ab582ebc5 --- /dev/null +++ b/bench/cloudflare/bench.ts @@ -0,0 +1,253 @@ +import { schemaTx } from "../../packages/ramose/src/db/internal.ts"; +import { signToken } from "../../packages/ramose/test/sign-local-token.ts"; +import { fmt, percentile } from "../lib.ts"; +import { BENCH_DATABASE, BenchSchema } from "./catalog.ts"; +import { + CAPABILITY_HEADER, + LANE_PATH, + MAX_LANE_PARALLEL, + MAX_LANE_REQUESTS, + type LaneReport, + type LaneRequest, + type LaneTarget, +} from "./lane.ts"; + +const url = process.env.RAMOSE_URL; +const capability = process.env.RAMOSE_BENCH_CAPABILITY; +if (!url || !capability) { + console.error("set RAMOSE_URL and RAMOSE_BENCH_CAPABILITY"); + process.exit(1); +} +const base = url.replace(/\/+$/, ""); +const lanes = Math.max(1, Number(process.argv[2] ?? 32)); +const parallel = Math.max(1, Math.min(MAX_LANE_PARALLEL, Number(process.argv[3] ?? 4))); +const seconds = Math.max(1, Number(process.argv[4] ?? 15)); +const laneRequests = Math.max( + parallel, + Math.min(MAX_LANE_REQUESTS, Number(process.env.BENCH_LANE_REQUESTS ?? 500)), +); +const label = process.env.BENCH_LABEL ?? ""; +const inFlight = lanes * parallel; + +type Exchange = { status: number; body: any; text: string }; + +const call = async (path: string, init: RequestInit, token?: string): Promise => { + const headers = new Headers(init.headers); + headers.set("content-type", "application/json"); + if (path.startsWith("/__test__/") || path.startsWith("/__bench__/")) { + headers.set(CAPABILITY_HEADER, capability); + } + if (token !== undefined) headers.set("authorization", `Bearer ${token}`); + const res = await fetch(`${base}${path}`, { ...init, headers }); + const text = await res.text(); + let body: unknown = null; + try { + body = text ? JSON.parse(text) : null; + } catch { + body = { error: text }; + } + return { status: res.status, body, text }; +}; + +const admin = (db: string, rest: string, body: unknown): Promise => + call(`/__test__/db/${encodeURIComponent(db)}${rest}`, { method: "POST", body: JSON.stringify(body) }); + +const requireOk = (what: string, r: Exchange): any => { + if (r.status !== 200) throw new Error(`${what} failed (${r.status}): ${r.text.slice(0, 400)}`); + return r.body; +}; + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +const settle = async (what: string, attempt: () => Promise, deadlineMs = 90_000): Promise => { + const deadline = Date.now() + deadlineMs; + let last: Exchange | undefined; + while (true) { + try { + last = await attempt(); + if (last.status === 200) return last.body; + if (last.status < 500 && last.status !== 404) break; + } catch (e) { + last = { status: 0, body: null, text: e instanceof Error ? e.message : String(e) }; + } + if (Date.now() >= deadline) break; + console.log(` ${what}: ${last.status} ${last.text.slice(0, 120)}; retrying while the deployment settles`); + await sleep(2_000); + } + return requireOk(what, last!); +}; + +const run = Date.now().toString(36); +let laneCounter = 0; + +const launchLane = async ( + target: LaneTarget, + db: string, + durationMs: number, + requests: number, + token?: string, +): Promise => { + const lane: LaneRequest = { + target, + db, + run, + lane: laneCounter++, + parallel, + maxRequests: requests, + durationMs, + ...(token === undefined ? {} : { token }), + }; + const r = await call(LANE_PATH, { method: "POST", body: JSON.stringify(lane) }); + requireOk("bench lane", r); + return r.body as LaneReport; +}; + +const load = async (name: string, target: LaneTarget, db: string, token?: string) => { + const warm = await launchLane(target, db, 10_000, parallel, token); + if (warm.ok === 0) { + throw new Error(`${name} warm-up failed: ${JSON.stringify(warm.failures)}`); + } + const lat: number[] = []; + const failures = new Map(); + const colos = new Map(); + const serverEvents = new Map(); + let done = 0, errors = 0, invocations = 0, laneErrors = 0; + const t0 = performance.now(); + const deadline = Date.now() + seconds * 1000; + await Promise.all(Array.from({ length: lanes }, async () => { + while (true) { + const remaining = deadline - Date.now(); + if (remaining < 200) break; + try { + const report = await launchLane(target, db, remaining, laneRequests, token); + invocations++; + done += report.ok; + errors += report.errors; + for (const ms of report.latencies) lat.push(ms); + for (const [error, count] of Object.entries(report.failures)) { + failures.set(error, (failures.get(error) ?? 0) + count); + } + colos.set(report.colo, (colos.get(report.colo) ?? 0) + 1); + for (const [event, count] of Object.entries(report.serverEvents)) { + serverEvents.set(event, (serverEvents.get(event) ?? 0) + count); + } + } catch (e) { + laneErrors++; + const message = e instanceof Error ? e.message : String(e); + failures.set(`lane: ${message}`, (failures.get(`lane: ${message}`) ?? 0) + 1); + await sleep(250); + } + } + })); + const ms = performance.now() - t0; + lat.sort((a, b) => a - b); + console.log( + `${name.padEnd(12)} lanes=${lanes} parallel=${parallel} in-flight=${inFlight}: ${done} ok in ${fmt(ms, 0)} ms → ${fmt((done / ms) * 1000, 0)}/s, errors=${errors}, lane invocations=${invocations}, lane failures=${laneErrors}`, + ); + console.log(` latency p50 ${fmt(percentile(lat, 50))} ms p95 ${fmt(percentile(lat, 95))} ms p99 ${fmt(percentile(lat, 99))} ms`); + console.log(` driver colos ${[...colos.entries()].map(([colo, n]) => `${colo}×${n}`).join(" ")}`); + for (const [error, count] of [...failures.entries()].sort((a, b) => b[1] - a[1]).slice(0, 5)) { + console.log(` error ×${count}: ${error}`); + } + for (const [event, count] of [...serverEvents.entries()].sort((a, b) => b[1] - a[1]).slice(0, 8)) { + console.log(` server ×${count}: ${event}`); + } + return { done, errors, ms, rate: (done / ms) * 1000, p50: percentile(lat, 50), p95: percentile(lat, 95), p99: percentile(lat, 99) }; +}; + +const transactorStats = async (db: string) => { + const info = requireOk("transactor info", await admin(db, "/info", {})); + const s = info.stats; + return { + txs: s.txs as number, + batches: s.batches as number, + avgBatch: s.batches ? s.txs / s.batches : 0, + maxBatch: s.maxBatch as number, + rejected: s.rejected as number, + indexRuns: s.indexRuns as number, + txPerSec: info.metrics.txPerSec as number, + queueDepth: info.metrics.queueDepth as number, + novelty: info.novelty as number, + opts: info.opts, + }; +}; + +type Stats = Awaited>; + +const serverSide = (before: Stats, after: Stats, wallMs: number) => { + const restarted = after.txs < before.txs || after.batches < before.batches; + if (restarted) { + console.log(" transactor restarted during the phase; server-side counters cover only the new instance"); + } + const txs = restarted ? after.txs : after.txs - before.txs; + const batches = restarted ? after.batches : after.batches - before.batches; + return { + "server txs": txs, + "server tx/s": after.txPerSec, + restarted, + rejected: restarted ? after.rejected : after.rejected - before.rejected, + "index runs": restarted ? after.indexRuns : after.indexRuns - before.indexRuns, + "wall ms/batch": +fmt(batches ? wallMs / batches : 0, 1), + "wall ms/tx": +fmt(txs ? wallMs / txs : 0, 2), + novelty: after.novelty, + }; +}; + +const attribute = (ident: string, type: string, extra: Record = {}) => ({ + ":db/ident": ident, + ":db/valueType": `:db.type/${type}`, + ":db/cardinality": ":db.cardinality/one", + ":db/optional": true, + ...extra, +}); + +const rows: Record[] = []; + +{ + const db = `bench-tx-${run}`; + await settle("transact schema", () => admin(db, "/transact", { + tx: [attribute(":k/id", "long", { ":db/unique": ":db.unique/identity" }), attribute(":k/v", "string")], + })); + const before = await transactorStats(db); + const r = await load("transact", "transact", db); + const after = await transactorStats(db); + const batches = Math.max(0, after.batches - before.batches); + rows.push({ phase: "transact", label, ...summary(r), batches, "avg batch": fmt(batches ? r.done / batches : 0, 1), "max batch": after.maxBatch, ...serverSide(before, after, r.ms) }); + console.log(` transactor opts ${JSON.stringify(after.opts)} batches=${batches} maxBatch=${after.maxBatch}`); +} + +if (process.env.BENCH_PING === "1") { + const db = `bench-tx-${run}`; + const r = await load("info", "info", db); + rows.push({ phase: "info", label, ...summary(r) }); +} + +if (process.env.BENCH_SOCKET === "1") { + const db = `bench-tx-${run}`; + const ping = await load("socket-ping", "socket-ping", db); + rows.push({ phase: "socket-ping", label, ...summary(ping) }); + const before = await transactorStats(db); + const r = await load("socket-write", "socket-write", db); + const after = await transactorStats(db); + const batches = Math.max(0, after.batches - before.batches); + rows.push({ phase: "socket-write", label, ...summary(r), batches, "avg batch": fmt(batches ? r.done / batches : 0, 1), "max batch": after.maxBatch, ...serverSide(before, after, r.ms) }); +} + +{ + const db = BENCH_DATABASE; + await settle("operation schema", () => admin(db, "/transact", { tx: schemaTx(BenchSchema) })); + const token = await signToken(db, "writer", "bench-writer", undefined, { exp: `${seconds + 300}s` }); + const before = await transactorStats(db); + const r = await load("operation", "operation", db, token); + const after = await transactorStats(db); + const batches = Math.max(0, after.batches - before.batches); + rows.push({ phase: "operation", label, ...summary(r), batches, "avg batch": fmt(batches ? r.done / batches : 0, 1), "max batch": after.maxBatch, ...serverSide(before, after, r.ms) }); + console.log(` transactor opts ${JSON.stringify(after.opts)} batches=${batches} maxBatch=${after.maxBatch}`); +} + +function summary(r: Awaited>) { + return { "per second": Math.round(r.rate), ok: r.done, errors: r.errors, "p50 ms": +fmt(r.p50), "p95 ms": +fmt(r.p95), "p99 ms": +fmt(r.p99) }; +} + +console.log(`\nCloudflare write throughput driven from Cloudflare with ${lanes} lanes × ${parallel} in flight (${inFlight} concurrent writers) over ${seconds}s per phase:`); +console.table(rows); diff --git a/bench/cloudflare/catalog.ts b/bench/cloudflare/catalog.ts new file mode 100644 index 000000000..f2670313a --- /dev/null +++ b/bench/cloudflare/catalog.ts @@ -0,0 +1,50 @@ +import * as EffectSchema from "effect/Schema"; +import { Entity, EntityId, Field, Schema, string } from "ramose/db"; + +export const BENCH_DATABASE = "bench-ops"; + +export const BenchUser = Entity("benchUser", { + sub: Field.unique(string(), "strict"), +}); + +export const BenchItem = Entity("benchItem", { + key: Field.unique(string(), "strict"), + value: string(), +}, { + operations: (Operation) => ({ + create: Operation({ + self: false, + input: EffectSchema.Struct({ + key: EffectSchema.String, + value: EffectSchema.String, + }), + output: EffectSchema.Struct({ id: EntityId }), + run(op, input) { + return { id: op.create({ key: input.key, value: input.value }) }; + }, + }), + }), +}); + +export const BenchSchema = Schema("bench-cloudflare", { + benchUser: BenchUser, + benchItem: BenchItem, +}); + +BenchSchema.applyPolicy( + { + principal: BenchUser.sub, + roles: ["writer"] as const, + }, + ({ policy, session }) => { + const writer = session.roles.writer; + policy.benchUser.read.where(writer); + policy.benchItem.read.where(writer); + policy.benchItem.operations.create.where(writer); + }, +); + +export const benchCatalogDeployment = Object.freeze({ + root: BenchSchema, + deployments: [{ database: BENCH_DATABASE }], +}); diff --git a/bench/cloudflare/lane.ts b/bench/cloudflare/lane.ts new file mode 100644 index 000000000..e25a2f3fb --- /dev/null +++ b/bench/cloudflare/lane.ts @@ -0,0 +1,212 @@ +export type LaneTarget = "transact" | "operation" | "info" | "socket-ping" | "socket-write"; + +export const MAX_LANE_PARALLEL = 6; +export const MAX_LANE_REQUESTS = 5_000; +export const LANE_PATH = "/__bench__/lane"; +export const SOCKET_PATH = "/__bench__/socket"; +export const CAPABILITY_HEADER = "x-ramose-test-capability"; + +export interface LaneRequest { + readonly target: LaneTarget; + readonly db: string; + readonly run: string; + readonly lane: number; + readonly parallel: number; + readonly maxRequests: number; + readonly durationMs: number; + readonly token?: string; +} + +export interface LaneReport { + readonly colo: string; + readonly serverEvents: Record; + readonly ok: number; + readonly errors: number; + readonly failures: Record; + readonly latencies: number[]; + readonly startedAt: number; + readonly endedAt: number; +} + +export interface LaneFetcher { + fetch(request: Request): Promise; +} + +const idOf = (lane: number, slot: number, i: number): number => + lane * 1_000_000 + slot * 100_000 + i; + +const buildRequest = ( + req: LaneRequest, + origin: string, + capability: string, + slot: number, + i: number, +): Request => { + const id = idOf(req.lane, slot, i); + const key = `${req.run}-${req.lane}-${slot}-${i}`; + if (req.target === "info") { + return new Request(`${origin}/__test__/db/${encodeURIComponent(req.db)}/info`, { + method: "POST", + headers: { "content-type": "application/json", [CAPABILITY_HEADER]: capability }, + body: "{}", + }); + } + if (req.target === "transact") { + return new Request(`${origin}/__test__/db/${encodeURIComponent(req.db)}/transact`, { + method: "POST", + headers: { "content-type": "application/json", [CAPABILITY_HEADER]: capability }, + body: JSON.stringify({ tx: [{ ":k/id": id, ":k/v": "x" }], clientTxId: key }), + }); + } + return new Request(`${origin}/db/${encodeURIComponent(req.db)}/op`, { + method: "POST", + headers: { + "content-type": "application/json", + ...(req.token === undefined ? {} : { authorization: `Bearer ${req.token}` }), + }, + body: JSON.stringify({ + invocationId: crypto.randomUUID(), + operation: { owner: { kind: "entity", name: "benchItem" }, localName: "create" }, + input: { key, value: "x" }, + }), + }); +}; + +type SocketFrame = { kind: string; id?: number; message?: string }; + +const runSocketLane = async ( + req: LaneRequest, + fetcher: LaneFetcher, + origin: string, + capability: string, + colo: string, + serverEvents: () => Record, +): Promise => { + const parallel = Math.max(1, Math.min(MAX_LANE_PARALLEL, Math.floor(req.parallel))); + const maxRequests = Math.max(1, Math.min(MAX_LANE_REQUESTS, Math.floor(req.maxRequests))); + const startedAt = Date.now(); + const deadline = startedAt + Math.max(0, req.durationMs); + const latencies: number[] = []; + const failures: Record = {}; + let ok = 0; + let errors = 0; + const response = await fetcher.fetch( + new Request(`${origin}${SOCKET_PATH}?db=${encodeURIComponent(req.db)}`, { + headers: { Upgrade: "websocket", [CAPABILITY_HEADER]: capability }, + }), + ); + const socket = (response as Response & { webSocket?: WebSocket }).webSocket; + if (response.status !== 101 || socket === undefined || socket === null) { + const text = await response.text(); + failures[`socket ${response.status} ${text.slice(0, 160)}`] = 1; + return { colo, serverEvents: serverEvents(), ok: 0, errors: 1, failures, latencies, startedAt, endedAt: Date.now() }; + } + socket.accept(); + const pending = new Map void }>(); + socket.addEventListener("message", (event) => { + if (typeof event.data !== "string") return; + let frame: SocketFrame; + try { + frame = JSON.parse(event.data) as SocketFrame; + } catch { + return; + } + if (typeof frame.id !== "number") return; + const waiter = pending.get(frame.id); + if (waiter === undefined) return; + pending.delete(frame.id); + waiter.resolve(frame); + }); + let closed: string | undefined; + socket.addEventListener("close", (event) => { + closed = `socket closed ${event.code} ${event.reason}`; + for (const [, waiter] of pending) waiter.resolve({ kind: "error", message: closed }); + pending.clear(); + }); + socket.addEventListener("error", () => { + closed = "socket error"; + }); + let nextId = 0; + let issued = 0; + const exchange = (frame: Record): Promise<{ frame: SocketFrame; ms: number }> => + new Promise((resolve) => { + const id = nextId++; + const t0 = performance.now(); + pending.set(id, { t0, resolve: (reply) => resolve({ frame: reply, ms: performance.now() - t0 }) }); + socket.send(JSON.stringify({ ...frame, id })); + }); + const slots = Array.from({ length: parallel }, async (_, slot) => { + let i = 0; + while (Date.now() < deadline && issued < maxRequests && closed === undefined) { + issued++; + const frame = req.target === "socket-ping" + ? { v: 1, kind: "ping" } + : { + v: 1, + kind: "write", + tx: [{ ":k/id": idOf(req.lane, slot, i), ":k/v": "x" }], + clientTxId: `${req.run}-${req.lane}-${slot}-${i}`, + }; + i++; + const { frame: reply, ms } = await exchange(frame); + latencies.push(Math.round(ms * 100) / 100); + const failed = reply.kind === "error"; + if (!failed) ok++; + else { + errors++; + const key = `frame error: ${(reply.message ?? "").slice(0, 160)}`; + failures[key] = (failures[key] ?? 0) + 1; + } + } + }); + await Promise.all(slots); + try { + socket.close(1000, "done"); + } catch {} + return { colo, serverEvents: serverEvents(), ok, errors, failures, latencies, startedAt, endedAt: Date.now() }; +}; + +export const runLane = async ( + req: LaneRequest, + fetcher: LaneFetcher, + origin: string, + capability: string, + colo: string, + serverEvents: () => Record, +): Promise => { + if (req.target === "socket-ping" || req.target === "socket-write") { + return runSocketLane(req, fetcher, origin, capability, colo, serverEvents); + } + const parallel = Math.max(1, Math.min(MAX_LANE_PARALLEL, Math.floor(req.parallel))); + const maxRequests = Math.max(1, Math.min(MAX_LANE_REQUESTS, Math.floor(req.maxRequests))); + const startedAt = Date.now(); + const deadline = startedAt + Math.max(0, req.durationMs); + const latencies: number[] = []; + const failures: Record = {}; + let ok = 0; + let errors = 0; + let issued = 0; + const slots = Array.from({ length: parallel }, async (_, slot) => { + let i = 0; + while (Date.now() < deadline && issued < maxRequests) { + issued++; + const t0 = performance.now(); + let error: string | undefined; + try { + const res = await fetcher.fetch(buildRequest(req, origin, capability, slot, i++)); + const text = await res.text(); + if (res.status !== 200) error = `${res.status} ${text.slice(0, 160)}`; + } catch (e) { + error = e instanceof Error ? e.message : String(e); + } + latencies.push(Math.round((performance.now() - t0) * 100) / 100); + if (error === undefined) ok++; + else { + errors++; + failures[error] = (failures[error] ?? 0) + 1; + } + } + }); + await Promise.all(slots); + return { colo, serverEvents: serverEvents(), ok, errors, failures, latencies, startedAt, endedAt: Date.now() }; +}; diff --git a/bench/cloudflare/worker.ts b/bench/cloudflare/worker.ts new file mode 100644 index 000000000..e5f97d332 --- /dev/null +++ b/bench/cloudflare/worker.ts @@ -0,0 +1,89 @@ +import * as Effect from "effect/Effect"; +import { deployOperationCatalogs } from "ramose/worker"; +import { setTelemetrySink, type TelemetryEvent } from "../../packages/ramose/src/internal/core/telemetry.ts"; +import type { RamoseEnv } from "../../packages/ramose/src/RamoseEnv.ts"; +import { + QueryReplicaDO, + createServer, + createTransactorDO, +} from "../../packages/ramose/src/worker/testing.ts"; +import { benchCatalogDeployment } from "./catalog.ts"; +import { internalHeaders } from "../../packages/ramose/src/internal/transactor/internal.ts"; +import { + CAPABILITY_HEADER, + LANE_PATH, + SOCKET_PATH, + type LaneFetcher, + type LaneRequest, + runLane, +} from "./lane.ts"; + +type BenchEnv = RamoseEnv & { BENCH_SELF: LaneFetcher }; + +const operationCatalogs = await Effect.runPromise( + deployOperationCatalogs(benchCatalogDeployment), +); + +const server = createServer({ operationCatalogs }); + +const SERVER_EVENT_LIMIT = 200; +const serverEvents = new Map(); +setTelemetrySink((e: TelemetryEvent) => { + console.log(JSON.stringify(e)); + if (e.level !== "warn" && e.level !== "error") return; + const detail = typeof e.error === "string" ? e.error : typeof e.code === "string" ? e.code : ""; + const key = `${e.component}.${e.event}${detail === "" ? "" : `: ${detail.slice(0, 160)}`}`; + if (serverEvents.size >= SERVER_EVENT_LIMIT && !serverEvents.has(key)) return; + serverEvents.set(key, (serverEvents.get(key) ?? 0) + 1); +}); +const drainServerEvents = (): Record => { + const out = Object.fromEntries(serverEvents); + serverEvents.clear(); + return out; +}; + +const sameSecret = (left: string, right: string): boolean => { + if (left.length !== right.length) return false; + let difference = 0; + for (let index = 0; index < left.length; index++) { + difference |= left.charCodeAt(index) ^ right.charCodeAt(index); + } + return difference === 0; +}; + +const grantedCapability = (request: Request, env: BenchEnv): string | undefined => { + const capability = env.RAMOSE_TEST_CAPABILITY; + if (env.RAMOSE_TEST_HOOKS !== "1" || capability === undefined || capability.length < 32) { + return undefined; + } + const supplied = request.headers.get(CAPABILITY_HEADER) ?? ""; + return sameSecret(supplied, capability) ? capability : undefined; +}; + +export default { + async fetch(request: Request, env: BenchEnv, ctx: ExecutionContext): Promise { + const url = new URL(request.url); + if (url.pathname === SOCKET_PATH) { + const capability = grantedCapability(request, env); + if (capability === undefined) return new Response("not found", { status: 404 }); + const db = url.searchParams.get("db") ?? ""; + const stub = env.TRANSACTOR.get(env.TRANSACTOR.idFromName(db)); + return stub.fetch(`https://transactor/subscribe?writer=1&db=${encodeURIComponent(db)}`, { + headers: { Upgrade: "websocket", ...internalHeaders(env) }, + }); + } + if (url.pathname === LANE_PATH && request.method === "POST") { + const capability = grantedCapability(request, env); + if (capability === undefined) return new Response("not found", { status: 404 }); + const lane = (await request.json()) as LaneRequest; + const colo = (request.cf as { colo?: string } | undefined)?.colo ?? "unknown"; + const report = await runLane(lane, env.BENCH_SELF, url.origin, capability, colo, drainServerEvents); + return new Response(JSON.stringify(report), { + headers: { "content-type": "application/json" }, + }); + } + return server.fetch(request, env, ctx); + }, +}; +export { QueryReplicaDO }; +export const TransactorDO = createTransactorDO(operationCatalogs); diff --git a/bench/operation.bench.ts b/bench/operation.bench.ts new file mode 100644 index 000000000..5d351cb0e --- /dev/null +++ b/bench/operation.bench.ts @@ -0,0 +1,119 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import * as Effect from "effect/Effect"; +import * as Result from "effect/Result"; +import { schemaTx } from "../packages/ramose/src/db/internal.ts"; +import { DatabaseId } from "../packages/ramose/src/internal/authorization/index.ts"; +import type { AuthenticatedCaller } from "../packages/ramose/src/internal/authorization/request.ts"; +import { + deployOperationCatalogsForVersion, + deployedOperationCatalogs, +} from "../packages/ramose/src/worker/operation-catalogs.ts"; +import { BENCH_DATABASE, BenchSchema, benchCatalogDeployment } from "./cloudflare/catalog.ts"; +import { fmt, percentile } from "./lib.ts"; +import { BenchHarness, attribute } from "./transactor-harness.ts"; + +const conc = Number(process.argv[2] ?? 64); +const seconds = Number(process.argv[3] ?? 5); +const mode = process.argv[4] ?? "both"; + +const dir = mkdtempSync(join(tmpdir(), "ramose-op-")); + +const operationCatalogs = await Effect.runPromise(deployOperationCatalogsForVersion( + benchCatalogDeployment, + { id: "bench", tag: "bench", timestamp: "1970-01-01T00:00:00.000Z" }, +)); +const deployed = deployedOperationCatalogs(operationCatalogs); +const database = DatabaseId.make(BENCH_DATABASE); +const bound = Result.getOrThrow(deployed.requireDatabase(database)); +const definition = bound.definition; + +const caller: AuthenticatedCaller = { + claims: { sub: "bench-writer" }, + classes: ["writer"], + exp: Math.floor(Date.now() / 1000) + 3600, +}; + +const harness = (label: string) => + new BenchHarness({ + dbName: BENCH_DATABASE, + file: join(dir, `${label}.sqlite`), + config: process.env.BENCH_INDEX === "1" ? {} : { indexTxThreshold: 1_000_000, indexIntervalMs: 1_000_000 }, + operationRuntime: { + catalogs: deployed, + environment: {}, + now: () => Date.now(), + }, + }); + +type Client = (id: number, i: number) => Promise; + +const drive = async (label: string, h: BenchHarness, client: Client) => { + const tx = h.transactor; + let done = 0, errors = 0; + let firstError: string | undefined; + const lat: number[] = []; + const deadline = Date.now() + seconds * 1000; + const run = async (id: number) => { + let i = 0; + while (Date.now() < deadline) { + const t0 = performance.now(); + try { + const result = await client(id, i++); + if (typeof result === "object" && result !== null && "_tag" in result && (result as { _tag: string })._tag !== "Completed") { + errors++; + firstError ??= JSON.stringify(result).slice(0, 200); + } else done++; + } catch (e) { + errors++; + firstError ??= e instanceof Error ? e.message : String(e); + } + lat.push(performance.now() - t0); + } + }; + const t0 = performance.now(); + await Promise.all(Array.from({ length: conc }, (_, i) => run(i))); + const ms = performance.now() - t0; + lat.sort((a, b) => a - b); + const rate = (done / ms) * 1000; + console.log(`${label.padEnd(10)} concurrency=${conc}: ${done} ok in ${fmt(ms, 0)} ms → ${fmt(rate, 0)}/s, errors=${errors}${firstError === undefined ? "" : ` (${firstError})`}`); + console.log(` ack p50 ${fmt(percentile(lat, 50))} ms p99 ${fmt(percentile(lat, 99))} ms batches=${tx.stats.batches} avgBatch=${fmt(tx.stats.txs / Math.max(1, tx.stats.batches), 1)} storage writes=${h.writes} µs/item=${fmt((ms * 1000) / Math.max(1, done))}`); + h.db.close(); + return rate; +}; + +const rows: Record[] = []; + +if (mode === "both" || mode === "transact") { + const h = harness("transact"); + const tx = h.transactor; + await tx.init(); + await tx.transact([attribute(":k/id", "long", { ":db/unique": ":db.unique/identity" }), attribute(":k/v", "string")]); + const rate = await drive("transact", h, (id, i) => tx.transact([{ ":k/id": id * 1_000_000 + i, ":k/v": "x" }])); + rows.push({ phase: "transact", "per second": Math.round(rate) }); +} + +if (mode === "both" || mode === "operation") { + const h = harness("operation"); + const tx = h.transactor; + await tx.init(); + await tx.transact(schemaTx(BenchSchema)); + await tx.provisionCatalog(definition); + const run = Date.now().toString(36); + const rate = await drive("operation", h, (id, i) => + tx.invoke({ + database, + catalogKey: definition.catalogKey, + unitHash: definition.unitHash, + owner: { kind: "entity", name: "benchItem" }, + localName: "create", + input: { key: `${run}-${id}-${i}`, value: "x" }, + caller, + invocationId: crypto.randomUUID(), + })); + rows.push({ phase: "operation", "per second": Math.round(rate) }); +} + +rmSync(dir, { recursive: true, force: true }); +console.table(rows); diff --git a/bench/transactor-harness.ts b/bench/transactor-harness.ts index 4ae151b81..e56c6bbc2 100644 --- a/bench/transactor-harness.ts +++ b/bench/transactor-harness.ts @@ -5,6 +5,7 @@ import { prefixedBucket, } from "../packages/ramose/src/internal/storage/index.ts"; import { MemoryBucket } from "../packages/ramose/src/internal/storage/memory.ts"; +import type { OperationRuntime } from "../packages/ramose/src/internal/authorization/operations-runtime.ts"; import { DEFAULT_CONFIG, type SocketLike, @@ -48,6 +49,7 @@ export type BenchHarnessOptions = { readonly dbName?: string; readonly file?: string; readonly config?: Partial; + readonly operationRuntime?: OperationRuntime; }; export class BenchHarness implements TransactorHost { @@ -74,7 +76,7 @@ export class BenchHarness implements TransactorHost { this.raw = bucket ?? new MemoryBucket(); this.bucket = prefixedBucket(this.raw, dbPrefix(this.dbName)); this.config = { ...DEFAULT_CONFIG, ...options.config }; - this.transactor = new Transactor(this); + this.transactor = new Transactor(this, options.operationRuntime); } transactionSync(run: () => A): A { diff --git a/package.json b/package.json index e1024f59c..68b545b70 100644 --- a/package.json +++ b/package.json @@ -36,9 +36,12 @@ "bench:join": "bun run bench/join.bench.ts", "bench:write": "bun run bench/write.bench.ts", "bench:transactor": "bun run bench/transactor.bench.ts", + "bench:op": "bun run bench/operation.bench.ts", "bench:write-do": "bun run bench/write-do.bench.ts", "bench:read-do": "bun run bench/read-do.bench.ts", "bench:indexer": "bun run bench/indexer.bench.ts", + "bench:cf": "bash bench/cloudflare.sh", + "bench:cf:compare": "bash bench/cloudflare-compare.sh", "dev": "bun run scripts/build-packages.ts && bun alchemy dev", "dev:todos": "bun run scripts/build-packages.ts && CI=${CI:-1} ALCHEMY_STATE=${ALCHEMY_STATE:-local} CLOUDFLARE_ACCOUNT_ID=${CLOUDFLARE_ACCOUNT_ID:-0123456789abcdef0123456789abcdef} CLOUDFLARE_API_TOKEN=${CLOUDFLARE_API_TOKEN:-x} bun alchemy dev examples/todos/alchemy.run.ts", "dev:reef": "bun run scripts/build-packages.ts && CI=${CI:-1} ALCHEMY_STATE=${ALCHEMY_STATE:-local} CLOUDFLARE_ACCOUNT_ID=${CLOUDFLARE_ACCOUNT_ID:-0123456789abcdef0123456789abcdef} CLOUDFLARE_API_TOKEN=${CLOUDFLARE_API_TOKEN:-x} bun alchemy dev examples/reef/alchemy.run.ts", diff --git a/packages/ramose/src/RamoseEnv.ts b/packages/ramose/src/RamoseEnv.ts index 77b41b313..100399e8a 100644 --- a/packages/ramose/src/RamoseEnv.ts +++ b/packages/ramose/src/RamoseEnv.ts @@ -26,6 +26,8 @@ export interface RamoseEnv { RAMOSE_GC_EVERY_N_INDEXES?: string; RAMOSE_RETAIN_ROOTS?: string; RAMOSE_MAX_BATCH?: string; + RAMOSE_BATCH_BUDGET_MS?: string; + RAMOSE_OP_COALESCE_MS?: string; RAMOSE_TIMING_YIELDS?: string; RAMOSE_QUERY_MAX_CELLS?: string; RAMOSE_LOG_LEVEL?: string; diff --git a/packages/ramose/src/internal/authorization/invocation-receipts.ts b/packages/ramose/src/internal/authorization/invocation-receipts.ts index 71ba23b87..1efa5f391 100644 --- a/packages/ramose/src/internal/authorization/invocation-receipts.ts +++ b/packages/ramose/src/internal/authorization/invocation-receipts.ts @@ -292,16 +292,53 @@ export const invocationDigestMaterial = ( }), }); -const hashCanonical = Effect.fn("Authorization.hashInvocationReceiptMaterial")( - function* (domain: string, material: JsonValue) { - return yield* Effect.tryPromise({ - try: () => sha256Hex( - UTF8.encode(`${domain}${canonicalizeJson(material)}`), - ), - catch: () => invalid("operation invocation must contain canonical JSON data"), - }); - }, -); + +const SCOPE_DIGEST_CACHE_LIMIT = 256; +const scopeDigests = new Map(); + +const scopeDigestFor = async ( + invocation: AuthoritativeOperationInvocation, +): Promise => { + const canonical = canonicalizeJson(invocationScopeMaterial(invocation)); + const cached = scopeDigests.get(canonical); + if (cached !== undefined) return cached; + const digest = await sha256Hex( + UTF8.encode(`${INVOCATION_SCOPE_DIGEST_DOMAIN}${canonical}`), + ); + if (scopeDigests.size >= SCOPE_DIGEST_CACHE_LIMIT) { + scopeDigests.delete(scopeDigests.keys().next().value!); + } + scopeDigests.set(canonical, digest); + return digest; +}; + +export const prepareInvocationReceiptDirect = async ( + invocation: AuthoritativeOperationInvocation, + operationVersion: OperationVersion, +): Promise => { + const invocationId = requireInvocationId(invocation.invocationId); + const principalId = invocationPrincipalId(invocation); + let scopeDigest: string; + let invocationDigest: string; + try { + [scopeDigest, invocationDigest] = await Promise.all([ + scopeDigestFor(invocation), + sha256Hex(UTF8.encode( + `${INVOCATION_DIGEST_DOMAIN}${canonicalizeJson(invocationDigestMaterial(invocation, operationVersion))}`, + )), + ]); + } catch { + throw invalid("operation invocation must contain canonical JSON data"); + } + return Object.freeze({ + version: INVOCATION_RECEIPT_VERSION, + principalId, + invocationId, + scopeDigest, + operationVersion, + invocationDigest, + }); +}; export const requireSuppliedOperationVersion = ( value: unknown, @@ -319,22 +356,11 @@ export const prepareInvocationReceipt = Effect.fn( invocation: AuthoritativeOperationInvocation, operationVersion: OperationVersion, ): Effect.fn.Return { - const invocationId = requireInvocationId(invocation.invocationId); - const principalId = invocationPrincipalId(invocation); - const [scopeDigest, invocationDigest] = yield* Effect.all([ - hashCanonical(INVOCATION_SCOPE_DIGEST_DOMAIN, invocationScopeMaterial(invocation)), - hashCanonical( - INVOCATION_DIGEST_DOMAIN, - invocationDigestMaterial(invocation, operationVersion), - ), - ]); - return Object.freeze({ - version: INVOCATION_RECEIPT_VERSION, - principalId, - invocationId, - scopeDigest, - operationVersion, - invocationDigest, + return yield* Effect.tryPromise({ + try: () => prepareInvocationReceiptDirect(invocation, operationVersion), + catch: (cause) => cause instanceof InvalidRequest + ? cause + : invalid("operation invocation must contain canonical JSON data"), }); }); diff --git a/packages/ramose/src/internal/authorization/operations-runtime.ts b/packages/ramose/src/internal/authorization/operations-runtime.ts index ba5e64043..72121fd59 100644 --- a/packages/ramose/src/internal/authorization/operations-runtime.ts +++ b/packages/ramose/src/internal/authorization/operations-runtime.ts @@ -28,7 +28,7 @@ import { } from "../../db/entityArg.ts"; import { markEngineTypeAssertion } from "../core/tx-provenance.ts"; import type { Connection, TxReport } from "../core/conn.ts"; -import { Index } from "../core/datom.ts"; +import { type Datom, Index } from "../core/datom.ts"; import type { Db, EntityRef } from "../core/db.ts"; import { toWireDatom } from "../core/log.ts"; import { query } from "../core/query/engine.ts"; @@ -140,6 +140,46 @@ export type CatalogOperationAdmission = { readonly expiresAtSeconds: number; readonly authoritativeNowMs: number; readonly target?: { readonly eid: number; readonly type: string }; + readonly targetWitness?: TargetVisibilityWitness; +}; + +type TargetVisibilityWitness = { + readonly eid: number; + readonly observations: readonly ReadAuthorizationObservation[]; + readonly lookupAttributeId: number | undefined; +}; + +const lookupAttributeIdOf = (db: Db, target: EntityRef): number | undefined => { + if (typeof target === "number") return undefined; + const ident = typeof target === "string" ? ":db/ident" : target[0]; + return db.schema.attr(ident)?.id; +}; + +const witnessTouchedBy = ( + witness: TargetVisibilityWitness, + db: Db, + datoms: readonly Datom[], +): boolean => { + const fieldAttributeId = (observation: ReadAuthorizationObservation & { readonly _tag: "field" }) => + observation.attributeId ?? db.schema.attr(observation.ident)?.id; + for (const datom of datoms) { + if (datom.e === witness.eid) return true; + if (witness.lookupAttributeId !== undefined && datom.a === witness.lookupAttributeId) return true; + for (const observation of witness.observations) { + if (observation.eid !== datom.e) continue; + switch (observation._tag) { + case "exists": + return true; + case "type": + if (datom.a === RAMOSE_TYPE) return true; + break; + case "field": + if (datom.a === fieldAttributeId(observation)) return true; + break; + } + } + } + return false; }; export const resolveOperationCatalog = Effect.fn( @@ -1223,6 +1263,7 @@ const targetAuthorizationState = async ( const captureInvocationReplayFence = async ( admission: CatalogOperationAdmission, dbAfter: Db, + txData: readonly Datom[], ): Promise => { const originalDb = admission.context.currentDb; const typeCache = new Map>(); @@ -1298,13 +1339,19 @@ const captureInvocationReplayFence = async ( }), }); } else { - const authorization = await targetAuthorizationState( - admission.resolved.deployed.definition, - admission.context.principal, - dbAfter, - admission.target.eid, - invocationTarget, - ); + const witness = admission.targetWitness; + const unchanged = witness !== undefined && + witness.eid === admission.target.eid && + !witnessTouchedBy(witness, originalDb, txData); + const authorization = unchanged + ? { visible: true as const, digest: "" } + : await targetAuthorizationState( + admission.resolved.deployed.definition, + admission.context.principal, + dbAfter, + admission.target.eid, + invocationTarget, + ); target = Object.freeze({ ...admission.target, referenceEid, @@ -1512,6 +1559,7 @@ const authorizeCatalogOperationOnDb = async ( } let target: { readonly eid: number; readonly type: string } | undefined; + let targetWitness: TargetVisibilityWitness | undefined; if (descriptor.id.target === "required") { if (invocation.target === undefined) throw deny(); if (replayFence !== undefined) { @@ -1582,8 +1630,26 @@ const authorizeCatalogOperationOnDb = async ( } } } else { - target = await resolveVisibleTarget(context, invocation.target); - if (target === undefined) throw deny(); + const eid = typeof invocation.target === "number" + ? invocation.target + : await context.currentDb.entid(invocation.target); + if (eid === undefined) throw deny(); + const authorization = await targetAuthorizationState( + deployed.definition, + context.principal, + context.currentDb, + eid, + invocation.target, + ); + if (!authorization.visible) throw deny(); + const concrete = await typeName(context.currentDb, eid); + if (concrete === undefined) throw deny(); + target = { eid, type: concrete }; + targetWitness = { + eid, + observations: authorization.observations, + lookupAttributeId: lookupAttributeIdOf(context.currentDb, invocation.target), + }; } if (target === undefined || !typeCompatible(deployed.definition, descriptor.id.owner, target.type)) { throw deny(); @@ -1630,6 +1696,7 @@ const authorizeCatalogOperationOnDb = async ( expiresAtSeconds, authoritativeNowMs, ...(target === undefined ? {} : { target }), + ...(targetWitness === undefined ? {} : { targetWitness }), }); }; @@ -1793,6 +1860,7 @@ export const executeCatalogOperation = async ( const replayFence = await captureInvocationReplayFence( admission, report.dbAfter, + report.txData, ); const requested = invocation.allocations ?? []; const extraction = extractAllocations( diff --git a/packages/ramose/src/internal/authorization/read-filter.ts b/packages/ramose/src/internal/authorization/read-filter.ts index 1c36afcde..2a1b4514c 100644 --- a/packages/ramose/src/internal/authorization/read-filter.ts +++ b/packages/ramose/src/internal/authorization/read-filter.ts @@ -192,6 +192,39 @@ export const uniqueCanonicalTypeName = ( const viewKey = (db: Db): string => `${db.basisT}:${db.asOfT ?? ""}:${db.isHistory ? 1 : 0}`; +type PolicyTables = { + readonly rules: ReadonlyMap; + readonly entityDecisions: ReadonlyMap; + readonly traitDecisions: ReadonlyMap; + readonly fieldDecisions: ReadonlyMap; +}; + +const policyTablesByUnit = new WeakMap(); + +const policyTables = (unit: InstalledCatalogUnitV2): PolicyTables => { + const cached = policyTablesByUnit.get(unit); + if (cached !== undefined) return cached; + const rules = new Map(); + for (const rule of unit.policy.rules) { + rules.set(rule.id, rule.expr); + } + const entityDecisions = new Map(); + for (const entry of unit.policy.decisions.entities) { + entityDecisions.set(entry.target.name, entry.decision); + } + const traitDecisions = new Map(); + for (const entry of unit.policy.decisions.traits) { + traitDecisions.set(entry.target.name, entry.decision); + } + const fieldDecisions = new Map(); + for (const entry of unit.policy.decisions.fields) { + fieldDecisions.set(fieldKey(entry.target), entry.decision); + } + const tables: PolicyTables = { rules, entityDecisions, traitDecisions, fieldDecisions }; + policyTablesByUnit.set(unit, tables); + return tables; +}; + export const compileReadFilter = (input: CompileReadFilterInput): DatomPredicate => { try { return compilePredicate(input); @@ -220,23 +253,7 @@ const compilePredicate = (input: CompileReadFilterInput): DatomPredicate => { if (attr !== undefined) attrFields.set(attr.id, field); } - const rules = new Map(); - for (const rule of unit.policy.rules) { - rules.set(rule.id, rule.expr); - } - - const entityDecisions = new Map(); - for (const entry of unit.policy.decisions.entities) { - entityDecisions.set(entry.target.name, entry.decision); - } - const traitDecisions = new Map(); - for (const entry of unit.policy.decisions.traits) { - traitDecisions.set(entry.target.name, entry.decision); - } - const fieldDecisions = new Map(); - for (const entry of unit.policy.decisions.fields) { - fieldDecisions.set(fieldKey(entry.target), entry.decision); - } + const { rules, entityDecisions, traitDecisions, fieldDecisions } = policyTables(unit); const typeMemo = new Map>(); const rowMemo = new Map>(); diff --git a/packages/ramose/src/internal/authorization/validation/catalog.ts b/packages/ramose/src/internal/authorization/validation/catalog.ts index 4ed9eff81..cb9fed68b 100644 --- a/packages/ramose/src/internal/authorization/validation/catalog.ts +++ b/packages/ramose/src/internal/authorization/validation/catalog.ts @@ -209,9 +209,34 @@ const validateOperationShape = ( } }); +const preparedCatalogs = new WeakMap< + CatalogDescriptor, + Map> +>(); + +const bindingTargetKey = (target: CatalogBindingTarget): string => + `${target.database}\0${target.catalog}\0${target.catalogVersion}\0${target.schemaFingerprint}`; + export const prepareAuthorizationCatalog = ( target: CatalogBindingTarget, descriptor: CatalogDescriptor, +): Result.Result => { + let byTarget = preparedCatalogs.get(descriptor); + if (byTarget === undefined) { + byTarget = new Map(); + preparedCatalogs.set(descriptor, byTarget); + } + const key = bindingTargetKey(target); + const cached = byTarget.get(key); + if (cached !== undefined) return cached; + const prepared = prepareAuthorizationCatalogUncached(target, descriptor); + byTarget.set(key, prepared); + return prepared; +}; + +const prepareAuthorizationCatalogUncached = ( + target: CatalogBindingTarget, + descriptor: CatalogDescriptor, ): Result.Result => Result.gen(function* () { yield* validateTarget(target, descriptor); diff --git a/packages/ramose/src/internal/core/conn.ts b/packages/ramose/src/internal/core/conn.ts index 52758b49d..45bd1c539 100644 --- a/packages/ramose/src/internal/core/conn.ts +++ b/packages/ramose/src/internal/core/conn.ts @@ -1,7 +1,7 @@ import { type Datom, ALL_INDEXES, COMPARATORS, type IndexId, Index } from "./datom.ts"; import { Db, type Roots } from "./db.ts"; -import { Novelty } from "./novelty.ts"; -import { DB_IDENT, FIRST_USER_EID, Schema, bootstrapDatoms } from "./schema.ts"; +import { Novelty, type NoveltyView } from "./novelty.ts"; +import { DB_IDENT, FIRST_USER_EID, Schema, bootstrapDatoms, definesSchema } from "./schema.ts"; import { MemStore } from "./store.ts"; import { type BuildOptions, type NodeSource, type NodeStore, buildTree, mergeTree } from "./tree.ts"; import type { CompositionIndex } from "./composition.ts"; @@ -78,10 +78,12 @@ export async function mergeRoots( build?: BuildOptions, ): Promise { const pick = (i: IndexId) => novelty.byIndex[i].all().filter((d) => d.t <= maxT); - const eavt = await mergeTree(store, Index.EAVT, roots.eavt, pick(Index.EAVT), build); - const aevt = await mergeTree(store, Index.AEVT, roots.aevt, pick(Index.AEVT), build); - const avet = await mergeTree(store, Index.AVET, roots.avet, pick(Index.AVET), build); - const vaet = await mergeTree(store, Index.VAET, roots.vaet, pick(Index.VAET), build); + const [eavt, aevt, avet, vaet] = await Promise.all([ + mergeTree(store, Index.EAVT, roots.eavt, pick(Index.EAVT), build), + mergeTree(store, Index.AEVT, roots.aevt, pick(Index.AEVT), build), + mergeTree(store, Index.AVET, roots.avet, pick(Index.AVET), build), + mergeTree(store, Index.VAET, roots.vaet, pick(Index.VAET), build), + ]); return { t: maxT, eavt, aevt, avet, vaet }; } @@ -214,17 +216,17 @@ export class Connection { this.composition === undefined ? undefined : { composition: this.composition }, ); const schemaAfter = this.schema.clone().apply(res.datoms); - const noveltyAfter = new Novelty(); - noveltyAfter.add( - this.novelty.byIndex[Index.EAVT].all(), - (a) => schemaAfter.isAvet(a), - (a) => schemaAfter.isVaet(a), - ); - noveltyAfter.add( - res.datoms, - (a) => schemaAfter.isAvet(a), - (a) => schemaAfter.isVaet(a), - ); + const isAvet = (a: number) => schemaAfter.isAvet(a); + const isVaet = (a: number) => schemaAfter.isVaet(a); + let noveltyAfter: NoveltyView; + if (res.datoms.some((d) => definesSchema(d.a))) { + const copied = new Novelty(); + copied.add(this.novelty.byIndex[Index.EAVT].all(), isAvet, isVaet); + copied.add(res.datoms, isAvet, isVaet); + noveltyAfter = copied; + } else { + noveltyAfter = this.novelty.overlay(res.datoms, isAvet, isVaet); + } const dbAfter = new Db({ store: this.store, roots: this.roots, diff --git a/packages/ramose/src/internal/core/db.ts b/packages/ramose/src/internal/core/db.ts index ee6768b64..456cbfb0a 100644 --- a/packages/ramose/src/internal/core/db.ts +++ b/packages/ramose/src/internal/core/db.ts @@ -9,7 +9,7 @@ import { inferTag, normalizeValue, } from "./datom.ts"; -import { Novelty, collapseCurrent, currentView, filterAsOf, mergeChunks, rawView } from "./novelty.ts"; +import { type NoveltyView, collapseCurrent, currentView, filterAsOf, mergeChunks, rawView } from "./novelty.ts"; import { type Attribute, DB_IDENT, Schema, isTxEid } from "./schema.ts"; import { type NodeRef, type NodeSource, estimateCount, scan, scanMany, sortedUnion } from "./tree.ts"; import { COMPARATORS } from "./datom.ts"; @@ -46,7 +46,7 @@ export type DatomPredicate = ( export interface DbOptions { store: NodeSource; roots: Roots; - novelty: Novelty; + novelty: NoveltyView; basisT: number; schema: Schema; nextEid: number; @@ -59,7 +59,7 @@ export interface DbOptions { export class Db { readonly store: NodeSource; readonly roots: Roots; - readonly novelty: Novelty; + readonly novelty: NoveltyView; readonly basisT: number; readonly schema: Schema; readonly nextEid: number; diff --git a/packages/ramose/src/internal/core/novelty.ts b/packages/ramose/src/internal/core/novelty.ts index fd3bb0e0f..c23ea6ac7 100644 --- a/packages/ramose/src/internal/core/novelty.ts +++ b/packages/ramose/src/internal/core/novelty.ts @@ -10,9 +10,45 @@ import { } from "./datom.ts"; import { type Chunk, lowerBound, sortedUnion, upperBound } from "./tree.ts"; -export class SortedNovelty { - private base: Datom[] = []; +export interface SortedNoveltyView { + readonly index: IndexId; + readonly cmp: DatomComparator; + readonly size: number; + all(): readonly Datom[]; + range(prefix: Prefix): Chunk | undefined; +} + +export interface NoveltyView { + readonly byIndex: Record; + readonly count: number; + readonly maxT: number; +} + +function rangeOf(index: IndexId, ds: readonly Datom[], prefix: Prefix): Chunk | undefined { + if (ds.length === 0) return undefined; + const s = lowerBound(index, ds, prefix); + if (s >= ds.length) return undefined; + const e = upperBound(index, ds, prefix); + if (s >= e) return undefined; + return { datoms: ds, start: s, end: e }; +} + +function unionChunks(cmp: DatomComparator, chunks: readonly Chunk[]): Chunk | undefined { + if (chunks.length === 0) return undefined; + if (chunks.length === 1) return chunks[0]; + let merged = sliceOf(chunks[0]); + for (let i = 1; i < chunks.length; i++) merged = sortedUnion(cmp, merged, sliceOf(chunks[i])); + return { datoms: merged, start: 0, end: merged.length }; +} + +function sliceOf(c: Chunk): readonly Datom[] { + return c.start === 0 && c.end === c.datoms.length ? c.datoms : c.datoms.slice(c.start, c.end); +} + +export class SortedNovelty implements SortedNoveltyView { + private runs: Datom[][] = []; private pending: Datom[] = []; + private stored = 0; readonly cmp: DatomComparator; constructor(readonly index: IndexId) { @@ -20,53 +56,102 @@ export class SortedNovelty { } get size(): number { - return this.base.length + this.pending.length; + return this.stored + this.pending.length; } add(datoms: readonly Datom[]): void { for (const d of datoms) this.pending.push(d); } - private flush(): Datom[] { + private flush(): readonly Datom[][] { if (this.pending.length > 0) { const p = this.pending; this.pending = []; p.sort(this.cmp); - const dp: Datom[] = []; + const run: Datom[] = []; for (let i = 0; i < p.length; i++) { - if (i === 0 || this.cmp(p[i - 1], p[i]) !== 0) dp.push(p[i]); + if (i === 0 || this.cmp(p[i - 1], p[i]) !== 0) run.push(p[i]); + } + this.runs.push(run); + this.stored += run.length; + while (this.runs.length >= 2 && this.runs[this.runs.length - 1].length >= this.runs[this.runs.length - 2].length) { + const top = this.runs.pop()!; + const below = this.runs.pop()!; + const merged = sortedUnion(this.cmp, below, top); + this.stored += merged.length - top.length - below.length; + this.runs.push(merged); } - this.base = this.base.length === 0 ? dp : sortedUnion(this.cmp, this.base, dp); } - return this.base; + return this.runs; + } + + private compact(): Datom[] { + const runs = this.flush(); + if (runs.length === 0) return []; + if (runs.length === 1) return runs[0]; + let merged = runs[0]; + for (let i = 1; i < runs.length; i++) merged = sortedUnion(this.cmp, merged, runs[i]); + this.runs = [merged]; + this.stored = merged.length; + return merged; } all(): readonly Datom[] { - return this.flush(); + return this.compact(); } range(prefix: Prefix): Chunk | undefined { - const ds = this.flush(); - if (ds.length === 0) return undefined; - const s = lowerBound(this.index, ds, prefix); - if (s >= ds.length) return undefined; - const e = upperBound(this.index, ds, prefix); - if (s >= e) return undefined; - return { datoms: ds, start: s, end: e }; + const runs = this.flush(); + if (runs.length === 1) return rangeOf(this.index, runs[0], prefix); + const hits: Chunk[] = []; + for (const run of runs) { + const c = rangeOf(this.index, run, prefix); + if (c !== undefined) hits.push(c); + } + return unionChunks(this.cmp, hits); } dropThrough(maxT: number): void { - const ds = this.flush(); - this.base = ds.filter((d) => d.t > maxT); + const kept = this.compact().filter((d) => d.t > maxT); + this.runs = kept.length === 0 ? [] : [kept]; + this.stored = kept.length; } clear(): void { - this.base = []; + this.runs = []; this.pending = []; + this.stored = 0; } } -export class Novelty { +export class LayeredSortedNovelty implements SortedNoveltyView { + readonly index: IndexId; + readonly cmp: DatomComparator; + + constructor(private readonly base: SortedNoveltyView, private readonly top: SortedNoveltyView) { + this.index = base.index; + this.cmp = base.cmp; + } + + get size(): number { + return this.base.size + this.top.size; + } + + all(): readonly Datom[] { + return sortedUnion(this.cmp, this.base.all(), this.top.all()); + } + + range(prefix: Prefix): Chunk | undefined { + const hits: Chunk[] = []; + const b = this.base.range(prefix); + if (b !== undefined) hits.push(b); + const t = this.top.range(prefix); + if (t !== undefined) hits.push(t); + return unionChunks(this.cmp, hits); + } +} + +export class Novelty implements NoveltyView { readonly byIndex: Record = { 0: new SortedNovelty(0), 1: new SortedNovelty(1), @@ -76,6 +161,21 @@ export class Novelty { private _count = 0; private _maxT = 0; + overlay(datoms: readonly Datom[], avet: (a: number) => boolean, vaet: (a: number) => boolean): NoveltyView { + const top = new Novelty(); + top.add(datoms, avet, vaet); + return { + byIndex: { + 0: new LayeredSortedNovelty(this.byIndex[0], top.byIndex[0]), + 1: new LayeredSortedNovelty(this.byIndex[1], top.byIndex[1]), + 2: new LayeredSortedNovelty(this.byIndex[2], top.byIndex[2]), + 3: new LayeredSortedNovelty(this.byIndex[3], top.byIndex[3]), + }, + count: this.count + top.count, + maxT: Math.max(this.maxT, top.maxT), + }; + } + add(datoms: readonly Datom[], avet: (a: number) => boolean, vaet: (a: number) => boolean): void { if (datoms.length === 0) return; this.byIndex[0].add(datoms); diff --git a/packages/ramose/src/internal/core/schema.ts b/packages/ramose/src/internal/core/schema.ts index b73859f60..37aa74dd7 100644 --- a/packages/ramose/src/internal/core/schema.ts +++ b/packages/ramose/src/internal/core/schema.ts @@ -13,6 +13,22 @@ export const DB_DOC = 62; export const RAMOSE_TYPE_IDENT = ":ramose/type"; +export function definesSchema(a: number): boolean { + switch (a) { + case DB_IDENT: + case DB_VALUE_TYPE: + case DB_CARDINALITY: + case DB_UNIQUE: + case DB_IS_COMPONENT: + case DB_INDEX: + case DB_OPTIONAL: + case DB_DOC: + return true; + default: + return false; + } +} + export const FIRST_USER_EID = 1000; export const TX_BASE = 2 ** 42; export function txEid(t: number): number { diff --git a/packages/ramose/src/internal/core/tx.ts b/packages/ramose/src/internal/core/tx.ts index e1c10056e..a3670c684 100644 --- a/packages/ramose/src/internal/core/tx.ts +++ b/packages/ramose/src/internal/core/tx.ts @@ -275,6 +275,17 @@ export async function expandTx( if (op.kind === "add" && op.a === ":db/ident" && typeof op.v === "string") newIdents.set(op.v, op.e); } + const avetProbes = new Map>(); + const firstAvet = (a: number, tv: TaggedValue): Promise => { + const k = a + "|" + valueKey(tv.vt, tv.v); + let p = avetProbes.get(k); + if (p === undefined) { + p = db.first(Index.AVET, { a, vt: tv.vt, v: tv.v }); + avetProbes.set(k, p); + } + return p; + }; + const lookupEntid = async (form: [string, unknown]): Promise => { try { return await db.entid(form); @@ -384,7 +395,7 @@ export async function expandTx( } if (tv.v === undefined) continue; const claimKey = attr.id + "|" + valueKey(tv.vt, tv.v); - const existing = await db.first(Index.AVET, { a: attr.id, vt: tv.vt, v: tv.v }); + const existing = await firstAvet(attr.id, tv); const canonical = aliasOf(op.e); if (existing) { const prev = tempids.get(canonical); @@ -453,7 +464,7 @@ export async function expandTx( if (seen !== undefined && seen !== e) { throw new TxError(`unique conflict: ${attr.ident} ${String(tv.v)} already asserted for ${seen}`, "tx/unique-conflict"); } - const other = await db.first(Index.AVET, { a: attr.id, vt: tv.vt, v: tv.v }); + const other = await firstAvet(attr.id, tv); if (other && other.e !== e) { const ov = await current(other.e, attr.id); if (ov.has(vk)) { diff --git a/packages/ramose/src/internal/transactor/host.ts b/packages/ramose/src/internal/transactor/host.ts index 5b4395a2d..cbd6e526c 100644 --- a/packages/ramose/src/internal/transactor/host.ts +++ b/packages/ramose/src/internal/transactor/host.ts @@ -22,7 +22,9 @@ export interface TransactorConfig { gcEveryNIndexes: number; retainRoots: number; maxBatch: number; + batchBudgetMs: number; timingYields: boolean; + socketWrites: boolean; } export const DEFAULT_CONFIG: TransactorConfig = { @@ -33,7 +35,9 @@ export const DEFAULT_CONFIG: TransactorConfig = { gcEveryNIndexes: 50, retainRoots: 20, maxBatch: 0, + batchBudgetMs: 20, timingYields: false, + socketWrites: false, }; export interface TransactorHost { diff --git a/packages/ramose/src/internal/transactor/index.ts b/packages/ramose/src/internal/transactor/index.ts index 7b601e581..653a6d6ce 100644 --- a/packages/ramose/src/internal/transactor/index.ts +++ b/packages/ramose/src/internal/transactor/index.ts @@ -2,6 +2,6 @@ export { Transactor, TransactorDeadError, type OperationAck, type TxAck, type Tr export { DEFAULT_CONFIG, type SocketLike, type SqlLike, type SqlCursorLike, type TransactorConfig, type TransactorHost } from "./host.ts"; export { Indexer, type IndexerOptions, type IndexRunResult } from "./indexer.ts"; export { type RamoseEnv, envInt } from "./env.ts"; -export { INTERNAL_HEADER, internalGate, internalHeaders, isInternal } from "./internal.ts"; +export { INTERNAL_HEADER, ROUTE_ROOT_HEADER, internalGate, internalHeaders, isInternal } from "./internal.ts"; export { TxMetrics, type AnalyticsEngineDatasetLike, type BatchPoint, type IndexPoint } from "./observability.ts"; export { TxRejected, TransactorDead, BadRequest, NotFound, Internal, type TransactorHttpError, toHttpError, statusOf, errorResponse } from "./errors.ts"; diff --git a/packages/ramose/src/internal/transactor/indexer.ts b/packages/ramose/src/internal/transactor/indexer.ts index 0f226173f..16f3e5fec 100644 --- a/packages/ramose/src/internal/transactor/indexer.ts +++ b/packages/ramose/src/internal/transactor/indexer.ts @@ -32,6 +32,7 @@ export class Indexer { private runs = 0; private lastRun: IndexRunResult | undefined; private lastGc: unknown; + private armedAt: number | null | undefined; private readonly log = componentLogger("indexer"); constructor( @@ -53,18 +54,25 @@ export class Indexer { async maybeSchedule(): Promise { if (this.t.txsSinceIndex >= this.opts.txThreshold) { - await this.t.host.setAlarm(this.t.host.now()); + const now = this.t.host.now(); + if (this.armedAt === undefined || this.armedAt === null || this.armedAt > now) await this.arm(now); } else await this.schedule(); } async schedule(): Promise { - const existing = await this.t.host.getAlarm(); - if (existing === null) await this.t.host.setAlarm(this.t.host.now() + this.opts.intervalMs); + if (this.armedAt === undefined) this.armedAt = await this.t.host.getAlarm(); + if (this.armedAt === null) await this.arm(this.t.host.now() + this.opts.intervalMs); + } + + private async arm(time: number): Promise { + await this.t.host.setAlarm(time); + this.armedAt = time; } async onAlarm(): Promise { + this.armedAt = null; const res = await this.runOnce(); - if (res.remainingTxs > 0) await this.t.host.setAlarm(this.t.host.now() + 50); + if (res.remainingTxs > 0) await this.arm(this.t.host.now() + 50); } async runNow(): Promise { diff --git a/packages/ramose/src/internal/transactor/internal.ts b/packages/ramose/src/internal/transactor/internal.ts index 3091ec966..a4d4558db 100644 --- a/packages/ramose/src/internal/transactor/internal.ts +++ b/packages/ramose/src/internal/transactor/internal.ts @@ -29,3 +29,5 @@ export function internalGate(env: SecretEnv, request: Request): Response | undef headers: { "content-type": "application/json" }, }); } + +export const ROUTE_ROOT_HEADER = "x-ramose-route-root"; diff --git a/packages/ramose/src/internal/transactor/transactor-do.ts b/packages/ramose/src/internal/transactor/transactor-do.ts index 03e68ed9d..a7e153e02 100644 --- a/packages/ramose/src/internal/transactor/transactor-do.ts +++ b/packages/ramose/src/internal/transactor/transactor-do.ts @@ -14,12 +14,14 @@ import { serverSealingKey } from "../replication/identity-root.ts"; import { dbPrefix, prefixedBucket } from "../storage/index.ts"; import { type RamoseEnv, envInt } from "./env.ts"; import { DEFAULT_CONFIG, type SocketLike, type TransactorConfig, type TransactorHost } from "./host.ts"; -import { internalGate } from "./internal.ts"; +import { ROUTE_ROOT_HEADER, internalGate } from "./internal.ts"; import { Transactor, type TxAck } from "./transactor.ts"; import type { RuntimeBoundaries } from "../runtime-boundaries.ts"; export type { TxAck }; +const WRITER_SOCKET_TAG = "ramose-test-writer"; + export function configFromEnv(env: RamoseEnv): TransactorConfig { return { ...DEFAULT_CONFIG, @@ -30,7 +32,9 @@ export function configFromEnv(env: RamoseEnv): TransactorConfig { gcEveryNIndexes: envInt(env.RAMOSE_GC_EVERY_N_INDEXES, DEFAULT_CONFIG.gcEveryNIndexes), retainRoots: envInt(env.RAMOSE_RETAIN_ROOTS, DEFAULT_CONFIG.retainRoots), maxBatch: envInt(env.RAMOSE_MAX_BATCH, DEFAULT_CONFIG.maxBatch), + batchBudgetMs: envInt(env.RAMOSE_BATCH_BUDGET_MS, DEFAULT_CONFIG.batchBudgetMs), timingYields: env.RAMOSE_TIMING_YIELDS === "1", + socketWrites: false, }; } @@ -53,6 +57,8 @@ class TransactorDOBase extends DurableObject { private readonly core: Transactor; private readonly databaseCatalogBindings: DatabaseCatalogBindings | undefined; private dbName: string | undefined; + private readonly provisionedRoots = new Set(); + private readonly provisioning = new Map>(); constructor( ctx: DurableObjectState, @@ -62,7 +68,8 @@ class TransactorDOBase extends DurableObject { private readonly testing?: TransactorTesting, ) { super(ctx, env); - if (testing?.enabled(env) === true) testing.reset(); + const testingOn = testing?.enabled(env) === true; + if (testingOn) testing?.reset(); ctx.storage.sql.exec(`CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT NOT NULL)`); const row = ctx.storage.sql.exec(`SELECT v FROM meta WHERE k = 'db'`).toArray()[0]; if (row) this.dbName = JSON.parse(row.v as string) as string; @@ -78,12 +85,15 @@ class TransactorDOBase extends DurableObject { get bucket() { return prefixedBucket(env.STORE, dbPrefix(host.dbName)); }, - sockets: () => ctx.getWebSockets() as unknown as SocketLike[], + sockets: () => + (testingOn + ? ctx.getWebSockets().filter((ws) => !ctx.getTags(ws).includes(WRITER_SOCKET_TAG)) + : ctx.getWebSockets()) as unknown as SocketLike[], getAlarm: () => ctx.storage.getAlarm(), setAlarm: (time) => ctx.storage.setAlarm(time), abort: (reason) => ctx.abort(reason), now: () => Date.now(), - config: configFromEnv(env), + config: { ...configFromEnv(env), socketWrites: testingOn }, ...(env.ANALYTICS !== undefined && { analytics: env.ANALYTICS }), }; this.core = new Transactor( @@ -100,6 +110,45 @@ class TransactorDOBase extends DurableObject { ); } + private async provisionCatalog(rootDatabase: string): Promise { + if (this.databaseCatalogBindings === undefined) { + throw new Error("catalog provisioning unavailable"); + } + const derivation: DatabaseRouteDerivation = Object.freeze({ + rootDatabase: DatabaseId.make(rootDatabase), + }); + const route = await Effect.runPromise(deriveResolvedDatabaseRoute( + this.databaseCatalogBindings, + derivation, + )); + if (route.database !== DatabaseId.make(this.dbName!)) { + throw new Error("database route derivation does not match this transactor"); + } + const deployed = Result.getOrThrow(resolveBoundCatalogDefinition( + this.databaseCatalogBindings, + route, + )); + const t = await this.core.provisionCatalog(deployed.definition); + this.provisionedRoots.add(rootDatabase); + return t; + } + + private provisionCatalogOnce(rootDatabase: string): Promise { + if (this.provisionedRoots.has(rootDatabase)) return Promise.resolve(); + let pending = this.provisioning.get(rootDatabase); + if (pending === undefined) { + pending = this.provisionCatalog(rootDatabase).then( + () => undefined, + (cause) => { + this.provisioning.delete(rootDatabase); + throw cause; + }, + ); + this.provisioning.set(rootDatabase, pending); + } + return pending; + } + private assign(db: string): void { if (this.dbName === db) return; if (this.dbName !== undefined) throw new Error(`transactor already bound to database ${this.dbName}`); @@ -143,6 +192,10 @@ class TransactorDOBase extends DurableObject { const from = Number(url.searchParams.get("from") ?? "0"); const pair = new WebSocketPair(); const [client, server] = [pair[0], pair[1]]; + if (testingEnabled && url.searchParams.get("writer") === "1") { + this.ctx.acceptWebSocket(server, [WRITER_SOCKET_TAG]); + return new Response(null, { status: 101, webSocket: client }); + } this.ctx.acceptWebSocket(server); this.core.onSubscribe(server, from); return new Response(null, { status: 101, webSocket: client }); @@ -172,23 +225,9 @@ class TransactorDOBase extends DurableObject { ) { throw new Error("invalid database route derivation"); } - const derivation: DatabaseRouteDerivation = Object.freeze({ - rootDatabase: DatabaseId.make( - (raw as { readonly rootDatabase: string }).rootDatabase, - ), - }); - const route = await Effect.runPromise(deriveResolvedDatabaseRoute( - this.databaseCatalogBindings, - derivation, - )); - if (route.database !== DatabaseId.make(this.dbName)) { - throw new Error("database route derivation does not match this transactor"); - } - const deployed = Result.getOrThrow(resolveBoundCatalogDefinition( - this.databaseCatalogBindings, - route, - )); - const t = await this.core.provisionCatalog(deployed.definition); + const t = await this.provisionCatalog( + (raw as { readonly rootDatabase: string }).rootDatabase, + ); return new Response(JSON.stringify({ t }), { headers: { "content-type": "application/json" }, }); @@ -201,6 +240,22 @@ class TransactorDOBase extends DurableObject { }); } } + const routeRoot = request.headers.get(ROUTE_ROOT_HEADER); + if ( + routeRoot !== null && + (url.pathname === "/invoke" || url.pathname === "/invoke-batch") + ) { + try { + await this.provisionCatalogOnce(routeRoot); + } catch (cause) { + return new Response(JSON.stringify({ + error: cause instanceof Error ? cause.message : "catalog provisioning failed", + }), { + status: 500, + headers: { "content-type": "application/json" }, + }); + } + } if (testingEnabled) { const testAdmin = await this.testing.handleAdmin( request, diff --git a/packages/ramose/src/internal/transactor/transactor.ts b/packages/ramose/src/internal/transactor/transactor.ts index efaf04102..204c519bf 100644 --- a/packages/ramose/src/internal/transactor/transactor.ts +++ b/packages/ramose/src/internal/transactor/transactor.ts @@ -34,6 +34,7 @@ import { } from "../../db/Errors.ts"; import { R2NodeStore, readCurrentRoot, recordToRoots, rootsToRecord } from "../storage/index.ts"; import * as Effect from "effect/Effect"; +import * as Result from "effect/Result"; import { BadRequest, NotFound, TransactorDeadError, Unavailable, errorResponse, toHttpError } from "./errors.ts"; import { type SocketLike, type TransactorHost } from "./host.ts"; import { Indexer } from "./indexer.ts"; @@ -63,9 +64,10 @@ import { parseEntityIdScope, parseInvocationAllocations, parseStoredInvocationReceipt, - prepareInvocationReceipt, + prepareInvocationReceiptDirect, requireSuppliedOperationVersion, - resolveOperationCatalog, + resolveDeployedCatalogDefinition, + type ResolvedOperationCatalog, resolveSealedInputRefs, resolveSealedTarget, transitionInvocationReceipt, @@ -171,6 +173,17 @@ const safeName = (host: TransactorHost): string | undefined => { const json = (body: unknown, status = 200, headers: Record = {}) => new Response(JSON.stringify(toJson(body)), { status, headers: { "content-type": "application/json", ...headers } }); +export const MAX_INVOKE_BATCH = 256; + +export type InvokeOutcome = { + readonly status: number; + readonly body: unknown; + readonly headers?: Record; +}; + +const LOG_ROWS_PER_INSERT = 32; +const logRowPlaceholders = (rows: number): string => Array.from({ length: rows }, () => "(?, ?, ?)").join(", "); + export class Transactor { private ready: Promise | undefined; private conn!: Connection; @@ -244,7 +257,7 @@ export class Transactor { const fresh = rootsToRecord(roots, { log_watermark: 0, next_eid: FIRST_USER_EID, codec: gzipCodec.name }); const boot = bootstrapDatoms(); this.host.transactionSync(() => { - this.appendLogRow({ t: 1, txInstant: this.host.now(), datoms: boot }); + this.appendLogRows([{ t: 1, txInstant: this.host.now(), datoms: boot }]); this.setMeta("root", fresh); this.setMeta("next_eid", FIRST_USER_EID); }); @@ -285,10 +298,23 @@ export class Transactor { private setMeta(k: string, v: unknown): void { this.host.sql.exec(`INSERT OR REPLACE INTO meta (k, v) VALUES (?, ?)`, k, JSON.stringify(v)); } - private appendLogRow(e: LogEntry): void { - const body = encodeLogChunk([e]); - const buf = body.buffer.slice(body.byteOffset, body.byteOffset + body.byteLength) as ArrayBuffer; - this.host.sql.exec(`INSERT INTO log (t, tx_instant, datoms) VALUES (?, ?, ?)`, e.t, e.txInstant, buf); + private appendLogRows(entries: readonly LogEntry[]): void { + for (let from = 0; from < entries.length; from += LOG_ROWS_PER_INSERT) { + const slice = entries.slice(from, from + LOG_ROWS_PER_INSERT); + const bindings: unknown[] = []; + for (const e of slice) { + const body = encodeLogChunk([e]); + bindings.push( + e.t, + e.txInstant, + body.buffer.slice(body.byteOffset, body.byteOffset + body.byteLength) as ArrayBuffer, + ); + } + this.host.sql.exec( + `INSERT INTO log (t, tx_instant, datoms) VALUES ${logRowPlaceholders(slice.length)}`, + ...bindings, + ); + } } private readInvocationReceipt( @@ -332,22 +358,40 @@ export class Transactor { ); } + private completeClaimedReceipt(receipt: TerminalInvocationReceipt): void { + const updated = this.host.sql.exec( + `UPDATE operation_receipts SET status = ?, receipt = ? + WHERE principal_id = ? AND invocation_id = ? AND status = 'claimed' + RETURNING invocation_id`, + receipt.status, + JSON.stringify(receipt), + receipt.principalId, + receipt.invocationId, + ).toArray(); + if (updated.length !== 1) { + throw new Error("durable invocation claim changed before completion"); + } + } + private claimInvocationReceipt( prepared: PreparedInvocationReceipt, + inspected?: ReturnType, ) { - return this.host.transactionSync(() => { - const stored = this.readInvocationReceipt( - prepared.principalId, - prepared.invocationId, - ); - const decision = decideInvocationReceipt(stored, prepared); + const apply = (decision: ReturnType) => { if (decision._tag === "Claim") { this.insertInvocationReceipt(decision.receipt); } else if (decision._tag === "Recover") { this.replaceInvocationReceipt(decision.receipt); } return decision; - }); + }; + if (inspected !== undefined) return apply(inspected); + return this.host.transactionSync(() => + apply(decideInvocationReceipt( + this.readInvocationReceipt(prepared.principalId, prepared.invocationId), + prepared, + )) + ); } private inspectInvocationReceipt( @@ -376,36 +420,14 @@ export class Transactor { }); } - private assertClaimIdentity( - stored: StoredInvocationReceipt | LegacyInvocationReceiptRow | undefined, - claim: ClaimedInvocationReceipt, - ): asserts stored is ClaimedInvocationReceipt { - if ( - stored === undefined || isLegacyInvocationReceiptRow(stored) || - stored.status !== "claimed" || stored.version !== claim.version || - stored.principalId !== claim.principalId || - stored.invocationId !== claim.invocationId || - stored.scopeDigest !== claim.scopeDigest || - stored.operationVersion !== claim.operationVersion || - stored.invocationDigest !== claim.invocationDigest - ) { - throw new Error("durable invocation claim changed before completion"); - } - } - private finishInvocationReceipt( claim: ClaimedInvocationReceipt, event: InvocationReceiptEvent, insideTransaction = false, ): TerminalInvocationReceipt { const finish = () => { - const stored = this.readInvocationReceipt( - claim.principalId, - claim.invocationId, - ); - this.assertClaimIdentity(stored, claim); - const terminal = transitionInvocationReceipt(stored, event); - this.replaceInvocationReceipt(terminal); + const terminal = transitionInvocationReceipt(claim, event); + this.completeClaimedReceipt(terminal); return terminal; }; return insideTransaction ? finish() : this.host.transactionSync(finish); @@ -711,13 +733,21 @@ export class Transactor { private takeBatch(): Pending[] { const max = this.host.config.maxBatch; - if (this.queue[0]?.operation !== undefined) return this.queue.splice(0, 1); - const operationAt = this.queue.findIndex((pending) => pending.operation !== undefined); - const available = operationAt < 0 ? this.queue.length : operationAt; - const count = max > 0 ? Math.min(available, max) : available; + const count = max > 0 ? Math.min(this.queue.length, max) : this.queue.length; return this.queue.splice(0, count); } + private deferOverBudget(batch: Pending[], position: number, startedAt: number): void { + const budget = this.host.config.batchBudgetMs; + if (budget <= 0 || position === 0 || position >= batch.length) return; + if (performance.now() - startedAt < budget) return; + this.deferFrom(batch, position); + } + + private deferFrom(batch: Pending[], from: number): void { + this.queue.unshift(...batch.splice(from)); + } + private async commitLoop(): Promise { try { await this.init(); @@ -748,8 +778,12 @@ export class Transactor { }; }[] = []; const batchAcks = new Map(); + const claimedInBatch = new Set(); const tResolve = performance.now(); - for (const p of batch) { + for (let position = 0; position < batch.length; position++) { + this.deferOverBudget(batch, position, tResolve); + if (position >= batch.length) break; + const p = batch[position]; if (p.operation !== undefined) { let claim: ClaimedInvocationReceipt | undefined; try { @@ -759,9 +793,18 @@ export class Transactor { const supplied = requireSuppliedOperationVersion( p.operation.operationVersion, ); - const resolved = await Effect.runPromise( - resolveOperationCatalog(this.operationRuntime, p.operation), + const resolvedCatalog = resolveDeployedCatalogDefinition( + this.operationRuntime.catalogs, + { + database: p.operation.database, + catalogKey: p.operation.catalogKey, + unitHash: p.operation.unitHash, + }, ); + if (Result.isFailure(resolvedCatalog)) throw opaqueOperationDenial(); + const resolved: ResolvedOperationCatalog = Object.freeze({ + deployed: resolvedCatalog.success, + }); this.bindComposition( resolved.deployed.definition.unitHash, resolved.deployed.definition.composition, @@ -844,9 +887,15 @@ export class Transactor { await resolveCompatibility("UpdateRequired"); continue; } - const prepared = await Effect.runPromise( - prepareInvocationReceipt(operation, operationVersion), + const prepared = await prepareInvocationReceiptDirect( + operation, + operationVersion, ); + const receiptKey = `${prepared.principalId}\0${prepared.invocationId}`; + if (claimedInBatch.has(receiptKey)) { + this.deferFrom(batch, position); + break; + } const inspected = this.inspectInvocationReceipt(prepared); if ( inspected._tag === "OperationChanged" || @@ -916,7 +965,7 @@ export class Transactor { resolved, ); - const decision = this.claimInvocationReceipt(prepared); + const decision = this.claimInvocationReceipt(prepared, inspected); if ( decision._tag === "Conflict" || decision._tag === "OperationChanged" || @@ -944,6 +993,7 @@ export class Transactor { continue; } claim = decision.receipt; + claimedInBatch.add(receiptKey); await this.boundaries.checkpoint("operation.claimed"); const executed = await executeCatalogOperation( this.conn, @@ -1068,7 +1118,7 @@ export class Transactor { for (const pending of acks) pending.assertFresh?.(); this.host.transactionSync(() => { this.boundaries.checkpointSync("transactor.commit.write"); - for (const e of entries) this.appendLogRow(e); + this.appendLogRows(entries); this.setMeta("next_eid", this.conn.nextEntityId); for (const pending of acks) { if (pending.receiptCompletion !== undefined) { @@ -1187,7 +1237,21 @@ export class Transactor { return; } if (msg?.kind === "resume" && typeof msg.from === "number") this.sendCatchUp(ws, msg.from); - else if (msg?.kind === "ping") ws.send(JSON.stringify({ v: 1, kind: "pong", t: this.conn.t })); + else if (msg?.kind === "ping") { + ws.send(JSON.stringify({ v: 1, kind: "pong", t: this.conn.t, ...(msg.id === undefined ? {} : { id: msg.id }) })); + } else if (msg?.kind === "write" && this.host.config.socketWrites && Array.isArray(msg.tx)) { + const id = msg.id; + const clientTxId = typeof msg.clientTxId === "string" && msg.clientTxId.length > 0 ? msg.clientTxId : undefined; + void this.transact(msg.tx, undefined, clientTxId).then( + (ack) => ws.send(JSON.stringify({ v: 1, kind: "ack", id, t: ack.t })), + (cause) => ws.send(JSON.stringify({ + v: 1, + kind: "error", + id, + message: cause instanceof Error ? cause.message : String(cause), + })), + ); + } } private sendCatchUp(ws: SocketLike, from: number): void { @@ -1217,7 +1281,7 @@ export class Transactor { earliestLogT: this.earliestLogT(), nextEid: this.conn.nextEntityId, subscribers: this.host.sockets().length, - opts: { timingYields: this.host.config.timingYields, maxBatch: this.host.config.maxBatch }, + opts: { timingYields: this.host.config.timingYields, maxBatch: this.host.config.maxBatch, batchBudgetMs: this.host.config.batchBudgetMs }, stats: this.stats, metrics: { txPerSec: round(this.txRate.rate(this.host.now())), @@ -1246,6 +1310,60 @@ export class Transactor { return Number(row?.count ?? 0); } + private parseInvocation(raw: unknown): AuthoritativeOperationInvocation { + const invocation = typeof raw === "object" && raw !== null && !Array.isArray(raw) + ? { + ...raw, + ...(Object.hasOwn(raw, "target") + ? { target: fromJson((raw as { readonly target?: unknown }).target) } + : {}), + } as AuthoritativeOperationInvocation + : undefined; + if ( + invocation === undefined || invocation.database !== safeName(this.host) + ) { + throw new BadRequest({ message: "invalid deployed operation invocation" }); + } + const entityIdScope = parseEntityIdScope(invocation.entityIdScope); + const allocations = parseInvocationAllocations(invocation.allocations); + if ( + allocations === undefined || + (invocation.sealedTarget !== undefined && + (typeof invocation.sealedTarget !== "string" || + entityIdScope === undefined)) || + (allocations.length > 0 && entityIdScope === undefined) + ) { + throw new BadRequest({ message: "invalid deployed operation invocation" }); + } + if ( + entityIdScope !== undefined && + (typeof invocation.entityIdKeyId !== "string" || + invocation.entityIdKeyId.length === 0) + ) { + throw new BadRequest({ message: "invalid deployed operation invocation" }); + } + return { + ...invocation, + ...(entityIdScope === undefined ? {} : { entityIdScope }), + ...(allocations.length === 0 ? {} : { allocations }), + }; + } + + private async invokeOutcome(raw: unknown): Promise { + try { + return { status: 200, body: await this.invoke(this.parseInvocation(raw)) }; + } catch (cause) { + const error = toHttpError(cause); + const response = errorResponse(error); + const retryAfter = response.headers.get("retry-after"); + return { + status: response.status, + body: await response.json(), + ...(retryAfter === null ? {} : { headers: { "retry-after": retryAfter } }), + }; + } + } + async handleRequest(request: Request): Promise { await this.init(); this.metrics.observeColo((request as { cf?: { colo?: string } }).cf?.colo ?? request.headers.get("x-ramose-colo") ?? undefined); @@ -1270,47 +1388,20 @@ export class Transactor { const path = url.pathname; if (path === "/invoke" && request.method === "POST") { const body = await request.json() as { invocation?: unknown }; - const raw = body?.invocation; - const invocation = typeof raw === "object" && raw !== null && !Array.isArray(raw) - ? { - ...raw, - ...(Object.hasOwn(raw, "target") - ? { target: fromJson((raw as { readonly target?: unknown }).target) } - : {}), - } as AuthoritativeOperationInvocation - : undefined; - if ( - invocation === undefined || invocation.database !== safeName(this.host) - ) { - throw new BadRequest({ message: "invalid deployed operation invocation" }); - } - const entityIdScope = parseEntityIdScope(invocation.entityIdScope); - const allocations = parseInvocationAllocations(invocation.allocations); - if ( - allocations === undefined || - (invocation.sealedTarget !== undefined && - (typeof invocation.sealedTarget !== "string" || - entityIdScope === undefined)) || - (allocations.length > 0 && entityIdScope === undefined) - ) { - throw new BadRequest({ message: "invalid deployed operation invocation" }); - } - if ( - entityIdScope !== undefined && - (typeof invocation.entityIdKeyId !== "string" || - invocation.entityIdKeyId.length === 0) - ) { - throw new BadRequest({ message: "invalid deployed operation invocation" }); - } - const resolved: AuthoritativeOperationInvocation = { - ...invocation, - ...(entityIdScope === undefined ? {} : { entityIdScope }), - ...(allocations.length === 0 ? {} : { allocations }), - }; + const resolved = this.parseInvocation(body?.invocation); return new Response(JSON.stringify(await this.invoke(resolved)), { headers: { "content-type": "application/json" }, }); } + if (path === "/invoke-batch" && request.method === "POST") { + const body = await request.json() as { invocations?: unknown }; + const raw = Array.isArray(body?.invocations) ? body.invocations : undefined; + if (raw === undefined || raw.length === 0 || raw.length > MAX_INVOKE_BATCH) { + throw new BadRequest({ message: "invalid deployed operation invocation batch" }); + } + const results = await Promise.all(raw.map((entry) => this.invokeOutcome(entry))); + return json({ results }); + } if (path === "/transact" && request.method === "POST") { const body = fromJson(await request.json()) as { tx?: TxData; diff --git a/packages/ramose/src/worker/authorized-operation.ts b/packages/ramose/src/worker/authorized-operation.ts index b75b9b0a8..b8b505c8c 100644 --- a/packages/ramose/src/worker/authorized-operation.ts +++ b/packages/ramose/src/worker/authorized-operation.ts @@ -17,7 +17,9 @@ import { fromJson, toJson } from "../internal/core/json.ts"; import type { EntityIdScope } from "../internal/replication/entity-id.ts"; import type { ServerSealingKey } from "../internal/replication/server-identity.ts"; import { makeEntityIdScope } from "../internal/replication/identity.ts"; -import { internalHeaders } from "../internal/transactor/index.ts"; +import { ROUTE_ROOT_HEADER, internalHeaders } from "../internal/transactor/index.ts"; +import { type InvokeOutcome } from "../internal/transactor/transactor.ts"; +import { DEFAULT_COALESCE_WINDOW_MS, coalesceInvocation } from "./operation-coalescer.ts"; import type { RamoseEnv } from "../RamoseEnv.ts"; import { serverSealingKey } from "./server-identity.ts"; import { @@ -47,17 +49,18 @@ type RoutedOperationRequest = ParsedOperationRequest & { readonly unitHash: OperationInvocation["unitHash"]; }; +const wireInvocation = (invocation: AuthoritativeOperationInvocation) => ({ + ...invocation, + ...(invocation.target === undefined ? {} : { target: toJson(invocation.target) }), +}); + export const serializeOperationInvocation = ( invocation: AuthoritativeOperationInvocation, -): string => { - const wireInvocation = { - ...invocation, - ...(invocation.target === undefined - ? {} - : { target: toJson(invocation.target) }), - }; - return JSON.stringify({ invocation: wireInvocation }); -}; +): string => JSON.stringify({ invocation: wireInvocation(invocation) }); + +export const serializeOperationInvocationBatch = ( + invocations: readonly AuthoritativeOperationInvocation[], +): string => JSON.stringify({ invocations: invocations.map(wireInvocation) }); const bad = (message: string): BadRequest => new BadRequest({ message }); const deny = (): Unauthorized => new Unauthorized({ status: 403 }); @@ -287,12 +290,116 @@ const sealPublicEntityRefs = async ( } }; +export type InvokeRouting = { + readonly rootDatabase?: string; +}; + +const coalesceWindowMs = (env: RamoseEnv): number => { + const raw = env.RAMOSE_OP_COALESCE_MS; + if (raw === undefined || raw === "") return DEFAULT_COALESCE_WINDOW_MS; + const parsed = Number(raw); + return Number.isFinite(parsed) && parsed >= 0 ? parsed : DEFAULT_COALESCE_WINDOW_MS; +}; + +const transactorHeaders = ( + env: RamoseEnv, + routing: InvokeRouting, +): Record => ({ + "content-type": "application/json", + ...internalHeaders(env), + ...(routing.rootDatabase === undefined ? {} : { [ROUTE_ROOT_HEADER]: routing.rootDatabase }), +}); + +const invokeSingle = async ( + env: RamoseEnv, + database: string, + routing: InvokeRouting, + invocation: AuthoritativeOperationInvocation, +): Promise => { + const stub = env.TRANSACTOR.get(env.TRANSACTOR.idFromName(database)); + const response = await stub.fetch( + `https://transactor/invoke?db=${encodeURIComponent(database)}`, + { + method: "POST", + headers: transactorHeaders(env, routing), + body: serializeOperationInvocation(invocation), + }, + ); + const text = await response.text(); + const retryAfter = response.headers.get("retry-after"); + let body: unknown; + try { + body = JSON.parse(text); + } catch { + body = undefined; + } + return { + status: response.status, + body, + ...(retryAfter === null ? {} : { headers: { "retry-after": retryAfter } }), + }; +}; + +const invokeBatch = async ( + env: RamoseEnv, + database: string, + routing: InvokeRouting, + invocations: readonly AuthoritativeOperationInvocation[], +): Promise => { + const stub = env.TRANSACTOR.get(env.TRANSACTOR.idFromName(database)); + const response = await stub.fetch( + `https://transactor/invoke-batch?db=${encodeURIComponent(database)}`, + { + method: "POST", + headers: transactorHeaders(env, routing), + body: serializeOperationInvocationBatch(invocations), + }, + ); + const text = await response.text(); + if (!response.ok) { + const retryAfter = response.headers.get("retry-after"); + let body: unknown; + try { + body = JSON.parse(text); + } catch { + body = undefined; + } + const shared: InvokeOutcome = { + status: response.status, + body, + ...(retryAfter === null ? {} : { headers: { "retry-after": retryAfter } }), + }; + return invocations.map(() => shared); + } + const parsed = JSON.parse(text) as { readonly results?: unknown }; + if (!Array.isArray(parsed?.results)) { + throw new Error("transactor returned an invalid invocation batch"); + } + return parsed.results as readonly InvokeOutcome[]; +}; + +const outcomeFailure = (outcome: InvokeOutcome): UpstreamError | OperationRejected => { + const text = JSON.stringify(outcome.body ?? null); + if (outcome.status === 409) { + return operationRejectedOf(text) ?? privateFailure(409); + } + if (outcome.status === 503) { + const retryAfter = outcome.headers?.["retry-after"]; + return privateFailure( + 503, + retryAfter === undefined ? undefined : { "retry-after": retryAfter }, + ); + } + return privateFailure(outcome.status >= 500 ? 500 : outcome.status); +}; + export const invokeAuthoritativeOperation = async ( env: RamoseEnv, database: string, origin: string, parsed: RoutedOperationRequest, caller: AuthenticatedCaller, + routing: InvokeRouting = {}, ): Promise => { const sealingScope = await invocationEntityIdScope( env, @@ -310,32 +417,27 @@ export const invokeAuthoritativeOperation = async ( entityIdKeyId: sealingScope.sealing.keyId, }), }; - const stub = env.TRANSACTOR.get(env.TRANSACTOR.idFromName(database)); - let response: Response; - let text: string; + const windowMs = coalesceWindowMs(env); + let outcome: InvokeOutcome; try { - response = await stub.fetch( - `https://transactor/invoke?db=${encodeURIComponent(database)}`, - { - method: "POST", - headers: { - "content-type": "application/json", - ...internalHeaders(env), - }, - body: serializeOperationInvocation(invocation), - }, - ); - text = await response.text(); + outcome = windowMs > 0 + ? await coalesceInvocation( + `${database}\0${routing.rootDatabase ?? ""}`, + invocation, + (batch) => invokeBatch(env, database, routing, batch), + windowMs, + ) + : await invokeSingle(env, database, routing, invocation); } catch { throw privateFailure(); } - if (!response.ok) { - throw operationFailureFromResponse(response, text); + if (outcome.status !== 200) { + throw outcomeFailure(outcome); } let result: AuthoritativeInvocationResult; try { result = parseAuthoritativeInvocationResult( - JSON.parse(text), + outcome.body, invocation.invocationId, ); } catch { diff --git a/packages/ramose/src/worker/handle.ts b/packages/ramose/src/worker/handle.ts index 6b812c0fb..43ab2f592 100644 --- a/packages/ramose/src/worker/handle.ts +++ b/packages/ramose/src/worker/handle.ts @@ -386,6 +386,7 @@ export const handle = ( database: string, catalogKey: CatalogId, unitHash: CatalogUnitHash, + rootDatabase?: string, ) => Effect.tryPromise({ try: () => invokeAuthoritativeOperation( env, @@ -410,6 +411,7 @@ export const handle = ( input: parsed.input, }, caller, + rootDatabase === undefined ? {} : { rootDatabase }, ), catch: (cause) => isRamoseError(cause) ? cause : fromThrown(cause, { stacks: env.RAMOSE_STAGE !== "prod", @@ -422,9 +424,7 @@ export const handle = ( const root = yield* Effect.fromResult( databaseBindings.root(DatabaseId.make(db)), ).pipe(Effect.mapError(() => new Unauthorized({ status: 403 }))); - yield* provisionResolvedDatabase(env, root, { - rootDatabase: root.database, - }); + return yield* invoke(db, rootProof.catalogKey, rootProof.unitHash, root.database); } return yield* invoke(db, rootProof.catalogKey, rootProof.unitHash); }); diff --git a/packages/ramose/src/worker/operation-coalescer.ts b/packages/ramose/src/worker/operation-coalescer.ts new file mode 100644 index 000000000..baed162d0 --- /dev/null +++ b/packages/ramose/src/worker/operation-coalescer.ts @@ -0,0 +1,87 @@ +import type { AuthoritativeOperationInvocation } from "../internal/authorization/index.ts"; +import { MAX_INVOKE_BATCH, type InvokeOutcome } from "../internal/transactor/transactor.ts"; + +export type InvokeDispatch = ( + invocations: readonly AuthoritativeOperationInvocation[], +) => Promise; + +type Waiter = { + readonly invocation: AuthoritativeOperationInvocation; + settled: boolean; + outcome?: InvokeOutcome; + failure?: unknown; +}; + +type Lane = { + readonly waiters: Waiter[]; + dispatched: boolean; +}; + +export const DEFAULT_COALESCE_WINDOW_MS = 2; +const MAX_WAIT_MS = 30_000; + +const lanes = new Map(); + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +const settleAll = (waiters: readonly Waiter[], apply: (waiter: Waiter, index: number) => void): void => { + waiters.forEach((waiter, index) => { + apply(waiter, index); + waiter.settled = true; + }); +}; + +const flushLane = async (key: string, lane: Lane, dispatch: InvokeDispatch): Promise => { + lane.dispatched = true; + if (lanes.get(key) === lane) lanes.delete(key); + const waiters = lane.waiters; + try { + const outcomes = await dispatch(waiters.map((w) => w.invocation)); + if (outcomes.length !== waiters.length) { + throw new Error("transactor returned a mismatched invocation batch"); + } + settleAll(waiters, (waiter, index) => { + waiter.outcome = outcomes[index]!; + }); + } catch (cause) { + settleAll(waiters, (waiter) => { + waiter.failure = cause; + }); + } +}; + +const settled = (waiter: Waiter): InvokeOutcome => { + if (waiter.failure !== undefined) throw waiter.failure; + return waiter.outcome!; +}; + +export const coalesceInvocation = async ( + key: string, + invocation: AuthoritativeOperationInvocation, + dispatch: InvokeDispatch, + windowMs = DEFAULT_COALESCE_WINDOW_MS, +): Promise => { + let lane = lanes.get(key); + if (lane === undefined || lane.dispatched) { + lane = { waiters: [], dispatched: false }; + lanes.set(key, lane); + } + const waiter: Waiter = { invocation, settled: false }; + lane.waiters.push(waiter); + if (lane.waiters.length >= MAX_INVOKE_BATCH) { + await flushLane(key, lane, dispatch); + return settled(waiter); + } + const deadline = Date.now() + MAX_WAIT_MS; + while (true) { + await sleep(Math.max(1, windowMs)); + if (waiter.settled) return settled(waiter); + if (!lane.dispatched) { + await flushLane(key, lane, dispatch); + return settled(waiter); + } + if (Date.now() > deadline) { + throw new Error("timed out waiting for a coalesced operation batch"); + } + } +}; diff --git a/packages/ramose/src/worker/testing.ts b/packages/ramose/src/worker/testing.ts index 21970316e..8c063b6c9 100644 --- a/packages/ramose/src/worker/testing.ts +++ b/packages/ramose/src/worker/testing.ts @@ -93,7 +93,20 @@ export const createServer = (options: ServerOptions = {}) => ({ } return await handleTestAdmin(request, runtimeEnv, new URL(request.url)); } catch (cause) { - return respond(asTestAdminError(cause), request, runtimeEnv); + const error = asTestAdminError(cause); + if (error._tag === "UpstreamError" && error.status >= 500) { + return new Response( + JSON.stringify({ error: "internal error", upstream: { status: error.status, body: error.body.slice(0, 400) } }), + { status: 500, headers: { "content-type": "application/json" } }, + ); + } + if (error._tag === "Internal") { + return new Response( + JSON.stringify({ error: "internal error", message: error.message.slice(0, 400) }), + { status: 500, headers: { "content-type": "application/json" } }, + ); + } + return respond(error, request, runtimeEnv); } } return runFetch( diff --git a/packages/ramose/test/internal/authorization/operations-runtime.test.ts b/packages/ramose/test/internal/authorization/operations-runtime.test.ts index 06b50485c..b120d8148 100644 --- a/packages/ramose/test/internal/authorization/operations-runtime.test.ts +++ b/packages/ramose/test/internal/authorization/operations-runtime.test.ts @@ -706,6 +706,28 @@ const ReplayTarget = Entity("replayTarget", { return { id: op.self }; }, }), + noteNoise: Operation({ + input: EffectSchema.Struct({ + noise: OperationEntityId, + note: EffectSchema.String, + }), + output: EffectSchema.Struct({ id: OperationEntityId }), + run(op, input) { + (op as any).set(ReplayNoise, input.noise, ReplayNoise.note, input.note); + return { id: op.self }; + }, + }), + renameGate: Operation({ + input: EffectSchema.Struct({ + gate: OperationEntityId, + name: EffectSchema.String, + }), + output: EffectSchema.Struct({ id: OperationEntityId }), + run(op, input) { + (op as any).set(ReplayGate, input.gate, ReplayGate.name, input.name); + return { id: op.self }; + }, + }), }), }); const ReplayApp = Schema("replay-fence", { @@ -727,6 +749,8 @@ ReplayApp.applyPolicy({ roles: ["member"] }, ({ policy, session, allOf }) => { policy.replayTarget.operations.deleteAndConsumeGate.where(session.hasRole("member")); policy.replayTarget.operations.archive.where(session.hasRole("member")); policy.replayTarget.operations.moveLookup.where(session.hasRole("member")); + policy.replayTarget.operations.noteNoise.where(session.hasRole("member")); + policy.replayTarget.operations.renameGate.where(session.hasRole("member")); }); const replayDatabase = DatabaseId.make("operation-replay-fence"); const replayArtifactHash = DigestHex.make("6".repeat(64)); @@ -810,7 +834,13 @@ const replayRuntime = (world: Awaited>) const replayInvocation = ( world: Awaited>, - localName: "deleteAndEcho" | "deleteAndConsumeGate" | "archive" | "moveLookup", + localName: + | "deleteAndEcho" + | "deleteAndConsumeGate" + | "archive" + | "moveLookup" + | "noteNoise" + | "renameGate", input: unknown, ) => ({ owner: { kind: "entity" as const, name: "replayTarget" }, @@ -1457,6 +1487,53 @@ describe("deployed operation runtime", () => { )).rejects.toBeInstanceOf(Unauthorized); }); + test("keeps a visible fence without rechecking when the write touches nothing the witness observed", async () => { + const world = await buildReplayFenceWorld(); + const runtime = replayRuntime(world); + const invocation = replayInvocation(world, "noteNoise", { + noise: world.noise, + note: "touched", + }); + const executed = await executeCatalogOperation(world.conn, runtime, invocation); + expect(executed.output).toEqual({ id: world.target }); + expect(executed.replayFence.target?.postCommit).toEqual({ kind: "visible" }); + expect((await world.conn.db().entity(world.noise))?.[":replayNoise/note"]).toBe("touched"); + await expect(authorizeCatalogOperationReplay( + world.conn, + runtime, + invocation, + executed.replayFence, + )).resolves.toBeUndefined(); + }); + + test("recomputes the fence when the write changes a policy fact the witness observed", async () => { + const world = await buildReplayFenceWorld(); + const runtime = replayRuntime(world); + const invocation = replayInvocation(world, "renameGate", { + gate: world.gate, + name: "Revoked", + }); + const executed = await executeCatalogOperation(world.conn, runtime, invocation); + expect(executed.output).toEqual({ id: world.target }); + const fenced = executed.replayFence.target; + expect(fenced?.postCommit.kind).toBe("hidden"); + if (fenced?.postCommit.kind !== "hidden") throw new Error("expected hidden replay fence"); + expect(/^[0-9a-f]{64}$/.test(fenced.postCommit.authorizationDigest)).toBe(true); + await expect(authorizeCatalogOperationReplay( + world.conn, + runtime, + invocation, + executed.replayFence, + )).resolves.toBeUndefined(); + await world.conn.transact([{ ":db/id": world.gate, ":replayGate/name": "Good" }]); + await expect(authorizeCatalogOperationReplay( + world.conn, + runtime, + invocation, + executed.replayFence, + )).rejects.toBeInstanceOf(Unauthorized); + }); + test("replays an exact post-commit lookup rebind without admitting its replacement", async () => { const world = await buildReplayFenceWorld(); const runtime = replayRuntime(world); diff --git a/packages/ramose/test/internal/core/novelty-runs.test.ts b/packages/ramose/test/internal/core/novelty-runs.test.ts new file mode 100644 index 000000000..aae6ab09f --- /dev/null +++ b/packages/ramose/test/internal/core/novelty-runs.test.ts @@ -0,0 +1,122 @@ +import { describe, expect, test } from "bun:test"; +import { COMPARATORS, type Datom, Index, ValueTag, comparePrefix, datomEquals } from "../../../src/internal/core/datom.ts"; +import { Novelty, SortedNovelty } from "../../../src/internal/core/novelty.ts"; +import { Schema } from "../../../src/internal/core/schema.ts"; +import { Connection } from "../../../src/internal/core/conn.ts"; +import { randDatoms, randInt, rng } from "./util.ts"; + +function sortDedup(index: 0 | 1 | 2 | 3, ds: readonly Datom[]): Datom[] { + const cmp = COMPARATORS[index]; + const s = ds.slice().sort(cmp); + return s.filter((d, i) => i === 0 || cmp(s[i - 1], d) !== 0); +} + +function chunkToArray(c: { datoms: readonly Datom[]; start: number; end: number } | undefined): Datom[] { + return c === undefined ? [] : c.datoms.slice(c.start, c.end); +} + +describe("SortedNovelty sorted runs", () => { + test("many small adds keep range and all consistent with a single sorted array", () => { + const r = rng(7); + for (const index of [Index.EAVT, Index.AEVT, Index.AVET, Index.VAET] as const) { + const n = new SortedNovelty(index); + let all: Datom[] = []; + for (let i = 0; i < 300; i++) { + const batch = randDatoms(r, randInt(r, 1, 6), { maxE: 40, maxA: 4, maxT: 50 }); + n.add(batch); + all = all.concat(batch); + if (i % 7 === 0) { + const d = all[randInt(r, 0, all.length - 1)]; + const p = index === Index.VAET ? { vt: d.vt, v: d.v } : index === Index.EAVT ? { e: d.e } : { a: d.a }; + const got = chunkToArray(n.range(p)); + const exp = sortDedup(index, all).filter((x) => comparePrefix(index, x, p) === 0); + expect(got.length).toBe(exp.length); + got.forEach((x, j) => expect(datomEquals(x, exp[j])).toBe(true)); + } + } + const got = n.all(); + const exp = sortDedup(index, all); + expect(got.length).toBe(exp.length); + got.forEach((x, j) => expect(datomEquals(x, exp[j])).toBe(true)); + expect(n.size).toBe(exp.length); + n.dropThrough(25); + expect(n.all().every((d) => d.t > 25)).toBe(true); + expect(n.size).toBe(exp.filter((d) => d.t > 25).length); + } + }); + + test("a duplicate added in a later run is reported once", () => { + const r = rng(11); + const n = new SortedNovelty(Index.EAVT); + const first = randDatoms(r, 5, { maxE: 3, maxA: 2, maxT: 2 }); + n.add(first); + n.all(); + n.add([first[2]]); + n.range({ e: first[2].e }); + n.add(first.slice(0, 2)); + const got = n.all(); + const exp = sortDedup(Index.EAVT, first); + expect(got.length).toBe(exp.length); + got.forEach((x, j) => expect(datomEquals(x, exp[j])).toBe(true)); + }); +}); + +describe("Novelty overlay", () => { + test("layered view reads like a copy of base plus the overlay datoms", () => { + const r = rng(13); + const schema = Schema.bootstrap(); + const isAvet = (a: number) => schema.isAvet(a) || a % 2 === 0; + const isVaet = (a: number) => schema.isVaet(a) || a % 3 === 0; + const base = new Novelty(); + const baseDatoms = randDatoms(r, 500, { maxE: 30, maxA: 6, maxT: 20 }); + base.add(baseDatoms, isAvet, isVaet); + const extra = randDatoms(r, 20, { maxE: 30, maxA: 6, maxT: 25 }); + const view = base.overlay(extra, isAvet, isVaet); + const copy = new Novelty(); + copy.add(baseDatoms, isAvet, isVaet); + copy.add(extra, isAvet, isVaet); + expect(view.maxT).toBe(copy.maxT); + for (const index of [Index.EAVT, Index.AEVT, Index.AVET, Index.VAET] as const) { + const expAll = copy.byIndex[index].all(); + const gotAll = view.byIndex[index].all(); + expect(gotAll.length).toBe(expAll.length); + gotAll.forEach((x, j) => expect(datomEquals(x, expAll[j])).toBe(true)); + for (let k = 0; k < 40; k++) { + const d = expAll.length === 0 ? undefined : expAll[randInt(r, 0, expAll.length - 1)]; + if (d === undefined) continue; + const p = index === Index.VAET ? { vt: d.vt, v: d.v } : index === Index.EAVT ? { e: d.e } : k % 2 ? { a: d.a } : {}; + const got = chunkToArray(view.byIndex[index].range(p)); + const exp = chunkToArray(copy.byIndex[index].range(p)); + expect(got.length).toBe(exp.length); + got.forEach((x, j) => expect(datomEquals(x, exp[j])).toBe(true)); + } + } + expect(base.byIndex[Index.EAVT].all().length).toBe(sortDedup(Index.EAVT, baseDatoms).length); + }); + + test("validators see the staged transaction without copying unindexed novelty", async () => { + const conn = await Connection.create(); + await conn.transact([ + { ":db/ident": ":item/name", ":db/valueType": ":db.type/string", ":db/cardinality": ":db.cardinality/one", ":db/unique": ":db.unique/identity" }, + { ":db/ident": ":item/parent", ":db/valueType": ":db.type/ref", ":db/cardinality": ":db.cardinality/one", ":db/optional": true }, + ]); + for (let i = 0; i < 200; i++) await conn.transact([{ ":item/name": `seed-${i}` }]); + const seeded = conn.noveltyCount; + const staged = await conn.transactValidated( + [{ ":db/id": "child", ":item/name": "child", ":item/parent": [":item/name", "seed-7"] }], + async ({ dbBefore, dbAfter, tempids }) => { + const child = tempids.child; + expect(await dbBefore.entid([":item/name", "child"])).toBeUndefined(); + expect(await dbAfter.entid([":item/name", "child"])).toBe(child); + expect(await dbAfter.entid([":item/name", "seed-7"])).toBe(await dbBefore.entid([":item/name", "seed-7"])); + const parent = await dbAfter.entid([":item/name", "seed-7"]); + const incoming = await dbAfter.datomsArray(Index.VAET, { vt: ValueTag.Ref, v: parent! }); + expect(incoming.some((d) => d.e === child)).toBe(true); + expect(dbAfter.novelty.count).toBe(seeded + 4); + return child; + }, + ); + expect(conn.noveltyCount).toBe(seeded + 4); + expect(await conn.db().entid([":item/name", "child"])).toBe(staged.value); + }); +}); diff --git a/packages/ramose/test/worker/operation-coalescer.test.ts b/packages/ramose/test/worker/operation-coalescer.test.ts new file mode 100644 index 000000000..d549a8fc9 --- /dev/null +++ b/packages/ramose/test/worker/operation-coalescer.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, test } from "bun:test"; +import type { AuthoritativeOperationInvocation } from "../../src/internal/authorization/index.ts"; +import { MAX_INVOKE_BATCH, type InvokeOutcome } from "../../src/internal/transactor/transactor.ts"; +import { coalesceInvocation } from "../../src/worker/operation-coalescer.ts"; + +const invocation = (id: number): AuthoritativeOperationInvocation => + ({ invocationId: `inv-${id}`, input: { id } }) as unknown as AuthoritativeOperationInvocation; + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +const echo = async (batch: readonly AuthoritativeOperationInvocation[]): Promise => + batch.map((entry) => ({ status: 200, body: { invocationId: entry.invocationId } })); + +describe("operation coalescer", () => { + test("groups invocations that arrive within the window into one dispatch", async () => { + const batches: number[] = []; + const dispatch = (batch: readonly AuthoritativeOperationInvocation[]) => { + batches.push(batch.length); + return echo(batch); + }; + const outcomes = await Promise.all( + Array.from({ length: 5 }, (_, i) => coalesceInvocation("db-a", invocation(i), dispatch, 5)), + ); + expect(batches).toEqual([5]); + expect(outcomes.map((o) => (o.body as { invocationId: string }).invocationId)).toEqual( + ["inv-0", "inv-1", "inv-2", "inv-3", "inv-4"], + ); + }); + + test("keeps separate keys in separate dispatches", async () => { + const keys: string[] = []; + const dispatch = (key: string) => (batch: readonly AuthoritativeOperationInvocation[]) => { + keys.push(key); + return echo(batch); + }; + await Promise.all([ + coalesceInvocation("db-b", invocation(1), dispatch("db-b"), 2), + coalesceInvocation("db-c", invocation(2), dispatch("db-c"), 2), + ]); + expect(keys.sort()).toEqual(["db-b", "db-c"]); + }); + + test("flushes early once a batch reaches the transactor limit", async () => { + const batches: number[] = []; + const dispatch = (batch: readonly AuthoritativeOperationInvocation[]) => { + batches.push(batch.length); + return echo(batch); + }; + await Promise.all( + Array.from({ length: MAX_INVOKE_BATCH + 3 }, (_, i) => + coalesceInvocation("db-d", invocation(i), dispatch, 50)), + ); + expect(batches).toEqual([MAX_INVOKE_BATCH, 3]); + }); + + test("rejects every waiter when the dispatch fails or is mismatched", async () => { + const failing = () => Promise.reject(new Error("down")); + await expect(Promise.all([ + coalesceInvocation("db-e", invocation(1), failing, 1), + coalesceInvocation("db-e", invocation(2), failing, 1), + ])).rejects.toThrow("down"); + await Promise.allSettled([sleep(5)]); + const short = async () => [{ status: 200, body: {} }]; + await expect(Promise.all([ + coalesceInvocation("db-f", invocation(1), short, 1), + coalesceInvocation("db-f", invocation(2), short, 1), + ])).rejects.toThrow("mismatched"); + }); +}); diff --git a/scripts/e2e-cloudflare.sh b/scripts/e2e-cloudflare.sh index 84a9e18f6..df3138eea 100755 --- a/scripts/e2e-cloudflare.sh +++ b/scripts/e2e-cloudflare.sh @@ -6,8 +6,9 @@ cd "$(dirname "$0")/.." fail() { echo "error: $*" >&2; exit 1; } -[ -n "${CLOUDFLARE_API_TOKEN:-}" ] || fail "CLOUDFLARE_API_TOKEN is not set (see .cursor/CLOUD.md)." -[ -n "${CLOUDFLARE_ACCOUNT_ID:-}" ] || fail "CLOUDFLARE_ACCOUNT_ID is not set (see .cursor/CLOUD.md)." +if [ -z "${CLOUDFLARE_API_TOKEN:-}" ] && [ ! -f "${HOME}/.alchemy/profiles.json" ]; then + fail "set CLOUDFLARE_API_TOKEN and CLOUDFLARE_ACCOUNT_ID (see .cursor/CLOUD.md), or log in with 'bun alchemy login'." +fi command -v bun >/dev/null 2>&1 || fail "bun is not on PATH." echo ">> Building the package ..."