feat(lambda): run the test-link agent over the authenticated TCP link

The lambdatestlink agent now loads its identity, seals fw_cfg and dials the
host exactly as the normal agent does, then carries the test stream over
that mutual-TLS connection instead of a device path. Nothing tenant-facing
starts until the host's first valid frame: a refusal after the handshake,
an early close or a shutdown never starts the bootstrap. Each VM serves one
invocation and then stops.

The normal agent is unchanged and still accepts no invocations. The tagged
tests join the race tier for the two packages that carry the build tag.

Makefile

@@ -326,6 +326,7 @@ test-cover:
 test-race:
 	@echo -e "\n....Running tests with race detector for $(GO_PROJECT_NAME)...."
 	$(_Q)LOG_IGNORE=1 SPINIFEX_SQS_R3=1 go test $(GOTESTFLAGS) -race -timeout 300s ./spinifex/... ./cmd/... ./internal/... ./contracts/... ./architecture/... $(_RACEQ)
+	$(_Q)LOG_IGNORE=1 go test $(GOTESTFLAGS) -race -tags=lambdatestlink -timeout 120s ./cmd/lambda-agent/ ./spinifex/agents/lambda/hostlink/ $(_RACEQ)
 
 # Unit tests for in-repo GitHub Actions (e.g. .github/actions/e2e-analyze).
 # Kept out of `test-cover` so coverage % isn't diluted by CI-only tooling.

cmd/lambda-agent/connect.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+)
+
+// Exit statuses: the guest refused to start, the control connection failed,
+// or the host closed it.
+const (
+	exitRefused = 2
+	exitLink    = 1
+	exitClosed  = 0
+)
+
+type startup struct {
+	fwCfgDir  string
+	fwCfgRoot string
+	unload    func() error
+}
+
+func defaultStartup() startup {
+	return startup{fwCfgDir: hostlink.FwCfgDir, fwCfgRoot: hostlink.FwCfgRoot, unload: hostlink.UnloadFwCfg}
+}
+
+// connect loads the identity, seals fw_cfg whether or not the load succeeded,
+// then dials the host. Under TLS 1.3 a host refusal is seen only on first read.
+func (s startup) connect(ctx context.Context) (*tls.Conn, int, error) {
+	id, loadErr := hostlink.LoadIdentity(s.fwCfgDir)
+	if err := hostlink.SealFwCfg(s.fwCfgRoot, s.unload); err != nil {
+		return nil, exitRefused, errors.Join(loadErr, err)
+	}
+	if loadErr != nil {
+		return nil, exitRefused, loadErr
+	}
+	conn, err := hostlink.Dial(ctx, id)
+	if err != nil {
+		if ctx.Err() != nil {
+			err = fmt.Errorf("%w: %w", err, context.Cause(ctx))
+		}
+		return nil, exitLink, err
+	}
+	return conn, 0, nil
+}

cmd/lambda-agent/helpers_test.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	lambdav1 "github.com/mulgadc/spinifex/contracts/lambda/v1"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/guestconn"
+	"github.com/stretchr/testify/require"
+)
+
+const scope lambdav1.Scope = "ap-southeast-2/000000000001/fn-1/env-1/gen-1"
+
+// guest is a fake fw_cfg tree; unload removes it as unloading the module would.
+type guest struct {
+	root     string
+	unloads  int
+	sealedAt chan struct{}
+}
+
+func newGuest(t *testing.T, entries map[string][]byte) *guest {
+	t.Helper()
+	g := &guest{root: filepath.Join(t.TempDir(), "qemu_fw_cfg"), sealedAt: make(chan struct{}, 1)}
+	for name, b := range entries {
+		dir := filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix, name)
+		require.NoError(t, os.MkdirAll(dir, 0o700))
+		require.NoError(t, os.WriteFile(filepath.Join(dir, "raw"), b, 0o600))
+	}
+	return g
+}
+
+func (g *guest) startup() startup {
+	return startup{
+		fwCfgDir:  filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix),
+		fwCfgRoot: g.root,
+		unload: func() error {
+			g.unloads++
+			g.sealedAt <- struct{}{}
+			return os.RemoveAll(g.root)
+		},
+	}
+}
+
+type host struct {
+	ln   *net.TCPListener
+	gen  *guestconn.Generation
+	conn chan accepted
+}
+
+func newHost(t *testing.T) *host {
+	t.Helper()
+	ln, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	require.NoError(t, err)
+	t.Cleanup(func() { ln.Close() })
+	gen, err := guestconn.NewGeneration(scope, time.Hour)
+	require.NoError(t, err)
+	return &host{ln: ln, gen: gen, conn: make(chan accepted, 1)}
+}
+
+func (h *host) entries() map[string][]byte { return h.gen.GuestEntries(h.ln.Addr().String()) }
+
+type accepted struct {
+	conn *tls.Conn
+	err  error
+}
+
+// accept runs the host end until ctx ends and reports what it accepted.
+func (h *host) accept(ctx context.Context) {
+	go func() {
+		c, err := h.gen.AcceptGuest(ctx, h.ln, 2*time.Second)
+		h.conn <- accepted{c, err}
+	}()
+}
+
+// guest waits for the host to authenticate the guest.
+func (h *host) guest(t *testing.T) *tls.Conn {
+	t.Helper()
+	a := wait(t, h.conn, "host to authenticate the guest")
+	require.NoError(t, a.err)
+	return a.conn
+}
+
+// wait bounds every test wait so a stalled host or startup fails promptly.
+func wait[T any](t *testing.T, ch <-chan T, what string) T {
+	t.Helper()
+	select {
+	case v := <-ch:
+		return v
+	case <-time.After(10 * time.Second):
+		t.Fatalf("timed out waiting for %s", what)
+		panic("unreachable")
+	}
+}

