[REG-10] W0.5 · Async context propagation — workers + recoverAsyncPanic (partial)

backend/internal/observability/worker.go

@@ -0,0 +1,45 @@
+package observability
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// Tracer returns the shared tracer for hand-instrumented spans outside the
+// otelgin/gorm/redisotel integrations — same instrumentation scope name
+// ("regen") as those, so all of Regen's spans group under one scope.
+func Tracer() trace.Tracer {
+	return otel.Tracer("regen")
+}
+
+// StartWorkerTick opens a fresh root span for one iteration of a periodic
+// background worker's loop.
+//
+// Deliberately rooted at context.Background(), not the worker's long-lived
+// lifecycle context (the ctx passed to Run): that context spans the whole
+// process, and parenting every tick under it would produce one unbounded
+// "trace" per worker for the server's entire lifetime instead of one span
+// per iteration — the wrong shape for "what happened on this particular
+// tick" queries.
+//
+// tracer is a parameter rather than always using the package-global Tracer()
+// so tests can inject an isolated tracer instead of mutating global otel
+// state (see REG-7's finding: the global TracerProvider is a shared,
+// order-dependent singleton, unsafe to save/restore per-test). Production
+// callers pass observability.Tracer().
+func StartWorkerTick(tracer trace.Tracer, name string) (context.Context, trace.Span) {
+	return tracer.Start(context.Background(), name)
+}
+
+// EndWorkerTick ends span, recording err as the span's error status when
+// non-nil. Callers pair this with StartWorkerTick, typically via defer.
+func EndWorkerTick(span trace.Span, err error) {
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+	}
+	span.End()
+}

backend/internal/observability/worker_test.go

@@ -0,0 +1,78 @@
+package observability
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"go.opentelemetry.io/otel/codes"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// newTestTracer builds an isolated tracer backed by a span recorder, so tests
+// never touch the process-global TracerProvider (see REG-7: it's a shared,
+// order-dependent singleton, unreliable to save/restore per-test).
+func newTestTracer(t *testing.T) (trace.Tracer, *tracetest.SpanRecorder) {
+	t.Helper()
+	sr := tracetest.NewSpanRecorder()
+	tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
+	t.Cleanup(func() { _ = tp.Shutdown(context.Background()) })
+	return tp.Tracer("test"), sr
+}
+
+func TestStartWorkerTick_ProducesAFreshRootSpanEachCall(t *testing.T) {
+	tracer, sr := newTestTracer(t)
+
+	_, span1 := StartWorkerTick(tracer, "escalation_worker.tick")
+	span1.End()
+	_, span2 := StartWorkerTick(tracer, "escalation_worker.tick")
+	span2.End()
+
+	spans := sr.Ended()
+	if len(spans) != 2 {
+		t.Fatalf("expected 2 independent spans, got %d", len(spans))
+	}
+	if spans[0].SpanContext().TraceID() == spans[1].SpanContext().TraceID() {
+		t.Error("expected each tick to start its own trace (fresh root span), got the same trace ID for both — ticks must not share one unbounded trace for the worker's lifetime")
+	}
+	for _, s := range spans {
+		if s.Name() != "escalation_worker.tick" {
+			t.Errorf("span name = %q, want %q", s.Name(), "escalation_worker.tick")
+		}
+	}
+}
+
+func TestEndWorkerTick_RecordsErrorStatusWhenErrGiven(t *testing.T) {
+	tracer, sr := newTestTracer(t)
+
+	_, span := StartWorkerTick(tracer, "push_cleanup_worker.tick")
+	EndWorkerTick(span, errors.New("boom"))
+
+	spans := sr.Ended()
+	if len(spans) != 1 {
+		t.Fatalf("expected 1 span, got %d", len(spans))
+	}
+	if spans[0].Status().Code != codes.Error {
+		t.Errorf("expected span status Error, got %v", spans[0].Status().Code)
+	}
+	if len(spans[0].Events()) == 0 {
+		t.Fatal("expected the error to be recorded as a span event")
+	}
+}
+
+func TestEndWorkerTick_NoErrorStatusWhenErrNil(t *testing.T) {
+	tracer, sr := newTestTracer(t)
+
+	_, span := StartWorkerTick(tracer, "holiday_worker.tick")
+	EndWorkerTick(span, nil)
+
+	spans := sr.Ended()
+	if len(spans) != 1 {
+		t.Fatalf("expected 1 span, got %d", len(spans))
+	}
+	if spans[0].Status().Code == codes.Error {
+		t.Error("expected no Error status when err is nil")
+	}
+}

backend/internal/services/async_panic_test.go

