From 3df4f66829171e3b9adb24e3ac31dc3639482341 Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Date: Sun, 18 Oct 2026 06:17:56 +0530 Subject: [PATCH 1/7] perf(bench): isolate gateway load processes Measure scheduled arrivals outside the gateway CPU budget. Add opt-in receipt trace regions without changing commit acknowledgment. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> --- Makefile | 9 +- bench/PROC.md | 85 +++++++++++++++ bench/loadgen/main.go | 204 ++++++++++++++++++++++++++++++++++++ bench/loadgen/main_test.go | 26 +++++ bench/proc.py | 97 +++++++++++++++++ bench/procgateway/main.go | 209 +++++++++++++++++++++++++++++++++++++ internal/receipt/ledger.go | 23 +++- 7 files changed, 650 insertions(+), 3 deletions(-) create mode 100644 bench/PROC.md create mode 100644 bench/loadgen/main.go create mode 100644 bench/loadgen/main_test.go create mode 100644 bench/proc.py create mode 100644 bench/procgateway/main.go diff --git a/Makefile b/Makefile index 56551e7..e5a3969 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: build build-verify run test docker lint clean bench-e2e bench-e2e-open-loop +.PHONY: build build-verify run test docker lint clean bench-e2e bench-e2e-open-loop bench-proc build: CGO_ENABLED=0 go build -o bin/agentgate ./cmd/agentgw @@ -14,6 +14,13 @@ test: BENCHTIME ?= 20000x BENCH_FILTER ?= ^BenchmarkHandleActE2E$$ +BENCH_PROC_ARGS ?= + +bench-proc: + mkdir -p .cache/tmp .cache/go-tmp .cache/go-build .cache/proc/bin + TMPDIR="$(CURDIR)/.cache/tmp" GOTMPDIR="$(CURDIR)/.cache/go-tmp" \ + GOCACHE="$(CURDIR)/.cache/go-build" go build -o .cache/proc/bin/ ./bench/procgateway ./bench/mockupstream ./bench/loadgen + python3 bench/proc.py $(BENCH_PROC_ARGS) bench-e2e-open-loop: BENCH_FILTER = ^BenchmarkHandleActOpenLoop$$ diff --git a/bench/PROC.md b/bench/PROC.md new file mode 100644 index 0000000..f3ea431 --- /dev/null +++ b/bench/PROC.md @@ -0,0 +1,85 @@ +# Separate-process latency benchmark + +Run `make bench-proc` from the repository root. No Docker or extra Go dependencies are needed. +Python 3 supervises 3 compiled Go processes: + +- `bench/mockupstream` listens on loopback port `19090`, with `GOMAXPROCS=2`. +- `bench/procgateway` listens on loopback port `18080`, with `GOMAXPROCS=4`. +- `bench/loadgen` sends requests with its own `GOMAXPROCS=2`. + +These processes share host CPUs. `GOMAXPROCS` isn't CPU pinning or a 4-vCPU quota. +Default process priority remains unchanged. No privileged priority change is needed. +Stop unrelated CPU and disk workloads before measuring. +Run on a dedicated Linux 4-vCPU gateway host before making deployment claims. + +## Stack and timing + +The thin gateway main wires the real gateway, authentication, encrypted vault, and signing stores. +It uses real SQLite files and migrations, plus the default receipt ledger and upstream pool. +All rate checks remain active, with high benchmark limits. +Requests use a scoped HMAC key and an unexpired cached bearer token. +No delegation token is presented. Production rejects invalid tokens through its unchanged verifier path. +Info logs still undergo JSON encoding, but writes go to `io.Discard`. +Terminal and remote log-sink delays aren't measured. + +Each case gets a fresh database and 256 warmup requests. +The default matrix measures 20,000 requests per case. +It includes fixed arrivals at 1,000, 2,500, 5,000, and 10,000 requests/s. +It also includes closed-loop concurrency 1 and 16. +Both modes run with 0 ms and 5 ms mock delays. + +Open-loop latency starts at each request's original scheduled arrival. +Late arrivals never reset the schedule or drop requests. +Dispatch lag and client connection waits stay in latency. +The client uses at most 1,024 connections. Pool waits remain measured. +Closed-loop timing starts immediately before `Client.Do`. +Both timers end after reading and closing the entire response body. + +The mock reports its measured handler duration, including sleep overshoot, inside `body.upstream_ns`. +Response encoding and writing stay in overhead. + +```text +overhead = response_completed - scheduled_arrival - upstream_ns +``` + +Subtraction occurs per request, before nearest-rank percentiles. +All requests must succeed. The ledger head must advance by exactly the measured count. +The client records request samples, connection reuse, and gateway timing headers. +Those headers exist only in this loopback benchmark main, not the public gateway. +`handler_ns` ends before response encoding. `receipt_ns` includes ledger queueing and committed acknowledgment. +Receipt time and auth time are nested inside handler time. Don't add overlapping spans. +These aren't isolated handler CPU costs. Both loopback paths and scheduler delays remain included. + +## Reproduction + +```sh +make bench-proc +make bench-proc BENCH_PROC_ARGS='--label repeat' +make bench-proc BENCH_PROC_ARGS='--label smoke --n 64' +make bench-proc BENCH_PROC_ARGS='--label profile --profile --rates 5000 --closed "" --delays 0' +``` + +Profiles change timing. Don't use profiled requests for the latency gate. +Each run prints its unique directory under ignored `.cache/proc/`. +Every case stores its CSV samples, summary JSON, private database, and gateway startup pragmas. +Credential files use mode `0600` and disappear when the gateway stops. +The supervisor stops all children, including after failures. +Keep databases and profiles private. Don't commit `.cache/`. + +Profiling records CPU, allocation, block, mutex, and execution trace files. +It also enables `GODEBUG=gctrace=1` in the gateway process. +Receipt trace regions distinguish head reads, signing, inserts, and SQLite COMMIT. +Request task logs identify individual load-generator sample indices. + +```sh +go tool pprof -top .cache/proc/bin/procgateway PATH/gateway.cpu +go tool pprof -top .cache/proc/bin/procgateway PATH/gateway.block +go tool pprof -top .cache/proc/bin/procgateway PATH/gateway.mutex +go tool pprof -top -alloc_space .cache/proc/bin/procgateway PATH/gateway.allocs +go tool trace -d=parsed PATH/gateway.trace +``` + +Set `TMPDIR`, `GOTMPDIR`, and `GOCACHE` under `.cache/` before invoking Go tools directly. +The Makefile already does this for builds. +`--batch-delay` tests existing collection waits without changing production defaults. +Only compare identical unprofiled matrices before and after each code change. diff --git a/bench/loadgen/main.go b/bench/loadgen/main.go new file mode 100644 index 0000000..4224581 --- /dev/null +++ b/bench/loadgen/main.go @@ -0,0 +1,204 @@ +// Command loadgen measures scheduled latency without coordinated omission. +package main + +import ( + "encoding/csv" + "encoding/json" + "flag" + "fmt" + "io" + "log" + "math" + "net/http" + "net/http/httptrace" + "os" + "runtime" + "slices" + "strconv" + "strings" + "sync" + "time" +) + +type sample struct { + Total, Upstream, Dispatch, Auth, Receipt, Handler int64 + Reused bool + Error string +} + +func percentile(values []int64, p float64) float64 { + slices.Sort(values) + return float64(values[max(0, int(math.Ceil(float64(len(values))*p))-1)]) / 1e6 +} + +func waitUntil(deadline time.Time) { + for { + left := time.Until(deadline) + if left <= 0 { + return + } + if left > 2*time.Millisecond { + time.Sleep(left - time.Millisecond) + } else { + runtime.Gosched() + } + } +} + +func main() { + url := flag.String("url", "http://127.0.0.1:18080/v1/act", "gateway URL") + credentials := flag.String("credentials", ".cache/proc/key.json", "credential JSON file") + count := flag.Int("n", 20000, "measured requests") + rate := flag.Int("rate", 1000, "fixed arrivals per second; 0 selects closed loop") + concurrency := flag.Int("c", 16, "closed-loop concurrency") + output := flag.String("output", ".cache/proc/samples.csv", "per-request CSV path") + flag.Parse() + if *count <= 0 || *rate < 0 || *concurrency <= 0 { + log.Fatal("n and c must be positive; rate must be non-negative") + } + data, err := os.ReadFile(*credentials) + if err != nil { + log.Fatal(err) + } + var secret struct{ Key string } + if err := json.Unmarshal(data, &secret); err != nil || secret.Key == "" { + log.Fatal("invalid credential file") + } + tr := &http.Transport{MaxIdleConns: 1024, MaxIdleConnsPerHost: 1024, MaxConnsPerHost: 1024} + defer tr.CloseIdleConnections() + client := &http.Client{Transport: tr, Timeout: 15 * time.Second} + act := func(scheduled time.Time, id int) sample { + const body = `{"service":"mock","action":"echo","on_behalf_of":"bench-user","params":{"worker":0}}` + req, _ := http.NewRequest(http.MethodPost, *url, strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+secret.Key) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Bench-ID", strconv.Itoa(id)) + result := sample{} + req = req.WithContext(httptrace.WithClientTrace(req.Context(), &httptrace.ClientTrace{ + GotConn: func(info httptrace.GotConnInfo) { result.Reused = info.Reused }, + })) + start := time.Now() + if scheduled.IsZero() { + scheduled = start + } + result.Dispatch = start.Sub(scheduled).Nanoseconds() + resp, err := client.Do(req) + if err != nil { + result.Error = err.Error() + return result + } + data, err := io.ReadAll(resp.Body) + resp.Body.Close() + result.Total = time.Since(scheduled).Nanoseconds() + if err != nil { + result.Error = err.Error() + return result + } + var envelope struct { + Status int + Body struct { + OK bool `json:"ok"` + UpstreamNS int64 `json:"upstream_ns"` + } + } + if err := json.Unmarshal(data, &envelope); err != nil || resp.StatusCode != 200 || envelope.Status != 200 || !envelope.Body.OK { + result.Error = fmt.Sprintf("status %d: %s", resp.StatusCode, data) + return result + } + result.Upstream = envelope.Body.UpstreamNS + result.Auth, _ = strconv.ParseInt(resp.Header.Get("X-Bench-Auth-Ns"), 10, 64) + result.Receipt, _ = strconv.ParseInt(resp.Header.Get("X-Bench-Receipt-Ns"), 10, 64) + result.Handler, _ = strconv.ParseInt(resp.Header.Get("X-Bench-Handler-Ns"), 10, 64) + if result.Upstream < 0 || result.Upstream > result.Total { + result.Error = "invalid upstream duration" + } + return result + } + head := func() uint64 { + base := strings.TrimSuffix(*url, "/v1/act") + resp, err := client.Get(base + "/bench/head") + if err != nil { + log.Fatal(err) + } + defer resp.Body.Close() + var body struct{ Head uint64 } + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil || resp.StatusCode != 200 { + log.Fatal("cannot read ledger head") + } + return body.Head + } + for range 256 { + if warmup := act(time.Time{}, -1); warmup.Error != "" { + log.Fatal(warmup.Error) + } + } + before := head() + samples := make([]sample, *count) + start := time.Now().Add(10 * time.Millisecond) + var done sync.WaitGroup + if *rate > 0 { + for i := range samples { + scheduled := start.Add(time.Duration(int64(i) * int64(time.Second) / int64(*rate))) + waitUntil(scheduled) + done.Add(1) + go func() { + defer done.Done() + samples[i] = act(scheduled, i) + }() + } + } else { + waitUntil(start) + for worker := range min(*concurrency, *count) { + done.Add(1) + go func() { + defer done.Done() + for i := worker; i < *count; i += *concurrency { + samples[i] = act(time.Time{}, i) + } + }() + } + } + done.Wait() + elapsed := time.Since(start) + after := head() + file, err := os.Create(*output) + if err != nil { + log.Fatal(err) + } + writer := csv.NewWriter(file) + writer.Write([]string{"total_ns", "upstream_ns", "dispatch_ns", "auth_ns", "receipt_ns", "handler_ns", "reused", "error"}) + overhead, dispatch := make([]int64, *count), make([]int64, *count) + errors, reused := 0, 0 + for i, s := range samples { + writer.Write([]string{ + strconv.FormatInt(s.Total, 10), strconv.FormatInt(s.Upstream, 10), strconv.FormatInt(s.Dispatch, 10), + strconv.FormatInt(s.Auth, 10), strconv.FormatInt(s.Receipt, 10), strconv.FormatInt(s.Handler, 10), + strconv.FormatBool(s.Reused), s.Error, + }) + overhead[i], dispatch[i] = s.Total-s.Upstream, s.Dispatch + if s.Error != "" { + errors++ + } + if s.Reused { + reused++ + } + } + writer.Flush() + if err := writer.Error(); err != nil { + log.Fatal(err) + } + if err := file.Close(); err != nil { + log.Fatal(err) + } + summary := map[string]any{ + "rate": *rate, "c": *concurrency, "n": *count, "errors": errors, "receipts": after - before, + "throughput": float64(*count) / elapsed.Seconds(), "reuse_pct": 100 * float64(reused) / float64(*count), + "p50_ms": percentile(slices.Clone(overhead), .50), "p90_ms": percentile(slices.Clone(overhead), .90), + "p99_ms": percentile(slices.Clone(overhead), .99), "p999_ms": percentile(overhead, .999), + "dispatch_p99_ms": percentile(dispatch, .99), + } + json.NewEncoder(os.Stdout).Encode(summary) + if errors != 0 || after-before != uint64(*count) { + os.Exit(1) + } +} diff --git a/bench/loadgen/main_test.go b/bench/loadgen/main_test.go new file mode 100644 index 0000000..e799a24 --- /dev/null +++ b/bench/loadgen/main_test.go @@ -0,0 +1,26 @@ +package main + +import ( + "testing" + "time" +) + +func TestPercentileNearestRank(t *testing.T) { + values := []int64{4e6, 1e6, 3e6, 2e6} + for _, test := range []struct { + p float64 + want float64 + }{{.5, 2}, {.9, 4}, {.99, 4}} { + if got := percentile(values, test.p); got != test.want { + t.Fatalf("percentile(%v) = %v, want %v", test.p, got, test.want) + } + } +} + +func TestWaitUntilPreservesLateSchedule(t *testing.T) { + scheduled := time.Now().Add(-time.Second) + waitUntil(scheduled) + if time.Since(scheduled) < time.Second { + t.Fatal("late arrivals must retain the original schedule") + } +} diff --git a/bench/proc.py b/bench/proc.py new file mode 100644 index 0000000..225a8af --- /dev/null +++ b/bench/proc.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Run the mock, real-stack gateway, and load generator in separate processes.""" +import argparse +import json +import os +from pathlib import Path +import secrets +import signal +import subprocess +import time +import urllib.request + + +def ready(url, process): + for _ in range(200): + if process.poll() is not None: + raise RuntimeError(f"server exited: {process.returncode}") + try: + with urllib.request.urlopen(url, timeout=1) as response: + if response.status == 200: + return + except OSError: + time.sleep(0.05) + raise RuntimeError(f"server not ready: {url}") + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--n", type=int, default=20000) + parser.add_argument("--label", default="baseline") + parser.add_argument("--profile", action="store_true") + parser.add_argument("--rates", default="1000,2500,5000,10000") + parser.add_argument("--closed", default="1,16") + parser.add_argument("--delays", default="0,5ms") + parser.add_argument("--batch-delay", default="0") + args = parser.parse_args() + root = Path(".cache/proc") / f"{args.label}-{secrets.token_hex(4)}" + root.mkdir(parents=True) + results = [] + for delay in args.delays.split(","): + cases = [("rate", int(r)) for r in args.rates.split(",") if r] + cases += [("c", int(c)) for c in args.closed.split(",") if c] + for mode, value in cases: + name = f"delay{delay}-{mode}{value}" + directory = root / name + directory.mkdir() + processes = [] + files = [] + try: + for binary, flags, procs in [ + ("mockupstream", ["-addr", "127.0.0.1:19090", "-delay", delay], "2"), + ("procgateway", [ + "-db", str(directory / "agentgate.db"), + "-credentials", str(directory / "key.json"), + "-batch-delay", args.batch_delay, + ] + (["-profile", str(directory / "gateway")] if args.profile else []), "4"), + ]: + output = open(directory / f"{binary}.log", "w") + files.append(output) + env = dict(os.environ, GOMAXPROCS=procs) + if args.profile and binary == "procgateway": + env["GODEBUG"] = "gctrace=1" + process = subprocess.Popen([f".cache/proc/bin/{binary}", *flags], env=env, + stdout=output, stderr=output) + processes.append(process) + ready("http://127.0.0.1:" + ("19090" if binary == "mockupstream" else "18080") + "/healthz", process) + command = [".cache/proc/bin/loadgen", "-n", str(args.n), + "-credentials", str(directory / "key.json"), + "-output", str(directory / "samples.csv"), + "-rate", str(value if mode == "rate" else 0), "-c", str(value if mode == "c" else 16)] + completed = subprocess.run(command, env=dict(os.environ, GOMAXPROCS="2"), + text=True, capture_output=True) + (directory / "loadgen.log").write_text(completed.stdout + completed.stderr) + if completed.returncode: + print(completed.stdout + completed.stderr, flush=True) + raise RuntimeError(f"load generator failed: {directory}") + result = dict(json.loads(completed.stdout), delay=delay, name=name) + results.append(result) + print(json.dumps(result), flush=True) + (directory / "summary.json").write_text(json.dumps(result, indent=2) + "\n") + finally: + for process in reversed(processes): + if process.poll() is None: + process.send_signal(signal.SIGTERM) + try: + process.wait(timeout=30) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + for file in files: + file.close() + (root / "results.json").write_text(json.dumps(results, indent=2) + "\n") + print(f"Results: {root}", flush=True) + + +if __name__ == "__main__": + main() diff --git a/bench/procgateway/main.go b/bench/procgateway/main.go new file mode 100644 index 0000000..81810da --- /dev/null +++ b/bench/procgateway/main.go @@ -0,0 +1,209 @@ +// Command procgateway wires the real gateway stack for separate-process benchmarks. +// It binds loopback only. Never deploy this benchmark fixture. +package main + +import ( + "context" + "crypto/rand" + "encoding/json" + "flag" + "fmt" + "io" + "log" + "log/slog" + "net" + "net/http" + "os" + "os/signal" + "runtime" + "runtime/pprof" + "runtime/trace" + "strconv" + "syscall" + "time" + + "github.com/Clawdlinux/agentgate/internal/auth" + agentdb "github.com/Clawdlinux/agentgate/internal/db" + "github.com/Clawdlinux/agentgate/internal/gateway" + "github.com/Clawdlinux/agentgate/internal/ratelimit" + "github.com/Clawdlinux/agentgate/internal/receipt" + "github.com/Clawdlinux/agentgate/internal/registry" + "github.com/Clawdlinux/agentgate/internal/signer" + "github.com/Clawdlinux/agentgate/internal/vault" +) + +type timings struct { + auth, receipt time.Duration +} +type timingKey struct{} +type authorizer struct{ *auth.KeyStore } + +func (a authorizer) Validate(ctx context.Context, key string) (*auth.AgentKey, error) { + start := time.Now() + result, err := a.KeyStore.Validate(ctx, key) + ctx.Value(timingKey{}).(*timings).auth = time.Since(start) + return result, err +} + +type recorder struct{ *receipt.Ledger } + +func (r recorder) Append(ctx context.Context, draft receipt.Draft) (receipt.Receipt, error) { + start := time.Now() + region := trace.StartRegion(ctx, "receipt.append") + result, err := r.Ledger.Append(ctx, draft) + region.End() + ctx.Value(timingKey{}).(*timings).receipt = time.Since(start) + return result, err +} + +type responseWriter struct { + http.ResponseWriter + timing *timings + start time.Time +} + +func (w responseWriter) WriteHeader(status int) { + w.Header().Set("X-Bench-Auth-Ns", strconv.FormatInt(w.timing.auth.Nanoseconds(), 10)) + w.Header().Set("X-Bench-Receipt-Ns", strconv.FormatInt(w.timing.receipt.Nanoseconds(), 10)) + w.Header().Set("X-Bench-Handler-Ns", strconv.FormatInt(time.Since(w.start).Nanoseconds(), 10)) + w.ResponseWriter.WriteHeader(status) +} + +func must(err error) { + if err != nil { + log.Fatal(err) + } +} + +func main() { + addr := flag.String("addr", "127.0.0.1:18080", "loopback listen address") + upstream := flag.String("upstream", "http://127.0.0.1:19090", "mock upstream URL") + path := flag.String("db", ".cache/proc/agentgate.db", "fresh SQLite path") + credentials := flag.String("credentials", ".cache/proc/key.json", "private client credential file") + profile := flag.String("profile", "", "profile prefix; empty disables profiling") + batchDelay := flag.Duration("batch-delay", 0, "receipt collection delay") + flag.Parse() + host, _, err := net.SplitHostPort(*addr) + must(err) + if ip := net.ParseIP(host); ip == nil || !ip.IsLoopback() { + log.Fatal("benchmark server requires a loopback address") + } + database, err := agentdb.Open(*path) + must(err) + defer database.Close() + must(agentdb.RunMigrations(database)) + var journal string + var synchronous, busyTimeout, foreignKeys int + must(database.QueryRow("PRAGMA journal_mode").Scan(&journal)) + must(database.QueryRow("PRAGMA synchronous").Scan(&synchronous)) + must(database.QueryRow("PRAGMA busy_timeout").Scan(&busyTimeout)) + must(database.QueryRow("PRAGMA foreign_keys").Scan(&foreignKeys)) + log.Printf("SQLite journal=%s synchronous=%d busy_timeout=%d foreign_keys=%d GOMAXPROCS=%d", + journal, synchronous, busyTimeout, foreignKeys, runtime.GOMAXPROCS(0)) + + master := make([]byte, 32) + _, err = rand.Read(master) + must(err) + defer clear(master) + pepper, err := auth.DeriveKeyPepper(master) + must(err) + keys, err := auth.NewKeyStore(database, pepper) + must(err) + _, key, err := keys.Create(context.Background(), "proc-bench", []string{"mock"}, []string{"bench-user"}) + must(err) + store, err := vault.NewSQLiteStore(database, master) + must(err) + cache, err := vault.NewCachedStore(store) + must(err) + token := make([]byte, 32) + _, err = rand.Read(token) + must(err) + must(cache.Put("bench-user", "mock", vault.Token{AccessToken: fmt.Sprintf("%x", token), TokenType: "Bearer"})) + signers, err := signer.NewStore(database, master) + must(err) + defer signers.Close() + _, private, err := signers.LoadOrCreateActive(1) + clear(private) + must(err) + ledger := receipt.NewLedger(database, signers, receipt.WithBatchDelay(*batchDelay)) + defer ledger.Close() + reg := registry.New() + must(reg.LoadBytes([]byte(fmt.Sprintf(`services: + mock: + base_url: %s + auth: + type: bearer + rate_limit: + requests_per_second: 1000000 + burst: 1000000 + actions: + echo: + method: POST + path: /echo + params: + worker: int +`, *upstream)))) + limiter := ratelimit.New(reg.RateLimits(), ratelimit.WithSettings(ratelimit.Settings{ + KeyRPS: 1e6, KeyBurst: 1e6, TenantRPS: 1e6, TenantBurst: 1e6, TenantMaxInflight: 1024, + })) + handler := gateway.New(gateway.Config{ + Registry: reg, Vault: cache, Authorizer: authorizer{keys}, Receipts: recorder{ledger}, Limiter: limiter, + Logger: slog.New(slog.NewJSONHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelInfo})), + }) + defer handler.CloseIdleConnections() + encoded, err := json.Marshal(map[string]string{"key": key}) + must(err) + must(os.WriteFile(*credentials, encoded, 0600)) + defer os.Remove(*credentials) + + if *profile != "" { + cpu, err := os.Create(*profile + ".cpu") + must(err) + defer cpu.Close() + must(pprof.StartCPUProfile(cpu)) + defer pprof.StopCPUProfile() + runtime.SetMutexProfileFraction(1) + runtime.SetBlockProfileRate(1) + traceFile, err := os.Create(*profile + ".trace") + must(err) + defer traceFile.Close() + must(trace.Start(traceFile)) + defer trace.Stop() + defer func() { + for _, name := range []string{"mutex", "block", "allocs", "heap"} { + file, err := os.Create(*profile + "." + name) + must(err) + must(pprof.Lookup(name).WriteTo(file, 0)) + must(file.Close()) + } + }() + } + + server := &http.Server{Addr: *addr, ReadHeaderTimeout: 5 * time.Second, Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/bench/head" { + head, _, err := ledger.Head(r.Context()) + if err != nil { + http.Error(w, err.Error(), 500) + return + } + json.NewEncoder(w).Encode(map[string]uint64{"head": head}) + return + } + timing := &timings{} + ctx, task := trace.NewTask(context.WithValue(r.Context(), timingKey{}, timing), "act") + defer task.End() + trace.Log(ctx, "request.id", r.Header.Get("X-Bench-ID")) + handler.ServeHTTP(responseWriter{w, timing, time.Now()}, r.WithContext(ctx)) + })} + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) + defer stop() + go func() { + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatal(err) + } + }() + <-ctx.Done() + shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + must(server.Shutdown(shutdown)) +} diff --git a/internal/receipt/ledger.go b/internal/receipt/ledger.go index 40c970f..0f9432a 100644 --- a/internal/receipt/ledger.go +++ b/internal/receipt/ledger.go @@ -13,6 +13,7 @@ import ( "encoding/json" "errors" "fmt" + "runtime/trace" "slices" "sync" "time" @@ -237,6 +238,9 @@ func (l *Ledger) flush(batch []appendRequest) { return err }) for i, request := range batch { + if trace.IsEnabled() { + trace.Log(request.ctx, "receipt.reply", "") + } if err != nil { request.reply <- appendResult{err: appendError(err)} } else { @@ -248,6 +252,11 @@ func (l *Ledger) flush(batch []appendRequest) { func (l *Ledger) commitBatch(batch []appendRequest, key signer.KeyRecord, priv ed25519.PrivateKey) ([]appendResult, error) { // Caller cancellation must not interrupt a shared transaction. ctx := context.Background() + region := trace.StartRegion(ctx, "receipt.transaction") + defer region.End() + if trace.IsEnabled() { + trace.Logf(ctx, "receipt.batch", "%d", len(batch)) + } conn, err := l.db.Conn(ctx) if err != nil { return nil, fmt.Errorf("acquire connection: %w", err) @@ -267,7 +276,9 @@ func (l *Ledger) commitBatch(batch []appendRequest, key signer.KeyRecord, priv e } }() + headRegion := trace.StartRegion(ctx, "receipt.head") head, prevHash, err := readHead(ctx, conn) + headRegion.End() if err != nil { return nil, fmt.Errorf("read head: %w", err) } @@ -293,18 +304,26 @@ func (l *Ledger) commitBatch(batch []appendRequest, key signer.KeyRecord, priv e } r := request.draft.receipt() r.Seq, r.PrevHash, r.TimestampUnixNS = head+1, prevHash, uint64(l.nowFn().UnixNano()) + signRegion := trace.StartRegion(ctx, "receipt.sign") r, err = receiptspec.Seal(r, storeSigner{kid: key.KID, priv: priv}) + signRegion.End() if err != nil { return nil, fmt.Errorf("sign receipt: %w", err) } - if err := insertReceipt(ctx, conn, r); err != nil { + insertRegion := trace.StartRegion(ctx, "receipt.insert") + err = insertReceipt(ctx, conn, r) + insertRegion.End() + if err != nil { return nil, fmt.Errorf("insert: %w", err) } results[i].receipt = r head, prevHash = r.Seq, r.EntryHash } - if _, err := conn.ExecContext(ctx, "COMMIT"); err != nil { + commitRegion := trace.StartRegion(ctx, "receipt.commit") + _, err = conn.ExecContext(ctx, "COMMIT") + commitRegion.End() + if err != nil { return nil, fmt.Errorf("commit: %w", err) } committed = true From 2d644f273be995c06b73fa506d5d88f4a02b9e23 Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Date: Sun, 18 Oct 2026 06:20:20 +0530 Subject: [PATCH 2/7] fix(bench): bound connection bursts and keep failures Keep transport queueing in scheduled latency. Limit client connections to avoid loopback accept bursts. Preserve failed-case evidence and correlate request tails with receipt trace phases. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> --- bench/PROC.md | 2 +- bench/analyze.py | 113 ++++++++++++++++++++++++++++++++++++++++++ bench/loadgen/main.go | 2 +- bench/proc.py | 7 ++- 4 files changed, 120 insertions(+), 4 deletions(-) create mode 100644 bench/analyze.py diff --git a/bench/PROC.md b/bench/PROC.md index f3ea431..0bb4a40 100644 --- a/bench/PROC.md +++ b/bench/PROC.md @@ -31,7 +31,7 @@ Both modes run with 0 ms and 5 ms mock delays. Open-loop latency starts at each request's original scheduled arrival. Late arrivals never reset the schedule or drop requests. Dispatch lag and client connection waits stay in latency. -The client uses at most 1,024 connections. Pool waits remain measured. +The client uses at most 128 connections. Pool waits remain measured. Closed-loop timing starts immediately before `Client.Do`. Both timers end after reading and closing the entire response body. diff --git a/bench/analyze.py b/bench/analyze.py new file mode 100644 index 0000000..9295503 --- /dev/null +++ b/bench/analyze.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""Summarize correlated tail samples and optional Go parsed trace regions.""" +import argparse +import collections +import csv +import json +import math +import re + + +def quantiles(values): + values = sorted(values) + return {name: values[math.ceil(len(values) * fraction) - 1] / 1e6 + for name, fraction in [("p50_ms", .5), ("p99_ms", .99), ("max_ms", 1)]} + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("samples") + parser.add_argument("--trace", help="go tool trace -d=parsed output") + args = parser.parse_args() + with open(args.samples) as file: + rows = [{key: int(value) for key, value in row.items() if key.endswith("_ns")} + for row in csv.DictReader(file)] + ranked = sorted(range(len(rows)), key=lambda i: rows[i]["total_ns"] - rows[i]["upstream_ns"]) + if not rows: + raise ValueError("no samples") + tail = ranked[min(len(rows) - 1, math.ceil(.99 * len(rows))):] + totals = collections.defaultdict(int) + for index in tail: + row = rows[index] + totals["dispatch"] += row["dispatch_ns"] + totals["auth"] += row["auth_ns"] + totals["receipt"] += row["receipt_ns"] + totals["other_gateway"] += row["handler_ns"] - row["auth_ns"] - row["receipt_ns"] - row["upstream_ns"] + totals["transport_and_response"] += row["total_ns"] - row["dispatch_ns"] - row["handler_ns"] + result = { + "tail_count": len(tail), + "tail_mean_ms": sum(totals.values()) / len(tail) / 1e6, + "tail_components": {key: {"mean_ms": value / len(tail) / 1e6, + "share_pct": value / sum(totals.values()) * 100} + for key, value in totals.items()}, + } + if args.trace: + starts, active, latest, tasks, appends, replies = {}, {}, {}, {}, {}, {} + durations = collections.defaultdict(list) + pauses = [] + pause_start = None + with open(args.trace) as file: + for line in file: + region = re.search(r'G=(-?\d+) (RegionBegin|RegionEnd) Time=(\d+) Task=(\d+) Type="([^"]+)"', line) + if region: + goroutine, event, timestamp, task, name = region.groups() + timestamp = int(timestamp) + key = (goroutine, task, name) + if event == "RegionBegin": + starts[key] = timestamp + if name == "receipt.transaction": + active[goroutine] = {"start": timestamp, "phases": collections.defaultdict(int)} + elif key in starts: + start = starts.pop(key) + durations[name].append(timestamp - start) + if name == "receipt.transaction": + transaction = active.pop(goroutine) + transaction["end"] = timestamp + latest[goroutine] = transaction + elif name == "receipt.append": + appends[task] = (start, timestamp) + elif name.startswith("receipt.") and goroutine in active: + active[goroutine]["phases"][name] += timestamp - start + message = re.search(r'G=(-?\d+) Log Time=\d+ Task=(\d+) Category="([^"]+)" Message="([^"]*)"', line) + if message: + goroutine, task, category, value = message.groups() + if category == "request.id" and value and int(value) >= 0: + tasks[int(value)] = task + if category == "receipt.reply" and goroutine in latest: + replies[task] = latest[goroutine] + pause = re.search(r'(RangeBegin|RangeEnd) Time=(\d+) Name="stop-the-world \(GC [^"]+"', line) + if pause: + event, timestamp = pause.groups() + if event == "RangeBegin": + pause_start = int(timestamp) + elif pause_start is not None: + pauses.append((pause_start, int(timestamp))) + pause_start = None + result["regions"] = {name: dict(quantiles(values), count=len(values)) for name, values in durations.items()} + result["gc_stw"] = dict(quantiles([end - start for start, end in pauses]), count=len(pauses)) if pauses else {} + phases = collections.defaultdict(int) + matched = 0 + for index in tail: + task = tasks.get(index) + if task not in appends or task not in replies: + continue + start, end = appends[task] + transaction = replies[task] + matched += 1 + phases["queue_and_writer_acquire"] += transaction["start"] - start + phases["ack_schedule"] += end - transaction["end"] + known = 0 + for name, duration in transaction["phases"].items(): + phases[name] += duration + known += duration + phases["transaction_other"] += transaction["end"] - transaction["start"] - known + phases["overlapping_gc_stw"] += sum(max(0, min(end, pe) - max(start, ps)) for ps, pe in pauses) + result["tail_receipt_trace"] = { + "matched": matched, + "mean_ms": {name: value / matched / 1e6 for name, value in phases.items()} if matched else {}, + } + print(json.dumps(result, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/bench/loadgen/main.go b/bench/loadgen/main.go index 4224581..b3006f9 100644 --- a/bench/loadgen/main.go +++ b/bench/loadgen/main.go @@ -64,7 +64,7 @@ func main() { if err := json.Unmarshal(data, &secret); err != nil || secret.Key == "" { log.Fatal("invalid credential file") } - tr := &http.Transport{MaxIdleConns: 1024, MaxIdleConnsPerHost: 1024, MaxConnsPerHost: 1024} + tr := &http.Transport{MaxIdleConns: 128, MaxIdleConnsPerHost: 128, MaxConnsPerHost: 128} defer tr.CloseIdleConnections() client := &http.Client{Transport: tr, Timeout: 15 * time.Second} act := func(scheduled time.Time, id int) sample { diff --git a/bench/proc.py b/bench/proc.py index 225a8af..1f1f5de 100644 --- a/bench/proc.py +++ b/bench/proc.py @@ -71,10 +71,11 @@ def main(): completed = subprocess.run(command, env=dict(os.environ, GOMAXPROCS="2"), text=True, capture_output=True) (directory / "loadgen.log").write_text(completed.stdout + completed.stderr) - if completed.returncode: + if completed.returncode and not completed.stdout.strip().startswith("{"): print(completed.stdout + completed.stderr, flush=True) raise RuntimeError(f"load generator failed: {directory}") - result = dict(json.loads(completed.stdout), delay=delay, name=name) + result = dict(json.loads(completed.stdout), delay=delay, name=name, + functional_pass=completed.returncode == 0) results.append(result) print(json.dumps(result), flush=True) (directory / "summary.json").write_text(json.dumps(result, indent=2) + "\n") @@ -91,6 +92,8 @@ def main(): file.close() (root / "results.json").write_text(json.dumps(results, indent=2) + "\n") print(f"Results: {root}", flush=True) + if any(not result["functional_pass"] for result in results): + raise SystemExit(1) if __name__ == "__main__": From 8373975ffb46a8a48268fa4692f751bf4b29205b Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Date: Sun, 18 Oct 2026 06:23:49 +0530 Subject: [PATCH 3/7] perf(bench): record gateway allocation deltas Capture GC counters outside the request timer. Stop tracing before writing profiles and skip stack lines during trace analysis. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> --- bench/analyze.py | 8 +++++--- bench/loadgen/main.go | 21 +++++++++++++++------ bench/procgateway/main.go | 19 +++++++++++++------ 3 files changed, 33 insertions(+), 15 deletions(-) diff --git a/bench/analyze.py b/bench/analyze.py index 9295503..e6a9352 100644 --- a/bench/analyze.py +++ b/bench/analyze.py @@ -48,7 +48,9 @@ def main(): pause_start = None with open(args.trace) as file: for line in file: - region = re.search(r'G=(-?\d+) (RegionBegin|RegionEnd) Time=(\d+) Task=(\d+) Type="([^"]+)"', line) + if not line.startswith("M="): + continue + region = re.search(r'G=(-?\d+) (RegionBegin|RegionEnd) Time=(\d+) Task=(\d+) Type="([^"]+)"', line) if "Region" in line else None if region: goroutine, event, timestamp, task, name = region.groups() timestamp = int(timestamp) @@ -68,14 +70,14 @@ def main(): appends[task] = (start, timestamp) elif name.startswith("receipt.") and goroutine in active: active[goroutine]["phases"][name] += timestamp - start - message = re.search(r'G=(-?\d+) Log Time=\d+ Task=(\d+) Category="([^"]+)" Message="([^"]*)"', line) + message = re.search(r'G=(-?\d+) Log Time=\d+ Task=(\d+) Category="([^"]+)" Message="([^"]*)"', line) if " Log " in line else None if message: goroutine, task, category, value = message.groups() if category == "request.id" and value and int(value) >= 0: tasks[int(value)] = task if category == "receipt.reply" and goroutine in latest: replies[task] = latest[goroutine] - pause = re.search(r'(RangeBegin|RangeEnd) Time=(\d+) Name="stop-the-world \(GC [^"]+"', line) + pause = re.search(r'(RangeBegin|RangeEnd) Time=(\d+) Name="stop-the-world \(GC [^"]+"', line) if "stop-the-world (GC" in line else None if pause: event, timestamp = pause.groups() if event == "RangeBegin": diff --git a/bench/loadgen/main.go b/bench/loadgen/main.go index b3006f9..3ee743c 100644 --- a/bench/loadgen/main.go +++ b/bench/loadgen/main.go @@ -114,18 +114,24 @@ func main() { } return result } - head := func() uint64 { + type snapshot struct { + Head uint64 + TotalAlloc uint64 `json:"total_alloc"` + NumGC uint64 `json:"num_gc"` + PauseTotalNS uint64 `json:"pause_total_ns"` + } + head := func() snapshot { base := strings.TrimSuffix(*url, "/v1/act") resp, err := client.Get(base + "/bench/head") if err != nil { log.Fatal(err) } defer resp.Body.Close() - var body struct{ Head uint64 } + var body snapshot if err := json.NewDecoder(resp.Body).Decode(&body); err != nil || resp.StatusCode != 200 { log.Fatal("cannot read ledger head") } - return body.Head + return body } for range 256 { if warmup := act(time.Time{}, -1); warmup.Error != "" { @@ -191,14 +197,17 @@ func main() { log.Fatal(err) } summary := map[string]any{ - "rate": *rate, "c": *concurrency, "n": *count, "errors": errors, "receipts": after - before, + "rate": *rate, "c": *concurrency, "n": *count, "errors": errors, "receipts": after.Head - before.Head, "throughput": float64(*count) / elapsed.Seconds(), "reuse_pct": 100 * float64(reused) / float64(*count), "p50_ms": percentile(slices.Clone(overhead), .50), "p90_ms": percentile(slices.Clone(overhead), .90), "p99_ms": percentile(slices.Clone(overhead), .99), "p999_ms": percentile(overhead, .999), - "dispatch_p99_ms": percentile(dispatch, .99), + "dispatch_p99_ms": percentile(dispatch, .99), + "gateway_alloc_bytes_per_request": float64(after.TotalAlloc-before.TotalAlloc) / float64(*count), + "gateway_gc_cycles": after.NumGC - before.NumGC, + "gateway_gc_pause_total_ms": float64(after.PauseTotalNS-before.PauseTotalNS) / 1e6, } json.NewEncoder(os.Stdout).Encode(summary) - if errors != 0 || after-before != uint64(*count) { + if errors != 0 || after.Head-before.Head != uint64(*count) { os.Exit(1) } } diff --git a/bench/procgateway/main.go b/bench/procgateway/main.go index 81810da..4c67925 100644 --- a/bench/procgateway/main.go +++ b/bench/procgateway/main.go @@ -93,13 +93,15 @@ func main() { defer database.Close() must(agentdb.RunMigrations(database)) var journal string - var synchronous, busyTimeout, foreignKeys int + var synchronous, busyTimeout, foreignKeys, autoCheckpoint, fullFSync int must(database.QueryRow("PRAGMA journal_mode").Scan(&journal)) must(database.QueryRow("PRAGMA synchronous").Scan(&synchronous)) must(database.QueryRow("PRAGMA busy_timeout").Scan(&busyTimeout)) must(database.QueryRow("PRAGMA foreign_keys").Scan(&foreignKeys)) - log.Printf("SQLite journal=%s synchronous=%d busy_timeout=%d foreign_keys=%d GOMAXPROCS=%d", - journal, synchronous, busyTimeout, foreignKeys, runtime.GOMAXPROCS(0)) + must(database.QueryRow("PRAGMA wal_autocheckpoint").Scan(&autoCheckpoint)) + must(database.QueryRow("PRAGMA fullfsync").Scan(&fullFSync)) + log.Printf("SQLite journal=%s synchronous=%d busy_timeout=%d foreign_keys=%d wal_autocheckpoint=%d fullfsync=%d GOMAXPROCS=%d", + journal, synchronous, busyTimeout, foreignKeys, autoCheckpoint, fullFSync, runtime.GOMAXPROCS(0)) master := make([]byte, 32) _, err = rand.Read(master) @@ -161,15 +163,15 @@ func main() { must(err) defer cpu.Close() must(pprof.StartCPUProfile(cpu)) - defer pprof.StopCPUProfile() runtime.SetMutexProfileFraction(1) runtime.SetBlockProfileRate(1) traceFile, err := os.Create(*profile + ".trace") must(err) defer traceFile.Close() must(trace.Start(traceFile)) - defer trace.Stop() defer func() { + trace.Stop() + pprof.StopCPUProfile() for _, name := range []string{"mutex", "block", "allocs", "heap"} { file, err := os.Create(*profile + "." + name) must(err) @@ -186,7 +188,12 @@ func main() { http.Error(w, err.Error(), 500) return } - json.NewEncoder(w).Encode(map[string]uint64{"head": head}) + var stats runtime.MemStats + runtime.ReadMemStats(&stats) + json.NewEncoder(w).Encode(map[string]uint64{ + "head": head, "total_alloc": stats.TotalAlloc, "num_gc": uint64(stats.NumGC), + "pause_total_ns": stats.PauseTotalNs, + }) return } timing := &timings{} From 03d6347d926db3fb4b3b89beb4e622f037dd6fce Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Date: Sun, 18 Oct 2026 06:24:16 +0530 Subject: [PATCH 4/7] perf(receipt): trim canonical buffer allocations Validate preflight drafts without building an unused preimage. Size the signed preimage from validated fields instead of maximum field sizes. Keep canonical bytes and synchronous commit acknowledgment unchanged. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> --- internal/db/sqlite_test.go | 33 +++++++++++++++++++++ internal/receipt/ledger.go | 2 +- internal/receipt/preflight_bench_test.go | 28 ++++++++++++++++++ pkg/receiptspec/encoding.go | 13 ++++++++- pkg/receiptspec/encoding_test.go | 37 ++++++++++++++++++++++++ 5 files changed, 111 insertions(+), 2 deletions(-) create mode 100644 internal/receipt/preflight_bench_test.go create mode 100644 pkg/receiptspec/encoding_test.go diff --git a/internal/db/sqlite_test.go b/internal/db/sqlite_test.go index ae49dbb..a63c314 100644 --- a/internal/db/sqlite_test.go +++ b/internal/db/sqlite_test.go @@ -7,11 +7,44 @@ package db import ( "database/sql" + "path/filepath" "testing" _ "modernc.org/sqlite" ) +func TestOpenDurabilityPragmasOnEveryConnection(t *testing.T) { + database, err := Open(filepath.Join(t.TempDir(), "durable.db")) + if err != nil { + t.Fatal(err) + } + defer database.Close() + // Hold both connections so the second check cannot reuse the first connection. + for range 2 { + connection, err := database.Conn(t.Context()) + if err != nil { + t.Fatal(err) + } + defer connection.Close() + var journal string + if err := connection.QueryRowContext(t.Context(), "PRAGMA journal_mode").Scan(&journal); err != nil { + t.Fatal(err) + } + if journal != "wal" { + t.Fatalf("journal_mode=%q, want wal", journal) + } + for query, want := range map[string]int{"PRAGMA synchronous": 2, "PRAGMA busy_timeout": 5000, "PRAGMA foreign_keys": 1} { + var got int + if err := connection.QueryRowContext(t.Context(), query).Scan(&got); err != nil { + t.Fatal(err) + } + if got != want { + t.Fatalf("%s=%d, want %d", query, got, want) + } + } + } +} + func TestRunMigrations_AppliesAllMigrationsCleanly(t *testing.T) { t.Parallel() database, err := sql.Open("sqlite", ":memory:") diff --git a/internal/receipt/ledger.go b/internal/receipt/ledger.go index 0f9432a..4b9fff4 100644 --- a/internal/receipt/ledger.go +++ b/internal/receipt/ledger.go @@ -154,7 +154,7 @@ func (l *Ledger) Append(ctx context.Context, draft Draft) (Receipt, error) { draft.DelegationChain = slices.Clone(draft.DelegationChain) preflight := draft.receipt() preflight.Seq, preflight.TimestampUnixNS, preflight.SignerKID = 1, 1, "preflight" - if _, err := receiptspec.CanonicalHashInput(preflight); err != nil { + if err := receiptspec.Validate(preflight); err != nil { return Receipt{}, appendError(err) } request := appendRequest{ctx: ctx, draft: draft, reply: make(chan appendResult, 1)} diff --git a/internal/receipt/preflight_bench_test.go b/internal/receipt/preflight_bench_test.go new file mode 100644 index 0000000..682ea4b --- /dev/null +++ b/internal/receipt/preflight_bench_test.go @@ -0,0 +1,28 @@ +package receipt + +import ( + "testing" + + "github.com/Clawdlinux/agentgate/pkg/receiptspec" +) + +func BenchmarkLedgerPreflight(b *testing.B) { + preflight := testDraft("mock", "echo").receipt() + preflight.Seq, preflight.TimestampUnixNS, preflight.SignerKID = 1, 1, "preflight" + b.Run("canonical", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + if _, err := receiptspec.CanonicalHashInput(preflight); err != nil { + b.Fatal(err) + } + } + }) + b.Run("validate", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + if err := receiptspec.Validate(preflight); err != nil { + b.Fatal(err) + } + } + }) +} diff --git a/pkg/receiptspec/encoding.go b/pkg/receiptspec/encoding.go index 0a66a94..5ec4400 100644 --- a/pkg/receiptspec/encoding.go +++ b/pkg/receiptspec/encoding.go @@ -21,7 +21,18 @@ func CanonicalHashInput(receipt Receipt) ([]byte, error) { return nil, err } - encoded := make([]byte, 0, 3110) + // Count fixed-width fields and length prefixes without reserving every field's maximum. + size := len(HashDomainV1) + 1 + 8 + 8 + 4 + 32 + 4 + 8 + 32 + for _, field := range []string{ + receipt.HumanPrincipal, receipt.AgentKeyID, receipt.Service, receipt.Action, + receipt.PolicyDecision, receipt.Error, receipt.SignerKID, + } { + size += 4 + len(field) + } + for _, element := range receipt.DelegationChain { + size += 4 + len(element) + } + encoded := make([]byte, 0, size) encoded = append(encoded, HashDomainV1...) encoded = append(encoded, 0) encoded = binary.LittleEndian.AppendUint64(encoded, receipt.Seq) diff --git a/pkg/receiptspec/encoding_test.go b/pkg/receiptspec/encoding_test.go new file mode 100644 index 0000000..38fb4b5 --- /dev/null +++ b/pkg/receiptspec/encoding_test.go @@ -0,0 +1,37 @@ +package receiptspec + +import ( + "strings" + "testing" +) + +func TestCanonicalHashInputExactCapacity(t *testing.T) { + for _, chain := range [][]string{nil, {}, {"first", "second"}, {strings.Repeat("a", 64)}} { + receipt := Receipt{ + Seq: 1, TimestampUnixNS: 1, HumanPrincipal: "human", AgentKeyID: "agent", + Service: "mock", Action: "echo", PolicyDecision: DecisionAllow, + StatusCode: 200, SignerKID: "signer", DelegationChain: chain, + } + encoded, err := CanonicalHashInput(receipt) + if err != nil { + t.Fatal(err) + } + if len(encoded) != cap(encoded) { + t.Fatalf("len=%d cap=%d: capacity must match the encoded length", len(encoded), cap(encoded)) + } + } +} + +func BenchmarkCanonicalHashInput(b *testing.B) { + receipt := Receipt{ + Seq: 1, TimestampUnixNS: 1, HumanPrincipal: "bench-user", AgentKeyID: "agent", + Service: "mock", Action: "echo", PolicyDecision: DecisionAllow, + StatusCode: 200, SignerKID: "signer", + } + b.ReportAllocs() + for b.Loop() { + if _, err := CanonicalHashInput(receipt); err != nil { + b.Fatal(err) + } + } +} From d643e7f34a0ad44a325a068f2d52f9100235ec92 Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Date: Sun, 18 Oct 2026 06:38:24 +0530 Subject: [PATCH 5/7] perf(bench): attribute connection and receipt tails Correlate receipt phases with the same tail requests. Record client acquisition separately and profile upstream reuse. Add macOS foreground-policy controls and analyzer regression tests. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> --- bench/PROC.md | 14 ++++++++ bench/analyze.py | 17 ++++++++-- bench/loadgen/main.go | 17 ++++++---- bench/proc.py | 11 +++++++ bench/procgateway/main.go | 10 ++++++ bench/test_analyze.py | 67 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 127 insertions(+), 9 deletions(-) create mode 100644 bench/test_analyze.py diff --git a/bench/PROC.md b/bench/PROC.md index 0bb4a40..23bd05a 100644 --- a/bench/PROC.md +++ b/bench/PROC.md @@ -9,6 +9,9 @@ Python 3 supervises 3 compiled Go processes: These processes share host CPUs. `GOMAXPROCS` isn't CPU pinning or a 4-vCPU quota. Default process priority remains unchanged. No privileged priority change is needed. +On macOS, `--foreground` runs `taskpolicy -B -p PID` on the gateway before warmup. +This clears background policy. It doesn't reserve CPUs. +The supervisor fails explicitly when that requested policy can't be applied. Stop unrelated CPU and disk workloads before measuring. Run on a dedicated Linux 4-vCPU gateway host before making deployment claims. @@ -57,6 +60,7 @@ make bench-proc make bench-proc BENCH_PROC_ARGS='--label repeat' make bench-proc BENCH_PROC_ARGS='--label smoke --n 64' make bench-proc BENCH_PROC_ARGS='--label profile --profile --rates 5000 --closed "" --delays 0' +make bench-proc BENCH_PROC_ARGS='--label foreground --foreground --rates 1000,5000 --closed "" --delays 0' ``` Profiles change timing. Don't use profiled requests for the latency gate. @@ -79,6 +83,16 @@ go tool pprof -top -alloc_space .cache/proc/bin/procgateway PATH/gateway.allocs go tool trace -d=parsed PATH/gateway.trace ``` +Save parsed trace output inside `.cache/`, then correlate the slowest 1%: + +```sh +python3 bench/analyze.py PATH/samples.csv --trace PATH/trace.txt +``` + +Shares describe those same requests, not sums of unrelated percentiles. +GC overlap is reported separately because it overlaps the receipt spans. +Profiled requests also record upstream connection wait and reuse through `httptrace`. + Set `TMPDIR`, `GOTMPDIR`, and `GOCACHE` under `.cache/` before invoking Go tools directly. The Makefile already does this for builds. `--batch-delay` tests existing collection waits without changing production defaults. diff --git a/bench/analyze.py b/bench/analyze.py index e6a9352..9188a63 100644 --- a/bench/analyze.py +++ b/bench/analyze.py @@ -33,9 +33,11 @@ def main(): totals["auth"] += row["auth_ns"] totals["receipt"] += row["receipt_ns"] totals["other_gateway"] += row["handler_ns"] - row["auth_ns"] - row["receipt_ns"] - row["upstream_ns"] - totals["transport_and_response"] += row["total_ns"] - row["dispatch_ns"] - row["handler_ns"] + totals["client_connection"] += row.get("client_conn_ns", 0) + totals["transport_and_response"] += row["total_ns"] - row["dispatch_ns"] - row["handler_ns"] - row.get("client_conn_ns", 0) result = { "tail_count": len(tail), + "client_connection_timing_available": "client_conn_ns" in rows[0], "tail_mean_ms": sum(totals.values()) / len(tail) / 1e6, "tail_components": {key: {"mean_ms": value / len(tail) / 1e6, "share_pct": value / sum(totals.values()) * 100} @@ -44,6 +46,7 @@ def main(): if args.trace: starts, active, latest, tasks, appends, replies = {}, {}, {}, {}, {}, {} durations = collections.defaultdict(list) + conn_starts, conn_waits, conn_reused = {}, [], 0 pauses = [] pause_start = None with open(args.trace) as file: @@ -70,13 +73,18 @@ def main(): appends[task] = (start, timestamp) elif name.startswith("receipt.") and goroutine in active: active[goroutine]["phases"][name] += timestamp - start - message = re.search(r'G=(-?\d+) Log Time=\d+ Task=(\d+) Category="([^"]+)" Message="([^"]*)"', line) if " Log " in line else None + message = re.search(r'G=(-?\d+) Log Time=(\d+) Task=(\d+) Category="([^"]+)" Message="([^"]*)"', line) if " Log " in line else None if message: - goroutine, task, category, value = message.groups() + goroutine, timestamp, task, category, value = message.groups() if category == "request.id" and value and int(value) >= 0: tasks[int(value)] = task if category == "receipt.reply" and goroutine in latest: replies[task] = latest[goroutine] + if category == "upstream.getconn": + conn_starts[task] = int(timestamp) + if category == "upstream.gotconn" and task in conn_starts: + conn_waits.append(int(timestamp) - conn_starts.pop(task)) + conn_reused += value == "true" pause = re.search(r'(RangeBegin|RangeEnd) Time=(\d+) Name="stop-the-world \(GC [^"]+"', line) if "stop-the-world (GC" in line else None if pause: event, timestamp = pause.groups() @@ -87,6 +95,9 @@ def main(): pause_start = None result["regions"] = {name: dict(quantiles(values), count=len(values)) for name, values in durations.items()} result["gc_stw"] = dict(quantiles([end - start for start, end in pauses]), count=len(pauses)) if pauses else {} + if conn_waits: + result["upstream_connections"] = dict(quantiles(conn_waits), count=len(conn_waits), + reuse_pct=100 * conn_reused / len(conn_waits)) phases = collections.defaultdict(int) matched = 0 for index in tail: diff --git a/bench/loadgen/main.go b/bench/loadgen/main.go index 3ee743c..c58572f 100644 --- a/bench/loadgen/main.go +++ b/bench/loadgen/main.go @@ -21,9 +21,9 @@ import ( ) type sample struct { - Total, Upstream, Dispatch, Auth, Receipt, Handler int64 - Reused bool - Error string + Total, Upstream, Dispatch, Auth, Receipt, Handler, ClientConn int64 + Reused bool + Error string } func percentile(values []int64, p float64) float64 { @@ -74,10 +74,14 @@ func main() { req.Header.Set("Content-Type", "application/json") req.Header.Set("X-Bench-ID", strconv.Itoa(id)) result := sample{} + var start time.Time req = req.WithContext(httptrace.WithClientTrace(req.Context(), &httptrace.ClientTrace{ - GotConn: func(info httptrace.GotConnInfo) { result.Reused = info.Reused }, + GotConn: func(info httptrace.GotConnInfo) { + result.Reused = info.Reused + result.ClientConn = time.Since(start).Nanoseconds() + }, })) - start := time.Now() + start = time.Now() if scheduled.IsZero() { scheduled = start } @@ -172,13 +176,14 @@ func main() { log.Fatal(err) } writer := csv.NewWriter(file) - writer.Write([]string{"total_ns", "upstream_ns", "dispatch_ns", "auth_ns", "receipt_ns", "handler_ns", "reused", "error"}) + writer.Write([]string{"total_ns", "upstream_ns", "dispatch_ns", "auth_ns", "receipt_ns", "handler_ns", "client_conn_ns", "reused", "error"}) overhead, dispatch := make([]int64, *count), make([]int64, *count) errors, reused := 0, 0 for i, s := range samples { writer.Write([]string{ strconv.FormatInt(s.Total, 10), strconv.FormatInt(s.Upstream, 10), strconv.FormatInt(s.Dispatch, 10), strconv.FormatInt(s.Auth, 10), strconv.FormatInt(s.Receipt, 10), strconv.FormatInt(s.Handler, 10), + strconv.FormatInt(s.ClientConn, 10), strconv.FormatBool(s.Reused), s.Error, }) overhead[i], dispatch[i] = s.Total-s.Upstream, s.Dispatch diff --git a/bench/proc.py b/bench/proc.py index 1f1f5de..ff8f831 100644 --- a/bench/proc.py +++ b/bench/proc.py @@ -5,6 +5,7 @@ import os from pathlib import Path import secrets +import shutil import signal import subprocess import time @@ -33,6 +34,7 @@ def main(): parser.add_argument("--closed", default="1,16") parser.add_argument("--delays", default="0,5ms") parser.add_argument("--batch-delay", default="0") + parser.add_argument("--foreground", action="store_true", help="clear macOS gateway background policy using taskpolicy") args = parser.parse_args() root = Path(".cache/proc") / f"{args.label}-{secrets.token_hex(4)}" root.mkdir(parents=True) @@ -63,6 +65,15 @@ def main(): process = subprocess.Popen([f".cache/proc/bin/{binary}", *flags], env=env, stdout=output, stderr=output) processes.append(process) + if binary == "procgateway" and args.foreground: + taskpolicy = shutil.which("taskpolicy") + if taskpolicy is None: + raise RuntimeError("taskpolicy is unavailable") + policy = subprocess.run([taskpolicy, "-B", "-p", str(process.pid)], + text=True, capture_output=True) + (directory / "taskpolicy.log").write_text(policy.stdout + policy.stderr) + if policy.returncode: + raise RuntimeError("taskpolicy failed; see taskpolicy.log") ready("http://127.0.0.1:" + ("19090" if binary == "mockupstream" else "18080") + "/healthz", process) command = [".cache/proc/bin/loadgen", "-n", str(args.n), "-credentials", str(directory / "key.json"), diff --git a/bench/procgateway/main.go b/bench/procgateway/main.go index 4c67925..50cf4c4 100644 --- a/bench/procgateway/main.go +++ b/bench/procgateway/main.go @@ -13,6 +13,7 @@ import ( "log/slog" "net" "net/http" + "net/http/httptrace" "os" "os/signal" "runtime" @@ -200,6 +201,15 @@ func main() { ctx, task := trace.NewTask(context.WithValue(r.Context(), timingKey{}, timing), "act") defer task.End() trace.Log(ctx, "request.id", r.Header.Get("X-Bench-ID")) + if *profile != "" { + traceContext := ctx + ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{ + GetConn: func(string) { trace.Log(traceContext, "upstream.getconn", "") }, + GotConn: func(info httptrace.GotConnInfo) { + trace.Log(traceContext, "upstream.gotconn", strconv.FormatBool(info.Reused)) + }, + }) + } handler.ServeHTTP(responseWriter{w, timing, time.Now()}, r.WithContext(ctx)) })} ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) diff --git a/bench/test_analyze.py b/bench/test_analyze.py new file mode 100644 index 0000000..3085f60 --- /dev/null +++ b/bench/test_analyze.py @@ -0,0 +1,67 @@ +import contextlib +import io +import json +import sys +import unittest +from unittest.mock import patch + +import analyze + + +class AnalyzeTest(unittest.TestCase): + def test_nearest_rank(self): + self.assertEqual(analyze.quantiles([4e6, 1e6, 2e6, 3e6]), + {"p50_ms": 2, "p99_ms": 4, "max_ms": 4}) + + def test_correlated_receipt_phases(self): + samples = ( + "total_ns,upstream_ns,dispatch_ns,auth_ns,receipt_ns,handler_ns,client_conn_ns\n" + "500,100,10,20,200,300,50\n" + "1000,100,10,20,500,800,50\n" + ) + events = [ + 'G=1 Log Time=0 Task=7 Category="request.id" Message="1"', + 'G=1 Log Time=20 Task=7 Category="upstream.getconn" Message=""', + 'G=1 Log Time=70 Task=7 Category="upstream.gotconn" Message="true"', + 'G=1 RegionBegin Time=100 Task=7 Type="receipt.append"', + 'G=2 RegionBegin Time=200 Task=0 Type="receipt.transaction"', + 'G=2 RegionBegin Time=210 Task=0 Type="receipt.head"', + 'G=2 RegionEnd Time=220 Task=0 Type="receipt.head"', + 'G=2 RegionBegin Time=230 Task=0 Type="receipt.sign"', + 'G=2 RegionEnd Time=240 Task=0 Type="receipt.sign"', + 'G=2 RegionBegin Time=250 Task=0 Type="receipt.insert"', + 'G=2 RegionEnd Time=260 Task=0 Type="receipt.insert"', + 'G=2 RegionBegin Time=300 Task=0 Type="receipt.commit"', + 'G=3 RangeBegin Time=350 Name="stop-the-world (GC mark termination)" Scope=Goroutine(3)', + 'G=3 RangeEnd Time=360 Name="stop-the-world (GC mark termination)" Scope=Goroutine(3)', + 'G=2 RegionEnd Time=450 Task=0 Type="receipt.commit"', + 'G=2 RegionEnd Time=500 Task=0 Type="receipt.transaction"', + 'G=2 Log Time=510 Task=7 Category="receipt.reply" Message=""', + 'G=1 RegionEnd Time=600 Task=7 Type="receipt.append"', + ] + trace = "\n".join("M=1 P=1 " + event for event in events) + output = io.StringIO() + + def open_fixture(path): + return io.StringIO(samples if path == "samples" else trace) + + with patch.object(sys, "argv", ["analyze", "samples", "--trace", "trace"]): + with patch("builtins.open", side_effect=open_fixture): + with contextlib.redirect_stdout(output): + analyze.main() + result = json.loads(output.getvalue()) + self.assertEqual(result["tail_count"], 1) + self.assertEqual(result["tail_mean_ms"], .0009) + self.assertEqual(result["tail_receipt_trace"]["matched"], 1) + phases = result["tail_receipt_trace"]["mean_ms"] + self.assertEqual(phases["queue_and_writer_acquire"], .0001) + self.assertEqual(phases["ack_schedule"], .0001) + self.assertEqual(phases["receipt.commit"], .00015) + self.assertEqual(phases["overlapping_gc_stw"], .00001) + self.assertEqual(result["upstream_connections"]["reuse_pct"], 100) + self.assertEqual(result["upstream_connections"]["p99_ms"], .00005) + self.assertEqual(result["gc_stw"]["count"], 1) + + +if __name__ == "__main__": + unittest.main() From 1b9ec19d6ff801d7817838af04ce1eb91cd0f0ba Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Date: Sun, 18 Oct 2026 06:43:17 +0530 Subject: [PATCH 6/7] docs(bench): report separate-process tail results Keep the failed gate explicit, including the 1.7-second transport tail. Record correlated receipt, GC, lock, and connection evidence. Qualify shared-host variance and pending Linux validation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> --- bench/RESULTS.md | 362 ++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 361 insertions(+), 1 deletion(-) diff --git a/bench/RESULTS.md b/bench/RESULTS.md index dea317e..45b142c 100644 --- a/bench/RESULTS.md +++ b/bench/RESULTS.md @@ -1,4 +1,364 @@ -# Phase 0: end-to-end act overhead +# Separate-process gateway tail investigation + +**Current verdict: FAIL. Separate processes don't establish `<1 ms p99` at moderate fixed arrival rates.** + +The allocation fix reduces gateway bytes/request by about 24% in a 5,000 requests/s comparison. +It doesn't resolve the tail gate. +Only the 0 ms, closed-loop c1 case passes in the new full matrix. +All fixed-rate cases fail in both after runs. +Linux 4-vCPU validation remains pending. + +## New method and limits + +Measured on 2026-10-18, starting from `origin/main`, `19084f8`. +The host remains Apple M4, 10 cores, 16 GiB, macOS `27.0.1`, Go `1.26.0`. +See [PROC.md](PROC.md) for the complete setup and timing contract. + +`make bench-proc` builds 3 Go binaries and supervises separate processes: + +| Process | Implementation | GOMAXPROCS | +| --- | --- | ---: | +| Gateway | `bench/procgateway`, wiring the real gateway stack | 4 | +| Upstream | `bench/mockupstream` | 2 | +| Load generator | `bench/loadgen` | 2 | + +The gateway retains real authentication, scoped authorization, active rate limits, encrypted caches, and the signed SQLite ledger. +No delegation token is presented. No production delegation check changed. +SQLite startup reports WAL and `synchronous=2`, meaning FULL. +Both checked database connections retain WAL, FULL, foreign keys, and the 5,000 ms busy timeout. +The new durability regression test enforces those settings on 2 simultaneous connections. +No receipt can acknowledge success before COMMIT returns. +Batch size remains 256. Collection delay remains 0. +The production upstream transport remains unchanged. +Info logs still undergo JSON encoding, with sink writes discarded. + +Each case uses a fresh database, 256 warmup requests, and 20,000 measured requests. +The mock supplies measured upstream handler nanoseconds inside its JSON response. +This subtracts actual sleep overshoot, not a nominal 5 ms. +Response encoding and network writes remain in overhead. +Open-loop time starts at the original scheduled arrival, including late dispatch and connection-pool waits. +Closed-loop time starts immediately before `Client.Do`. +Both end after reading and closing the response body. +Subtraction precedes sorting. Percentiles use nearest rank. +The client caps connections at 128. This bounds connection bursts, not arrivals. +Every scheduled request still runs. Its original timestamp never resets. + +The first full before and after matrices each completed 240,000 measured requests without errors. +Every case's ledger head advanced by exactly 20,000. +The repeated after fixed-rate matrix also completed all 160,000 requests. +Initial 1,024-connection experiments encountered loopback connection resets. +Those failed experiments aren't passing evidence and aren't included in the main comparison. + +`GOMAXPROCS=4` isn't CPU isolation, pinning, or a VM quota. +Background CPU activity was observed during this shared-host investigation. +Other test workloads were deferred when coordination allowed. +Interactive host processes remained active. +These runs have no isolated-host confidence interval. +Don't treat a better single run as proof of a causal p99 improvement. +The earlier shared-generator method was flawed, but its FAIL wasn't proved to be method-only. + +## Separate-process before and after + +All latency values are milliseconds. Throughput includes measured upstream time. +Before uses the unoptimized production paths, with the new process harness. +After includes `d15cf71`, the receipt allocation fix. +The repeat reruns every fixed rate after that same production fix. + +| Upstream | Load | Before p99 | After p50 | After p99 | Repeat after p99 | After requests/s | After gate | +| --- | --- | ---: | ---: | ---: | ---: | ---: | --- | +| 0 ms | 1,000/s | 5.785 | 0.250 | 1.952 | 4.535 | 1,000 | FAIL | +| 0 ms | 2,500/s | 33.733 | 0.218 | 5.586 | 33.822 | 2,500 | FAIL | +| 0 ms | 5,000/s | 50.720 | 0.193 | 2.734 | 8.194 | 5,000 | FAIL | +| 0 ms | 10,000/s | 73.768 | 0.419 | 5.270 | 69.471 | 9,999 | FAIL | +| 0 ms | Closed c1 | 0.690 | 0.165 | 0.438 | Not repeated | 5,274 | PASS, single run | +| 0 ms | Closed c16 | 5.611 | 0.845 | 3.643 | Not repeated | 16,312 | FAIL | +| 5 ms | 1,000/s | 21.760 | 0.267 | 2.304 | 6.233 | 1,000 | FAIL | +| 5 ms | 2,500/s | 5.590 | 0.293 | 4.457 | 9.699 | 2,498 | FAIL | +| 5 ms | 5,000/s | 45.674 | 22.866 | 1,702.730 | 59.228 | 3,777 | FAIL | +| 5 ms | 10,000/s | 149.782 | 0.705 | 25.411 | 100.367 | 9,790 | FAIL | +| 5 ms | Closed c1 | 4.802 | 0.506 | 2.302 | Not repeated | 147.6 | FAIL | +| 5 ms | Closed c16 | 4.434 | 1.073 | 4.113 | Not repeated | 2,477 | FAIL | + +The 1.7-second after tail isn't discarded as an outlier. +Its slowest 1% averages 1,715.248 ms overhead. +Receipt append accounts for 58.481 ms, or 3.41%, of that conditional mean. +Transport, request acceptance, client connection waits, and response completion account for 1,655.857 ms, or 96.54%. +Dispatch accounts for 0.0067 ms. Authentication accounts for 0.0016 ms. +Other gateway work accounts for 0.902 ms. +That run predates explicit client-connection timing, so its residual can't be split retrospectively. +The subsequent same-rate run still fails, with p99 59.228 ms. +This variability prevents an honest claim that receipt allocation alone fixes the observed p99. + +Local evidence directories, all ignored: + +| Evidence | Directory under `.cache/proc/` | +| --- | --- | +| Full before matrix | `bounded-before-c9598240` | +| Full after matrix | `after-9e36bcec` | +| Repeated after fixed rates | `repeat-after-69b46760` | +| Controlled 5,000/s allocation baseline | `controlled-before-f9aabf7a` | +| Before 5,000/s profile | `prof-before-c5807e20` | +| After profiles, 0 ms and 5 ms | `prof-after-0f54c36b` | +| Foreground-policy control | `foreground-f8e7de70` | +| Profiles with client-connection timing | `latest-prof-892308cc` | + +These private files aren't committed. Each directory contains CSV samples and JSON summaries. +Profiles remain local, with the databases. + +### macOS foreground-policy control + +`taskpolicy -B -p PID` succeeded before each gateway warmup. +This clears background policy. It doesn't pin CPUs or isolate the generator. +All 4 cases completed without errors, with 20,000 committed receipts each. +These additional unprofiled controls still miss the gate. +They aren't evidence that foreground policy worsens latency. Host activity wasn't controlled. + +| Upstream | Offered requests/s | Overhead p99, ms | Dispatch p99, ms | Gate | +| --- | ---: | ---: | ---: | --- | +| 0 ms | 1,000 | 85.127 | 36.857 | FAIL | +| 0 ms | 5,000 | 5.352 | 0.0408 | FAIL | +| 5 ms | 1,000 | 10.809 | 0.0610 | FAIL | +| 5 ms | 5,000 | 65.628 | 0.9912 | FAIL | + +The first control recorded 204.066 ms of aggregate gateway GC pauses. +Its large dispatch lag also demonstrates scheduler pressure outside receipt COMMIT. +No assumed platform penalty is subtracted from any gate. + +## Tail contributors + +Profiles are gateway-only. The generator and mock run in different processes. +CPU, block, mutex, allocation, execution trace, and `GODEBUG=gctrace=1` evidence were collected. +Request IDs correlate client samples with receipt transactions and committed replies. +Profiled numbers aren't substituted into the gate table. + +### Correlated receipt-heavy tail + +The quieter after profile uses 5,000/s and 0 ms upstream. +Its overhead p99 is 3.723 ms. +The slowest 200 requests average 4.503 ms overhead. +The following partition uses those same requests. It doesn't add unrelated percentile ranks. + +| Contributor | Conditional mean, ms | Share of that tail mean | +| --- | ---: | ---: | +| Load-generator dispatch | 0.023 | 0.51% | +| Authentication, including key cache | 0.001 | 0.03% | +| Receipt append, including all waits | 4.137 | 91.87% | +| Other gateway work, minus upstream time | 0.119 | 2.65% | +| Transport and response completion | 0.222 | 4.94% | + +Client-connection timing wasn't recorded in that profile's CSV. +Its transport row therefore includes the client connection wait. +Upstream `httptrace` was available: reuse was 99.975%, with connection wait p99 0.0093 ms. +Its maximum upstream connection wait was 0.355 ms. +That doesn't support a connection-pool change as the primary fix for this receipt-heavy tail. + +All 200 tail requests matched their committed receipt transactions. +Within their receipt wait, these trace spans averaged: + +| Receipt contributor | Conditional mean, ms | +| --- | ---: | +| Queueing, preflight, and writer acquisition | 2.470 | +| Head read | 0.0285 | +| Signing the entire batch | 0.4065 | +| Inserting the entire batch | 0.5500 | +| SQLite COMMIT | 0.5774 | +| Other transaction work | 0.0251 | +| Reply scheduling after the transaction | 0.0790 | + +Queueing accounts for about 54.9% of the full tail mean. +COMMIT accounts for about 12.8%. +Batch signing and inserts account for about 9.0% and 12.2%. +These are wall-clock spans. They include descheduling, not just crypto or SQLite CPU. +Each caller waits for its entire batch, so batch work is deliberately attributed to every waiting caller. +Trace timer boundaries differ slightly from the response-header timer. + +Across all transactions, rather than only the tail requests: + +| Trace region | Before p99, ms | After p99, ms | +| --- | ---: | ---: | +| Head read | 0.0473 | 0.0623 | +| Receipt signing | 0.0509 | 0.0588 | +| Receipt insert | 0.0922 | 0.1228 | +| SQLite COMMIT | 0.2134 | 0.2897 | +| Entire transaction | 0.4897 | 0.7171 | +| Caller append wait | 12.0573 | 3.4212 | + +After COMMIT p50 is 0.0628 ms, with maximum 4.186 ms. +The automatic WAL checkpoint threshold remains 1,000 pages. +The existing macOS `fullfsync` setting remains 0. SQLite synchronous mode remains FULL. +COMMIT includes SQLite's WAL sync and any automatic checkpoint work. +A separate kernel fsync-duration distribution wasn't obtained. +macOS `fs_usage` refused unprivileged execution. No privileged trace was run. +Don't describe these COMMIT wall times as isolated fsync syscall times. +Linux syscall-level sync validation remains pending. + +### Locks, GC, JSON, logging, and connections + +The receipt-heavy after CPU profile spans 4.31s, with 4.69s of attributed CPU samples. +System calls account for 59.91% of samples. +`Ledger.commitBatch` accounts for 50.75% cumulatively. +SQLite `_full_fsync` accounts for 6.40% cumulatively. +These paths overlap. CPU shares aren't p99 wall-time shares. +`slog.Info` accounts for 0.64% of CPU samples. Authentication accounts for 0.21%. +JSON encoding wasn't a sampled CPU hotspot in this short run. +JSON decoder buffers still account for 5.69% of sampled allocation bytes. + +The block profile attributes 6.26s of aggregate wait to `Ledger.Append`. +It attributes 1.20s to upstream dispatch. +Those are summed goroutine waits, not elapsed process time. +Ledger submission and acknowledgment use channels, not a contended limiter lock. + +The mutex profile records 3.12 ms total observed delay over 20,256 requests. +Limiter admission accounts for about 0.32 ms cumulatively. Slot release accounts for 0.34 ms. +The upstream pool accounts for 0.25 ms. +Vault-cache and key-cache locks aren't reported as contention hotspots. +For scale, distributing every observed mutex delay over 200 tail requests averages 0.0156 ms each. +This doesn't justify adding lock sharding here. + +The receipt-heavy trace contains 144 GC stop-the-world spans. +Their p99 is 0.146 ms, with maximum 0.148 ms. +Their mean overlap with a tail receipt append is 0.0101 ms. +GC overlap is already inside the receipt spans. Don't add it again. +That overlap is about 0.22% of the conditional tail mean. +It doesn't explain this run's multi-millisecond queueing by itself. + +The profiled 5 ms case has upstream connection reuse of 99.131%. +Upstream connection wait p99 is 0.1125 ms, with maximum 7.404 ms. +Its slowest 1% averages 50.926 ms overhead. +Receipt wait contributes 19.96%, dispatch 12.52%, other gateway work 21.38%, and transport completion 46.13%. +Again, its older CSV can't split client connection waits from the last row. + +The later instrumented 5 ms profile explicitly records client connection acquisition. +Its slowest 1% averages 61.788 ms overhead: + +| Contributor | Share | +| --- | ---: | +| Client connection acquisition and pool queue | 39.25% | +| Receipt append | 22.52% | +| Other gateway work | 18.84% | +| Remaining transport and response completion | 15.64% | +| Dispatch | 3.75% | +| Authentication | 0.01% | + +Upstream reuse there is 98.233%, with connection-wait p99 0.431 ms. +Client acquisition includes dialing and pool waiting. It isn't a gateway receipt cost. +All of it remains in the conservative end-to-end gate. +The later 0 ms profile also shows severe scheduler pressure. +Its conditional tail assigns 75.64% to client connection acquisition and 5.44% to receipts. +Its GC pause maximum reaches 47.858 ms. Its dispatch p99 reaches 72.119 ms. +These changing shares are why a single syscall CPU profile couldn't establish the original root cause. +The remaining tail includes host scheduling and transport queueing, alongside the serialized durable writer. +No fixed share is claimed for uninstrumented JSON, logging, or kernel socket substeps. + +## Safe fix applied + +`d15cf71`, `perf(receipt): trim canonical buffer allocations`: + +- Validate preflight drafts through the same `receiptspec.Validate` contract, without encoding an unused preimage. +- Reserve the exact validated preimage length instead of every field's maximum size. + +Canonical bytes, receipt hashes, and signatures remain unchanged. +Signing and inserts still finish before synchronous COMMIT and acknowledgment. +Authentication, rate limits, and delegation behavior remain unchanged. +No GC setting, batch delay, transport limit, or SQLite durability setting changed. +Golden vectors and chain verification tests pass. + +The microbenchmarks use a compiler overlay for the original encoding and ledger sources. +They run the same benchmark code before and after, on this host's 10 default Go execution slots. +These aren't load-gate measurements. + +| Microbenchmark | Before | After | +| --- | ---: | ---: | +| Preflight production path | 489.5 ns, 3,200 B, 1 allocation | 44.23 ns, 0 B, 0 allocations | +| Canonical preimage encoding | 251.7 ns, 3,200 B, 1 allocation | 89.53 ns, 192 B, 1 allocation | + +At 5,000/s with 0 ms upstream, a separate controlled before run allocated 25,977 B/request. +The full after run allocated 19,704 B/request, about 24.2% less. +That control's p99 was 19.950 ms, rather than the earlier matrix's 50.720 ms. +The after p99 was 2.734 ms, then 8.194 ms on repetition. +Host variability prevents a firm causal claim about the p99 change. +The allocation benefit is verified independently. + +For that controlled comparison, GC cycles fell from 201 to 184 per 20,000 requests. +Total GC pause time fell from 11.391 ms to 8.984 ms. +Those process counters don't isolate each tail request. +They don't explain the second-scale transport residual. + +## Reproduction for this investigation + +```sh +make bench-proc BENCH_PROC_ARGS='--label before' +make bench-proc BENCH_PROC_ARGS='--label after' +make bench-proc BENCH_PROC_ARGS='--label repeat-after --closed ""' +make bench-proc BENCH_PROC_ARGS='--label profile --profile --rates 5000 --closed "" --delays 0,5ms' +``` + +The first command must run before applying `d15cf71`, or with its parent production sources overlaid. +Keep the harness identical during a paired comparison. +Use `.cache/` for overlays, Go scratch files, and profiles. +Run `go test` allocation benchmarks with `-run '^$' -benchmem`. +The benchmark names are `BenchmarkLedgerPreflight` and `BenchmarkCanonicalHashInput`. +See [PROC.md](PROC.md) for profile decoding and correlated tail analysis. + +## Investigation commits and validation + +| Commit | Change | +| --- | --- | +| `8632eb0` | Separate-process harness and receipt trace regions | +| `d770739` | Bound client connection bursts and retain failed-case evidence | +| `36164ba` | Record gateway allocation and GC deltas | +| `d15cf71` | Reduce receipt preimage allocations without changing signed bytes | +| `9d39315` | Correlate connection and receipt tails, with analyzer regression tests | + +The production allocation fix is isolated in `d15cf71`. +The harness commits don't change production transport, authorization, or durability settings. +All commits are signed off and include the requested co-author trailer. +Nothing was pushed. + +Final checks use `GOMAXPROCS=4`, with `TMPDIR`, `GOTMPDIR`, and `GOCACHE` inside `.cache/`. +`SEMGREP_SEND_METRICS=on` allows automatic local rule selection. +`SEMGREP_ENABLE_VERSION_CHECK=0` disables the unrelated version check. +Semgrep ran locally. No remote code analysis was enabled. + +| Command | Result | +| --- | --- | +| `go build ./...` | PASS | +| `go vet ./...` | PASS | +| `go vet -tags=bench ./bench` | PASS | +| `go test -race ./...` | PASS, every normal package | +| `go test -tags=bench ./bench ./bench/mockupstream -run Test -count=1` | PASS | +| `python3 -m unittest discover -s bench -p 'test_*.py'` | PASS, 2 analyzer tests | +| `semgrep --config auto --error --quiet ./internal ./cmd` | PASS, exit 0 | +| `make bench-proc BENCH_PROC_ARGS='--label final-smoke --n 64'` | PASS, all 12 functional cases | +| Full before, after, and repeated fixed-rate commands above | PASS functional checks; latency gates FAIL | +| Foreground-policy control | PASS functional checks; all 4 latency gates FAIL | +| Gateway-only CPU, block, mutex, allocation, and trace collection | PASS | + +The final smoke evidence is under `.cache/proc/final-smoke-0abec342`. +Small smoke samples aren't latency-gate evidence. +No cache, database, or profile file is committed. + +## Remaining work and tuning boundaries + +The current evidence doesn't justify limiter sharding or weaker receipt durability. +Don't replace FULL with NORMAL or acknowledge an uncommitted receipt. +Don't drop late arrivals or subtract assumed scheduler penalties. +Keep the zero collection delay until a repeated paired experiment supports a different setting. + +`GOGC=200` is an optional deployment experiment when memory headroom allows. +Pair it with a `GOMEMLIMIT` below the pod memory budget. +Leave space for SQLite's native allocations, socket buffers, and the runtime. +Neither setting is hardcoded or claimed as a measured fix here. +GC pause reduction alone hasn't established the p99 gate. + +Dedicated Linux validation must isolate the gateway's 4 vCPUs from the generator and mock. +Repeat each fixed rate and both upstream delays. +Capture syscall-level sync timing there, alongside the same per-request receipt and connection spans. +Until then, no moderate-load production p99 pass is claimed. + +--- + +# Historical Phase 0: in-process act overhead **Verdict: FAIL overall. The `<1 ms p99` exit gate is not met across the tested load levels.** From dc5c05e6748ded9b5b44d68d9c89d05ed116d0e5 Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> Date: Sun, 18 Oct 2026 06:52:16 +0530 Subject: [PATCH 7/7] fix(bench): invalidate failed-request latency rows Exclude failed samples from latency statistics and suppress p99 on invalid rows. Keep successful-only diagnostic percentiles and test partial and total failures. Attribute p99 movement to host noise and clarify production trace instrumentation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Shreyansh Sancheti <43677304+shreyanshjain7174@users.noreply.github.com> --- bench/PROC.md | 3 +++ bench/RESULTS.md | 17 +++++++++--- bench/loadgen/main.go | 55 ++++++++++++++++++++++++++++---------- bench/loadgen/main_test.go | 44 ++++++++++++++++++++++++++++++ 4 files changed, 102 insertions(+), 17 deletions(-) diff --git a/bench/PROC.md b/bench/PROC.md index 23bd05a..06c0b16 100644 --- a/bench/PROC.md +++ b/bench/PROC.md @@ -47,6 +47,9 @@ overhead = response_completed - scheduled_arrival - upstream_ns Subtraction occurs per request, before nearest-rank percentiles. All requests must succeed. The ledger head must advance by exactly the measured count. +Failed samples are excluded from latency percentiles. +A failed request or receipt-count mismatch marks the summary invalid and emits `p99_ms: null`. +Other percentiles, when available, describe successful samples only. Invalid rows still exit with failure. The client records request samples, connection reuse, and gateway timing headers. Those headers exist only in this loopback benchmark main, not the public gateway. `handler_ns` ends before response encoding. `receipt_ns` includes ledger queueing and committed acknowledgment. diff --git a/bench/RESULTS.md b/bench/RESULTS.md index 45b142c..2766987 100644 --- a/bench/RESULTS.md +++ b/bench/RESULTS.md @@ -42,6 +42,10 @@ Both end after reading and closing the response body. Subtraction precedes sorting. Percentiles use nearest rank. The client caps connections at 128. This bounds connection bursts, not arrivals. Every scheduled request still runs. Its original timestamp never resets. +Failed samples are excluded from latency percentiles. +Any failed request marks the summary invalid and emits `p99_ms: null`. +Other available percentiles describe successful requests only. They aren't a passing gate for an invalid row. +Receipt-count mismatches also invalidate the row and suppress p99. The first full before and after matrices each completed 240,000 measured requests without errors. Every case's ledger head advanced by exactly 20,000. @@ -183,6 +187,10 @@ Across all transactions, rather than only the tail requests: | Entire transaction | 0.4897 | 0.7171 | | Caller append wait | 12.0573 | 3.4212 | +This is a noisy, single-run comparison, not evidence of an allocation-driven tail improvement. +Caller append p99 fell while transaction p99 increased in the same table. +Don't attribute that movement to `d15cf71`. + After COMMIT p50 is 0.0628 ms, with maximum 4.186 ms. The automatic WAL checkpoint threshold remains 1,000 pages. The existing macOS `fullfsync` setting remains 0. SQLite synchronous mode remains FULL. @@ -275,9 +283,8 @@ These aren't load-gate measurements. At 5,000/s with 0 ms upstream, a separate controlled before run allocated 25,977 B/request. The full after run allocated 19,704 B/request, about 24.2% less. That control's p99 was 19.950 ms, rather than the earlier matrix's 50.720 ms. -The after p99 was 2.734 ms, then 8.194 ms on repetition. -Host variability prevents a firm causal claim about the p99 change. -The allocation benefit is verified independently. +The 50.720 → 2.734 → 8.194 ms p99 movement is shared-host noise, not an effect of `d15cf71`. +That commit saves about 6.2 KB of allocations per request, verified independently. For that controlled comparison, GC cycles fell from 201 to 184 per 20,000 requests. Total GC pause time fell from 11.391 ms to 8.984 ms. @@ -302,6 +309,8 @@ See [PROC.md](PROC.md) for profile decoding and correlated tail analysis. ## Investigation commits and validation +These IDs identify the measured pre-rebase commits. + | Commit | Change | | --- | --- | | `8632eb0` | Separate-process harness and receipt trace regions | @@ -311,6 +320,8 @@ See [PROC.md](PROC.md) for profile decoding and correlated tail analysis. | `9d39315` | Correlate connection and receipt tails, with analyzer regression tests | The production allocation fix is isolated in `d15cf71`. +`8632eb0` also adds production trace instrumentation in `internal/receipt/ledger.go`. +That instrumentation is a no-op when tracing is off. The harness commits don't change production transport, authorization, or durability settings. All commits are signed off and include the requested co-author trailer. Nothing was pushed. diff --git a/bench/loadgen/main.go b/bench/loadgen/main.go index c58572f..f24072a 100644 --- a/bench/loadgen/main.go +++ b/bench/loadgen/main.go @@ -31,6 +31,34 @@ func percentile(values []int64, p float64) float64 { return float64(values[max(0, int(math.Ceil(float64(len(values))*p))-1)]) / 1e6 } +func summarizeLatencies(samples []sample) map[string]any { + overhead, dispatch := make([]int64, 0, len(samples)), make([]int64, 0, len(samples)) + failures := 0 + for _, s := range samples { + if s.Error != "" { + failures++ + continue + } + overhead = append(overhead, s.Total-s.Upstream) + dispatch = append(dispatch, s.Dispatch) + } + summary := map[string]any{ + "errors": failures, "successful": len(overhead), "valid": failures == 0 && len(overhead) > 0, + "p50_ms": nil, "p90_ms": nil, "p99_ms": nil, "p999_ms": nil, "dispatch_p99_ms": nil, + } + if len(overhead) == 0 { + return summary + } + summary["p50_ms"] = percentile(slices.Clone(overhead), .50) + summary["p90_ms"] = percentile(slices.Clone(overhead), .90) + summary["p999_ms"] = percentile(slices.Clone(overhead), .999) + summary["dispatch_p99_ms"] = percentile(dispatch, .99) + if failures == 0 { + summary["p99_ms"] = percentile(overhead, .99) + } + return summary +} + func waitUntil(deadline time.Time) { for { left := time.Until(deadline) @@ -177,19 +205,14 @@ func main() { } writer := csv.NewWriter(file) writer.Write([]string{"total_ns", "upstream_ns", "dispatch_ns", "auth_ns", "receipt_ns", "handler_ns", "client_conn_ns", "reused", "error"}) - overhead, dispatch := make([]int64, *count), make([]int64, *count) - errors, reused := 0, 0 - for i, s := range samples { + reused := 0 + for _, s := range samples { writer.Write([]string{ strconv.FormatInt(s.Total, 10), strconv.FormatInt(s.Upstream, 10), strconv.FormatInt(s.Dispatch, 10), strconv.FormatInt(s.Auth, 10), strconv.FormatInt(s.Receipt, 10), strconv.FormatInt(s.Handler, 10), strconv.FormatInt(s.ClientConn, 10), strconv.FormatBool(s.Reused), s.Error, }) - overhead[i], dispatch[i] = s.Total-s.Upstream, s.Dispatch - if s.Error != "" { - errors++ - } if s.Reused { reused++ } @@ -201,18 +224,22 @@ func main() { if err := file.Close(); err != nil { log.Fatal(err) } - summary := map[string]any{ - "rate": *rate, "c": *concurrency, "n": *count, "errors": errors, "receipts": after.Head - before.Head, - "throughput": float64(*count) / elapsed.Seconds(), "reuse_pct": 100 * float64(reused) / float64(*count), - "p50_ms": percentile(slices.Clone(overhead), .50), "p90_ms": percentile(slices.Clone(overhead), .90), - "p99_ms": percentile(slices.Clone(overhead), .99), "p999_ms": percentile(overhead, .999), - "dispatch_p99_ms": percentile(dispatch, .99), + summary := summarizeLatencies(samples) + for key, value := range map[string]any{ + "rate": *rate, "c": *concurrency, "n": *count, "receipts": after.Head - before.Head, + "throughput": float64(summary["successful"].(int)) / elapsed.Seconds(), + "reuse_pct": 100 * float64(reused) / float64(*count), "gateway_alloc_bytes_per_request": float64(after.TotalAlloc-before.TotalAlloc) / float64(*count), "gateway_gc_cycles": after.NumGC - before.NumGC, "gateway_gc_pause_total_ms": float64(after.PauseTotalNS-before.PauseTotalNS) / 1e6, + } { + summary[key] = value + } + if after.Head-before.Head != uint64(*count) { + summary["valid"], summary["p99_ms"] = false, nil } json.NewEncoder(os.Stdout).Encode(summary) - if errors != 0 || after.Head-before.Head != uint64(*count) { + if !summary["valid"].(bool) { os.Exit(1) } } diff --git a/bench/loadgen/main_test.go b/bench/loadgen/main_test.go index e799a24..69fec8e 100644 --- a/bench/loadgen/main_test.go +++ b/bench/loadgen/main_test.go @@ -1,10 +1,54 @@ package main import ( + "encoding/json" "testing" "time" ) +func TestSummarizeLatencies(t *testing.T) { + successes := []sample{ + {Total: 10e6, Upstream: 2e6, Dispatch: 1e6}, + {Total: 20e6, Upstream: 4e6, Dispatch: 3e6}, + } + failures := []sample{{Error: "request failed"}, {Error: "body read failed"}, {Error: "invalid response"}} + for _, test := range []struct { + name string + samples []sample + successful, errors int + valid bool + p50, p99 any + }{ + {"success", successes, 2, 0, true, float64(8), float64(16)}, + {"partial failure", append(append([]sample{}, failures...), successes...), 2, 3, false, float64(8), nil}, + {"all failed", failures, 0, 3, false, nil, nil}, + {"empty", nil, 0, 0, false, nil, nil}, + } { + t.Run(test.name, func(t *testing.T) { + summary := summarizeLatencies(test.samples) + for key, want := range map[string]any{ + "successful": test.successful, "errors": test.errors, "valid": test.valid, + "p50_ms": test.p50, "p99_ms": test.p99, + } { + if got := summary[key]; got != want { + t.Fatalf("%s=%v, want %v", key, got, want) + } + } + data, err := json.Marshal(summary) + if err != nil { + t.Fatal(err) + } + var encoded map[string]json.RawMessage + if err := json.Unmarshal(data, &encoded); err != nil { + t.Fatal(err) + } + if !test.valid && string(encoded["p99_ms"]) != "null" { + t.Fatalf("invalid row must encode p99_ms as null: %s", data) + } + }) + } +} + func TestPercentileNearestRank(t *testing.T) { values := []int64{4e6, 1e6, 3e6, 2e6} for _, test := range []struct {