test(lambda): sample heap only across the measured lifetimes and join the sampler

The sampler now starts after the post-GC baseline and stops after both
replies are read, and cleanup joins it if the test fails first.

Makefile

@@ -90,6 +90,7 @@ go_build:
 LAMBDA_TESTLINK_DIR := ./bin/lambda-testlink
 LAMBDA_TESTLINK_PKG := github.com/mulgadc/spinifex/spinifex/runtime/daemon
 LAMBDA_TESTLINK_PIN := $(LAMBDA_TESTLINK_PKG).lambdaHardKillSHA256
+LAMBDA_TESTLINK_GW := github.com/mulgadc/spinifex/spinifex/runtime/roles/awsgw
 LAMBDA_DIGEST_RE := ^[0-9a-f]{64}$$
 build-lambda-testlink: override export LAMBDA_IMAGE_DIGEST := $(value LAMBDA_IMAGE_DIGEST)
 build-lambda-testlink:
@@ -104,8 +105,8 @@ build-lambda-testlink:
 	@digest=$$(sha256sum $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so | cut -d' ' -f1); \
 	test $${#digest} -eq 64 || { echo "initializer digest missing"; exit 1; }; \
 	echo "initializer sha256 $$digest"; \
-	pin="-X $(LAMBDA_TESTLINK_PIN)=$$digest -X $(LAMBDA_TESTLINK_PKG).lambdaImageDigest=$$LAMBDA_IMAGE_DIGEST"; \
-	LAMBDA_TESTLINK_INITIALIZER=$(abspath $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so) go test -count=1 -tags lambdatestlink -ldflags "$$pin" -run '^TestLambda(HardKillPin|ImageDigestPin)$$' ./spinifex/runtime/daemon/ && \
+	pin="-X $(LAMBDA_TESTLINK_PIN)=$$digest -X $(LAMBDA_TESTLINK_PKG).lambdaImageDigest=$$LAMBDA_IMAGE_DIGEST -X $(LAMBDA_TESTLINK_GW).lambdaImageDigest=$$LAMBDA_IMAGE_DIGEST"; \
+	LAMBDA_TESTLINK_INITIALIZER=$(abspath $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so) go test -count=1 -tags lambdatestlink -ldflags "$$pin" -run '^TestLambda(HardKillPin|ImageDigestPin)$$' ./spinifex/runtime/daemon/ ./spinifex/runtime/roles/awsgw/ && \
 	go build $(GO_BUILD_MOD) -tags lambdatestlink -ldflags "$(subst -s -w ,,$(LDFLAGS)) $$pin" -o $(LAMBDA_TESTLINK_DIR).tmp/$(GO_PROJECT_NAME) cmd/spinifex/main.go
 	@mkdir -p $(LAMBDA_TESTLINK_DIR)
 	mv $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so $(LAMBDA_TESTLINK_DIR).tmp/$(GO_PROJECT_NAME) $(LAMBDA_TESTLINK_DIR)/

contracts/lambda/v1/README.md

@@ -22,6 +22,11 @@ Tenant code may be able to reach the listener, but it cannot open the agent's co
 For each generation the executor creates ephemeral self-signed host and guest certificates whose single URI SAN is `spinifex-lambda://<Region>/<account>/<function ID>/<environment ID>/<generation>/<role>`, with role `host` or `guest`.
 Each end validates the peer certificate against the one pinned peer certificate and separately requires that URI SAN with the opposite role; hostname verification is not relied on.
 ALPN carries the L-GUEST major version, `spinifex-lambda-guest/1`, and an unsupported version fails the handshake. Builds with the `lambdatestlink` tag offer only `spinifex-lambda-testlink/2` on both host and guest, whose first frame carries the execution-role credential, so a test-link end and a release end fail the handshake; release builds are unchanged.
+In the same builds, the gateway reaches the host daemon on the local abstract socket `spinifex-lambda-testlink-control/3`, and each request carries the runtime-image digest the gateway was built with; the daemon refuses a missing or different digest before starting the attempt, and a gateway and daemon on different socket names never connect.
+On that socket a request's raw payload is at most 6 MiB, checked by the gateway before encoding and by the daemon after decoding.
+Its encoded frame is at most the base64 length of 6 MiB plus 64 KiB of metadata headroom and 64 bytes of field slack, and a receiver checks the frame length before allocating; metadata has no separate bound within the frame.
+The credential travels as its own frame of at most 8 KiB, and the reply has the same payload and frame bounds as the request.
+The daemon holds at most two connections at once, from accept to close, and leaves the rest unaccepted in the listen backlog.
 Under TLS 1.3 the guest's handshake can complete before the host refuses its certificate, so the guest treats the connection as accepted only after the host's first message.
 
 **Bootstrap material:** the executor passes these fw_cfg entries under `opt/spinifex/lambda/`, and the agent reads them as root and unloads `qemu_fw_cfg` before user code starts.

spinifex/agents/lambda/hostlink/test_stream.go

@@ -321,6 +321,19 @@ func (s *TestStream) HostSend(req Request) error {
 	return s.write(frame{Type: typeInvoke, Request: &req})
 }
 
+// CheckInvokeFits encodes req as HostSend would and returns ErrFrameTooLarge
+// when the frame exceeds the bound, so a caller can refuse before launch.
+func CheckInvokeFits(req Request) error {
+	b, err := json.Marshal(frame{Type: typeInvoke, Request: &req})
+	if err != nil {
+		return err
+	}
+	if len(b) > maxFrame {
+		return ErrFrameTooLarge
+	}
+	return nil
+}
+
 // HostShutdown writes a shutdown frame, as the host side of the carrier.
 func (s *TestStream) HostShutdown() error {
 	return s.write(frame{Type: typeShutdown})

spinifex/runtime/compute/lambda/testlink/contract.go

@@ -49,6 +49,8 @@ type Request struct {
 	CodeURL     string          `json:"code_url"`
 	CodeSize    int64           `json:"code_size"`
 	CodeSHA256  string          `json:"code_sha256"`
+	// ImageDigest is the runtime image the gateway expects to run the code.
+	ImageDigest string `json:"image_digest"`
 	Payload     []byte          `json:"payload"`
 
 	// cred arrives beside the request on the control socket, never inside it.

spinifex/runtime/compute/lambda/testlink/control.go

@@ -23,7 +23,7 @@ import (
 // ControlName is the abstract Unix socket name the node's daemon serves. An
 // abstract name has no directory entry; the kernel frees it when the last
 // reference to the listening socket closes.
-const ControlName = "spinifex-lambda-testlink-control"
+const ControlName = "spinifex-lambda-testlink-control/3"
 
 // Service accounts at each end of the control socket.
 const (
@@ -31,10 +31,36 @@ const (
 	DaemonUser  = "spinifex-daemon"
 )
 
+// joinAfterFunc arms f on ctx's cancellation and returns a stop that, unlike
+// context.AfterFunc's, waits for f when it has already started.
+func joinAfterFunc(ctx context.Context, f func()) func() {
+	done := make(chan struct{})
+	stop := context.AfterFunc(ctx, func() { defer close(done); f() })
+	return func() {
+		if !stop() {
+			<-done
+		}
+	}
+}
+
 // MaxCredentialBytes bounds the encoded credential; a session's three values
 // fit well inside it.
 const MaxCredentialBytes = 8 << 10
 
+// Declared control-socket bounds, independent of any message bus. Request and
+// Reply payloads are capped raw; the frame cap leaves MetadataHeadroomBytes
+// beside a full base64 payload, and metadata has no bound of its own.
+const (
+	MaxPayloadBytes       = 6 << 20
+	MetadataHeadroomBytes = 64 << 10
+	// payloadFieldSlack covers the payload key, quotes and comma.
+	payloadFieldSlack = 64
+	MaxFrameBytes     = int64((MaxPayloadBytes+2)/3*4 + MetadataHeadroomBytes + payloadFieldSlack)
+	// MaxAdmitted bounds connections the daemon holds at once, from accept
+	// to close; the daemon runs one attempt, so one more is answered busy.
+	MaxAdmitted = 2
+)
+
 // Credential is one execution-role session for one attempt; it goes to the
 // guest in the first frame and never on NATS, to disk or into a log.
 type Credential = hostlink.Credential
@@ -118,8 +144,8 @@ func readFrame(r io.Reader, limit int64) ([]byte, error) {
 // ControlServer answers one request per connection from the gateway's UID.
 type ControlServer struct {
 	GatewayUID int
-	// RequestLimit bounds the encoded Request alone, as the NATS path did, and
-	// ReplyLimit the encoded Reply; a reply over it becomes unknown.
+	// RequestLimit bounds the encoded Request frame and ReplyLimit the
+	// encoded Reply; a reply over it, or over MaxPayloadBytes, becomes unknown.
 	RequestLimit int64
 	ReplyLimit   int64
 	// IOTimeout bounds reading the request and writing the reply.
@@ -129,14 +155,25 @@ type ControlServer struct {
 
 // Serve accepts until ctx ends or the listener fails, then waits for every
 // connection it started.
+// At most MaxAdmitted connections are held at once, each from accept to
+// close; others wait unaccepted in the kernel's listen backlog.
 func (s *ControlServer) Serve(ctx context.Context, ln *net.UnixListener) error {
-	stop := context.AfterFunc(ctx, func() { _ = ln.Close() })
-	defer stop()
+	// Serve may return before the callback runs, so it closes the listener
+	// too, and joins a callback already started before returning.
+	defer joinAfterFunc(ctx, func() { _ = ln.Close() })()
+	defer func() { _ = ln.Close() }()
 	var wg sync.WaitGroup
 	defer wg.Wait()
+	slots := make(chan struct{}, MaxAdmitted)
 	for {
+		select {
+		case slots <- struct{}{}:
+		case <-ctx.Done():
+			return nil
+		}
 		c, err := ln.AcceptUnix()
 		if err != nil {
+			<-slots
 			if ctx.Err() != nil {
 				return nil
 			}
@@ -145,6 +182,10 @@ func (s *ControlServer) Serve(ctx context.Context, ln *net.UnixListener) error {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			defer func() { <-slots }()
+			// Closing on cancellation unblocks a read or write in progress; the
+			// close is joined before the slot is released.
+			defer joinAfterFunc(ctx, func() { _ = c.Close() })()
 			s.serveConn(ctx, c)
 		}()
 	}
@@ -176,6 +217,10 @@ func (s *ControlServer) serveConn(ctx context.Context, c *net.UnixConn) {
 	_ = c.SetReadDeadline(time.Time{})
 
 	rep := s.Handle(ctx, req, cred)
+	// The raw payload is checked before the reply is encoded.
+	if len(rep.Payload) > MaxPayloadBytes {
+		rep = Reply{Outcome: OutcomeUnknown, Error: "result too large for the test path"}
+	}
 	body, err := json.Marshal(rep)
 	if err == nil && int64(len(body)) > s.ReplyLimit {
 		body, err = json.Marshal(Reply{Outcome: OutcomeUnknown, Error: "result too large for the test path"})
@@ -218,8 +263,11 @@ func (c *ControlClient) Exchange(ctx context.Context, req []byte, cred Credentia
 	if replyBy.IsZero() || !replyBy.After(time.Now()) {
 		return Reply{}, fmt.Errorf("%w: reply bound is zero or already past", ErrNotSent)
 	}
+	// The dial stops at replyBy even when ctx has no deadline.
+	dialCtx, cancelDial := context.WithDeadline(ctx, replyBy)
 	var d net.Dialer
-	conn, err := d.DialContext(ctx, "unix", "@"+c.Name)
+	conn, err := d.DialContext(dialCtx, "unix", "@"+c.Name)
+	cancelDial()
 	if err != nil {
 		return Reply{}, fmt.Errorf("%w: dial test-link control: %w", ErrNotSent, err)
 	}
@@ -269,6 +317,9 @@ func (c *ControlClient) Exchange(ctx context.Context, req []byte, cred Credentia
 	if err := json.Unmarshal(body, &rep); err != nil {
 		return Reply{}, errors.New("unreadable test-link reply")
 	}
+	if len(rep.Payload) > MaxPayloadBytes {
+		return Reply{}, fmt.Errorf("test-link reply payload of %d bytes exceeds %d", len(rep.Payload), MaxPayloadBytes)
+	}
 	return rep, nil
 }
 

spinifex/runtime/compute/lambda/testlink/control_bounds_test.go

@@ -0,0 +1,569 @@
+//go:build lambdatestlink
+
+//test:in-package — reads frames with the unexported readFrame and drives ControlServer limits directly
+package testlink
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"net"
+	"os"
+	"runtime"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+)
+
+// fullServe runs a ControlServer at the declared bounds on ln and returns a
+// cancel that waits for Serve.
+func fullServe(t *testing.T, ln *net.UnixListener, handle func(context.Context, []byte, Credential) Reply) func() {
+	t.Helper()
+	srv := &ControlServer{GatewayUID: os.Getuid(), RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes, IOTimeout: 60 * time.Second, Handle: handle}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- srv.Serve(ctx, ln) }()
+	var once sync.Once
+	stop := func() {
+		once.Do(func() {
+			cancel()
+			select {
+			case err := <-done:
+				require.NoError(t, err, "Serve returned an error on shutdown")
+			case <-time.After(20 * time.Second):
+				t.Fatal("Serve did not return")
+			}
+		})
+	}
+	t.Cleanup(stop)
+	return stop
+}
+
+func fullListen(t *testing.T, name string) *net.UnixListener {
+	t.Helper()
+	ln, err := ListenControl(name)
+	require.NoError(t, err)
+	return ln
+}
+
+func fullClient(name string) *ControlClient {
+	return &ControlClient{Name: name, DaemonUID: os.Getuid(), RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes, IOTimeout: 60 * time.Second}
+}
+
+// requestOfSize returns an encoded Request with a payload of payloadLen and
+// a FunctionARN padded so the frame is exactly size bytes.
+func requestOfSize(t *testing.T, payloadLen, size int) []byte {
+	t.Helper()
+	req := Request{AttemptID: "0123456789ab", Payload: bytes.Repeat([]byte{7}, payloadLen)}
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+	req.FunctionARN = "a"
+	b, err = json.Marshal(req)
+	require.NoError(t, err)
+	require.LessOrEqual(t, len(b), size)
+	req.FunctionARN = strings.Repeat("a", 1+size-len(b))
+	b, err = json.Marshal(req)
+	require.NoError(t, err)
+	require.Len(t, b, size)
+	return b
+}
+
+// replyOfSize returns a Reply with a payload of payloadLen whose encoding is
+// exactly size bytes.
+func replyOfSize(t *testing.T, payloadLen, size int) Reply {
+	t.Helper()
+	rep := Reply{Outcome: OutcomeDone}
+	if payloadLen > 0 {
+		rep.Payload = bytes.Repeat([]byte{9}, payloadLen)
+	}
+	b, err := json.Marshal(rep)
+	require.NoError(t, err)
+	rep.Error = "e"
+	b, err = json.Marshal(rep)
+	require.NoError(t, err)
+	require.LessOrEqual(t, len(b), size)
+	rep.Error = strings.Repeat("e", 1+size-len(b))
+	b, err = json.Marshal(rep)
+	require.NoError(t, err)
+	require.Len(t, b, size)
+	return rep
+}
+
+func recvBounded[T any](t *testing.T, ch chan T) T {
+	t.Helper()
+	select {
+	case v := <-ch:
+		return v
+	case <-time.After(30 * time.Second):
+		t.Fatal("no value within 30 s")
+		var zero T
+		return zero
+	}
+}
+
+func allocatedDuring(f func()) uint64 {
+	var before, after runtime.MemStats
+	runtime.GC()
+	runtime.ReadMemStats(&before)
+	f()
+	runtime.ReadMemStats(&after)
+	return after.TotalAlloc - before.TotalAlloc
+}
+
+// Test 5: a frame over MaxFrameBytes is rejected on its header, before the
+// receiver allocates its body.
+func TestReadFrame_OverBoundRejectedBeforeAllocation(t *testing.T) {
+	var hdr [4]byte
+	binary.BigEndian.PutUint32(hdr[:], uint32(MaxFrameBytes+1))
+	var err error
+	alloc := allocatedDuring(func() { _, err = readFrame(bytes.NewReader(hdr[:]), MaxFrameBytes) })
+	require.ErrorContains(t, err, "outside 1..")
+	require.Less(t, alloc, uint64(1<<20), "no body buffer was allocated")
+}
+
+// Test 5: a direct peer's frame with a small payload and metadata far over
+// 64 KiB, within MaxFrameBytes, reaches Handle: metadata has no separate bound.
+func TestControl_LargeMetadataWithinTheFrameIsAccepted(t *testing.T) {
+	name := controlName(t)
+	got := make(chan []byte, 1)
+	fullServe(t, fullListen(t, name), func(_ context.Context, req []byte, _ Credential) Reply {
+		got <- req
+		return Reply{Outcome: OutcomeDone}
+	})
+	req := requestOfSize(t, 16, 1<<20)
+	_, err := fullClient(name).Exchange(context.Background(), req, testCred, time.Now().Add(30*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, req, recvBounded(t, got))
+}
+
+// Test 6a: a Request and a Reply of exactly MaxFrameBytes with a full 6 MiB
+// payload are both carried byte-identical.
+func TestControl_AtBoundRequestAndReplyBothFit(t *testing.T) {
+	name := controlName(t)
+	want := replyOfSize(t, MaxPayloadBytes, int(MaxFrameBytes))
+	got := make(chan []byte, 1)
+	fullServe(t, fullListen(t, name), func(_ context.Context, req []byte, _ Credential) Reply {
+		got <- req
+		return want
+	})
+	req := requestOfSize(t, MaxPayloadBytes, int(MaxFrameBytes))
+	rep, err := fullClient(name).Exchange(context.Background(), req, testCred, time.Now().Add(60*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, req, recvBounded(t, got))
+	require.Equal(t, want, rep)
+}
+
+// Test 6a: an at-bound reply with no payload, whose payload field the encoder
+// omits, still fits.
+func TestControl_AtBoundReplyWithoutPayloadFits(t *testing.T) {
+	name := controlName(t)
+	want := replyOfSize(t, 0, int(MaxFrameBytes))
+	fullServe(t, fullListen(t, name), func(context.Context, []byte, Credential) Reply { return want })
+	rep, err := fullClient(name).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(60*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, want, rep)
+}
+
+// Test 6b: the real server answers a handler's oversized payload, and a reply
+// encoding over MaxFrameBytes, with the small unknown reply.
+func TestControl_ServerReplacesOversizedRepliesWithUnknown(t *testing.T) {
+	over := Reply{Outcome: OutcomeDone, Payload: make([]byte, MaxPayloadBytes+1)}
+	for name, rep := range map[string]Reply{
+		"payload over 6 MiB":        over,
+		"encoding over frame bound": replyOfSize(t, MaxPayloadBytes, int(MaxFrameBytes)+1),
+	} {
+		t.Run(name, func(t *testing.T) {
+			sock := controlName(t)
+			fullServe(t, fullListen(t, sock), func(context.Context, []byte, Credential) Reply { return rep })
+			got, err := fullClient(sock).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(60*time.Second))
+			require.NoError(t, err)
+			require.Equal(t, Reply{Outcome: OutcomeUnknown, Error: "result too large for the test path"}, got)
+		})
+	}
+	// The oversized payload is refused before the reply is encoded: one
+	// exchange allocates far less than an 8 MiB base64 encoding.
+	sock := controlName(t)
+	fullServe(t, fullListen(t, sock), func(context.Context, []byte, Credential) Reply { return over })
+	var err error
+	alloc := allocatedDuring(func() {
+		_, err = fullClient(sock).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(60*time.Second))
+	})
+	require.NoError(t, err)
+	require.Less(t, alloc, uint64(4<<20), "the oversized reply was encoded before the payload check")
+}
+
+// directPeer serves one connection by reading the request and credential,
+// then writing with body; the peer's own error is returned on the channel.
+func directPeer(t *testing.T, name string, body func(net.Conn) error) chan error {
+	t.Helper()
+	ln := fullListen(t, name)
+	t.Cleanup(func() { _ = ln.Close() })
+	peerErr := make(chan error, 1)
+	go func() {
+		peerErr <- func() error {
+			c, err := ln.AcceptUnix()
+			if err != nil {
+				return err
+			}
+			defer func() { _ = c.Close() }()
+			if _, err := readFrame(c, MaxFrameBytes); err != nil {
+				return err
+			}
+			if _, err := readFrame(c, MaxCredentialBytes); err != nil {
+				return err
+			}
+			return body(c)
+		}()
+	}()
+	return peerErr
+}
+
+// Test 6: replies from a direct peer daemon over the payload bound or the
+// frame bound are errors after sending, so the gateway reports unknown.
+func TestControl_ClientRejectsOversizedReplies(t *testing.T) {
+	overPayload, err := json.Marshal(Reply{Outcome: OutcomeDone, Payload: make([]byte, MaxPayloadBytes+1)})
+	require.NoError(t, err)
+	require.LessOrEqual(t, int64(len(overPayload)), MaxFrameBytes)
+	cases := map[string]struct {
+		send func(net.Conn) error
+		want string
+	}{
+		"payload over 6 MiB": {func(c net.Conn) error { return writeFrame(c, overPayload) }, "reply payload of 6291457 bytes exceeds 6291456"},
+		"frame over bound": {func(c net.Conn) error {
+			var hdr [4]byte
+			binary.BigEndian.PutUint32(hdr[:], uint32(MaxFrameBytes+1))
+			_, err := c.Write(hdr[:])
+			return err
+		}, "outside 1.."},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			sock := controlName(t)
+			peerErr := directPeer(t, sock, tc.send)
+			_, err := fullClient(sock).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(30*time.Second))
+			require.NoError(t, recvBounded(t, peerErr), "the peer read the request and wrote its reply")
+			require.ErrorContains(t, err, tc.want)
+			require.NotErrorIs(t, err, ErrNotSent)
+		})
+	}
+}
+
+// Test 12: the credential bound is unchanged beside a full-size request.
+func TestControl_CredentialOverBoundRefusedWithAFullRequest(t *testing.T) {
+	cred := testCred
+	b, err := json.Marshal(cred)
+	require.NoError(t, err)
+	cred.SessionToken += strings.Repeat("t", MaxCredentialBytes+1-len(b))
+	b, err = json.Marshal(cred)
+	require.NoError(t, err)
+	require.Len(t, b, MaxCredentialBytes+1)
+
+	name := controlName(t)
+	s, _ := serveControl(t, name, os.Getuid(), nil)
+	_, err = fullClient(name).Exchange(context.Background(), requestOfSize(t, MaxPayloadBytes, int(MaxFrameBytes)), cred, time.Now().Add(30*time.Second))
+	require.ErrorIs(t, err, ErrNotSent)
+	require.Zero(t, s.calls.Load())
+}
+
+// pausedHandlers counts concurrent Handle calls and blocks each until released.
+type pausedHandlers struct {
+	in, max atomic.Int32
+	calls   atomic.Int32
+	entered chan []byte
+	release chan struct{}
+	once    sync.Once
+}
+
+func newPaused(t *testing.T) *pausedHandlers {
+	p := &pausedHandlers{entered: make(chan []byte, 64), release: make(chan struct{})}
+	t.Cleanup(p.free)
+	return p
+}
+
+func (p *pausedHandlers) free() { p.once.Do(func() { close(p.release) }) }
+
+func (p *pausedHandlers) handle(ctx context.Context, req []byte, _ Credential) Reply {
+	n := p.in.Add(1)
+	for {
+		m := p.max.Load()
+		if n <= m || p.max.CompareAndSwap(m, n) {
+			break
+		}
+	}
+	p.calls.Add(1)
+	p.entered <- req
+	select {
+	case <-p.release:
+	case <-ctx.Done():
+	}
+	p.in.Add(-1)
+	return Reply{Outcome: OutcomeDone}
+}
+
+func (p *pausedHandlers) waitEntered(t *testing.T, n int) {
+	t.Helper()
+	for range n {
+		select {
+		case <-p.entered:
+		case <-time.After(30 * time.Second):
+			t.Fatal("handler not entered")
+		}
+	}
+}
+
+// serverConns reads the kernel's socket table for the abstract name: server
+// ends Accept returned are connected and held by this process; a connection
+// still in the listen backlog is connecting, with no inode until accepted.
+func serverConns(t *testing.T, name string) (accepted, backlog int) {
+	t.Helper()
+	held := map[string]bool{}
+	fds, err := os.ReadDir("/proc/self/fd")
+	require.NoError(t, err)
+	for _, fd := range fds {
+		if target, err := os.Readlink("/proc/self/fd/" + fd.Name()); err == nil && strings.HasPrefix(target, "socket:[") {
+			held[strings.TrimSuffix(strings.TrimPrefix(target, "socket:["), "]")] = true
+		}
+	}
+	table, err := os.ReadFile("/proc/net/unix")
+	require.NoError(t, err)
+	for _, line := range strings.Split(string(table), "\n")[1:] {
+		f := strings.Fields(line)
+		// Num RefCount Protocol Flags Type St Inode Path.
+		if len(f) != 8 || f[7] != "@"+name {
+			continue
+		}
+		switch {
+		case f[5] == "03" && held[f[6]]:
+			accepted++
+		case f[5] == "02" && f[6] == "0":
+			backlog++
+		}
+	}
+	return accepted, backlog
+}
+
+// Test 8: eight full-size requests at once; with two admitted connections
+// paused in Handle, the kernel shows two accepted and six still in the
+// backlog, and at most two Handle calls ever run at once.
+func TestControl_AdmissionHoldsAtMostTwoFullSizeRequests(t *testing.T) {
+	name := controlName(t)
+	p := newPaused(t)
+	fullServe(t, fullListen(t, name), p.handle)
+	req := requestOfSize(t, MaxPayloadBytes, int(MaxFrameBytes))
+
+	var wg sync.WaitGroup
+	errs := make(chan error, 8)
+	for range 8 {
+		wg.Go(func() {
+			_, err := fullClient(name).Exchange(context.Background(), req, testCred, time.Now().Add(90*time.Second))
+			errs <- err
+		})
+	}
+	p.waitEntered(t, MaxAdmitted)
+	deadline := time.Now().Add(30 * time.Second)
+	for {
+		accepted, backlog := serverConns(t, name)
+		require.LessOrEqual(t, accepted, MaxAdmitted, "a third connection was accepted")
+		if backlog == 8-MaxAdmitted {
+			break
+		}
+		require.True(t, time.Now().Before(deadline), "the other clients never reached the backlog: %d accepted, %d waiting", accepted, backlog)
+		runtime.Gosched()
+	}
+	for range 20 {
+		accepted, backlog := serverConns(t, name)
+		require.Equal(t, MaxAdmitted, accepted, "only the admitted connections are accepted")
+		require.Equal(t, 8-MaxAdmitted, backlog, "the other six stay unaccepted")
+		<-time.After(50 * time.Millisecond)
+	}
+	require.Equal(t, int32(MaxAdmitted), p.calls.Load())
+
+	p.free()
+	waitGroupBounded(t, &wg, 90*time.Second)
+	close(errs)
+	for err := range errs {
+		require.NoError(t, err)
+	}
+	require.Equal(t, int32(8), p.calls.Load())
+	require.LessOrEqual(t, p.max.Load(), int32(MaxAdmitted))
+}
+
+// Test 9: a gateway that wrote its whole request while unaccepted, then gave
+// up and closed, reports unknown; the daemon later admits it and Handle runs.
+func TestControl_GatewayGivesUpButBufferedRequestIsStillHandled(t *testing.T) {
+	name := controlName(t)
+	p := newPaused(t)
+	fullServe(t, fullListen(t, name), p.handle)
+	for range MaxAdmitted {
+		go func() {
+			_, _ = fullClient(name).Exchange(context.Background(), []byte(`{"attempt_id":"held"}`), testCred, time.Now().Add(60*time.Second))
+		}()
+	}
+	p.waitEntered(t, MaxAdmitted)
+
+	late := []byte(`{"attempt_id":"late"}`)
+	c := fullClient(name)
+	c.IOTimeout = 2 * time.Second
+	_, err := c.Exchange(context.Background(), late, testCred, time.Now().Add(2*time.Second))
+	require.Error(t, err)
+	require.NotErrorIs(t, err, ErrNotSent, "the request was written, so the outcome is unknown")
+
+	p.free()
+	deadline := time.After(30 * time.Second)
+	for {
+		select {
+		case req := <-p.entered:
+			if bytes.Equal(req, late) {
+				return // Handle ran for a request whose gateway reported unknown.
+			}
+		case <-deadline:
+			t.Fatal("the buffered request was never handled")
+		}
+	}
+}
+
+// smallBacklogListener binds name with a listen backlog of one.
+func smallBacklogListener(t *testing.T, name string) *net.UnixListener {
+	t.Helper()
+	fd, err := unix.Socket(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_CLOEXEC, 0)
+	require.NoError(t, err)
+	require.NoError(t, unix.Bind(fd, &unix.SockaddrUnix{Name: "@" + name}))
+	require.NoError(t, unix.Listen(fd, 1))
+	f := os.NewFile(uintptr(fd), name)
+	defer func() { _ = f.Close() }()
+	l, err := net.FileListener(f)
+	require.NoError(t, err)
+	return l.(*net.UnixListener)
+}
+
+// Test 10: with both slots held and the backlog full, a gateway connect made
+// with no context deadline fails or stops at replyBy as ErrNotSent, and no
+// request reaches Handle.
+func TestControl_FullBacklogConnectIsNotSent(t *testing.T) {
+	name := controlName(t)
+	p := newPaused(t)
+	fullServe(t, smallBacklogListener(t, name), p.handle)
+	for range MaxAdmitted {
+		go func() {
+			_, _ = fullClient(name).Exchange(context.Background(), []byte(`{"attempt_id":"held"}`), testCred, time.Now().Add(60*time.Second))
+		}()
+	}
+	p.waitEntered(t, MaxAdmitted)
+
+	// Fill the backlog with raw connections that never write.
+	var fillers []net.Conn
+	t.Cleanup(func() {
+		for _, c := range fillers {
+			_ = c.Close()
+		}
+	})
+	for range 64 {
+		d := net.Dialer{Timeout: 200 * time.Millisecond}
+		c, err := d.Dial("unix", "@"+name)
+		if err != nil {
+			break
+		}
+		fillers = append(fillers, c)
+	}
+	require.Less(t, len(fillers), 64, "the backlog never filled")
+
+	began := time.Now()
+	_, err := fullClient(name).Exchange(context.Background(), []byte(`{"attempt_id":"x"}`), testCred, time.Now().Add(2*time.Second))
+	require.ErrorIs(t, err, ErrNotSent)
+	require.Less(t, time.Since(began), 10*time.Second)
+	require.Equal(t, int32(MaxAdmitted), p.calls.Load(), "no further request reached Handle")
+}
+
+// Test 15 (measurement, not a bound): two admitted full-size lifetimes, each
+// decoding and fit-checking its request as the daemon does and answering with
+// a 6 MiB reply, measured from a post-GC baseline just before the first dial
+// until both replies are read.
+func TestControl_MeasureTwoAdmittedFullSizeLifetimes(t *testing.T) {
+	name := controlName(t)
+	var mu sync.Mutex
+	entered := 0
+	both := make(chan struct{})
+	fitErrs := make(chan error, MaxAdmitted)
+	fullServe(t, fullListen(t, name), func(ctx context.Context, b []byte, cred Credential) Reply {
+		var req Request
+		if err := json.Unmarshal(b, &req); err != nil {
+			return Reply{Outcome: OutcomeRefused, Error: err.Error()}
+		}
+		req.cred = cred
+		fitErrs <- hostlink.CheckInvokeFits(guestRequest(req, 0))
+		mu.Lock()
+		entered++
+		if entered == MaxAdmitted {
+			close(both)
+		}
+		mu.Unlock()
+		select {
+		case <-both:
+		case <-ctx.Done():
+			return Reply{Outcome: OutcomeUnknown}
+		}
+		return Reply{Outcome: OutcomeDone, Payload: bytes.Repeat([]byte{9}, MaxPayloadBytes)}
+	})
+	req := requestOfSize(t, MaxPayloadBytes, int(MaxFrameBytes))
+
+	var wg sync.WaitGroup
+	var reps [MaxAdmitted]Reply
+	var errs [MaxAdmitted]error
+	var before, after runtime.MemStats
+	runtime.GC()
+	runtime.ReadMemStats(&before)
+
+	// HeapAlloc is sampled every 2 ms from the baseline until both replies are
+	// read, so its maximum is a sampled process-wide figure, not a true peak.
+	var heapMax uint64
+	samples := 0
+	stopSampling, sampled := make(chan struct{}), make(chan struct{})
+	go func() {
+		defer close(sampled)
+		tick := time.NewTicker(2 * time.Millisecond)
+		defer tick.Stop()
+		for {
+			var m runtime.MemStats
+			runtime.ReadMemStats(&m)
+			heapMax, samples = max(heapMax, m.HeapAlloc), samples+1
+			select {
+			case <-tick.C:
+			case <-stopSampling:
+				return
+			}
+		}
+	}()
+	var stopOnce sync.Once
+	stop := func() {
+		stopOnce.Do(func() { close(stopSampling) })
+		<-sampled
+	}
+	t.Cleanup(stop)
+
+	for i := range MaxAdmitted {
+		wg.Go(func() {
+			reps[i], errs[i] = fullClient(name).Exchange(context.Background(), req, testCred, time.Now().Add(60*time.Second))
+		})
+	}
+	waitGroupBounded(t, &wg, 60*time.Second)
+	stop()
+	runtime.ReadMemStats(&after)
+
+	for i := range MaxAdmitted {
+		require.NoError(t, errs[i])
+		require.Equal(t, OutcomeDone, reps[i].Outcome)
+		require.Len(t, reps[i].Payload, MaxPayloadBytes, "each lifetime ends with its full-size reply read")
+		require.NoError(t, recvBounded(t, fitErrs), "the full-size request fits the guest link")
+	}
+	// Excludes the executor's marshaling, HostSend, the guest and QEMU.
+	t.Logf("two admitted full-size lifetimes, process-wide (client and server): %d bytes in %d allocations; sampled max HeapAlloc %d bytes over %d samples",
+		after.TotalAlloc-before.TotalAlloc, after.Mallocs-before.Mallocs, heapMax, samples)
+}

spinifex/runtime/compute/lambda/testlink/control_daemon_test.go

@@ -0,0 +1,96 @@
+//go:build lambdatestlink
+
+//test:in-package — reads the daemon's unexported slot and owner fixtures
+package testlink
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/execution"
+)
+
+// A direct peer's request whose raw payload is 6 MiB + 1 but whose frame is
+// within MaxFrameBytes passes the frame check, reaches the real daemon, and
+// is refused there before Start with no address or tap taken.
+func TestControlDaemon_RawPayloadOverBoundRefusedBeforeStart(t *testing.T) {
+	code := []byte("zip bytes")
+	srv := codeServer(t, code)
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: srv.Client(), WorkDir: t.TempDir()})
+
+	req := baseRequest(srv.URL, code)
+	req.Grant, req.ImageDigest = g.att.Grant, g.image.Digest
+	req.Payload = make([]byte, MaxPayloadBytes+1)
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+	require.LessOrEqual(t, int64(len(b)), MaxFrameBytes, "the frame alone would admit it")
+
+	name := controlName(t)
+	fullServe(t, fullListen(t, name), d.Handle)
+	rep, err := fullClient(name).Exchange(context.Background(), b, testCred, time.Now().Add(30*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, OutcomeRefused, rep.Outcome, rep.Error)
+	require.Contains(t, rep.Error, "payload")
+	require.Empty(t, ips.allocated)
+	require.Empty(t, taps.created)
+	require.Equal(t, execution.StateGranted, g.state(t))
+}
+
+// With one real attempt started and held in its code fetch, a second request
+// through the real control server is refused before Start as busy.
+func TestControlDaemon_SecondRequestRefusedWhileAnAttemptIsHeld(t *testing.T) {
+	code := []byte("zip bytes")
+	fetching, release := make(chan struct{}), make(chan struct{})
+	var fetchOnce, releaseOnce sync.Once
+	free := func() { releaseOnce.Do(func() { close(release) }) }
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fetchOnce.Do(func() { close(fetching) })
+		select {
+		case <-release:
+		case <-r.Context().Done():
+			return
+		}
+		_, _ = w.Write(code)
+	}))
+	// Cleanups run last-registered first: free the held fetch before Close.
+	t.Cleanup(srv.Close)
+	t.Cleanup(free)
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: srv.Client(), WorkDir: t.TempDir()})
+	name := controlName(t)
+	fullServe(t, fullListen(t, name), d.Handle)
+
+	first := make(chan error, 1)
+	go func() {
+		_, err := fullClient(name).Exchange(context.Background(), g.request(srv.URL, code), testCred, time.Now().Add(60*time.Second))
+		first <- err
+	}()
+	// On every path the held attempt is released and its exchange joined.
+	t.Cleanup(func() {
+		free()
+		select {
+		case <-first:
+		case <-time.After(30 * time.Second):
+			t.Error("the held exchange did not return")
+		}
+	})
+	recvBounded(t, fetching)
+	require.Equal(t, execution.StateRunning, g.state(t), "the first attempt started")
+
+	rep, err := fullClient(name).Exchange(context.Background(), g.request(srv.URL, code), testCred, time.Now().Add(30*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, OutcomeRefused, rep.Outcome)
+	require.Contains(t, rep.Error, "another")
+	require.Empty(t, ips.allocated, "nothing was allocated for either attempt yet")
+	require.Empty(t, taps.created)
+}

spinifex/runtime/compute/lambda/testlink/control_lifecycle_test.go

@@ -0,0 +1,169 @@
+//go:build lambdatestlink
+
+//test:in-package — exercises the unexported joinAfterFunc and Serve's admission lifecycle
+package testlink
+
+import (
+	"context"
+	"net"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+)
+
+// waitClosed fails the test if ch is not closed within d.
+func waitClosed(t *testing.T, ch <-chan struct{}, d time.Duration, what string) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(d):
+		t.Fatal(what)
+	}
+}
+
+// waitGroupBounded fails the test if wg does not finish within d.
+func waitGroupBounded(t *testing.T, wg *sync.WaitGroup, d time.Duration) {
+	t.Helper()
+	done := make(chan struct{})
+	go func() { wg.Wait(); close(done) }()
+	waitClosed(t, done, d, "goroutines did not finish")
+}
+
+// A stop after the callback has started waits for it to finish, so no
+// close is still running when its caller releases a slot or returns.
+func TestJoinAfterFunc_StopWaitsForAStartedCallback(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	started, release := make(chan struct{}), make(chan struct{})
+	var once sync.Once
+	unblock := func() { once.Do(func() { close(release) }) }
+	t.Cleanup(unblock)
+	stop := joinAfterFunc(ctx, func() { close(started); <-release })
+	cancel()
+	waitClosed(t, started, 5*time.Second, "callback never started")
+
+	returned := make(chan struct{})
+	go func() { stop(); close(returned) }()
+	select {
+	case <-returned:
+		t.Fatal("stop returned while the callback was still running")
+	case <-time.After(100 * time.Millisecond):
+	}
+	unblock()
+	waitClosed(t, returned, 5*time.Second, "stop never returned after the callback finished")
+}
+
+func TestJoinAfterFunc_StopBeforeCancelNeverRunsTheCallback(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	ran := make(chan struct{}, 1)
+	stop := joinAfterFunc(ctx, func() { ran <- struct{}{} })
+	stop()
+	cancel()
+	select {
+	case <-ran:
+		t.Fatal("callback ran after stop")
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+// startServe runs Serve with the given IOTimeout and handler and returns a
+// cancel that waits for Serve to return, reporting how long that took.
+func startServe(t *testing.T, name string, io time.Duration, handle func(context.Context, []byte, Credential) Reply) func() time.Duration {
+	t.Helper()
+	ln, err := ListenControl(name)
+	require.NoError(t, err)
+	srv := &ControlServer{GatewayUID: os.Getuid(), RequestLimit: testRequestLimit, ReplyLimit: 4096, IOTimeout: io, Handle: handle}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- srv.Serve(ctx, ln) }()
+	var once sync.Once
+	var took time.Duration
+	stop := func() time.Duration {
+		once.Do(func() {
+			began := time.Now()
+			cancel()
+			select {
+			case err := <-done:
+				require.NoError(t, err)
+			case <-time.After(20 * time.Second):
+				t.Fatal("Serve did not return after cancellation")
+			}
+			took = time.Since(began)
+		})
+		return took
+	}
+	t.Cleanup(func() { stop() })
+	return stop
+}
+
+// Shutdown closes an admitted peer stalled mid-request and a handler that
+// honours cancellation, well inside the 30 s read deadline, leaving no goroutine.
+func TestServe_ShutdownClosesAStalledPeerAndACancelledHandler(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+	name := controlName(t)
+	inHandle := make(chan struct{})
+	stop := startServe(t, name, 30*time.Second, func(ctx context.Context, _ []byte, _ Credential) Reply {
+		close(inHandle)
+		<-ctx.Done()
+		return Reply{Outcome: OutcomeUnknown}
+	})
+
+	stalled, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: "@" + name, Net: "unix"})
+	require.NoError(t, err)
+	defer func() { _ = stalled.Close() }()
+	// Half a frame header: the server blocks reading the rest.
+	_, err = stalled.Write([]byte{0, 0})
+	require.NoError(t, err)
+
+	exchanged := make(chan error, 1)
+	go func() {
+		_, err := client(name, os.Getuid()).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(30*time.Second))
+		exchanged <- err
+	}()
+	select {
+	case <-inHandle:
+	case <-time.After(5 * time.Second):
+		t.Fatal("second admitted connection never reached Handle")
+	}
+
+	took := stop()
+	require.Less(t, took, 5*time.Second, "shutdown waited on the read deadline")
+	select {
+	case <-exchanged:
+	case <-time.After(5 * time.Second):
+		t.Fatal("gateway exchange never ended after shutdown")
+	}
+	require.NoError(t, stalled.SetReadDeadline(time.Now().Add(5*time.Second)))
+	_, err = stalled.Read(make([]byte, 1))
+	require.Error(t, err, "the stalled peer's connection was closed")
+}
+
+// Cancellation racing normal completion never leaves a close callback or
+// connection goroutine behind once Serve returns.
+func TestServe_CancellationRacingCompletionLeavesNothingRunning(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+	for i := range 50 {
+		name := controlName(t)
+		stop := startServe(t, name, 5*time.Second, func(context.Context, []byte, Credential) Reply {
+			return Reply{Outcome: OutcomeDone, Payload: []byte("ok")}
+		})
+		// Alternate cancelling before, during and after the exchanges.
+		if i%3 == 0 {
+			stop()
+		}
+		var wg sync.WaitGroup
+		for range MaxAdmitted {
+			wg.Go(func() {
+				_, _ = client(name, os.Getuid()).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(5*time.Second))
+			})
+		}
+		if i%3 == 1 {
+			stop()
+		}
+		waitGroupBounded(t, &wg, 15*time.Second)
+		stop()
+	}
+}

