From b9394b17dc3f9d005db1f3835367f86fb2489f16 Mon Sep 17 00:00:00 2001 From: Joel Scheuner Date: Wed, 3 Jun 2026 18:14:08 +0200 Subject: [PATCH 1/2] feat(init): add runtime exit error reporting via supervisor and events API Ports the supervisor and events API from PR #41 to enable proper error reporting when a Lambda runtime process exits unexpectedly (e.g. sys.exit() or missing wrapper script), instead of LocalStack timing out with a generic error. - Add LocalStackSupervisor: wraps ProcessSupervisor, detects unexpected runtime-* process exits and emits SendFault(RuntimeExit) events - Add LocalStackEventsAPI: wraps StandaloneEventsAPI, overrides SendFault to forward errors to LocalStack via SendStatus(error, ...) - Wire both into SandboxBuilder via SetEventsAPI / SetSupervisor - Refactor NewCustomInteropServer to accept a pre-created *LocalStackAdapter shared with the events API - Improve SendInitErrorResponse: properly deserialises the payload, includes RequestId, and sends asynchronously (non-blocking) Enables test_lambda_runtime_exit and test_lambda_runtime_wrapper_not_found. Co-Authored-By: Claude Sonnet 4.6 --- cmd/localstack/custom_interop.go | 51 ++++++++++--- cmd/localstack/events.go | 55 ++++++++++++++ cmd/localstack/main.go | 22 +++++- cmd/localstack/supervisor.go | 124 +++++++++++++++++++++++++++++++ 4 files changed, 239 insertions(+), 13 deletions(-) create mode 100644 cmd/localstack/events.go create mode 100644 cmd/localstack/supervisor.go diff --git a/cmd/localstack/custom_interop.go b/cmd/localstack/custom_interop.go index 6b89d65..ca02d25 100644 --- a/cmd/localstack/custom_interop.go +++ b/cmd/localstack/custom_interop.go @@ -97,15 +97,12 @@ type ErrorResponse struct { StackTrace []string `json:"stackTrace,omitempty"` } -func NewCustomInteropServer(lsOpts *LsOpts, delegate interop.Server, logCollector *LogCollector) (server *CustomInteropServer) { +func NewCustomInteropServer(lsOpts *LsOpts, adapter *LocalStackAdapter, delegate interop.Server, logCollector *LogCollector) (server *CustomInteropServer) { server = &CustomInteropServer{ - delegate: delegate.(*rapidcore.Server), - port: lsOpts.InteropPort, - upstreamEndpoint: lsOpts.RuntimeEndpoint, - localStackAdapter: &LocalStackAdapter{ - UpstreamEndpoint: lsOpts.RuntimeEndpoint, - RuntimeId: lsOpts.RuntimeId, - }, + delegate: delegate.(*rapidcore.Server), + port: lsOpts.InteropPort, + upstreamEndpoint: lsOpts.RuntimeEndpoint, + localStackAdapter: adapter, } // TODO: extract this @@ -219,12 +216,44 @@ func (c *CustomInteropServer) SendErrorResponse(invokeID string, resp *interop.E return c.delegate.SendErrorResponse(invokeID, resp) } -// SendInitErrorResponse writes error response during init to a shared memory and sends GIRD FAULT. +// SendInitErrorResponse forwards the init error to LocalStack and then propagates it to the delegate. func (c *CustomInteropServer) SendInitErrorResponse(resp *interop.ErrorInvokeResponse) error { log.Traceln("SendInitErrorResponse called") - if err := c.localStackAdapter.SendStatus(Error, resp.Payload); err != nil { - log.Fatalln("Failed to send init error to LocalStack " + err.Error() + ". Exiting.") + + // Deserialize the raw payload so we can include the requestId and structured fields. + var parsed struct { + ErrorMessage string `json:"errorMessage"` + ErrorType string `json:"errorType"` + StackTrace []string `json:"stackTrace,omitempty"` + } + if err := json.Unmarshal(resp.Payload, &parsed); err != nil { + log.WithError(err).Warn("Failed to parse init error payload; forwarding raw payload") + if err := c.localStackAdapter.SendStatus(Error, resp.Payload); err != nil { + log.WithError(err).WithField("runtime-id", c.localStackAdapter.RuntimeId). + Error("Failed to send init error to LocalStack") + } + return c.delegate.SendInitErrorResponse(resp) + } + + adaptedResp := ErrorResponse{ + ErrorMessage: parsed.ErrorMessage, + ErrorType: parsed.ErrorType, + RequestId: c.delegate.GetCurrentInvokeID(), + StackTrace: parsed.StackTrace, + } + body, err := json.Marshal(adaptedResp) + if err != nil { + log.WithError(err).Error("Failed to marshal adapted init error response") + body = resp.Payload } + + go func() { + if err := c.localStackAdapter.SendStatus(Error, body); err != nil { + log.WithError(err).WithField("runtime-id", c.localStackAdapter.RuntimeId). + Error("Failed to send init error to LocalStack") + } + }() + return c.delegate.SendInitErrorResponse(resp) } diff --git a/cmd/localstack/events.go b/cmd/localstack/events.go new file mode 100644 index 0000000..bf6fa00 --- /dev/null +++ b/cmd/localstack/events.go @@ -0,0 +1,55 @@ +package main + +import ( + "encoding/json" + "fmt" + "sync" + + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/interop" + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry" +) + +// LocalStackEventsAPI intercepts fault events and forwards them to LocalStack as error status callbacks. +type LocalStackEventsAPI struct { + *telemetry.StandaloneEventsAPI + adapter *LocalStackAdapter + requestID string + mu sync.RWMutex +} + +func NewLocalStackEventsAPI(adapter *LocalStackAdapter) *LocalStackEventsAPI { + return &LocalStackEventsAPI{ + adapter: adapter, + StandaloneEventsAPI: new(telemetry.StandaloneEventsAPI), + } +} + +func (ev *LocalStackEventsAPI) SendFault(data interop.FaultData) error { + _ = ev.StandaloneEventsAPI.SendFault(data) + + requestID := string(data.RequestID) + if data.RequestID == "" { + ev.mu.RLock() + requestID = ev.requestID + ev.mu.RUnlock() + } + + resp := ErrorResponse{ + ErrorMessage: fmt.Sprintf("RequestId: %s Error: %s", requestID, data.ErrorMessage), + ErrorType: string(data.ErrorType), + } + + payload, err := json.Marshal(resp) + if err != nil { + return err + } + + return ev.adapter.SendStatus(Error, payload) +} + +func (ev *LocalStackEventsAPI) SetCurrentRequestID(id interop.RequestID) { + ev.mu.Lock() + defer ev.mu.Unlock() + ev.requestID = string(id) + ev.StandaloneEventsAPI.SetCurrentRequestID(id) +} diff --git a/cmd/localstack/main.go b/cmd/localstack/main.go index b03d877..23abadc 100644 --- a/cmd/localstack/main.go +++ b/cmd/localstack/main.go @@ -179,6 +179,20 @@ func main() { localStackLogsEgressApi := NewLocalStackLogsEgressAPI(logCollector) tracer := NewLocalStackTracer() + // Create LocalStack adapter upfront so it can be shared with the events API and interop server + lsAdapter := &LocalStackAdapter{ + UpstreamEndpoint: lsOpts.RuntimeEndpoint, + RuntimeId: lsOpts.RuntimeId, + } + + // Events API forwards runtime fault events (unexpected exits) to LocalStack as error callbacks + lsEventsAPI := NewLocalStackEventsAPI(lsAdapter) + + // Supervisor intercepts runtime process terminations and emits fault events via the events API + supervisorCtx, cancelSupervisor := context.WithCancel(context.Background()) + + localStackSupv := NewLocalStackSupervisor(supervisorCtx, lsEventsAPI) + // build sandbox sandbox := rapidcore. NewSandboxBuilder(). @@ -186,11 +200,15 @@ func main() { AddShutdownFunc(func() { log.Debugln("Stopping file watcher") cancelFileWatcher() + log.Debugln("Stopping supervisor") + cancelSupervisor() }). SetExtensionsFlag(true). SetInitCachingFlag(true). SetLogsEgressAPI(localStackLogsEgressApi). - SetTracer(tracer) + SetTracer(tracer). + SetEventsAPI(lsEventsAPI). + SetSupervisor(localStackSupv) // Corresponds to the 'AWS_LAMBDA_RUNTIME_API' environment variable. // We need to ensure the runtime server is up before the INIT phase, @@ -211,7 +229,7 @@ func main() { runDaemon(d) // async defaultInterop := sandbox.DefaultInteropServer() - interopServer := NewCustomInteropServer(lsOpts, defaultInterop, logCollector) + interopServer := NewCustomInteropServer(lsOpts, lsAdapter, defaultInterop, logCollector) sandbox.SetInteropServer(interopServer) if len(handler) > 0 { sandbox.SetHandler(handler) diff --git a/cmd/localstack/supervisor.go b/cmd/localstack/supervisor.go new file mode 100644 index 0000000..161a940 --- /dev/null +++ b/cmd/localstack/supervisor.go @@ -0,0 +1,124 @@ +package main + +import ( + "context" + "fmt" + "strings" + "sync/atomic" + + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/fatalerror" + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/interop" + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/supervisor" + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/supervisor/model" + "github.com/google/uuid" + log "github.com/sirupsen/logrus" +) + +// LocalStackSupervisor wraps a ProcessSupervisor and intercepts runtime process termination events. +// When a runtime process exits unexpectedly it sends a fault event via the EventsAPI so LocalStack +// receives a proper error instead of timing out. +type LocalStackSupervisor struct { + model.ProcessSupervisor + eventsChan chan model.Event + eventsAPI interop.EventsAPI + + isShuttingDown *atomic.Bool +} + +func NewLocalStackSupervisor(ctx context.Context, evs interop.EventsAPI) *LocalStackSupervisor { + var isShuttingDown atomic.Bool + ls := &LocalStackSupervisor{ + ProcessSupervisor: supervisor.NewLocalSupervisor(), + eventsAPI: evs, + eventsChan: make(chan model.Event), + isShuttingDown: &isShuttingDown, + } + + go ls.loop(ctx) + + return ls +} + +func (ls *LocalStackSupervisor) loop(ctx context.Context) { + inCh, err := ls.ProcessSupervisor.Events(ctx, nil) + if err != nil { + panic(err) + } + defer close(ls.eventsChan) + for { + select { + case event, ok := <-inCh: + if !ok { + return + } + + select { + case ls.eventsChan <- event: + case <-ctx.Done(): + return + } + + if ls.isShuttingDown.Load() { + continue + } + + termination := event.Event.ProcessTerminated() + if termination == nil { + continue + } + + if !strings.Contains(*termination.Name, "runtime-") { + log.Debugf("Ignoring non-runtime process termination: %s", *termination.Name) + continue + } + + if termination.Signaled() != nil { + log.Debugf("Runtime process signalled: %d", *termination.Signo) + } + + faultData := interop.FaultData{ + RequestID: interop.RequestID(uuid.NewString()), + ErrorMessage: fmt.Errorf("Runtime exited without providing a reason"), + ErrorType: fatalerror.RuntimeExit, + } + if !termination.Success() { + faultData.ErrorMessage = fmt.Errorf("Runtime exited with error: %s", termination.String()) + } + + if err := ls.eventsAPI.SendFault(faultData); err != nil { + log.WithError(err).Error("Failed to send runtime fault event") + } + case <-ctx.Done(): + return + } + } +} + +func (ls *LocalStackSupervisor) Exec(ctx context.Context, request *model.ExecRequest) error { + if request.Domain == "runtime" { + ls.isShuttingDown.Store(false) + } + return ls.ProcessSupervisor.Exec(ctx, request) +} + +func (ls *LocalStackSupervisor) Terminate(ctx context.Context, request *model.TerminateRequest) error { + defer func() { + if request.Domain == "runtime" && strings.HasPrefix(request.Name, "runtime-") { + ls.isShuttingDown.Store(true) + } + }() + return ls.ProcessSupervisor.Terminate(ctx, request) +} + +func (ls *LocalStackSupervisor) Kill(ctx context.Context, request *model.KillRequest) error { + defer func() { + if request.Domain == "runtime" && strings.HasPrefix(request.Name, "runtime-") { + ls.isShuttingDown.Store(true) + } + }() + return ls.ProcessSupervisor.Kill(ctx, request) +} + +func (ls *LocalStackSupervisor) Events(ctx context.Context, _ *model.EventsRequest) (<-chan model.Event, error) { + return ls.eventsChan, nil +} From a08c667d935129771c795e39282961f33440eec2 Mon Sep 17 00:00:00 2001 From: Joel Scheuner Date: Wed, 3 Jun 2026 19:45:50 +0200 Subject: [PATCH 2/2] fix(init): include requestId in init error response even when empty MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Use *string for the RequestId field in ErrorResponse so that an empty string is serialized (not omitted by omitempty), while nil — used for fault events — stays omitted. Fixes test_lambda_runtime_error snapshot mismatch where requestId: "" was expected but absent. Co-Authored-By: Claude Sonnet 4.6 --- cmd/localstack/custom_interop.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/cmd/localstack/custom_interop.go b/cmd/localstack/custom_interop.go index ca02d25..d46ff91 100644 --- a/cmd/localstack/custom_interop.go +++ b/cmd/localstack/custom_interop.go @@ -93,7 +93,9 @@ type InvokeRequest struct { type ErrorResponse struct { ErrorMessage string `json:"errorMessage"` ErrorType string `json:"errorType,omitempty"` - RequestId string `json:"requestId,omitempty"` + // RequestId uses *string so that an empty string "" is serialized (not omitted), + // while nil is omitted — init errors always set this field, fault events leave it nil. + RequestId *string `json:"requestId,omitempty"` StackTrace []string `json:"stackTrace,omitempty"` } @@ -235,10 +237,11 @@ func (c *CustomInteropServer) SendInitErrorResponse(resp *interop.ErrorInvokeRes return c.delegate.SendInitErrorResponse(resp) } + requestId := c.delegate.GetCurrentInvokeID() adaptedResp := ErrorResponse{ ErrorMessage: parsed.ErrorMessage, ErrorType: parsed.ErrorType, - RequestId: c.delegate.GetCurrentInvokeID(), + RequestId: &requestId, StackTrace: parsed.StackTrace, } body, err := json.Marshal(adaptedResp)