test(lambda): add code update methods to the seam fake

Hold one commit after the owner has read the record and checked its
upload, let a competing commit win, then release it: the held commit
must lose its conditional write and leave its upload owed.

cmd/lambda-agent/start_testlink.go

@@ -12,12 +12,14 @@ import (
 	"net/http"
 	"os"
 	"os/exec"
+	"sort"
 	"strings"
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/agents/lambda/agent"
 	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
 	"github.com/mulgadc/spinifex/spinifex/agents/lambda/runtimeapi"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/lambdaenv"
 )
 
 const (
@@ -91,6 +93,9 @@ func (t testlink) run(parent context.Context) (int, error) {
 	}
 	cred := *first.Credential
 	first.Credential = nil
+	if err := lambdaenv.Validate(cred.Environment); err != nil {
+		return exitRefused, errors.New("first frame carries an invalid function environment")
+	}
 	if sdkErr != nil {
 		slog.Warn("lambda-agent: SDK gateway settings unavailable", "err", sdkErr)
 	}
@@ -142,19 +147,31 @@ func (t testlink) run(parent context.Context) (int, error) {
 
 // bootstrapEnv supplies the runtime, execution-role and SDK settings.
 func bootstrapEnv(runtimeAPI, handler string, cred hostlink.Credential, region string, sdk *sdkConfig, caPath string) []string {
-	env := []string{
-		"AWS_LAMBDA_RUNTIME_API=" + runtimeAPI,
-		"_HANDLER=" + handler,
-		"LAMBDA_TASK_ROOT=" + taskRoot,
-		"PATH=/bin:/sbin:/usr/bin",
-		"AWS_ACCESS_KEY_ID=" + cred.AccessKeyID,
-		"AWS_SECRET_ACCESS_KEY=" + cred.SecretAccessKey,
-		"AWS_SESSION_TOKEN=" + cred.SessionToken,
-		"AWS_REGION=" + region,
-		"AWS_DEFAULT_REGION=" + region,
+	vars := map[string]string{
+		"AWS_LAMBDA_RUNTIME_API": runtimeAPI,
+		"_HANDLER":               handler,
+		"LAMBDA_TASK_ROOT":       taskRoot,
+		"PATH":                   "/bin:/sbin:/usr/bin",
+		"AWS_ACCESS_KEY_ID":      cred.AccessKeyID,
+		"AWS_SECRET_ACCESS_KEY":  cred.SecretAccessKey,
+		"AWS_SESSION_TOKEN":      cred.SessionToken,
+		"AWS_REGION":             region,
+		"AWS_DEFAULT_REGION":     region,
 	}
 	if sdk != nil {
-		env = append(env, "AWS_ENDPOINT_URL="+sdk.endpoint, "AWS_CA_BUNDLE="+caPath)
+		vars["AWS_ENDPOINT_URL"], vars["AWS_CA_BUNDLE"] = sdk.endpoint, caPath
+	}
+	for k, v := range cred.Environment {
+		vars[k] = v
+	}
+	keys := make([]string, 0, len(vars))
+	for k := range vars {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	env := make([]string, 0, len(keys))
+	for _, k := range keys {
+		env = append(env, k+"="+vars[k])
 	}
 	return env
 }

cmd/lambda-agent/start_testlink_test.go

@@ -221,6 +221,48 @@ func TestTestlinkProvidesSDKSettingsToBootstrap(t *testing.T) {
 	assert.Equal(t, os.FileMode(0o644), info.Mode().Perm())
 }
 
+func TestTestlinkProvidesFunctionEnvironment(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(t, h.entries()))
+	conn := h.guest(t)
+	defer conn.Close()
+	hs := hostlink.NewTestStream(conn)
+	cred := testCredential
+	cred.Environment = map[string]string{"GREETING": "hello $WORLD", "PATH": "/my/bin"}
+	req := hostlink.Request{Credential: &cred, GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", DeadlineMs: time.Now().Add(5 * time.Second).UnixMilli(), Payload: []byte(`{}`)}
+	require.NoError(t, hs.HostSend(req))
+	_, err := hs.HostResult(ctx)
+	require.NoError(t, err)
+	require.NoError(t, wait(t, done, "agent to stop").err)
+	assert.Equal(t, "hello $WORLD", envValue(b.env, "GREETING"))
+	assert.Equal(t, "/my/bin", envValue(b.env, "PATH"))
+	assert.Equal(t, testCredential.SecretAccessKey, envValue(b.env, "AWS_SECRET_ACCESS_KEY"))
+}
+
+func TestTestlinkRejectsReservedFunctionEnvironment(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(t, h.entries()))
+	conn := h.guest(t)
+	defer conn.Close()
+	cred := testCredential
+	cred.Environment = map[string]string{"AWS_SECRET_ACCESS_KEY": "forged"}
+	req := hostlink.Request{Credential: &cred, GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", DeadlineMs: time.Now().Add(5 * time.Second).UnixMilli(), Payload: []byte(`{}`)}
+	require.NoError(t, hostlink.NewTestStream(conn).HostSend(req))
+	result := wait(t, done, "agent to refuse the environment")
+	assert.Equal(t, exitRefused, result.code)
+	assert.Error(t, result.err)
+	assert.NotContains(t, result.err.Error(), "forged")
+	assert.Zero(t, b.n)
+}
+
 var testCredential = hostlink.Credential{AccessKeyID: "ASIAGUESTKEY", SecretAccessKey: "guest-secret", SessionToken: "guest-token"}
 
 func envValue(env []string, name string) string {

contracts/lambda/v1/README.md

@@ -21,14 +21,14 @@ Tenant code may be able to reach the listener, but it cannot open the agent's co
 **Authentication:** mutual TLS 1.3, bound to the environment ID and realization generation, using material the user runtime cannot read (ADR-0009:S17).
 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.
+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/3` on both host and guest, whose first frame carries the execution-role credential and function environment, 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/4`, 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 credential and environment travel as their own frame of at most 40 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.
+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 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).
 

spinifex/agents/lambda/hostlink/link.go

@@ -28,11 +28,12 @@ type Request struct {
 	Credential *Credential `json:"credential,omitempty"`
 }
 
-// Credential is one execution-role session. It is never logged or stored.
+// Credential holds one execution-role session and its function environment.
 type Credential struct {
-	AccessKeyID     string `json:"access_key_id"`
-	SecretAccessKey string `json:"secret_access_key"`
-	SessionToken    string `json:"session_token"`
+	AccessKeyID     string            `json:"access_key_id"`
+	SecretAccessKey string            `json:"secret_access_key"`
+	SessionToken    string            `json:"session_token"`
+	Environment     map[string]string `json:"environment,omitempty"`
 }
 
 // LogValue keeps every value out of logs.

spinifex/agents/lambda/hostlink/protocol_testlink.go

@@ -4,5 +4,5 @@ package hostlink
 
 // Protocol is the only ALPN protocol a test-link host and guest offer. It
 // differs from the release protocol because the first frame carries the
-// execution credential, which an older guest would ignore.
-const Protocol = "spinifex-lambda-testlink/2"
+// execution credential and function environment, which an older guest would ignore.
+const Protocol = "spinifex-lambda-testlink/3"

spinifex/domains/lambda/awsapi/handler_test.go

@@ -49,6 +49,12 @@ type localClient struct {
 	failCodeStored bool
 	passOwnerClock bool
 	rollbacks      int
+	// beginErr fails BeginCodeUpdate before the owner sees it.
+	beginErr error
+	// loseUpdatedReply commits CodeUpdated, then reports a transport error.
+	loseUpdatedReply bool
+	// beforeCommit runs once, unlocked, before the next CodeUpdated reaches the owner.
+	beforeCommit func()
 }
 
 var _ function.Client = (*localClient)(nil)
@@ -104,6 +110,37 @@ func (c *localClient) RollbackCreate(ctx context.Context, account, region, name
 	return err
 }
 
+func (c *localClient) BeginCodeUpdate(ctx context.Context, account, region string, rec function.Record, rev uint64, codeRevisionID string) error {
+	c.mu.Lock()
+	fail := c.beginErr
+	c.mu.Unlock()
+	if fail != nil {
+		return fail
+	}
+	_, err := c.svc.BeginCodeUpdate(ctx, &function.CodeUpdateBeginRequest{Region: region, FunctionName: rec.FunctionName,
+		FunctionID: rec.FunctionID, Revision: rev, RevisionID: codeRevisionID}, account)
+	return err
+}
+
+func (c *localClient) CodeUpdated(ctx context.Context, account, region string, rec function.Record, rev uint64, codeRevisionID, sha string, size int64) (function.Record, uint64, error) {
+	c.mu.Lock()
+	hook, lose := c.beforeCommit, c.loseUpdatedReply
+	c.beforeCommit = nil
+	c.mu.Unlock()
+	if hook != nil {
+		hook()
+	}
+	r, err := c.svc.CodeUpdated(ctx, &function.CodeUpdatedRequest{Region: region, FunctionName: rec.FunctionName, FunctionID: rec.FunctionID,
+		Revision: rev, RevisionID: codeRevisionID, SHA256: sha, Size: size}, account)
+	if err != nil {
+		return function.Record{}, 0, err
+	}
+	if lose {
+		return function.Record{}, 0, errors.New("nats: timeout")
+	}
+	return r.Record, r.Revision, nil
+}
+
 // memStore is an in-memory code store; objects is keyed by Ref.
 type memStore struct {
 	mu      sync.Mutex
@@ -181,6 +218,16 @@ func newHarness(t *testing.T, statements ...iamdomain.Statement) (*gwtest.Harnes
 
 // uploads returns every upload record in the account bucket, by function ID.
 func (e *env) uploads(t *testing.T) map[string]function.Upload {
+	t.Helper()
+	out := map[string]function.Upload{}
+	for _, up := range e.uploadsByRevision(t) {
+		out[up.FunctionID] = up
+	}
+	return out
+}
+
+// uploadsByRevision returns every upload record in the account bucket, by code revision ID.
+func (e *env) uploadsByRevision(t *testing.T) map[string]function.Upload {
 	t.Helper()
 	kv := e.kv(t)
 	keys, err := kv.ListKeys(t.Context())
@@ -195,7 +242,7 @@ func (e *env) uploads(t *testing.T) map[string]function.Upload {
 		var up function.Upload
 		require.NoError(t, json.Unmarshal(entry.Value(), &up))
 		require.Equal(t, function.KVUploadKey(up.Region, up.FunctionID, up.RevisionID), k)
-		out[up.FunctionID] = up
+		out[up.RevisionID] = up
 	}
 	return out
 }
@@ -203,13 +250,13 @@ func (e *env) uploads(t *testing.T) map[string]function.Upload {
 // requireTracked asserts every code copy in the store has an upload record.
 func (e *env) requireTracked(t *testing.T) {
 	t.Helper()
-	ups := e.uploads(t)
+	ups := e.uploadsByRevision(t)
 	e.store.mu.Lock()
 	defer e.store.mu.Unlock()
 	for ref := range e.store.objects {
-		up, ok := ups[ref.FunctionID]
-		require.True(t, ok, "code copy %s has no upload record", ref.FunctionID)
-		require.Equal(t, ref.RevisionID, up.RevisionID)
+		up, ok := ups[ref.RevisionID]
+		require.True(t, ok, "code copy %s/%s has no upload record", ref.FunctionID, ref.RevisionID)
+		require.Equal(t, ref.FunctionID, up.FunctionID)
 	}
 }
 
@@ -553,7 +600,6 @@ func TestCreateFunction_ParameterValidation(t *testing.T) {
 		"missing role":    func(b map[string]any) { b["Role"] = "arn:aws:iam::" + gwtest.AccountID + ":role/ghost" },
 		"not a role arn":  func(b map[string]any) { b["Role"] = "arn:aws:iam::" + gwtest.AccountID + ":user/alice" },
 		"malformed body":  func(b map[string]any) { b["Timeout"] = "soon" },
-		"empty env name":  func(b map[string]any) { b["Environment"] = map[string]any{"Variables": map[string]string{"": "x"}} },
 		"description len": func(b map[string]any) { b["Description"] = strings.Repeat("d", 257) },
 	}
 	for name, mutate := range cases {
@@ -566,6 +612,31 @@ func TestCreateFunction_ParameterValidation(t *testing.T) {
 	}
 }
 
+func TestCreateFunction_EnvironmentValidation(t *testing.T) {
+	for name, tc := range map[string]struct {
+		vars map[string]string
+		code string
+	}{
+		"empty name":           {map[string]string{"": "secret"}, awserrors.ErrorValidationException},
+		"one character":        {map[string]string{"A": "secret"}, awserrors.ErrorValidationException},
+		"invalid character":    {map[string]string{"BAD-NAME": "secret"}, awserrors.ErrorValidationException},
+		"invalid and reserved": {map[string]string{"A": "secret", "AWS_REGION": "secret"}, awserrors.ErrorValidationException},
+		"reserved":             {map[string]string{"AWS_SECRET_ACCESS_KEY": "secret"}, awserrors.ErrorInvalidParameterValueException},
+		"nul value":            {map[string]string{"VALUE": "secret\x00"}, awserrors.ErrorInvalidParameterValueException},
+		"too large":            {map[string]string{"VALUE": strings.Repeat("x", 4092)}, awserrors.ErrorInvalidParameterValueException},
+	} {
+		t.Run(name, func(t *testing.T) {
+			e := newEnv(t)
+			_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
+				b["Environment"] = map[string]any{"Variables": tc.vars}
+			}))
+			testutil.RequireAWSCode(t, err, tc.code)
+			assert.NotContains(t, err.Error(), "secret")
+			assert.Zero(t, e.store.count())
+		})
+	}
+}
+
 func TestCreateFunction_RoleErrorsDoNotLeakAcrossAccounts(t *testing.T) {
 	e := newEnv(t)
 	_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) { b["Role"] = foreignRoleARN }))
@@ -765,13 +836,14 @@ func TestAuthorization(t *testing.T) {
 
 func TestRegistration_ResolvesActionsAndInventory(t *testing.T) {
 	reg := awsapi.NewRegistration(awsapi.Deps{})
-	assert.Equal(t, []string{"CreateEventSourceMapping", "CreateFunction", "DeleteEventSourceMapping", "GetEventSourceMapping", "GetFunction", "GetFunctionCodeSigningConfig", "GetFunctionConfiguration", "Invoke", "ListEventSourceMappings", "ListVersionsByFunction"}, reg.Inventory.Registered)
+	assert.Equal(t, []string{"CreateEventSourceMapping", "CreateFunction", "DeleteEventSourceMapping", "GetEventSourceMapping", "GetFunction", "GetFunctionCodeSigningConfig", "GetFunctionConfiguration", "Invoke", "ListEventSourceMappings", "ListVersionsByFunction", "UpdateFunctionCode"}, reg.Inventory.Registered)
 	assert.Equal(t, "ListVersionsByFunction", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2015-03-31/functions/x/versions", nil)))
 	assert.Equal(t, "DeleteEventSourceMapping", reg.ResolveAction(httptest.NewRequest(http.MethodDelete, "/2015-03-31/event-source-mappings/x", nil)))
 	assert.Equal(t, "GetFunctionCodeSigningConfig", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2020-06-30/functions/x/code-signing-config", nil)))
 	assert.Equal(t, "CreateFunction", reg.ResolveAction(httptest.NewRequest(http.MethodPost, "/2015-03-31/functions", nil)))
 	assert.Equal(t, "GetFunction", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2015-03-31/functions/x", nil)))
 	assert.Equal(t, "GetFunctionConfiguration", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2015-03-31/functions/x/configuration", nil)))
+	assert.Equal(t, "UpdateFunctionCode", reg.ResolveAction(httptest.NewRequest(http.MethodPut, "/2015-03-31/functions/x/code", nil)))
 	assert.Empty(t, reg.ResolveAction(httptest.NewRequest(http.MethodDelete, "/2015-03-31/functions/x", nil)))
 }
 

spinifex/domains/lambda/awsapi/registration.go

@@ -138,7 +138,7 @@ func (d Deps) serve(w http.ResponseWriter, inv dispatch.Invocation) error {
 		if decodeErr == nil {
 			resource = functionARN(inv.Region, inv.AccountID, aws.StringValue(c.Create.FunctionName))
 		}
-	case "GetFunction", "GetFunctionConfiguration", "GetFunctionCodeSigningConfig", "ListVersionsByFunction":
+	case "GetFunction", "GetFunctionConfiguration", "GetFunctionCodeSigningConfig", "ListVersionsByFunction", "UpdateFunctionCode":
 		if name, _, err := resolveFunctionName(inv.Region, inv.AccountID, params[0]); err == nil {
 			resource = functionARN(inv.Region, inv.AccountID, name)
 		}
@@ -167,8 +167,8 @@ func (d Deps) serve(w http.ResponseWriter, inv dispatch.Invocation) error {
 			return err
 		}
 	}
-	// Only CreateFunction and GetFunction touch the code store.
-	needsCodes := action == "CreateFunction" || action == "GetFunction"
+	// Only CreateFunction, GetFunction and UpdateFunctionCode touch the code store.
+	needsCodes := action == "CreateFunction" || action == "GetFunction" || action == "UpdateFunctionCode"
 	if d.Meta == nil || (d.Codes == nil && needsCodes) {
 		return errors.New(awserrors.ErrorServiceException)
 	}

spinifex/domains/lambda/awsapi/routes.go

@@ -50,6 +50,10 @@ var routes = []route{
 		Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
 			return d.getFunction(ctx, c)
 		}},
+	{Method: "PUT", Pattern: "/2015-03-31/functions/{FunctionName}/code", Action: "UpdateFunctionCode",
+		Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
+			return d.updateFunctionCode(ctx, c)
+		}},
 	{Method: "GET", Pattern: "/2015-03-31/functions/{FunctionName}/configuration", Action: "GetFunctionConfiguration",
 		Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
 			return d.getFunctionConfiguration(ctx, c)

spinifex/domains/lambda/awsapi/updatecode.go

@@ -0,0 +1,113 @@
+package awsapi
+
+import (
+	"context"
+	"log/slog"
+	"net/http"
+	"strconv"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+)
+
+// ownerError passes an AWS-coded owner refusal through and turns anything
+// else, such as an owner with no responder for the subject, into a retryable
+// ServiceException, logging the stage that failed.
+func ownerError(ctx context.Context, stage string, rec function.Record, codeRevisionID string, err error) error {
+	if _, coded := awserrors.ResolveErrorCode(err); coded {
+		return err
+	}
+	slog.ErrorContext(ctx, "Lambda: code update owner request failed", "stage", stage,
+		"function_id", rec.FunctionID, "revision_id", codeRevisionID, "err", err)
+	return serviceError("the function code could not be updated; retry the request")
+}
+
+// updateFunctionCode implements UpdateFunctionCode for an inline ZIP. The new
+// copy is written under a fresh code revision while the record still names the
+// old one, and the record moves to it only by compare-and-swap.
+func (d Deps) updateFunctionCode(ctx context.Context, c *call) (reply, error) {
+	in, err := decodeUpdateCode(c.Body)
+	if err != nil {
+		return reply{}, err
+	}
+	name, q, err := resolveFunctionName(c.Region, c.AccountID, c.Params[0])
+	if err != nil {
+		return reply{}, err
+	}
+	if q != "" {
+		return reply{}, invalidFunctionName()
+	}
+	if err := validateUpdateCode(in); err != nil {
+		return reply{}, err
+	}
+	rec, revision, err := d.lookupTarget(ctx, c, name, "")
+	if err != nil {
+		return reply{}, err
+	}
+	if rec.State != function.StateActive || rec.Code.State != function.CodeStored {
+		return reply{}, awserrors.Errorf(awserrors.ErrorResourceConflictException,
+			"The operation cannot be performed at this time. The function is currently in the following state: %s", rec.State)
+	}
+	if want := aws.StringValue(in.RevisionId); in.RevisionId != nil && want != strconv.FormatUint(revision, 10) {
+		return reply{}, awserrors.Errorf(awserrors.ErrorPreconditionFailedException,
+			"The Revision Id provided does not match the latest Revision Id. Call the GetFunction/GetAlias API to retrieve the latest Revision Id")
+	}
+	// The environment is opened before anything is written, so a commit is
+	// never followed by a response that cannot be rendered.
+	var env map[string]string
+	if rec.EnvCipher != "" {
+		if d.AEAD == nil {
+			return reply{}, serviceError("environment decryption is unavailable")
+		}
+		if env, err = function.OpenEnv(d.AEAD, rec.AccountID, rec.FunctionID, rec.EnvCipher); err != nil {
+			slog.ErrorContext(ctx, "Lambda: open environment failed", "function_id", rec.FunctionID, "err", err)
+			return reply{}, serviceError("could not read the function environment")
+		}
+	}
+	if aws.BoolValue(in.DryRun) {
+		cfg, err := configuration(rec, revision, env)
+		if err != nil {
+			return reply{}, serviceError("could not build the function configuration")
+		}
+		return reply{Status: http.StatusOK, Body: cfg}, nil
+	}
+
+	codeRevisionID, err := randomID()
+	if err != nil {
+		return reply{}, serviceError("could not generate a revision identifier")
+	}
+	if err := d.Meta.BeginCodeUpdate(ctx, c.AccountID, c.Region, rec, revision, codeRevisionID); err != nil {
+		return reply{}, ownerError(ctx, "begin", rec, codeRevisionID, err)
+	}
+	ref := codestore.Ref{AccountID: rec.AccountID, FunctionID: rec.FunctionID, RevisionID: codeRevisionID}
+	store, err := d.Codes(ctx)
+	if err != nil {
+		owedCopy(ctx, ref, "code store unavailable", err)
+		return reply{}, serviceError("the function code could not be stored")
+	}
+	sha, err := store.Put(ctx, ref, in.ZipFile)
+	if err != nil {
+		owedCopy(ctx, ref, "code write failed", err)
+		return reply{}, serviceError("the function code could not be stored")
+	}
+	updated, updatedRev, err := d.Meta.CodeUpdated(ctx, c.AccountID, c.Region, rec, revision, codeRevisionID, sha, int64(len(in.ZipFile)))
+	if err != nil {
+		// The commit may have landed with its reply lost; nothing is undone.
+		owedCopy(ctx, ref, "commit not confirmed", err)
+		return reply{}, ownerError(ctx, "commit", rec, codeRevisionID, err)
+	}
+	cfg, err := configuration(updated, updatedRev, env)
+	if err != nil {
+		return reply{}, serviceError("could not build the function configuration")
+	}
+	return reply{Status: http.StatusOK, Body: cfg}, nil
+}
+
+// owedCopy logs a code update whose new copy may exist without the function
+// naming it; its upload record stays owed for later cleanup.
+func owedCopy(ctx context.Context, ref codestore.Ref, why string, cause error) {
+	slog.WarnContext(ctx, "Lambda: code update did not complete; code object cleanup may be owed", "reason", why,
+		"account_id", ref.AccountID, "function_id", ref.FunctionID, "revision_id", ref.RevisionID, "err", cause)
+}

spinifex/domains/lambda/awsapi/updatecode_test.go

@@ -0,0 +1,353 @@
+package awsapi_test
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/lambda"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/internal/testkit/gwtest"
+	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const updatePath = "/2015-03-31/functions/hello/code"
+
+var updateZip = buildZip("hello, version two")
+
+func codeSHA(code []byte) string {
+	sum := sha256.Sum256(code)
+	return base64.StdEncoding.EncodeToString(sum[:])
+}
+
+// activate moves every stored function to Active through the owner's own pass.
+func (e *env) activate(t *testing.T) {
+	t.Helper()
+	e.meta.svc.SetReadiness(func(context.Context, function.Record) error { return nil })
+	require.NoError(t, e.meta.svc.ActivatePending(t.Context()))
+}
+
+// record reads the stored function record and its revision.
+func (e *env) record(t *testing.T, name string) (function.Record, uint64) {
+	t.Helper()
+	entry, err := e.kv(t).Get(t.Context(), function.KVFunctionKey(gwtest.Region, name))
+	require.NoError(t, err)
+	var rec function.Record
+	require.NoError(t, json.Unmarshal(entry.Value(), &rec))
+	return rec, entry.Revision()
+}
+
+// newUpdateEnv creates an Active "hello" with an environment and non-default configuration.
+func newUpdateEnv(t *testing.T, statements ...iamdomain.Statement) *env {
+	t.Helper()
+	e := newEnv(t, statements...)
+	_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
+		b["Environment"] = map[string]any{"Variables": map[string]string{"MODE": "prod"}}
+		b["Description"] = "demo"
+		b["Timeout"] = 30
+		b["MemorySize"] = 256
+	}))
+	require.NoError(t, err)
+	e.activate(t)
+	return e
+}
+
+func updateBody(mutate func(map[string]any)) map[string]any {
+	b := map[string]any{"ZipFile": updateZip}
+	if mutate != nil {
+		mutate(b)
+	}
+	return b
+}
+
+func decodeConfig(t *testing.T, raw []byte) lambda.FunctionConfiguration {
+	t.Helper()
+	var cfg lambda.FunctionConfiguration
+	require.NoError(t, json.Unmarshal(raw, &cfg))
+	return cfg
+}
+
+// requireUnchanged asserts the record still names the original copy at its original revision.
+func requireUnchanged(t *testing.T, e *env, before function.Record, beforeRev uint64) {
+	t.Helper()
+	after, rev := e.record(t, "hello")
+	assert.Equal(t, beforeRev, rev, "the function record was written")
+	assert.Equal(t, before.Code, after.Code)
+}
+
+func TestUpdateFunctionCode_SwapsTheCodeAndKeepsEverythingElse(t *testing.T) {
+	e := newUpdateEnv(t)
+	before, beforeRev := e.record(t, "hello")
+
+	rec, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) {
+		b["Architectures"] = []string{"x86_64"}
+		b["RevisionId"] = strconv.FormatUint(beforeRev, 10)
+	}))
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, rec.Code)
+	cfg := decodeConfig(t, rec.Body.Bytes())
+	assert.Equal(t, codeSHA(updateZip), *cfg.CodeSha256)
+	assert.Equal(t, int64(len(updateZip)), *cfg.CodeSize)
+	assert.Equal(t, "Active", *cfg.State)
+	assert.Equal(t, "Successful", *cfg.LastUpdateStatus)
+	assert.Equal(t, "prod", *cfg.Environment.Variables["MODE"])
+	assert.Equal(t, "demo", *cfg.Description)
+	assert.Equal(t, int64(30), *cfg.Timeout)
+	assert.Equal(t, int64(256), *cfg.MemorySize)
+	assert.Equal(t, roleARN, *cfg.Role)
+
+	after, afterRev := e.record(t, "hello")
+	assert.Equal(t, strconv.FormatUint(afterRev, 10), *cfg.RevisionId)
+	assert.Greater(t, afterRev, beforeRev)
+	assert.Equal(t, before.FunctionID, after.FunctionID)
+	assert.Equal(t, before.EnvCipher, after.EnvCipher)
+	assert.Equal(t, before.CreatedAt, after.CreatedAt)
+	assert.NotEqual(t, before.Code.RevisionID, after.Code.RevisionID)
+	assert.Equal(t, function.CodeStored, after.Code.State)
+	assert.Equal(t, function.StateActive, after.State)
+
+	// Both copies stay, each under its own revision and tracked.
+	require.Equal(t, 2, e.store.count())
+	assert.Equal(t, testZip, e.store.objects[codestore.Ref{AccountID: gwtest.AccountID, FunctionID: before.FunctionID, RevisionID: before.Code.RevisionID}])
+	assert.Equal(t, updateZip, e.store.objects[codestore.Ref{AccountID: gwtest.AccountID, FunctionID: after.FunctionID, RevisionID: after.Code.RevisionID}])
+	e.requireTracked(t)
+	ups := e.uploadsByRevision(t)
+	assert.Equal(t, function.UploadAttached, ups[before.Code.RevisionID].State)
+	assert.Equal(t, function.UploadAttached, ups[after.Code.RevisionID].State)
+	assert.Equal(t, afterRev, ups[after.Code.RevisionID].AttachedRevision)
+
+	got, err := e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
+	require.NoError(t, err)
+	var out lambda.GetFunctionOutput
+	require.NoError(t, json.Unmarshal(got.Body.Bytes(), &out))
+	assert.Equal(t, codeSHA(updateZip), *out.Configuration.CodeSha256)
+}
+
+func TestUpdateFunctionCode_RefusesWhatThisBuildDoesNotImplement(t *testing.T) {
+	cases := map[string]func(map[string]any){
+		"S3 bucket":           func(b map[string]any) { delete(b, "ZipFile"); b["S3Bucket"] = "code" },
+		"S3 key":              func(b map[string]any) { b["S3Key"] = "fn.zip" },
+		"S3 object version":   func(b map[string]any) { b["S3ObjectVersion"] = "1" },
+		"image":               func(b map[string]any) { delete(b, "ZipFile"); b["ImageUri"] = "repo/fn:latest" },
+		"publish":             func(b map[string]any) { b["Publish"] = true },
+		"arm64":               func(b map[string]any) { b["Architectures"] = []string{"arm64"} },
+		"two architectures":   func(b map[string]any) { b["Architectures"] = []string{"x86_64", "arm64"} },
+		"empty architectures": func(b map[string]any) { b["Architectures"] = []string{} },
+		"unknown arch":        func(b map[string]any) { b["Architectures"] = []string{"sparc"} },
+		"no source":           func(b map[string]any) { delete(b, "ZipFile") },
+		"not a zip":           func(b map[string]any) { b["ZipFile"] = []byte("plain bytes") },
+		"unmodelled member":   func(b map[string]any) { b["SourceKMSKeyArn"] = "arn:aws:kms:x" },
+	}
+	e := newUpdateEnv(t)
+	before, beforeRev := e.record(t, "hello")
+	for name, mutate := range cases {
+		t.Run(name, func(t *testing.T) {
+			_, err := e.do(http.MethodPut, updatePath, updateBody(mutate))
+			testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
+		})
+	}
+	t.Run("publish names versions", func(t *testing.T) {
+		_, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) { b["Publish"] = true }))
+		testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
+		assert.Contains(t, errMessage(err), "Publish")
+		assert.Contains(t, errMessage(err), "versions are not yet supported")
+	})
+	t.Run("qualified name", func(t *testing.T) {
+		_, err := e.do(http.MethodPut, "/2015-03-31/functions/hello:1/code", updateBody(nil))
+		testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
+	})
+	requireUnchanged(t, e, before, beforeRev)
+	assert.Equal(t, 1, e.store.count())
+	assert.Len(t, e.uploadsByRevision(t), 1)
+}
+
+func TestUpdateFunctionCode_MissingOrInactiveFunction(t *testing.T) {
+	e := newEnv(t)
+	_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
+	testutil.RequireAWSCode(t, err, awserrors.ErrorResourceNotFoundException)
+
+	_, err = e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
+	require.NoError(t, err)
+	before, beforeRev := e.record(t, "hello")
+	_, err = e.do(http.MethodPut, updatePath, updateBody(nil))
+	testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
+	assert.Contains(t, errMessage(err), "Pending")
+	requireUnchanged(t, e, before, beforeRev)
+	assert.Equal(t, 1, e.store.count())
+}
+
+func TestUpdateFunctionCode_Authorization(t *testing.T) {
+	arn := "arn:aws:lambda:" + gwtest.Region + ":" + gwtest.AccountID + ":function:hello"
+	t.Run("denied on the function ARN", func(t *testing.T) {
+		e := newUpdateEnv(t, gwtest.Statement("Allow", "*", "*"), gwtest.Statement("Deny", "lambda:UpdateFunctionCode", arn))
+		before, beforeRev := e.record(t, "hello")
+		_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
+		gwtest.AssertDenied(t, err)
+		requireUnchanged(t, e, before, beforeRev)
+		assert.Equal(t, 1, e.store.count())
+	})
+	t.Run("another function's grant does not reach it", func(t *testing.T) {
+		e := newUpdateEnv(t, gwtest.Statement("Allow", "lambda:CreateFunction", "*"), gwtest.Statement("Allow", "iam:PassRole", "*"),
+			gwtest.Statement("Allow", "lambda:UpdateFunctionCode", arn+"-other"))
+		_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
+		gwtest.AssertDenied(t, err)
+	})
+	t.Run("allowed on the function ARN", func(t *testing.T) {
+		e := newUpdateEnv(t, gwtest.Statement("Allow", "lambda:CreateFunction", "*"), gwtest.Statement("Allow", "iam:PassRole", "*"),
+			gwtest.Statement("Allow", "lambda:UpdateFunctionCode", arn))
+		_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
+		require.NoError(t, err)
+	})
+}
+
+func TestUpdateFunctionCode_DryRunWritesNothing(t *testing.T) {
+	e := newUpdateEnv(t)
+	before, beforeRev := e.record(t, "hello")
+	rec, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) { b["DryRun"] = true }))
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, rec.Code)
+	cfg := decodeConfig(t, rec.Body.Bytes())
+	assert.Equal(t, codeSHA(testZip), *cfg.CodeSha256, "a dry run reports the code that still runs")
+	assert.Equal(t, strconv.FormatUint(beforeRev, 10), *cfg.RevisionId)
+	requireUnchanged(t, e, before, beforeRev)
+	assert.Equal(t, 1, e.store.count())
+	assert.Len(t, e.uploadsByRevision(t), 1)
+
+	_, err = e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) { b["DryRun"] = true; b["Publish"] = true }))
+	testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
+}
+
+func TestUpdateFunctionCode_StaleRevisionIDIsAPreconditionFailure(t *testing.T) {
+	e := newUpdateEnv(t)
+	before, beforeRev := e.record(t, "hello")
+	_, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) {
+		b["RevisionId"] = strconv.FormatUint(beforeRev-1, 10)
+	}))
+	testutil.RequireAWSCode(t, err, awserrors.ErrorPreconditionFailedException)
+	requireUnchanged(t, e, before, beforeRev)
+	assert.Equal(t, 1, e.store.count())
+	assert.Len(t, e.uploadsByRevision(t), 1)
+}
+
+func TestUpdateFunctionCode_FailedCopyLeavesTheFunctionOnItsOldCode(t *testing.T) {
+	e := newUpdateEnv(t)
+	before, beforeRev := e.record(t, "hello")
+	e.store.putErr = errors.New("predastore: unavailable")
+	_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
+	testutil.RequireAWSCode(t, err, awserrors.ErrorServiceException)
+	requireUnchanged(t, e, before, beforeRev)
+
+	ups := e.uploadsByRevision(t)
+	require.Len(t, ups, 2, "the attempted upload stays recorded")
+	for rev, up := range ups {
+		if rev == before.Code.RevisionID {
+			assert.Equal(t, function.UploadAttached, up.State)
+			continue
+		}
+		assert.Equal(t, function.UploadOwed, up.State, "an upload whose copy may exist stays owed")
+	}
+}
+
+func TestUpdateFunctionCode_OwnerWithoutTheSubjectIsRetryableAndChangesNothing(t *testing.T) {
+	e := newUpdateEnv(t)
+	before, beforeRev := e.record(t, "hello")
+	e.meta.beginErr = fmt.Errorf("NATS request to %s: %w", function.SubjectCodeUpdateBegin, nats.ErrNoResponders)
+	_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
+	testutil.RequireAWSCode(t, err, awserrors.ErrorServiceException)
+	assert.False(t, awserrors.IsTerminal(err), "the caller is told to retry once the owner is upgraded")
+	requireUnchanged(t, e, before, beforeRev)
+	assert.Equal(t, 1, e.store.count())
+	assert.Len(t, e.uploadsByRevision(t), 1)
+}
+
+func TestUpdateFunctionCode_CompetingUpdatesCommitOnce(t *testing.T) {
+	e := newUpdateEnv(t)
+	other := buildZip("hello, competing version")
+	// The second update starts and completes while the first is writing its copy.
+	e.store.beforePut = func() {
+		_, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) { b["ZipFile"] = other }))
+		assert.NoError(t, err)
+	}
+	_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
+	testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
+
+	after, _ := e.record(t, "hello")
+	cfg, err := e.do(http.MethodGet, "/2015-03-31/functions/hello/configuration", nil)
+	require.NoError(t, err)
+	assert.Equal(t, codeSHA(other), *decodeConfig(t, cfg.Body.Bytes()).CodeSha256, "the winner's code stays")
+	assert.Equal(t, 3, e.store.count(), "the original, the winner's and the loser's copies")
+	e.requireTracked(t)
+	owed := 0
+	for rev, up := range e.uploadsByRevision(t) {
+		if up.State == function.UploadOwed {
+			owed++
+			assert.NotEqual(t, after.Code.RevisionID, rev)
+		}
+	}
+	assert.Equal(t, 1, owed, "only the losing upload is owed")
+}
+
+func TestUpdateFunctionCode_LostCommitReplyIsAnErrorButTheUpdateStands(t *testing.T) {
+	e := newUpdateEnv(t)
+	before, _ := e.record(t, "hello")
+	e.meta.loseUpdatedReply = true
+	_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
+	testutil.RequireAWSCode(t, err, awserrors.ErrorServiceException)
+
+	after, _ := e.record(t, "hello")
+	assert.NotEqual(t, before.Code.RevisionID, after.Code.RevisionID, "the committed update is not rolled back")
+	cfg, err := e.do(http.MethodGet, "/2015-03-31/functions/hello/configuration", nil)
+	require.NoError(t, err)
+	assert.Equal(t, codeSHA(updateZip), *decodeConfig(t, cfg.Body.Bytes()).CodeSha256)
+	require.Equal(t, 2, e.store.count(), "neither copy is deleted")
+	ups := e.uploadsByRevision(t)
+	require.Len(t, ups, 2)
+	assert.Equal(t, function.UploadAttached, ups[before.Code.RevisionID].State)
+	assert.Equal(t, function.UploadAttached, ups[after.Code.RevisionID].State)
+}
+
+func TestUpdateFunctionCode_AnInvocationInFlightKeepsItsOldCopy(t *testing.T) {
+	e := newUpdateEnv(t)
+	ex := &fakeExec{result: okResult, started: make(chan struct{}), release: make(chan struct{})}
+	e.deps.Exec = ex
+	before, _ := e.record(t, "hello")
+	done := make(chan error, 1)
+	go func() {
+		_, err := e.invoke(t.Context(), invokePath, nil, nil)
+		done <- err
+	}()
+	<-ex.started
+
+	_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
+	require.NoError(t, err)
+	close(ex.release)
+	require.NoError(t, <-done)
+
+	ex.mu.Lock()
+	inflight := ex.calls[0].Function.Code
+	ex.started = nil
+	ex.mu.Unlock()
+	assert.Equal(t, before.Code, inflight, "the running attempt stays bound to the copy it read")
+	assert.Equal(t, testZip, e.store.objects[codestore.Ref{AccountID: gwtest.AccountID, FunctionID: before.FunctionID, RevisionID: inflight.RevisionID}],
+		"the old copy is still there for it")
+
+	_, err = e.invoke(t.Context(), invokePath, nil, nil)
+	require.NoError(t, err)
+	after, _ := e.record(t, "hello")
+	ex.mu.Lock()
+	defer ex.mu.Unlock()
+	assert.Equal(t, after.Code, ex.calls[1].Function.Code, "a later attempt reads the new copy")
+	assert.NotEqual(t, before.Code.RevisionID, ex.calls[1].Function.Code.RevisionID)
+}

