feat: execute a request's invocations concurrently

README.md

@@ -180,7 +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 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)))

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.

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

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.

server/bench_test.go

@@ -0,0 +1,103 @@
+package server_test
+
+import (
+	"crypto/sha256"
+	"fmt"
+	"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(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)
+				}
+			}
+		})
+
+		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)
+				}
+			}
+		})
+
+		b.Run(fmt.Sprintf("%s/%dx1", name, n), 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()
+			}
+		})
+	}
+}

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)
+	})
+}