spinifex/runtime/compute/lambda/testlink/control_test.go

@@ -435,3 +435,25 @@ func TestControl_WriteErrorAfterDeliveryIsUnknown(t *testing.T) {
 	require.Equal(t, []byte(`{"attempt_id":"a"}`), s.req.Load())
 	require.Equal(t, testCred, s.cred.Load())
 }
+
+// legacyControlName is the protocol name before the request carried the image
+// digest; an old daemon still serves it and ignores the field.
+const legacyControlName = "spinifex-lambda-testlink-control"
+
+// A gateway and daemon on different control protocol names never connect, in
+// either direction. The scratch prefix isolates the binds; the names after it
+// are the production constant and its predecessor.
+func TestControl_MixedProtocolNamesNeverConnect(t *testing.T) {
+	prefix := controlName(t) + "/"
+	for name, tc := range map[string]struct{ daemon, gateway string }{
+		"old daemon, new gateway": {prefix + legacyControlName, prefix + ControlName},
+		"new daemon, old gateway": {prefix + ControlName, prefix + legacyControlName},
+	} {
+		t.Run(name, func(t *testing.T) {
+			s, _ := serveControl(t, tc.daemon, os.Getuid(), nil)
+			_, err := client(tc.gateway, os.Getuid()).Exchange(t.Context(), []byte(`{"attempt_id":"a"}`), testCred, soon())
+			require.ErrorIs(t, err, ErrNotSent)
+			assert.Equal(t, int32(0), s.calls.Load())
+		})
+	}
+}

