test(lambda): hold a daemon invocation while serving ends on listener failure

The wake test now waits for the start pass before storing code, so only
the code-stored wake can activate the second function.

contracts/lambda/v1/README.md

@@ -28,6 +28,9 @@ Its encoded frame is at most the base64 length of 6 MiB plus 64 KiB of metadata
 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.
+In the same builds, the function owner moves a created function with stored code from `Pending` to `Active` once this node's daemon serves the pinned image and the cluster configuration has exactly one node; a function with environment variables or a code copy over the test-link bound stays `Pending`, and `StateReason` names the unmet condition.
+Release builds never leave `Pending`.
+The Mulga deploy role installs a test-link build with `deploy_lambda_testlink=true`; this profile does not meet Q-199 fail-closed revocation, and its guests take management addresses whose record can expire (mulga-0hsub).
 
 **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/domains/lambda/function/activate.go

@@ -0,0 +1,155 @@
+package function
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Readiness reports why a function whose code is stored cannot run yet, or nil
+// when it can. Its error text becomes the function's StateReason.
+type Readiness func(ctx context.Context, rec Record) error
+
+// SetReadiness enables activation. Without it no function ever leaves Pending.
+func (s *MetaService) SetReadiness(r Readiness) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.ready = r
+	if s.nudge == nil {
+		s.nudge = make(chan struct{}, 1)
+	}
+}
+
+func (s *MetaService) readiness() (Readiness, chan struct{}) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.ready, s.nudge
+}
+
+// wake asks the activation loop for an early pass; it never blocks.
+func (s *MetaService) wake() {
+	if _, nudge := s.readiness(); nudge != nil {
+		select {
+		case nudge <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// RunActivation activates pending functions until ctx ends: once at start, on
+// every code-stored completion, and every interval, so a crash between storing
+// the code and activating it is converged by the next pass.
+func (s *MetaService) RunActivation(ctx context.Context, interval time.Duration) {
+	ready, nudge := s.readiness()
+	if ready == nil {
+		return
+	}
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		if err := s.ActivatePending(ctx); err != nil && ctx.Err() == nil {
+			slog.Warn("Lambda: activation pass incomplete; retrying", "err", err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-nudge:
+		case <-t.C:
+		}
+	}
+}
+
+// ActivatePending makes one pass over every account's functions, moving each
+// Pending function with stored code to Active when readiness allows it.
+func (s *MetaService) ActivatePending(ctx context.Context) error {
+	ready, _ := s.readiness()
+	if ready == nil {
+		return nil
+	}
+	var accounts []string
+	names := s.js.KeyValueStoreNames(ctx)
+	for name := range names.Name() {
+		if id, ok := strings.CutPrefix(name, KVBucketAccountPrefix); ok {
+			accounts = append(accounts, id)
+		}
+	}
+	if err := names.Error(); err != nil {
+		return fmt.Errorf("lambda: list account buckets: %w", err)
+	}
+	var errs []error
+	for _, accountID := range accounts {
+		if err := s.activateAccount(ctx, accountID, ready); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+func (s *MetaService) activateAccount(ctx context.Context, accountID string, ready Readiness) error {
+	kv, err := s.bucket(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	keys, err := kv.ListKeysFiltered(ctx, kvFunctionKeyPrefix+">")
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("lambda: list functions of account %s: %w", accountID, err)
+	}
+	var errs []error
+	for key := range keys.Keys() {
+		if err := s.activateOne(ctx, kv, key, ready); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// activateOne writes the readiness outcome by compare-and-swap at the revision
+// it read, so a concurrent update or delete wins and is never overwritten.
+func (s *MetaService) activateOne(ctx context.Context, kv jetstream.KeyValue, key string, ready Readiness) error {
+	entry, err := kv.Get(ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) || errors.Is(err, jetstream.ErrKeyDeleted) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("lambda: read %s: %w", key, err)
+	}
+	rec, err := decode(entry.Value())
+	if err != nil {
+		return err
+	}
+	if rec.State != StatePending || rec.Code.State != CodeStored {
+		return nil
+	}
+	if notReady := ready(ctx, rec); notReady != nil {
+		if rec.StateReason == notReady.Error() {
+			return nil
+		}
+		rec.StateReasonCode, rec.StateReason = ReasonCodeCreating, notReady.Error()
+	} else {
+		rec.State, rec.StateReasonCode, rec.StateReason = StateActive, "", ""
+	}
+	data, err := json.Marshal(rec)
+	if err != nil {
+		return err
+	}
+	if _, err := kv.Update(ctx, key, data, entry.Revision()); err != nil {
+		if errors.Is(err, jetstream.ErrKeyExists) {
+			return nil
+		}
+		return fmt.Errorf("lambda: write readiness of %s: %w", key, err)
+	}
+	if rec.State == StateActive {
+		slog.Info("Lambda: function active", "account_id", rec.AccountID, "region", rec.Region,
+			"function_name", rec.FunctionName, "function_id", rec.FunctionID)
+	}
+	return nil
+}

spinifex/domains/lambda/function/activate_test.go

@@ -0,0 +1,120 @@
+package function_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	actAccount = "000000000001"
+	actRegion  = "ap-southeast-2"
+	actDigest  = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+)
+
+func newOwner(t *testing.T) *function.MetaService {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	return function.NewMetaService(testutil.NewJetStream(t, nc))
+}
+
+func create(t *testing.T, svc *function.MetaService, name string) uint64 {
+	t.Helper()
+	resp, err := svc.Create(t.Context(), &function.CreateRequest{Region: actRegion, Record: function.Record{
+		FunctionID: "id-" + name, FunctionName: name, Role: "arn:aws:iam::000000000001:role/r",
+		Runtime: "provided.al2023", Handler: "bootstrap", Timeout: 3, MemorySize: 128,
+		Code: function.Code{RevisionID: "rev-" + name},
+	}}, actAccount)
+	require.NoError(t, err)
+	return resp.Revision
+}
+
+func get(t *testing.T, svc *function.MetaService, name string) *function.GetResponse {
+	t.Helper()
+	resp, err := svc.Get(t.Context(), &function.GetRequest{Region: actRegion, FunctionName: name}, actAccount)
+	require.NoError(t, err)
+	return resp
+}
+
+func storeCode(t *testing.T, svc *function.MetaService, name string, rev uint64) {
+	t.Helper()
+	_, err := svc.CodeStored(t.Context(), &function.CodeStoredRequest{Region: actRegion, FunctionName: name, Revision: rev, SHA256: actDigest, Size: 9}, actAccount)
+	require.NoError(t, err)
+}
+
+func TestActivation_WithoutReadinessNothingLeavesPending(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	svc.RunActivation(ctx, time.Millisecond)
+
+	got := get(t, svc, "hello").Record
+	assert.Equal(t, function.StatePending, got.State)
+	assert.Equal(t, function.ReasonCreating, got.StateReason)
+}
+
+func TestActivation_UnmetConditionStaysPendingWithItsReason(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+	svc.SetReadiness(func(context.Context, function.Record) error { return errors.New("The node is not serving.") })
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	first := get(t, svc, "hello")
+	assert.Equal(t, function.StatePending, first.Record.State)
+	assert.Equal(t, function.ReasonCodeCreating, first.Record.StateReasonCode)
+	assert.Equal(t, "The node is not serving.", first.Record.StateReason)
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	assert.Equal(t, first.Revision, get(t, svc, "hello").Revision, "an unchanged reason is not rewritten")
+}
+
+func TestActivation_ReadyStoredFunctionBecomesActive(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+	storingRev := create(t, svc, "storing")
+	var seen []string
+	svc.SetReadiness(func(_ context.Context, rec function.Record) error {
+		seen = append(seen, rec.FunctionName)
+		return nil
+	})
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+
+	got := get(t, svc, "hello").Record
+	assert.Equal(t, function.StateActive, got.State)
+	assert.Empty(t, got.StateReason)
+	assert.Empty(t, got.StateReasonCode)
+	assert.Equal(t, function.CodeStored, got.Code.State)
+	storing := get(t, svc, "storing")
+	assert.Equal(t, function.StatePending, storing.Record.State, "a function still storing its code is never activated")
+	assert.Equal(t, storingRev, storing.Revision)
+	assert.Equal(t, []string{"hello"}, seen)
+}
+
+func TestActivation_CodeStoredWakesTheLoop(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "first", create(t, svc, "first"))
+	// Readiness is enabled after first is stored, so no wake is buffered and
+	// only the start pass can activate first.
+	svc.SetReadiness(func(context.Context, function.Record) error { return nil })
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan struct{})
+	go func() { defer close(done); svc.RunActivation(ctx, time.Hour) }()
+	t.Cleanup(func() { cancel(); <-done })
+	require.Eventually(t, func() bool { return get(t, svc, "first").Record.State == function.StateActive },
+		10*time.Second, 20*time.Millisecond, "the start pass activates what was already stored")
+
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+
+	require.Eventually(t, func() bool { return get(t, svc, "hello").Record.State == function.StateActive },
+		10*time.Second, 20*time.Millisecond, "after the start pass, only the code-stored wake can activate within the hour")
+}

spinifex/domains/lambda/function/meta.go

@@ -44,6 +44,8 @@ type MetaService struct {
 
 	mu      sync.Mutex
 	buckets map[string]jetstream.KeyValue
+	ready   Readiness
+	nudge   chan struct{}
 }
 
 // NewMetaService builds the service over a JetStream handle.
@@ -306,6 +308,7 @@ func (s *MetaService) CodeStored(ctx context.Context, req *CodeStoredRequest, ac
 		return nil, fmt.Errorf("lambda: update function record: %w", err)
 	}
 	attach(ctx, kv, up, upRev, rev)
+	s.wake()
 	return &CodeStoredResponse{Record: rec, Revision: rev}, nil
 }
 

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

@@ -151,6 +151,9 @@ type ControlServer struct {
 	// IOTimeout bounds reading the request and writing the reply.
 	IOTimeout time.Duration
 	Handle    func(ctx context.Context, req []byte, cred Credential) Reply
+	// AcceptStopped, if set, runs once Serve stops accepting, before it waits
+	// for the connections it started.
+	AcceptStopped func()
 }
 
 // Serve accepts until ctx ends or the listener fails, then waits for every
@@ -164,6 +167,9 @@ func (s *ControlServer) Serve(ctx context.Context, ln *net.UnixListener) error {
 	defer func() { _ = ln.Close() }()
 	var wg sync.WaitGroup
 	defer wg.Wait()
+	if s.AcceptStopped != nil {
+		defer s.AcceptStopped()
+	}
 	slots := make(chan struct{}, MaxAdmitted)
 	for {
 		select {

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

@@ -462,3 +462,45 @@ func TestControl_MixedProtocolNamesNeverConnect(t *testing.T) {
 		})
 	}
 }
+
+func TestControl_AcceptStoppedRunsBeforeTheDrain(t *testing.T) {
+	name := controlName(t)
+	ln, err := ListenControl(name)
+	require.NoError(t, err)
+	inHandle, release := make(chan struct{}), make(chan struct{})
+	stopped := make(chan struct{})
+	srv := &ControlServer{GatewayUID: os.Getuid(), RequestLimit: testRequestLimit, ReplyLimit: 4096, IOTimeout: 5 * time.Second,
+		Handle: func(context.Context, []byte, Credential) Reply {
+			close(inHandle)
+			<-release
+			return Reply{Outcome: OutcomeDone}
+		},
+		AcceptStopped: func() { close(stopped) }}
+	done := make(chan error, 1)
+	go func() { done <- srv.Serve(context.Background(), ln) }()
+	go func() { _, _ = client(name, os.Getuid()).Exchange(context.Background(), []byte("{}"), testCred, soon()) }()
+	select {
+	case <-inHandle:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the request never reached Handle")
+	}
+
+	require.NoError(t, ln.Close(), "the listener fails under an outstanding invocation")
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("AcceptStopped did not run while the invocation was outstanding")
+	}
+	select {
+	case err := <-done:
+		t.Fatalf("Serve returned before its connection finished: %v", err)
+	default:
+	}
+	close(release)
+	select {
+	case err := <-done:
+		require.Error(t, err, "a listener failure is reported")
+	case <-time.After(5 * time.Second):
+		t.Fatal("Serve did not return after its connection finished")
+	}
+}

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

@@ -91,9 +91,13 @@ type Daemon struct {
 	Initializer       string
 	InitializerSHA256 string
 
-	busy atomic.Bool
+	busy    atomic.Bool
+	serving atomic.Bool
 }
 
+// Serving reports whether the control socket is accepting invocations.
+func (d *Daemon) Serving() bool { return d.serving.Load() }
+
 // unresolved names an attempt whose marker is still on disk. A marker goes
 // only once the attempt's VM stop is proven and everything it held is
 // released, so it outlives a daemon restart; nothing resolves it on its own.
@@ -108,6 +112,12 @@ func (d *Daemon) unresolved() (string, error) {
 	return strings.TrimSuffix(filepath.Base(markers[0]), markerSuffix), nil
 }
 
+// validHandler is the handler a guest run accepts: 1..128 bytes, no
+// whitespace or NUL.
+func validHandler(h string) bool {
+	return h != "" && len(h) <= 128 && strings.IndexFunc(h, unicode.IsSpace) < 0 && !strings.ContainsRune(h, 0)
+}
+
 // controlIOTimeout bounds reading a request from, and writing a reply to, the
 // gateway on the control socket.
 const controlIOTimeout = 30 * time.Second
@@ -119,14 +129,24 @@ func (d *Daemon) ServeControl(ctx context.Context, name string, gatewayUID int)
 	if err != nil {
 		return err
 	}
-	srv := &ControlServer{GatewayUID: gatewayUID, RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes,
-		IOTimeout: controlIOTimeout, Handle: d.Handle}
+	d.serve(ctx, ln, d.controlServer(gatewayUID))
+	return nil
+}
+
+// controlServer answers the gateway with d.Handle, and stops d reporting
+// serving as soon as it stops accepting, before its connections drain.
+func (d *Daemon) controlServer(gatewayUID int) *ControlServer {
+	return &ControlServer{GatewayUID: gatewayUID, RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes,
+		IOTimeout: controlIOTimeout, Handle: d.Handle, AcceptStopped: func() { d.serving.Store(false) }}
+}
+
+func (d *Daemon) serve(ctx context.Context, ln *net.UnixListener, srv *ControlServer) {
+	d.serving.Store(true)
 	go func() {
 		if err := srv.Serve(ctx, ln); err != nil {
 			slog.Error("Lambda test link: control socket stopped; not serving new invocations", "err", err)
 		}
 	}()
-	return nil
 }
 
 // Handle runs one request under its grant, giving the guest cred in its first
