add trace contex, el headers and failure replay (#70)

* feat(server): extract W3C trace context from incoming HTTP requests

Replace the default TraceLayer span with a custom MakeSpan that, when the
otel feature is enabled, extracts traceparent/tracestate from request
headers via the global propagator and sets the result as the request
span's parent. The whole existing explicit-parent chain (request span ->
handler -> request_proof -> fetch_witness / prove) now joins the caller's
distributed trace. The request span is also promoted from the debug-level
tower-http default to info level so it survives info-level filters.

Without the otel feature the custom MakeSpan only creates the span, with
no OpenTelemetry dependency.

Verified by a new otel-gated test using an in-memory span exporter that
asserts the exported request span keeps the caller's trace id and is
parented under the caller's span id.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(server): record cluster job_id on the prove span and in logs

Declare an empty job_id field on the per-attempt prove span and pass that
span explicitly through zkVMInstance::prove into the zisk cluster backend,
which records the id on it as soon as the cluster assigns one, so traces
correlate a prove span with the cluster-side job. Recording on the
explicitly passed span rather than Span::current() means the id cannot
silently land elsewhere if an intermediate span is ever introduced.

Also emit an info log with the job id at creation so the correlation is
available in plain logs without the otel feature.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(server): support custom HTTP headers for EL JSON-RPC requests

Add an optional top-level el_headers map (default empty) next to
el_endpoint, applied as reqwest default headers when constructing the
ElClient, so every EL JSON-RPC request (chain config, blocks, witnesses)
carries them, e.g. for authenticated endpoints.

Since header values typically carry credentials they are kept out of any
Debug output: the parsed HeaderValues are marked sensitive, and Config
implements Debug manually so el_headers values format as "<redacted>".

Config validation fails at startup on invalid header names or values, and
rejects keys that differ only in case (header names are case-insensitive;
HashMap iteration order would otherwise decide which duplicate wins).
ElClient::new now propagates HTTP-client construction errors instead of
panicking, so all startup failures surface as errors.

Documented in the README and the example configs; covered by config
tests.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(client): propagate W3C trace context on outbound requests

Behind a new otel cargo feature, inject the current tracing span's
OpenTelemetry context (traceparent/tracestate) into the headers of every
outbound request (request_proof, subscribe_proof_events, get_proof,
verify_proof) via the global propagator, using a hand-rolled
HeaderInjector mirroring ere_server_client's OtelPropagation middleware
and the server crate's HeaderExtractor. For the SSE subscription the
context is captured eagerly at call time, not at first poll. Without the
feature the helper returns an empty header map, so default builds change
no behavior and pull in no OpenTelemetry dependencies.

The crate docs call out that the calling application must install a
global text-map propagator (the OpenTelemetry default is a no-op), or no
traceparent header is emitted even with the feature enabled.

This completes the propagation chain for external callers such as
proofessoor: caller span -> zkboost-client traceparent -> zkboost server
request span -> ere zkVM server.

Verified by an otel-gated integration test that serves requests on a
local listener and asserts both a unary request and the SSE subscribe
arrive with a traceparent carrying the active span's trace id.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(server): set otel span kind and HTTP attributes on the request span

Set otel.kind=server and basic HTTP semantic-convention attributes
(http.request.method, url.path, url.query) on the request span. They are
recorded as OTLP-only attributes, not tracing fields, so the log format
of default-feature builds is unchanged.

Also move the request-span trace-context test into its own
integration-test binary (tests/otel_request_span.rs): it installs a
process-global tracing subscriber so spans created on the server's tasks
are observed, and sharing a process with parallel tests both raced on the
global subscriber and poisoned the global callsite interest cache, which
previously required a warm-up/retry workaround. The relocated test
additionally asserts the span kind and HTTP attributes.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(server): replay proof failures to late SSE subscribers

Previously only completed proofs were cached and replayed on subscribe;
a subscriber that missed a live proof_failure broadcast could never learn
the terminal outcome of a request. Terminal failures are now cached in a
bounded LRU (same size as the completed-proof cache) and replayed to
per-root SSE subscribers exactly like completions, with unchanged event
shapes. A later successful proof for the same (root, proof_type) evicts
the stale failure entry so both outcomes are never replayed.

Event delivery on the subscribe endpoint is at-least-once with
latest-wins semantics: a replayed stale failure can still precede the
completion of a retry that has already succeeded, and subscribers must
treat the most recent terminal event per (root, proof_type) as
authoritative. This is documented on the handler and in the README.

Tests: subscribe-after-failure receives the replayed failure (handler
unit test + end-to-end re-subscribe assertions for proving error and
timeout), LRU bound eviction, cached-vs-broadcast event equality, and
completion evicting a stale failure.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* chore(client): simplify unused dependency lint

* fix(server): clear stale failure on retry admission

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: chetanyb

Cargo.lock

@@ -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

README.md

@@ -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/Cargo.toml

@@ -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/src/lib.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/client/tests/otel_propagation.rs

@@ -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"]