spinifex/runtime/compute/lambda/testlink/daemon.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+	"math"
 	"net"
 	"net/http"
 	"net/url"
@@ -113,12 +114,12 @@ const controlIOTimeout = 30 * time.Second
 
 // ServeControl binds the control socket name and answers the gateway's
 // requests on it until ctx ends. A bind error means this daemon does not serve.
-func (d *Daemon) ServeControl(ctx context.Context, name string, gatewayUID int, limit int64) error {
+func (d *Daemon) ServeControl(ctx context.Context, name string, gatewayUID int) error {
 	ln, err := ListenControl(name)
 	if err != nil {
 		return err
 	}
-	srv := &ControlServer{GatewayUID: gatewayUID, RequestLimit: limit, ReplyLimit: limit,
+	srv := &ControlServer{GatewayUID: gatewayUID, RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes,
 		IOTimeout: controlIOTimeout, Handle: d.Handle}
 	go func() {
 		if err := srv.Serve(ctx, ln); err != nil {
@@ -149,6 +150,9 @@ func (d *Daemon) Handle(ctx context.Context, data []byte, cred Credential) Reply
 	if d.Image == nil {
 		return refused("runtime image not prepared")
 	}
+	if req.ImageDigest != d.Image.Digest {
+		return refused("runtime image digest differs from the prepared image")
+	}
 	if !d.busy.CompareAndSwap(false, true) {
 		return refused("another test-link attempt holds this node's slot")
 	}
@@ -162,6 +166,12 @@ func (d *Daemon) Handle(ctx context.Context, data []byte, cred Credential) Reply
 	if id, err := d.unresolved(); err != nil || id != "" {
 		return refused("an unresolved test-link attempt holds this node's slot")
 	}
+	// The guest frame is checked at its longest deadline encoding, MinInt64,
+	// before Start: control and guest frames bound different fields. It runs
+	// after the slot is taken so a busy refusal never encodes the payload.
+	if err := hostlink.CheckInvokeFits(guestRequest(req, math.MinInt64)); err != nil {
+		return refused("request does not fit the guest link")
+	}
 	// Without pidfs a restarted daemon could not identify the VM it left.
 	if _, err := selfInode(); err != nil {
 		return refused(err.Error())
@@ -193,7 +203,7 @@ func (d *Daemon) attempt(ctx context.Context, req Request, launch func(code []by
 	if err := d.Grants.Start(ctx, req.Grant, req.Identity()); err != nil {
 		if errors.Is(err, execution.ErrOutcomeUnknown) {
 			slog.Error("Lambda test link: grant start outcome unknown; node held", "attempt", req.AttemptID, "err", err)
-			return Reply{Outcome: OutcomeUnknown, Error: "grant start outcome unknown"}, true
+			return Reply{Outcome: OutcomeUnknown, Error: "grant start outcome unknown"}, false
 		}
 		return refused(err.Error()), true
 	}
@@ -251,8 +261,25 @@ func removeMarker(marker string) bool {
 	return true
 }
 
+// guestRequest is the invocation frame sent to the guest; the pre-start fit
+// check and the send share it so they cannot drift.
+func guestRequest(req Request, deadlineMs int64) hostlink.Request {
+	return hostlink.Request{
+		GrantID:     req.Grant.GrantID,
+		AttemptID:   req.AttemptID,
+		RequestID:   req.RequestID,
+		FunctionARN: req.FunctionARN,
+		DeadlineMs:  deadlineMs,
+		Payload:     req.Payload,
+		Credential:  &req.cred,
+	}
+}
+
 // CheckRequest is the daemon's validation of a test-link request.
 func CheckRequest(req Request) error {
+	if len(req.Payload) > MaxPayloadBytes {
+		return fmt.Errorf("payload of %d bytes exceeds %d", len(req.Payload), MaxPayloadBytes)
+	}
 	if !attemptIDPattern.MatchString(req.AttemptID) {
 		return errors.New("attempt ID must be 12 lowercase hex digits")
 	}
@@ -577,15 +604,7 @@ func (d *Daemon) session(ctx context.Context, req Request, gen *guestconn.Genera
 	if expired(hardDeadline) {
 		return Reply{}, ErrDeadlineExpired
 	}
-	if err := stream.HostSend(hostlink.Request{
-		GrantID:     req.Grant.GrantID,
-		AttemptID:   req.AttemptID,
-		RequestID:   req.RequestID,
-		FunctionARN: req.FunctionARN,
-		DeadlineMs:  deadline.UnixMilli(),
-		Payload:     req.Payload,
-		Credential:  &req.cred,
-	}); err != nil {
+	if err := stream.HostSend(guestRequest(req, deadline.UnixMilli())); err != nil {
 		return Reply{}, pastDeadline(deadline, fmt.Errorf("send invocation: %w", err))
 	}
 	resp, err := stream.HostResult(rctx)

spinifex/runtime/compute/lambda/testlink/guestframe_test.go

@@ -0,0 +1,146 @@
+//go:build lambdatestlink
+
+//test:in-package — drives Daemon.Handle with the package's grant fixtures and guestRequest
+package testlink
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/execution"
+)
+
+// maxCredential encodes to exactly MaxCredentialBytes.
+func maxCredential(t *testing.T) Credential {
+	t.Helper()
+	c := Credential{AccessKeyID: "ASIA" + strings.Repeat("K", 16), SecretAccessKey: strings.Repeat("s", 40)}
+	b, err := json.Marshal(c)
+	require.NoError(t, err)
+	c.SessionToken = strings.Repeat("t", MaxCredentialBytes-len(b))
+	b, err = json.Marshal(c)
+	require.NoError(t, err)
+	require.Len(t, b, MaxCredentialBytes)
+	return c
+}
+
+// A request the control socket accepts but whose guest frame would not fit
+// is refused before Start, so no grant starts and nothing is allocated.
+func TestDaemon_GuestFrameOverflowRefusedBeforeStart(t *testing.T) {
+	code := []byte("zip bytes")
+	good := codeServer(t, code)
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: good.Client(), WorkDir: t.TempDir()})
+
+	req := baseRequest(good.URL, code)
+	req.Grant, req.ImageDigest = g.att.Grant, g.image.Digest
+	req.Payload = make([]byte, MaxPayloadBytes)
+	req.FunctionARN = strings.Repeat("a", 60000)
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+	require.LessOrEqual(t, int64(len(b)), MaxFrameBytes, "the control frame fits")
+	require.NoError(t, CheckRequest(req))
+
+	rep := d.Handle(context.Background(), b, maxCredential(t))
+	require.Equal(t, OutcomeRefused, rep.Outcome, rep.Error)
+	require.Contains(t, rep.Error, "does not fit the guest link")
+	require.Equal(t, execution.StateGranted, g.state(t), "the grant was never started")
+	require.False(t, d.busy.Load(), "a fit refusal releases the slot")
+	require.Empty(t, ips.allocated)
+	require.Empty(t, taps.created)
+
+	// A later valid request for the same grant is not refused as busy.
+	rep = d.Handle(context.Background(), g.request(good.URL, code), testCred)
+	require.NotContains(t, rep.Error, "holds this node's slot")
+	require.NotEqual(t, execution.StateGranted, g.state(t), "the valid request started the grant")
+}
+
+// The largest Invoke ARN, a full payload, a maximal credential and full IDs
+// reach a guest reader byte-identical through HostSend.
+func TestGuestRequest_LargestInvokeFitsTheGuestLink(t *testing.T) {
+	cred := maxCredential(t)
+	req := Request{
+		Grant:     execution.Grant{GrantID: strings.Repeat("f", 32)},
+		AttemptID: strings.Repeat("a", 12),
+		RequestID: strings.Repeat("b", 12),
+		// Region of representative length; name and qualifier at Invoke's maxima.
+		FunctionARN: "arn:aws:lambda:ap-southeast-2:123456789012:function:" + strings.Repeat("n", 64) + ":" + strings.Repeat("q", 128),
+		Payload:     make([]byte, MaxPayloadBytes),
+		cred:        cred,
+	}
+	for i := range req.Payload {
+		req.Payload[i] = byte(i)
+	}
+	sent := guestRequest(req, math.MinInt64)
+	require.NoError(t, hostlink.CheckInvokeFits(sent))
+
+	hostEnd, guestEnd := net.Pipe()
+	t.Cleanup(func() { _ = hostEnd.Close(); _ = guestEnd.Close() })
+	host, guest := hostlink.NewTestStream(hostEnd), hostlink.NewTestStream(guestEnd)
+	sendErr := make(chan error, 1)
+	go func() { sendErr <- host.HostSend(sent) }()
+
+	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
+	defer cancel()
+	got, err := guest.Next(ctx)
+	require.NoError(t, err)
+	select {
+	case err := <-sendErr:
+		require.NoError(t, err)
+	case <-time.After(30 * time.Second):
+		t.Fatal("HostSend never returned")
+	}
+	require.Equal(t, sent, got)
+}
+
+// At a request size where only the minus sign of MinInt64 overflows the guest
+// frame, the daemon still refuses before Start: its check uses the longest
+// int64 encoding, not MaxInt64.
+func TestDaemon_GuestFrameCheckUsesTheLongestDeadlineEncoding(t *testing.T) {
+	code := []byte("zip bytes")
+	good := codeServer(t, code)
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: good.Client(), WorkDir: t.TempDir()})
+	cred := maxCredential(t)
+
+	req := baseRequest(good.URL, code)
+	req.Grant, req.ImageDigest = g.att.Grant, g.image.Digest
+	req.Payload = make([]byte, MaxPayloadBytes)
+	req.cred = cred
+	fits := func(n int, deadline int64) bool {
+		r := req
+		r.FunctionARN = strings.Repeat("a", n)
+		return hostlink.CheckInvokeFits(guestRequest(r, deadline)) == nil
+	}
+	// The longest ARN that fits at MaxInt64.
+	lo, hi := 0, 70000
+	require.True(t, fits(lo, math.MaxInt64))
+	require.False(t, fits(hi, math.MaxInt64))
+	for hi-lo > 1 {
+		if mid := (lo + hi) / 2; fits(mid, math.MaxInt64) {
+			lo = mid
+		} else {
+			hi = mid
+		}
+	}
+	require.False(t, fits(lo, math.MinInt64), "MinInt64 is one byte longer")
+
+	req.FunctionARN = strings.Repeat("a", lo)
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+	require.LessOrEqual(t, int64(len(b)), MaxFrameBytes)
+	rep := d.Handle(context.Background(), b, cred)
+	require.Equal(t, OutcomeRefused, rep.Outcome, rep.Error)
+	require.Contains(t, rep.Error, "does not fit the guest link")
+	require.Equal(t, execution.StateGranted, g.state(t))
+	require.False(t, d.busy.Load(), "a fit refusal releases the slot")
+}

spinifex/runtime/compute/lambda/testlink/testlink_test.go

@@ -121,13 +121,18 @@ func baseRequest(url string, code []byte) Request {
 	return Request{
 		AttemptID: "0123456789ab", RequestID: "r", Region: "ap-southeast-2", AccountID: "123456789012",
 		FunctionID: "fn1", RevisionID: "rev1", Handler: "bootstrap.handler", TimeoutMs: 3000, CodeURL: url, CodeSize: int64(len(code)),
-		CodeSHA256: hex.EncodeToString(sum[:]),
+		CodeSHA256: hex.EncodeToString(sum[:]), ImageDigest: testImageDigest,
 	}
 }
 
 func (g *granted) request(url string, code []byte) []byte {
+	return g.requestFor(url, code, g.image.Digest)
+}
+
+func (g *granted) requestFor(url string, code []byte, imageDigest string) []byte {
 	req := baseRequest(url, code)
 	req.Grant = g.att.Grant
+	req.ImageDigest = imageDigest
 	b, _ := json.Marshal(req)
 	return b
 }
@@ -148,6 +153,10 @@ func TestDaemon_RefusesBeforeAllocating(t *testing.T) {
 			b, _ := json.Marshal(baseRequest(good.URL, code))
 			return b
 		}, execution.StateGranted},
+		// A gateway built for another runtime image, or an old one that sends
+		// no digest, is refused before Start.
+		"image digest differs": {func(g *granted) []byte { return g.requestFor(good.URL, code, strings.Repeat("0", 64)) }, execution.StateGranted},
+		"image digest missing": {func(g *granted) []byte { return g.requestFor(good.URL, code, "") }, execution.StateGranted},
 		// Refused after Start with nothing launched: the daemon records the end.
 		"digest mismatch": {func(g *granted) []byte { return g.request(wrong.URL, code) }, execution.StateEnded},
 		"size mismatch":   {func(g *granted) []byte { return g.request(short.URL, code) }, execution.StateEnded},
@@ -164,6 +173,7 @@ func TestDaemon_RefusesBeforeAllocating(t *testing.T) {
 			rep := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: client, WorkDir: t.TempDir()}).Handle(context.Background(), tc.data(g), testCred)
 			require.Equal(t, OutcomeRefused, rep.Outcome, rep.Error)
 			require.Empty(t, ips.allocated)
+			require.Empty(t, taps.created)
 			require.NotContains(t, rep.Error, good.URL)
 			require.Equal(t, tc.state, g.state(t))
 		})
@@ -198,6 +208,7 @@ func TestDaemon_OwnerRefusalComesBeforeAnyAllocation(t *testing.T) {
 			ips := &fakeIPs{}
 			rep := g.daemon(&Daemon{IPs: ips, Taps: &fakeTaps{}, HTTP: srv.Client(), WorkDir: t.TempDir()}).Handle(context.Background(), data, testCred)
 			require.Equal(t, OutcomeRefused, rep.Outcome)
+			require.NotContains(t, rep.Error, "image digest", "the owner, not the digest check, refuses")
 			require.False(t, fetched)
 			require.Empty(t, ips.allocated)
 			require.Equal(t, execution.StateGranted, g.state(t))
@@ -447,3 +458,65 @@ func TestCheckRequest_RefusesUnusableHandler(t *testing.T) {
 		})
 	}
 }