@@ -293,8 +313,7 @@ func CheckRequest(req Request) error {
 	if req.RequestID == "" || req.RevisionID == "" {
 		return errors.New("request and revision IDs are required")
 	}
-	if req.Handler == "" || len(req.Handler) > 128 || strings.IndexFunc(req.Handler, unicode.IsSpace) >= 0 ||
-		strings.ContainsRune(req.Handler, 0) {
+	if !validHandler(req.Handler) {
 		return errors.New("handler must be 1..128 characters with no whitespace")
 	}
 	if req.TimeoutMs <= 0 || req.TimeoutMs > maxTimeoutMs {

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

@@ -0,0 +1,38 @@
+//go:build lambdatestlink
+
+package testlink
+
+import (
+	"context"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+)
+
+// Readiness admits a function the test link can run, and only while d serves
+// the pinned image on the cluster's only node, so no gateway or daemon is
+// handed an Active function it cannot run.
+func (d *Daemon) Readiness(imagePin string, nodes int) function.Readiness {
+	return func(_ context.Context, rec function.Record) error {
+		if rec.EnvCipher != "" {
+			return notReady("The function cannot run here: test-link Lambda does not support environment variables.")
+		}
+		if !validHandler(rec.Handler) {
+			return notReady("The function cannot run here: test-link Lambda runs only a handler of 1 to 128 characters with no whitespace.")
+		}
+		if rec.Code.Size <= 0 || rec.Code.Size > MaxCodeBytes {
+			return notReady("The function cannot run here: its code is larger than test-link Lambda accepts.")
+		}
+		if nodes != 1 {
+			return notReady("The function cannot run here: test-link Lambda activates only on a single-node cluster.")
+		}
+		if !d.Serving() || d.Image == nil || d.Image.Digest != imagePin {
+			return notReady(function.ReasonCreating)
+		}
+		return nil
+	}
+}
+
+// notReady is an unmet activation condition, worded as the StateReason.
+type notReady string
+
+func (n notReady) Error() string { return string(n) }

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

@@ -0,0 +1,173 @@
+//go:build lambdatestlink
+
+//test:in-package — reuses the package fixtures writeImage, testImageDigest and controlName
+package testlink
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	readyAccount = "000000000001"
+	readyRegion  = "ap-southeast-2"
+)
+
+// storedFunction creates a function through the real owner and records its
+// code as stored, leaving it Pending the way CreateFunction does.
+func storedFunction(t *testing.T, svc *function.MetaService, name string) {
+	t.Helper()
+	storedFunctionWith(t, svc, name, "", 9)
+}
+
+func storedFunctionWith(t *testing.T, svc *function.MetaService, name, envCipher string, size int64) {
+	t.Helper()
+	storedFunctionHandler(t, svc, name, "bootstrap", envCipher, size)
+}
+
+func storedFunctionHandler(t *testing.T, svc *function.MetaService, name, handler, envCipher string, size int64) {
+	t.Helper()
+	created, err := svc.Create(t.Context(), &function.CreateRequest{Region: readyRegion, Record: function.Record{
+		FunctionID: "id-" + name, FunctionName: name, Role: "arn:aws:iam::000000000001:role/r",
+		Runtime: "provided.al2023", Handler: handler, Timeout: 3, MemorySize: 128, EnvCipher: envCipher,
+		Code: function.Code{RevisionID: "rev-" + name},
+	}}, readyAccount)
+	require.NoError(t, err)
+	_, err = svc.CodeStored(t.Context(), &function.CodeStoredRequest{Region: readyRegion, FunctionName: name,
+		Revision: created.Revision, SHA256: testImageDigest, Size: size}, readyAccount)
+	require.NoError(t, err)
+}
+
+func functionState(t *testing.T, svc *function.MetaService, name string) function.Record {
+	t.Helper()
+	got, err := svc.Get(t.Context(), &function.GetRequest{Region: readyRegion, FunctionName: name}, readyAccount)
+	require.NoError(t, err)
+	require.True(t, got.Found)
+	return got.Record
+}
+
+func readyOwner(t *testing.T) *function.MetaService {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	return function.NewMetaService(testutil.NewJetStream(t, nc))
+}
+
+func TestReadiness_ActivatesOnlyOnceTheDaemonServesThePinnedImage(t *testing.T) {
+	svc := readyOwner(t)
+	im, err := PrepareImage(writeImage(t), testImageDigest)
+	require.NoError(t, err)
+	t.Cleanup(im.Close)
+	d := &Daemon{Node: "node1", Image: im}
+	svc.SetReadiness(d.Readiness(testImageDigest, 1))
+	storedFunction(t, svc, "hello")
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	waiting := functionState(t, svc, "hello")
+	assert.Equal(t, function.StatePending, waiting.State, "a prepared image that is not served activates nothing")
+	assert.Equal(t, function.ReasonCreating, waiting.StateReason)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	require.NoError(t, d.ServeControl(ctx, controlName(t), os.Getuid()))
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	active := functionState(t, svc, "hello")
+	assert.Equal(t, function.StateActive, active.State)
+	assert.Empty(t, active.StateReason)
+
+	cancel()
+	require.Eventually(t, func() bool { return !d.Serving() }, 5*time.Second, 10*time.Millisecond,
+		"a stopped control socket is reported as not serving")
+	storedFunction(t, svc, "later")
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	assert.Equal(t, function.StatePending, functionState(t, svc, "later").State)
+}
+
+func TestReadiness_RefusesAnotherPinOrAnotherNode(t *testing.T) {
+	im, err := PrepareImage(writeImage(t), testImageDigest)
+	require.NoError(t, err)
+	t.Cleanup(im.Close)
+	d := &Daemon{Node: "node1", Image: im}
+	require.NoError(t, d.ServeControl(t.Context(), controlName(t), os.Getuid()))
+
+	for _, tc := range []struct {
+		name   string
+		pin    string
+		nodes  int
+		reason string
+	}{
+		{"other pin", "0000000000000000000000000000000000000000000000000000000000000000", 1, function.ReasonCreating},
+		{"two nodes", testImageDigest, 2, "The function cannot run here: test-link Lambda activates only on a single-node cluster."},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			svc := readyOwner(t)
+			svc.SetReadiness(d.Readiness(tc.pin, tc.nodes))
+			storedFunction(t, svc, "hello")
+			require.NoError(t, svc.ActivatePending(t.Context()))
+			got := functionState(t, svc, "hello")
+			assert.Equal(t, function.StatePending, got.State)
+			assert.Equal(t, tc.reason, got.StateReason)
+		})
+	}
+}
+
+func TestReadiness_RefusesFunctionsTheTestLinkCannotRun(t *testing.T) {
+	im, err := PrepareImage(writeImage(t), testImageDigest)
+	require.NoError(t, err)
+	t.Cleanup(im.Close)
+	d := &Daemon{Node: "node1", Image: im}
+	require.NoError(t, d.ServeControl(t.Context(), controlName(t), os.Getuid()))
+	svc := readyOwner(t)
+	svc.SetReadiness(d.Readiness(testImageDigest, 1))
+	storedFunctionWith(t, svc, "env", "sealed", 9)
+	storedFunctionWith(t, svc, "huge", "", MaxCodeBytes+1)
+	storedFunctionWith(t, svc, "edge", "", MaxCodeBytes)
+	storedFunctionHandler(t, svc, "spaced", "boot\u00a0strap", "", 9)
+	storedFunctionHandler(t, svc, "longhandler", strings.Repeat("h", 129), "", 9)
+	storedFunctionHandler(t, svc, "edgehandler", strings.Repeat("h", 128), "", 9)
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	assert.Equal(t, "The function cannot run here: test-link Lambda does not support environment variables.", functionState(t, svc, "env").StateReason)
+	assert.Equal(t, function.StatePending, functionState(t, svc, "env").State)
+	assert.Equal(t, "The function cannot run here: its code is larger than test-link Lambda accepts.", functionState(t, svc, "huge").StateReason)
+	assert.Equal(t, function.StatePending, functionState(t, svc, "huge").State)
+	assert.Equal(t, function.StateActive, functionState(t, svc, "edge").State, "the largest accepted code copy activates")
+	for _, name := range []string{"spaced", "longhandler"} {
+		assert.Equal(t, function.StatePending, functionState(t, svc, name).State, name)
+		assert.Equal(t, "The function cannot run here: test-link Lambda runs only a handler of 1 to 128 characters with no whitespace.", functionState(t, svc, name).StateReason, name)
+	}
+	assert.Equal(t, function.StateActive, functionState(t, svc, "edgehandler").State, "the longest accepted handler activates")
+}
+
+func TestReadiness_ServingEndsWhenAcceptingStops(t *testing.T) {
+	d := &Daemon{Node: "node1"}
+	name := controlName(t)
+	ln, err := ListenControl(name)
+	require.NoError(t, err)
+	inHandle, release := make(chan struct{}), make(chan struct{})
+	srv := d.controlServer(os.Getuid())
+	srv.Handle = func(context.Context, []byte, Credential) Reply {
+		close(inHandle)
+		<-release
+		return Reply{Outcome: OutcomeDone}
+	}
+	d.serve(t.Context(), ln, srv)
+	require.True(t, d.Serving())
+	go func() { _, _ = (&ControlClient{Name: name, DaemonUID: os.Getuid(), RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes, IOTimeout: 5 * time.Second}).Exchange(context.Background(), []byte("{}"), testCred, time.Now().Add(5*time.Second)) }()
+	select {
+	case <-inHandle:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the request never reached Handle")
+	}
+	defer close(release)
+
+	require.NoError(t, ln.Close(), "the listener fails under an outstanding invocation")
+	assert.Eventually(t, func() bool { return !d.Serving() }, 2*time.Second, 10*time.Millisecond,
+		"serving ends while the invocation is still held, not after the drain")
+}

spinifex/runtime/daemon/daemon_lambda_testlink.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/domains/lambda/execution"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/host"
@@ -88,8 +89,17 @@ func (d *Daemon) startLambdaTestLink() {
 	}
 	slog.Warn("Lambda test link serving: test-only, not for production; the guest does not verify owner grant authority",
 		"node", d.node, "image_digest", tl.Image.Digest)
+	if d.lambdaMetaService != nil && d.clusterConfig != nil {
+		d.lambdaMetaService.SetReadiness(tl.Readiness(lambdaImageDigest, len(d.clusterConfig.Nodes)))
+		go d.lambdaMetaService.RunActivation(d.ctx, lambdaActivationInterval)
+	}
 }
 
+// lambdaActivationInterval bounds how long a function whose activation was
+// missed, by a crash or an earlier refusal, waits for the next pass.
+const lambdaActivationInterval = 30 * time.Second
+
+
 // lambdaTestLinkReady runs the serving checks and, only once they pass,
 // prepares the pinned image into tl; a refusal leaves tl.Image nil.
 func lambdaTestLinkReady(tl *testlink.Daemon, imageDir, imagePin string, trustErr error) error {