Skip to content

Commit e253319

Browse files
committed
feat(redis): W3C traceparent transport-header wiring (completes ADR-0028)
Redis now carries traceparent via a transport-level frame (back-compatible with bare values; reliable-queue RPUSH/BLMOVE/LREM semantics intact), so all babelqueue-go transports (in-memory, AMQP, SQS, Redis) propagate the header. Also bump otel/amqp/sqs/redis module require on the core to v1.6.0 — the version that introduced the HeaderPublisher/Headers seam — so external consumers resolve a core that actually has it (the prior v1.0.0/v1.3.0 requires predated the seam).
1 parent 2c9e1b3 commit e253319

8 files changed

Lines changed: 447 additions & 9 deletions

File tree

amqp/go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ module github.com/babelqueue/babelqueue-go/amqp
33
go 1.21
44

55
require (
6-
github.com/babelqueue/babelqueue-go v1.3.0
6+
github.com/babelqueue/babelqueue-go v1.6.0
77
github.com/rabbitmq/amqp091-go v1.10.0
88
)
99

otel/go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ module github.com/babelqueue/babelqueue-go/otel
33
go 1.21
44

55
require (
6-
github.com/babelqueue/babelqueue-go v1.3.0
6+
github.com/babelqueue/babelqueue-go v1.6.0
77
go.opentelemetry.io/otel v1.24.0
88
go.opentelemetry.io/otel/sdk v1.24.0
99
go.opentelemetry.io/otel/trace v1.24.0

redis/README.md

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,4 +34,27 @@ Bring your own client with `redis.NewWithClient(client)` (custom pool / TLS / au
3434
The BabelQueue core stays zero-dependency; this module is the only place `go-redis`
3535
is pulled in.
3636

37+
## Out-of-band headers (`traceparent`, ADR-0028)
38+
39+
The transport implements the optional `babelqueue.HeaderPublisher` capability, so
40+
out-of-band transport headers — e.g. a W3C `traceparent` for cross-hop span linkage —
41+
ride **beside** the frozen wire envelope, never inside it. Redis stores only the raw
42+
list value (the `LREM` ack handle *is* that value), so there is no native per-message
43+
metadata channel; the transport instead owns a small JSON **frame**, distinct from the
44+
wire envelope:
45+
46+
```json
47+
{"__bq_frame":1,"headers":{"traceparent":"00-…-01"},"body":"<raw wire envelope>"}
48+
```
49+
50+
- `Publish` stores the **bare** envelope byte-for-byte (no frame).
51+
- `PublishWithHeaders` writes a **frame** only when the header map is non-empty;
52+
with no usable headers it degrades to a byte-identical bare `Publish`.
53+
- `Pop` detects frame-vs-bare by the reserved `__bq_frame` sentinel (the frozen
54+
envelope never carries it). A frame is unwrapped — `Body` is the raw envelope,
55+
`Headers` carries the frame's headers; the `Handle` stays the **stored frame**, so
56+
the reliable-queue `RPUSH`/`BLMOVE`/`LREM` semantics are untouched. A bare value
57+
(an older or non-otel publisher, or a cross-version queue) consumes exactly as before:
58+
`Headers` is nil and the handle is the bare value.
59+
3760
Full spec: **[babelqueue.com](https://babelqueue.com)** · [MIT](../LICENSE) © Muhammet Şafak

redis/frame_test.go

Lines changed: 189 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,189 @@
1+
package redis
2+
3+
import (
4+
"encoding/json"
5+
"testing"
6+
)
7+
8+
// These are pure unit tests for the header-frame logic (ADR-0028). They need no broker:
9+
// they exercise the frame marshal/unframe round-trip, bare-value back-compat, and the
10+
// Ack-handle invariant directly against the unexported helpers, so the framing is fully
11+
// covered without a live Redis.
12+
13+
// bareEnvelope is a minimal stand-in for a frozen wire envelope value. The only property
14+
// that matters here is that it is a JSON object WITHOUT the reserved "__bq_frame"
15+
// sentinel, so unframe must treat it as bare.
16+
const bareEnvelope = `{"job":"urn:babel:test:ping","data":{"n":1},"meta":{"id":"abc"}}`
17+
18+
func TestUnframeBareEnvelopePassesThrough(t *testing.T) {
19+
body, headers := unframe(bareEnvelope)
20+
if body != bareEnvelope {
21+
t.Errorf("body = %q, want the value verbatim", body)
22+
}
23+
if headers != nil {
24+
t.Errorf("headers = %v, want nil for a bare value", headers)
25+
}
26+
}
27+
28+
func TestUnframeNonJSONPassesThrough(t *testing.T) {
29+
for _, v := range []string{"", "not json", "[1,2,3]", `"a string"`, "42"} {
30+
body, headers := unframe(v)
31+
if body != v {
32+
t.Errorf("unframe(%q) body = %q, want verbatim", v, body)
33+
}
34+
if headers != nil {
35+
t.Errorf("unframe(%q) headers = %v, want nil", v, headers)
36+
}
37+
}
38+
}
39+
40+
func TestUnframeJSONObjectWithoutSentinelIsBare(t *testing.T) {
41+
// A JSON object that happens to have a "headers" / "body" shape but no sentinel must
42+
// NOT be mistaken for a frame — only the reserved key marks a frame.
43+
v := `{"headers":{"x":"y"},"body":"spoof"}`
44+
body, headers := unframe(v)
45+
if body != v {
46+
t.Errorf("body = %q, want verbatim (no sentinel = bare)", body)
47+
}
48+
if headers != nil {
49+
t.Errorf("headers = %v, want nil", headers)
50+
}
51+
}
52+
53+
func TestFrameRoundTrip(t *testing.T) {
54+
in := map[string]string{
55+
"traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
56+
"tracestate": "vendor=value",
57+
}
58+
raw, err := json.Marshal(headerFrame{Version: frameVersion, Headers: in, Body: bareEnvelope})
59+
if err != nil {
60+
t.Fatal(err)
61+
}
62+
63+
body, headers := unframe(string(raw))
64+
if body != bareEnvelope {
65+
t.Errorf("body = %q, want the unframed wire envelope %q", body, bareEnvelope)
66+
}
67+
if len(headers) != len(in) {
68+
t.Fatalf("headers = %v, want %v", headers, in)
69+
}
70+
for k, v := range in {
71+
if headers[k] != v {
72+
t.Errorf("headers[%q] = %q, want %q", k, headers[k], v)
73+
}
74+
}
75+
}
76+
77+
func TestSanitizeHeadersDropsBlanks(t *testing.T) {
78+
got := sanitizeHeaders(map[string]string{
79+
"traceparent": "00-trace-01",
80+
"": "no-key",
81+
"empty-val": "",
82+
})
83+
if len(got) != 1 || got["traceparent"] != "00-trace-01" {
84+
t.Errorf("sanitizeHeaders = %v, want only the non-blank traceparent", got)
85+
}
86+
if got := sanitizeHeaders(nil); got != nil {
87+
t.Errorf("sanitizeHeaders(nil) = %v, want nil", got)
88+
}
89+
if got := sanitizeHeaders(map[string]string{"": ""}); got != nil {
90+
t.Errorf("sanitizeHeaders(all-blank) = %v, want nil", got)
91+
}
92+
}
93+
94+
func TestFrameValueWithoutHeadersStaysBare(t *testing.T) {
95+
// Plain Publish stores `body`; PublishWithHeaders with nil/empty/blank headers must
96+
// store the byte-identical bare value so nothing regresses and cross-version queues
97+
// interoperate.
98+
for name, headers := range map[string]map[string]string{
99+
"nil": nil,
100+
"empty": {},
101+
"all-blank": {"": "", "k": ""},
102+
} {
103+
got, err := frameValue(bareEnvelope, headers)
104+
if err != nil {
105+
t.Fatalf("%s: %v", name, err)
106+
}
107+
if got != bareEnvelope {
108+
t.Errorf("%s: frameValue = %q, want byte-identical bare body %q", name, got, bareEnvelope)
109+
}
110+
}
111+
}
112+
113+
func TestFrameValueWithHeadersIsAFrame(t *testing.T) {
114+
got, err := frameValue(bareEnvelope, map[string]string{"traceparent": "00-trace-01"})
115+
if err != nil {
116+
t.Fatal(err)
117+
}
118+
if got == bareEnvelope {
119+
t.Fatal("with headers the stored value must be a frame, not the bare body")
120+
}
121+
if !containsSentinel(got) {
122+
t.Errorf("frame %q is missing the __bq_frame sentinel", got)
123+
}
124+
}
125+
126+
// TestStoredValueAckHandleInvariant models the full RPUSH→BLMOVE→LREM cycle without a
127+
// broker: frameValue is what RPUSH stores and what BLMOVE returns (the Pop Handle), and
128+
// Ack's LREM matches on that exact value. So the handle must equal the stored value
129+
// byte-for-byte in BOTH the framed and bare cases, and unframe must recover the right
130+
// body/headers from it.
131+
func TestStoredValueAckHandleInvariant(t *testing.T) {
132+
cases := []struct {
133+
name string
134+
headers map[string]string
135+
wantHeaders map[string]string
136+
}{
137+
{"bare (no headers)", nil, nil},
138+
{"framed (traceparent)", map[string]string{"traceparent": "00-trace-01"}, map[string]string{"traceparent": "00-trace-01"}},
139+
}
140+
for _, c := range cases {
141+
t.Run(c.name, func(t *testing.T) {
142+
stored, err := frameValue(bareEnvelope, c.headers) // what RPUSH writes
143+
if err != nil {
144+
t.Fatal(err)
145+
}
146+
// BLMOVE returns `stored`; Pop sets Handle = stored, then unframes it.
147+
handle := stored
148+
body, headers := unframe(stored)
149+
if body != bareEnvelope {
150+
t.Errorf("body = %q, want %q", body, bareEnvelope)
151+
}
152+
if len(headers) != len(c.wantHeaders) {
153+
t.Errorf("headers = %v, want %v", headers, c.wantHeaders)
154+
}
155+
for k, v := range c.wantHeaders {
156+
if headers[k] != v {
157+
t.Errorf("headers[%q] = %q, want %q", k, headers[k], v)
158+
}
159+
}
160+
// LREM matches on the handle, which MUST equal the stored value.
161+
if handle != stored {
162+
t.Errorf("Ack handle %q != stored value %q — LREM would not match", handle, stored)
163+
}
164+
})
165+
}
166+
}
167+
168+
// TestFrameBodyIsNotTheWireEnvelope locks GR-1: the wire envelope is never mutated by
169+
// framing. The frame is a separate JSON object whose "body" field holds the verbatim
170+
// envelope; the envelope bytes themselves are untouched.
171+
func TestFrameBodyIsNotTheWireEnvelope(t *testing.T) {
172+
raw, err := json.Marshal(headerFrame{
173+
Version: frameVersion,
174+
Headers: map[string]string{"traceparent": "00-trace-01"},
175+
Body: bareEnvelope,
176+
})
177+
if err != nil {
178+
t.Fatal(err)
179+
}
180+
// The stored frame must NOT be the bare envelope (it wraps it) ...
181+
if string(raw) == bareEnvelope {
182+
t.Fatal("frame must differ from the bare envelope")
183+
}
184+
// ... and unframing must reproduce the exact original envelope bytes.
185+
body, _ := unframe(string(raw))
186+
if body != bareEnvelope {
187+
t.Errorf("unframed body = %q, want the original envelope %q", body, bareEnvelope)
188+
}
189+
}

redis/go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ module github.com/babelqueue/babelqueue-go/redis
33
go 1.21
44

55
require (
6-
github.com/babelqueue/babelqueue-go v1.0.0
6+
github.com/babelqueue/babelqueue-go v1.6.0
77
github.com/redis/go-redis/v9 v9.7.3
88
)
99

0 commit comments

Comments
 (0)