+
+// unconfirmedStart lets the owner write the start, then reports the outcome
+// as unknown, as a lost acknowledgement would.
+type unconfirmedStart struct{ *execution.Store }
+
+func (u unconfirmedStart) Start(ctx context.Context, g execution.Grant, want execution.Identity) error {
+	if err := u.Store.Start(ctx, g, want); err != nil {
+		return err
+	}
+	return execution.ErrOutcomeUnknown
+}
+
+func TestDaemon_UnknownStartOutcomeHoldsTheSlot(t *testing.T) {
+	code := []byte("x")
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, WorkDir: t.TempDir()})
+	d.Grants = unconfirmedStart{g.store}
+
+	rep := d.Handle(context.Background(), g.request("https://h/x", code), testCred)
+	require.Equal(t, OutcomeUnknown, rep.Outcome)
+	require.True(t, d.busy.Load(), "an unconfirmed start may have taken the node")
+	require.Empty(t, ips.allocated)
+	require.Equal(t, execution.StateRunning, g.state(t))
+
+	rep = d.Handle(context.Background(), g.request("https://h/x", code), testCred)
+	require.Equal(t, OutcomeRefused, rep.Outcome)
+	require.Contains(t, rep.Error, "holds this node's slot")
+}
+
+// A daemon whose image is replaced between two Invokes refuses the second
+// before Start, though the first passed the digest check.
+func TestDaemon_ImageReplacedBetweenInvokesRefusesBeforeStart(t *testing.T) {
+	code := []byte("zip bytes")
+	wrong := codeServer(t, []byte("other one"))
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: wrong.Client(), WorkDir: t.TempDir()})
+
+	rep := d.Handle(context.Background(), g.request(wrong.URL, code), testCred)
+	require.Equal(t, OutcomeRefused, rep.Outcome)
+	require.Equal(t, execution.StateEnded, g.state(t), "the first request passed the digest check and started")
+
+	second, err := g.store.Issue(context.Background(), execution.Attempt{
+		Grant:    execution.Grant{AttemptID: "0123456789ac", EnvironmentID: "env-0123456789ac", Generation: 1, Node: testNode},
+		Identity: baseRequest("", code).Identity(),
+		LaunchBy: time.Now().Add(time.Minute),
+	})
+	require.NoError(t, err)
+	d.Image = &Image{Digest: strings.Repeat("1", 64)}
+	req := baseRequest(wrong.URL, code)
+	req.AttemptID, req.Grant, req.ImageDigest = second.Grant.AttemptID, second.Grant, g.image.Digest
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+
+	rep = d.Handle(context.Background(), b, testCred)
+	require.Equal(t, OutcomeRefused, rep.Outcome)
+	require.Contains(t, rep.Error, "image digest differs")
+	require.Equal(t, execution.StateGranted, g.state(t), "the second grant was never started")
+	require.Empty(t, ips.allocated)
+	require.Empty(t, taps.created)
+}

