fix(sink): make label output deterministic, thread context, and join batch errors

Prometheus serialized labels in map-iteration order, breaking the documented
stable output and golden assertions; sort the keys. The statsd Aggregator
dropped the caller context before the registry transform, losing deadline and
trace propagation; thread it through metricFor. The reservoir non-batch
dispatch kept only the last error; join them so no per-payload failure is lost.

Signed-off-by: Joshua Temple <joshua.temple@stablekernel.com>

Author: joshua-temple

sink/prometheus/determinism_test.go

@@ -0,0 +1,68 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package prometheus_test
+
+import (
+	"context"
+	"net/http"
+	"testing"
+
+	csink "github.com/stablekernel/crucible/sink"
+	prom "github.com/stablekernel/crucible/sink/prometheus"
+)
+
+// TestPushMetrics_MultiLabelDeterministicOrder verifies that a metric with
+// several labels serializes its labels in a stable, sorted order regardless of
+// map iteration order. Serializing the same metric many times must produce
+// byte-identical output.
+func TestPushMetrics_MultiLabelDeterministicOrder(t *testing.T) {
+	t.Parallel()
+
+	metrics := []prom.Metric{
+		{
+			Name:  "http_requests_total",
+			Type:  prom.TypeCounter,
+			Value: "7",
+			Labels: map[string]string{
+				"method": "GET",
+				"status": "200",
+				"region": "us-east-1",
+				"app":    "api",
+			},
+		},
+	}
+
+	const want = `http_requests_total{app="api",method="GET",region="us-east-1",status="200"} 7`
+
+	var first string
+	for i := 0; i < 50; i++ {
+		fd := &fakeDoer{status: http.StatusOK}
+		reg := prom.NewRegistry()
+		csink.Register(reg, func(_ context.Context, _ deployFinished) csink.Op[prom.Doer] {
+			return prom.PushMetrics("http://gw", "job", metrics)
+		})
+		outlet := prom.New(fd, reg)
+		if err := outlet.Sink(context.Background(), deployFinished{Env: "prod"}); err != nil {
+			t.Fatalf("Sink() error = %v", err)
+		}
+		if i == 0 {
+			first = fd.body
+			if !contains(first, want) {
+				t.Fatalf("body = %q, want it to contain %q", first, want)
+			}
+			continue
+		}
+		if fd.body != first {
+			t.Fatalf("label order is not deterministic:\n run 0: %q\n run %d: %q", first, i, fd.body)
+		}
+	}
+}
+
+func contains(haystack, needle string) bool {
+	for i := 0; i+len(needle) <= len(haystack); i++ {
+		if haystack[i:i+len(needle)] == needle {
+			return true
+		}
+	}
+	return false
+}

sink/prometheus/prometheus.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"sort"
 	"strings"
 
 	csink "github.com/stablekernel/crucible/sink"
