fix: OTel trace propagation, /readyz health wiring, description update

OTel trace context (W3C traceparent) auto-propagated through gRPC
(PushRequest metadata), Kafka (message headers), and HTTP (request
headers). All gated behind #[cfg(feature = "otel")] — zero overhead
when disabled. Shared propagation helpers in transport/propagation.rs.

/readyz endpoint now aggregates from HealthRegistry when health feature
is enabled. /health/detailed returns per-component JSON status.
MetricsManager readiness also checks HealthRegistry.

Fix test ordering issue with HealthRegistry global state.
Update crate description. 671 tests pass.

Author: catinspace-au

Cargo.toml

@@ -11,7 +11,7 @@ name = "hyperi-rustlib"
 version = "1.17.0-dev.16"
 edition = "2024"
 rust-version = "1.94"
-description = "Shared utility library for HyperI Rust applications"
+description = "Opinionated Rust framework for high-throughput data pipelines at PB scale. Auto-wiring config, logging, metrics, tracing, health, and graceful shutdown — built from many years of production infrastructure experience."
 license = "FSL-1.1-ALv2"
 repository = "https://github.com/hyperi-io/hyperi-rustlib"
 publish = true

src/http_server/server.rs

@@ -196,6 +196,11 @@ impl HttpServer {
             );
         }
 
