feat(otel): W3C traceparent transport-header propagation + AMQP/SQS wiring (ADR-0028)

Implements ADR-0025 Option 2: true cross-hop parent-child span linkage via the
W3C traceparent header carried out-of-band on the transport (frozen envelope,
GR-1 intact). The otel submodule injects/extracts traceparent over a broker-
agnostic header seam (HeaderPublisher.PublishWithHeaders / ReceivedMessage.
Headers). AMQP (amqp091.Table headers) and SQS (MessageAttributes) implement the
seam — they now carry traceparent without clobbering existing bq-* headers and
respect the SQS 10-attribute limit. No header present falls back to v0.1
(trace_id-derived trace). Core gains an additive, zero-dep HeadersFromContext.
Redis stays deferred (no header side-channel). Other SDKs follow per-SDK MINOR.

Author: muhammetsafak

amqp/README.md

@@ -32,6 +32,17 @@ app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
 app.Consume(ctx)
 ```
 
+## Out-of-band transport headers
+
+This transport implements the optional `babelqueue.HeaderPublisher` capability:
+out-of-band headers ride in the AMQP message headers (`amqp091.Table`) **beside the
+frozen envelope, never in it** (GR-1), and are surfaced back to the consumer on
+`ReceivedMessage.Headers`. So the optional [`otel`](../otel) module's W3C
+`traceparent` (ADR-0028) — and the `bq-replay-bypass` marker (ADR-0027) —
+propagate over RabbitMQ for true cross-hop span parent-child linkage. The header is
+merged on top of the contract `x-*` headers without clobbering them (a contract
+header always wins a key collision).
+
 The BabelQueue core stays zero-dependency; this module is the only place
 `amqp091-go` is pulled in.
 

amqp/amqp.go

@@ -6,6 +6,12 @@
 // consumer can route on properties.type without parsing the body. Consuming uses
 // basic.get + manual ack (at-least-once), matching the PHP RabbitMQ driver.
 //
+// It also implements the optional [babelqueue.HeaderPublisher] capability: out-of-band
+// transport headers (e.g. a W3C traceparent for cross-hop span linkage, ADR-0028, or
+// the bq-replay-bypass marker, ADR-0027) ride in the AMQP message headers
+// (amqp091.Table) beside the frozen envelope — never in it (GR-1) — and are surfaced
+// back to the consumer on [babelqueue.ReceivedMessage.Headers].
+//
 //	tr := amqp.New("amqp://guest:guest@localhost:5672/")
 //	app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
 //
@@ -14,6 +20,7 @@ package amqp
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"time"
 
@@ -74,6 +81,20 @@ func (t *Transport) declare(ch *amqp091.Channel, queue string) error {
 // Publish declares the durable queue and publishes body with persistent delivery
 // and the contract AMQP properties.
 func (t *Transport) Publish(ctx context.Context, queue, body string) error {
+	return t.publish(ctx, queue, body, nil)
+}
+
+// PublishWithHeaders publishes body together with out-of-band transport headers
+// ([babelqueue.HeaderPublisher]). The headers ride in the AMQP message headers
+// (amqp091.Table) beside the frozen envelope (GR-1) — e.g. a W3C traceparent for
+// cross-hop span linkage (ADR-0028). They are merged on top of the contract x-*
+// headers without clobbering them (a contract header always wins a key collision),
+// and empty/blank keys are skipped. A nil/empty map behaves exactly like [Transport.Publish].
+func (t *Transport) PublishWithHeaders(ctx context.Context, queue, body string, headers map[string]string) error {
+	return t.publish(ctx, queue, body, headers)
+}
+
+func (t *Transport) publish(ctx context.Context, queue, body string, headers map[string]string) error {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	ch, err := t.ensure()
@@ -83,33 +104,84 @@ func (t *Transport) Publish(ctx context.Context, queue, body string) error {
 	if err := t.declare(ch, queue); err != nil {
 		return err
 	}
-	return ch.PublishWithContext(ctx, "", queue, false, false, t.publishing(body))
+	return ch.PublishWithContext(ctx, "", queue, false, false, t.publishing(body, headers))
 }
 
-func (t *Transport) publishing(body string) amqp091.Publishing {
+// publishing builds the AMQP message for body. extra carries optional out-of-band
+// transport headers that are merged into the message header table without overwriting
+// the contract x-* headers (the contract wins a key collision).
+func (t *Transport) publishing(body string, extra map[string]string) amqp091.Publishing {
 	pub := amqp091.Publishing{
 		ContentType:     "application/json",
 		ContentEncoding: "utf-8",
 		DeliveryMode:    amqp091.Persistent,
 		AppId:           "babelqueue",
 		Body:            []byte(body),
 	}
+	headers := amqp091.Table{}
+	mergeHeaders(headers, extra)
 	if env, err := babelqueue.Decode([]byte(body)); err == nil {
 		pub.Type = env.Job
 		pub.CorrelationId = env.TraceID
 		pub.MessageId = env.Meta.ID
-		headers := amqp091.Table{"x-attempts": env.Attempts}
+		// Contract x-* headers are written last so they win any key collision with an
+		// out-of-band header — the cross-language projection must never be clobbered.
+		headers["x-attempts"] = env.Attempts
 		if env.Meta.SchemaVersion != 0 {
 			headers["x-schema-version"] = env.Meta.SchemaVersion
 		}
 		if env.Meta.Lang != "" {
 			headers["x-source-lang"] = env.Meta.Lang
 		}
+	}
+	if len(headers) > 0 {
 		pub.Headers = headers
 	}
 	return pub
 }
 
+// mergeHeaders writes the out-of-band string headers into table as AMQP values,
+// skipping empty keys and values. It never removes or overwrites a key already in
+// table, so callers can seed contract headers afterwards and keep them authoritative.
+func mergeHeaders(table amqp091.Table, headers map[string]string) {
+	for k, v := range headers {
+		if k == "" || v == "" {
+			continue
+		}
+		if _, exists := table[k]; exists {
+			continue
+		}
+		table[k] = v
+	}
+}
+
+// headersFromTable maps an inbound AMQP header table to a flat map[string]string,
+// stringifying values defensively (AMQP headers are typed: strings, ints, bytes…).
+// It returns nil when no headers are present so a header-less delivery stays
+// header-less on [babelqueue.ReceivedMessage.Headers].
+func headersFromTable(table amqp091.Table) map[string]string {
+	if len(table) == 0 {
+		return nil
+	}
+	out := make(map[string]string, len(table))
+	for k, v := range table {
+		switch val := v.(type) {
+		case string:
+			out[k] = val
+		case []byte:
+			out[k] = string(val)
+		case nil:
+			// skip nil-valued headers
+		default:
+			out[k] = fmt.Sprintf("%v", val)
+		}
+	}
+	if len(out) == 0 {
+		return nil
+	}
+	return out
+}
+
 // Pop reserves the next message with basic.get (manual ack). When the queue is
 // empty it waits up to timeout (heartbeat-safe) and returns (nil, nil).
 func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error) {
@@ -139,7 +211,12 @@ func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration
 		}
 		return nil, nil
 	}
-	return &babelqueue.ReceivedMessage{Body: string(delivery.Body), Queue: queue, Handle: delivery}, nil
+	return &babelqueue.ReceivedMessage{
+		Body:    string(delivery.Body),
+		Queue:   queue,
+		Handle:  delivery,
+		Headers: headersFromTable(delivery.Headers),
+	}, nil
 }
 
 // Ack acknowledges the reserved delivery.
@@ -164,4 +241,7 @@ func (t *Transport) Close() error {
 	return nil
 }
 
-var _ babelqueue.Transport = (*Transport)(nil)
+var (
+	_ babelqueue.Transport       = (*Transport)(nil)
+	_ babelqueue.HeaderPublisher = (*Transport)(nil)
+)

amqp/amqp_test.go

@@ -58,6 +58,48 @@ func TestAMQPRoundTrip(t *testing.T) {
 	}
 }
 
+// TestAMQPCarriesTraceparentHeader is an integration test: it publishes a message
+// with an out-of-band traceparent header (HeaderPublisher) and asserts the live
+// broker delivers it back on ReceivedMessage.Headers, beside the unchanged envelope.
+// It skips cleanly when no RabbitMQ broker is reachable.
+func TestAMQPCarriesTraceparentHeader(t *testing.T) {
+	tr := bqamqp.New(brokerURL())
+	defer tr.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
+	defer cancel()
+
+	const queue = "bq-test-amqp-headers"
+	const traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+
+	env, err := babelqueue.Make("urn:babel:test:traced", map[string]any{"n": 1}, babelqueue.WithQueue(queue))
+	if err != nil {
+		t.Fatal(err)
+	}
+	body, _ := env.Encode()
+
+	if err := tr.PublishWithHeaders(ctx, queue, string(body), map[string]string{"traceparent": traceparent}); err != nil {
+		t.Skipf("rabbitmq not reachable: %v", err)
+	}
+
+	msg, err := tr.Pop(ctx, queue, time.Second)
+	if err != nil {
+		t.Fatalf("Pop: %v", err)
+	}
+	if msg == nil {
+		t.Fatal("expected a message")
+	}
+	defer func() { _ = tr.Ack(ctx, msg) }()
+
+	if msg.Headers["traceparent"] != traceparent {
+		t.Errorf("traceparent header = %q, want %q", msg.Headers["traceparent"], traceparent)
+	}
+	// The envelope must be byte-identical — the header rides out of band (GR-1).
+	if msg.Body != string(body) {
+		t.Errorf("envelope body changed: header propagation must not touch the body")
+	}
+}
+
 // TestAMQPAppEndToEnd runs a publish + drain through the App over RabbitMQ.
 func TestAMQPAppEndToEnd(t *testing.T) {
 	tr := bqamqp.New(brokerURL())

amqp/conformance_test.go

@@ -45,7 +45,7 @@ func TestRabbitMQConformance(t *testing.T) {
 		t.Fatalf("fixture: %v", err)
 	}
 
-	pub := (&Transport{}).publishing(string(body))
+	pub := (&Transport{}).publishing(string(body), nil)
 
 	checks := map[string]struct{ got, want string }{
 		"type":           {pub.Type, proj.Properties["type"]},

amqp/go.mod

@@ -3,7 +3,7 @@ module github.com/babelqueue/babelqueue-go/amqp
 go 1.21
 
 require (
-	github.com/babelqueue/babelqueue-go v1.0.0
+	github.com/babelqueue/babelqueue-go v1.3.0
 	github.com/rabbitmq/amqp091-go v1.10.0
 )
 

amqp/headers_test.go

@@ -0,0 +1,113 @@
+package amqp
+
+import (
+	"testing"
+
+	babelqueue "github.com/babelqueue/babelqueue-go"
+	amqp091 "github.com/rabbitmq/amqp091-go"
+)
+
+// TestPublishingMergesHeadersWithoutClobbering proves an out-of-band header
+// (traceparent) is written into the AMQP header table beside the contract x-*
+// headers, and that a contract header always wins a key collision.
+func TestPublishingMergesHeadersWithoutClobbering(t *testing.T) {
+	env, _ := babelqueue.Make("urn:babel:orders:created", map[string]any{"x": 1}, babelqueue.WithQueue("orders"))
+	body, _ := env.Encode()
+
+	tr := &Transport{}
+	pub := tr.publishing(string(body), map[string]string{
+		"traceparent":      "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
+		"tracestate":       "vendor=value",
+		"x-source-lang":    "ATTACKER", // must NOT overwrite the contract header
+		"":                 "blank-key-skipped",
+		"bq-replay-bypass": "1",
+	})
+
+	if got, _ := pub.Headers["traceparent"].(string); got != "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" {
+		t.Errorf("traceparent not carried: %q", got)
+	}
+	if got, _ := pub.Headers["tracestate"].(string); got != "vendor=value" {
+		t.Errorf("tracestate not carried: %q", got)
+	}
+	if got, _ := pub.Headers["bq-replay-bypass"].(string); got != "1" {
+		t.Errorf("bq-replay-bypass not carried: %q", got)
+	}
+	// The contract projection must survive the merge unchanged.
+	if got, _ := pub.Headers["x-source-lang"].(string); got != env.Meta.Lang {
+		t.Errorf("contract x-source-lang clobbered: got %q, want %q", got, env.Meta.Lang)
+	}
+	if v, _ := pub.Headers["x-schema-version"].(int); v != env.Meta.SchemaVersion {
+		t.Errorf("x-schema-version = %v, want %d", pub.Headers["x-schema-version"], env.Meta.SchemaVersion)
+	}
+	if _, exists := pub.Headers[""]; exists {
+		t.Error("blank header key must be skipped")
+	}
+}
+
+// TestPublishingNilHeadersIsByteForBytePlainPublish proves PublishWithHeaders with
+// no extra headers produces exactly the same message as a plain Publish would.
+func TestPublishingNilHeadersUnchanged(t *testing.T) {
+	env, _ := babelqueue.Make("urn:babel:orders:created", map[string]any{"x": 1})
+	body, _ := env.Encode()
+
+	tr := &Transport{}
+	plain := tr.publishing(string(body), nil)
+
+	if plain.Type != env.Job || plain.CorrelationId != env.TraceID || plain.MessageId != env.Meta.ID {
+		t.Error("contract properties changed under nil headers")
+	}
+	// Only the contract x-* headers should be present.
+	for k := range plain.Headers {
+		switch k {
+		case "x-attempts", "x-schema-version", "x-source-lang":
+		default:
+			t.Errorf("unexpected header %q on a plain publish", k)
+		}
+	}
+}
+
+func TestHeadersFromTable(t *testing.T) {
+	if got := headersFromTable(nil); got != nil {
+		t.Errorf("nil table must surface nil headers, got %v", got)
+	}
+	if got := headersFromTable(amqp091.Table{}); got != nil {
+		t.Errorf("empty table must surface nil headers, got %v", got)
+	}
+
+	got := headersFromTable(amqp091.Table{
+		"traceparent":      "00-trace-span-01",
+		"x-attempts":       3,             // ints stringify
+		"bq-replay-bypass": []byte("1"),   // byte slices stringify
+		"x-source-lang":    "go",
+		"nil-value":        nil, // skipped
+	})
+	if got["traceparent"] != "00-trace-span-01" {
+		t.Errorf("traceparent = %q", got["traceparent"])
+	}
+	if got["x-attempts"] != "3" {
+		t.Errorf("x-attempts (int) = %q, want \"3\"", got["x-attempts"])
+	}
+	if got["bq-replay-bypass"] != "1" {
+		t.Errorf("bq-replay-bypass ([]byte) = %q, want \"1\"", got["bq-replay-bypass"])
+	}
+	if _, ok := got["nil-value"]; ok {
+		t.Error("nil-valued header must be skipped")
+	}
+}
+
+// TestHeaderRoundTrip proves a traceparent injected on the produce side survives a
+// table → map[string]string extraction on the consume side, fully in-process.
+func TestHeaderRoundTrip(t *testing.T) {
+	const tp = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+	env, _ := babelqueue.Make("urn:babel:orders:created", map[string]any{"x": 1})
+	body, _ := env.Encode()
+
+	pub := (&Transport{}).publishing(string(body), map[string]string{"traceparent": tp})
+	// pub.Headers is what RabbitMQ would deliver back as delivery.Headers.
+	out := headersFromTable(pub.Headers)
+	if out["traceparent"] != tp {
+		t.Fatalf("round-trip lost traceparent: got %q, want %q", out["traceparent"], tp)
+	}
+}
+
+var _ babelqueue.HeaderPublisher = (*Transport)(nil)