cmd/lambda-agent/main.go

@@ -6,8 +6,8 @@
 // never starts the function's bootstrap: the message protocol and grant
 // fencing do not exist yet. The agent never listens for the host.
 //
-// A lambdatestlink build instead relays invocations over an unauthenticated
-// test carrier for local proofs; it is not the guest protocol.
+// A lambdatestlink build opens the same connection, then relays one invocation
+// over a test carrier for local proofs; the carrier is not the guest protocol.
 package main
 
 import (

cmd/lambda-agent/start_tcp.go

@@ -9,56 +9,28 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
-
-	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
-)
-
-// Exit statuses: the guest refused to start, the control connection failed,
-// or the host closed it.
-const (
-	exitRefused = 2
-	exitLink    = 1
-	exitClosed  = 0
 )
 
 // errNoProtocol is returned when the host sends anything: this build defines no
 // message protocol, so it accepts no work.
 var errNoProtocol = errors.New("host sent data, but this build accepts no invocations")
 
-type startup struct {
-	fwCfgDir  string
-	fwCfgRoot string
-	unload    func() error
-}
-
 func start(ctx context.Context) int {
-	s := startup{fwCfgDir: hostlink.FwCfgDir, fwCfgRoot: hostlink.FwCfgRoot, unload: hostlink.UnloadFwCfg}
-	code, err := s.run(ctx)
+	code, err := defaultStartup().run(ctx)
 	if err != nil {
 		slog.Error("lambda-agent stopping", "exit", code, "err", err)
 	}
 	return code
 }
 