@@ -106,14 +107,18 @@ func marshalMetrics(metrics []Metric) string {
 		b.WriteString(m.Name)
 		if len(m.Labels) > 0 {
 			b.WriteByte('{')
-			first := true
-			// Iterate in a stable order for deterministic output.
-			for k, v := range m.Labels {
-				if !first {
+			// Sort the label keys so the serialized output is deterministic
+			// regardless of map iteration order.
+			keys := make([]string, 0, len(m.Labels))
+			for k := range m.Labels {
+				keys = append(keys, k)
+			}
+			sort.Strings(keys)
+			for i, k := range keys {
+				if i > 0 {
 					b.WriteByte(',')
 				}
-				fmt.Fprintf(&b, `%s="%s"`, k, v)
-				first = false
+				fmt.Fprintf(&b, `%s="%s"`, k, m.Labels[k])
 			}
 			b.WriteByte('}')
 		}

sink/reservoir.go

@@ -4,6 +4,7 @@ package sink
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"time"
 
@@ -81,8 +82,9 @@ type reservoir struct {
 	flushLatency telemetry.Histogram
 	dropped      telemetry.Counter
 
-	mu  sync.Mutex
-	buf []any
+	mu     sync.Mutex
+	buf    []any
+	closed bool // set by Shutdown; Sink passes through afterward
 
 	cancel context.CancelFunc
 	wg     sync.WaitGroup
@@ -121,8 +123,15 @@ func Reservoir(inner Outlet, opts ...ReservoirOption) Outlet {
 
 // Sink buffers payload, flushing synchronously if the buffer reaches the batch
 // size. An over-cap payload is dropped and counted, never blocking the caller.
+// After Shutdown the background loop is gone, so a payload would otherwise be
+// stranded in the buffer; once closed, Sink dispatches each payload synchronously
+// to inner instead of buffering it.
 func (r *reservoir) Sink(ctx context.Context, payload any) error {
 	r.mu.Lock()
+	if r.closed {
+		r.mu.Unlock()
+		return r.dispatch(ctx, []any{payload})
+	}
 	if r.maxBuf > 0 && len(r.buf) >= r.maxBuf {
 		r.mu.Unlock()
 		r.dropped.Add(ctx, 1)
@@ -158,6 +167,9 @@ func (r *reservoir) Shutdown(ctx context.Context) error {
 			r.cancel()
 		}
 		r.wg.Wait()
+		r.mu.Lock()
+		r.closed = true
+		r.mu.Unlock()
 	})
 	return r.Flush(ctx)
 }
@@ -183,11 +195,13 @@ func (r *reservoir) dispatch(ctx context.Context, batch []any) error {
 	if b, ok := r.inner.(BatchOutlet); ok {
 		err = b.SinkBatch(ctx, batch)
 	} else {
+		var errs []error
 		for _, p := range batch {
 			if e := r.inner.Sink(ctx, p); e != nil {
-				err = e
+				errs = append(errs, e)
 			}
 		}
+		err = errors.Join(errs...)
 	}
 	r.flushLatency.Record(ctx, float64(r.now().Sub(start).Milliseconds()))
 	return err

sink/reservoir_errors_test.go

@@ -0,0 +1,91 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package sink
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+)
+
+// erroringOutlet fails its first failN Sink calls with a distinct error each,
+// then succeeds. It is not a BatchOutlet, so the reservoir takes the per-payload
+// dispatch path where errors must be joined.
+type erroringOutlet struct {
+	mu     sync.Mutex
+	errs   []error
+	called int
+}
+
+func (o *erroringOutlet) Sink(_ context.Context, _ any) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	i := o.called
+	o.called++
+	if i < len(o.errs) {
+		return o.errs[i]
+	}
+	return nil
+}
+
+// TestReservoirNonBatchJoinsAllErrors verifies the per-payload dispatch path
+// joins every error rather than keeping only the last one.
+func TestReservoirNonBatchJoinsAllErrors(t *testing.T) {
+	t.Parallel()
+
+	err1 := errors.New("first failed")
+	err2 := errors.New("second failed")
+	out := &erroringOutlet{errs: []error{err1, err2}}
+	r := Reservoir(out, WithBatchSize(0), WithBatchInterval(0)).(*reservoir)
+
+	_ = r.Sink(context.Background(), "a")
+	_ = r.Sink(context.Background(), "b")
+	err := r.Flush(context.Background())
+	if err == nil {
+		t.Fatal("Flush() = nil, want a joined error")
+	}
+	if !errors.Is(err, err1) {
+		t.Errorf("joined error does not contain the first error: %v", err)
+	}
+	if !errors.Is(err, err2) {
+		t.Errorf("joined error does not contain the second error: %v", err)
+	}
+}
+
+// TestReservoirSinkAfterShutdownPassesThrough verifies a payload sunk after
+// Shutdown is dispatched synchronously to inner rather than stranded in the
+// buffer with no loop left to flush it.
+func TestReservoirSinkAfterShutdownPassesThrough(t *testing.T) {
+	t.Parallel()
+
+	bucket := NewBucket()
+	r := Reservoir(bucket, WithBatchSize(100), WithBatchInterval(0)).(*reservoir)
+	if err := r.Shutdown(context.Background()); err != nil {
+		t.Fatalf("Shutdown() error = %v", err)
+	}
+
+	if err := r.Sink(context.Background(), "late"); err != nil {
+		t.Fatalf("Sink() after Shutdown error = %v", err)
+	}
+	all := bucket.All()
+	if len(all) != 1 || all[0] != "late" {
+		t.Fatalf("inner has %v, want the post-shutdown payload delivered immediately", all)
+	}
+}
+
+// TestReservoirSinkAfterShutdownSurfacesError verifies the pass-through path
+// still returns inner's error to the caller.
+func TestReservoirSinkAfterShutdownSurfacesError(t *testing.T) {
+	t.Parallel()
+
+	boom := errors.New("inner down")
+	out := &erroringOutlet{errs: []error{boom}}
+	r := Reservoir(out, WithBatchSize(100), WithBatchInterval(0)).(*reservoir)
+	if err := r.Shutdown(context.Background()); err != nil {
+		t.Fatalf("Shutdown() error = %v", err)
+	}
+	if err := r.Sink(context.Background(), "late"); !errors.Is(err, boom) {
+		t.Fatalf("Sink() after Shutdown = %v, want %v", err, boom)
+	}
+}

sink/statsd/options_test.go

@@ -0,0 +1,109 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package statsd_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	csink "github.com/stablekernel/crucible/sink"
+	statsdsink "github.com/stablekernel/crucible/sink/statsd"
+)
+
+// TestWithName_AppearsInFlushError verifies WithName overrides the outlet name
+// carried on the *sink.Error returned from a failing flush.
+func TestWithName_AppearsInFlushError(t *testing.T) {
+	t.Parallel()
+
+	boom := errors.New("emit failed")
+	fc := &fakeClient{err: boom}
+	agg := statsdsink.NewAggregator(fc,
+		statsdsink.WithName("custom-statsd"),
+		statsdsink.WithInterval(0), // no background loop; Flush is the only emit
+	)
+
+	if err := agg.Sink(context.Background(), statsdsink.Metric{Type: statsdsink.TypeCount, Name: "n", Int: 1, Rate: 1}); err != nil {
+		t.Fatalf("Sink() error = %v", err)
+	}
+	f, ok := agg.(csink.Flusher)
+	if !ok {
+		t.Fatal("Aggregator does not implement sink.Flusher")
+	}
+	err := f.Flush(context.Background())
+	if !errors.Is(err, boom) {
+		t.Fatalf("Flush() = %v, want to wrap %v", err, boom)
+	}
+	var se *csink.Error
+	if !errors.As(err, &se) || se.Outlet != "custom-statsd" {
+		t.Fatalf("error = %+v, want Outlet=custom-statsd", se)
+	}
+}
+
+// TestWithName_EmptyIgnored verifies an empty name leaves the default in place.
+func TestWithName_EmptyIgnored(t *testing.T) {
+	t.Parallel()
+
+	boom := errors.New("emit failed")
+	fc := &fakeClient{err: boom}
+	agg := statsdsink.NewAggregator(fc, statsdsink.WithName(""), statsdsink.WithInterval(0))
+	_ = agg.Sink(context.Background(), statsdsink.Metric{Type: statsdsink.TypeCount, Name: "n", Int: 1, Rate: 1})
+
+	err := agg.(csink.Flusher).Flush(context.Background())
+	var se *csink.Error
+	if !errors.As(err, &se) || se.Outlet != "statsd" {
+		t.Fatalf("error = %+v, want default Outlet=statsd", se)
+	}
+}
+
+// TestWithClock_NonNilAcceptedNilIgnored verifies WithClock installs a non-nil
+// clock and ignores a nil one. The aggregator stays usable in both cases.
+func TestWithClock_NonNilAcceptedNilIgnored(t *testing.T) {
+	t.Parallel()
+
+	fixed := time.Unix(42, 0)
+	clock := func() time.Time { return fixed }
+
+	for _, tc := range []struct {
+		name string
+		now  func() time.Time
+	}{
+		{"non-nil clock", clock},
+		{"nil clock falls back to default", nil},
+	} {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			fc := &fakeClient{}
+			agg := statsdsink.NewAggregator(fc, statsdsink.WithClock(tc.now), statsdsink.WithInterval(0))
+			if err := agg.Sink(context.Background(), statsdsink.Metric{Type: statsdsink.TypeGauge, Name: "g", Value: 1, Rate: 1}); err != nil {
+				t.Fatalf("Sink() error = %v", err)
+			}
+			if err := agg.(csink.Flusher).Flush(context.Background()); err != nil {
+				t.Fatalf("Flush() error = %v", err)
+			}
+			if got := len(fc.snapshot()); got != 1 {
+				t.Fatalf("emitted %d metrics, want 1", got)
+			}
+		})
+	}
+}
+
+// TestDial_ReturnsUsableClient verifies Dial constructs a Client over a UDP
+// address without requiring a live StatsD server. The datadog SDK resolves the
+// address lazily, so a valid host:port yields a ready client.
+func TestDial_ReturnsUsableClient(t *testing.T) {
+	t.Parallel()
+
+	c, err := statsdsink.Dial("127.0.0.1:8125")
+	if err != nil {
+		t.Fatalf("Dial() error = %v", err)
+	}
+	if c == nil {
+		t.Fatal("Dial() returned a nil Client")
+	}
+	if err := c.Count("dial.test", 1, nil, 1); err != nil {
+		t.Fatalf("Count() on dialed client error = %v", err)
+	}
+}

