Merge branch 'main' into conf-workflows-e2e-add-chain-write

Author: vreff

enclave/nitro/host/host.go

@@ -1056,7 +1056,10 @@ func main() {
 	if err := mainServer.Shutdown(shutdownCtx); err != nil {
 		lggr.Errorw("main server shutdown error", "error", err)
 	}
-	if err := telemetry.close(shutdownCtx); err != nil {
+	// Give telemetry its own timeout budget
+	flushCtx, flushCancel := context.WithTimeout(context.Background(), *shutdownTimeout)
+	defer flushCancel()
+	if err := telemetry.close(flushCtx); err != nil {
 		lggr.Errorw("telemetry shutdown error", "error", err)
 	}
 

enclave/server/response_emitter.go

@@ -2,8 +2,10 @@ package server
 
 import (
 	"encoding/json"
+	"maps"
 	"math"
 	"net/http"
+	"sync"
 	"time"
 
 	"github.com/smartcontractkit/chainlink-confidential-compute/types"
@@ -16,6 +18,7 @@ const durationBucketSeconds = 0.01
 // ResponseEmitter collects metrics to be included in the response payload
 // instead of emitting them via OTel.
 type ResponseEmitter struct {
+	mu      sync.RWMutex
 	metrics []types.MetricEvent
 	// Captured at construction so WriteErrorResponse can report the total request
 	// duration on error, not just on the success path.
@@ -33,9 +36,13 @@ func NewResponseEmitter() *ResponseEmitter {
 }
 
 func (e *ResponseEmitter) Emit(event string, details map[string]any) {
+	details = roundDurations(maps.Clone(details))
+
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	e.metrics = append(e.metrics, types.MetricEvent{
 		Event:   event,
-		Details: roundDurations(details),
+		Details: details,
 	})
 }
 
@@ -59,16 +66,29 @@ func roundDurations(details map[string]any) map[string]any {
 // Duplicate event names collapse to the last occurrence; use GetMetricEvents
 // when per-occurrence counts matter (e.g. repeated capability calls).
 func (e *ResponseEmitter) GetMetrics() map[string]any {
-	result := make(map[string]any)
-	for _, m := range e.metrics {
-		result[m.Event] = m.Details
-	}
-	return result
+	metrics, _ := e.Snapshot()
+	return metrics
 }
 
 // GetMetricEvents returns every collected event in order, preserving duplicates.
 func (e *ResponseEmitter) GetMetricEvents() []types.MetricEvent {
-	return e.metrics
+	_, events := e.Snapshot()
+	return events
+}
+
+// Snapshot returns consistent map and ordered representations of all events.
+func (e *ResponseEmitter) Snapshot() (map[string]any, []types.MetricEvent) {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+
+	metrics := make(map[string]any)
+	events := make([]types.MetricEvent, len(e.metrics))
+	for i, event := range e.metrics {
+		details := maps.Clone(event.Details)
+		events[i] = types.MetricEvent{Event: event.Event, Details: details}
+		metrics[event.Event] = details
+	}
+	return metrics, events
 }
 
 // WriteErrorResponse writes a JSON error response that includes the accumulated
@@ -87,10 +107,11 @@ func (e *ResponseEmitter) WriteErrorResponse(w http.ResponseWriter, msg string,
 
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(statusCode)
+	metrics, metricEvents := e.Snapshot()
 	resp := types.EnclaveErrorResponse{
 		Error:        msg,
-		Metrics:      e.GetMetrics(),
-		MetricEvents: e.GetMetricEvents(),
+		Metrics:      metrics,
+		MetricEvents: metricEvents,
 	}
 	_ = json.NewEncoder(w).Encode(resp)
 }

enclave/server/response_emitter_test.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -35,3 +36,28 @@ func TestResponseEmitter_ErrorResponseCarriesMetricEvents(t *testing.T) {
 	// the full ordered list (2 capability_execution + 1 request_completed).
 	require.Len(t, e.GetMetricEvents(), 2)
 }
+
+func TestResponseEmitter_ConcurrentEmit(t *testing.T) {
+	const (
+		parallelCalls = 30 // chainlink-common's default per-execution pending-call limit
+		eventsPerCall = 3  // capability_started, capability_execution, capability_finished
+	)
+
+	e := NewResponseEmitter()
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Add(parallelCalls)
+	for call := range parallelCalls {
+		go func() {
+			defer wg.Done()
+			<-start
+			e.Emit("capability_started", map[string]any{"step_ref": call})
+			e.Emit("capability_execution", map[string]any{"step_ref": call})
+			e.Emit("capability_finished", map[string]any{"step_ref": call})
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	assert.Len(t, e.GetMetricEvents(), parallelCalls*eventsPerCall)
+}

enclave/server/server.go

@@ -756,8 +756,7 @@ func (s *enclaveServer) handleExecute(w http.ResponseWriter, r *http.Request) {
 		responseEmitter.WriteErrorResponse(w, fmt.Sprintf("error creating attestation: %v", err), http.StatusInternalServerError)
 		return
 	}
-	resp.Metrics = responseEmitter.GetMetrics()
-	resp.MetricEvents = responseEmitter.GetMetricEvents()
+	resp.Metrics, resp.MetricEvents = responseEmitter.Snapshot()
 	resp.Attestation = att
 
 	respBytes, err := json.Marshal(resp)