From 1595bf40180828d2f93983d52b7bfd0a8d0e31bc Mon Sep 17 00:00:00 2001 From: forhappy Date: Mon, 19 Oct 2026 10:47:37 -0700 Subject: [PATCH] feat: add Cloudflare benchmark deployment --- .github/workflows/rust.yml | 4 +- deploy/cloudflare/Dockerfile | 6 + deploy/cloudflare/README.md | 178 + deploy/cloudflare/agent.py | 206 ++ deploy/cloudflare/provision.py | 111 + deploy/cloudflare/worker.js | 141 + .../cloudflare-perf-throughput-latency.svg | 3108 +++++++++++++++++ docs/README.md | 2 +- docs/cloudflare-benchmark.md | 180 + docs/data/cloudflare-benchmark.csv | 16 + docs/performance.md | 1 + scripts/check-docs.py | 2 +- scripts/perf/bench.py | 3 + scripts/perf/cloudflare.py | 336 ++ scripts/perf/cloudflare_report.py | 112 + scripts/perf/plot.py | 11 +- scripts/perf/requirements-cloudflare.txt | 2 + scripts/perf/test_cloudflare.py | 102 + 18 files changed, 4514 insertions(+), 7 deletions(-) create mode 100644 deploy/cloudflare/Dockerfile create mode 100644 deploy/cloudflare/README.md create mode 100644 deploy/cloudflare/agent.py create mode 100644 deploy/cloudflare/provision.py create mode 100644 deploy/cloudflare/worker.js create mode 100644 diagram/cloudflare-perf-throughput-latency.svg create mode 100644 docs/cloudflare-benchmark.md create mode 100644 docs/data/cloudflare-benchmark.csv create mode 100644 scripts/perf/cloudflare.py create mode 100644 scripts/perf/cloudflare_report.py create mode 100644 scripts/perf/requirements-cloudflare.txt create mode 100644 scripts/perf/test_cloudflare.py diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index cb07a68..6ac4563 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -26,8 +26,10 @@ jobs: - name: Check configurable benchmark tooling run: | python3 -m venv "$RUNNER_TEMP/perf-venv" - "$RUNNER_TEMP/perf-venv/bin/pip" install -r scripts/perf/requirements.txt + "$RUNNER_TEMP/perf-venv/bin/pip" install -r scripts/perf/requirements-cloudflare.txt "$RUNNER_TEMP/perf-venv/bin/python" -m unittest discover -s scripts/perf -p 'test_*.py' + "$RUNNER_TEMP/perf-venv/bin/python" -m py_compile deploy/cloudflare/agent.py deploy/cloudflare/provision.py scripts/perf/cloudflare.py scripts/perf/cloudflare_report.py + node --check deploy/cloudflare/worker.js build: runs-on: ubuntu-24.04 steps: diff --git a/deploy/cloudflare/Dockerfile b/deploy/cloudflare/Dockerfile new file mode 100644 index 0000000..52a92ee --- /dev/null +++ b/deploy/cloudflare/Dockerfile @@ -0,0 +1,6 @@ +FROM python:3.12-slim-bookworm@sha256:34386ef0cb081344d7ec1c103ba398e6e9f64e9ab3a1509accc92a4e24a07258 +RUN pip install --no-cache-dir websockets==15.0.1 +COPY beyonddb /opt/beyonddb +COPY agent.py /opt/agent.py +RUN mkdir -p /data /config +ENTRYPOINT ["python3", "/opt/agent.py"] diff --git a/deploy/cloudflare/README.md b/deploy/cloudflare/README.md new file mode 100644 index 0000000..1ce9b6a --- /dev/null +++ b/deploy/cloudflare/README.md @@ -0,0 +1,178 @@ +# Cloudflare benchmark deployment + +Run three BeyondDB database Containers, each requesting **1 vCPU, 3072 MiB RAM +and 5000 MB ephemeral disk**. R2 holds Cellar's durable state. This fixture uses +one table with **30 ordered data ranges**, ten on each node. Metadata and +coordinator Cells share each node's resources. + +| Path | Implementation | +| --- | --- | +| Public DynamoDB | HTTPS Worker → Container port 8080 → loopback ExtendDB listener | +| Private peers | Loopback TCP forwarders → secret-gated WSS Worker route → native Cellar peer port | +| Durability | Direct R2 S3 API; writes wait for bucket publication; follower durability disabled | +| Node lifecycle | Explicitly named Durable Objects; native Container API, with monitored process lifetime | +| Resource evidence | Cloudflare instance API allocation, guest/process observations and disk usage | +| Load generator | Workstation outside all three database-node budgets; WAN/Worker/bridge costs are included | + +The bridge preserves native Cellar bytes and authentication. It does not interpret +or replay mutations. Public requests cannot reach the private gateway without the +separate tunnel secret. Keep the generated credentials, keys and raw reports in an +ignored private run directory. Do not mount R2 as a SQLite filesystem. + +## Prepare + +Use a reviewed **Linux x86-64 release binary** built with the committed lockfile. +Cloudflare requires `linux/amd64`; the earlier local ARM binary is unsuitable. +Docker must be available to Wrangler. Install client dependencies separately: + +```sh +python3 -m venv benchmarks/cloudflare-venv +benchmarks/cloudflare-venv/bin/pip install -r scripts/perf/requirements-cloudflare.txt +``` + +Load the protected credential file into the shell environment; do not paste it +into commands, logs or Git. Provisioning expects `CF_ACCOUNT_ID`, `CF_API_TOKEN`, +`CF_ACCESS_KEY_ID`, `CF_SECRET_ACCESS_KEY` and `CF_S3_API_ENDPOINT`. +The API token needs permission to deploy Workers/Containers and create R2 buckets; +the S3 credentials must access the new dedicated bucket. + +```sh +set -a +source "$HOME/.cf.env" +set +a +export CLOUDFLARE_ACCOUNT_ID="$CF_ACCOUNT_ID" +export CLOUDFLARE_API_TOKEN="$CF_API_TOKEN" + +CF_PERF_PY="$PWD/benchmarks/cloudflare-venv/bin/python" +CF_PERF_RUN="$PWD/benchmarks/YYYY-MM-DD-cloudflare-benchmark" +"$CF_PERF_PY" deploy/cloudflare/provision.py \ + --run-dir "$CF_PERF_RUN" --binary /absolute/path/to/linux-amd64/beyonddb +``` + +Provisioning creates a dedicated R2 bucket and checks conditional create/update, +stale-write rejection and read-back. It saves source/binary/lockfile fingerprints, +private keys, Worker secrets and the generated Wrangler configuration. An existing +run directory is rejected. Preserve it after any failure; do not reinitialize it. + +## Deploy and start + +```sh +npm exec --yes --package=wrangler@4.151.2 -- wrangler deploy \ + --config "$CF_PERF_RUN/wrangler.json" +npm exec --yes --package=wrangler@4.151.2 -- wrangler secret bulk \ + "$CF_PERF_RUN/worker-secrets.json" --config "$CF_PERF_RUN/wrangler.json" + +# Initial provisioning only. Later starts omit --bootstrap. +"$CF_PERF_PY" scripts/perf/cloudflare.py start \ + --run-dir "$CF_PERF_RUN" --node 1 --bootstrap +"$CF_PERF_PY" scripts/perf/cloudflare.py ready --run-dir "$CF_PERF_RUN" --node 1 +``` + +Wait for signed `ListTables` to succeed through node 1 before starting nodes 2 +and 3. Start those nodes without `--bootstrap`, then verify signed calls through +each node. A running Container is not proof that its database is ready. + +```sh +"$CF_PERF_PY" scripts/perf/cloudflare.py start --run-dir "$CF_PERF_RUN" --node 2 +"$CF_PERF_PY" scripts/perf/cloudflare.py start --run-dir "$CF_PERF_RUN" --node 3 +"$CF_PERF_PY" scripts/perf/cloudflare.py ready --run-dir "$CF_PERF_RUN" +"$CF_PERF_PY" scripts/perf/cloudflare.py status --run-dir "$CF_PERF_RUN" +``` + +The Worker exposes only three fixed node names. The `durable_object` scheduling +policy has no `max_instances` configuration; the router never creates arbitrary +instance names. A western North America hint does not guarantee one datacenter; +record every node's actual location. The configured inactivity timeout is one +hour; a stopped node requires an explicit `start` without bootstrap. + +## Prepare and measure + +Run the local private tunnel forwarders in a separate terminal. They provide the +existing perf control protocol with access to the remote native peer ports: + +```sh +"$CF_PERF_PY" deploy/cloudflare/agent.py --tunnels "$CF_PERF_RUN/secrets.json" +``` + +With all three signed endpoints ready, prepare once and then measure: + +```sh +"$CF_PERF_PY" scripts/perf/cloudflare.py prepare --run-dir "$CF_PERF_RUN" +"$CF_PERF_PY" scripts/perf/cloudflare.py measure \ + --run-dir "$CF_PERF_RUN" --label main \ + --operations get query put transact_get transact_write \ + --clients 1 8 30 60 --seconds 60 --warmup 5 +``` + +Preparation pauses native balancing for controlled first placement, installs the +table through normal DynamoDB `CreateTable`, seeds every ordered interval and +verifies authoritative owner inventories. The measurement reuses +[`run_case`](../../scripts/perf/bench.py): signed requests, uniform 1-KiB payloads, +strong point/range reads, two-item cross-node transactions and disabled SDK +retries. It keeps successful samples, error latencies, actual elapsed time, +per-range/ingress coverage, resource observations and acknowledged-write read-backs. + +Use a fresh result label for another sweep. Do not rerun `prepare` over an existing +table or creation plan. If `CreateTable` times out, preparation observes the same +generation becoming active and verifies empty ranges before seeding; it never +resubmits creation. `resume-creation` refuses to replay a prior seed submission. +Failed/incomplete windows exit unsuccessfully while retaining their data; +failed mutations may have uncertain outcomes. Cloudflare guest CPU count and +memory totals can include platform overhead; use the instance API to verify the +requested allocation and retain guest observations separately. + +Generate a report and optional chart from every retained sweep: + +```sh +"$CF_PERF_PY" scripts/perf/cloudflare_report.py \ + --input "$CF_PERF_RUN/results-main.json" --output "$CF_PERF_RUN/report" +# Optional plotting dependency, kept outside the runtime image. +"$CF_PERF_PY" -m pip install -r scripts/perf/requirements-plot.txt +"$CF_PERF_PY" scripts/perf/plot.py \ + --input "$CF_PERF_RUN/report/combined-results.json" --output "$CF_PERF_RUN/report" +``` + +## Retain diagnostics and stop + +```sh +"$CF_PERF_PY" scripts/perf/cloudflare.py logs --run-dir "$CF_PERF_RUN" +"$CF_PERF_PY" scripts/perf/cloudflare.py stop --run-dir "$CF_PERF_RUN" +``` + +Collect final inventories and logs before stopping: Container disk is ephemeral. +A clean signal drains BeyondDB; startup failures retain diagnostic access for ten +minutes. R2 data, deployment metadata and encryption keys must remain available +for recovery. A restart omits bootstrap and restores state through Cellar. +Never infer durable user-item recovery from a successful credential lookup alone; +read back acknowledged items after the relevant restart. + +For a planned all-node cold restart, first retain final logs and inventories, +stop all three and observe them stopped before starting any node. Start without +`--bootstrap`, wait for signed readiness, then verify the original acknowledged +keys and one seed value per range: + +```sh +"$CF_PERF_PY" scripts/perf/cloudflare.py start --run-dir "$CF_PERF_RUN" --node 1 +"$CF_PERF_PY" scripts/perf/cloudflare.py start --run-dir "$CF_PERF_RUN" --node 2 +"$CF_PERF_PY" scripts/perf/cloudflare.py start --run-dir "$CF_PERF_RUN" --node 3 +"$CF_PERF_PY" scripts/perf/cloudflare.py ready --run-dir "$CF_PERF_RUN" +"$CF_PERF_PY" scripts/perf/cloudflare.py check-writes --run-dir "$CF_PERF_RUN" \ + --results "$CF_PERF_RUN/results-main.json" --label cold-restart +``` + +This read-only check retains every result, including missing/different values and +request errors. It samples original successful writes; it does not resolve failed +mutations or qualify every crash/transaction recovery boundary. Cold recovery can +change ownership, so retain a new authoritative inventory rather than assuming +the pre-restart ten-per-node placement. Start every recovery peer before waiting +for signed readiness: restored metadata may need a different owner. + +Compare requests/s with p50/p95/p99, successful counts, errors, coverage, resource +use and locations. These closed-loop experiments do not establish an arrival-rate +SLO or peak fleet capacity. See [performance interpretation](../../docs/performance.md#read-the-report) +and the [recorded Cloudflare experiment](../../docs/cloudflare-benchmark.md), whose +cold recovery required operator assistance after the five-minute readiness guard. + +References: [Container sizes](https://developers.cloudflare.com/containers/platform/limits/), +[Container API](https://developers.cloudflare.com/containers/api/durable-object-container/), +[R2 conditional operations](https://developers.cloudflare.com/r2/api/s3/api/). diff --git a/deploy/cloudflare/agent.py b/deploy/cloudflare/agent.py new file mode 100644 index 0000000..4eec70d --- /dev/null +++ b/deploy/cloudflare/agent.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +"""Preserve native Cellar peer bytes over authenticated WSS; supervise one node. + +Public HTTP terminates TLS at the Worker and reaches loopback through a raw TCP +forwarder. Private peer ports are exposed only by the Worker's secret-gated +WebSocket bridge. The same bridge supports the unmodified perf control protocol. +""" +import argparse +import asyncio +from contextlib import suppress +import json +import os +from pathlib import Path +import resource +import signal +import subprocess +import time + +from websockets.asyncio.client import connect + + +async def copy_stream(reader, writer): + try: + while data := await reader.read(65536): + writer.write(data) + await writer.drain() + finally: + writer.close() + + +async def local_forward(reader, writer, port): + try: + remote_reader, remote_writer = await asyncio.open_connection('127.0.0.1', port) + await asyncio.gather(copy_stream(reader, remote_writer), copy_stream(remote_reader, writer)) + except (OSError, ConnectionError): + writer.close() + + +async def peer_forward(reader, writer, index, settings): + try: + url = settings['worker_url'].replace('https://', 'wss://') + f'/_peer/{index}' + async with connect(url, additional_headers={'Authorization': 'Bearer ' + settings['tunnel_token']}, + max_size=None, compression=None, open_timeout=30, + user_agent_header='BeyondDBPeerBridge/0.1', + ping_interval=20, ping_timeout=30) as ws: + async def outbound(): + while data := await reader.read(65536): + await ws.send(data) + await ws.close() + async def inbound(): + async for data in ws: + if not isinstance(data, bytes): + raise ValueError('unexpected nonbinary peer frame') + writer.write(data) + await writer.drain() + tasks = [asyncio.create_task(outbound()), asyncio.create_task(inbound())] + try: + await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + except Exception as error: + # Do not log URLs, auth headers or request bytes. + print(f'peer bridge {index}: {type(error).__name__}', flush=True) + finally: + writer.close() + with suppress(OSError): + await writer.wait_closed() + + +def read_optional(path): + try: + return Path(path).read_text() + except OSError: + return None + + +def snapshot(process): + return { + 'node': int(os.environ['NODE_INDEX']), 'observed_at': time.time(), + 'pid': process.pid, 'exit_code': process.poll(), + 'platform': {k: v for k, v in os.environ.items() if k.startswith('CLOUDFLARE_')}, + 'cpu_count': os.cpu_count(), 'meminfo': read_optional('/proc/meminfo'), + 'process_status': read_optional(f'/proc/{process.pid}/status'), + 'process_limits': read_optional(f'/proc/{process.pid}/limits'), + 'agent_nofile_limit': resource.getrlimit(resource.RLIMIT_NOFILE), + 'process_open_fds': len(list(Path(f'/proc/{process.pid}/fd').iterdir())) if process.poll() is None else None, + 'cgroup': {p: read_optional('/sys/fs/cgroup/' + p) for p in + ('cpu.max', 'cpu.stat', 'memory.max', 'memory.current', 'memory.events', 'io.stat')}, + 'disk': dict(zip(('total', 'used', 'free'), __import__('shutil').disk_usage('/data'))), + } + + +async def status_handler(reader, writer, process): + try: + line = await reader.readline() + path = line.split()[1].decode() + await reader.readuntil(b'\r\n\r\n') + if path == '/logs': + body = Path('/data/server.log').read_bytes()[-48000:] + kind = 'text/plain' + else: + body = json.dumps(snapshot(process)).encode() + kind = 'application/json' + writer.write(f'HTTP/1.1 200 OK\r\nContent-Type: {kind}\r\nContent-Length: {len(body)}\r\nConnection: close\r\n\r\n'.encode() + body) + await writer.drain() + except (OSError, ValueError, asyncio.IncompleteReadError): + pass + finally: + writer.close() + + +def configure(): + index = int(os.environ['NODE_INDEX']) + root = Path('/config'); root.mkdir(mode=0o700, exist_ok=True) + Path('/data').mkdir(exist_ok=True) + for name, env in [('peer.key', 'PEER_KEY_HEX'), ('encryption.key', 'ENCRYPTION_KEY_HEX')]: + p = root/name; p.write_bytes(bytes.fromhex(os.environ[env])); p.chmod(0o600) + policy = {'Version': '2012-10-17', 'Statement': [{'Effect': 'Allow', 'Action': 'dynamodb:*', 'Resource': '*'}]} + (root/'policy.json').write_text(json.dumps(policy)) + config = { + 'node_name': f'perf30-node{index}', 'data_dir': '/data', 'region': 'us-east-1', + 'public_bind': '127.0.0.1:18000', 'public_endpoint': os.environ['WORKER_URL'], + 'peer_bind': '0.0.0.0:9000', 'peer_advertise': f'127.0.0.1:{19000+index}', + 'peer_key_file': '/config/peer.key', 'encryption_key_file': '/config/encryption.key', + 'accounts': ['123456789012'], 'initial_partitions': 30, 'max_active_cells': 2048, + 'split_threshold_bytes': 1073741824, 'auth_cache_enabled': False, + 'shutdown_mode': 'preserve', + 'fleet': {'bucket': os.environ['R2_BUCKET'], 'endpoint': os.environ['R2_ENDPOINT'], + 'prefix': 'ordered-v1', 'follower_durability': False}, + 'bootstrap': {'account_id': '123456789012', 'access_key_id': os.environ['DDB_ACCESS_KEY'], + 'principal_name': 'perf', 'policy_name': 'benchmark', 'policy_file': '/config/policy.json'}, + } + # The executable's loopback listener requires an HTTP advertised endpoint; + # the public HTTPS proxy is recorded separately by the deployment manifest. + config['public_endpoint'] = 'http://127.0.0.1:18000' + (root/'node.json').write_text(json.dumps(config)) + return index + + +async def main(settings=None): + if settings is not None: + servers = [await asyncio.start_server(lambda r, w, i=i: peer_forward(r, w, i, settings), + '127.0.0.1', 19000+i) for i in range(1, 4)] + await asyncio.Event().wait() + return + index = configure() + try: + resource.setrlimit(resource.RLIMIT_NOFILE, (65536, 65536)) + except (ValueError, PermissionError) as error: + raise RuntimeError('The benchmark requires an effective nofile limit of 65536') from error + print('Effective nofile limit:', resource.getrlimit(resource.RLIMIT_NOFILE), flush=True) + settings = {'worker_url': os.environ['WORKER_URL'], 'tunnel_token': os.environ['TUNNEL_TOKEN']} + servers = [] + for target in range(1, 4): + handler = (lambda r,w: local_forward(r,w,9000)) if target == index else ( + lambda r,w,i=target: peer_forward(r,w,i,settings)) + servers.append(await asyncio.start_server(handler, '127.0.0.1', 19000+target)) + env = dict(os.environ, AWS_VIRTUAL_HOSTED_STYLE_REQUEST='false', RUST_LOG='info') + command = ['/opt/beyonddb', '/config/node.json'] + bootstrap = os.environ.get('BOOTSTRAP') == '1' + if bootstrap: + command.append('--bootstrap') + log = open('/data/server.log', 'ab', buffering=0) + process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env) + process.stdin.write(os.environ['DDB_SECRET_KEY'].encode() if bootstrap else b'') + process.stdin.close() + servers.append(await asyncio.start_server(lambda r,w: local_forward(r,w,18000), '0.0.0.0', 8080)) + servers.append(await asyncio.start_server(lambda r,w: status_handler(r,w,process), '0.0.0.0', 8090)) + async def capture_logs(): + sensitive = [os.environ[k] for k in ('AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY', + 'PEER_KEY_HEX', 'ENCRYPTION_KEY_HEX', 'DDB_SECRET_KEY', 'TUNNEL_TOKEN')] + while line := await asyncio.to_thread(process.stdout.readline): + text = line.decode(errors='replace') + for secret in sensitive: + text = text.replace(secret, '[redacted]') + log.write(text.encode()) + print(text, end='', flush=True) + log_task = asyncio.create_task(capture_logs()) + loop = asyncio.get_running_loop() + stopping = asyncio.Event() + def request_stop(): + stopping.set() + if process.poll() is None: + process.send_signal(signal.SIGTERM) + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, request_stop) + while process.poll() is None: + await asyncio.sleep(1) + await log_task + print('BeyondDB exited', process.returncode, flush=True) + if not stopping.is_set(): + # Retain startup-failure diagnostics without pretending the DB is ready. + with suppress(asyncio.TimeoutError): + await asyncio.wait_for(stopping.wait(), 600) + for server in servers: + server.close() + raise SystemExit(process.returncode) + + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument('--tunnels', type=Path) + args = parser.parse_args() + asyncio.run(main(json.loads(args.tunnels.read_text()) if args.tunnels else None)) diff --git a/deploy/cloudflare/provision.py b/deploy/cloudflare/provision.py new file mode 100644 index 0000000..860da96 --- /dev/null +++ b/deploy/cloudflare/provision.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +"""Create a dedicated three-node benchmark configuration and R2 bucket. + +Requires the CF_* variables from a protected credential file in the environment. +This creates billable resources; it never overwrites a retained run directory. +""" +import argparse +import hashlib +import json +import os +from pathlib import Path +import secrets +import shutil +import subprocess +import time +import urllib.request + +import boto3 +from botocore.config import Config + + +def api(path, value=None): + request = urllib.request.Request( + f"https://api.cloudflare.com/client/v4/accounts/{os.environ['CF_ACCOUNT_ID']}/{path}", + data=None if value is None else json.dumps(value).encode(), + headers={'Authorization': 'Bearer '+os.environ['CF_API_TOKEN'], 'Content-Type': 'application/json'}) + with urllib.request.urlopen(request, timeout=30) as response: + result = json.load(response) + if not result.get('success'): + raise RuntimeError('Cloudflare API request failed') + return result['result'] + + +def private(path, value): + with path.open('x', opener=lambda p,f: os.open(p,f,0o600)) as stream: + stream.write(json.dumps(value,indent=2)+'\n') + + +def main(): + parser=argparse.ArgumentParser(description=__doc__) + parser.add_argument('--run-dir',required=True,type=Path) + parser.add_argument('--binary',required=True,type=Path) + args=parser.parse_args() + required=('CF_ACCOUNT_ID','CF_API_TOKEN','CF_ACCESS_KEY_ID','CF_SECRET_ACCESS_KEY','CF_S3_API_ENDPOINT') + if any(not os.environ.get(k) for k in required): + parser.error('Source the protected CF_* credential file first') + binary=args.binary.read_bytes() + if binary[:4] != b'\x7fELF' or int.from_bytes(binary[18:20],'little') != 62: + parser.error('A Linux x86-64 release binary is required') + root=args.run_dir.resolve() + root.mkdir(parents=True,exist_ok=True,mode=0o700) + if any(root.iterdir()): + parser.error('Use a fresh empty run directory; retain old manifests, credentials and data') + suffix=secrets.token_hex(3) + name='beyonddb-perf-'+suffix + subdomain=api('workers/subdomain')['subdomain'] + settings={'worker_url':f'https://{name}.{subdomain}.workers.dev', + 'tunnel_token':secrets.token_urlsafe(48),'admin_token':secrets.token_urlsafe(48), + 'access_key':'AKIABDBCFPERF00001','secret':secrets.token_urlsafe(40), + 'peer_key':secrets.token_bytes(32).hex(),'encryption_key':secrets.token_bytes(32).hex()} + repo=Path(__file__).resolve().parents[2] + meta={'worker':name,'bucket':name,'prefix':'ordered-v1','nodes':3,'cells_per_node':10, + 'source_revision':subprocess.check_output(['git','rev-parse','HEAD'],cwd=repo,text=True).strip(), + 'created_at':time.time(),'vcpu':1,'memory_mib':3072,'disk_mb':5000, + 'binary_sha256':hashlib.sha256(binary).hexdigest(), + 'lockfile_sha256':hashlib.sha256((repo/'Cargo.lock').read_bytes()).hexdigest(), + 'status':'configuration-created'} + private(root/'deployment.json',meta) + private(root/'secrets.json',settings) + worker_secrets={'R2_ENDPOINT':os.environ['CF_S3_API_ENDPOINT'],'R2_BUCKET':name, + 'R2_ACCESS_KEY_ID':os.environ['CF_ACCESS_KEY_ID'], + 'R2_SECRET_ACCESS_KEY':os.environ['CF_SECRET_ACCESS_KEY'], + 'PEER_KEY_HEX':settings['peer_key'],'ENCRYPTION_KEY_HEX':settings['encryption_key'], + 'DDB_ACCESS_KEY':settings['access_key'],'DDB_SECRET_KEY':settings['secret'], + 'TUNNEL_TOKEN':settings['tunnel_token'],'ADMIN_TOKEN':settings['admin_token']} + private(root/'worker-secrets.json',worker_secrets) + image=root/'image';image.mkdir() + shutil.copyfile(args.binary,image/'beyonddb');(image/'beyonddb').chmod(0o755) + for f in ('agent.py','Dockerfile'):shutil.copyfile(repo/'deploy/cloudflare'/f,image/f) + shutil.copyfile(repo/'deploy/cloudflare/worker.js',root/'worker.js') + config={'name':name,'main':'worker.js','compatibility_date':'2026-10-19', + 'account_id':os.environ['CF_ACCOUNT_ID'],'workers_dev':True, + 'containers':[{'class_name':'BeyondNode','scheduling_policy':'durable_object', + 'images':{'base':{'dockerfile':'./image/Dockerfile','build_context':'./image'}}}], + 'durable_objects':{'bindings':[{'name':'NODES','class_name':'BeyondNode'}]}, + 'exports':{'BeyondNode':{'type':'durable-object','storage':'sqlite'}}, + 'observability':{'enabled':True,'head_sampling_rate':1}, + 'vars':{'WORKER_URL':settings['worker_url']}} + private(root/'wrangler.json',config) + api('r2/buckets',{'name':name}) + s3=boto3.client('s3',region_name='auto',endpoint_url=os.environ['CF_S3_API_ENDPOINT'], + aws_access_key_id=os.environ['CF_ACCESS_KEY_ID'], + aws_secret_access_key=os.environ['CF_SECRET_ACCESS_KEY'], + config=Config(retries={'total_max_attempts':1},request_checksum_calculation='when_required')) + key='preflight/conditional-write' + first=s3.put_object(Bucket=name,Key=key,Body=b'first',IfNoneMatch='*') + for condition in ({'IfNoneMatch':'*'},{'IfMatch':'"not-current"'}): + try: + s3.put_object(Bucket=name,Key=key,Body=b'invalid',**condition) + except s3.exceptions.ClientError as error: + if error.response['ResponseMetadata']['HTTPStatusCode'] != 412:raise + else:raise RuntimeError('R2 accepted a stale conditional write') + s3.put_object(Bucket=name,Key=key,Body=b'second',IfMatch=first['ETag']) + if s3.get_object(Bucket=name,Key=key)['Body'].read()!=b'second': + raise RuntimeError('R2 conditional update read-back mismatch') + meta['r2_conditional_preflight']='passed' + (root/'deployment.json').write_text(json.dumps(meta,indent=2)+'\n') + print(f'Prepared {name}; conditional R2 writes verified. Private files: {root}') + + +if __name__=='__main__':main() diff --git a/deploy/cloudflare/worker.js b/deploy/cloudflare/worker.js new file mode 100644 index 0000000..c0509a7 --- /dev/null +++ b/deploy/cloudflare/worker.js @@ -0,0 +1,141 @@ +import { DurableObject } from 'cloudflare:workers'; + +// Exactly three explicitly named database nodes; no request-driven random instances. +export class BeyondNode extends DurableObject { + constructor(ctx, env) { + super(ctx, env); + this.monitoring = false; + if (ctx.container?.running) this.watch(); + } + watch() { + if (this.monitoring) return; + this.monitoring = true; + this.ctx.waitUntil(this.ctx.container.monitor().then( + () => this.ctx.storage.put('lastStop', { at: Date.now(), outcome: 'stopped' }), + error => this.ctx.storage.put('lastStop', { at: Date.now(), outcome: 'failed', message: error.message, exitCode: error.exitCode }), + ).finally(() => { this.monitoring = false; })); + } + async start(index, bootstrap = false) { + const c = this.ctx.container; + if (!c.running) { + c.start({ + image: c.images.base, + instance: { vcpu: 1, memoryMib: 3072, diskMb: 5000 }, + enableInternet: true, + env: { + NODE_INDEX: String(index), + BOOTSTRAP: bootstrap ? '1' : '0', + WORKER_URL: this.env.WORKER_URL, + R2_ENDPOINT: this.env.R2_ENDPOINT, + R2_BUCKET: this.env.R2_BUCKET, + AWS_ACCESS_KEY_ID: this.env.R2_ACCESS_KEY_ID, + AWS_SECRET_ACCESS_KEY: this.env.R2_SECRET_ACCESS_KEY, + AWS_REGION: 'auto', + PEER_KEY_HEX: this.env.PEER_KEY_HEX, + ENCRYPTION_KEY_HEX: this.env.ENCRYPTION_KEY_HEX, + DDB_ACCESS_KEY: this.env.DDB_ACCESS_KEY, + DDB_SECRET_KEY: this.env.DDB_SECRET_KEY, + TUNNEL_TOKEN: this.env.TUNNEL_TOKEN, + }, + labels: { benchmark: 'beyonddb-perf', node: String(index) }, + }); + await c.setInactivityTimeout(60 * 60 * 1000); + } + this.watch(); + return { running: c.running, info: await c.inspect() }; + } + async fetch(request) { + if (new URL(request.url).pathname.startsWith('/_peer/')) return this.tunnel(); + return this.publicFetch(request); + } + async publicFetch(request) { + const upstream = new Request(request.url.replace(/^https:/, 'http:'), request); + return this.ctx.container.getTcpPort(8080).fetch(upstream); + } + async inspectNode() { + const c = this.ctx.container; + if (!c.running) return { running: false, lastStop: await this.ctx.storage.get('lastStop') }; + const response = await c.getTcpPort(8090).fetch('http://agent/status'); + return { running: c.running, info: await c.inspect(), agent: await response.json() }; + } + async logs() { + const response = await this.ctx.container.getTcpPort(8090).fetch('http://agent/logs'); + return new Response(response.body, { headers: { 'content-type': 'text/plain' } }); + } + async stop() { + if (!this.ctx.container.running) return { running: false }; + this.ctx.container.signal(15); + return { stopping: true }; + } + async tunnel() { + // The public Worker validates the independent tunnel secret before this RPC. + const socket = this.ctx.container.getTcpPort(9000).connect('10.0.0.1:9000'); + await socket.opened; + const [client, server] = Object.values(new WebSocketPair()); + server.accept(); + server.binaryType = 'arraybuffer'; + const writer = socket.writable.getWriter(); + let writes = Promise.resolve(); + let closed = false; + const close = () => { + if (closed) return; + closed = true; + socket.close(); + try { server.close(1000); } catch {} + }; + server.addEventListener('message', event => { + writes = writes.then(async () => { + if (typeof event.data === 'string') throw new Error('Peer frames must be binary'); + const buffer = event.data instanceof Blob ? await event.data.arrayBuffer() : event.data; + await writer.write(new Uint8Array(buffer)); + }).catch(error => { console.error('Peer writer failed', error.message); close(); }); + }); + server.addEventListener('close', close); + server.addEventListener('error', close); + this.ctx.waitUntil((async () => { + try { + const reader = socket.readable.getReader(); + while (!closed) { + const { value, done } = await reader.read(); + if (done) break; + server.send(value.buffer.slice(value.byteOffset, value.byteOffset + value.byteLength)); + } + } catch (error) { console.error('Peer reader failed', error.message); } finally { close(); } + })()); + return new Response(null, { status: 101, webSocket: client }); + } +} + +function authorized(request, token) { + return typeof token === 'string' && token.length >= 32 && + request.headers.get('authorization') === `Bearer ${token}`; +} +function node(env, index) { + if (!Number.isInteger(index) || index < 1 || index > 3) throw new Error('Invalid node'); + return env.NODES.get(env.NODES.idFromName(`node-${index}`), { locationHint: 'wnam' }); +} +export default { + async fetch(request, env) { + const url = new URL(request.url); + const peer = url.pathname.match(/^\/_peer\/([1-3])$/); + if (peer) { + if (!authorized(request, env.TUNNEL_TOKEN)) return new Response('Forbidden', { status: 403 }); + if (request.headers.get('upgrade')?.toLowerCase() !== 'websocket') return new Response('WebSocket required', { status: 400 }); + return node(env, Number(peer[1])).fetch(request); + } + const admin = url.pathname.match(/^\/_admin\/([1-3])\/(start|status|logs|stop)$/); + if (admin) { + if (!authorized(request, env.ADMIN_TOKEN)) return new Response('Forbidden', { status: 403 }); + const index = Number(admin[1]), target = node(env, index); + if (admin[2] === 'start' && request.method === 'POST') return Response.json(await target.start(index, index === 1 && url.searchParams.get('bootstrap') === '1')); + if (admin[2] === 'stop' && request.method === 'POST') return Response.json(await target.stop()); + if (admin[2] === 'status' && request.method === 'GET') return Response.json(await target.inspectNode()); + if (admin[2] === 'logs' && request.method === 'GET') return target.logs(); + return new Response('Method not allowed', { status: 405 }); + } + if (url.pathname !== '/' && url.pathname !== '/health') return new Response('Not found', { status: 404 }); + const index = Number(request.headers.get('x-beyonddb-node') || '1'); + if (![1, 2, 3].includes(index)) return new Response('Invalid ingress node', { status: 400 }); + return node(env, index).fetch(request); + }, +}; diff --git a/diagram/cloudflare-perf-throughput-latency.svg b/diagram/cloudflare-perf-throughput-latency.svg new file mode 100644 index 0000000..d0fce63 --- /dev/null +++ b/diagram/cloudflare-perf-throughput-latency.svg @@ -0,0 +1,3108 @@ + + + + + + + + 2026-10-19T10:31:05.709941 + image/svg+xml + + + Matplotlib v3.11.2, https://matplotlib.org/ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/docs/README.md b/docs/README.md index 22d251d..cd913e4 100644 --- a/docs/README.md +++ b/docs/README.md @@ -18,7 +18,7 @@ below to evaluate a deployment; these guides do not establish a production SLA. | Send signed requests | [AWS CLI guide](user-guide.md) | | Check an operation or option | [API coverage](api.md) | | Inspect failures and operational limits | [Operations](operations.md) | -| Measure a workload | [Performance](performance.md) · [Local benchmark](local-benchmark.md) | +| Measure a workload | [Performance](performance.md) · [Local benchmark](local-benchmark.md) · [Cloudflare + R2](cloudflare-benchmark.md) | ## Design and evidence diff --git a/docs/cloudflare-benchmark.md b/docs/cloudflare-benchmark.md new file mode 100644 index 0000000..3c78541 --- /dev/null +++ b/docs/cloudflare-benchmark.md @@ -0,0 +1,180 @@ +# BeyondDB on Cloudflare Containers and R2 + +**Measured October 19, 2026:** three database nodes, one table across **30 ordered +data Cells**, ten per node. All 15 measured windows completed with **8,831 +successful API calls, zero measured errors and full range coverage**. A separate +60-client transaction-read warmup returned **two `ServiceUnavailable` errors**; +the pressure sweep therefore exited unsuccessfully. Transaction latency is high +and this run does not qualify production capacity or a latency SLO. + +| Deployment | Recorded setting | +| --- | --- | +| Each database node | **1 vCPU / 3072 MiB memory / 5000 MB ephemeral disk**, confirmed through Cloudflare's instance API | +| Locations during measurement | Node 1: `sea01`; node 2: `sjc08`; node 3: `lax09`; western North America, across metros | +| Table placement | 30 ordered ranges; balancing paused; authoritative ownership verified before and after both sweeps | +| Other Cells | Account, credentials, directory, inventory and transaction coordinators share the same budgets; end of measurement: **852 total owned Cells** | +| Durability | Dedicated R2 bucket; direct S3 API; acknowledgement waits for bucket publication; follower durability disabled | +| Executable | Unchanged release source `57bec2e`; Cellar `=0.1.0`; committed ExtendDB revision and lockfile | +| Runtime transport | Public HTTPS Worker → Container → loopback ExtendDB; authenticated WSS bridge carries native Cellar peer bytes | +| Client | External workstation; WAN, signing, authorization, Worker ingress, routing and durability included | + +Cloudflare's custom instances require at least **3 GiB per vCPU**. The memory +allocation above is 3 GiB, rather than 3 decimal GB. Guest CPU count and +`MemTotal` include platform details and are not the allocation proof. See the +[platform limits](https://developers.cloudflare.com/containers/platform/limits/). + +## Throughput and response latency + +Each API runs in its own closed-loop window. The reference sweep uses eight +clients and 60 seconds per window; the pressure sweep uses 30 then 60 clients +and 30 seconds per window. Each has five seconds of separate warmup. Actual +elapsed time includes in-flight completion drain. + +| Workload | Request contents | +| --- | --- | +| `GetItem` | One existing item, strongly consistent | +| `Query` | One PK with numeric SK `BETWEEN 1 AND 2`, two returned items, strongly consistent | +| `PutItem` | One new item | +| `TransactGetItems` | Two existing items in different data Cells and nodes | +| `TransactWriteItems` | Two new items in different data Cells and nodes; fresh client token | + +Payloads contain a uniform 1-KiB value. Traffic rotates across every range and all +three ingress nodes. SDK retries and authorization caching are disabled. There +are no intentional key conflicts, hot keys, secondary indexes, TTL or Streams. + +![Successful API throughput and p99 by concurrency; the red cross marks the transaction-read window whose separate warmup failed](../diagram/cloudflare-perf-throughput-latency.svg) + +| API | 8 clients: req/s | 30 clients: req/s | 60 clients: req/s | +| --- | ---: | ---: | ---: | +| GetItem | 12.279 | 40.142 | 48.973 | +| Query | 11.402 | 35.395 | 46.810 | +| PutItem | 3.720 | 12.246 | 23.937 | +| TransactGetItems | 0.328 | 1.023 | 1.450 | +| TransactWriteItems | 0.461 | 1.188 | 1.746 | + +At **60 clients**, the successful-call latency distribution was: + +| API | p50 ms | p95 ms | p99 ms | Successful samples | Measured errors | Separate warmup errors | +| --- | ---: | ---: | ---: | ---: | ---: | ---: | +| GetItem | 1032.634 | 3094.719 | 3664.695 | 1,515 | 0 | 0 | +| Query | 725.847 | 3969.660 | 5638.317 | 1,486 | 0 | 0 | +| PutItem | 2269.819 | 3923.116 | 5264.701 | 779 | 0 | 0 | +| TransactGetItems | 29141.885 | 35150.918 | 37312.382 | 90 | 0 | 2 | +| TransactWriteItems | 26464.494 | 32634.609 | 33301.932 | 102 | 0 | 0 | + +Requests/s counts API calls. Two-item Query and transaction item rates are twice +their request rates. Percentiles use nearest rank over successful calls only. +Every transaction window, every PutItem window and all eight-client windows +have fewer than 1,000 successful samples: their p99 estimates are weak. The +[reviewed CSV](data/cloudflare-benchmark.csv) retains every window's counts, +elapsed time, errors, coverage and percentiles. + +These results show that adding clients increases throughput while response +latency remains substantial. They do not isolate the cause. The deployment +includes cross-metro peers, the WSS adapter, R2 and shared metadata paths; phase +tracing and controlled comparisons are needed to attribute their costs. The +earlier [local benchmark](local-benchmark.md) uses different infrastructure and +does not provide an equivalent baseline. + +## Resource and durability observations + +| Node | Owned Cells at end of measurement | Peak sampled server RSS MiB | Maximum sampled disk use GB | Peak sampled open FDs | +| --- | ---: | ---: | ---: | ---: | +| 1 | 372 | 468.4 | 0.524 | 3,492 | +| 2 | 237 | 717.9 | 0.548 | 2,275 | +| 3 | 243 | 581.0 | 0.490 | 2,325 | + +Samples were collected approximately every five seconds. These are observed +values rather than hard peaks; server RSS excludes Python, the guest and other +platform memory. Disk observations include the image/runtime and retained setup +history. All nodes used a verified `nofile=65536:65536`. Guest cgroup CPU/I/O +counters were retained; root CPU/memory limit files were unavailable. + +R2 preflight verified conditional creation, conditional replacement, rejection +of stale conditions and read-back. After the six write windows, **180 acknowledged +values** were read back consistently: one per data range per write window. +This samples successful writes rather than proving every written item or every +transaction recovery boundary. + +### Operator-assisted cold recovery + +All three Containers were observed stopped before any restart. Startup omitted +bootstrap and reused the same binary, image, keys and R2 roots. Cloudflare starts +stopped Containers from a [fresh ephemeral disk](https://developers.cloudflare.com/containers/concepts/architecture/#use-snapshots); +no filesystem snapshot or retained local data volume was restored. + +| Observation | Result | +| --- | --- | +| All Containers stopped | `17:31:02 UTC` | +| Initial signed-readiness guard | Node 1 exceeded **300 seconds**; peers were started during this initial wait | +| Recovery discovery | **531 registered coordinator shards**; startup's recovery pass precedes public serving | +| Operator assistance | Six concurrent native **read-only pending-boundary queries** prefetched all 531 existing coordinator Cells; all final queries succeeded; 19 intermediate stale-route responses were retained while selecting the serving owner | +| All three signed endpoints ready | `17:45:40 UTC` | +| Consistent original-value checks | **210/210 passed**: 180 acknowledged write witnesses plus 30 seed values | +| Post-recovery table ownership | **30 unique resident data Cells**, ten per node; same three locations and verified resource allocation | + +The prefetch ran from `17:41:50` to `17:45:32 UTC`. It did not skip the startup +recovery pass, alter authority records or resubmit mutations. The failed initial +probe, routing responses, initial timeout, prefetch script and read-back results +remain in the private artifacts. These timings include staggered starts and +operator assistance; **unattended recovery time remains unqualified**. The result +establishes sampled recovery of user values from R2 for this experiment, rather +than every crash, owner-loss or transaction-atomicity boundary. + +## Retained setup failures + +The initial setup was not clean. All failed attempts and the exact measured +harness remain in the private local run directory. + +| Observation | Handling and evidence boundary | +| --- | --- | +| HTTPS forwarded to a plaintext internal port; WebSocket binary messages arrived as `Blob` | Fixed the deployment adapter's scheme and binary conversion before measurement; native protocol bytes remain unchanged | +| Restart attempted bootstrap over an existing credential | Bootstrap became explicit and first-start-only; existing identity was retained | +| Native process exhausted file descriptors | Raised and verified soft/hard limits to 65,536; retained the original failure logs; the old hard limit was not directly observed | +| Node 2 self-fenced during setup after ambiguous/unconfirmed R2 node-lease renewals | Retained its log and restarted normally; the underlying cause is unisolated; lease deadlines and fencing were unchanged | +| `CreateTable` exceeded the client's 45-second timeout | Observed the original generation becoming `ACTIVE`; no duplicate creation submission; verified empty ranges before seeding 240 acknowledged items | +| Earlier restored ownership was 19/1/10 | Verified the earlier task-created table was empty and retired it with normal `DeleteTable`; created the measured generation with controlled 10/10/10 placement; retained earlier roots/history | +| Two transaction-read warmup 503s at 60 clients | Kept both errors; no retry or durability/admission change; the pressure sweep exits nonzero | +| Cold signed readiness exceeded the five-minute observation guard | Retained the timeout; all peers eventually became ready with read-only operator prefetch; recovery checks passed without weakening runtime gates | + +No runtime authority, lease or SQLite records were edited. A timeout after +submission does not establish that a mutation failed to execute. Recovery and +diagnostics must retain that distinction. + +## Reproduce and inspect + +Use the [deployment runbook](../deploy/cloudflare/README.md) for protected +credentials, image deployment, readiness, controlled placement, measurement, +reporting and shutdown. This fixture creates billable resources. Keep its keys +and retained data together; do not rerun provisioning over an existing directory. + +The measured sweeps used: + +```sh +"$CF_PERF_PY" scripts/perf/cloudflare.py measure \ + --run-dir "$CF_PERF_RUN" --label reference \ + --clients 8 --seconds 60 --warmup 5 \ + --operations get query put transact_get transact_write + +"$CF_PERF_PY" scripts/perf/cloudflare.py measure \ + --run-dir "$CF_PERF_RUN" --label pressure \ + --clients 30 60 --seconds 30 --warmup 5 \ + --operations get query put transact_get transact_write +``` + +| Reproducibility field | Recorded value | +| --- | --- | +| Full source revision | `57bec2e52fb02697ad7c3744afdd82cb77c450f0` | +| Binary SHA-256 | `cb7cb9817b85e386baa33bb8fe056c1dc7f0f79646134047b0f94640be44220c` | +| Cargo.lock SHA-256 | `acb3beb63e18043d10855a2ff6835a910c40978a3ba4f952d420be9b989a9f98` | +| Measured image digest | `sha256:3f839c02ce2c9e794e3cd2f5206d72046ce7e38ee8212c43deb8adcc0fc2d2a1` | +| Build | Rust 1.98.1; cargo-zigbuild 0.23.4; Zig 0.14.1; LLVM ar/ranlib; `x86_64-unknown-linux-gnu.2.36` | +| Client read timeout | 45 seconds; SDK retry count zero | +| Local retained evidence | `benchmarks/2026-10-19-cloudflare-benchmark/` | + +Raw artifacts include all successful latency samples, failures, warmup files, +before/after authoritative inventories, resource samples, logs, deployment +fingerprints and the exact measured harness. They remain ignored and private. +Failed/incomplete windows must remain in reports. A closed-loop test has no +coordinated-omission correction and does not establish an arrival-rate SLO, +automatic-balancing performance or unattended recovery qualification. diff --git a/docs/data/cloudflare-benchmark.csv b/docs/data/cloudflare-benchmark.csv new file mode 100644 index 0000000..723dcae --- /dev/null +++ b/docs/data/cloudflare-benchmark.csv @@ -0,0 +1,16 @@ +label,operation,clients,completed,errors,elapsed_s,requests_per_s,items_per_s,p50_ms,p95_ms,p99_ms,covers_all_cells,write_witnesses_verified,warmup_errors +reference,get,8,744,0,60.593,12.279,12.279,381.705,2021.797,3249.688,True,0,0 +reference,query,8,731,0,64.114,11.402,22.803,386.957,2047.585,2867.136,True,0,0 +reference,put,8,235,0,63.17,3.72,3.72,1893.612,3417.02,4132.406,True,30,0 +reference,transact_get,8,29,0,88.369,0.328,0.656,19855.678,23915.324,29522.735,True,0,0 +reference,transact_write,8,32,0,69.465,0.461,0.921,15921.666,22660.738,22787.345,True,30,0 +pressure,get,30,1284,0,31.986,40.142,40.142,455.356,2239.338,3069.228,True,0,0 +pressure,query,30,1265,0,35.739,35.395,70.79,472.257,2114.483,2708.979,True,0,0 +pressure,put,30,419,0,34.215,12.246,12.246,1978.447,4525.999,5283.109,True,30,0 +pressure,transact_get,30,60,0,58.647,1.023,2.046,24470.018,29568.249,31799.085,True,0,0 +pressure,transact_write,30,60,0,50.498,1.188,2.376,21668.138,25667.197,26635.034,True,30,0 +pressure,get,60,1515,0,30.935,48.973,48.973,1032.634,3094.719,3664.695,True,0,0 +pressure,query,60,1486,0,31.745,46.81,93.62,725.847,3969.66,5638.317,True,0,0 +pressure,put,60,779,0,32.544,23.937,23.937,2269.819,3923.116,5264.701,True,30,0 +pressure,transact_get,60,90,0,62.054,1.45,2.901,29141.885,35150.918,37312.382,True,0,2 +pressure,transact_write,60,102,0,58.412,1.746,3.492,26464.494,32634.609,33301.932,True,30,0 diff --git a/docs/performance.md b/docs/performance.md index 1c71162..9e120d7 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -7,6 +7,7 @@ server and client preparation steps. | I need to… | Start here | | --- | --- | | Read the Cellar measurements and their limits | [Recorded result](#current-local-result) | +| Inspect the three-node Cloudflare + R2 experiment | [Cloudflare benchmark](cloudflare-benchmark.md) | | Choose a harness and run a comparison | [Run a benchmark](#run-a-benchmark) | | Understand throughput, tail latency and errors | [Read the report](#read-the-report) | | Investigate an earlier optimization or release | [Detailed evidence](#detailed-design-and-recorded-evidence) | diff --git a/scripts/check-docs.py b/scripts/check-docs.py index d8e007e..3b00063 100644 --- a/scripts/check-docs.py +++ b/scripts/check-docs.py @@ -46,7 +46,7 @@ def anchors(text): def main(): paths = sorted([ROOT / 'README.md', ROOT / 'AGENTS.md', *ROOT.glob('docs/**/*.md'), - *ROOT.glob('crates/*/README.md')]) + *ROOT.glob('crates/*/README.md'), *ROOT.glob('deploy/**/*.md')]) errors, counts = [], Counter() for path in paths: relative = path.relative_to(ROOT) diff --git a/scripts/perf/bench.py b/scripts/perf/bench.py index dcf50bf..fee17c8 100644 --- a/scripts/perf/bench.py +++ b/scripts/perf/bench.py @@ -97,11 +97,14 @@ def prepare(control, output): manifest["ranges"].append({"ordinal": ordinal, "node": topology.owner(ordinal), "partition": partition.hex(), "scope": scope}) print(f"Warmed range {ordinal:02d} on {topology.node_names[topology.owner(ordinal)]}", flush=True) + save(output / "table-plan.json", manifest) control.sdk().create_table(TableName=TABLE, BillingMode="PAY_PER_REQUEST", AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}, {"AttributeName": "sk", "AttributeType": "N"}], KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}, {"AttributeName": "sk", "KeyType": "RANGE"}]) + manifest["creation_returned"] = True + save(output / "table-plan.json", manifest) clients = [control.sdk(i) for i in topology.indices] for row in manifest["ranges"]: _, state = control.query(row["node"], DATA_NS, bytes.fromhex(row["partition"]), 4, None) diff --git a/scripts/perf/cloudflare.py b/scripts/perf/cloudflare.py new file mode 100644 index 0000000..8f7b0d0 --- /dev/null +++ b/scripts/perf/cloudflare.py @@ -0,0 +1,336 @@ +#!/usr/bin/env python3 +"""Run the existing ordered-range workload against three Cloudflare Containers. + +Uses private deployment.json/secrets.json in an ignored benchmark directory. +Run agent.py --tunnels against secrets.json for the native private control path. +Cloudflare/R2 credentials come only from the calling environment. +""" +import argparse +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from datetime import datetime, timezone +import json +import os +from pathlib import Path +import time +import threading +import urllib.error +import urllib.request + +import boto3 +from botocore.config import Config +from botocore.exceptions import ReadTimeoutError, ClientError + +from bench import key, prepare, run_case, save, verify +from control import Control, ACCOUNT_NS, DATA_NS +from topology import Topology + + +class CloudControl(Control): + def __init__(self, root): + self.root = Path(root) + self.meta = json.loads((self.root/'deployment.json').read_text()) + self.credentials = json.loads((self.root/'secrets.json').read_text()) + self.topology = Topology(self.meta['nodes'], self.meta['cells_per_node']) + self.key = bytes.fromhex(self.credentials['peer_key']) + self.bucket, self.prefix = self.meta['bucket'], self.meta['prefix'] + self.s3 = boto3.client('s3', endpoint_url=os.environ['CF_S3_API_ENDPOINT'], region_name='auto', + aws_access_key_id=os.environ['CF_ACCESS_KEY_ID'], + aws_secret_access_key=os.environ['CF_SECRET_ACCESS_KEY'], + config=Config(retries={'total_max_attempts': 1}, request_checksum_calculation='when_required')) + + def sdk(self, index=0, pool=128): + client = boto3.client('dynamodb', endpoint_url=self.credentials['worker_url'], region_name='us-east-1', + aws_access_key_id=self.credentials['access_key'], + aws_secret_access_key=self.credentials['secret'], + config=Config(max_pool_connections=pool, connect_timeout=10, read_timeout=45, + retries={'total_max_attempts': 1}, user_agent_extra='BeyondDBBenchmark/0.1')) + def ingress(request, **kwargs): + request.headers['x-beyonddb-node'] = str(index+1) + client.meta.events.register('before-sign.dynamodb', ingress) + return client + + def admin(self, index, action, method='GET'): + request = urllib.request.Request(self.credentials['worker_url'] + f'/_admin/{index+1}/{action}', + headers={'Authorization': 'Bearer '+self.credentials['admin_token'], 'User-Agent': 'BeyondDBBenchmark/0.1'}, method=method) + with urllib.request.urlopen(request, timeout=60) as response: + body = response.read() + return body.decode() if action == 'logs' else json.loads(body) + + def resources(self): + result = {self.topology.node_names[i]: self.admin(i, 'status') for i in self.topology.indices} + account = os.environ['CF_ACCOUNT_ID'] + base = f'https://api.cloudflare.com/client/v4/accounts/{account}/containers/applications' + headers = {'Authorization': 'Bearer '+os.environ['CF_API_TOKEN']} + if 'application_id' not in self.meta: + with urllib.request.urlopen(urllib.request.Request(base, headers=headers), timeout=30) as response: + applications = json.load(response)['result'] + app = next(a for a in applications if a['name'] == self.meta['worker']+'-beyondnode') + self.meta['application_id'] = app['id'] + save(self.root/'deployment.json', self.meta) + api = f"{base}/{self.meta['application_id']}/instances" + request = urllib.request.Request(api, headers=headers) + with urllib.request.urlopen(request, timeout=30) as response: + instances = json.load(response)['result']['instances'] + for index, name in enumerate(self.topology.node_names): + value = result[name] + if not value['running'] or value['agent']['exit_code'] is not None: + raise RuntimeError(f'{name} is not running') + if min(value['agent']['agent_nofile_limit']) < 65536: + raise RuntimeError(f'{name}: nofile limit is below the qualified fixture setting') + instance = next(i for i in instances if i['name'] == f'node-{index+1}') + allocation = instance['configuration'] + if allocation != {'vcpu': 1, 'memory': 3072, 'disk': 5000}: + raise RuntimeError(f'{name}: unexpected Cloudflare instance allocation') + value['cloudflare_instance'] = instance + # Guest cpu_count/MemTotal can include platform scheduling and VM + # overhead; retain them without treating them as cgroup limits. + return result + + + + +def resume_creation(control, root): + plan = json.loads((root/'table-plan.json').read_text()) + if plan.get('creation_returned'): + raise RuntimeError('Seeding may already have started; do not replay ambiguous item writes') + if any(root.glob('seed-range-*.json')): + raise RuntimeError('A prior seed submission exists; retain its outcome instead of resubmitting') + deadline = time.monotonic()+180 + while True: + _, record = control.query(0, ACCOUNT_NS, bytes(4), 7, plan['table']) + if record and record['id'] != plan['table_id']: + raise RuntimeError('Catalog generation differs from the original creation plan') + if record: + status = control.sdk().describe_table(TableName=plan['table'])['Table']['TableStatus'] + if status == 'ACTIVE': break + if time.monotonic() >= deadline: + raise RuntimeError('Original table creation has not become ACTIVE; retain its ambiguous outcome') + time.sleep(3) + for row in plan['ranges']: + partition = bytes.fromhex(row['partition']) + _, state = control.query(row['node'],DATA_NS,partition,4,None) + if state is None or state['spec']['table']['id'] != plan['table_id']: + raise RuntimeError('Original range installation does not match the persisted plan') + row['installed_spec'] = state['spec'] + _, usage = control.query(row['node'],DATA_NS,partition,7,None) + if usage['item_count'] != 0: + raise RuntimeError('Fixture already contains items; do not replay ambiguous seed writes') + verify(control, plan) + clients = [control.sdk(i) for i in control.topology.indices] + stop = threading.Event() + def seed(ordinal): + row=plan['ranges'][ordinal]; attempted=[] + try: + for sample in range(4): + for sk in (1,2): + if stop.is_set():return + item_key=key(ordinal,sample,sk,control.topology) + attempted.append({'key':item_key,'outcome':'submitted'}) + save(root/f'seed-range-{ordinal}.json',attempted) + clients[row['node']].put_item(TableName=plan['table'],Item={**item_key,'payload':{'S':'x'*1024}}) + attempted[-1]['outcome']='acknowledged' + save(root/f'seed-range-{ordinal}.json',attempted) + except Exception: + stop.set();raise + order=[node*control.topology.cells_per_node+offset + for offset in range(control.topology.cells_per_node) for node in control.topology.indices] + with ThreadPoolExecutor(max_workers=6) as pool: + list(pool.map(seed,order)) + if stop.is_set(): raise RuntimeError('Seed submission failed; retain uncertain outcomes without retry') + plan['seed_items']=control.topology.cells*8 + plan['creation_recovered_after_timeout']=True + save(root/'table.json',plan) + save(root/'ownership-prepared.json',verify(control,plan)) + print(f"Recovered original creation; verified {control.topology.cells} ranges and seeded {plan['seed_items']} items",flush=True) + + +@contextmanager +def resource_samples(control, root, label): + path = root/f'resource-samples-{label}.jsonl' + if path.exists(): + raise RuntimeError('Resource samples already exist; use a fresh result label') + stop = threading.Event() + def collect(): + with path.open('x') as stream: + while not stop.is_set(): + row = {'observed_at': datetime.now(timezone.utc).isoformat(), 'nodes': {}} + for index, name in enumerate(control.topology.node_names): + try: row['nodes'][name] = control.admin(index, 'status') + except Exception as error: row['nodes'][name] = {'sampling_error': type(error).__name__} + stream.write(json.dumps(row)+'\n'); stream.flush() + stop.wait(5) + thread = threading.Thread(target=collect, daemon=True) + thread.start() + try: + yield + finally: + stop.set(); thread.join(timeout=15) + + +def measure(control, root, args): + destination = root/f'results-{args.label}.json' + if destination.exists(): + raise RuntimeError('Result label already exists; preserve previous measurements and select a fresh label') + manifest = json.loads((root/'table.json').read_text()) + result = {'platform': 'cloudflare-containers-r2', 'topology': control.topology.to_dict(), + 'source_revision': control.meta['source_revision'], 'model': 'closed_loop', + 'profile': 'release', 'payload_bytes': 1024, 'strong_reads': True, + 'sdk_retries': 0, 'client_read_timeout_s': 45, 'transaction_items': 2, + 'transaction_placement': 'different cells and nodes', 'warmup_s': args.warmup, + 'durability': 'R2 bucket publication; follower durability disabled', + 'client_location': 'external workstation; YVR edge observed; HTTPS Worker ingress; WAN included', + 'label': args.label, 'operation_order': args.operations, 'client_order': args.clients, + 'warmup_errors': 0, 'cases': [], + 'started_at': datetime.now(timezone.utc).isoformat()} + save(root/f'ownership-before-{args.label}.json', verify(control, manifest)) + clients = [control.sdk(i, max(args.clients)) for i in control.topology.indices] + run_id = __import__('uuid').uuid4().hex[:12] + for workers in args.clients: + for operation in args.operations: + phase = f'{args.label}-{operation}-{workers}' + if args.warmup: + warmup = run_case(clients, operation, workers, args.warmup, run_id, phase+'-warmup', control.topology) + save(root/f'warmup-{phase}.json', warmup) + result['warmup_errors'] += warmup['errors'] + save(destination, result) + before = control.resources() + case = run_case(clients, operation, workers, args.seconds, run_id, phase, control.topology) + result['cases'].append(case) + save(destination, result) + case['resources_before'] = before + case['resources_after'] = control.resources() + checked = 0 + for cell, item_key in case['write_witnesses'].items(): + response = clients[control.topology.owner(int(cell))].get_item(TableName=manifest['table'], Key=item_key, ConsistentRead=True) + if response.get('Item', {}).get('payload', {}).get('S') != 'x'*1024: + raise RuntimeError('Acknowledged write did not read back') + checked += 1 + case['write_witnesses_verified'] = checked + save(destination, result) + print(json.dumps({k: case[k] for k in ('operation','clients','completed','errors','elapsed_s','requests_per_s','p50_ms','p95_ms','p99_ms','covers_all_cells')}), flush=True) + save(root/f'ownership-after-{args.label}.json', verify(control, manifest)) + result['completed_at'] = datetime.now(timezone.utc).isoformat() + save(destination, result) + if result['warmup_errors'] or any(c['errors'] or not c['covers_all_cells'] for c in result['cases']): + raise SystemExit('Failed/incomplete windows retained; inspect results') + + +def ready(control, indices, timeout): + """Repeat only a pure query while a newly started node becomes ready.""" + for index in indices: + deadline = time.monotonic() + timeout + client = control.sdk(index) + while True: + try: + client.list_tables() + print(f'Node {index+1}: signed ListTables succeeded', flush=True) + break + except Exception: + if time.monotonic() >= deadline: + raise RuntimeError(f'Node {index+1} did not become ready') from None + time.sleep(3) + + +def check_writes(control, root, paths, label): + """Read original acknowledged keys; never resubmit their mutations.""" + destination = root/f'write-check-{label}.json' + if destination.exists(): + raise RuntimeError('Write-check label already exists; retain previous verification') + manifest = json.loads((root/'table.json').read_text()) + checks = [] + for path in paths: + for case in json.loads(path.read_text())['cases']: + for cell, item_key in case['write_witnesses'].items(): + checks.append({'source': path.name, 'phase': case['phase'], + 'cell': int(cell), 'key': item_key, 'kind': 'acknowledged-write'}) + if not checks: + raise RuntimeError('No acknowledged write witnesses in the supplied results') + for cell in range(control.topology.cells): + checks.append({'cell': cell, 'key': key(cell, topology=control.topology), 'kind': 'seed'}) + result = {'started_at': datetime.now(timezone.utc).isoformat(), + 'planned_checks': len(checks), 'checks': []} + save(destination, result) + clients = [control.sdk(i) for i in control.topology.indices] + def read(check): + try: + response = clients[check['cell'] % control.topology.nodes].get_item( + TableName=manifest['table'], Key=check['key'], ConsistentRead=True) + check['verified'] = response.get('Item', {}).get('payload', {}).get('S') == 'x'*1024 + if not check['verified']: + check['error'] = 'Acknowledged value missing or different' + except Exception as error: + check.update(verified=False, error_type=type(error).__name__) + return check + with ThreadPoolExecutor(max_workers=6) as pool: + for check in pool.map(read, checks): + result['checks'].append(check) + save(destination, result) + result['verified'] = sum(c['verified'] for c in result['checks']) + result['completed_at'] = datetime.now(timezone.utc).isoformat() + save(destination, result) + print(f"Verified {result['verified']} of {len(checks)} original values", flush=True) + if result['verified'] != len(checks): + raise SystemExit('Write recovery check failed; retained all observations') + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('action', choices=['start','ready','status','logs','stop','prepare','resume-creation','measure','verify','check-writes']) + parser.add_argument('--run-dir', required=True, type=Path) + parser.add_argument('--node', type=int, choices=[1,2,3]) + parser.add_argument('--bootstrap', action='store_true', help='first provisioning only; requires start --node 1') + parser.add_argument('--operations', nargs='+', choices=['get','query','put','transact_get','transact_write'], + default=['get','query','put','transact_get','transact_write']) + parser.add_argument('--clients', type=int, nargs='+', default=[1,8,30,60]) + parser.add_argument('--seconds', type=int, default=60) + parser.add_argument('--warmup', type=int, default=5) + parser.add_argument('--label', default='main') + parser.add_argument('--results', type=Path, nargs='+', help='original result files for check-writes') + parser.add_argument('--ready-timeout', type=int, default=180) + args = parser.parse_args() + if min(args.clients) < 1 or args.seconds < 1 or args.warmup < 0 or args.ready_timeout < 1: + parser.error('positive clients/seconds/ready-timeout and nonnegative warmup required') + if args.bootstrap and (args.action != 'start' or args.node != 1): + parser.error('--bootstrap requires start --node 1') + if args.action == 'check-writes' and not args.results: + parser.error('check-writes requires --results') + control = CloudControl(args.run_dir) + if args.action == 'prepare': + if (args.run_dir/'table.json').exists() or (args.run_dir/'table-plan.json').exists(): + parser.error('Retained creation plan exists; observe it with resume-creation instead of recreating') + try: + prepare(control, args.run_dir) + except (ReadTimeoutError, ClientError) as error: + if not (args.run_dir/'table-plan.json').exists():raise + plan=json.loads((args.run_dir/'table-plan.json').read_text()) + save(args.run_dir/'creation-failure.json',{'error_type':type(error).__name__, + 'client_timeout_s':45,'creation_returned':plan.get('creation_returned',False), + 'outcome':'uncertain; observe original generation without resubmitting CreateTable'}) + if plan.get('creation_returned'):raise + resume_creation(control,args.run_dir) + elif args.action == 'resume-creation': + resume_creation(control,args.run_dir) + elif args.action == 'measure': + with resource_samples(control,args.run_dir,args.label): + measure(control,args.run_dir,args) + elif args.action == 'verify': + print(json.dumps(verify(control,json.loads((args.run_dir/'table.json').read_text())))) + elif args.action == 'ready': + ready(control, [args.node-1] if args.node else control.topology.indices, args.ready_timeout) + elif args.action == 'check-writes': + check_writes(control,args.run_dir,args.results,args.label) + else: + indices = [args.node-1] if args.node else control.topology.indices + for index in indices: + action = args.action + ('?bootstrap=1' if args.bootstrap else '') + result = control.admin(index,action,'POST' if args.action in ('start','stop') else 'GET') + if args.action == 'logs': + path=args.run_dir/f'node{index+1}.log';path.write_text(result);print(f'Saved {path}') + else: + save(args.run_dir/f'{args.action}-node{index+1}.json',result) + print(json.dumps({'node':index+1,**result})) + + +if __name__ == '__main__': + main() diff --git a/scripts/perf/cloudflare_report.py b/scripts/perf/cloudflare_report.py new file mode 100644 index 0000000..d60493f --- /dev/null +++ b/scripts/perf/cloudflare_report.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python3 +"""Report all retained Cloudflare windows, including failures and partial runs.""" +import argparse +import csv +import json +from pathlib import Path + +from report import LABELS + + +def warmup_errors(source, run): + """Raw warmup files survive an interruption before the aggregate is saved.""" + paths = list(source.glob(f"warmup-{run['label']}-*.json")) + if paths: + return sum(json.loads(path.read_text())['errors'] for path in paths) + return run.get('warmup_errors', 0) + + +def main(): + parser=argparse.ArgumentParser(description=__doc__) + parser.add_argument('--input',nargs='+',type=Path,required=True) + parser.add_argument('--output',type=Path,required=True) + parser.add_argument('--summary-clients',type=int,default=8) + args=parser.parse_args() + meta=json.loads((args.input[0].parent/'deployment.json').read_text()) + runs=[json.loads(p.read_text()) for p in args.input] + cases=[] + for source,run in zip(args.input,runs): + for case in run['cases']: + warmup_path=source.parent/f"warmup-{case['phase']}.json" + errors=json.loads(warmup_path.read_text())['errors'] if warmup_path.exists() else None + cases.append(dict(case,label=run['label'],warmup_errors=errors)) + args.output.mkdir(parents=True,exist_ok=True) + columns=['label','operation','clients','completed','errors','elapsed_s','requests_per_s', + 'items_per_s','p50_ms','p95_ms','p99_ms','covers_all_cells','write_witnesses_verified','warmup_errors'] + with (args.output/'windows.csv').open('w') as stream: + writer=csv.DictWriter(stream,fieldnames=columns,extrasaction='ignore',lineterminator='\n');writer.writeheader();writer.writerows(cases) + warmups=[warmup_errors(p.parent,r) for p,r in zip(args.input,runs)] + warmup=sum(warmups) + failures=sum(c['errors'] for c in cases) + successes=sum(c['completed'] for c in cases) + complete=all('completed_at' in r for r in runs) + lines=['# BeyondDB on Cloudflare Containers and R2','', + f"Runtime source `{meta['source_revision']}`; binary SHA-256 `{meta['binary_sha256']}`.",'', + '**Three database nodes: 1 vCPU, 3072 MiB memory, 5000 MB ephemeral disk each.** ' + 'One measured table spans 30 ordered ranges, ten per node. Metadata/coordinator Cells ' + 'and the peer bridge share those node budgets. Native balancing is paused for controlled placement.','', + 'Durability: direct R2 S3 bucket publication; follower durability disabled. ' + 'Uniform 1-KiB payloads, strong GetItem/Query, two-item cross-node transactions and disabled SDK retries. ' + 'External workstation traffic includes WAN, HTTPS Worker ingress and private WSS peer transport.','', + f"**{len(cases)} retained windows: {successes:,} successful calls, {failures:,} measured errors and {warmup:,} warmup errors.** " + f"All requested runs reached final inventory verification: **{complete}**.",'', + 'Earlier setup failures and their empty generation are retained in the local run record. ' + 'The empty failed table was retired through the normal API before the measured generation. ' + 'Disk, metadata and resource observations include retained earlier roots.','', + f'## At {args.summary_clients} clients','', + '| API | Requests/s | p50 ms | p95 ms | p99 ms | Successes | Errors | Coverage |', + '| --- | ---: | ---: | ---: | ---: | ---: | ---: | --- |'] + def row(c,include_label=False): + values=[LABELS[c['operation']],c['requests_per_s'],c['p50_ms'],c['p95_ms'],c['p99_ms'], + c['completed'],c['errors'],'all 30' if c['covers_all_cells'] else 'incomplete'] + return '| '+' | '.join('—' if v is None else str(v) for v in values)+' |' + for op in LABELS: + for c in cases: + if c['clients']==args.summary_clients and c['operation']==op:lines.append(row(c)) + lines+=['','![Successful throughput and successful-call p99; failed windows remain in the tables](throughput-latency.svg)','', + 'Percentiles use nearest rank over successful calls only. Fewer than 1,000 successes ' + 'give a weak p99 estimate. Failed calls retain their own latencies in raw JSON. ' + 'Zero or missing successful samples do not establish a latency distribution.','', + 'Throughput divides successful calls by actual elapsed time, including completion drain. ' + 'Query and transaction requests each contain two items; Get/Put contain one. ' + 'Workers stop after three errors. Failed/incomplete windows are not healthy capacity.','', + '## Every retained window',''] + for r,w in zip(runs,warmups): + lines += [f"### {r['label']}",'', + '| API / clients | Requests/s | p50 ms | p95 ms | p99 ms | Successes | Errors | Coverage |', + '| --- | ---: | ---: | ---: | ---: | ---: | ---: | --- |'] + for c in r['cases']: + lines.append(row(c).replace(LABELS[c['operation']],f"{LABELS[c['operation']]} / {c['clients']}",1)) + lines += ['',f"Separate warmup errors: **{w}**. Completion recorded: **{'completed_at' in r}**.",''] + samples=[] + for r,p in zip(runs,args.input): + path=p.parent/f"resource-samples-{r['label']}.jsonl" + if path.exists():samples += [json.loads(line) for line in path.read_text().splitlines() if line.strip()] + lines+=['## Resource observations','', + '| Node | Location(s) | Peak sampled process RSS MiB | Maximum sampled disk use GB | Peak sampled open FDs |', + '| --- | --- | ---: | ---: | ---: |'] + for i in range(1,4): + values=[s['nodes'][f'perf30-node{i}']['agent'] for s in samples + if s['nodes'].get(f'perf30-node{i}',{}).get('agent')] + rss=[] + for v in values: + for line in (v.get('process_status') or '').splitlines(): + if line.startswith('VmRSS:'):rss.append(int(line.split()[1])/1024) + locations=sorted({v['platform']['CLOUDFLARE_LOCATION'] for v in values}) + disk=max((v['disk']['used']/1e9 for v in values),default=0) + fds=max((v.get('process_open_fds') or 0 for v in values),default=0) + lines.append(f"| {i} | {', '.join(locations) or 'unavailable'} | {max(rss,default=0):.1f} | {disk:.3f} | {fds} |") + lines+=['','These are sampled observations, not hard peaks. Process RSS excludes other guest memory. ' + 'Cloudflare instance configuration is the allocation proof; guest CPU count/MemTotal ' + 'can include platform scheduling and VM overhead. Full guest/process/cgroup observations ' + 'and sampling errors remain in the raw artifacts.','', + 'Closed-loop characterization only: no coordinated-omission correction, arrival-rate SLO ' + 'or peak fleet-capacity claim. Recovery and broader production qualification require separate evidence.',''] + (args.output/'report.md').write_text('\n'.join(lines)) + combined=dict(runs[0],cases=cases,warmup_errors=warmup, + runs=[{'label':r['label'],'completed': 'completed_at' in r} for r in runs]) + (args.output/'combined-results.json').write_text(json.dumps(combined,indent=2)+'\n') + print(f'Reported {len(cases)} windows; {failures} measured errors; {warmup} warmup errors; complete={complete}') + + +if __name__=='__main__':main() diff --git a/scripts/perf/plot.py b/scripts/perf/plot.py index be6f40e..3b35166 100644 --- a/scripts/perf/plot.py +++ b/scripts/perf/plot.py @@ -33,7 +33,7 @@ def main(): good = [c for c in rows if c[metric] is not None and c[metric] > 0] ax.plot([c["clients"] for c in good], [c[metric] for c in good], marker="o", color=color, linewidth=2, label=label) - bad = [c for c in good if c["errors"] or not covers_all(c)] + bad = [c for c in good if c["errors"] or c.get("warmup_errors") or not covers_all(c)] if bad: ax.scatter([c["clients"] for c in bad], [c[metric] for c in bad], color="red", marker="x", zorder=5, s=90) @@ -46,14 +46,17 @@ def main(): axes[0].set_ylabel("Requests / second (log scale)") axes[1].set_title("p99 response latency") axes[1].set_ylabel("Milliseconds (log scale)") - fig.suptitle(f"BeyondDB · {topology.nodes} × 1 vCPU / 3 GiB / 5 GiB data · {topology.cells} data cells / one table", fontsize=13, fontweight="bold") + cloud = result.get("platform") == "cloudflare-containers-r2" + disk = "5 GB ephemeral disk" if cloud else "5 GiB data" + fig.suptitle(f"BeyondDB · {topology.nodes} × 1 vCPU / 3 GiB / {disk} · {topology.cells} data cells / one table", fontsize=13, fontweight="bold") handles, labels = axes[0].get_legend_handles_labels() - if any(c["errors"] or not covers_all(c) for c in cases): + if any(c["errors"] or c.get("warmup_errors") or not covers_all(c) for c in cases): handles.append(Line2D([], [], color="red", marker="x", linestyle="none")) labels.append("Errors or incomplete coverage") fig.legend(handles, labels, loc="lower center", ncol=3, bbox_to_anchor=(0.5, 0.06), frameon=False) seconds = ",".join(str(s) for s in sorted({c["requested_s"] for c in cases})) - fig.text(0.5, 0.02, f"1 KiB payload · signed SDK · {seconds} s closed-loop windows · bucket durability · local shared VM", ha="center", color="#555555", fontsize=9) + environment = "Cloudflare Containers + R2 · external client / WAN included" if cloud else "bucket durability · local shared VM" + fig.text(0.5, 0.02, f"1 KiB payload · signed SDK · {seconds} s closed-loop windows · {environment}", ha="center", color="#555555", fontsize=9) fig.subplots_adjust(top=0.82, bottom=0.25, left=0.07, right=0.98, wspace=0.26) args.output.mkdir(parents=True, exist_ok=True) fig.savefig(args.output / "throughput-latency.png", dpi=180, facecolor="white") diff --git a/scripts/perf/requirements-cloudflare.txt b/scripts/perf/requirements-cloudflare.txt new file mode 100644 index 0000000..7c78427 --- /dev/null +++ b/scripts/perf/requirements-cloudflare.txt @@ -0,0 +1,2 @@ +-r requirements.txt +websockets==15.0.1 diff --git a/scripts/perf/test_cloudflare.py b/scripts/perf/test_cloudflare.py new file mode 100644 index 0000000..f1f491e --- /dev/null +++ b/scripts/perf/test_cloudflare.py @@ -0,0 +1,102 @@ +"""Offline byte-preservation check for the Cloudflare deployment peer bridge.""" +import asyncio +import importlib.util +import json +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +import cloudflare_report +from cloudflare import check_writes +from topology import Topology + + +class CloudflareRecovery(unittest.TestCase): + def test_failed_readback_is_retained_without_resubmission(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root/'table.json').write_text('{"table":"Offline"}') + source = root/'results.json' + source.write_text(json.dumps({'cases': [{ + 'phase': 'offline-put', 'write_witnesses': {'0': { + 'pk': {'S': 'original-key'}, 'sk': {'N': '1'}}}}]})) + class Client: + def __init__(self): self.calls = [] + def get_item(self, **request): + self.calls.append(request) + raise TimeoutError('offline fault') + client = Client() + control = SimpleNamespace(topology=Topology(3, 1), sdk=lambda i: client) + with self.assertRaises(SystemExit): + check_writes(control, root, [source], 'offline') + result = json.loads((root/'write-check-offline.json').read_text()) + self.assertEqual(result['verified'], 0) + self.assertEqual(len(result['checks']), 4) + self.assertEqual(len(client.calls), 4) + self.assertTrue(all(r['ConsistentRead'] for r in client.calls)) + self.assertEqual(client.calls[0]['Key']['pk']['S'], 'original-key') + self.assertTrue(all(c['error_type'] == 'TimeoutError' for c in result['checks'])) + + +class CloudflareReports(unittest.TestCase): + def test_interrupted_window_and_unsaved_warmup_errors_remain_visible(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root/'deployment.json').write_text(json.dumps({ + 'source_revision': 'offline-revision', 'binary_sha256': 'offline-digest'})) + # The warmup completed, but the process died before saving its total. + (root/'warmup-pressure-transact_get-60.json').write_text('{"errors": 2}') + result = {'label': 'pressure', 'warmup_errors': 0, 'cases': [{ + 'operation': 'transact_get', 'phase': 'pressure-transact_get-60', 'clients': 60, 'completed': 0, + 'errors': 3, 'elapsed_s': 45, 'requests_per_s': 0, 'items_per_s': 0, + 'p50_ms': None, 'p95_ms': None, 'p99_ms': None, 'covers_all_cells': False}]} + source = root/'results-pressure.json' + source.write_text(json.dumps(result)) + output = root/'report' + with patch('sys.argv', ['cloudflare_report.py', '--input', str(source), + '--output', str(output)]): + cloudflare_report.main() + combined = json.loads((output/'combined-results.json').read_text()) + self.assertEqual(combined['warmup_errors'], 2) + self.assertEqual(combined['cases'][0]['errors'], 3) + self.assertFalse(combined['runs'][0]['completed']) + self.assertIn('0 successful calls, 3 measured errors and 2 warmup errors', + (output/'report.md').read_text()) + self.assertIn('pressure,transact_get,60,0,3,', (output/'windows.csv').read_text()) + +try: + from websockets.asyncio.server import serve +except ImportError: + serve = None + + +@unittest.skipIf(serve is None, 'install requirements-cloudflare.txt for bridge checks') +class CloudflareBridge(unittest.IsolatedAsyncioTestCase): + async def test_preserves_native_headers_and_large_binary_stream(self): + source = Path(__file__).resolve().parents[2]/'deploy/cloudflare/agent.py' + spec = importlib.util.spec_from_file_location('beyonddb_cloud_agent', source) + agent = importlib.util.module_from_spec(spec) + spec.loader.exec_module(agent) + observed = [] + async def echo(ws): + observed.append((ws.request.path, ws.request.headers.get('Authorization'))) + async for payload in ws: + await ws.send(payload) + async with serve(echo, '127.0.0.1', 0, compression=None) as remote: + settings = {'worker_url': f'ws://127.0.0.1:{remote.sockets[0].getsockname()[1]}', + 'tunnel_token': 'offline-token'} + proxy = await asyncio.start_server(lambda r,w: agent.peer_forward(r,w,2,settings), '127.0.0.1', 0) + try: + reader, writer = await asyncio.open_connection('127.0.0.1', proxy.sockets[0].getsockname()[1]) + payload = (b'POST /peer/tunnel HTTP/1.1\r\nUpgrade: cellar-tunnel\r\n\r\n' + + bytes(range(256))*1024) + writer.write(payload[:7]); await writer.drain() + writer.write(payload[7:]); await writer.drain() + received = await asyncio.wait_for(reader.readexactly(len(payload)), 10) + self.assertEqual(received, payload) + self.assertEqual(observed, [('/_peer/2', 'Bearer offline-token')]) + writer.close(); await writer.wait_closed() + finally: + proxy.close(); await proxy.wait_closed()