@@ -0,0 +1,152 @@
+package services
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"testing"
+
+	"github.com/FluidifyAI/Regen/backend/internal/observability"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+)
+
+// withTestAsyncSpanTracer swaps the package-level asyncSpanTracer for the
+// duration of the test, backed by an isolated span recorder. A plain
+// package-level var (not otel's global TracerProvider) is reliably
+// save/restorable — unlike the process-global TracerProvider, which REG-7
+// found to be a shared, order-dependent singleton unsafe to save/restore
+// per-test.
+func withTestAsyncSpanTracer(t *testing.T) *tracetest.SpanRecorder {
+	t.Helper()
+	sr := tracetest.NewSpanRecorder()
+	tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
+	t.Cleanup(func() { _ = tp.Shutdown(context.Background()) })
+
+	prev := asyncSpanTracer
+	asyncSpanTracer = tp.Tracer("test")
+	t.Cleanup(func() { asyncSpanTracer = prev })
+
+	return sr
+}
+
+func TestRecoverAsyncPanic_LogsViaErrorContextWhenPanicRecovered(t *testing.T) {
+	withTestAsyncSpanTracer(t)
+
+	var buf bytes.Buffer
+	prevLogger := slog.Default()
+	slog.SetDefault(slog.New(observability.NewContextHandler(slog.NewJSONHandler(&buf, nil))))
+	defer slog.SetDefault(prevLogger)
+
+	func() {
+		defer recoverAsyncPanic(context.Background(), "testOp", "incident_id", "abc-123")
+		panic("boom")
+	}()
+
+	out := buf.String()
+	if !bytes.Contains(buf.Bytes(), []byte("testOp")) {
+		t.Errorf("expected the log line to name the op, got: %q", out)
+	}
+	if !bytes.Contains(buf.Bytes(), []byte("boom")) {
+		t.Errorf("expected the log line to include the panic value, got: %q", out)
+	}
+	if !bytes.Contains(buf.Bytes(), []byte("abc-123")) {
+		t.Errorf("expected the log line to include the extra fields, got: %q", out)
+	}
+}
+
+func TestRecoverAsyncPanic_DoesNotReturnAPanicToTheCaller(t *testing.T) {
+	withTestAsyncSpanTracer(t)
+
+	// The whole point of this helper: a panic inside the deferred block must
+	// not propagate and crash the goroutine/test process.
+	func() {
+		defer recoverAsyncPanic(context.Background(), "testOp")
+		panic("should be recovered")
+	}()
+}
+
+func TestRecoverAsyncPanic_NoOpWhenNoPanicOccurs(t *testing.T) {
+	sr := withTestAsyncSpanTracer(t)
+
+	var buf bytes.Buffer
+	prevLogger := slog.Default()
+	slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, nil)))
+	defer slog.SetDefault(prevLogger)
+
+	func() {
+		defer recoverAsyncPanic(context.Background(), "testOp")
+		// no panic
+	}()
+
+	if buf.Len() != 0 {
+		t.Errorf("expected no log output when nothing panicked, got: %q", buf.String())
+	}
+	if len(sr.Ended()) != 0 {
+		t.Errorf("expected no span when nothing panicked, got %d", len(sr.Ended()))
+	}
+}
+
+func TestRecoverAsyncPanic_RecordsErrorOnASpanLinkedToTheOriginatingSpan(t *testing.T) {
+	sr := withTestAsyncSpanTracer(t)
+
+	// A real originating span, standing in for the eventual request-scoped
+	// span REG-157 will thread through — proves the link, once fed a real
+	// context, actually connects to the right trace.
+	originTP := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
+	defer originTP.Shutdown(context.Background())
+	originCtx, originSpan := originTP.Tracer("test").Start(context.Background(), "originating-request")
+	originSpan.End()
+
+	func() {
+		defer recoverAsyncPanic(originCtx, "sendTelegramIncidentCreated", "incident_id", "abc-123")
+		panic("telegram send failed")
+	}()
+
+	spans := sr.Ended()
+	var panicSpan sdktrace.ReadOnlySpan
+	for _, s := range spans {
+		if s.Name() == "sendTelegramIncidentCreated" {
+			panicSpan = s
+		}
+	}
+	if panicSpan == nil {
+		t.Fatalf("expected a span named %q, got: %v", "sendTelegramIncidentCreated", spanNamesFor(spans))
+	}
+
+	links := panicSpan.Links()
+	if len(links) != 1 {
+		t.Fatalf("expected 1 link to the originating span, got %d", len(links))
+	}
+	if links[0].SpanContext.TraceID() != originSpan.SpanContext().TraceID() {
+		t.Error("the panic span's link points at a different trace than the originating span")
+	}
+
+	// Linked, not parented: the panic span must NOT be a child of the
+	// originating span (which may well have already ended by the time an
+	// async goroutine panics) — it's causally related, not nested.
+	if panicSpan.Parent().IsValid() {
+		t.Error("expected the panic span to have no parent (linked, not nested) — the async goroutine may outlive the originating span")
+	}
+}
+
+func TestRecoverAsyncPanic_HandlesBackgroundContextGracefully(t *testing.T) {
+	withTestAsyncSpanTracer(t)
+
+	// This is the real, current state of every one of the 16 call sites in
+	// incident_service.go today: no real request context available yet
+	// (see REG-157). Must not panic or error just because the link source
+	// has no valid span.
+	func() {
+		defer recoverAsyncPanic(context.Background(), "testOp")
+		panic("boom")
+	}()
+}
+
+func spanNamesFor(spans []sdktrace.ReadOnlySpan) []string {
+	names := make([]string, len(spans))
+	for i, s := range spans {
+		names[i] = s.Name()
+	}
+	return names
+}