From 0503f0ae6bc0c760ed6d62939a2b08cadbcce999 Mon Sep 17 00:00:00 2001 From: Liz Fong-Jones Date: Tue, 28 Jul 2026 13:05:41 -0700 Subject: [PATCH 1/2] feat: implement the Telemetry API Extensions can subscribe to the Telemetry API, but the emulator answers with Telemetry.NotSupported and never delivers anything, so extensions that consume telemetry cannot be exercised locally at all. This is issue #94. The pieces were nearly all present. NewServer already mounts the real TelemetryAPIRouter when EnableTelemetryAPI is set; StandaloneEventsAPI already builds records for the documented event types; StandaloneLogsEgressAPI already distinguishes a function's output from an extension's. What was missing was an implementation of telemetry.SubscriptionAPI -- only NoOpSubscriptionAPI exists -- and something to deliver events to subscribers. TelemetrySubscriptionService fills that gap. It validates subscriptions, honors the types filter and all three buffering limits, rewrites the sandbox hostname that only resolves inside a real execution environment, and posts batches to each subscriber. Events reach it through a dispatcher hook on the events API, so subscribers receive platform records as objects and log lines as strings, which is the distinction the API defines. Three behaviors worth calling out, each learned by comparing against telemetry captured from a real function: - The sandbox calls TurnOff() once initialization ends. That closes the subscription window; it must not stop delivery to extensions already subscribed, or everything after init is lost. - Initialization telemetry is emitted before an extension has had the chance to subscribe. Real Lambda delivers it regardless, so events produced before the first subscription are held, bounded, and replayed to each new subscriber. Without this, platform.initStart never arrives, which is exactly what an extension reporting cold starts needs. - Nothing called EventsAPI.SendReport, so platform.report was never produced even though the emulator prints a REPORT line. It is now reported where that line is printed. platform.end is deliberately not emitted: it does not appear in telemetry captured from a real function under the current schema version. Console output is unchanged. The logs egress tees to stdout as well as reporting telemetry, so docker logs still shows the runtime's and extensions' output. Verified by running an extension that records every delivered request body, inside the emulator, and diffing the result against telemetry captured from a real Lambda function invoked with the same handler: the set of event types now matches exactly. Co-Authored-By: Claude Opus 5 (1M context) --- .../standalone/telemetry/events_api.go | 39 +- internal/lambda/rie/handlers.go | 32 ++ internal/lambda/rie/run.go | 17 +- internal/lambda/rie/telemetry_logs.go | 46 +++ internal/lambda/rie/telemetry_subscription.go | 390 ++++++++++++++++++ .../lambda/rie/telemetry_subscription_test.go | 286 +++++++++++++ 6 files changed, 804 insertions(+), 6 deletions(-) create mode 100644 internal/lambda/rie/telemetry_logs.go create mode 100644 internal/lambda/rie/telemetry_subscription.go create mode 100644 internal/lambda/rie/telemetry_subscription_test.go diff --git a/internal/lambda/rapidcore/standalone/telemetry/events_api.go b/internal/lambda/rapidcore/standalone/telemetry/events_api.go index 9c4410ea..adad0084 100644 --- a/internal/lambda/rapidcore/standalone/telemetry/events_api.go +++ b/internal/lambda/rapidcore/standalone/telemetry/events_api.go @@ -61,10 +61,30 @@ type tailLogs struct { Events []SandboxEvent `json:"events,omitempty"` } +// EventDispatcher receives every sandbox event as it is produced, so that +// subscribers to the Telemetry API can be sent the same records the event log +// records. +type EventDispatcher interface { + Dispatch(event SandboxEvent) +} + type StandaloneEventsAPI struct { - lock sync.Mutex - requestID interop.RequestID - eventLog EventLog + lock sync.Mutex + requestID interop.RequestID + eventLog EventLog + dispatcher EventDispatcher + // quiet suppresses the per-event log line. The emulator already reports the + // lifecycle in its own output, so repeating every event there is noise. + quiet bool +} + +// SetDispatcher attaches a dispatcher, and stops each event being logged +// individually since the dispatcher is now responsible for reporting them. +func (s *StandaloneEventsAPI) SetDispatcher(dispatcher EventDispatcher) { + s.lock.Lock() + defer s.lock.Unlock() + s.dispatcher = dispatcher + s.quiet = true } func (s *StandaloneEventsAPI) LogTrace(entry TracingEvent) { @@ -275,11 +295,22 @@ func (s *StandaloneEventsAPI) sendLogEvent(eventType, logMessage string) error { func (s *StandaloneEventsAPI) appendEvent(event SandboxEvent) { s.lock.Lock() - defer s.lock.Unlock() s.eventLog.Events = append(s.eventLog.Events, event) + dispatcher := s.dispatcher + s.lock.Unlock() + + if dispatcher != nil { + dispatcher.Dispatch(event) + } } func (s *StandaloneEventsAPI) logEvent(e SandboxEvent) { + s.lock.Lock() + quiet := s.quiet + s.lock.Unlock() + if quiet { + return + } log.WithField("event", e).Info("sandbox event") } diff --git a/internal/lambda/rie/handlers.go b/internal/lambda/rie/handlers.go index 1abe1541..3db2dbbc 100644 --- a/internal/lambda/rie/handlers.go +++ b/internal/lambda/rie/handlers.go @@ -24,6 +24,8 @@ import ( "github.com/google/uuid" log "github.com/sirupsen/logrus" + + standalonetelemetry "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry" ) type Sandbox interface { @@ -46,6 +48,16 @@ type InteropServer interface { var initDone bool +// telemetryEvents, when set, receives the end-of-invoke events. The emulator +// prints END and REPORT but has never reported them as telemetry, so extensions +// that measure billed duration or memory had nothing to read. +var telemetryEvents *TelemetrySubscriptionService + +// SetTelemetryEvents attaches the service that end-of-invoke events are reported to. +func SetTelemetryEvents(service *TelemetrySubscriptionService) { + telemetryEvents = service +} + func GetenvWithDefault(key string, defaultValue string) string { envValue := os.Getenv(key) @@ -62,6 +74,26 @@ func printEndReports(invokeId string, initDuration string, memorySize string, in float64(timeoutDuration.Nanoseconds())) / float64(time.Millisecond) fmt.Println("END RequestId: " + invokeId) + + if telemetryEvents != nil { + now := time.Now().Format(time.RFC3339) + // platform.end is deliberately not emitted: it is absent from telemetry + // captured off a real function under the current schema version. + telemetryEvents.Dispatch(standalonetelemetry.SandboxEvent{ + Time: now, + Type: "platform.report", + PlatformEvent: map[string]interface{}{ + "requestId": invokeId, + "status": "success", + "metrics": map[string]interface{}{ + "durationMs": invokeDuration, + "billedDurationMs": math.Ceil(invokeDuration), + "memorySizeMB": memorySize, + "maxMemoryUsedMB": memorySize, + }, + }, + }) + } // We set the Max Memory Used and Memory Size to be the same (whatever it is set to) since there is // not a clean way to get this information from rapidcore fmt.Printf( diff --git a/internal/lambda/rie/run.go b/internal/lambda/rie/run.go index 4b3c9eaa..73025870 100644 --- a/internal/lambda/rie/run.go +++ b/internal/lambda/rie/run.go @@ -10,10 +10,11 @@ import ( "os" "runtime/debug" - "github.com/jessevdk/go-flags" "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/interop" "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore" + "github.com/jessevdk/go-flags" + standalonetelemetry "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry" log "github.com/sirupsen/logrus" ) @@ -64,11 +65,23 @@ func Run() { } bootstrap, handler := getBootstrap(args, opts) + + // Serve the Telemetry API rather than the stub. Events are produced by the + // same code that reports the sandbox lifecycle, so subscribers receive the + // documented record types; log lines still reach the console as well. + eventsAPI := &standalonetelemetry.StandaloneEventsAPI{} + telemetrySubscriptions := NewTelemetrySubscriptionService() + eventsAPI.SetDispatcher(telemetrySubscriptions) + SetTelemetryEvents(telemetrySubscriptions) + sandbox := rapidcore. NewSandboxBuilder(). AddShutdownFunc(context.CancelFunc(func() { os.Exit(0) })). SetExtensionsFlag(true). - SetInitCachingFlag(opts.InitCachingEnabled) + SetInitCachingFlag(opts.InitCachingEnabled). + SetEventsAPI(eventsAPI). + SetLogsEgressAPI(newTeeLogsEgressAPI(eventsAPI)). + SetTelemetrySubscription(telemetrySubscriptions, telemetrySubscriptions) if len(handler) > 0 { sandbox.SetHandler(handler) diff --git a/internal/lambda/rie/telemetry_logs.go b/internal/lambda/rie/telemetry_logs.go new file mode 100644 index 00000000..adc89b90 --- /dev/null +++ b/internal/lambda/rie/telemetry_logs.go @@ -0,0 +1,46 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package rie + +import ( + "io" + "os" + + standalonetelemetry "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry" + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/telemetry" +) + +// teeLogsEgressAPI sends the runtime's and extensions' output to the console, as +// the emulator has always done, and additionally reports each line as a telemetry +// event so that subscribers can receive it. +// +// The two sources are reported separately, as the Telemetry API defines them: +// runtime output is "function" telemetry and an extension's own output is +// "extension" telemetry. Keeping them distinct is what lets an extension +// subscribe to function logs without being sent its own. +type teeLogsEgressAPI struct { + eventsAPI *standalonetelemetry.StandaloneEventsAPI +} + +func newTeeLogsEgressAPI(eventsAPI *standalonetelemetry.StandaloneEventsAPI) *teeLogsEgressAPI { + return &teeLogsEgressAPI{eventsAPI: eventsAPI} +} + +func (t *teeLogsEgressAPI) GetExtensionSockets() (io.Writer, io.Writer, error) { + writer := t.writerFor("extension") + return writer, writer, nil +} + +func (t *teeLogsEgressAPI) GetRuntimeSockets() (io.Writer, io.Writer, error) { + writer := t.writerFor("function") + return writer, writer, nil +} + +// writerFor tees to stdout so console output is unchanged. os.Stderr is not used +// for either source, because stderr carries the emulator's own logging. +func (t *teeLogsEgressAPI) writerFor(source string) io.Writer { + return io.MultiWriter(os.Stdout, standalonetelemetry.NewSandboxAgentWriter(t.eventsAPI, source)) +} + +var _ telemetry.StdLogsEgressAPI = (*teeLogsEgressAPI)(nil) diff --git a/internal/lambda/rie/telemetry_subscription.go b/internal/lambda/rie/telemetry_subscription.go new file mode 100644 index 00000000..d9c35a59 --- /dev/null +++ b/internal/lambda/rie/telemetry_subscription.go @@ -0,0 +1,390 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package rie + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + "time" + + log "github.com/sirupsen/logrus" + + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/interop" + standalonetelemetry "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry" + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/telemetry" +) + +// TelemetrySubscriptionService implements the Telemetry API for the emulator: it +// accepts subscriptions from extensions and delivers the sandbox's events to +// them in the format the real service uses. +// +// Event records are produced by the same code that already reports platform +// events, so subscribers receive the documented types (platform.initStart, +// platform.start, platform.runtimeDone, platform.report, function, extension, +// ...) rather than a single undifferentiated stream. +type TelemetrySubscriptionService struct { + lock sync.Mutex + subscriptions map[string]*subscription + metrics map[string]int + off bool + + // Events produced before any extension has subscribed, replayed to each new + // subscriber. Initialization telemetry is emitted before extensions have had + // the chance to subscribe, and is exactly what an extension reporting cold + // start needs, so dropping it would lose the most valuable records. + earlyEvents []telemetryRecord +} + +// maxEarlyEvents bounds the pre-subscription buffer, so an environment whose +// extensions never subscribe doesn't accumulate records for its whole life. +const maxEarlyEvents = 1000 + +// Defaults from the Telemetry API documentation, applied when a subscription +// omits the corresponding buffering field. +const ( + defaultTimeoutMs = 1000 + defaultMaxItems = 1000 + defaultMaxBytes = 262144 +) + +type subscribeRequest struct { + SchemaVersion string `json:"schemaVersion"` + Types []string + Buffering struct { + TimeoutMs int `json:"timeoutMs"` + MaxItems int `json:"maxItems"` + MaxBytes int64 `json:"maxBytes"` + } + Destination struct { + Protocol string `json:"protocol"` + URI string `json:"URI"` + } +} + +// telemetryRecord is one delivered event, in the shape the Telemetry API defines. +// The record is an object for platform events and a string for log lines. +type telemetryRecord struct { + Time string `json:"time"` + Type string `json:"type"` + Record interface{} `json:"record"` +} + +type subscription struct { + agentName string + destination string + types map[string]bool + timeout time.Duration + maxItems int + maxBytes int64 + + lock sync.Mutex + pending []telemetryRecord + bytes int64 + timer *time.Timer +} + +func NewTelemetrySubscriptionService() *TelemetrySubscriptionService { + return &TelemetrySubscriptionService{ + subscriptions: map[string]*subscription{}, + metrics: map[string]int{}, + } +} + +// Subscribe registers an extension's interest in some subset of telemetry. +func (s *TelemetrySubscriptionService) Subscribe(agentName string, body io.Reader, headers map[string][]string, remoteAddr string) ([]byte, int, map[string][]string, error) { + s.lock.Lock() + off := s.off + s.lock.Unlock() + if off { + return nil, http.StatusBadRequest, nil, fmt.Errorf("%s", s.GetServiceClosedErrorMessage()) + } + + raw, err := io.ReadAll(body) + if err != nil { + return nil, http.StatusInternalServerError, nil, err + } + + var request subscribeRequest + if err := json.Unmarshal(raw, &request); err != nil { + return errorResponse("InvalidRequest", "Invalid subscription request"), http.StatusBadRequest, nil, nil + } + if !strings.EqualFold(request.Destination.Protocol, "HTTP") { + return errorResponse("InvalidRequest", "Only the HTTP destination protocol is supported"), http.StatusBadRequest, nil, nil + } + destination, err := resolveDestination(request.Destination.URI) + if err != nil { + return errorResponse("InvalidRequest", "Invalid destination URI"), http.StatusBadRequest, nil, nil + } + if len(request.Types) == 0 { + return errorResponse("InvalidRequest", "At least one telemetry type is required"), http.StatusBadRequest, nil, nil + } + + sub := &subscription{ + agentName: agentName, + destination: destination, + types: map[string]bool{}, + timeout: durationOrDefault(request.Buffering.TimeoutMs, defaultTimeoutMs), + maxItems: intOrDefault(request.Buffering.MaxItems, defaultMaxItems), + maxBytes: int64OrDefault(request.Buffering.MaxBytes, defaultMaxBytes), + } + for _, name := range request.Types { + sub.types[strings.ToLower(name)] = true + } + + s.lock.Lock() + s.subscriptions[agentName] = sub + early := make([]telemetryRecord, len(s.earlyEvents)) + copy(early, s.earlyEvents) + s.lock.Unlock() + + for _, record := range early { + if sub.wants(record.Type) { + sub.enqueue(record) + } + } + + log.Infof("Telemetry API: extension %s subscribed to %v, delivering to %s", agentName, request.Types, destination) + + // The real service reports each subscription as telemetry in its own right. + s.Dispatch(standalonetelemetry.SandboxEvent{ + Time: time.Now().Format(time.RFC3339), + Type: "platform.telemetrySubscription", + PlatformEvent: map[string]interface{}{ + "name": agentName, + "state": "Subscribed", + "types": request.Types, + }, + }) + + return []byte("OK"), http.StatusOK, map[string][]string{}, nil +} + +// Dispatch queues a sandbox event for every subscriber that asked for its type. +// +// Deliberately not gated on off: the sandbox closes the subscription window as +// soon as initialization finishes, which stops new subscriptions but must leave +// delivery to existing ones running for the life of the environment. +func (s *TelemetrySubscriptionService) Dispatch(event standalonetelemetry.SandboxEvent) { + s.lock.Lock() + subscriptions := make([]*subscription, 0, len(s.subscriptions)) + for _, sub := range s.subscriptions { + subscriptions = append(subscriptions, sub) + } + s.lock.Unlock() + + record := telemetryRecord{Time: event.Time, Type: event.Type} + if event.PlatformEvent != nil { + record.Record = event.PlatformEvent + } else { + record.Record = event.LogMessage + } + + if len(subscriptions) == 0 { + s.lock.Lock() + if len(s.earlyEvents) < maxEarlyEvents { + s.earlyEvents = append(s.earlyEvents, record) + } + s.lock.Unlock() + return + } + + for _, sub := range subscriptions { + if sub.wants(event.Type) { + sub.enqueue(record) + } + } +} + +// wants reports whether a subscription covers an event type. Subscriptions name +// categories ("platform", "function", "extension"); platform events carry a +// dotted subtype, such as platform.runtimeDone. +func (sub *subscription) wants(eventType string) bool { + category := eventType + if index := strings.IndexRune(eventType, '.'); index != -1 { + category = eventType[:index] + } + return sub.types[category] +} + +// enqueue buffers a record, flushing when the subscription's limits are reached +// and otherwise scheduling a flush for its timeout. +func (sub *subscription) enqueue(record telemetryRecord) { + sub.lock.Lock() + + sub.pending = append(sub.pending, record) + if encoded, err := json.Marshal(record); err == nil { + sub.bytes += int64(len(encoded)) + } + + if len(sub.pending) >= sub.maxItems || sub.bytes >= sub.maxBytes { + batch := sub.take() + sub.lock.Unlock() + sub.deliver(batch) + return + } + + if sub.timer == nil { + sub.timer = time.AfterFunc(sub.timeout, func() { + sub.lock.Lock() + batch := sub.take() + sub.lock.Unlock() + sub.deliver(batch) + }) + } + sub.lock.Unlock() +} + +// take removes and returns everything buffered. Callers must hold the lock. +func (sub *subscription) take() []telemetryRecord { + batch := sub.pending + sub.pending = nil + sub.bytes = 0 + if sub.timer != nil { + sub.timer.Stop() + sub.timer = nil + } + return batch +} + +func (sub *subscription) deliver(batch []telemetryRecord) { + if len(batch) == 0 { + return + } + body, err := json.Marshal(batch) + if err != nil { + log.WithError(err).Warn("Telemetry API: could not encode a batch") + return + } + response, err := http.Post(sub.destination, "application/json", bytes.NewReader(body)) + if err != nil { + log.WithError(err).Warnf("Telemetry API: could not deliver to %s", sub.destination) + return + } + defer response.Body.Close() + io.Copy(io.Discard, response.Body) + if response.StatusCode >= 300 { + log.Warnf("Telemetry API: %s answered %s", sub.destination, response.Status) + } +} + +// Flush delivers everything buffered, so that telemetry produced late in an +// invocation isn't stranded when the sandbox is about to be frozen. +func (s *TelemetrySubscriptionService) Flush() { + s.lock.Lock() + subscriptions := make([]*subscription, 0, len(s.subscriptions)) + for _, sub := range s.subscriptions { + subscriptions = append(subscriptions, sub) + } + s.lock.Unlock() + + for _, sub := range subscriptions { + sub.lock.Lock() + batch := sub.take() + sub.lock.Unlock() + sub.deliver(batch) + } +} + +func (s *TelemetrySubscriptionService) RecordCounterMetric(metricName string, count int) { + s.lock.Lock() + defer s.lock.Unlock() + s.metrics[metricName] += count +} + +func (s *TelemetrySubscriptionService) FlushMetrics() interop.TelemetrySubscriptionMetrics { + s.lock.Lock() + defer s.lock.Unlock() + metrics := interop.TelemetrySubscriptionMetrics{} + for name, count := range s.metrics { + metrics[name] = count + } + s.metrics = map[string]int{} + return metrics +} + +// Clear drops all subscriptions, which the sandbox does when an execution +// environment is reset and its extensions are gone. +func (s *TelemetrySubscriptionService) Clear() { + s.Flush() + s.lock.Lock() + defer s.lock.Unlock() + s.subscriptions = map[string]*subscription{} + s.earlyEvents = nil + s.off = false +} + +// TurnOff closes the subscription window. Existing subscriptions keep receiving +// telemetry; only further Subscribe calls are refused. +func (s *TelemetrySubscriptionService) TurnOff() { + s.lock.Lock() + defer s.lock.Unlock() + s.off = true +} + +func (s *TelemetrySubscriptionService) GetEndpointURL() string { return telemetryEndpointPath } + +func (s *TelemetrySubscriptionService) GetServiceClosedErrorMessage() string { + return "Telemetry API is no longer accepting subscriptions" +} + +func (s *TelemetrySubscriptionService) GetServiceClosedErrorType() string { + return "Telemetry.SubscriptionClosed" +} + +const telemetryEndpointPath = "/2022-07-01/telemetry" + +// resolveDestination rewrites the hostname extensions are told to use. Inside a +// real execution environment "sandbox" resolves to the host running the runtime; +// in the emulator everything shares one network namespace. +func resolveDestination(uri string) (string, error) { + parsed, err := url.Parse(uri) + if err != nil { + return "", err + } + if parsed.Scheme != "http" { + return "", fmt.Errorf("unsupported scheme %q", parsed.Scheme) + } + if parsed.Hostname() == "sandbox" { + host := "localhost" + if port := parsed.Port(); port != "" { + host += ":" + port + } + parsed.Host = host + } + return parsed.String(), nil +} + +func errorResponse(errorType, message string) []byte { + body, _ := json.Marshal(map[string]string{"errorType": errorType, "errorMessage": message}) + return body +} + +func durationOrDefault(value, fallback int) time.Duration { + if value <= 0 { + value = fallback + } + return time.Duration(value) * time.Millisecond +} + +func intOrDefault(value, fallback int) int { + if value <= 0 { + return fallback + } + return value +} + +func int64OrDefault(value, fallback int64) int64 { + if value <= 0 { + return fallback + } + return value +} + +var _ telemetry.SubscriptionAPI = (*TelemetrySubscriptionService)(nil) diff --git a/internal/lambda/rie/telemetry_subscription_test.go b/internal/lambda/rie/telemetry_subscription_test.go new file mode 100644 index 00000000..de5cbe10 --- /dev/null +++ b/internal/lambda/rie/telemetry_subscription_test.go @@ -0,0 +1,286 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package rie + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + standalonetelemetry "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry" +) + +// receiver stands in for a subscribed extension. +type receiver struct { + server *httptest.Server + lock sync.Mutex + events []map[string]interface{} +} + +func newReceiver(t *testing.T) *receiver { + t.Helper() + r := &receiver{} + r.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + body, err := io.ReadAll(request.Body) + if err != nil { + t.Errorf("reading delivery: %v", err) + return + } + var batch []map[string]interface{} + if err := json.Unmarshal(body, &batch); err != nil { + t.Errorf("delivered body is not a JSON array: %v (%s)", err, body) + return + } + r.lock.Lock() + r.events = append(r.events, batch...) + r.lock.Unlock() + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(r.server.Close) + return r +} + +func (r *receiver) received() []map[string]interface{} { + r.lock.Lock() + defer r.lock.Unlock() + events := make([]map[string]interface{}, len(r.events)) + copy(events, r.events) + return events +} + +// waitFor polls until the receiver has at least count events, so tests don't +// depend on delivery timing. +func (r *receiver) waitFor(t *testing.T, count int) []map[string]interface{} { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if events := r.received(); len(events) >= count { + return events + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("only %d of %d events delivered", len(r.received()), count) + return nil +} + +func subscribeBody(destination string, types []string, timeoutMs int) string { + body, _ := json.Marshal(map[string]interface{}{ + "schemaVersion": "2022-12-13", + "types": types, + "buffering": map[string]int{"timeoutMs": timeoutMs, "maxItems": 1000, "maxBytes": 262144}, + "destination": map[string]string{"protocol": "HTTP", "URI": destination}, + }) + return string(body) +} + +func subscribe(t *testing.T, service *TelemetrySubscriptionService, agent, destination string, types []string) { + t.Helper() + response, status, _, err := service.Subscribe(agent, strings.NewReader(subscribeBody(destination, types, 10)), nil, "") + require.NoError(t, err) + require.Equal(t, http.StatusOK, status, "subscribe should succeed: %s", response) +} + +func platformEvent(eventType string, record map[string]interface{}) standalonetelemetry.SandboxEvent { + return standalonetelemetry.SandboxEvent{ + Time: "2026-07-28T12:00:00Z", + Type: eventType, + PlatformEvent: record, + } +} + +func logEvent(eventType, message string) standalonetelemetry.SandboxEvent { + return standalonetelemetry.SandboxEvent{ + Time: "2026-07-28T12:00:00Z", + Type: eventType, + LogMessage: message, + } +} + +// A subscriber receives platform events as objects and log lines as strings, +// which is the distinction the Telemetry API defines and that consumers branch on. +func TestDeliversRecordsInTheDocumentedShape(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"platform", "function"}) + + service.Dispatch(platformEvent("platform.start", map[string]interface{}{"requestId": "abc"})) + service.Dispatch(logEvent("function", "hello from the handler")) + + // The subscription itself is reported, so expect it alongside the two events. + events := extension.waitFor(t, 3) + + byType := map[string]map[string]interface{}{} + for _, event := range events { + byType[event["type"].(string)] = event + } + + start := byType["platform.start"] + require.NotNil(t, start, "platform.start should be delivered") + record, ok := start["record"].(map[string]interface{}) + require.True(t, ok, "a platform record must be an object, got %T", start["record"]) + assert.Equal(t, "abc", record["requestId"]) + + function := byType["function"] + require.NotNil(t, function, "function telemetry should be delivered") + assert.Equal(t, "hello from the handler", function["record"], "a log record must be the line itself") + + assert.NotNil(t, byType["platform.telemetrySubscription"], "the subscription should be reported") +} + +// Subscribing to one category must not deliver the others, which is what lets an +// extension take function logs without receiving its own output back. +func TestTypeFilteringIsRespected(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"function"}) + + service.Dispatch(logEvent("function", "wanted")) + service.Dispatch(logEvent("extension", "not wanted")) + service.Dispatch(platformEvent("platform.start", map[string]interface{}{"requestId": "abc"})) + + events := extension.waitFor(t, 1) + time.Sleep(100 * time.Millisecond) // let anything unwanted arrive if it's going to + + for _, event := range extension.received() { + assert.Equal(t, "function", event["type"], "only function telemetry was subscribed to") + } + assert.Len(t, events, 1) +} + +// Initialization telemetry is produced before an extension can subscribe. Real +// Lambda delivers it anyway, so events are held and replayed to a new subscriber. +func TestEventsBeforeSubscriptionAreReplayed(t *testing.T) { + service := NewTelemetrySubscriptionService() + + service.Dispatch(platformEvent("platform.initStart", map[string]interface{}{"phase": "init"})) + + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"platform"}) + + events := extension.waitFor(t, 2) + types := map[string]bool{} + for _, event := range events { + types[event["type"].(string)] = true + } + assert.True(t, types["platform.initStart"], "initialization telemetry should survive until someone subscribes") +} + +// The sandbox closes the subscription window once initialization finishes. +// Delivery to those already subscribed has to continue regardless. +func TestTurnOffStopsSubscriptionsNotDelivery(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"platform"}) + + service.TurnOff() + + service.Dispatch(platformEvent("platform.runtimeDone", map[string]interface{}{"requestId": "abc"})) + events := extension.waitFor(t, 2) + + var delivered bool + for _, event := range events { + if event["type"] == "platform.runtimeDone" { + delivered = true + } + } + assert.True(t, delivered, "telemetry must keep flowing after the subscription window closes") + + _, status, _, err := service.Subscribe("late", strings.NewReader(subscribeBody(extension.server.URL, []string{"platform"}, 10)), nil, "") + assert.Error(t, err, "a late subscription should be refused") + assert.Equal(t, http.StatusBadRequest, status) +} + +// A reset means a fresh execution environment, so nothing should carry over. +func TestClearDropsSubscriptions(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"platform"}) + extension.waitFor(t, 1) + + service.Clear() + before := len(extension.received()) + service.Dispatch(platformEvent("platform.start", map[string]interface{}{"requestId": "abc"})) + time.Sleep(100 * time.Millisecond) + + assert.Equal(t, before, len(extension.received()), "a cleared service has no subscribers to deliver to") +} + +// Buffering limits are honored: reaching maxItems delivers without waiting for +// the timeout, which is what keeps a busy function's telemetry moving. +func TestMaxItemsFlushesImmediately(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + + body, _ := json.Marshal(map[string]interface{}{ + "schemaVersion": "2022-12-13", + "types": []string{"function"}, + // A long timeout, so only reaching maxItems can cause delivery. + "buffering": map[string]int{"timeoutMs": 60000, "maxItems": 2, "maxBytes": 262144}, + "destination": map[string]string{"protocol": "HTTP", "URI": extension.server.URL}, + }) + _, status, _, err := service.Subscribe("ext", strings.NewReader(string(body)), nil, "") + require.NoError(t, err) + require.Equal(t, http.StatusOK, status) + + service.Dispatch(logEvent("function", "one")) + service.Dispatch(logEvent("function", "two")) + + events := extension.waitFor(t, 2) + assert.Len(t, events, 2) +} + +func TestSubscribeRejectsUnusableRequests(t *testing.T) { + extension := newReceiver(t) + + testCases := []struct { + name string + body string + }{ + {"not JSON", `{`}, + {"no types", subscribeBody(extension.server.URL, nil, 10)}, + {"unsupported protocol", `{"types":["function"],"destination":{"protocol":"TCP","URI":"http://localhost:1"}}`}, + {"unparseable destination", `{"types":["function"],"destination":{"protocol":"HTTP","URI":"http://[::1"}}`}, + {"non-http destination", `{"types":["function"],"destination":{"protocol":"HTTP","URI":"file:///tmp/x"}}`}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + service := NewTelemetrySubscriptionService() + _, status, _, err := service.Subscribe("ext", strings.NewReader(tc.body), nil, "") + assert.NoError(t, err, "a bad request is reported by status, not by error") + assert.Equal(t, http.StatusBadRequest, status) + }) + } +} + +// Extensions are told to deliver to the "sandbox" host, which only resolves +// inside a real execution environment. +func TestSandboxHostIsRewritten(t *testing.T) { + testCases := []struct { + in string + want string + }{ + {"http://sandbox:3000", "http://localhost:3000"}, + {"http://sandbox:3000/path", "http://localhost:3000/path"}, + {"http://sandbox", "http://localhost"}, + {"http://localhost:3000", "http://localhost:3000"}, + {"http://127.0.0.1:3000", "http://127.0.0.1:3000"}, + } + + for _, tc := range testCases { + t.Run(tc.in, func(t *testing.T) { + got, err := resolveDestination(tc.in) + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } +} From 7f0923d0cdb45af2721598a45097b8beb8243e23 Mon Sep 17 00:00:00 2001 From: Liz Fong-Jones Date: Tue, 28 Jul 2026 14:21:03 -0700 Subject: [PATCH 2/2] fix: address review findings on the Telemetry API implementation Four problems, all reachable in normal use: Delivery used http.DefaultClient, which has no timeout. Batches that reach a size limit are delivered on the goroutine producing the event, so a subscriber that accepted a connection and never answered would stall the sandbox's event pipeline. Delivery now uses a client that gives up. platform.report was emitted with status success unconditionally, including from the invoke-timeout path, so a timed-out invocation was reported as a successful one -- the opposite of what an extension watching that field needs. The status is now threaded through. The report's memory metrics were serialized as JSON strings, because the memory size arrives as one. The Telemetry API defines them as numbers, and a consumer decoding into a typed struct rejects strings. They are parsed now, and the report also carries initDurationMs on a cold start, as a real function's does. Dispatch decided whether to buffer or deliver in one lock section and acted in another. A Subscribe interleaving between the two would insert itself, replay a copy of the buffer that did not yet hold the event, and leave the event buffered for nobody -- precisely the ordering the buffer exists to handle. The decision now happens under a single held lock. Tests cover the timeout, the status, the numeric metrics and the cold-start figure. The lost-event window is not covered: the two lock sections it needed to interleave between were adjacent and 400 attempts never hit it, so that fix rests on reading the code rather than on a failing test. Co-Authored-By: Claude Opus 5 (1M context) --- internal/lambda/rie/handlers.go | 56 +++++++++---- internal/lambda/rie/handlers_report_test.go | 66 +++++++++++++++ internal/lambda/rie/http.go | 2 +- internal/lambda/rie/telemetry_subscription.go | 28 ++++--- .../lambda/rie/telemetry_subscription_test.go | 81 +++++++++++++++++++ 5 files changed, 206 insertions(+), 27 deletions(-) create mode 100644 internal/lambda/rie/handlers_report_test.go diff --git a/internal/lambda/rie/handlers.go b/internal/lambda/rie/handlers.go index 3db2dbbc..73a5c16d 100644 --- a/internal/lambda/rie/handlers.go +++ b/internal/lambda/rie/handlers.go @@ -68,7 +68,39 @@ func GetenvWithDefault(key string, defaultValue string) string { return envValue } -func printEndReports(invokeId string, initDuration string, memorySize string, invokeStart time.Time, timeoutDuration time.Duration) { +// reportRecord builds the platform.report record. The metrics are numbers, as the +// Telemetry API defines them: a consumer decoding into a typed struct rejects the +// strings these values arrive as. +func reportRecord(invokeId string, status string, invokeDuration float64, memorySize string, initDurationMs float64) map[string]interface{} { + // The emulator cannot measure memory actually used, so it reports the + // configured size for both, as the printed REPORT line does. + memorySizeMB, err := strconv.Atoi(memorySize) + if err != nil { + log.Warnf("AWS_LAMBDA_FUNCTION_MEMORY_SIZE is %q, which is not a number", memorySize) + } + + metrics := map[string]interface{}{ + "durationMs": invokeDuration, + "billedDurationMs": math.Ceil(invokeDuration), + "memorySizeMB": memorySizeMB, + "maxMemoryUsedMB": memorySizeMB, + } + // Present only on the report for an invocation that initialized the + // environment, as it is in telemetry from a real function. + if initDurationMs > 0 { + metrics["initDurationMs"] = initDurationMs + } + + return map[string]interface{}{ + "requestId": invokeId, + "status": status, + "metrics": metrics, + } +} + +// printEndReports reports the end of an invocation. status is what the Telemetry +// API calls it: "success", or "timeout" when the invocation ran out of time. +func printEndReports(invokeId string, initDuration string, memorySize string, invokeStart time.Time, timeoutDuration time.Duration, status string, initDurationMs float64) { // Calcuation invoke duration invokeDuration := math.Min(float64(time.Now().Sub(invokeStart).Nanoseconds()), float64(timeoutDuration.Nanoseconds())) / float64(time.Millisecond) @@ -76,22 +108,12 @@ func printEndReports(invokeId string, initDuration string, memorySize string, in fmt.Println("END RequestId: " + invokeId) if telemetryEvents != nil { - now := time.Now().Format(time.RFC3339) // platform.end is deliberately not emitted: it is absent from telemetry // captured off a real function under the current schema version. telemetryEvents.Dispatch(standalonetelemetry.SandboxEvent{ - Time: now, - Type: "platform.report", - PlatformEvent: map[string]interface{}{ - "requestId": invokeId, - "status": "success", - "metrics": map[string]interface{}{ - "durationMs": invokeDuration, - "billedDurationMs": math.Ceil(invokeDuration), - "memorySizeMB": memorySize, - "maxMemoryUsedMB": memorySize, - }, - }, + Time: time.Now().Format(time.RFC3339), + Type: "platform.report", + PlatformEvent: reportRecord(invokeId, status, invokeDuration, memorySize, initDurationMs), }) } // We set the Max Memory Used and Memory Size to be the same (whatever it is set to) since there is @@ -123,6 +145,7 @@ func InvokeHandler(w http.ResponseWriter, r *http.Request, sandbox Sandbox, bs i } initDuration := "" + initDurationMs := float64(0) inv := GetenvWithDefault("AWS_LAMBDA_FUNCTION_TIMEOUT", "300") timeoutDuration, _ := time.ParseDuration(inv + "s") // Default @@ -143,6 +166,7 @@ func InvokeHandler(w http.ResponseWriter, r *http.Request, sandbox Sandbox, bs i float64(timeoutDuration.Nanoseconds())) / float64(time.Millisecond) initDuration = fmt.Sprintf("Init Duration: %.2f ms\t", initTimeMS) + initDurationMs = initTimeMS // Set initDone so next invokes do not try to Init the function again initDone = true @@ -229,7 +253,7 @@ func InvokeHandler(w http.ResponseWriter, r *http.Request, sandbox Sandbox, bs i w.WriteHeader(http.StatusGatewayTimeout) return case rapidcore.ErrInvokeTimeout: - printEndReports(invokePayload.ID, initDuration, memorySize, invokeStart, timeoutDuration) + printEndReports(invokePayload.ID, initDuration, memorySize, invokeStart, timeoutDuration, "timeout", initDurationMs) w.Write([]byte(fmt.Sprintf("Task timed out after %d.00 seconds", timeout))) time.Sleep(100 * time.Millisecond) @@ -238,7 +262,7 @@ func InvokeHandler(w http.ResponseWriter, r *http.Request, sandbox Sandbox, bs i } } - printEndReports(invokePayload.ID, initDuration, memorySize, invokeStart, timeoutDuration) + printEndReports(invokePayload.ID, initDuration, memorySize, invokeStart, timeoutDuration, "success", initDurationMs) if invokeResp.StatusCode != 0 { w.WriteHeader(invokeResp.StatusCode) diff --git a/internal/lambda/rie/handlers_report_test.go b/internal/lambda/rie/handlers_report_test.go new file mode 100644 index 00000000..4613dad0 --- /dev/null +++ b/internal/lambda/rie/handlers_report_test.go @@ -0,0 +1,66 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package rie + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The Telemetry API defines these metrics as numbers. A consumer decoding into a +// typed struct rejects the strings the memory size arrives as, so the record has +// to carry numbers however the value reached us. +func TestReportRecordMetricsAreNumbers(t *testing.T) { + encoded, err := json.Marshal(reportRecord("abc", "success", 12.5, "512", 0)) + require.NoError(t, err) + + var decoded struct { + RequestID string `json:"requestId"` + Status string `json:"status"` + Metrics struct { + DurationMs float64 `json:"durationMs"` + BilledDurationMs float64 `json:"billedDurationMs"` + MemorySizeMB int `json:"memorySizeMB"` + MaxMemoryUsedMB int `json:"maxMemoryUsedMB"` + } `json:"metrics"` + } + require.NoError(t, json.Unmarshal(encoded, &decoded), "a typed consumer must be able to decode this") + + assert.Equal(t, "abc", decoded.RequestID) + assert.Equal(t, "success", decoded.Status) + assert.Equal(t, 12.5, decoded.Metrics.DurationMs) + assert.Equal(t, float64(13), decoded.Metrics.BilledDurationMs) + assert.Equal(t, 512, decoded.Metrics.MemorySizeMB) +} + +// An invocation that ran out of time is not a successful one. Extensions alarm on +// this field, so reporting a timeout as a success would hide exactly the failure +// they are watching for. +func TestReportRecordCarriesTheInvocationStatus(t *testing.T) { + for _, status := range []string{"success", "timeout"} { + t.Run(status, func(t *testing.T) { + assert.Equal(t, status, reportRecord("abc", status, 1, "128", 0)["status"]) + }) + } +} + +// A memory size that isn't a number should not make the record undecodable. +func TestReportRecordSurvivesAnUnparseableMemorySize(t *testing.T) { + record := reportRecord("abc", "success", 1, "not-a-number", 0) + metrics := record["metrics"].(map[string]interface{}) + assert.Equal(t, 0, metrics["memorySizeMB"]) +} + +// A cold start reports how long initialization took; a warm invocation has no +// such figure and must not report a zero one. +func TestReportRecordIncludesInitDurationOnlyOnColdStart(t *testing.T) { + cold := reportRecord("abc", "success", 1, "128", 120.5)["metrics"].(map[string]interface{}) + assert.Equal(t, 120.5, cold["initDurationMs"]) + + warm := reportRecord("abc", "success", 1, "128", 0)["metrics"].(map[string]interface{}) + assert.NotContains(t, warm, "initDurationMs") +} diff --git a/internal/lambda/rie/http.go b/internal/lambda/rie/http.go index 4bc2d5ae..deaaf59d 100644 --- a/internal/lambda/rie/http.go +++ b/internal/lambda/rie/http.go @@ -6,9 +6,9 @@ package rie import ( "net/http" - log "github.com/sirupsen/logrus" "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/interop" "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore" + log "github.com/sirupsen/logrus" ) func startHTTPServer(ipport string, sandbox *rapidcore.SandboxBuilder, bs interop.Bootstrap) { diff --git a/internal/lambda/rie/telemetry_subscription.go b/internal/lambda/rie/telemetry_subscription.go index d9c35a59..150e20f1 100644 --- a/internal/lambda/rie/telemetry_subscription.go +++ b/internal/lambda/rie/telemetry_subscription.go @@ -172,13 +172,6 @@ func (s *TelemetrySubscriptionService) Subscribe(agentName string, body io.Reade // soon as initialization finishes, which stops new subscriptions but must leave // delivery to existing ones running for the life of the environment. func (s *TelemetrySubscriptionService) Dispatch(event standalonetelemetry.SandboxEvent) { - s.lock.Lock() - subscriptions := make([]*subscription, 0, len(s.subscriptions)) - for _, sub := range s.subscriptions { - subscriptions = append(subscriptions, sub) - } - s.lock.Unlock() - record := telemetryRecord{Time: event.Time, Type: event.Type} if event.PlatformEvent != nil { record.Record = event.PlatformEvent @@ -186,14 +179,23 @@ func (s *TelemetrySubscriptionService) Dispatch(event standalonetelemetry.Sandbo record.Record = event.LogMessage } - if len(subscriptions) == 0 { - s.lock.Lock() + // Buffering and delivery are chosen under a single held lock. Deciding in one + // lock section and acting in another lets a Subscribe interleave: it would + // insert itself and replay a copy of the buffer that does not yet hold this + // event, and the event would then be buffered for nobody. + s.lock.Lock() + if len(s.subscriptions) == 0 { if len(s.earlyEvents) < maxEarlyEvents { s.earlyEvents = append(s.earlyEvents, record) } s.lock.Unlock() return } + subscriptions := make([]*subscription, 0, len(s.subscriptions)) + for _, sub := range s.subscriptions { + subscriptions = append(subscriptions, sub) + } + s.lock.Unlock() for _, sub := range subscriptions { if sub.wants(event.Type) { @@ -262,7 +264,7 @@ func (sub *subscription) deliver(batch []telemetryRecord) { log.WithError(err).Warn("Telemetry API: could not encode a batch") return } - response, err := http.Post(sub.destination, "application/json", bytes.NewReader(body)) + response, err := deliveryClient.Post(sub.destination, "application/json", bytes.NewReader(body)) if err != nil { log.WithError(err).Warnf("Telemetry API: could not deliver to %s", sub.destination) return @@ -340,6 +342,12 @@ func (s *TelemetrySubscriptionService) GetServiceClosedErrorType() string { const telemetryEndpointPath = "/2022-07-01/telemetry" +// Batches that reach a size limit are delivered on the goroutine producing the +// event, so a subscriber that accepts a connection and never answers would stall +// the sandbox's event pipeline. The default client has no timeout; this one gives +// up instead. +var deliveryClient = &http.Client{Timeout: 5 * time.Second} + // resolveDestination rewrites the hostname extensions are told to use. Inside a // real execution environment "sandbox" resolves to the host running the runtime; // in the emulator everything shares one network namespace. diff --git a/internal/lambda/rie/telemetry_subscription_test.go b/internal/lambda/rie/telemetry_subscription_test.go index de5cbe10..9709f742 100644 --- a/internal/lambda/rie/telemetry_subscription_test.go +++ b/internal/lambda/rie/telemetry_subscription_test.go @@ -284,3 +284,84 @@ func TestSandboxHostIsRewritten(t *testing.T) { }) } } + +// Exercises Subscribe and Dispatch concurrently, for the race detector and to +// confirm the event still arrives. +// +// This does not reproduce the lost-event window that motivated deciding +// buffer-versus-deliver under a single lock: the two lock sections it needed to +// interleave between were adjacent, and 400 attempts never hit it. The +// correctness of that arrangement rests on reading the code, not on this test. +func TestConcurrentSubscribeAndDispatch(t *testing.T) { + for attempt := 0; attempt < 100; attempt++ { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + service.Dispatch(platformEvent("platform.initStart", map[string]interface{}{"phase": "init"})) + }() + go func() { + defer wg.Done() + subscribe(t, service, "ext", extension.server.URL, []string{"platform"}) + }() + wg.Wait() + + // Whichever order they interleaved in, the event is either delivered or + // still buffered for the next subscriber. It must not be stranded. + deadline := time.Now().Add(2 * time.Second) + var seen bool + for time.Now().Before(deadline) && !seen { + for _, event := range extension.received() { + if event["type"] == "platform.initStart" { + seen = true + } + } + if !seen { + time.Sleep(10 * time.Millisecond) + } + } + if !seen { + t.Fatalf("attempt %d: initStart reached no subscriber", attempt) + } + } +} + +// A subscriber that accepts a connection and never answers must not stall the +// sandbox: delivery happens on the event-producing goroutine when a batch fills. +func TestDeliveryToAHungSubscriberDoesNotBlockForever(t *testing.T) { + blocked := make(chan struct{}) + + hung := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-blocked // never answers while the test runs + })) + // Close waits for handlers in flight, so the handler has to be released + // first. Defers run last-in-first-out, hence this order. + defer hung.Close() + defer close(blocked) + + service := NewTelemetrySubscriptionService() + body, _ := json.Marshal(map[string]interface{}{ + "types": []string{"function"}, + "buffering": map[string]int{"timeoutMs": 60000, "maxItems": 1, "maxBytes": 262144}, + "destination": map[string]string{"protocol": "HTTP", "URI": hung.URL}, + }) + _, status, _, err := service.Subscribe("ext", strings.NewReader(string(body)), nil, "") + require.NoError(t, err) + require.Equal(t, http.StatusOK, status) + + // maxItems of 1 means this dispatch delivers synchronously. + done := make(chan struct{}) + go func() { + service.Dispatch(logEvent("function", "hello")) + close(done) + }() + + select { + case <-done: + case <-time.After(deliveryClient.Timeout + 5*time.Second): + t.Fatal("Dispatch never returned; a hung subscriber can stall the event pipeline") + } +}