spinifex/domains/lambda/awsapi/validate.go

@@ -4,6 +4,7 @@ import (
 	"archive/zip"
 	"bytes"
 	"encoding/json"
+	"errors"
 	"reflect"
 	"regexp"
 	"strings"
@@ -13,16 +14,16 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/lambdaenv"
 )
 
 const (
-	supportedRuntime  = "provided.al2023"
-	defaultTimeout    = 3
-	maxTimeout        = 900
-	defaultMemory     = 128
-	maxMemory         = 10240
-	maxDescription    = 256
-	maxEnvironmentLen = 4096
+	supportedRuntime = "provided.al2023"
+	defaultTimeout   = 3
+	maxTimeout       = 900
+	defaultMemory    = 128
+	maxMemory        = 10240
+	maxDescription   = 256
 	// maxBodyBytes bounds a CreateFunction body: the base64 of the largest
 	// code copy plus room for the rest of the document.
 	maxBodyBytes = (codestore.MaxCodeBytes/3+1)*4 + 1<<20
@@ -52,6 +53,15 @@ func unsupported(field string) error {
 	return invalid("%s is not supported", field)
 }
 
+// refuseField names a refused member; Publish also says why, since it waits on
+// function versions rather than being out of scope.
+func refuseField(field string) error {
+	if field == "Publish" {
+		return invalid("Publish is not supported: function versions are not yet supported")
+	}
+	return unsupported(field)
+}
+
 // decodeCreate reads a CreateFunction body strictly: a member this build does
 // not model is refused rather than silently dropped.
 func decodeCreate(body []byte) (*lambda.CreateFunctionInput, error) {
@@ -209,14 +219,12 @@ func validateCreate(in *lambda.CreateFunctionInput) (*createParams, error) {
 	if pt := aws.StringValue(in.PackageType); pt != "" && pt != lambda.PackageTypeZip {
 		return nil, invalid("PackageType must be %s", lambda.PackageTypeZip)
 	}
-	if in.Code == nil || len(in.Code.ZipFile) == 0 {
-		return nil, invalid("Code.ZipFile is required")
+	var zipFile []byte
+	if in.Code != nil {
+		zipFile = in.Code.ZipFile
 	}
-	if len(in.Code.ZipFile) > codestore.MaxCodeBytes {
-		return nil, awserrors.Errorf(awserrors.ErrorRequestEntityTooLarge, "Code.ZipFile exceeds %d bytes", codestore.MaxCodeBytes)
-	}
-	if _, err := zip.NewReader(bytes.NewReader(in.Code.ZipFile), int64(len(in.Code.ZipFile))); err != nil {
-		return nil, invalid("Could not unzip uploaded file. Please check your file, then try to upload again.")
+	if err := checkZip("Code.ZipFile", zipFile); err != nil {
+		return nil, err
 	}
 	handler := aws.StringValue(in.Handler)
 	if handler == "" {
@@ -253,21 +261,82 @@ func validateCreate(in *lambda.CreateFunctionInput) (*createParams, error) {
 	}, nil
 }
 
+// checkZip applies the inline ZIP rules CreateFunction and UpdateFunctionCode
+// share: present, within the code store bound, and readable as a ZIP.
+func checkZip(field string, code []byte) error {
+	if len(code) == 0 {
+		return invalid("%s is required", field)
+	}
+	if len(code) > codestore.MaxCodeBytes {
+		return awserrors.Errorf(awserrors.ErrorRequestEntityTooLarge, "%s exceeds %d bytes", field, codestore.MaxCodeBytes)
+	}
+	if _, err := zip.NewReader(bytes.NewReader(code), int64(len(code))); err != nil {
+		return invalid("Could not unzip uploaded file. Please check your file, then try to upload again.")
+	}
+	return nil
+}
+
+// decodeUpdateCode reads an UpdateFunctionCode body as strictly as
+// decodeCreate reads CreateFunction.
+func decodeUpdateCode(body []byte) (*lambda.UpdateFunctionCodeInput, error) {
+	in := new(lambda.UpdateFunctionCodeInput)
+	dec := json.NewDecoder(bytes.NewReader(body))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(in); err != nil {
+		return nil, invalid("the request body is not a valid UpdateFunctionCode document: %v", err)
+	}
+	return in, nil
+}
+
+// validateUpdateCode refuses every code source but an inline ZIP, publishing
+// and an architecture change, then checks the ZIP.
+func validateUpdateCode(in *lambda.UpdateFunctionCodeInput) error {
+	checks := []struct {
+		field string
+		set   bool
+	}{
+		{"S3Bucket", isSet(in.S3Bucket)},
+		{"S3Key", isSet(in.S3Key)},
+		{"S3ObjectVersion", isSet(in.S3ObjectVersion)},
+		{"ImageUri", isSet(in.ImageUri)},
+		{"Publish", aws.BoolValue(in.Publish)},
+	}
+	for _, c := range checks {
+		if c.set {
+			return refuseField(c.field)
+		}
+	}
+	if in.Architectures != nil {
+		if len(in.Architectures) == 0 {
+			return invalid("1 validation error detected: Value at 'architectures' failed to satisfy constraint: Member must have length greater than or equal to 1")
+		}
+		if len(in.Architectures) > 1 {
+			return invalid("1 validation error detected: Value at 'architectures' failed to satisfy constraint: Member must have length less than or equal to 1")
+		}
+		switch a := aws.StringValue(in.Architectures[0]); a {
+		case lambda.ArchitectureX8664:
+		case lambda.ArchitectureArm64:
+			return unsupported("Architectures " + a)
+		default:
+			return invalid("1 validation error detected: Value at 'architectures' failed to satisfy constraint: Member must satisfy enum value set: [x86_64, arm64]")
+		}
+	}
+	return checkZip("ZipFile", in.ZipFile)
+}
+
 func envVars(e *lambda.Environment) (map[string]string, error) {
 	if e == nil || len(e.Variables) == 0 {
 		return nil, nil
 	}
 	vars := make(map[string]string, len(e.Variables))
-	size := 0
 	for k, v := range e.Variables {
-		if k == "" {
-			return nil, invalid("Environment variable names must not be empty")
-		}
 		vars[k] = aws.StringValue(v)
-		size += len(k) + len(vars[k])
 	}
-	if size > maxEnvironmentLen {
-		return nil, invalid("Environment variables exceed %d bytes", maxEnvironmentLen)
+	if err := lambdaenv.Validate(vars); err != nil {
+		if errors.Is(err, lambdaenv.ErrName) {
+			return nil, awserrors.Errorf(awserrors.ErrorValidationException, "Environment variable names must match [a-zA-Z][a-zA-Z0-9_]+")
+		}
+		return nil, invalid("%s", err)
 	}
 	return vars, nil
 }

spinifex/domains/lambda/function/client.go

@@ -20,6 +20,10 @@ type Client interface {
 	CodeStored(ctx context.Context, accountID, region, name string, revision uint64, sha256Hex string, size int64) (Record, uint64, error)
 	// RollbackCreate frees the name while the record is still storing at revision.
 	RollbackCreate(ctx context.Context, accountID, region, name string, revision uint64) error
+	// BeginCodeUpdate records an owed upload of codeRevisionID for the function at revision.
+	BeginCodeUpdate(ctx context.Context, accountID, region string, rec Record, revision uint64, codeRevisionID string) error
+	// CodeUpdated commits the stored copy and returns the updated record and revision.
+	CodeUpdated(ctx context.Context, accountID, region string, rec Record, revision uint64, codeRevisionID, sha256Hex string, size int64) (Record, uint64, error)
 }
 
 // NATSClient is the Client over NATS request/reply to the daemon.
@@ -65,3 +69,20 @@ func (c *NATSClient) RollbackCreate(ctx context.Context, accountID, region, name
 		&RollbackCreateRequest{Region: region, FunctionName: name, Revision: revision}, metaRequestTimeout, accountID)
 	return err
 }
+
+func (c *NATSClient) BeginCodeUpdate(ctx context.Context, accountID, region string, rec Record, revision uint64, codeRevisionID string) error {
+	_, err := natsmsg.NATSRequest[CodeUpdateBeginResponse](ctx, c.conn, SubjectCodeUpdateBegin,
+		&CodeUpdateBeginRequest{Region: region, FunctionName: rec.FunctionName, FunctionID: rec.FunctionID, Revision: revision, RevisionID: codeRevisionID},
+		metaRequestTimeout, accountID)
+	return err
+}
+
+func (c *NATSClient) CodeUpdated(ctx context.Context, accountID, region string, rec Record, revision uint64, codeRevisionID, sha256Hex string, size int64) (Record, uint64, error) {
+	resp, err := natsmsg.NATSRequest[CodeUpdatedResponse](ctx, c.conn, SubjectCodeUpdated,
+		&CodeUpdatedRequest{Region: region, FunctionName: rec.FunctionName, FunctionID: rec.FunctionID, Revision: revision,
+			RevisionID: codeRevisionID, SHA256: sha256Hex, Size: size}, metaRequestTimeout, accountID)
+	if err != nil {
+		return Record{}, 0, err
+	}
+	return resp.Record, resp.Revision, nil
+}

spinifex/domains/lambda/function/record.go

@@ -103,10 +103,12 @@ func KVFunctionKey(region, name string) string { return kvFunctionKeyPrefix + re
 
 // NATS subjects served by the daemon's metadata service.
 const (
-	SubjectCreate         = "lambda.function.create"
-	SubjectGet            = "lambda.function.get"
-	SubjectCodeStored     = "lambda.function.code-stored"
-	SubjectRollbackCreate = "lambda.function.rollback-create"
+	SubjectCreate          = "lambda.function.create"
+	SubjectGet             = "lambda.function.get"
+	SubjectCodeStored      = "lambda.function.code-stored"
+	SubjectRollbackCreate  = "lambda.function.rollback-create"
+	SubjectCodeUpdateBegin = "lambda.function.code-update-begin"
+	SubjectCodeUpdated     = "lambda.function.code-updated"
 )
 
 // CreateRequest carries the record the gateway built; the daemon owns the
@@ -158,3 +160,33 @@ type RollbackCreateRequest struct {
 
 // RollbackCreateResponse is empty: success means the name is free.
 type RollbackCreateResponse struct{}
+
+// CodeUpdateBeginRequest asks for an owed upload of new code for the function
+// as it stood at Revision; RevisionID names the new code copy.
+type CodeUpdateBeginRequest struct {
+	Region       string `json:"region"`
+	FunctionName string `json:"functionName"`
+	FunctionID   string `json:"functionId"`
+	Revision     uint64 `json:"revision"`
+	RevisionID   string `json:"revisionId"`
+}
+
+// CodeUpdateBeginResponse is empty: success means the upload is owed.
+type CodeUpdateBeginResponse struct{}
+
+// CodeUpdatedRequest names the stored new copy for the function at Revision.
+type CodeUpdatedRequest struct {
+	Region       string `json:"region"`
+	FunctionName string `json:"functionName"`
+	FunctionID   string `json:"functionId"`
+	Revision     uint64 `json:"revision"`
+	RevisionID   string `json:"revisionId"`
+	SHA256       string `json:"sha256"`
+	Size         int64  `json:"size"`
+}
+
+// CodeUpdatedResponse returns the committed record and its revision.
+type CodeUpdatedResponse struct {
+	Record   Record `json:"record"`
+	Revision uint64 `json:"revision"`
+}

spinifex/domains/lambda/function/update.go

@@ -0,0 +1,121 @@
+package function
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// readServing loads the record a code update names and checks it is still the
+// one the caller read: same revision and function, Active, with stored code.
+func readServing(ctx context.Context, kv jetstream.KeyValue, region, name, functionID string, revision uint64) (Record, error) {
+	entry, err := kv.Get(ctx, KVFunctionKey(region, name))
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return Record{}, awserrors.Errorf(awserrors.ErrorResourceNotFoundException, "Function not found: %s", name)
+	}
+	if err != nil {
+		return Record{}, fmt.Errorf("lambda: get function record: %w", err)
+	}
+	if entry.Revision() != revision {
+		return Record{}, conflict("Function was modified by another request: %s", name)
+	}
+	rec, err := decode(entry.Value())
+	if err != nil {
+		return Record{}, err
+	}
+	if rec.FunctionID != functionID {
+		return Record{}, conflict("Function was modified by another request: %s", name)
+	}
+	if rec.State != StateActive || rec.Code.State != CodeStored {
+		return Record{}, conflict("The operation cannot be performed at this time. The function is currently in the following state: %s", rec.State)
+	}
+	return rec, nil
+}
+
+func validUpdate(region, name, functionID, revisionID, accountID string) error {
+	if !regionRE.MatchString(region) {
+		return badRegion()
+	}
+	if !ValidName(name) || !idRE.MatchString(functionID) || !idRE.MatchString(revisionID) || accountID == "" {
+		return awserrors.Errorf(awserrors.ErrorInvalidParameterValueException, "code update request is incomplete")
+	}
+	return nil
+}
+
+// BeginCodeUpdate records an owed upload for a new code copy of the function
+// at the caller's revision. The function record is left unchanged, so invokes
+// keep running the current copy while the new one is written.
+func (s *MetaService) BeginCodeUpdate(ctx context.Context, req *CodeUpdateBeginRequest, accountID string) (*CodeUpdateBeginResponse, error) {
+	if err := validUpdate(req.Region, req.FunctionName, req.FunctionID, req.RevisionID, accountID); err != nil {
+		return nil, err
+	}
+	kv, err := s.bucket(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+	rec, err := readServing(ctx, kv, req.Region, req.FunctionName, req.FunctionID, req.Revision)
+	if err != nil {
+		return nil, err
+	}
+	if rec.Code.RevisionID == req.RevisionID {
+		return nil, conflict("Function code revision is already in use: %s", req.FunctionName)
+	}
+	data, err := json.Marshal(Upload{
+		FormatVersion: FormatVersion, AccountID: accountID, Region: req.Region, FunctionName: req.FunctionName,
+		FunctionID: req.FunctionID, RevisionID: req.RevisionID, State: UploadOwed, CreatedAt: s.now().UTC(),
+	})
+	if err != nil {
+		return nil, err
+	}
+	if _, err := kv.Create(ctx, KVUploadKey(req.Region, req.FunctionID, req.RevisionID), data); err != nil {
+		if errors.Is(err, jetstream.ErrKeyExists) {
+			return nil, conflict("Function code revision is already in use: %s", req.FunctionName)
+		}
+		return nil, fmt.Errorf("lambda: record code upload: %w", err)
+	}
+	return &CodeUpdateBeginResponse{}, nil
+}
+
+// CodeUpdated points the function at its new stored copy by compare-and-swap at
+// the revision the update began from; only the code and last-modified time
+// change. The previous copy and its upload record are kept.
+func (s *MetaService) CodeUpdated(ctx context.Context, req *CodeUpdatedRequest, accountID string) (*CodeUpdatedResponse, error) {
+	if err := validUpdate(req.Region, req.FunctionName, req.FunctionID, req.RevisionID, accountID); err != nil {
+		return nil, err
+	}
+	if !digestRE.MatchString(req.SHA256) || req.Size <= 0 {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValueException, "code digest is invalid")
+	}
+	kv, err := s.bucket(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+	rec, err := readServing(ctx, kv, req.Region, req.FunctionName, req.FunctionID, req.Revision)
+	if err != nil {
+		return nil, err
+	}
+	next := rec
+	next.Code = Code{RevisionID: req.RevisionID, SHA256: req.SHA256, Size: req.Size, State: CodeStored}
+	up, upRev, err := checkUpload(ctx, kv, next)
+	if err != nil {
+		return nil, err
+	}
+	next.LastModified = s.now().UTC()
+	data, err := json.Marshal(next)
+	if err != nil {
+		return nil, err
+	}
+	rev, err := kv.Update(ctx, KVFunctionKey(req.Region, req.FunctionName), data, req.Revision)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrKeyExists) {
+			return nil, conflict("Function was modified by another request: %s", req.FunctionName)
+		}
+		return nil, fmt.Errorf("lambda: update function record: %w", err)
+	}
+	attach(ctx, kv, up, upRev, rev)
+	return &CodeUpdatedResponse{Record: next, Revision: rev}, nil
+}

spinifex/domains/lambda/function/update_test.go

@@ -0,0 +1,211 @@
+package function
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const updateDigest = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+
+// mustServe creates "hello" with stored code and activates it.
+func mustServe(t *testing.T, svc *MetaService) *GetResponse {
+	t.Helper()
+	rev := mustCreate(t, svc, "hello", "f1")
+	_, err := svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: rev, SHA256: testDigest, Size: 9}, testAccount)
+	require.NoError(t, err)
+	svc.SetReadiness(func(context.Context, Record) error { return nil })
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	got := mustGet(t, svc, "hello")
+	require.Equal(t, StateActive, got.Record.State)
+	return got
+}
+
+func beginReq(revision uint64, codeRevisionID string) *CodeUpdateBeginRequest {
+	return &CodeUpdateBeginRequest{Region: testRegion, FunctionName: "hello", FunctionID: "f1", Revision: revision, RevisionID: codeRevisionID}
+}
+
+func commitReq(revision uint64, codeRevisionID string) *CodeUpdatedRequest {
+	return &CodeUpdatedRequest{Region: testRegion, FunctionName: "hello", FunctionID: "f1", Revision: revision,
+		RevisionID: codeRevisionID, SHA256: updateDigest, Size: 12}
+}
+
+func uploadOf(t *testing.T, svc *MetaService, codeRevisionID string) (Upload, bool) {
+	t.Helper()
+	kv, err := svc.bucket(t.Context(), testAccount)
+	require.NoError(t, err)
+	entry, err := kv.Get(t.Context(), KVUploadKey(testRegion, "f1", codeRevisionID))
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return Upload{}, false
+	}
+	require.NoError(t, err)
+	var up Upload
+	require.NoError(t, json.Unmarshal(entry.Value(), &up))
+	return up, true
+}
+
+func TestCodeUpdate_CommitChangesOnlyTheCodeReference(t *testing.T) {
+	svc, c := newTestService(t)
+	before := mustServe(t, svc)
+	c.t = c.t.Add(time.Minute)
+
+	_, err := svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, "rev-2"), testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, before.Revision, mustGet(t, svc, "hello").Revision, "beginning an update leaves the record as it was")
+	up, ok := uploadOf(t, svc, "rev-2")
+	require.True(t, ok)
+	assert.Equal(t, UploadOwed, up.State)
+
+	resp, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-2"), testAccount)
+	require.NoError(t, err)
+	got := mustGet(t, svc, "hello")
+	assert.Equal(t, resp.Revision, got.Revision)
+	assert.Equal(t, Code{RevisionID: "rev-2", SHA256: updateDigest, Size: 12, State: CodeStored}, got.Record.Code)
+	assert.Equal(t, c.t, got.Record.LastModified)
+
+	want := before.Record
+	want.Code, want.LastModified = got.Record.Code, got.Record.LastModified
+	assert.Equal(t, want, got.Record, "identity, configuration and state are unchanged")
+
+	up, _ = uploadOf(t, svc, "rev-2")
+	assert.Equal(t, UploadAttached, up.State)
+	assert.Equal(t, resp.Revision, up.AttachedRevision)
+	old, ok := uploadOf(t, svc, "rev-f1")
+	require.True(t, ok, "the previous copy stays tracked")
+	assert.Equal(t, UploadAttached, old.State)
+}
+
+func TestCodeUpdate_OnlyOneOfTwoUpdatesFromTheSameRevisionCommits(t *testing.T) {
+	svc, _ := newTestService(t)
+	before := mustServe(t, svc)
+	for _, id := range []string{"rev-a", "rev-b"} {
+		_, err := svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, id), testAccount)
+		require.NoError(t, err)
+	}
+	_, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-b"), testAccount)
+	require.NoError(t, err)
+	_, err = svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-a"), testAccount)
+	testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
+
+	assert.Equal(t, "rev-b", mustGet(t, svc, "hello").Record.Code.RevisionID)
+	loser, _ := uploadOf(t, svc, "rev-a")
+	assert.Equal(t, UploadOwed, loser.State, "the losing copy stays owed for cleanup")
+
+	_, err = svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, "rev-c"), testAccount)
+	testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
+	_, ok := uploadOf(t, svc, "rev-c")
+	assert.False(t, ok, "an update from a stale revision records nothing")
+}
+
+// Both commits read the same record before either writes: the loser is held
+// between its checks and its conditional write while the winner commits.
+func TestCodeUpdate_InterleavedCommitLosesTheConditionalWrite(t *testing.T) {
+	svc, c := newTestService(t)
+	before := mustServe(t, svc)
+	for _, id := range []string{"rev-a", "rev-b"} {
+		_, err := svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, id), testAccount)
+		require.NoError(t, err)
+	}
+	paused, release := make(chan struct{}), make(chan struct{})
+	var calls atomic.Int32
+	svc.now = func() time.Time {
+		if calls.Add(1) == 1 {
+			close(paused)
+			<-release
+		}
+		return c.now()
+	}
+	loserErr := make(chan error, 1)
+	go func() {
+		_, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-a"), testAccount)
+		loserErr <- err
+	}()
+	<-paused
+	won, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-b"), testAccount)
+	require.NoError(t, err)
+	close(release)
+	testutil.RequireAWSCode(t, <-loserErr, awserrors.ErrorResourceConflictException)
+
+	got := mustGet(t, svc, "hello")
+	assert.Equal(t, won.Revision, got.Revision)
+	assert.Equal(t, "rev-b", got.Record.Code.RevisionID)
+	winner, _ := uploadOf(t, svc, "rev-b")
+	assert.Equal(t, UploadAttached, winner.State)
+	assert.Equal(t, won.Revision, winner.AttachedRevision)
+	loser, ok := uploadOf(t, svc, "rev-a")
+	require.True(t, ok, "the losing copy stays tracked")
+	assert.Equal(t, UploadOwed, loser.State)
+}
+
+func TestCodeUpdate_CommitNeedsItsOwnOwedUpload(t *testing.T) {
+	svc, _ := newTestService(t)
+	before := mustServe(t, svc)
+	_, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-never-begun"), testAccount)
+	testutil.RequireAWSCode(t, err, awserrors.ErrorServiceException)
+
+	_, err = svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, "rev-f1"), testAccount)
+	testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
+
+	other := commitReq(before.Revision, "rev-2")
+	_, err = svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, "rev-2"), testAccount)
+	require.NoError(t, err)
+	other.FunctionID = "f-other"
+	_, err = svc.CodeUpdated(t.Context(), other, testAccount)
+	testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
+	assert.Equal(t, before.Revision, mustGet(t, svc, "hello").Revision)
+}
+
+func TestCodeUpdate_RefusesAFunctionThatIsNotServing(t *testing.T) {
+	svc, _ := newTestService(t)
+	rev := mustCreate(t, svc, "hello", "f1")
+	_, err := svc.BeginCodeUpdate(t.Context(), beginReq(rev, "rev-2"), testAccount)
+	testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
+	_, ok := uploadOf(t, svc, "rev-2")
+	assert.False(t, ok)
+
+	ghost := beginReq(1, "rev-2")
+	ghost.FunctionName = "ghost"
+	_, err = svc.BeginCodeUpdate(t.Context(), ghost, testAccount)
+	testutil.RequireAWSCode(t, err, awserrors.ErrorResourceNotFoundException)
+}
+
+func TestCodeUpdate_RefusesMalformedRequests(t *testing.T) {
+	svc, _ := newTestService(t)
+	before := mustServe(t, svc)
+	begins := map[string]func(*CodeUpdateBeginRequest){
+		"region":      func(r *CodeUpdateBeginRequest) { r.Region = "Bad Region" },
+		"name":        func(r *CodeUpdateBeginRequest) { r.FunctionName = "a/b" },
+		"function id": func(r *CodeUpdateBeginRequest) { r.FunctionID = "" },
+		"revision id": func(r *CodeUpdateBeginRequest) { r.RevisionID = "../x" },
+	}
+	for name, mutate := range begins {
+		req := beginReq(before.Revision, "rev-2")
+		mutate(req)
+		t.Run("begin "+name, func(t *testing.T) {
+			_, err := svc.BeginCodeUpdate(t.Context(), req, testAccount)
+			testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
+		})
+	}
+	commits := map[string]func(*CodeUpdatedRequest){
+		"digest": func(r *CodeUpdatedRequest) { r.SHA256 = "xyz" },
+		"size":   func(r *CodeUpdatedRequest) { r.Size = 0 },
+	}
+	for name, mutate := range commits {
+		req := commitReq(before.Revision, "rev-2")
+		mutate(req)
+		t.Run("commit "+name, func(t *testing.T) {
+			_, err := svc.CodeUpdated(t.Context(), req, testAccount)
+			testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
+		})
+	}
+	assert.Equal(t, before.Revision, mustGet(t, svc, "hello").Revision)
+}

