From f5b16616b7bdd1aa8364ca135793a404e8d4c627 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20Bajto=C5=A1?= Date: Tue, 22 Sep 2026 09:21:11 +0200 Subject: [PATCH 1/5] feat: execute a request's invocations concurrently MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The server ran the invocations of one request one after another, so a request carrying 100 invocations took longer than 100 requests carrying one each, and batching made large multipart uploads slower rather than faster. Run each invocation on its own goroutine, the way net/http runs each connection, admitted through a per-request semaphore the way HTTP/2 and QUIC servers cap concurrent streams per connection. WithMaxConcurrency sets the cap; the default of 100 matches quic-go and the RFC 9113 floor, zero removes the cap and a negative value panics. The semaphore is acquired before spawning, so an oversized request waits in the loop rather than as a pile of blocked goroutines. Receipts and metadata are collected into a slice indexed by request position and merged after every invocation finishes, so the response is the same regardless of which handler finished first and needs no lock. Dispatcher errors are joined and surface after all siblings have completed; they still fail the batch as before. Benchmark, one request of 100 invocations, 1ms sleeping handler: sequential 151ms, concurrent 4.6ms, against 4.5ms for 100 concurrent requests of one. The CPU-bound handler goes from 18.9ms to 3.7ms. Fixes FIL-1261 Assisted-by: Claude:claude-fable-5-1 Signed-off-by: Miroslav Bajtoš --- README.md | 3 +- execution/batch/batch.go | 11 +-- execution/execution.go | 5 ++ notes.md | 1 + server/bench_test.go | 102 +++++++++++++++++++++++ server/concurrency_test.go | 165 +++++++++++++++++++++++++++++++++++++ server/http.go | 99 ++++++++++++++++------ server/http_test.go | 4 + server/options.go | 24 ++++++ 9 files changed, 384 insertions(+), 30 deletions(-) create mode 100644 server/bench_test.go create mode 100644 server/concurrency_test.go diff --git a/README.md b/README.md index ac838e3..d958bf0 100644 --- a/README.md +++ b/README.md @@ -180,7 +180,8 @@ if out := resp.Receipt().Out(); out.IsOK() { ``` Many invocations can be sent in one request. The service executes every -invocation addressed to it and answers each with its own receipt: +invocation addressed to it concurrently, up to `server.WithMaxConcurrency` +at a time (100 by default), and answers each with its own receipt: ```go res, err := c.ExecuteBatch(batch.NewRequest(context.Background(), invs, batch.WithDelegations(dlg))) diff --git a/execution/batch/batch.go b/execution/batch/batch.go index 8a184e9..07ba425 100644 --- a/execution/batch/batch.go +++ b/execution/batch/batch.go @@ -6,11 +6,12 @@ // that travel alongside them (proofs, receipts, context invocations), and a // [Response] holds the receipts keyed by the task they ran. // -// Container encoding sorts tokens bytewise, so an executor sees no particular -// order: a batch must not depend on one invocation running before another, and -// receipts are addressed by task rather than by position. A container also -// holds at most 8192 tokens, counting the invocations, their proofs and any -// receipts together, so a batch and its metadata must fit within that. +// Container encoding sorts tokens bytewise, and the server executes the +// invocations of a request concurrently, so a batch must not depend on one +// invocation running before another, and receipts are addressed by task +// rather than by position. A container also holds at most 8192 tokens, +// counting the invocations, their proofs and any receipts together, so a batch +// and its metadata must fit within that. // // The single-invocation API in the execution package is a special case of // this one: a batch of one. diff --git a/execution/execution.go b/execution/execution.go index f10da56..d7ca229 100644 --- a/execution/execution.go +++ b/execution/execution.go @@ -42,4 +42,9 @@ type Executor interface { } // HandlerFunc is a function that can handle a specific UCAN invocation. +// +// A server calls handlers concurrently: across requests, as any HTTP handler, +// and within one request, since the invocations of a batch execute at the +// same time. A handler must not rely on another invocation of the same request +// having run before it, and any state it shares needs its own locking. type HandlerFunc = func(Request, Response) error diff --git a/notes.md b/notes.md index f59f961..447cec4 100644 --- a/notes.md +++ b/notes.md @@ -23,3 +23,4 @@ * Containers hold at most `container.MaxTokens` (8192) tokens. The container spec sets no limit; this is the cbor-gen/dag-json-gen array limit, pinned with a `maxlen` struct tag on `ContainerModel.Ctn1` so a generator upgrade cannot move it silently. The exported constant lets callers budget before encoding. * Server is a HTTP `RoundTripper` * A panic while executing an invocation fails its own task: `Dispatcher.Execute` recovers it and issues a short-lived `ExecutionFailure` receipt, with the same TTL a returned handler error gets. The boundary is the whole of `Execute` because validation runs application code too (a `WithDIDResolver`, a `WithVerifierFactories` factory, a nonstandard-signature verifier), and a panic there would otherwise escape the worker goroutine the server runs each invocation on. There is one receipt for every panic, and it does not say that a panic happened, let alone where: a client cannot act on the distinction, and a developer debugging it needs the log line anyway, since the panic value never reaches the client (it was not written for one and can carry internals). The value goes to a `PanicLogger` callback (`WithPanicLogger` on the dispatcher and the server); the default prints it with the stack through the standard `log` package and `runtime.Stack`, as net/http does, so the library links neither `log/slog` nor `runtime/debug` into a consumer binary. Before this, a panic escaped to net/http, which dropped the connection with no receipt; with invocations executing on their own goroutines it would have crashed the process. +* The server executes the invocations of one request concurrently, one goroutine each, admitted through a per-request semaphore (`server.WithMaxConcurrency`, default 100, zero for no cap). This is what every Go server surveyed does for streams on a connection (grpc-go, x/net/http2, quic-go, go-libp2p): goroutine per unit, a semaphore or queue in front. The cap is per request rather than server-wide because that is the HTTP/2 analogue and because a process-wide bound belongs at the listener (`netutil.LimitListener`, `HTTP2Config.MaxConcurrentStreams`), where the Go team put it when asked for one in net/http (golang/go#6012). The default of 100 matches quic-go and the RFC 9113 floor; x/net/http2 uses 250 and grpc-go none. Receipts are merged in request order after every invocation has finished, so the response does not depend on which handler finished first. diff --git a/server/bench_test.go b/server/bench_test.go new file mode 100644 index 0000000..a2b3be3 --- /dev/null +++ b/server/bench_test.go @@ -0,0 +1,102 @@ +package server_test + +import ( + "crypto/sha256" + "sync" + "testing" + "time" + + "github.com/fil-forge/ucantone/execution" + "github.com/fil-forge/ucantone/execution/batch" + "github.com/fil-forge/ucantone/ipld/datamodel" + "github.com/fil-forge/ucantone/multikey/ed25519" + "github.com/fil-forge/ucantone/server" + "github.com/fil-forge/ucantone/testutil" + "github.com/fil-forge/ucantone/ucan" + "github.com/fil-forge/ucantone/ucan/invocation" +) + +// BenchmarkBatch compares one request carrying n invocations against n +// concurrent requests carrying one each, for a handler that waits on I/O +// (modelled as a 1ms sleep) and one that burns CPU (hashing 256 KiB). The +// sequential variant caps the request at one invocation at a time, which is +// how every request executed before the invocations of a request ran +// concurrently. +func BenchmarkBatch(b *testing.B) { + const n = 100 + service, err := ed25519.GenerateIssuer() + if err != nil { + b.Fatal(err) + } + alice, err := ed25519.GenerateIssuer() + if err != nil { + b.Fatal(err) + } + + invs := make([]ucan.Invocation, 0, n) + for i := range n { + inv, err := invocation.Invoke(alice, alice.DID(), testutil.TestEchoCommand, datamodel.Map{"i": int64(i)}, invocation.WithAudience(service.DID())) + if err != nil { + b.Fatal(err) + } + invs = append(invs, inv) + } + + payload := make([]byte, 256<<10) + handlers := []struct { + name string + handler execution.HandlerFunc + }{ + {"io", func(req execution.Request, res execution.Response) error { + time.Sleep(time.Millisecond) + return res.SetSuccess(datamodel.Map{}) + }}, + {"cpu", func(req execution.Request, res execution.Response) error { + sum := sha256.Sum256(payload) + return res.SetSuccess(datamodel.Map{"sum": sum[:]}) + }}, + } + + for _, h := range handlers { + name, handler := h.name, h.handler + srv := server.NewHTTP(service) + srv.Handle(testutil.TestEchoCommand, handler) + + // A cap of one is how every request executed before invocations ran + // concurrently. + sequential := server.NewHTTP(service, server.WithMaxConcurrency(1)) + sequential.Handle(testutil.TestEchoCommand, handler) + + b.Run(name+"/1x100/sequential", func(b *testing.B) { + for b.Loop() { + if _, err := sequential.ExecuteBatch(batch.NewRequest(b.Context(), invs)); err != nil { + b.Fatal(err) + } + } + }) + + b.Run(name+"/1x100", func(b *testing.B) { + for b.Loop() { + if _, err := srv.ExecuteBatch(batch.NewRequest(b.Context(), invs)); err != nil { + b.Fatal(err) + } + } + }) + + b.Run(name+"/100x1", func(b *testing.B) { + for b.Loop() { + var wg sync.WaitGroup + for _, inv := range invs { + wg.Add(1) + go func() { + defer wg.Done() + if _, err := srv.ExecuteBatch(batch.NewRequest(b.Context(), []ucan.Invocation{inv})); err != nil { + b.Error(err) + } + }() + } + wg.Wait() + } + }) + } +} diff --git a/server/concurrency_test.go b/server/concurrency_test.go new file mode 100644 index 0000000..a841144 --- /dev/null +++ b/server/concurrency_test.go @@ -0,0 +1,165 @@ +package server_test + +import ( + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/fil-forge/ucantone/execution" + "github.com/fil-forge/ucantone/execution/batch" + "github.com/fil-forge/ucantone/ipld/datamodel" + "github.com/fil-forge/ucantone/server" + "github.com/fil-forge/ucantone/testutil" + "github.com/fil-forge/ucantone/ucan" + "github.com/fil-forge/ucantone/ucan/container" + "github.com/fil-forge/ucantone/ucan/invocation" + "github.com/stretchr/testify/require" +) + +// barrierHandler blocks every invocation until n of them have entered, so a +// batch of n completes only when they all run at the same time. Under +// sequential execution or a cap below n the first one times out instead. +func barrierHandler(n int) execution.HandlerFunc { + var entered atomic.Int32 + release := make(chan struct{}) + return func(req execution.Request, res execution.Response) error { + if entered.Add(1) == int32(n) { + close(release) + } + select { + case <-release: + return res.SetSuccess(datamodel.Map{}) + case <-time.After(5 * time.Second): + return fmt.Errorf("timed out waiting for %d concurrent invocations", n) + } + } +} + +func echoBatch(t *testing.T, alice, service ucan.Issuer, n int) []ucan.Invocation { + invs := make([]ucan.Invocation, 0, n) + for i := range n { + inv, err := invocation.Invoke(alice, alice.DID(), testutil.TestEchoCommand, datamodel.Map{"i": int64(i)}, invocation.WithAudience(service.DID())) + require.NoError(t, err) + invs = append(invs, inv) + } + return invs +} + +func requireAllOK(t *testing.T, res *batch.Response, invs []ucan.Invocation) { + failures := map[string]any{} + for _, inv := range invs { + rcpt, ok := res.Receipt(inv.Task().Link()) + require.True(t, ok, "missing receipt for %s", inv.Task().Link()) + if _, x := rcpt.Out().Unpack(); x != nil { + failures[inv.Task().Link().String()] = testutil.ResultMap(t, x)["message"] + } + } + require.Empty(t, failures) +} + +func TestHTTPServerConcurrency(t *testing.T) { + service := testutil.RandomIssuer(t) + alice := testutil.RandomIssuer(t) + + t.Run("invocations of one request execute concurrently", func(t *testing.T) { + const n = 10 + srv := server.NewHTTP(service) + srv.Handle(testutil.TestEchoCommand, barrierHandler(n)) + invs := echoBatch(t, alice, service, n) + + res, err := srv.ExecuteBatch(batch.NewRequest(t.Context(), invs)) + require.NoError(t, err) + requireAllOK(t, res, invs) + }) + + t.Run("WithMaxConcurrency(0) removes the cap", func(t *testing.T) { + n := server.DefaultMaxConcurrency + 50 + srv := server.NewHTTP(service, server.WithMaxConcurrency(0)) + srv.Handle(testutil.TestEchoCommand, barrierHandler(n)) + invs := echoBatch(t, alice, service, n) + + res, err := srv.ExecuteBatch(batch.NewRequest(t.Context(), invs)) + require.NoError(t, err) + requireAllOK(t, res, invs) + }) + + t.Run("WithMaxConcurrency caps how many run at once", func(t *testing.T) { + const cap, n = 2, 5 + srv := server.NewHTTP(service, server.WithMaxConcurrency(cap)) + + // Each handler waits until cap invocations are in flight before it + // finishes, so the peak reaches cap by construction, and the + // semaphore keeps it from going past. + var inFlight, peak atomic.Int32 + full := make(chan struct{}) + var closeFull sync.Once + srv.Handle(testutil.TestEchoCommand, func(req execution.Request, res execution.Response) error { + now := inFlight.Add(1) + defer inFlight.Add(-1) + for { + seen := peak.Load() + if now <= seen || peak.CompareAndSwap(seen, now) { + break + } + } + if now == cap { + closeFull.Do(func() { close(full) }) + } + select { + case <-full: + return res.SetSuccess(datamodel.Map{}) + case <-time.After(5 * time.Second): + return fmt.Errorf("timed out waiting for %d in-flight invocations", cap) + } + }) + invs := echoBatch(t, alice, service, n) + + res, err := srv.ExecuteBatch(batch.NewRequest(t.Context(), invs)) + require.NoError(t, err) + requireAllOK(t, res, invs) + require.Equal(t, int32(cap), peak.Load()) + }) + + t.Run("WithMaxConcurrency panics on a negative value", func(t *testing.T) { + require.PanicsWithValue(t, "server.WithMaxConcurrency: n must not be negative", func() { + server.WithMaxConcurrency(-1) + }) + }) + + t.Run("response metadata stays in request order", func(t *testing.T) { + const n = 20 + srv := server.NewHTTP(service) + // Later invocations finish first, so metadata gathered in completion + // order would come back reversed. + srv.Handle(testutil.TestEchoCommand, func(req execution.Request, res execution.Response) error { + i := testutil.ArgsMap(t, req.Invocation())["i"].(int64) + time.Sleep(time.Duration(n-i) * time.Millisecond) + claim, err := invocation.Invoke(service, service.DID(), testutil.ConsoleLogCommand, datamodel.Map{"i": i}) + if err != nil { + return err + } + if err := res.SetMetadata(container.New(container.WithInvocations(claim))); err != nil { + return err + } + return res.SetSuccess(datamodel.Map{}) + }) + invs := echoBatch(t, alice, service, n) + + res, err := srv.ExecuteBatch(batch.NewRequest(t.Context(), invs)) + require.NoError(t, err) + requireAllOK(t, res, invs) + + require.NotNil(t, res.Metadata()) + var got []int64 + for _, claim := range res.Metadata().Invocations() { + got = append(got, testutil.ArgsMap(t, claim)["i"].(int64)) + } + want := make([]int64, n) + for i := range want { + want[i] = int64(i) + } + require.Equal(t, want, got) + }) +} diff --git a/server/http.go b/server/http.go index 0f8ccfd..9032bdf 100644 --- a/server/http.go +++ b/server/http.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "net/http" + "sync" "github.com/fil-forge/ucantone/execution" "github.com/fil-forge/ucantone/execution/batch" @@ -16,10 +17,11 @@ import ( ) type HTTPServer struct { - id ucan.Issuer - executor *dispatcher.Dispatcher - codec transport.InboundCodec[*http.Request, *http.Response] - listeners []EventListener + id ucan.Issuer + executor *dispatcher.Dispatcher + codec transport.InboundCodec[*http.Request, *http.Response] + listeners []EventListener + maxConcurrency int } var ( @@ -30,7 +32,8 @@ var ( // NewHTTP creates a new server capable of handling UCAN invocations over HTTP. func NewHTTP(id ucan.Issuer, options ...HTTPOption) *HTTPServer { cfg := httpServerConfig{ - codec: transport.DefaultHTTPInboundCodec, + codec: transport.DefaultHTTPInboundCodec, + maxConcurrency: DefaultMaxConcurrency, } for _, opt := range options { opt(&cfg) @@ -44,10 +47,11 @@ func NewHTTP(id ucan.Issuer, options ...HTTPOption) *HTTPServer { } executor := dispatcher.New(id, dispatcherOpts...) return &HTTPServer{ - id: id, - codec: cfg.codec, - executor: executor, - listeners: cfg.listeners, + id: id, + codec: cfg.codec, + executor: executor, + listeners: cfg.listeners, + maxConcurrency: cfg.maxConcurrency, } } @@ -85,17 +89,28 @@ func (s *HTTPServer) Execute(req execution.Request) (execution.Response, error) // skipped and get no receipt. Each handler sees every token of // the request as its metadata, and the tokens handlers attach to their // responses are gathered into the metadata of the returned response. +// +// Invocations execute concurrently, each on its own goroutine, with at most +// [WithMaxConcurrency] of them running at once. Handlers must therefore be +// safe to call concurrently within one request, as they already must be across +// requests. Receipts and metadata are gathered in request order once every +// invocation has finished, so the response does not depend on which handler +// finished first. func (s *HTTPServer) ExecuteBatch(req *batch.Request) (*batch.Response, error) { // Every handler sees the same tokens, so the container is built once, on // the first invocation addressed to this server, and shared by every // handler in the batch. var execMeta ucan.Container - var receipts []ucan.Receipt - var invocations []ucan.Invocation - var delegations []ucan.Delegation - var extraReceipts []ucan.Receipt - for _, inv := range req.Invocations() { + // Each goroutine writes only its own slot, so the slice needs no lock and + // the merge below runs in request order. + results := make([]batchResult, len(req.Invocations())) + var wg sync.WaitGroup + var slots chan struct{} + if s.maxConcurrency > 0 { + slots = make(chan struct{}, s.maxConcurrency) + } + for i, inv := range req.Invocations() { aud := inv.Audience() if !aud.Defined() { aud = inv.Subject() @@ -106,22 +121,50 @@ func (s *HTTPServer) ExecuteBatch(req *batch.Request) (*batch.Response, error) { if execMeta == nil { execMeta = batchMetadata(req) } - execReq := execution.NewRequest(req.Context(), inv, execution.WithRequestMetadata(execMeta)) + // Acquire before spawning, so a request larger than the cap waits in + // this loop instead of as a pile of blocked goroutines. + if slots != nil { + slots <- struct{}{} + } + wg.Add(1) + go func() { + defer wg.Done() + if slots != nil { + defer func() { <-slots }() + } + execReq := execution.NewRequest(req.Context(), inv, execution.WithRequestMetadata(execMeta)) + res, err := s.executor.Execute(execReq) + results[i] = batchResult{executed: true, response: res, err: err} + }() + } + wg.Wait() - res, err := s.executor.Execute(execReq) - if err != nil { + var errs error + var receipts []ucan.Receipt + var invocations []ucan.Invocation + var delegations []ucan.Delegation + var extraReceipts []ucan.Receipt + for i, inv := range req.Invocations() { + result := results[i] + if !result.executed { + continue + } + if result.err != nil { // This shouldn't really happen, executor only returns an error when // result or metadata cannot be set, which is likely a developer error. - return nil, fmt.Errorf("executing task %s: %w", inv.Task().Link(), err) + errs = errors.Join(errs, fmt.Errorf("executing task %s: %w", inv.Task().Link(), result.err)) + continue } - - receipts = append(receipts, res.Receipt()) - if res.Metadata() != nil { - invocations = append(invocations, res.Metadata().Invocations()...) - delegations = append(delegations, res.Metadata().Delegations()...) - extraReceipts = append(extraReceipts, res.Metadata().Receipts()...) + receipts = append(receipts, result.response.Receipt()) + if result.response.Metadata() != nil { + invocations = append(invocations, result.response.Metadata().Invocations()...) + delegations = append(delegations, result.response.Metadata().Delegations()...) + extraReceipts = append(extraReceipts, result.response.Metadata().Receipts()...) } } + if errs != nil { + return nil, errs + } var options []batch.ResponseOption if len(invocations) > 0 || len(delegations) > 0 || len(extraReceipts) > 0 { @@ -147,6 +190,14 @@ func batchMetadata(req *batch.Request) ucan.Container { return container.New(options...) } +// batchResult is the outcome of executing one invocation of a batch. executed +// is false for invocations addressed elsewhere, which get no receipt. +type batchResult struct { + executed bool + response execution.Response + err error +} + func (s *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { resp, err := s.RoundTrip(r) if err != nil { diff --git a/server/http_test.go b/server/http_test.go index be6ec28..8a23559 100644 --- a/server/http_test.go +++ b/server/http_test.go @@ -4,6 +4,7 @@ import ( "context" "io" "net/http" + "sync" "testing" "time" @@ -178,9 +179,12 @@ func TestHTTPServerBatch(t *testing.T) { t.Run("batch handlers share one metadata container", func(t *testing.T) { server := server.NewHTTP(service) + var mu sync.Mutex var seen []ucan.Container server.Handle(testutil.TestEchoCommand, func(req execution.Request, res execution.Response) error { + mu.Lock() seen = append(seen, req.Metadata()) + mu.Unlock() return res.SetSuccess(datamodel.Map{}) }) diff --git a/server/options.go b/server/options.go index 0d2bf47..9a30acb 100644 --- a/server/options.go +++ b/server/options.go @@ -17,8 +17,17 @@ type httpServerConfig struct { receiptTimestamps bool listeners []EventListener panicLogger dispatcher.PanicLogger + maxConcurrency int } +// DefaultMaxConcurrency is how many invocations of one request execute at the +// same time unless [WithMaxConcurrency] says otherwise. It is the value +// quic-go uses for MaxIncomingStreams and the floor RFC 9113 recommends for +// SETTINGS_MAX_CONCURRENT_STREAMS; x/net/http2 defaults to 250. Nobody +// documents a derivation for these numbers: they bound how much one peer can +// have in flight while leaving batches of everyday size unthrottled. +const DefaultMaxConcurrency = 100 + func WithHTTPCodec(codec transport.InboundCodec[*http.Request, *http.Response]) HTTPOption { return func(cfg *httpServerConfig) { cfg.codec = codec @@ -39,6 +48,21 @@ func WithReceiptTimestamps(enabled bool) HTTPOption { } } +// WithMaxConcurrency caps how many invocations of one request execute at the +// same time. Every invocation still gets its own goroutine; the cap is a +// per-request semaphore that limits how many are admitted at once, the way an +// HTTP/2 or QUIC server caps concurrent streams per connection. Zero removes +// the cap, for deployments that bound concurrency at the listener instead. +// A negative value panics. Defaults to [DefaultMaxConcurrency]. +func WithMaxConcurrency(n int) HTTPOption { + if n < 0 { + panic("server.WithMaxConcurrency: n must not be negative") + } + return func(cfg *httpServerConfig) { + cfg.maxConcurrency = n + } +} + // WithPanicLogger sets the function the server's dispatcher records recovered // panics with, from handlers and from validation code alike. See // [dispatcher.WithPanicLogger] for the default and the contract. A nil logger From 1a6b363d02c83845f36a91370dddb4abbee4691b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20Bajto=C5=A1?= Date: Tue, 22 Sep 2026 09:42:19 +0200 Subject: [PATCH 2/5] test: derive batch benchmark names from the size MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sub-benchmark names spelled out the batch size, so measuring a different size meant editing four lines instead of one constant. Assisted-by: Claude:claude-fable-5-1 Signed-off-by: Miroslav Bajtoš --- server/bench_test.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/server/bench_test.go b/server/bench_test.go index a2b3be3..df09a8e 100644 --- a/server/bench_test.go +++ b/server/bench_test.go @@ -2,6 +2,7 @@ package server_test import ( "crypto/sha256" + "fmt" "sync" "testing" "time" @@ -67,7 +68,7 @@ func BenchmarkBatch(b *testing.B) { sequential := server.NewHTTP(service, server.WithMaxConcurrency(1)) sequential.Handle(testutil.TestEchoCommand, handler) - b.Run(name+"/1x100/sequential", func(b *testing.B) { + b.Run(fmt.Sprintf("%s/1x%d/sequential", name, n), func(b *testing.B) { for b.Loop() { if _, err := sequential.ExecuteBatch(batch.NewRequest(b.Context(), invs)); err != nil { b.Fatal(err) @@ -75,7 +76,7 @@ func BenchmarkBatch(b *testing.B) { } }) - b.Run(name+"/1x100", func(b *testing.B) { + b.Run(fmt.Sprintf("%s/1x%d", name, n), func(b *testing.B) { for b.Loop() { if _, err := srv.ExecuteBatch(batch.NewRequest(b.Context(), invs)); err != nil { b.Fatal(err) @@ -83,7 +84,7 @@ func BenchmarkBatch(b *testing.B) { } }) - b.Run(name+"/100x1", func(b *testing.B) { + b.Run(fmt.Sprintf("%s/%dx1", name, n), func(b *testing.B) { for b.Loop() { var wg sync.WaitGroup for _, inv := range invs { From 340af77ac7b14698967f82526b91913c69be4d12 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20Bajto=C5=A1?= Date: Tue, 22 Sep 2026 09:54:44 +0200 Subject: [PATCH 3/5] docs: clarify the scope and settings of WithMaxConcurrency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Say that the cap is per request rather than per server, how to force serial execution with a cap of one, and drop the sentence about the default's derivation that read as confusing. Addresses review comments on #66. Assisted-by: Claude:claude-fable-5-1 Signed-off-by: Miroslav Bajtoš --- server/http.go | 12 +++++++----- server/options.go | 14 ++++++++------ 2 files changed, 15 insertions(+), 11 deletions(-) diff --git a/server/http.go b/server/http.go index 9032bdf..51cd153 100644 --- a/server/http.go +++ b/server/http.go @@ -91,11 +91,13 @@ func (s *HTTPServer) Execute(req execution.Request) (execution.Response, error) // responses are gathered into the metadata of the returned response. // // Invocations execute concurrently, each on its own goroutine, with at most -// [WithMaxConcurrency] of them running at once. Handlers must therefore be -// safe to call concurrently within one request, as they already must be across -// requests. Receipts and metadata are gathered in request order once every -// invocation has finished, so the response does not depend on which handler -// finished first. +// [WithMaxConcurrency] of them running at once. That cap applies to this +// request alone: concurrent requests each get their own, so the server as a +// whole runs up to the cap times the number of requests in flight. Handlers +// must therefore be safe to call concurrently within one request, as they +// already must be across requests. Receipts and metadata are gathered in +// request order once every invocation has finished, so the response does not +// depend on which handler finished first. func (s *HTTPServer) ExecuteBatch(req *batch.Request) (*batch.Response, error) { // Every handler sees the same tokens, so the container is built once, on // the first invocation addressed to this server, and shared by every diff --git a/server/options.go b/server/options.go index 9a30acb..bc0ba85 100644 --- a/server/options.go +++ b/server/options.go @@ -23,9 +23,7 @@ type httpServerConfig struct { // DefaultMaxConcurrency is how many invocations of one request execute at the // same time unless [WithMaxConcurrency] says otherwise. It is the value // quic-go uses for MaxIncomingStreams and the floor RFC 9113 recommends for -// SETTINGS_MAX_CONCURRENT_STREAMS; x/net/http2 defaults to 250. Nobody -// documents a derivation for these numbers: they bound how much one peer can -// have in flight while leaving batches of everyday size unthrottled. +// SETTINGS_MAX_CONCURRENT_STREAMS; x/net/http2 defaults to 250. const DefaultMaxConcurrency = 100 func WithHTTPCodec(codec transport.InboundCodec[*http.Request, *http.Response]) HTTPOption { @@ -51,9 +49,13 @@ func WithReceiptTimestamps(enabled bool) HTTPOption { // WithMaxConcurrency caps how many invocations of one request execute at the // same time. Every invocation still gets its own goroutine; the cap is a // per-request semaphore that limits how many are admitted at once, the way an -// HTTP/2 or QUIC server caps concurrent streams per connection. Zero removes -// the cap, for deployments that bound concurrency at the listener instead. -// A negative value panics. Defaults to [DefaultMaxConcurrency]. +// HTTP/2 or QUIC server caps concurrent streams per connection. It does not +// limit the server as a whole: each request in flight gets its own cap. +// +// One runs the invocations of a request one after another, for handlers that +// are not safe to call concurrently within a request. Zero removes the cap, +// for deployments that bound concurrency at the listener instead. A negative +// value panics. Defaults to [DefaultMaxConcurrency]. func WithMaxConcurrency(n int) HTTPOption { if n < 0 { panic("server.WithMaxConcurrency: n must not be negative") From 4fd6eb5bcc6c8c0177931c620e4a929c90802a68 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20Bajto=C5=A1?= Date: Tue, 22 Sep 2026 09:58:48 +0200 Subject: [PATCH 4/5] improve code comment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Miroslav Bajtoš --- server/options.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/server/options.go b/server/options.go index bc0ba85..ab72093 100644 --- a/server/options.go +++ b/server/options.go @@ -52,10 +52,11 @@ func WithReceiptTimestamps(enabled bool) HTTPOption { // HTTP/2 or QUIC server caps concurrent streams per connection. It does not // limit the server as a whole: each request in flight gets its own cap. // -// One runs the invocations of a request one after another, for handlers that -// are not safe to call concurrently within a request. Zero removes the cap, -// for deployments that bound concurrency at the listener instead. A negative -// value panics. Defaults to [DefaultMaxConcurrency]. +// - WithMaxConcurrency(1) runs the invocations of a request serially, one after another. +// - WithMaxConcurrency(0) removes the cap. +// - A negative value panics. +// +// Defaults to [DefaultMaxConcurrency]. func WithMaxConcurrency(n int) HTTPOption { if n < 0 { panic("server.WithMaxConcurrency: n must not be negative") From a23cedfb7a259e4c15ceb53540c932a390196b4f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20Bajto=C5=A1?= Date: Tue, 22 Sep 2026 16:53:25 +0200 Subject: [PATCH 5/5] docs: name the default cap instead of the option MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The README and the ExecuteBatch comment read "up to WithMaxConcurrency at a time", as if the option were a number. Name DefaultMaxConcurrency and say the option changes it. While here, replace the manual Add/Done pair with WaitGroup.Go. Assisted-by: Claude:claude-fable-5-1 Signed-off-by: Miroslav Bajtoš --- README.md | 5 +++-- server/http.go | 13 ++++++------- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index d958bf0..649c38f 100644 --- a/README.md +++ b/README.md @@ -180,8 +180,9 @@ if out := resp.Receipt().Out(); out.IsOK() { ``` Many invocations can be sent in one request. The service executes every -invocation addressed to it concurrently, up to `server.WithMaxConcurrency` -at a time (100 by default), and answers each with its own receipt: +invocation addressed to it concurrently, up to `server.DefaultMaxConcurrency` +(100) at a time unless `server.WithMaxConcurrency` sets another cap, and +answers each with its own receipt: ```go res, err := c.ExecuteBatch(batch.NewRequest(context.Background(), invs, batch.WithDelegations(dlg))) diff --git a/server/http.go b/server/http.go index 51cd153..7aaf559 100644 --- a/server/http.go +++ b/server/http.go @@ -91,9 +91,10 @@ func (s *HTTPServer) Execute(req execution.Request) (execution.Response, error) // responses are gathered into the metadata of the returned response. // // Invocations execute concurrently, each on its own goroutine, with at most -// [WithMaxConcurrency] of them running at once. That cap applies to this -// request alone: concurrent requests each get their own, so the server as a -// whole runs up to the cap times the number of requests in flight. Handlers +// [DefaultMaxConcurrency] of them running at once unless [WithMaxConcurrency] +// sets another cap. The cap applies to this request alone: concurrent +// requests each get their own, so the server as a whole runs up to the cap +// times the number of requests in flight. Handlers // must therefore be safe to call concurrently within one request, as they // already must be across requests. Receipts and metadata are gathered in // request order once every invocation has finished, so the response does not @@ -128,16 +129,14 @@ func (s *HTTPServer) ExecuteBatch(req *batch.Request) (*batch.Response, error) { if slots != nil { slots <- struct{}{} } - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { if slots != nil { defer func() { <-slots }() } execReq := execution.NewRequest(req.Context(), inv, execution.WithRequestMetadata(execMeta)) res, err := s.executor.Execute(execReq) results[i] = batchResult{executed: true, response: res, err: err} - }() + }) } wg.Wait()