-// run loads the identity, seals fw_cfg whether or not the load succeeded,
-// then authenticates to the host and holds the connection until it closes.
+// run authenticates to the host and holds the connection until it closes.
 func (s startup) run(ctx context.Context) (int, error) {
-	id, loadErr := hostlink.LoadIdentity(s.fwCfgDir)
-	if err := hostlink.SealFwCfg(s.fwCfgRoot, s.unload); err != nil {
-		return exitRefused, errors.Join(loadErr, err)
-	}
-	if loadErr != nil {
-		return exitRefused, loadErr
-	}
-	conn, err := hostlink.Dial(ctx, id)
+	conn, code, err := s.connect(ctx)
 	if err != nil {
-		if ctx.Err() != nil {
-			err = fmt.Errorf("%w: %w", err, context.Cause(ctx))
-		}
-		return exitLink, err
+		return code, err
 	}
 	defer conn.Close()
-	slog.Info("control connection open; invocations are not enabled in this build", "host", id.HostAddr)
+	slog.Info("control connection open; invocations are not enabled in this build", "host", conn.RemoteAddr().String())
 	if err := awaitHostClose(ctx, conn); err != nil {
 		return exitLink, err
 	}

cmd/lambda-agent/start_tcp_test.go

@@ -4,11 +4,9 @@ package main
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"net"
 	"os"
-	"path/filepath"
 	"testing"
 	"time"
 
@@ -18,89 +16,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-const scope lambdav1.Scope = "ap-southeast-2/000000000001/fn-1/env-1/gen-1"
-
-// guest is a fake fw_cfg tree; unload removes it as unloading the module would.
-type guest struct {
-	root     string
-	unloads  int
-	sealedAt chan struct{}
-}
-
-func newGuest(t *testing.T, entries map[string][]byte) *guest {
-	t.Helper()
-	g := &guest{root: filepath.Join(t.TempDir(), "qemu_fw_cfg"), sealedAt: make(chan struct{}, 1)}
-	for name, b := range entries {
-		dir := filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix, name)
-		require.NoError(t, os.MkdirAll(dir, 0o700))
-		require.NoError(t, os.WriteFile(filepath.Join(dir, "raw"), b, 0o600))
-	}
-	return g
-}
-
-func (g *guest) startup() startup {
-	return startup{
-		fwCfgDir:  filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix),
-		fwCfgRoot: g.root,
-		unload: func() error {
-			g.unloads++
-			g.sealedAt <- struct{}{}
-			return os.RemoveAll(g.root)
-		},
-	}
-}
-
-type host struct {
-	ln   *net.TCPListener
-	gen  *guestconn.Generation
-	conn chan accepted
-}
-
-func newHost(t *testing.T) *host {
-	t.Helper()
-	ln, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)})
-	require.NoError(t, err)
-	t.Cleanup(func() { ln.Close() })
-	gen, err := guestconn.NewGeneration(scope, time.Hour)
-	require.NoError(t, err)
-	return &host{ln: ln, gen: gen, conn: make(chan accepted, 1)}
-}
-
-func (h *host) entries() map[string][]byte { return h.gen.GuestEntries(h.ln.Addr().String()) }
-
-type accepted struct {
-	conn *tls.Conn
-	err  error
-}
-
-// accept runs the host end until ctx ends and reports what it accepted.
-func (h *host) accept(ctx context.Context) {
-	go func() {
-		c, err := h.gen.AcceptGuest(ctx, h.ln, 2*time.Second)
-		h.conn <- accepted{c, err}
-	}()
-}
-
-// guest waits for the host to authenticate the guest.
-func (h *host) guest(t *testing.T) *tls.Conn {
-	t.Helper()
-	a := wait(t, h.conn, "host to authenticate the guest")
-	require.NoError(t, a.err)
-	return a.conn
-}
-
-// wait bounds every test wait so a stalled host or startup fails promptly.
-func wait[T any](t *testing.T, ch <-chan T, what string) T {
-	t.Helper()
-	select {
-	case v := <-ch:
-		return v
-	case <-time.After(10 * time.Second):
-		t.Fatalf("timed out waiting for %s", what)
-		panic("unreachable")
-	}
-}
-
 type result struct {
 	code int
 	err  error

cmd/lambda-agent/start_testlink.go

@@ -11,6 +11,7 @@ import (
 	"net"
 	"net/http"
 	"os"
+	"os/exec"
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/agents/lambda/agent"
@@ -23,34 +24,56 @@ const (
 	taskRoot    = "/var/task"
 )
 
-var (
-	linkPath = flag.String("link", os.Getenv("LAMBDA_AGENT_LINK"), "test carrier device path")
-	handler  = flag.String("handler", os.Getenv("_HANDLER"), "function handler passed to the bootstrap")
-)
+var handler = flag.String("handler", os.Getenv("_HANDLER"), "function handler passed to the bootstrap")
 
-func start(parent context.Context) int {
-	if *linkPath == "" {
-		slog.Error("no test carrier configured; refusing to start")
-		return 2
+// testlink relays one invocation over the test carrier on the authenticated
+// connection; the carrier is not the guest protocol.
+type testlink struct {
+	startup
+	runtimeAddr    string
+	startBootstrap func(ctx context.Context, env []string) (*exec.Cmd, error)
+}
+
+func start(ctx context.Context) int {
+	t := testlink{
+		startup:     defaultStartup(),
+		runtimeAddr: runtimeAddr,
+		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
+			return agent.StartBootstrap(ctx, taskRoot+"/bootstrap", env, os.Stdout, os.Stderr)
+		},
 	}
-	// The bootstrap is tenant code; fw_cfg must be unreadable before it runs.
-	if err := hostlink.SealFwCfg(hostlink.FwCfgRoot, hostlink.UnloadFwCfg); err != nil {
-		slog.Error("seal fw_cfg; refusing to start", "err", err)
-		return 2
-	}
-	dev, err := os.OpenFile(*linkPath, os.O_RDWR, 0)
+	code, err := t.run(ctx)
 	if err != nil {
-		slog.Error("open test carrier; refusing to start", "err", err)
-		return 2
+		slog.Error("lambda-agent stopping", "exit", code, "err", err)
+	}
+	return code
+}
+
+func (t testlink) run(parent context.Context) (int, error) {
+	conn, code, err := t.connect(parent)
+	if err != nil {
+		return code, err
+	}
+	defer conn.Close()
+	link := hostlink.NewTestStream(conn)
+
+	// The handshake can return before the host refuses the guest, so nothing
+	// tenant-facing starts until the host's first valid frame arrives.
+	first, err := link.Next(parent)
+	if errors.Is(err, hostlink.ErrShutdown) {
+		return exitClosed, nil
+	}
+	if err != nil {
+		return exitLink, fmt.Errorf("first frame from host: %w", err)
+	}
+	if parent.Err() != nil {
+		return exitLink, context.Cause(parent)
 	}
-	defer dev.Close()
-	link := hostlink.NewTestStream(dev)
 
 	srv := runtimeapi.New()
-	ln, err := net.Listen("tcp", runtimeAddr)
+	ln, err := net.Listen("tcp", t.runtimeAddr)
 	if err != nil {
-		slog.Error("listen runtime api", "err", err)
-		return 1
+		return exitLink, fmt.Errorf("listen runtime api: %w", err)
 	}
 	hs := &http.Server{Handler: srv.Handler(), ReadHeaderTimeout: 10 * time.Second}
 	go func() { _ = hs.Serve(ln) }()
@@ -58,16 +81,14 @@ func start(parent context.Context) int {
 
 	ctx, cancel := context.WithCancelCause(parent)
 	defer cancel(nil)
-
-	cmd, err := agent.StartBootstrap(ctx, taskRoot+"/bootstrap", []string{
-		"AWS_LAMBDA_RUNTIME_API=" + runtimeAddr,
+	cmd, err := t.startBootstrap(ctx, []string{
+		"AWS_LAMBDA_RUNTIME_API=" + ln.Addr().String(),
 		"_HANDLER=" + *handler,
 		"LAMBDA_TASK_ROOT=" + taskRoot,
 		"PATH=/bin:/sbin:/usr/bin",
-	}, os.Stdout, os.Stderr)
+	})
 	if err != nil {
-		slog.Error("start bootstrap", "err", err)
-		return 1
+		return exitLink, fmt.Errorf("start bootstrap: %w", err)
 	}
 	exited := make(chan struct{})
 	go func() {
@@ -76,12 +97,32 @@ func start(parent context.Context) int {
 		cancel(fmt.Errorf("bootstrap exited: %w", err))
 	}()
 
-	err = agent.Run(ctx, link, srv)
+	err = agent.Run(ctx, &oneInvocation{Link: link, first: &first}, srv)
 	cancel(errors.New("agent stopping"))
 	<-exited
 	if err != nil {
-		slog.Error("agent stopped", "err", err)
-		return 1
+		return exitLink, err
 	}
-	return 0
+	return exitClosed, nil
+}
+
+// oneInvocation serves the host's first request and then stops the agent:
+// each test VM runs a single invocation.
+type oneInvocation struct {
+	hostlink.Link
+	first *hostlink.Request
+}
+
+var _ hostlink.Link = (*oneInvocation)(nil)
+
+func (o *oneInvocation) Next(ctx context.Context) (hostlink.Request, error) {
+	if err := ctx.Err(); err != nil {
+		return hostlink.Request{}, err
+	}
+	if o.first == nil {
+		return hostlink.Request{}, hostlink.ErrShutdown
+	}
+	req := *o.first
+	o.first = nil
+	return req, nil
 }

cmd/lambda-agent/start_testlink_test.go

@@ -0,0 +1,179 @@
+//go:build lambdatestlink
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"testing"
+	"time"
+
+	lambdav1 "github.com/mulgadc/spinifex/contracts/lambda/v1"
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/guestconn"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const helperEnv = "LAMBDA_AGENT_TEST_BOOTSTRAP"
+
+// TestHelperBootstrap is the fake function bootstrap, run as a subprocess: it
+// echoes one invocation through the Runtime API, then waits for the next.
+func TestHelperBootstrap(t *testing.T) {
+	if os.Getenv(helperEnv) != "1" {
+		return
+	}
+	base := "http://" + os.Getenv("AWS_LAMBDA_RUNTIME_API") + "/2018-06-01/runtime/invocation/"
+	resp, err := http.Get(base + "next")
+	require.NoError(t, err)
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	resp.Body.Close()
+	id := resp.Header.Get("Lambda-Runtime-Aws-Request-Id")
+	post, err := http.Post(base+id+"/response", "application/json", bytes.NewReader(body))
+	require.NoError(t, err)
+	post.Body.Close()
+	_, _ = http.Get(base + "next")
+}
+
+type bootstrapStarts struct {
+	n       int
+	started chan struct{}
+}
+
+func (b *bootstrapStarts) testlink(g *guest) testlink {
+	b.started = make(chan struct{}, 1)
+	return testlink{
+		startup:     g.startup(),
+		runtimeAddr: "127.0.0.1:0",
+		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
+			b.n++
+			b.started <- struct{}{}
+			cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestHelperBootstrap$")
+			cmd.Env = append(env, helperEnv+"=1")
+			return cmd, cmd.Start()
+		},
+	}
+}
+
+func runTestlinkAsync(ctx context.Context, tl testlink) <-chan result {
+	out := make(chan result, 1)
+	go func() {
+		code, err := tl.run(ctx)
+		out <- result{code, err}
+	}()
+	return out
+}
+
+type result struct {
+	code int
+	err  error
+}
+
+func TestTestlinkServesOneInvocationThenStops(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	var b bootstrapStarts
+	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
+
+	conn := h.guest(t)
+	defer conn.Close()
+	select {
+	case <-b.started:
+		t.Fatal("bootstrap started before the host sent a frame")
+	case <-time.After(100 * time.Millisecond):
+	}
+	hs := hostlink.NewTestStream(conn)
+	req := hostlink.Request{GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", DeadlineMs: time.Now().Add(5 * time.Second).UnixMilli(), Payload: []byte(`{"a":40,"b":2}`)}
+	require.NoError(t, hs.HostSend(req))
+	got, err := hs.HostResult(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, hostlink.Response{GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", Payload: req.Payload}, got)
+
+	// A second invocation on the same VM is never served. The agent may
+	// already have closed the connection, so a failed send is a refusal too.
+	second := req
+	second.RequestID = "r-2"
+	sendErr := hs.HostSend(second)
+	res := wait(t, done, "agent to stop after one invocation")
+	assert.Equal(t, exitClosed, res.code)
+	assert.NoError(t, res.err)
+	if sendErr == nil {
+		_, err = hs.HostResult(ctx)
+		assert.Error(t, err, "agent answered a second invocation")
+	}
+	assert.Equal(t, 1, b.n)
+}
+
+func TestTestlinkShutdownBeforeInvocationNeverStartsBootstrap(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	var b bootstrapStarts
+	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
+
+	conn := h.guest(t)
+	defer conn.Close()
+	require.NoError(t, hostlink.NewTestStream(conn).HostShutdown())
+	res := wait(t, done, "agent to stop")
+	assert.Equal(t, exitClosed, res.code)
+	assert.NoError(t, res.err)
+	assert.Zero(t, b.n)
+}
+
+// The guest's handshake completes, but the host pins another generation's
+// guest; the refusal arrives on first read and the bootstrap never starts.
+func TestTestlinkHostRefusalNeverStartsBootstrap(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	other, err := guestconn.NewGeneration(scope, time.Hour)
+	require.NoError(t, err)
+	entries := h.entries()
+	wrong := other.GuestEntries(h.ln.Addr().String())
+	entries[lambdav1.EntryGuestKey], entries[lambdav1.EntryGuestCert] = wrong[lambdav1.EntryGuestKey], wrong[lambdav1.EntryGuestCert]
+	var b bootstrapStarts
+
+	res := wait(t, runTestlinkAsync(ctx, b.testlink(newGuest(t, entries))), "agent to stop")
+	assert.Equal(t, exitLink, res.code)
+	require.Error(t, res.err)
+	assert.Contains(t, res.err.Error(), "bad certificate")
+	assert.Zero(t, b.n)
+}
+
+func TestTestlinkHostCloseBeforeFrameNeverStartsBootstrap(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	var b bootstrapStarts
+	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
+
+	require.NoError(t, h.guest(t).Close())
+	res := wait(t, done, "agent to stop")
+	assert.Equal(t, exitLink, res.code)
+	assert.ErrorContains(t, res.err, "first frame from host")
+	assert.Zero(t, b.n)
+}
+
+// A cancelled context is reported before the buffered request is delivered,
+// and the request stays undelivered.
+func TestOneInvocationHonoursCancelledContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	o := &oneInvocation{first: &hostlink.Request{RequestID: "r-1"}}
+	_, err := o.Next(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+
+	req, err := o.Next(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, "r-1", req.RequestID)
+}