spinifex/foundation/aws/errors/awserrors.go

@@ -629,6 +629,7 @@ var (
 	ErrorServiceException               = "ServiceException"
 	ErrorRequestEntityTooLargeException = "RequestEntityTooLargeException"
 	ErrorInvalidRequestContentException = "InvalidRequestContentException"
+	ErrorPreconditionFailedException    = "PreconditionFailedException"
 )
 
 // ResolveErrorCode returns the first registered AWS error code in err's unwrap tree.
@@ -1431,6 +1432,7 @@ var ErrorLookup = map[string]ErrorMessage{
 	ErrorServiceException:                  {HTTPCode: 500, Message: "The Lambda service encountered an internal error."},
 	ErrorRequestEntityTooLargeException:    {HTTPCode: 413, Message: "The request payload exceeded the Invoke request body JSON input quota."},
 	ErrorInvalidRequestContentException:    {HTTPCode: 400, Message: "The request body could not be parsed as JSON."},
+	ErrorPreconditionFailedException:       {HTTPCode: 412, Message: "The RevisionId provided does not match the latest RevisionId for the Lambda function or alias."},
 
 	// Route 53.
 	ErrorNoSuchHostedZone:          {HTTPCode: 404, Message: "No hosted zone exists with the ID that you specified."},

spinifex/foundation/aws/lambdaenv/env.go

@@ -0,0 +1,50 @@
+// Package lambdaenv validates function environment variables.
+package lambdaenv
+
+import (
+	"errors"
+	"regexp"
+	"strings"
+)
+
+var (
+	ErrName     = errors.New("invalid environment variable name")
+	ErrReserved = errors.New("reserved environment variable name")
+	ErrValue    = errors.New("environment variable contains a NUL byte")
+	ErrSize     = errors.New("environment variables exceed 4096 bytes")
+)
+
+var namePattern = regexp.MustCompile(`^[a-zA-Z][a-zA-Z0-9_]+$`)
+
+var reserved = map[string]struct{}{
+	"AWS_ACCESS_KEY": {}, "AWS_ACCESS_KEY_ID": {}, "AWS_SECRET_ACCESS_KEY": {}, "AWS_SESSION_TOKEN": {},
+	"AWS_DEFAULT_REGION": {}, "AWS_REGION": {}, "AWS_EXECUTION_ENV": {},
+	"AWS_LAMBDA_RUNTIME_API": {}, "AWS_LAMBDA_FUNCTION_NAME": {}, "AWS_LAMBDA_FUNCTION_MEMORY_SIZE": {},
+	"AWS_LAMBDA_FUNCTION_VERSION": {}, "AWS_LAMBDA_INITIALIZATION_TYPE": {},
+	"AWS_LAMBDA_LOG_GROUP_NAME": {}, "AWS_LAMBDA_LOG_STREAM_NAME": {},
+	"AWS_LAMBDA_MAX_CONCURRENCY": {}, "AWS_LAMBDA_METADATA_API": {}, "AWS_LAMBDA_METADATA_TOKEN": {},
+	"LAMBDA_TASK_ROOT": {}, "LAMBDA_RUNTIME_DIR": {},
+}
+
+// Validate checks values without including customer names or values in errors.
+func Validate(vars map[string]string) error {
+	for k := range vars {
+		if !namePattern.MatchString(k) {
+			return ErrName
+		}
+	}
+	size := 0
+	for k, v := range vars {
+		if _, ok := reserved[k]; ok {
+			return ErrReserved
+		}
+		if strings.ContainsRune(v, '\x00') {
+			return ErrValue
+		}
+		size += len(k) + len(v)
+	}
+	if size > 4096 {
+		return ErrSize
+	}
+	return nil
+}

spinifex/foundation/aws/lambdaenv/env_test.go

@@ -0,0 +1,33 @@
+package lambdaenv_test
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/lambdaenv"
+)
+
+func TestValidate(t *testing.T) {
+	for name, tc := range map[string]struct {
+		vars map[string]string
+		want error
+	}{
+		"absent": {},
+		"literal values and configurable settings": {vars: map[string]string{"GREETING": "hello $WORLD", "PATH": "/my/bin", "AWS_ENDPOINT_URL": "https://example.test"}},
+		"maximum size":                   {vars: map[string]string{"AA": strings.Repeat("x", 4094)}},
+		"empty name":                     {vars: map[string]string{"": "x"}, want: lambdaenv.ErrName},
+		"one-character name":             {vars: map[string]string{"A": "x"}, want: lambdaenv.ErrName},
+		"invalid character":              {vars: map[string]string{"bad-name": "x"}, want: lambdaenv.ErrName},
+		"reserved credential":            {vars: map[string]string{"AWS_SECRET_ACCESS_KEY": "x"}, want: lambdaenv.ErrReserved},
+		"invalid name precedes reserved": {vars: map[string]string{"A": "x", "AWS_REGION": "x"}, want: lambdaenv.ErrName},
+		"nul value":                      {vars: map[string]string{"VALUE": "x\x00"}, want: lambdaenv.ErrValue},
+		"over maximum":                   {vars: map[string]string{"AA": strings.Repeat("x", 4095)}, want: lambdaenv.ErrSize},
+	} {
+		t.Run(name, func(t *testing.T) {
+			if got := lambdaenv.Validate(tc.vars); !errors.Is(got, tc.want) {
+				t.Fatalf("Validate() = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}

spinifex/ingress/aws/pipeline/lambda_invoke_wire_test.go

@@ -47,6 +47,14 @@ func (activeMeta) RollbackCreate(context.Context, string, string, string, uint64
 	return errors.New("unused")
 }
 
+func (activeMeta) BeginCodeUpdate(context.Context, string, string, function.Record, uint64, string) error {
+	return errors.New("unused")
+}
+
+func (activeMeta) CodeUpdated(context.Context, string, string, function.Record, uint64, string, string, int64) (function.Record, uint64, error) {
+	return function.Record{}, 0, errors.New("unused")
+}
+
 // countingExec records how many invocations reached it.
 type countingExec struct{ calls atomic.Int32 }
 

spinifex/runtime/compute/lambda/guestconn/protocol_testlink_test.go

@@ -18,8 +18,12 @@ import (
 
 // releaseOnly is cfg as a release build would offer it: ALPNv1 alone.
 func releaseOnly(cfg *tls.Config) *tls.Config {
+	return protocolOnly(cfg, lambdav1.ALPNv1)
+}
+
+func protocolOnly(cfg *tls.Config, protocol string) *tls.Config {
 	cfg = cfg.Clone()
-	cfg.NextProtos = []string{lambdav1.ALPNv1}
+	cfg.NextProtos = []string{protocol}
 	verify := cfg.VerifyConnection
 	cfg.VerifyConnection = func(cs tls.ConnectionState) error {
 		cs.NegotiatedProtocol = hostlink.Protocol // let only the ALPN offer differ
@@ -28,6 +32,42 @@ func releaseOnly(cfg *tls.Config) *tls.Config {
 	return cfg
 }
 
+func TestTestLinkRejectsPreviousProtocolBothWays(t *testing.T) {
+	const previous = "spinifex-lambda-testlink/2"
+	t.Run("old guest", func(t *testing.T) {
+		ln := listen(t)
+		g := newGeneration(t, gen1)
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		hostSide := acceptAsync(g, ln, ctx, 2*time.Second)
+		id := guestFor(t, g, ln.Addr().String())
+		raw, err := net.Dial("tcp", ln.Addr().String())
+		require.NoError(t, err)
+		defer raw.Close()
+		require.Error(t, tls.Client(raw, protocolOnly(id.ClientConfig(), previous)).HandshakeContext(ctx))
+		require.Error(t, (<-hostSide).err)
+	})
+	t.Run("old host", func(t *testing.T) {
+		ln := listen(t)
+		g := newGeneration(t, gen1)
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		hostErr := make(chan error, 1)
+		go func() {
+			c, err := ln.Accept()
+			if err != nil {
+				hostErr <- err
+				return
+			}
+			defer c.Close()
+			hostErr <- tls.Server(c, protocolOnly(g.ServerConfig(), previous)).HandshakeContext(ctx)
+		}()
+		_, err := hostlink.Dial(ctx, guestFor(t, g, ln.Addr().String()))
+		require.Error(t, err)
+		require.Error(t, <-hostErr)
+	})
+}
+
 func TestTestLinkProtocolIsNotTheReleaseProtocol(t *testing.T) {
 	assert.NotEqual(t, lambdav1.ALPNv1, hostlink.Protocol)
 }

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/3"
+const ControlName = "spinifex-lambda-testlink-control/4"
 
 // Service accounts at each end of the control socket.
 const (
@@ -43,9 +43,8 @@ func joinAfterFunc(ctx context.Context, f func()) func() {
 	}
 }
 
-// MaxCredentialBytes bounds the encoded credential; a session's three values
-// fit well inside it.
-const MaxCredentialBytes = 8 << 10
+// MaxCredentialBytes includes the credential and JSON-escaped function environment.
+const MaxCredentialBytes = 40 << 10
 
 // Declared control-socket bounds, independent of any message bus. Request and
 // Reply payloads are capped raw; the frame cap leaves MetadataHeadroomBytes
@@ -61,8 +60,8 @@ const (
 	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.
+// Credential carries one execution-role session and its function environment
+// to the guest, never on NATS, to disk or into a log.
 type Credential = hostlink.Credential
 
 // ErrNotSent means the exchange failed before any request byte was written, so

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

@@ -83,6 +83,17 @@ func TestControl_OneRequestOneReply(t *testing.T) {
 	assert.Equal(t, testCred, s.cred.Load())
 }
 
+func TestControl_FullEnvironmentFitsWithCredential(t *testing.T) {
+	name := controlName(t)
+	s, _ := serveControl(t, name, os.Getuid(), nil)
+	cred := testCred
+	cred.SessionToken = string(bytes.Repeat([]byte("t"), 8<<10-128))
+	cred.Environment = map[string]string{"AA": string(bytes.Repeat([]byte{1}, 4094))}
+	_, err := client(name, os.Getuid()).Exchange(t.Context(), []byte(`{}`), cred, soon())
+	require.NoError(t, err)
+	assert.Equal(t, cred, s.cred.Load())
+}
+
 func TestControl_RequestSizeBoundary(t *testing.T) {
 	name := controlName(t)
 	s, _ := serveControl(t, name, os.Getuid(), nil)
@@ -219,7 +230,7 @@ func TestControl_BadFramesAreRefused(t *testing.T) {
 		"incomplete credential": append(rawFrame([]byte("{}")),
 			rawFrame([]byte(`{"access_key_id":"ASIA"}`))...),
 		"oversized credential": append(rawFrame([]byte("{}")),
-			rawFrame(bytes.Repeat([]byte("c"), MaxCredentialBytes+1))...),
+			rawFrame(bytes.Repeat([]byte("c"), MaxCredentialBytes+1))[:4]...),
 	}
 	for name, frame := range cases {
 		t.Run(name, func(t *testing.T) {
@@ -441,6 +452,7 @@ func TestControl_WriteErrorAfterDeliveryIsUnknown(t *testing.T) {
 const (
 	legacyControlName = "spinifex-lambda-testlink-control"
 	v2ControlName     = "spinifex-lambda-testlink-control/2"
+	v3ControlName     = "spinifex-lambda-testlink-control/3"
 )
 
 // A gateway and daemon on different control protocol names never connect, in
@@ -453,6 +465,8 @@ func TestControl_MixedProtocolNamesNeverConnect(t *testing.T) {
 		"new daemon, unversioned gateway": {prefix + ControlName, prefix + legacyControlName},
 		"/2 daemon, new gateway":          {prefix + v2ControlName, prefix + ControlName},
 		"new daemon, /2 gateway":          {prefix + ControlName, prefix + v2ControlName},
+		"/3 daemon, new gateway":          {prefix + v3ControlName, prefix + ControlName},
+		"new daemon, /3 gateway":          {prefix + ControlName, prefix + v3ControlName},
 	} {
 		t.Run(name, func(t *testing.T) {
 			s, _ := serveControl(t, tc.daemon, os.Getuid(), nil)
@@ -478,7 +492,9 @@ func TestControl_AcceptStoppedRunsBeforeTheDrain(t *testing.T) {
 		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()) }()
+	go func() {
+		_, _ = client(name, os.Getuid()).Exchange(context.Background(), []byte("{}"), testCred, soon())
+	}()
 	select {
 	case <-inHandle:
 	case <-time.After(5 * time.Second):

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

@@ -64,7 +64,9 @@ func TestDaemon_StartThenFailedLaunchedWriteIsUnknown(t *testing.T) {
 	owner := &failLaunched{Grants: d.Grants}
 	d.Grants = owner
 
-	rep := d.Handle(context.Background(), g.request(srv.URL, code), testCred)
+	cred := testCred
+	cred.Environment = map[string]string{"GREETING": "private-function-environment"}
+	rep := d.Handle(context.Background(), g.request(srv.URL, code), cred)
 	// An identity is read only from a live process's pidfd, so one was started.
 	require.Len(t, owner.got, 1, "no process was started, so this run proves nothing about after-start replies")
 	require.NotZero(t, owner.got[0].PID)
@@ -72,7 +74,7 @@ func TestDaemon_StartThenFailedLaunchedWriteIsUnknown(t *testing.T) {
 	require.Equal(t, OutcomeUnknown, rep.Outcome, rep.Error)
 	require.Equal(t, ErrLaunchNotRecorded.Error(), rep.Error)
 
-	secrets := []string{testCred.AccessKeyID, testCred.SecretAccessKey, testCred.SessionToken, srv.URL}
+	secrets := []string{cred.AccessKeyID, cred.SecretAccessKey, cred.SessionToken, cred.Environment["GREETING"], srv.URL}
 	files := 0
 	require.NoError(t, filepath.WalkDir(d.WorkDir, func(p string, e fs.DirEntry, err error) error {
 		if err != nil || e.IsDir() {

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

@@ -13,9 +13,6 @@ import (
 // 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.")
 		}

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

@@ -133,8 +133,7 @@ func TestReadiness_RefusesFunctionsTheTestLinkCannotRun(t *testing.T) {
 	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, function.StateActive, 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")
@@ -159,7 +158,9 @@ func TestReadiness_ServingEndsWhenAcceptingStops(t *testing.T) {
 	}
 	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)) }()
+	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):

spinifex/runtime/daemon/daemon.go

@@ -1359,6 +1359,8 @@ func (d *Daemon) subscribeAll() error {
 			natsSub{lambdafunction.SubjectGet, handleNATSRequest(d.node, d.lambdaMetaService.Get), "spinifex-workers"},
 			natsSub{lambdafunction.SubjectCodeStored, handleNATSRequest(d.node, d.lambdaMetaService.CodeStored), "spinifex-workers"},
 			natsSub{lambdafunction.SubjectRollbackCreate, handleNATSRequest(d.node, d.lambdaMetaService.RollbackCreate), "spinifex-workers"},
+			natsSub{lambdafunction.SubjectCodeUpdateBegin, handleNATSRequest(d.node, d.lambdaMetaService.BeginCodeUpdate), "spinifex-workers"},
+			natsSub{lambdafunction.SubjectCodeUpdated, handleNATSRequest(d.node, d.lambdaMetaService.CodeUpdated), "spinifex-workers"},
 		)
 	}
 

spinifex/runtime/daemon/daemon_lambda_update_test.go

@@ -0,0 +1,90 @@
+package daemon
+
+import (
+	"context"
+	"testing"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	lambdafunction "github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const lambdaTestAccount = "000000000001"
+
+func serveLambdaSubjects(t *testing.T, nc *nats.Conn, svc *lambdafunction.MetaService, withUpdate bool) {
+	t.Helper()
+	handlers := map[string]nats.MsgHandler{
+		lambdafunction.SubjectCreate:         asMsgHandler(handleNATSRequest("node-1", svc.Create)),
+		lambdafunction.SubjectGet:            asMsgHandler(handleNATSRequest("node-1", svc.Get)),
+		lambdafunction.SubjectCodeStored:     asMsgHandler(handleNATSRequest("node-1", svc.CodeStored)),
+		lambdafunction.SubjectRollbackCreate: asMsgHandler(handleNATSRequest("node-1", svc.RollbackCreate)),
+	}
+	if withUpdate {
+		handlers[lambdafunction.SubjectCodeUpdateBegin] = asMsgHandler(handleNATSRequest("node-1", svc.BeginCodeUpdate))
+		handlers[lambdafunction.SubjectCodeUpdated] = asMsgHandler(handleNATSRequest("node-1", svc.CodeUpdated))
+	}
+	for subject, h := range handlers {
+		sub, err := nc.QueueSubscribe(subject, "spinifex-workers", h)
+		require.NoError(t, err)
+		t.Cleanup(func() { _ = sub.Unsubscribe() })
+	}
+	require.NoError(t, nc.Flush())
+}
+
+// A gateway that can update code may reach a daemon that predates it: the
+// request must fail as a retryable transport error and change nothing, and the
+// existing Get path must read a record an upgraded daemon has updated.
+func TestLambdaCodeUpdate_PreviousDaemonRefusesWithoutMutation(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	svc := lambdafunction.NewMetaService(js)
+	svc.SetReadiness(func(context.Context, lambdafunction.Record) error { return nil })
+	client := lambdafunction.NewNATSClient(nc)
+	ctx := t.Context()
+	const region, digest = "ap-southeast-2", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+
+	serveLambdaSubjects(t, nc, svc, false)
+	rev, err := client.Create(ctx, lambdaTestAccount, region, lambdafunction.Record{
+		FunctionID: "f1", FunctionName: "hello", Role: "arn:aws:iam::000000000001:role/r", Runtime: "provided.al2023",
+		Handler: "bootstrap", Timeout: 3, MemorySize: 128, Code: lambdafunction.Code{RevisionID: "rev-1"},
+	})
+	require.NoError(t, err)
+	_, _, err = client.CodeStored(ctx, lambdaTestAccount, region, "hello", rev, digest, 9)
+	require.NoError(t, err)
+	require.NoError(t, svc.ActivatePending(ctx))
+	rec, rev, err := client.Get(ctx, lambdaTestAccount, region, "hello")
+	require.NoError(t, err)
+	require.Equal(t, lambdafunction.StateActive, rec.State)
+
+	err = client.BeginCodeUpdate(ctx, lambdaTestAccount, region, rec, rev, "rev-2")
+	require.ErrorIs(t, err, nats.ErrNoResponders)
+	_, coded := awserrors.ResolveErrorCode(err)
+	assert.False(t, coded, "the gateway maps an uncoded failure to a retryable ServiceException")
+	_, _, err = client.CodeUpdated(ctx, lambdaTestAccount, region, rec, rev, "rev-2", digest, 9)
+	require.ErrorIs(t, err, nats.ErrNoResponders)
+	same, sameRev, err := client.Get(ctx, lambdaTestAccount, region, "hello")
+	require.NoError(t, err)
+	assert.Equal(t, rev, sameRev)
+	assert.Equal(t, rec, same)
+	kv, err := js.KeyValue(ctx, lambdafunction.KVAccountBucket(lambdaTestAccount))
+	require.NoError(t, err)
+	_, err = kv.Get(ctx, lambdafunction.KVUploadKey(region, "f1", "rev-2"))
+	assert.ErrorIs(t, err, jetstream.ErrKeyNotFound, "no upload was recorded")
+
+	// Once the daemon serves the new subjects the update lands, and the old
+	// Get request still decodes the updated record.
+	serveLambdaSubjects(t, nc, svc, true)
+	const newDigest = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+	require.NoError(t, client.BeginCodeUpdate(ctx, lambdaTestAccount, region, rec, rev, "rev-2"))
+	_, updatedRev, err := client.CodeUpdated(ctx, lambdaTestAccount, region, rec, rev, "rev-2", newDigest, 12)
+	require.NoError(t, err)
+	got, gotRev, err := client.Get(ctx, lambdaTestAccount, region, "hello")
+	require.NoError(t, err)
+	assert.Equal(t, updatedRev, gotRev)
+	assert.Equal(t, lambdafunction.FormatVersion, got.FormatVersion)
+	assert.Equal(t, lambdafunction.Code{RevisionID: "rev-2", SHA256: newDigest, Size: 12, State: lambdafunction.CodeStored}, got.Code)
+	assert.Equal(t, lambdafunction.StateActive, got.State)
+}

spinifex/runtime/roles/awsgw/awsgw.go

@@ -407,7 +407,7 @@ func launchService(config *config.ClusterConfig) error {
 		Roles:  iamdomain.NewRoleResolver(iamService),
 		Trusts: lambdaRoleTrusts(iamService),
 		AEAD:   lambdaKey.AEAD,
-		Exec:   lambdaTestExecutor(natsConn, config.Node, lambdaCodeStore, stsService),
+		Exec:   lambdaTestExecutor(natsConn, config.Node, lambdaCodeStore, stsService, lambdaKey.AEAD),
 		Mappings: func(ctx context.Context) (lambdaawsapi.MappingStore, error) {
 			return lambdaMappings(ctx)
 		},

spinifex/runtime/roles/awsgw/lambda_testlink.go

@@ -4,6 +4,7 @@ package awsgw
 
 import (
 	"context"
+	"crypto/cipher"
 	"log/slog"
 	"time"
 
@@ -18,7 +19,7 @@ import (
 
 // lambdaTestExecutor sends each Invoke to this node's test-link daemon over
 // the control socket. Without the daemon's service account it does not serve.
-func lambdaTestExecutor(nc *nats.Conn, node string, codes *codestore.Manager, sessions lambdaEnvironmentRoleIssuer) awsapi.Executor {
+func lambdaTestExecutor(nc *nats.Conn, node string, codes *codestore.Manager, sessions lambdaEnvironmentRoleIssuer, aead cipher.AEAD) awsapi.Executor {
 	daemonUID, err := testlink.LookupUID(testlink.DaemonUser)
 	if err != nil {
 		slog.Error("Lambda test link: control peer account unavailable; Invoke is not served", "err", err)
@@ -31,6 +32,7 @@ func lambdaTestExecutor(nc *nats.Conn, node string, codes *codestore.Manager, se
 		Control:      testLinkControl(daemonUID),
 		RequestLimit: testlink.MaxFrameBytes,
 		Sessions:     sessions,
+		AEAD:         aead,
 		Node:         node,
 		ImageDigest:  lambdaImageDigest,
 		Codes: func(ctx context.Context) (codePresigner, error) {

spinifex/runtime/roles/awsgw/lambda_testlink_executor.go

@@ -4,6 +4,7 @@ package awsgw
 
 import (
 	"context"
+	"crypto/cipher"
 	"crypto/rand"
 	"encoding/hex"
 	"encoding/json"
@@ -21,6 +22,7 @@ import (
 	stsdomain "github.com/mulgadc/spinifex/spinifex/domains/sts"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/lambdaenv"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/testlink"
 )
 
@@ -58,12 +60,13 @@ type testLinkExecutor struct {
 	// RequestLimit bounds the encoded request frame, without its credential.
 	RequestLimit int64
 	Sessions     lambdaEnvironmentRoleIssuer
+	AEAD         cipher.AEAD
 	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
+	Codes       func(context.Context) (codePresigner, error)
+	Grants      func(context.Context) (grantIssuer, error)
+	now         func() time.Time
 }
 
 var _ awsapi.Executor = (*testLinkExecutor)(nil)
@@ -78,11 +81,16 @@ func (e *testLinkExecutor) InvokeSync(ctx context.Context, in awsapi.SyncInvocat
 		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.
+	var environment map[string]string
 	if rec.EnvCipher != "" {
-		return awsapi.SyncResult{}, &awsapi.RefusalError{Err: awserrors.Errorf(awserrors.ErrorServiceException,
-			"the test execution path does not support function environment variables")}
+		if e.AEAD == nil {
+			return awsapi.SyncResult{}, &awsapi.RefusalError{Err: awserrors.Errorf(awserrors.ErrorServiceException, "the function environment is unavailable")}
+		}
+		var err error
+		environment, err = function.OpenEnv(e.AEAD, rec.AccountID, rec.FunctionID, rec.EnvCipher)
+		if err != nil || lambdaenv.Validate(environment) != nil {
+			return awsapi.SyncResult{}, &awsapi.RefusalError{Err: awserrors.Errorf(awserrors.ErrorServiceException, "the function environment is unavailable")}
+		}
 	}
 	if rec.Code.SHA256 == "" || rec.Code.Size <= 0 || rec.Code.Size > testlink.MaxCodeBytes {
 		return awsapi.SyncResult{}, &awsapi.RefusalError{Err: fmt.Errorf("code copy of %d bytes is outside the test bound", rec.Code.Size)}
@@ -152,6 +160,7 @@ func (e *testLinkExecutor) InvokeSync(ctx context.Context, in awsapi.SyncInvocat
 		AccessKeyID:     aws.StringValue(session.Credentials.AccessKeyId),
 		SecretAccessKey: aws.StringValue(session.Credentials.SecretAccessKey),
 		SessionToken:    aws.StringValue(session.Credentials.SessionToken),
+		Environment:     environment,
 	}
 	revoke := func() error {
 		rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), revokeTimeout)

spinifex/runtime/roles/awsgw/lambda_testlink_executor_test.go

@@ -4,6 +4,8 @@ package awsgw
 
 import (
 	"context"
+	"crypto/aes"
+	"crypto/cipher"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -165,6 +167,61 @@ func TestLambdaTestLinkExecutor_RefusesWithoutSending(t *testing.T) {
 	}
 }
 
+func TestLambdaTestLinkExecutor_OpensEnvironmentBeforeDispatch(t *testing.T) {
+	block, err := aes.NewCipher(make([]byte, 32))
+	require.NoError(t, err)
+	aead, err := cipher.NewGCM(block)
+	require.NoError(t, err)
+	vars := map[string]string{"GREETING": "private environment value"}
+	sealed, err := function.SealEnv(aead, "123456789012", "fn1", vars)
+	require.NoError(t, err)
+	input := invocation([]byte(`{}`))
+	input.Function.EnvCipher = sealed
+	n := &fakeControl{max: 1 << 20, reply: &testlink.Reply{Outcome: testlink.OutcomeDone}}
+	e := executor(n, nil, grantStore(t))
+	e.AEAD = aead
+	_, err = e.InvokeSync(t.Context(), input)
+	require.NoError(t, err)
+	require.Len(t, n.creds, 1)
+	require.Equal(t, vars, n.creds[0].Environment)
+	require.NotContains(t, string(n.sent[0]), vars["GREETING"])
+}
+
+func TestLambdaTestLinkExecutor_RefusesUnreadableEnvironment(t *testing.T) {
+	block, err := aes.NewCipher(make([]byte, 32))
+	require.NoError(t, err)
+	aead, err := cipher.NewGCM(block)
+	require.NoError(t, err)
+	wrongAccount, err := function.SealEnv(aead, "000000000001", "fn1", map[string]string{"GREETING": "private value"})
+	require.NoError(t, err)
+	reserved, err := function.SealEnv(aead, "123456789012", "fn1", map[string]string{"AWS_REGION": "private value"})
+	require.NoError(t, err)
+	for name, tc := range map[string]struct {
+		sealed string
+		aead   cipher.AEAD
+	}{
+		"key missing":       {wrongAccount, nil},
+		"wrong account":     {wrongAccount, aead},
+		"reserved variable": {reserved, aead},
+		"unreadable":        {"not ciphertext", aead},
+	} {
+		t.Run(name, func(t *testing.T) {
+			input := invocation(nil)
+			input.Function.EnvCipher = tc.sealed
+			n, store, sessions := &fakeControl{max: 1 << 20}, grantStore(t), &fakeSessions{}
+			e := executorWith(n, sessions, nil, store)
+			e.AEAD = tc.aead
+			_, err := e.InvokeSync(t.Context(), input)
+			var refusal *awsapi.RefusalError
+			require.ErrorAs(t, err, &refusal)
+			require.NotContains(t, err.Error(), "private value")
+			require.Empty(t, n.sent)
+			require.Empty(t, sessions.issued)
+			require.Empty(t, grantState(t, store))
+		})
+	}
+}
+
 // A gateway built without a valid image pin refuses before presigning,
 // granting, issuing or connecting.
 func TestLambdaTestLinkExecutor_InvalidImagePinRefusesBeforeConnecting(t *testing.T) {

spinifex/runtime/roles/awsgw/lambda_testlink_off.go

@@ -3,6 +3,7 @@
 package awsgw
 
 import (
+	"crypto/cipher"
 	"github.com/mulgadc/spinifex/spinifex/domains/lambda/awsapi"
 	"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -11,7 +12,7 @@ import (
 
 // Without the test link there is no executor, so Invoke refuses before it
 // presigns, sends or allocates anything, and Lambda stays unrouted.
-func lambdaTestExecutor(*nats.Conn, string, *codestore.Manager, lambdaEnvironmentRoleIssuer) awsapi.Executor {
+func lambdaTestExecutor(*nats.Conn, string, *codestore.Manager, lambdaEnvironmentRoleIssuer, cipher.AEAD) awsapi.Executor {
 	return nil
 }
 

spinifex/runtime/roles/awsgw/lambda_testlink_off_test.go

@@ -40,6 +40,14 @@ func (*spyFunctions) RollbackCreate(context.Context, string, string, string, uin
 	return errors.New("unused")
 }
 
+func (*spyFunctions) BeginCodeUpdate(context.Context, string, string, function.Record, uint64, string) error {
+	return errors.New("unused")
+}
+
+func (*spyFunctions) CodeUpdated(context.Context, string, string, function.Record, uint64, string, string, int64) (function.Record, uint64, error) {
+	return function.Record{}, 0, errors.New("unused")
+}
+
 // spyExec counts invocations that would start an attempt.
 type spyExec struct{ calls atomic.Int32 }
 
@@ -51,7 +59,7 @@ func (e *spyExec) InvokeSync(context.Context, lambdaawsapi.SyncInvocation) (lamb
 // A default build composes the gateway with Lambda unrouted and no executor: a
 // signed Invoke is refused before any Lambda handler runs and starts no attempt.
 func TestDefaultBuild_InvokeNeverReachesLambda(t *testing.T) {
-	require.Nil(t, lambdaTestExecutor(nil, "node1", nil, nil), "a default build has no Lambda executor")
+	require.Nil(t, lambdaTestExecutor(nil, "node1", nil, nil, nil), "a default build has no Lambda executor")
 
 	meta, exec := &spyFunctions{}, &spyExec{}
 	lambda := lambdaawsapi.NewRegistration(lambdaawsapi.Deps{Meta: meta, Exec: exec})

spinifex/runtime/roles/awsgw/lambda_testlink_reply_test.go

@@ -48,6 +48,14 @@ func (storedFunction) RollbackCreate(context.Context, string, string, string, ui
 	return errors.New("unused")
 }
 
+func (storedFunction) BeginCodeUpdate(context.Context, string, string, function.Record, uint64, string) error {
+	return errors.New("unused")
+}
+
+func (storedFunction) CodeUpdated(context.Context, string, string, function.Record, uint64, string, string, int64) (function.Record, uint64, error) {
+	return function.Record{}, 0, errors.New("unused")
+}
+
 // invokeThrough sends one Invoke through the Lambda handler to the real
 // test-link executor; fakeControl stands in for the daemon and answers with rep.
 func invokeThrough(t *testing.T, rep testlink.Reply) (*httptest.ResponseRecorder, error) {

spinifex/runtime/roles/awsgw/lambda_testlink_seam_env_test.go

@@ -5,6 +5,7 @@ package awsgw
 
 import (
 	"context"
+	"crypto/cipher"
 	"crypto/rand"
 	"encoding/hex"
 	"encoding/json"
@@ -17,6 +18,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/iam"
+	"github.com/mulgadc/bluebottle/pkg/masterkey"
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
@@ -51,6 +53,7 @@ type seamEnv struct {
 	roleARN  string
 	grants   *execution.Store
 	sessions jetstream.KeyValue
+	aead     cipher.AEAD
 }
 
 func newSeamEnv(t *testing.T) *seamEnv {
@@ -58,6 +61,8 @@ func newSeamEnv(t *testing.T) *seamEnv {
 	_, nc, _ := testutil.StartTestJetStream(t)
 	masterKey, err := iamdomain.GenerateMasterKey()
 	require.NoError(t, err)
+	lambdaKey, err := masterkey.New(masterKey)
+	require.NoError(t, err)
 	iamSvc, err := iamdomain.NewIAMServiceImpl(t.Context(), nc, masterKey)
 	require.NoError(t, err)
 	stsSvc, err := stsdomain.NewSTSServiceImpl(t.Context(), nc, iamSvc, masterKey, seamNoJWKS{})
@@ -78,7 +83,7 @@ func newSeamEnv(t *testing.T) *seamEnv {
 	require.NoError(t, err)
 	sessions, err := js.KeyValue(t.Context(), stsdomain.KVBucketSessionCredentials)
 	require.NoError(t, err)
-	return &seamEnv{nc: nc, iam: iamSvc, sts: stsSvc, roleARN: aws.StringValue(role.Role.Arn), grants: grants, sessions: sessions}
+	return &seamEnv{nc: nc, iam: iamSvc, sts: stsSvc, roleARN: aws.StringValue(role.Role.Arn), grants: grants, sessions: sessions, aead: lambdaKey.AEAD}
 }
 
 // resolve authenticates cred through the real verifier and resolves its

spinifex/runtime/roles/awsgw/lambda_testlink_seam_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/testlink"
+	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/require"
 )
 
@@ -72,6 +73,14 @@ func (seamFunction) RollbackCreate(context.Context, string, string, string, uint
 	return errors.New("unused")
 }
 
+func (seamFunction) BeginCodeUpdate(context.Context, string, string, function.Record, uint64, string) error {
+	return errors.New("unused")
+}
+
+func (seamFunction) CodeUpdated(context.Context, string, string, function.Record, uint64, string, string, int64) (function.Record, uint64, error) {
+	return function.Record{}, 0, errors.New("unused")
+}
+
 type urlPresigner struct {
 	url   string
 	calls *atomic.Int32
@@ -212,7 +221,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: testlink.MaxFrameBytes, Sessions: env.sts, Node: "node1", ImageDigest: testImageDigest,
+	exec := &testLinkExecutor{Control: daemon.client, RequestLimit: testlink.MaxFrameBytes, Sessions: env.sts, AEAD: env.aead, Node: "node1", ImageDigest: testImageDigest,
 		Codes:  func(context.Context) (codePresigner, error) { return urlPresigner{codeURL, presigns}, nil },
 		Grants: func(context.Context) (grantIssuer, error) { return grants, nil },
 	}
@@ -375,16 +384,42 @@ func TestLambdaTestLinkSeam_RuntimeSeesInvokedARNAndHandler(t *testing.T) {
 	}
 }
 
-// A function with environment variables is refused before any code link is
-// presigned, any grant is issued or anything is sent to a node.
-func TestLambdaTestLinkSeam_EnvironmentVariablesRefusedBeforeDispatch(t *testing.T) {
+func TestLambdaTestLinkSeam_EnvironmentReachesGuestWithoutLeaking(t *testing.T) {
+	env := newSeamEnv(t)
+	var subscriptions []*nats.Subscription
+	for _, subject := range []string{">", "$JS.>", "$KV.>"} {
+		sub, err := env.nc.SubscribeSync(subject)
+		require.NoError(t, err)
+		t.Cleanup(func() { _ = sub.Unsubscribe() })
+		subscriptions = append(subscriptions, sub)
+	}
+	require.NoError(t, env.nc.Flush())
+	vars := map[string]string{"GREETING": "private-environment-marker"}
+	sealed, err := function.SealEnv(env.aead, seamAccount, "fn1", vars)
+	require.NoError(t, err)
 	srv := codeServer(t)
-	run := invokeComposedAt(t, "/2015-03-31/functions/f/invocations", seamFunction{timeout: 10, envCipher: "sealed"},
-		srv.URL+"/code", srv.Client(), func(ctx context.Context, _ net.Conn) { <-ctx.Done() })
-	require.Equal(t, http.StatusInternalServerError, run.resp.StatusCode)
-	require.Equal(t, "ServiceException", run.resp.Header.Get("X-Amzn-Errortype"))
-	require.Contains(t, string(run.body), "does not support function environment variables")
-	require.Zero(t, run.presigns, "a code link was presigned")
-	require.Zero(t, run.grants, "a grant was issued")
-	require.Empty(t, run.requests, "the request reached a node")
+	creds := make(chan hostlink.Credential, 1)
+	run := invokeComposedIn(t, env, "/2015-03-31/functions/f/invocations", seamFunction{timeout: 10, envCipher: sealed},
+		srv.URL+"/code", srv.Client(), agentGuestWith(guestOpts{creds: creds}, func(base string) { respond(base, `{}`) }))
+	require.Equal(t, http.StatusOK, run.resp.StatusCode, string(run.body))
+	require.Equal(t, vars, recv(t, creds, "guest credential").Environment)
+	for _, request := range run.requests {
+		require.NotContains(t, string(request), vars["GREETING"])
+	}
+	for _, reply := range run.replies {
+		require.NotContains(t, string(reply), vars["GREETING"])
+	}
+	require.NoError(t, env.nc.Flush())
+	for _, sub := range subscriptions {
+		for {
+			msg, err := sub.NextMsg(50 * time.Millisecond)
+			if errors.Is(err, nats.ErrTimeout) {
+				break
+			}
+			require.NoError(t, err)
+			require.NotContains(t, msg.Subject, vars["GREETING"])
+			require.NotContains(t, string(msg.Data), vars["GREETING"])
+		}
+	}
+	require.NotContains(t, run.log, vars["GREETING"])
 }