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.

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)
 	}
 }
 
@@ -789,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

@@ -53,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) {
@@ -210,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 == "" {
@@ -254,6 +261,69 @@ 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

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/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/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/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 }
 

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_test.go

@@ -73,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