feat(otel): carry W3C traceparent via a Redis __bq_frame (ADR-0028)

Implements the babelqueue-core 1.5.0 header seam so traceparent rides the broker
beside the frozen envelope (GR-1), merge-not-clobber + (Redis) bare-value back-
compat; no header falls back to the v0.1 trace_id mapping. Core dep bumped to
1.5.0; package bumped to 1.1.0.

Author: muhammetsafak

CHANGELOG.md

@@ -7,6 +7,31 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 The envelope wire format is versioned separately by `meta.schema_version`
 (currently **1**) — see the contract at [babelqueue.com](https://babelqueue.com).
 
+## [Unreleased]
+
+### Added
+- **OpenTelemetry `traceparent` transport wiring (ADR-0028, v0.2).** Redis lists have no
+  native per-message metadata channel and the reliable-queue stores the bare body (the
+  `LREM` ack handle *is* the stored value), so to carry an out-of-band header (e.g. a W3C
+  `traceparent`) beside the frozen envelope the transport now uses a self-contained,
+  transport-owned JSON frame distinct from the wire envelope:
+  `{"__bq_frame":1,"headers":…,"body":<raw envelope>}` (`RedisFrame`, mirroring the Go/PHP
+  `__bq_frame`). `RedisPublisher.publishWithHeaders(Envelope, Map<String,String>)` is the
+  produce-side seam the optional core `com.babelqueue.otel.HeaderSender` wires to: with
+  usable headers it `RPUSH`es the frame; with none it `RPUSH`es the bare envelope
+  byte-for-byte (back-compat). `RedisConsumer` transparently unframes a reserved value —
+  bare-value back-compat detection via the reserved `__bq_frame` sentinel (a frozen
+  envelope never carries it), the `LREM` ack handle stays the stored value — and surfaces
+  the headers to a new header-aware `RedisConsumer.HeaderHandler` (`(envelope, body,
+  headers)`), the consume-side seam for `Tracing.wrapHandler(tracer, handler, Supplier)`.
+  The existing `(envelope, body)` `BabelHandler` registration is unchanged. The wire
+  envelope is never touched (GR-1); `trace_id` is preserved (GR-4); `schema_version` stays
+  **1**. No new runtime dependency — the frame codec is self-contained and the header seam
+  is a plain `Map<String,String>` (GR-7).
+
+### Changed
+- Require `com.babelqueue:babelqueue-core 1.5.0` (the out-of-band header-carrier seam).
+
 ## [1.0.0] - 2026-06-14
 
 ### Added

README.md

@@ -75,6 +75,34 @@ The envelope is unchanged (`schema_version` stays `1`); Redis is purely additive
 > consumer reading a queue produced by the Laravel driver must replicate Laravel's
 > reserve/ack semantics.
 
+## Trace propagation (OpenTelemetry `traceparent`, ADR-0028)
+
+Redis lists carry no native metadata channel, so to propagate a W3C `traceparent` (out of
+band, beside the frozen envelope — GR-1) the transport wraps the bare envelope in a small,
+transport-owned JSON frame distinct from the wire envelope:
+`{"__bq_frame":1,"headers":…,"body":<raw envelope>}` (the same `__bq_frame` Go and PHP
+use). The optional core `com.babelqueue.otel` module drives it:
+
+```java
+// produce: HeaderSender -> RedisPublisher.publishWithHeaders (frames only when headers exist)
+RedisPublisher publisher = RedisPublisher.create(redis, "orders");
+Tracing.publish(tracer, "urn:babel:orders:created", Map.of("order_id", 1042), "orders",
+    (envelope, headers) -> publisher.publishWithHeaders(envelope, headers));
+
+// consume: a header-aware handler receives the surfaced headers for wrapHandler's Supplier
+RedisConsumer.builder(redis, "orders")
+    .handler("urn:babel:orders:created", (env, body, headers) ->
+        Tracing.wrapHandler(tracer, h, () -> headers).handle(env))
+    .build();
+```
+
+A header-less `publish(...)` stores the bare envelope **byte-for-byte**; a bare
+(un-framed) value still consumes — the consumer detects a frame by the reserved
+`__bq_frame` sentinel (a frozen envelope never carries it), and the `LREM` ack handle
+stays the stored value, so cross-version queues interoperate. The existing
+`(env, body)` handler keeps working. Requires `babelqueue-core` ≥ 1.5.0; no new runtime
+dependency (the frame codec is self-contained, the seam is a plain `Map<String,String>`).
+
 ## Build & test
 
 ```bash

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.babelqueue</groupId>
     <artifactId>babelqueue-redis</artifactId>
-    <version>1.0.0</version>
+    <version>1.1.0</version>
     <packaging>jar</packaging>
 
     <name>BabelQueue Redis</name>
@@ -49,12 +49,25 @@
         <maven.compiler.release>17</maven.compiler.release>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
-        <babelqueue-core.version>1.0.0</babelqueue-core.version>
+        <babelqueue-core.version>1.5.0</babelqueue-core.version>
         <lettuce.version>6.5.5.RELEASE</lettuce.version>
+        <opentelemetry.version>1.45.0</opentelemetry.version>
         <junit.version>5.10.3</junit.version>
         <mockito.version>5.12.0</mockito.version>
     </properties>
 
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>io.opentelemetry</groupId>
+                <artifactId>opentelemetry-bom</artifactId>
+                <version>${opentelemetry.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
     <dependencies>
         <!-- The framework-agnostic core: the one codec + contracts. Transitive. -->
         <dependency>
@@ -89,6 +102,23 @@
             <version>20240303</version>
             <scope>test</scope>
         </dependency>
+        <!--
+          The core's OTel module is optional (not transitive), so the end-to-end
+          traceparent cross-hop test (Tracing.publish HeaderSender -> wrapHandler
+          Supplier over the __bq_frame) pulls the SDK + in-memory exporter in test scope.
+          Runtime use of this transport needs no OTel dependency at all — the header
+          seam is a plain Map<String,String> and the frame codec is self-contained.
+        -->
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk-testing</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/com/babelqueue/redis/JsonLite.java

@@ -0,0 +1,141 @@
+package com.babelqueue.redis;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * A minimal, dependency-free reader for the Redis transport's own {@code __bq_frame}
+ * header frame (ADR-0028) — and nothing more. The frame is a flat JSON object whose values
+ * are exactly: an integer (the {@code __bq_frame} version), a string (the {@code body},
+ * the raw wire envelope), and a nested string→string object (the {@code headers}). This
+ * parser supports only that shape; it is never used on the wire envelope, which the core
+ * codec owns. Any input it cannot parse — including a perfectly valid but differently
+ * shaped JSON document — yields {@code null}, so {@link RedisFrame#unframe} falls back to
+ * treating the value as a bare envelope.
+ *
+ * <p>Keeping the grammar this narrow keeps the transport zero-dependency (GR-7) without
+ * carrying a general-purpose JSON parser's surface.
+ */
+final class JsonLite {
+
+    private final String s;
+    private int pos;
+
+    private JsonLite(String s) {
+        this.s = s;
+    }
+
+    /**
+     * Parses {@code raw} as a frame object: a JSON object mapping string keys to a string,
+     * an integer, or a nested string→string object. Returns the map, or {@code null} on any
+     * deviation from that shape.
+     */
+    static Map<String, Object> parseFrame(String raw) {
+        try {
+            JsonLite p = new JsonLite(raw);
+            Map<String, Object> obj = p.parseObject(true);
+            return p.atEnd() ? obj : null;
+        } catch (RuntimeException e) {
+            return null;
+        }
+    }
+
+    /** Parses an object; {@code topLevel} values may be string/int/object, nested values string only. */
+    private Map<String, Object> parseObject(boolean topLevel) {
+        expect('{');
+        Map<String, Object> obj = new LinkedHashMap<>();
+        if (peek() == '}') {
+            pos++;
+            return obj;
+        }
+        while (true) {
+            String key = parseString();
+            expect(':');
+            obj.put(key, topLevel ? parseTopValue() : parseString());
+            char c = next();
+            if (c == '}') {
+                return obj;
+            }
+            if (c != ',') {
+                throw new IllegalStateException("expected , or }");
+            }
+        }
+    }
+
+    /** A top-level frame value is a string, a non-negative integer, or a nested string→string object. */
+    private Object parseTopValue() {
+        char c = peek();
+        if (c == '"') {
+            return parseString();
+        }
+        if (c == '{') {
+            return parseObject(false);
+        }
+        return parseInt();
+    }
+
+    private String parseString() {
+        expect('"');
+        StringBuilder sb = new StringBuilder();
+        while (true) {
+            char c = next();
+            if (c == '"') {
+                return sb.toString();
+            }
+            if (c == '\\') {
+                char e = next();
+                switch (e) {
+                    case '"' -> sb.append('"');
+                    case '\\' -> sb.append('\\');
+                    case '/' -> sb.append('/');
+                    case 'n' -> sb.append('\n');
+                    case 'r' -> sb.append('\r');
+                    case 't' -> sb.append('\t');
+                    case 'b' -> sb.append('\b');
+                    case 'f' -> sb.append('\f');
+                    case 'u' -> {
+                        sb.append((char) Integer.parseInt(s.substring(pos, pos + 4), 16));
+                        pos += 4;
+                    }
+                    default -> throw new IllegalStateException("bad escape");
+                }
+            } else {
+                sb.append(c);
+            }
+        }
+    }
+
+    private Long parseInt() {
+        int start = pos;
+        while (pos < s.length() && s.charAt(pos) >= '0' && s.charAt(pos) <= '9') {
+            pos++;
+        }
+        if (pos == start) {
+            throw new IllegalStateException("expected integer");
+        }
+        return Long.parseLong(s.substring(start, pos));
+    }
+
+    private boolean atEnd() {
+        return pos == s.length();
+    }
+
+    private char peek() {
+        if (pos >= s.length()) {
+            throw new IllegalStateException("unexpected end of input");
+        }
+        return s.charAt(pos);
+    }
+
+    private char next() {
+        char c = peek();
+        pos++;
+        return c;
+    }
+
+    private void expect(char c) {
+        if (next() != c) {
+            throw new IllegalStateException("expected " + c);
+        }
+    }
+}