+        #[cfg(all(feature = "health", feature = "serde_json"))]
+        if self.config.enable_health_endpoints {
+            router = router.route("/health/detailed", get(health_detailed));
+        }
+
         #[cfg(feature = "config")]
         if self.config.enable_config_endpoint {
             router = router.route("/config", get(config_dump));
@@ -240,14 +245,35 @@ async fn health_live() -> impl IntoResponse {
 }
 
 /// Readiness endpoint handler.
+///
+/// Checks the local ready flag AND (when the `health` feature is enabled)
+/// the global [`HealthRegistry`](crate::health::HealthRegistry). Both must
+/// be true for a 200 response; otherwise 503.
 async fn health_ready(ready: Arc<AtomicBool>) -> impl IntoResponse {
-    if ready.load(Ordering::SeqCst) {
+    let locally_ready = ready.load(Ordering::SeqCst);
+
+    #[cfg(feature = "health")]
+    let registry_ready = crate::health::HealthRegistry::is_ready();
+    #[cfg(not(feature = "health"))]
+    let registry_ready = true;
+
+    if locally_ready && registry_ready {
         (StatusCode::OK, "OK")
     } else {
         (StatusCode::SERVICE_UNAVAILABLE, "NOT READY")
     }
 }
 
+/// Detailed health endpoint returning per-component status as JSON.
+///
+/// Returns the output of [`HealthRegistry::to_json()`](crate::health::HealthRegistry::to_json),
+/// which includes overall status and each registered component's state.
+#[cfg(all(feature = "health", feature = "serde_json"))]
+async fn health_detailed() -> impl IntoResponse {
+    let json = crate::health::HealthRegistry::to_json();
+    axum::Json(json)
+}
+
 /// Config registry dump endpoint handler (redacted).
 #[cfg(feature = "config")]
 async fn config_dump() -> impl IntoResponse {
@@ -332,6 +358,9 @@ mod tests {
 
     #[tokio::test]
     async fn test_health_ready_when_ready() {
+        #[cfg(feature = "health")]
+        crate::health::HealthRegistry::reset();
+
         let config = HttpServerConfig::default();
         let server = HttpServer::new(config);
         server.set_ready(true);

src/metrics/mod.rs

@@ -759,7 +759,14 @@ async fn handle_connection(
     } else if request_line.starts_with("GET /readyz")
         || request_line.starts_with("GET /health/ready")
     {
-        let ready = readiness_fn.as_ref().is_none_or(|f| f());
+        let callback_ready = readiness_fn.as_ref().is_none_or(|f| f());
+
+        #[cfg(feature = "health")]
+        let registry_ready = crate::health::HealthRegistry::is_ready();
+        #[cfg(not(feature = "health"))]
+        let registry_ready = true;
+
+        let ready = callback_ready && registry_ready;
         if ready {
             ("200 OK", r#"{"status":"ready"}"#.to_string())
         } else {
@@ -787,11 +794,22 @@ async fn handle_connection(
 }
 
 /// Readiness response helper for axum endpoints.
+///
+/// Checks the caller-supplied readiness callback AND (when the `health`
+/// feature is enabled) the global [`HealthRegistry`](crate::health::HealthRegistry).
+/// Both must be true for a 200 response.
 #[cfg(all(feature = "metrics", feature = "http-server"))]
 fn readiness_response(rf: Option<ReadinessFn>) -> axum::response::Response {
     use axum::response::IntoResponse;
 
-    let ready = rf.as_ref().is_none_or(|f| f());
+    let callback_ready = rf.as_ref().is_none_or(|f| f());
+
+    #[cfg(feature = "health")]
+    let registry_ready = crate::health::HealthRegistry::is_ready();
+    #[cfg(not(feature = "health"))]
+    let registry_ready = true;
+
+    let ready = callback_ready && registry_ready;
     if ready {
         (
             [(axum::http::header::CONTENT_TYPE, "application/json")],

src/transport/grpc/mod.rs

@@ -245,6 +245,12 @@ impl TransportSender for GrpcTransport {
             metadata.insert("topic".to_string(), key.to_string());
         }
 
+        // Inject W3C traceparent into gRPC metadata for distributed tracing
+        #[cfg(feature = "otel")]
+        if let Some(tp) = super::propagation::current_traceparent() {
+            metadata.insert(super::propagation::TRACEPARENT_HEADER.to_string(), tp);
+        }
+
         let request = proto::PushRequest {
             payload: payload.to_vec(),
             format: proto::Format::Auto.into(),
@@ -392,6 +398,14 @@ impl proto::dfe_transport_server::DfeTransport for DfeTransportServiceImpl {
         let req = request.into_inner();
         let seq = self.sequence.fetch_add(1, Ordering::Relaxed);
 
+        // Extract W3C traceparent from incoming gRPC metadata for distributed tracing
+        #[cfg(feature = "otel")]
+        if let Some(tp) = req.metadata.get(super::propagation::TRACEPARENT_HEADER)
+            && super::propagation::is_valid_traceparent(tp)
+        {
+            tracing::Span::current().record("traceparent", tp.as_str());
+        }
+
         let format = PayloadFormat::detect(&req.payload);
         let key = req.metadata.get("topic").map(|s| Arc::from(s.as_str()));
 

src/transport/http.rs

@@ -307,12 +307,27 @@ struct ReceiverState {
 async fn ingest_handler(
     axum::extract::State(state): axum::extract::State<ReceiverState>,
     axum::extract::ConnectInfo(addr): axum::extract::ConnectInfo<std::net::SocketAddr>,
+    headers: axum::http::HeaderMap,
     body: axum::body::Bytes,
 ) -> axum::http::StatusCode {
     if body.is_empty() {
         return axum::http::StatusCode::BAD_REQUEST;
     }
 
+    // Extract W3C traceparent from incoming HTTP headers for distributed tracing
+    #[cfg(feature = "otel")]
+    if let Some(tp) = headers
+        .get(super::propagation::TRACEPARENT_HEADER)
+        .and_then(|v| v.to_str().ok())
+        && super::propagation::is_valid_traceparent(tp)
+    {
+        tracing::Span::current().record("traceparent", tp);
+    }
+
+    // Suppress unused variable warning when otel feature is disabled
+    #[cfg(not(feature = "otel"))]
+    let _ = &headers;
+
     let seq = state.sequence.fetch_add(1, Ordering::Relaxed);
     let format = PayloadFormat::detect(&body);
     let timestamp_ms = chrono::Utc::now().timestamp_millis();
@@ -384,14 +399,20 @@ impl TransportSender for HttpTransport {
         #[cfg(feature = "metrics")]
         let start = std::time::Instant::now();
 
-        let result = match self
+        // Build request with optional W3C traceparent header for distributed tracing
+        let request_builder = self
             .client
             .post(&url)
-            .header("content-type", "application/octet-stream")
-            .body(payload.to_vec())
-            .send()
-            .await
-        {
+            .header("content-type", "application/octet-stream");
+
+        #[cfg(feature = "otel")]
+        let request_builder = if let Some(tp) = super::propagation::current_traceparent() {
+            request_builder.header(super::propagation::TRACEPARENT_HEADER, tp)
+        } else {
+            request_builder
+        };
+
+        let result = match request_builder.body(payload.to_vec()).send().await {
             Ok(resp) if resp.status().is_success() => {
                 #[cfg(feature = "logger")]
                 tracing::debug!(url = %url, bytes = payload.len(), "HTTP transport: POST sent");

src/transport/kafka/mod.rs

@@ -293,6 +293,18 @@ impl TransportSender for KafkaTransport {
 
         let record: FutureRecord<'_, str, [u8]> = FutureRecord::to(key).payload(payload);
 
+        // Inject W3C traceparent into Kafka message headers for distributed tracing
+        #[cfg(feature = "otel")]
+        let record = if let Some(tp) = super::propagation::current_traceparent() {
+            let headers = rdkafka::message::OwnedHeaders::new().insert(rdkafka::message::Header {
+                key: super::propagation::TRACEPARENT_HEADER,
+                value: Some(tp.as_str()),
+            });
+            record.headers(headers)
+        } else {
+            record
+        };
+
         #[cfg(feature = "metrics")]
         let start = std::time::Instant::now();
 
@@ -371,6 +383,26 @@ impl TransportReceiver for KafkaTransport {
         if let Some(result) = self.consumer.poll(timeout) {
             match result {
                 Ok(msg) => {
+                    // Extract W3C traceparent from Kafka headers (first message only,
+                    // to associate the batch span with the upstream trace)
+                    #[cfg(feature = "otel")]
+                    if let Some(headers) = msg.headers() {
+                        use rdkafka::message::Headers;
+                        for idx in 0..headers.count() {
+                            if let Some(Ok(header)) = headers.try_get_as::<[u8]>(idx)
+                                && header.key == super::propagation::TRACEPARENT_HEADER
+                            {
+                                if let Some(value) = header.value
+                                    && let Ok(tp) = std::str::from_utf8(value)
+                                    && super::propagation::is_valid_traceparent(tp)
+                                {
+                                    tracing::Span::current().record("traceparent", tp);
+                                }
+                                break;
+                            }
+                        }
+                    }
+
                     let topic_str = msg.topic();
                     let topic: Arc<str> = get_or_insert_topic(&mut local_cache, topic_str);
                     let payload = msg.payload().map_or_else(Vec::new, |p| p.to_vec());

src/transport/mod.rs

@@ -53,6 +53,7 @@ mod detect;
 mod error;
 pub mod factory;
 mod payload;
+pub mod propagation;
 mod traits;
 mod types;