add trace contex, el headers and failure replay

README.md

@@ -54,6 +54,11 @@ port = 3000
 # Ethereum execution layer JSON-RPC endpoint (required)
 el_endpoint = "http://localhost:8545"
 
+# Optional HTTP headers applied to every EL JSON-RPC request (e.g. authentication).
+# Header names are case-insensitive; names differing only in case are rejected.
+# [el_headers]
+# Authorization = "Bearer <token>"
+
 # Optional local EL chain config JSON file.
 # chain_config_path = "path/to/chain_config.json"
 
@@ -129,6 +134,13 @@ The following endpoints are available:
 | `GET`  | `/health`                                                      | Health check                                             |
 | `GET`  | `/metrics`                                                     | Prometheus metrics                                       |
 
+Proof events on the SSE stream are delivered at-least-once with latest-wins semantics:
+terminal results (completions and failures) that happened before subscribing are replayed
+from bounded caches. Admitting a retry evicts its previous failure, although a concurrent
+subscription that already snapshotted that failure may still deliver it before the retry's
+terminal event. Clients should treat the most recent terminal event per
+`(new_payload_request_root, proof_type)` as authoritative.
+
 See [openapi.json](openapi.json) for the full API specification ([rendered](https://petstore.swagger.io/?url=https://raw.githubusercontent.com/eth-act/zkboost/master/openapi.json)).
 
 ## Observability

crates/client/Cargo.toml

@@ -8,16 +8,40 @@ license.workspace = true
 [lints]
 workspace = true
 
+[features]
+default = []
+otel = [
+    "dep:opentelemetry",
+    "dep:tracing",
+    "dep:tracing-opentelemetry",
+]
+
 [dependencies]
 async-stream.workspace = true
 bytes.workspace = true
 futures.workspace = true
+opentelemetry = { workspace = true, optional = true }
 reqwest = { workspace = true, features = ["json"] }
 reqwest-eventsource.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
 tokio-stream.workspace = true
+tracing = { workspace = true, optional = true }
+tracing-opentelemetry = { workspace = true, optional = true }
 url.workspace = true
 
 zkboost-types.workspace = true
+
+[dev-dependencies]
+axum.workspace = true
+opentelemetry.workspace = true
+opentelemetry_sdk.workspace = true
+tokio = { workspace = true, features = ["macros", "net", "rt"] }
+tracing.workspace = true
+tracing-opentelemetry.workspace = true
+tracing-subscriber = { workspace = true, features = ["registry"] }
+
+[[test]]
+name = "otel_propagation"
+required-features = ["otel"]

crates/client/src/lib.rs

@@ -8,6 +8,25 @@
 //! - [`get_proof`](zkBoostClient::get_proof) - download completed proof bytes
 //! - [`verify_proof`](zkBoostClient::verify_proof) - verify a proof against the server
 //!
+//! # Distributed tracing
+//!
+//! With the `otel` cargo feature enabled, every outbound request carries the current
+//! `tracing` span's W3C trace context (`traceparent`/`tracestate`), injected via the global
+//! OpenTelemetry propagator, so proofs requested from an instrumented caller join its
+//! distributed trace. Without the feature no OpenTelemetry dependency is pulled in and requests
+//! are sent unchanged.
+//!
+//! Note that the global propagator defaults to a no-op: the calling application must install
+//! one, e.g.
+//!
+//! ```ignore
+//! opentelemetry::global::set_text_map_propagator(
+//!     opentelemetry_sdk::propagation::TraceContextPropagator::new(),
+//! );
+//! ```
+//!
+//! otherwise no `traceparent` header is emitted even with the feature enabled.
+//!
 //! # Example
 //!
 //! ```ignore
@@ -22,7 +41,7 @@
 //! # }
 //! ```
 
-#![warn(unused_crate_dependencies)]
+#![cfg_attr(not(test), warn(unused_crate_dependencies))]
 
 pub mod error;
 
@@ -47,6 +66,43 @@ pub use {
 
 const APPLICATION_OCTET_STREAM: &str = "application/octet-stream";
 
+/// [`Injector`](opentelemetry::propagation::Injector) over HTTP headers, used to inject W3C trace
+/// context (`traceparent`/`tracestate`) into outbound requests.
+#[cfg(feature = "otel")]
+struct HeaderInjector<'a>(&'a mut reqwest::header::HeaderMap);
+
+#[cfg(feature = "otel")]
+impl opentelemetry::propagation::Injector for HeaderInjector<'_> {
+    fn set(&mut self, key: &str, value: String) {
+        if let Ok(name) = reqwest::header::HeaderName::from_bytes(key.as_bytes())
+            && let Ok(value) = reqwest::header::HeaderValue::from_str(&value)
+        {
+            self.0.insert(name, value);
+        }
+    }
+}
+
+/// Returns headers carrying the current `tracing` span's W3C trace context
+/// (`traceparent`/`tracestate`), injected via the global OpenTelemetry propagator.
+///
+/// Without the `otel` feature this returns an empty map, so requests are sent unchanged. With
+/// the feature, the map is still empty unless the application has installed a global text-map
+/// propagator (the OpenTelemetry default is a no-op).
+fn trace_context_headers() -> reqwest::header::HeaderMap {
+    #[cfg_attr(not(feature = "otel"), expect(unused_mut))]
+    let mut headers = reqwest::header::HeaderMap::new();
+    #[cfg(feature = "otel")]
+    {
+        use tracing_opentelemetry::OpenTelemetrySpanExt;
+
+        let context = tracing::Span::current().context();
+        opentelemetry::global::get_text_map_propagator(|propagator| {
+            propagator.inject_context(&context, &mut HeaderInjector(&mut headers));
+        });
+    }
+    headers
+}
+
 /// HTTP client for the zkboost Proof Node API.
 #[derive(Debug, Clone)]
 #[allow(non_camel_case_types)]
@@ -65,6 +121,9 @@ impl zkBoostClient {
     }
 
     /// Creates a new client with a custom [`reqwest::Client`].
+    ///
+    /// Use this to customize transport behavior, e.g. attach extra headers to every request via
+    /// [`reqwest::ClientBuilder::default_headers`] (for authenticated endpoints).
     pub fn with_http_client(endpoint: Url, http_client: reqwest::Client) -> Self {
         Self {
             endpoint,
@@ -93,6 +152,7 @@ impl zkBoostClient {
         let response = self
             .http_client
             .post(url)
+            .headers(trace_context_headers())
             .header(CONTENT_TYPE, APPLICATION_OCTET_STREAM)
             .body(body.to_ssz())
             .send()
@@ -111,14 +171,17 @@ impl zkBoostClient {
         &self,
         filter_root: Option<Hash256>,
     ) -> impl Stream<Item = Result<ProofEvent, Error>> + Send + '_ {
+        // Capture the trace context eagerly so the subscription joins the span current at call
+        // time, not whichever span happens to be current when the stream is first polled.
+        let trace_headers = trace_context_headers();
         async_stream::try_stream! {
             let mut url = self.endpoint.join("/v1/execution_proof_requests")?;
             if let Some(new_payload_request_root) = filter_root {
                 url.query_pairs_mut()
                     .append_pair("new_payload_request_root", &new_payload_request_root.to_string());
             }
 
-            let builder = self.http_client.get(url);
+            let builder = self.http_client.get(url).headers(trace_headers);
             let mut es = EventSource::new(builder)
                 .map_err(|e| Error::Sse(format!("failed to create event source: {e}")))?;
 
@@ -150,7 +213,8 @@ impl zkBoostClient {
             "/v1/execution_proofs/{new_payload_request_root}/{proof_type}"
         ))?;
 
-        let response = error_for_status(self.http_client.get(url).send().await?).await?;
+        let request = self.http_client.get(url).headers(trace_context_headers());
+        let response = error_for_status(request.send().await?).await?;
         Ok(response.bytes().await?)
     }
 
@@ -176,6 +240,7 @@ impl zkBoostClient {
         let response = self
             .http_client
             .post(url)
+            .headers(trace_context_headers())
             .header(CONTENT_TYPE, APPLICATION_OCTET_STREAM)
             .body(body.to_ssz())
             .send()

crates/client/tests/otel_propagation.rs

@@ -0,0 +1,102 @@
+//! Tests that outbound client requests carry the calling span's W3C trace context
+//! (`traceparent`) when the `otel` feature is enabled.
+
+use std::sync::{Arc, Mutex};
+
+use axum::{Router, extract::State, http::HeaderMap};
+use opentelemetry::trace::{TraceContextExt, TracerProvider};
+use opentelemetry_sdk::{propagation::TraceContextPropagator, trace::SdkTracerProvider};
+use tokio_stream::StreamExt;
+use tracing::Instrument;
+use tracing_opentelemetry::OpenTelemetrySpanExt;
+use tracing_subscriber::layer::SubscriberExt;
+use zkboost_client::{Hash256, ProofType, zkBoostClient};
+
+/// `traceparent` header values captured by the local server, one per request received.
+type CapturedTraceparents = Arc<Mutex<Vec<Option<String>>>>;
+
+async fn capture_traceparent(State(captured): State<CapturedTraceparents>, headers: HeaderMap) {
+    let traceparent = headers
+        .get("traceparent")
+        .and_then(|value| value.to_str().ok())
+        .map(str::to_owned);
+    captured.lock().unwrap().push(traceparent);
+}
+
+/// Extracts the trace id from a W3C `traceparent` header (`00-<trace_id>-<span_id>-<flags>`).
+fn trace_id_of(traceparent: &str) -> &str {
+    traceparent
+        .split('-')
+        .nth(1)
+        .expect("traceparent should have a trace id field")
+}
+
+/// Sends requests from within tracing spans and asserts each request arrives with a
+/// `traceparent` header carrying the active span's trace id.
+#[tokio::test]
+async fn test_outbound_requests_carry_active_span_trace_context() {
+    opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new());
+    let provider = SdkTracerProvider::builder().build();
+    let subscriber = tracing_subscriber::registry().with(
+        tracing_opentelemetry::OpenTelemetryLayer::new(provider.tracer("test")),
+    );
+    let _guard = tracing::subscriber::set_default(subscriber);
+
+    // Local server capturing the `traceparent` header of every request.
+    let captured = CapturedTraceparents::default();
+    let app = Router::new()
+        .fallback(capture_traceparent)
+        .with_state(captured.clone());
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let endpoint = format!("http://{}", listener.local_addr().unwrap());
+    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+
+    let client = zkBoostClient::new(endpoint.parse().unwrap());
+
+    // Unary request: `get_proof` awaited inside a span.
+    let unary_span = tracing::info_span!("unary_caller");
+    let unary_trace_id = unary_span
+        .context()
+        .span()
+        .span_context()
+        .trace_id()
+        .to_string();
+    assert_ne!(
+        unary_trace_id, "00000000000000000000000000000000",
+        "span should carry a valid OpenTelemetry context"
+    );
+    client
+        .get_proof(Hash256::ZERO, ProofType::RethZisk)
+        .instrument(unary_span)
+        .await
+        .unwrap();
+
+    // SSE subscription: the context is captured when `subscribe_proof_events` is called, even
+    // though the request is only sent once the stream is polled (outside the span, here). The
+    // plain 200 response fails SSE content-type validation, which is fine: the request has
+    // already reached the server with its headers by then.
+    let sse_span = tracing::info_span!("sse_caller");
+    let sse_trace_id = sse_span
+        .context()
+        .span()
+        .span_context()
+        .trace_id()
+        .to_string();
+    let stream = {
+        let _entered = sse_span.enter();
+        client.subscribe_proof_events(None)
+    };
+    let mut stream = std::pin::pin!(stream);
+    let _ = stream.next().await;
+
+    let captured = captured.lock().unwrap();
+    assert_eq!(captured.len(), 2);
+    let unary_traceparent = captured[0]
+        .as_deref()
+        .expect("get_proof should send traceparent");
+    assert_eq!(trace_id_of(unary_traceparent), unary_trace_id);
+    let sse_traceparent = captured[1]
+        .as_deref()
+        .expect("subscribe should send traceparent");
+    assert_eq!(trace_id_of(sse_traceparent), sse_trace_id);
+}

crates/server/Cargo.toml

@@ -77,6 +77,11 @@ zkboost-types.workspace = true
 
 [dev-dependencies]
 futures.workspace = true
+opentelemetry_sdk = { workspace = true, features = ["testing"] }
 
 # local
 zkboost-client.workspace = true
+
+[[test]]
+name = "otel_request_span"
+required-features = ["otel"]