sink/statsd/statsd.go

@@ -250,8 +250,8 @@ func NewAggregator(client Client, opts ...AggregatorOption) csink.Outlet {
 // directly; any other payload is resolved through the registry installed with
 // WithRegistry, and an unregistered type returns sink.ErrUnregistered. The
 // first Sink starts the background flush loop when an interval is configured.
-func (a *Aggregator) Sink(_ context.Context, payload any) error {
-	m, ok := a.metricFor(payload)
+func (a *Aggregator) Sink(ctx context.Context, payload any) error {
+	m, ok := a.metricFor(ctx, payload)
 	if !ok {
 		return csink.ErrUnregistered
 	}
@@ -263,7 +263,9 @@ func (a *Aggregator) Sink(_ context.Context, payload any) error {
 }
 
 // metricFor resolves payload to a Metric, either directly or via the registry.
-func (a *Aggregator) metricFor(payload any) (Metric, bool) {
+// The caller's context is threaded into the registry transformer so deadline
+// and trace propagation survive the resolution step.
+func (a *Aggregator) metricFor(ctx context.Context, payload any) (Metric, bool) {
 	if m, ok := payload.(Metric); ok {
 		return m, true
 	}
@@ -274,7 +276,7 @@ func (a *Aggregator) metricFor(payload any) (Metric, bool) {
 	if !ok {
 		return Metric{}, false
 	}
-	return transform(context.Background(), payload), true
+	return transform(ctx, payload), true
 }
 
 // Flush swaps the live window for a fresh one and emits the captured window to