spinifex/runtime/daemon/daemon_lambda_testlink.go

@@ -78,7 +78,7 @@ func (d *Daemon) startLambdaTestLink() {
 			if err != nil {
 				return err
 			}
-			return tl.ServeControl(d.ctx, testlink.ControlName, gatewayUID, d.natsConn.MaxPayload())
+			return tl.ServeControl(d.ctx, testlink.ControlName, gatewayUID)
 		},
 		func() { releaseLambdaTestLinkImage(tl) },
 	)

spinifex/runtime/roles/awsgw/lambda_testlink.go

@@ -24,17 +24,15 @@ func lambdaTestExecutor(nc *nats.Conn, node string, codes *codestore.Manager, se
 		slog.Error("Lambda test link: control peer account unavailable; Invoke is not served", "err", err)
 		return nil
 	}
+	if !testlink.ValidDigest(lambdaImageDigest) {
+		slog.Error("Lambda test link: runtime image pin missing or malformed; every Invoke is refused")
+	}
 	return &testLinkExecutor{
-		Control: &testlink.ControlClient{
-			Name:         testlink.ControlName,
-			DaemonUID:    daemonUID,
-			RequestLimit: nc.MaxPayload(),
-			ReplyLimit:   nc.MaxPayload(),
-			IOTimeout:    controlWriteTimeout,
-		},
-		RequestLimit: nc.MaxPayload(),
+		Control:      testLinkControl(daemonUID),
+		RequestLimit: testlink.MaxFrameBytes,
 		Sessions:     sessions,
 		Node:         node,
+		ImageDigest:  lambdaImageDigest,
 		Codes: func(ctx context.Context) (codePresigner, error) {
 			store, err := codes.Store(ctx)
 			if err != nil {
@@ -52,6 +50,21 @@ func lambdaTestExecutor(nc *nats.Conn, node string, codes *codestore.Manager, se
 	}
 }
 
+// testLinkControl is the gateway's client for this node's daemon socket.
+func testLinkControl(daemonUID int) *testlink.ControlClient {
+	return &testlink.ControlClient{
+		Name:         testlink.ControlName,
+		DaemonUID:    daemonUID,
+		RequestLimit: testlink.MaxFrameBytes,
+		ReplyLimit:   testlink.MaxFrameBytes,
+		IOTimeout:    controlWriteTimeout,
+	}
+}
+
+// lambdaImageDigest pins the runtime image this gateway expects the daemon
+// to run; make build-lambda-testlink sets it to the daemon's pin.
+var lambdaImageDigest string
+
 // controlWriteTimeout bounds writing one request to the local daemon.
 const controlWriteTimeout = 30 * time.Second
 

spinifex/runtime/roles/awsgw/lambda_testlink_executor.go

@@ -55,11 +55,12 @@ var _ grantIssuer = (*execution.Store)(nil)
 // testLinkExecutor sends each invocation to the test-link daemon on Node.
 type testLinkExecutor struct {
 	Control controlExchanger
-	// RequestLimit bounds the encoded request without its credential: the
-	// NATS message limit the path used before the socket.
+	// RequestLimit bounds the encoded request frame, without its credential.
 	RequestLimit int64
 	Sessions     lambdaEnvironmentRoleIssuer
 	Node         string
+	// ImageDigest is the runtime image every request requires of the daemon.
+	ImageDigest string
 	Codes  func(context.Context) (codePresigner, error)
 	Grants func(context.Context) (grantIssuer, error)
 	now    func() time.Time
@@ -73,6 +74,10 @@ var _ awsapi.Executor = (*testLinkExecutor)(nil)
 // session is revoked on every path after it is issued.
 func (e *testLinkExecutor) InvokeSync(ctx context.Context, in awsapi.SyncInvocation) (awsapi.SyncResult, error) {
 	rec := in.Function
+	if !testlink.ValidDigest(e.ImageDigest) {
+		return awsapi.SyncResult{}, &awsapi.RefusalError{Err: awserrors.Errorf(awserrors.ErrorServiceException,
+			"the test execution path has no valid runtime image pin")}
+	}
 	// The test link cannot deliver environment variables, so it refuses a
 	// function that has any before presigning, granting or sending.
 	if rec.EnvCipher != "" {
@@ -159,6 +164,11 @@ func (e *testLinkExecutor) InvokeSync(ctx context.Context, in awsapi.SyncInvocat
 		return err
 	}
 
+	// The raw payload is checked before encoding allocates its base64 copy.
+	if len(in.Payload) > testlink.MaxPayloadBytes {
+		_ = revoke()
+		return refuse(fmt.Errorf("payload of %d bytes exceeds %d", len(in.Payload), testlink.MaxPayloadBytes))
+	}
 	body, err := json.Marshal(testlink.Request{
 		Grant:       att.Grant,
 		AttemptID:   attemptID,
@@ -174,13 +184,14 @@ func (e *testLinkExecutor) InvokeSync(ctx context.Context, in awsapi.SyncInvocat
 		CodeURL:     url,
 		CodeSize:    rec.Code.Size,
 		CodeSHA256:  rec.Code.SHA256,
+		ImageDigest: e.ImageDigest,
 		Payload:     in.Payload,
 	})
 	if err != nil {
 		_ = revoke()
 		return refuse(errors.New("encode test-link request"))
 	}
-	// The encoded request, base64 payload included, keeps the NATS-era bound.
+	// The encoded frame, base64 payload included, is checked after encoding.
 	if int64(len(body)) > e.RequestLimit {
 		_ = revoke()
 		return refuse(fmt.Errorf("encoded request of %d bytes exceeds the %d-byte message limit", len(body), e.RequestLimit))

spinifex/runtime/roles/awsgw/lambda_testlink_executor_test.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"runtime"
 	"strings"
 	"testing"
 	"time"
@@ -114,8 +115,11 @@ func executor(n *fakeControl, presignErr error, store *execution.Store) *testLin
 	return executorWith(n, &fakeSessions{}, presignErr, store)
 }
 
+// testImageDigest stands in for the build's pinned runtime image digest.
+const testImageDigest = "ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12"
+
 func executorWith(n *fakeControl, sessions *fakeSessions, presignErr error, store *execution.Store) *testLinkExecutor {
-	return &testLinkExecutor{Control: n, RequestLimit: n.max, Sessions: sessions, Node: "node1",
+	return &testLinkExecutor{Control: n, RequestLimit: n.max, Sessions: sessions, Node: "node1", ImageDigest: testImageDigest,
 		Codes:  func(context.Context) (codePresigner, error) { return fakePresigner{err: presignErr}, nil },
 		Grants: func(context.Context) (grantIssuer, error) { return store, nil },
 	}
@@ -161,6 +165,25 @@ func TestLambdaTestLinkExecutor_RefusesWithoutSending(t *testing.T) {
 	}
 }
 
+// A gateway built without a valid image pin refuses before presigning,
+// granting, issuing or connecting.
+func TestLambdaTestLinkExecutor_InvalidImagePinRefusesBeforeConnecting(t *testing.T) {
+	for name, pin := range map[string]string{"missing": "", "malformed": "not-a-digest"} {
+		t.Run(name, func(t *testing.T) {
+			n, store := &fakeControl{max: 1 << 20, reply: &testlink.Reply{Outcome: testlink.OutcomeDone}}, grantStore(t)
+			sessions := &fakeSessions{}
+			e := executorWith(n, sessions, nil, store)
+			e.ImageDigest = pin
+			_, err := e.InvokeSync(context.Background(), invocation(nil))
+			var refusal *awsapi.RefusalError
+			require.ErrorAs(t, err, &refusal)
+			require.Empty(t, n.sent)
+			require.Empty(t, sessions.issued)
+			require.Empty(t, grantState(t, store), "no grant taken")
+		})
+	}
+}
+
 func TestLambdaTestLinkExecutor_BusyNodeSendsNothing(t *testing.T) {
 	store := grantStore(t)
 	held := &fakeControl{max: 1 << 20, err: errors.New("no reply")}
@@ -277,6 +300,7 @@ func TestLambdaTestLinkExecutor_IssuesForTheAttemptAndSendsOnTheSocket(t *testin
 	var sent testlink.Request
 	require.NoError(t, json.Unmarshal(n.sent[0], &sent))
 	require.Equal(t, testScope.FunctionARN+":$LATEST", sent.FunctionARN, "the runtime still sees the invoked ARN")
+	require.Equal(t, testImageDigest, sent.ImageDigest, "the daemon checks the gateway's pinned image")
 	require.NotContains(t, string(n.sent[0]), testSecret, "the credential travels beside the request, not in it")
 	require.Equal(t, testSecret, n.creds[0].SecretAccessKey)
 
@@ -399,3 +423,161 @@ func TestLambdaTestLinkExecutor_SilentDaemonIsUnknownAndRevoked(t *testing.T) {
 	require.Less(t, time.Since(start), 10*time.Second)
 	require.Len(t, sessions.revoked, 1)
 }
+
+// withdrawFault fails Withdraw before or after the owner writes it.
+type withdrawFault struct {
+	*execution.Store
+	afterWrite bool
+}
+
+func (w withdrawFault) Withdraw(ctx context.Context, g execution.Grant) error {
+	if !w.afterWrite {
+		return errors.New("owner unavailable")
+	}
+	if err := w.Store.Withdraw(ctx, g); err != nil {
+		return err
+	}
+	return execution.ErrOutcomeUnknown
+}
+
+// A daemon refusal with a failed Withdraw leaves the grant startable: a
+// replayed Start succeeds. A lost Withdraw acknowledgement leaves it fenced.
+func TestLambdaTestLinkExecutor_RefusalWithdrawFaults(t *testing.T) {
+	for name, tc := range map[string]struct {
+		afterWrite bool
+		state      string
+		replay     bool
+	}{
+		"fails before write": {state: execution.StateGranted, replay: true},
+		"ack lost":           {afterWrite: true, state: execution.StateWithdrawn},
+	} {
+		t.Run(name, func(t *testing.T) {
+			store := grantStore(t)
+			n := &fakeControl{max: 1 << 20, reply: &testlink.Reply{Outcome: testlink.OutcomeRefused, Error: "image differs"}}
+			e := executor(n, nil, store)
+			e.Grants = func(context.Context) (grantIssuer, error) { return withdrawFault{store, tc.afterWrite}, nil }
+			_, err := e.InvokeSync(context.Background(), invocation(nil))
+			require.Error(t, err)
+			require.Len(t, n.sent, 1)
+			require.Equal(t, tc.state, grantState(t, store))
+
+			var sent testlink.Request
+			require.NoError(t, json.Unmarshal(n.sent[0], &sent))
+			err = store.Start(context.Background(), sent.Grant, sent.Identity())
+			if tc.replay {
+				require.NoError(t, err, "an unwithdrawn grant can still start")
+			} else {
+				require.Error(t, err, "a withdrawn grant cannot start")
+			}
+		})
+	}
+}
+
+// The digest field counts toward the request bound: the exact encoded size
+// is sent and one byte less is refused unsent.
+func TestLambdaTestLinkExecutor_RequestBoundIncludesImageDigest(t *testing.T) {
+	probe := &fakeControl{max: 1 << 20, reply: &testlink.Reply{Outcome: testlink.OutcomeDone}}
+	_, err := executor(probe, nil, grantStore(t)).InvokeSync(context.Background(), invocation(make([]byte, 1000)))
+	require.NoError(t, err)
+	require.Len(t, probe.sent, 1)
+	require.Contains(t, string(probe.sent[0]), `"image_digest":"`+testImageDigest+`"`)
+	size := int64(len(probe.sent[0]))
+
+	at := &fakeControl{max: size, reply: &testlink.Reply{Outcome: testlink.OutcomeDone}}
+	_, err = executor(at, nil, grantStore(t)).InvokeSync(context.Background(), invocation(make([]byte, 1000)))
+	require.NoError(t, err)
+	require.Len(t, at.sent, 1)
+	require.Equal(t, size, int64(len(at.sent[0])), "the generated IDs are fixed-length, so the at-limit request is the probed size")
+
+	under := &fakeControl{max: size - 1}
+	_, err = executor(under, nil, grantStore(t)).InvokeSync(context.Background(), invocation(make([]byte, 1000)))
+	var refusal *awsapi.RefusalError
+	require.ErrorAs(t, err, &refusal)
+	require.Empty(t, under.sent)
+}
+
+// realCodeStore signs download links with the production code store, using a
+// session of the shape STS issues and a representative IPv6 off-host endpoint.
+func realCodeStore(t *testing.T) *codestore.Store {
+	t.Helper()
+	store, err := codestore.Open(codestore.OpenConfig{
+		Region: "ap-southeast-2", Endpoint: "http://127.0.0.1:1", SigningRegion: "ap-southeast-2",
+		PresignEndpoint: "https://[2001:db8:ffff:ffff:ffff:ffff:ffff:ffff]:65535",
+		Assume: func() (*sts.AssumeRoleOutput, error) {
+			return &sts.AssumeRoleOutput{Credentials: &sts.Credentials{
+				AccessKeyId: aws.String("ASIA" + strings.Repeat("K", 16)), SecretAccessKey: aws.String(strings.Repeat("s", 40)),
+				SessionToken: aws.String(strings.Repeat("t", 44)), Expiration: aws.Time(time.Now().Add(time.Hour)),
+			}}, nil
+		},
+	})
+	require.NoError(t, err)
+	return store
+}
+
+// longestInvocation carries the longest function name and qualifier Invoke
+// accepts, full-length generated IDs, and the given payload and handler.
+func longestInvocation(payload []byte, handler string) awsapi.SyncInvocation {
+	in := invocation(payload)
+	in.Qualifier = strings.Repeat("q", 128)
+	in.Function.FunctionName = strings.Repeat("n", 64)
+	in.Function.FunctionID = strings.Repeat("f", 32)
+	in.Function.Code.RevisionID = strings.Repeat("r", 32)
+	in.Function.Handler = handler
+	return in
+}
+
+func TestLambdaTestLinkExecutor_SenderBoundsAtFrameLimit(t *testing.T) {
+	codes := realCodeStore(t)
+	run := func(in awsapi.SyncInvocation) (*fakeControl, *fakeSessions, *execution.Store, error) {
+		n := &fakeControl{max: testlink.MaxFrameBytes, reply: &testlink.Reply{Outcome: testlink.OutcomeDone}}
+		sessions, store := &fakeSessions{}, grantStore(t)
+		e := executorWith(n, sessions, nil, store)
+		e.Codes = func(context.Context) (codePresigner, error) { return codes, nil }
+		_, err := e.InvokeSync(context.Background(), in)
+		return n, sessions, store, err
+	}
+	refusedUnsent := func(n *fakeControl, sessions *fakeSessions, store *execution.Store, err error) {
+		t.Helper()
+		var refusal *awsapi.RefusalError
+		require.ErrorAs(t, err, &refusal)
+		require.Empty(t, n.sent)
+		require.Len(t, sessions.revoked, 1, "the issued session is revoked")
+		require.Equal(t, execution.StateWithdrawn, grantState(t, store))
+	}
+
+	full, _, _, err := run(longestInvocation(make([]byte, testlink.MaxPayloadBytes), "bootstrap.h"))
+	require.NoError(t, err)
+	require.Len(t, full.sent, 1)
+	sent := string(full.sent[0])
+	// Asserted without printing, since the frame holds a signed URL.
+	require.True(t, strings.Contains(sent, `"code_url":"https://[2001:db8:ffff:ffff:ffff:ffff:ffff:ffff]:65535/`), "the off-host endpoint is signed")
+	require.True(t, strings.Contains(sent, "X-Amz-Signature="), "the code URL is a real presigned link")
+	headroom := testlink.MaxFrameBytes - int64(len(full.sent[0]))
+	t.Logf("full-size request: %d of %d frame bytes, %d bytes of headroom", len(full.sent[0]), testlink.MaxFrameBytes, headroom)
+
+	// The generated IDs and signed URL are fixed-length, so a longer handler
+	// moves the frame by exactly its extra bytes.
+	atLimit := "bootstrap.h" + strings.Repeat("h", int(headroom))
+	at, _, _, err := run(longestInvocation(make([]byte, testlink.MaxPayloadBytes), atLimit))
+	require.NoError(t, err)
+	require.Len(t, at.sent, 1)
+	require.Equal(t, testlink.MaxFrameBytes, int64(len(at.sent[0])))
+
+	refusedUnsent(run(longestInvocation(make([]byte, testlink.MaxPayloadBytes), atLimit+"h")))
+
+	// Refused before encoding: base64 alone would allocate 8 MiB, so a run
+	// that allocates under the payload size never encoded it.
+	over := longestInvocation(make([]byte, testlink.MaxPayloadBytes+1), "bootstrap.h")
+	n := &fakeControl{max: testlink.MaxFrameBytes}
+	sessions, store := &fakeSessions{}, grantStore(t)
+	e := executorWith(n, sessions, nil, store)
+	e.Codes = func(context.Context) (codePresigner, error) { return codes, nil }
+	var before, after runtime.MemStats
+	runtime.ReadMemStats(&before)
+	_, err = e.InvokeSync(context.Background(), over)
+	runtime.ReadMemStats(&after)
+	allocated := after.TotalAlloc - before.TotalAlloc
+	t.Logf("6 MiB + 1 refusal: %d bytes in %d allocations, process-wide", allocated, after.Mallocs-before.Mallocs)
+	require.Less(t, allocated, uint64(testlink.MaxPayloadBytes))
+	refusedUnsent(n, sessions, store, err)
+}

spinifex/runtime/roles/awsgw/lambda_testlink_pin_test.go

@@ -0,0 +1,35 @@
+//go:build lambdatestlink
+
+//test:in-package — reads the unexported ldflags image pin
+package awsgw
+
+import (
+	"os"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/testlink"
+)
+
+// TestLambdaImageDigestPin runs from make build-lambda-testlink with the
+// daemon's ldflags: the gateway carries the same image digest as the daemon.
+func TestLambdaImageDigestPin(t *testing.T) {
+	want := os.Getenv("LAMBDA_IMAGE_DIGEST")
+	if want == "" {
+		t.Skip("run by make build-lambda-testlink with a pinned image digest")
+	}
+	if lambdaImageDigest != want || !testlink.ValidDigest(lambdaImageDigest) {
+		t.Fatalf("pinned image digest %q, want %q", lambdaImageDigest, want)
+	}
+}
+
+// The gateway dials the production control name, so a mixed-version daemon on
+// the previous name is never reached.
+func TestLambdaTestLinkControlUsesTheProductionName(t *testing.T) {
+	c := testLinkControl(4242)
+	if c.Name != testlink.ControlName || c.DaemonUID != 4242 {
+		t.Fatalf("control client %q uid %d, want %q uid 4242", c.Name, c.DaemonUID, testlink.ControlName)
+	}
+	if c.RequestLimit != testlink.MaxFrameBytes || c.ReplyLimit != testlink.MaxFrameBytes {
+		t.Fatalf("control limits %d/%d, want the declared frame bound %d", c.RequestLimit, c.ReplyLimit, testlink.MaxFrameBytes)
+	}
+}

spinifex/runtime/roles/awsgw/lambda_testlink_seam_env_test.go

@@ -161,7 +161,7 @@ func startSeamDaemon(t *testing.T, d *testlink.Daemon, guest guestFunc) *seamDae
 	ln, err := testlink.ListenControl(name)
 	require.NoError(t, err)
 	s := &seamDaemon{entered: make(chan testlink.Request, 4), done: make(chan testlink.Reply, 4)}
-	srv := &testlink.ControlServer{GatewayUID: os.Getuid(), RequestLimit: 1 << 20, ReplyLimit: 1 << 20, IOTimeout: 5 * time.Second,
+	srv := &testlink.ControlServer{GatewayUID: os.Getuid(), RequestLimit: testlink.MaxFrameBytes, ReplyLimit: testlink.MaxFrameBytes, IOTimeout: 5 * time.Second,
 		Handle: func(ctx context.Context, req []byte, cred testlink.Credential) testlink.Reply {
 			var parsed testlink.Request
 			_ = json.Unmarshal(req, &parsed)
@@ -189,7 +189,7 @@ func startSeamDaemon(t *testing.T, d *testlink.Daemon, guest guestFunc) *seamDae
 			t.Errorf("control server did not stop within %s", seamWait)
 		}
 	})
-	s.client = &testlink.ControlClient{Name: name, DaemonUID: os.Getuid(), RequestLimit: 1 << 20, ReplyLimit: 1 << 20, IOTimeout: 5 * time.Second}
+	s.client = &testlink.ControlClient{Name: name, DaemonUID: os.Getuid(), RequestLimit: testlink.MaxFrameBytes, ReplyLimit: testlink.MaxFrameBytes, IOTimeout: 5 * time.Second}
 	return s
 }
 

spinifex/runtime/roles/awsgw/lambda_testlink_seam_payload_test.go

@@ -0,0 +1,113 @@
+//go:build lambdatestlink && lambdatestlinkseam
+
+//test:in-package — drives the unexported test-link executor and seam fixtures
+package awsgw
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/runtimeapi"
+	lambdaawsapi "github.com/mulgadc/spinifex/spinifex/domains/lambda/awsapi"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/testlink"
+	"github.com/mulgadc/spinifex/internal/testkit/gwfix"
+)
+
+// jsonString returns a JSON string literal of exactly n bytes.
+func jsonString(n int) []byte {
+	return []byte(`"` + strings.Repeat("a", n-2) + `"`)
+}
+
+// invokeComposedBody is invokeComposedIn with a signed Invoke carrying body.
+func invokeComposedBody(t *testing.T, body []byte, codeURL string, client *http.Client, guest guestFunc) seamRun {
+	t.Helper()
+	env := newSeamEnv(t)
+	exec, daemon, grants, presigns := composedExecutor(t, env, codeURL, client, guest)
+	reg := lambdaawsapi.NewRegistration(lambdaawsapi.Deps{Meta: seamFunction{timeout: 30, role: env.roleARN}, Exec: exec})
+	req := httptest.NewRequest(http.MethodPost, "/2015-03-31/functions/f/invocations", bytes.NewReader(body))
+	req.Host = "localhost:9999"
+	gwfix.SignRequestAs(t, req, body, gwfix.AccessKey, gwfix.SecretKey, lambdaawsapi.ServiceName, time.Now().UTC())
+	w := httptest.NewRecorder()
+	defaultGateway(t, lambdaTestRoutes(reg)...).ServeHTTP(w, req)
+	resp := w.Result()
+	out, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	reqs, replies := daemon.recorded()
+	return seamRun{env: env, resp: resp, body: out, requests: reqs, replies: replies, presigns: presigns.Load(), grants: grants.issued.Load()}
+}
+
+// echoGuest is a function runtime that returns its event unchanged and
+// reports its own result on done.
+func echoGuest(done chan<- error) func(base string) {
+	client := &http.Client{Timeout: 30 * time.Second}
+	return func(base string) {
+		done <- func() error {
+			resp, err := client.Get(base + "invocation/next")
+			if err != nil {
+				return err
+			}
+			event, err := io.ReadAll(resp.Body)
+			_ = resp.Body.Close()
+			if err != nil {
+				return err
+			}
+			if resp.StatusCode != http.StatusOK {
+				return fmt.Errorf("next: status %d", resp.StatusCode)
+			}
+			id := resp.Header.Get("Lambda-Runtime-Aws-Request-Id")
+			r, err := client.Post(base+"invocation/"+id+"/response", "application/json", bytes.NewReader(event))
+			if err != nil {
+				return err
+			}
+			_, _ = io.Copy(io.Discard, r.Body)
+			_ = r.Body.Close()
+			if r.StatusCode != http.StatusAccepted {
+				return fmt.Errorf("response: status %d", r.StatusCode)
+			}
+			return nil
+		}()
+	}
+}
+
+// A 6 MiB event crosses the real gateway, control socket and guest link,
+// and the 6 MiB response returns byte-identical. The seam daemon uses
+// HandleWithGuest, so the busy, image and fit checks are tested elsewhere.
+func TestLambdaTestLinkSeam_SixMiBRequestAndResponseRoundTrip(t *testing.T) {
+	require.Equal(t, testlink.MaxPayloadBytes, runtimeapi.MaxResponseBytes)
+	event := jsonString(testlink.MaxPayloadBytes)
+	srv := codeServer(t)
+	guestDone := make(chan error, 1)
+	run := invokeComposedBody(t, event, srv.URL+"/code", srv.Client(), agentGuest(0, echoGuest(guestDone)))
+	require.Equal(t, http.StatusOK, run.resp.StatusCode, string(run.body[:min(len(run.body), 300)]))
+	require.Empty(t, run.resp.Header.Get("X-Amz-Function-Error"))
+	require.Len(t, run.body, testlink.MaxPayloadBytes)
+	require.True(t, bytes.Equal(event, run.body), "response differs from the event")
+	require.Len(t, run.replies, 1)
+	select {
+	case err := <-guestDone:
+		require.NoError(t, err, "the echo runtime completed")
+	case <-time.After(30 * time.Second):
+		t.Fatal("the echo runtime did not complete")
+	}
+}
+
+// A 6 MiB + 1 event is refused with 413 before anything is presigned,
+// granted or sent to the daemon.
+func TestLambdaTestLinkSeam_OverSixMiBRequestRefusedBeforeDispatch(t *testing.T) {
+	srv := codeServer(t)
+	run := invokeComposedBody(t, jsonString(testlink.MaxPayloadBytes+1), srv.URL+"/code", srv.Client(), agentGuest(0, echoGuest(make(chan error, 1))))
+	require.Equal(t, http.StatusRequestEntityTooLarge, run.resp.StatusCode)
+	require.Contains(t, run.resp.Header.Get("X-Amzn-Errortype"), "RequestEntityTooLargeException")
+	require.Zero(t, run.presigns)
+	require.Zero(t, run.grants)
+	require.Empty(t, run.requests)
+	require.Zero(t, run.env.sessionCount(t))
+}

spinifex/runtime/roles/awsgw/lambda_testlink_seam_test.go

@@ -212,7 +212,7 @@ func composedExecutor(t *testing.T, env *seamEnv, codeURL string, client *http.C
 	grants := &countedGrants{grantIssuer: env.grants}
 	presigns := &atomic.Int32{}
 	daemon := startSeamDaemon(t, &testlink.Daemon{Node: "node1", Grants: env.grants, HTTP: client}, guest)
-	exec := &testLinkExecutor{Control: daemon.client, RequestLimit: 1 << 20, Sessions: env.sts, Node: "node1",
+	exec := &testLinkExecutor{Control: daemon.client, RequestLimit: testlink.MaxFrameBytes, Sessions: env.sts, Node: "node1", ImageDigest: testImageDigest,
 		Codes:  func(context.Context) (codePresigner, error) { return urlPresigner{codeURL, presigns}, nil },
 		Grants: func(context.Context) (grantIssuer, error) { return grants, nil },
 	}