From ca9efbc5a2cbcc89494115804e1f0ee366e0a16a Mon Sep 17 00:00:00 2001 From: Frando Date: Fri, 2 Oct 2026 11:45:35 +0200 Subject: [PATCH 01/12] refactor!: move the opentelemetry integration into irpc-opentelemetry irpc currently depends on `opentelemetry` and `tracing-opentelemetry` behind the `tracing-opentelemetry` feature, and exposes their types in its public API. Their versions change often, so irpc 1.0 could not update them without a major release (#111). This PR removes both dependencies from irpc. irpc keeps the wire format and gets a small `Propagator` trait: one method writes the context of a span into the text headers of a request, one sets the parent of a span from them. An application installs one propagator per process with `irpc::span_propagation::set_propagator`. The propagator does not change the wire format: a protocol with `span_propagation` always sends the `Option`, and without a propagator its value is `None`. The new crate `irpc-opentelemetry` (0.x) implements the trait with the global text map propagator of `opentelemetry`, and can follow new `opentelemetry` versions with breaking releases. Other tracing backends can implement the trait too. The span propagation tests and example move to the new crate. A new test in irpc checks the hook with a fake propagator. For cargo-release, the new crate has its own version and does not share the version of irpc. * removed: the feature `tracing-opentelemetry`. Add `irpc-opentelemetry` and call `irpc_opentelemetry::install()` once at startup. * removed: `SpanContextCarrier::from_current` and `SpanContextCarrier::to_context`, and the `Injector` and `Extractor` impls of `SpanContextCarrier`. * changed: `rpc` enables the `rt` feature of `tokio`, for the task-local of the span context. --- .github/workflows/ci.yml | 6 +- Cargo.lock | 24 ++- Cargo.toml | 32 +--- irpc-opentelemetry/Cargo.toml | 43 +++++ .../examples}/span_propagation.rs | 11 +- irpc-opentelemetry/src/lib.rs | 80 ++++++++++ .../tests}/span_propagation.rs | 2 +- .../tests}/span_propagation_concurrent.rs | 3 +- release.toml | 3 + src/lib.rs | 10 +- src/rpc.rs | 2 +- src/span_propagation.rs | 150 +++++++++--------- tests/span_propagator.rs | 84 ++++++++++ 13 files changed, 324 insertions(+), 126 deletions(-) create mode 100644 irpc-opentelemetry/Cargo.toml rename {examples => irpc-opentelemetry/examples}/span_propagation.rs (94%) create mode 100644 irpc-opentelemetry/src/lib.rs rename {tests => irpc-opentelemetry/tests}/span_propagation.rs (98%) rename {tests => irpc-opentelemetry/tests}/span_propagation_concurrent.rs (98%) create mode 100644 tests/span_propagator.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9ff4ad5e..f23b1b35 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -87,8 +87,8 @@ jobs: run: cargo check --locked --no-default-features --features spans --bins --tests --examples - name: cargo check (irpc, feature rpc) run: cargo check --locked --no-default-features --features rpc --bins --tests --examples - - name: cargo check (irpc, feature tracing-opentelemetry) - run: cargo check --locked --no-default-features --features tracing-opentelemetry --bins --tests --examples + - name: cargo check (irpc-opentelemetry) + run: cargo check --locked -p irpc-opentelemetry --bins --tests --examples test-release: runs-on: ${{ matrix.target.os }} @@ -144,6 +144,8 @@ jobs: - name: irpc run: cargo docs-rs --locked -p irpc + - name: irpc-opentelemetry + run: cargo docs-rs --locked -p irpc-opentelemetry wasm_build: name: Build wasm32 diff --git a/Cargo.lock b/Cargo.lock index 75f72fee..eb659426 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1544,9 +1544,6 @@ dependencies = [ "n0-tracing-test", "noq", "num-traits", - "opentelemetry", - "opentelemetry-otlp", - "opentelemetry_sdk", "portable-atomic", "postcard", "rand 0.10.3", @@ -1559,7 +1556,6 @@ dependencies = [ "tokio", "tokio-util", "tracing", - "tracing-opentelemetry", "tracing-subscriber", "trybuild", ] @@ -1573,6 +1569,26 @@ dependencies = [ "syn 3.0.6", ] +[[package]] +name = "irpc-opentelemetry" +version = "0.1.0" +dependencies = [ + "anyhow", + "clap", + "iroh", + "irpc", + "n0-error", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", + "rand 0.10.3", + "serde", + "tokio", + "tracing", + "tracing-opentelemetry", + "tracing-subscriber", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" diff --git a/Cargo.toml b/Cargo.toml index 9d3abc1c..9ab1e612 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -46,9 +46,6 @@ n0-future = { workspace = true } futures-util = { workspace = true, optional = true } # for the derive reexport/feature irpc-derive = { version = "0.17.0", path = "./irpc-derive", optional = true } -# for remote span propagation when tracing-opentelemetry feature is enabled -opentelemetry = { version = "0.32", optional = true } -tracing-opentelemetry = { version = "0.33", optional = true } # used for the iroh transport iroh = { workspace = true, optional = true } portable-atomic = { version = "1", optional = true } @@ -84,14 +81,10 @@ futures-util.workspace = true hex = "0.4.3" rand = "0.10.3" n0-tracing-test = "0.3.0" -opentelemetry = "0.32" -opentelemetry_sdk = { version = "0.32", features = ["testing"] } -opentelemetry-otlp = { version = "0.32", default-features = false, features = ["http-proto", "reqwest-blocking-client", "trace"] } -tracing-opentelemetry = "0.33" [features] # enable the remote transport -rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "tokio/io-util"] +rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "tokio/io-util", "tokio/rt"] # enable the iroh transport iroh = ["rpc", "dep:iroh", "dep:portable-atomic", "dep:getrandom"] # add helpers to create noq endpoints, for tests and for rpc on localhost @@ -101,8 +94,6 @@ spans = ["dep:tracing"] stream = ["dep:futures-util"] derive = ["dep:irpc-derive"] varint-util = ["dep:postcard", "dep:smallvec", "tokio/io-util"] -# enable OpenTelemetry span context propagation across remote connections -tracing-opentelemetry = ["dep:opentelemetry", "dep:tracing-opentelemetry", "rpc"] default = ["rpc", "iroh", "spans", "stream", "derive"] [[example]] @@ -149,21 +140,13 @@ required-features = ["iroh", "spans", "stream", "derive"] name = "server-shared-state" required-features = ["iroh", "spans", "stream", "derive"] -[[example]] -name = "span_propagation" -required-features = ["iroh", "spans", "stream", "derive"] - -[[test]] -name = "span_propagation" -required-features = ["iroh", "spans", "stream", "derive"] - [[test]] -name = "span_propagation_concurrent" -required-features = ["iroh", "spans", "stream", "derive"] +name = "span_propagator" +required-features = ["noq_endpoint_setup", "derive", "spans"] [workspace] resolver = "2" -members = ["irpc-derive"] +members = ["irpc-derive", "irpc-opentelemetry"] [package.metadata.docs.rs] all-features = true @@ -217,11 +200,4 @@ allowed_external_types = [ "futures_core::stream::Stream", "futures_lite::future::Boxed", "futures_sink::Sink", - - # only under the tracing-opentelemtry feature - # TODO: it would be preferable to find a way to not have this in the public API for 1.0 - # because otherwise we can't update the opentelemetry version - "opentelemetry::context::Context", - "opentelemetry::propagation::Extractor", - "opentelemetry::propagation::Injector", ] diff --git a/irpc-opentelemetry/Cargo.toml b/irpc-opentelemetry/Cargo.toml new file mode 100644 index 00000000..d37a9697 --- /dev/null +++ b/irpc-opentelemetry/Cargo.toml @@ -0,0 +1,43 @@ +[package] +name = "irpc-opentelemetry" +version = "0.1.0" +edition = "2024" +authors = ["n0 team"] +keywords = ["rpc", "opentelemetry", "tracing"] +categories = ["network-programming"] +license = "Apache-2.0/MIT" +repository = "https://github.com/n0-computer/irpc" +description = "OpenTelemetry span propagation for irpc" +rust-version = "1.91" + +[dependencies] +irpc = { version = "0.17.0", path = "..", default-features = false, features = ["rpc"] } +opentelemetry = "0.32" +tracing = { workspace = true } +tracing-opentelemetry = "0.33" + +[dev-dependencies] +anyhow = { workspace = true } +clap = { version = "4.5.41", features = ["derive"] } +# Tests and examples need a crypto provider, which iroh's default features supply. +iroh = { workspace = true, default-features = true } +irpc = { version = "0.17.0", path = "..", features = ["iroh", "derive", "spans"] } +n0-error = { workspace = true } +opentelemetry_sdk = { version = "0.32", features = ["testing"] } +opentelemetry-otlp = { version = "0.32", default-features = false, features = ["http-proto", "reqwest-blocking-client", "trace"] } +rand = "0.10.3" +serde = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tracing-subscriber = { workspace = true, features = ["fmt", "env-filter"] } + +[package.metadata.cargo_check_external_types] +allowed_external_types = [ + "irpc::*", + # Re-exported. This crate is 0.x and follows the version of opentelemetry. + "opentelemetry", + "tracing_opentelemetry", +] + +[package.metadata.release] +# Released on its own: the version follows the opentelemetry releases, not irpc. +shared-version = false diff --git a/examples/span_propagation.rs b/irpc-opentelemetry/examples/span_propagation.rs similarity index 94% rename from examples/span_propagation.rs rename to irpc-opentelemetry/examples/span_propagation.rs index 68058f7e..0e30abbe 100644 --- a/examples/span_propagation.rs +++ b/irpc-opentelemetry/examples/span_propagation.rs @@ -1,4 +1,4 @@ -//! Distributed span propagation across irpc-iroh, visualized in Jaeger. +//! Distributed span propagation over irpc with iroh, visualized in Jaeger. //! //! Start Jaeger (UI on 16686, OTLP HTTP on 4318): //! @@ -9,13 +9,13 @@ //! Server (prints its endpoint id): //! //! ```sh -//! cargo run --features tracing-opentelemetry --example span_propagation -- server +//! cargo run -p irpc-opentelemetry --example span_propagation -- server //! ``` //! //! Client: //! //! ```sh -//! cargo run --features tracing-opentelemetry --example span_propagation -- client +//! cargo run -p irpc-opentelemetry --example span_propagation -- client //! ``` //! //! Open , pick the `example-client` service, and each @@ -49,7 +49,7 @@ const ALPN: &[u8] = b"irpc-iroh/span_propagation/1"; const DEFAULT_OTLP_ENDPOINT: &str = "http://localhost:4318/v1/traces"; #[derive(Parser, Debug)] -#[command(about = "Distributed span propagation demo over irpc-iroh")] +#[command(about = "Distributed span propagation demo over irpc with iroh")] struct Cli { /// OTLP HTTP/protobuf endpoint to export spans to. Defaults to Jaeger's standard port. #[arg(long, global = true, default_value = DEFAULT_OTLP_ENDPOINT)] @@ -84,6 +84,7 @@ fn init_tracing( opentelemetry::global::set_text_map_propagator( opentelemetry_sdk::propagation::TraceContextPropagator::new(), ); + irpc_opentelemetry::install()?; // (2) Build the OTLP HTTP/protobuf exporter pointed at the collector // (Jaeger's OTLP ingestor by default). The `reqwest-blocking-client` @@ -183,7 +184,7 @@ async fn server(otlp_endpoint: &str) -> Result<()> { println!("server endpoint id: {}", router.endpoint().id()); println!("run the client with:"); println!( - " cargo run --features tracing-opentelemetry --example span_propagation -- client {}", + " cargo run -p irpc-opentelemetry --example span_propagation -- client {}", router.endpoint().id() ); println!("press ctrl+c to stop"); diff --git a/irpc-opentelemetry/src/lib.rs b/irpc-opentelemetry/src/lib.rs new file mode 100644 index 00000000..90a4d7d3 --- /dev/null +++ b/irpc-opentelemetry/src/lib.rs @@ -0,0 +1,80 @@ +//! OpenTelemetry span propagation for irpc. +//! +//! irpc carries the span context of a request in text headers, but does not +//! know any tracing backend. This crate connects it to OpenTelemetry: the +//! client writes the context of the current span with the global text map +//! propagator of `opentelemetry`, and the server sets the parent of the request +//! span from it. +//! +//! Install it once at startup, after the `opentelemetry` setup: +//! +//! ```no_run +//! opentelemetry::global::set_text_map_propagator( +//! opentelemetry_sdk::propagation::TraceContextPropagator::new(), +//! ); +//! irpc_opentelemetry::install().expect("no other propagator is installed"); +//! ``` +//! +//! Spans reach OpenTelemetry through the `tracing-opentelemetry` layer. This +//! crate uses the versions of `opentelemetry` and `tracing-opentelemetry` that +//! it re-exports. The application must use the same versions, because both +//! crates keep their state in statics. + +use irpc::span_propagation::{Propagator, PropagatorAlreadySet, SpanContextCarrier}; +pub use opentelemetry; +use opentelemetry::propagation::{Extractor, Injector}; +pub use tracing_opentelemetry; +use tracing_opentelemetry::OpenTelemetrySpanExt; + +/// A [`Propagator`] that uses the global text map propagator of `opentelemetry`. +#[derive(Debug, Default, Clone, Copy)] +pub struct OtelPropagator; + +impl Propagator for OtelPropagator { + fn inject(&self, span: &tracing::Span, carrier: &mut SpanContextCarrier) { + // The `tracing-opentelemetry` layer keeps the context in the tracing span. + let context = span.context(); + opentelemetry::global::get_text_map_propagator(|propagator| { + propagator.inject_context(&context, &mut Headers(carrier)); + }); + } + + fn set_parent(&self, span: &tracing::Span, carrier: &SpanContextCarrier) { + let context = opentelemetry::global::get_text_map_propagator(|propagator| { + propagator + .extract_with_context(&opentelemetry::Context::current(), &HeadersRef(carrier)) + }); + let _ = span.set_parent(context); + } +} + +/// Installs [`OtelPropagator`] as the span propagator of irpc. +/// +/// # Errors +/// +/// Returns [`PropagatorAlreadySet`] if a propagator is installed already. +pub fn install() -> Result<(), PropagatorAlreadySet> { + irpc::span_propagation::set_propagator(OtelPropagator) +} + +/// Writes headers into a [`SpanContextCarrier`]. +struct Headers<'a>(&'a mut SpanContextCarrier); + +impl Injector for Headers<'_> { + fn set(&mut self, key: &str, value: String) { + self.0.set(key, value); + } +} + +/// Reads headers from a [`SpanContextCarrier`]. +struct HeadersRef<'a>(&'a SpanContextCarrier); + +impl Extractor for HeadersRef<'_> { + fn get(&self, key: &str) -> Option<&str> { + self.0.get(key) + } + + fn keys(&self) -> Vec<&str> { + self.0.keys().collect() + } +} diff --git a/tests/span_propagation.rs b/irpc-opentelemetry/tests/span_propagation.rs similarity index 98% rename from tests/span_propagation.rs rename to irpc-opentelemetry/tests/span_propagation.rs index 459b1f44..ccbfe0d4 100644 --- a/tests/span_propagation.rs +++ b/irpc-opentelemetry/tests/span_propagation.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "tracing-opentelemetry")] mod span_propagation { use iroh::{Endpoint, endpoint::presets, protocol::Router}; @@ -42,6 +41,7 @@ mod span_propagation { opentelemetry::global::set_text_map_propagator( opentelemetry_sdk::propagation::TraceContextPropagator::new(), ); + irpc_opentelemetry::install().expect("no other propagator is installed"); // Use an in-memory exporter so we can inspect exported spans after the test. let exporter = InMemorySpanExporter::default(); diff --git a/tests/span_propagation_concurrent.rs b/irpc-opentelemetry/tests/span_propagation_concurrent.rs similarity index 98% rename from tests/span_propagation_concurrent.rs rename to irpc-opentelemetry/tests/span_propagation_concurrent.rs index c4a75f53..8514362c 100644 --- a/tests/span_propagation_concurrent.rs +++ b/irpc-opentelemetry/tests/span_propagation_concurrent.rs @@ -12,8 +12,6 @@ //! tracing subscriber and tracer provider, which would conflict with the //! test in `tests/span_propagation.rs`. -#![cfg(feature = "tracing-opentelemetry")] - use iroh::{Endpoint, endpoint::presets, protocol::Router}; use irpc::{ Client, WithChannels, @@ -52,6 +50,7 @@ async fn span_propagation_concurrent() -> n0_error::Result<()> { opentelemetry::global::set_text_map_propagator( opentelemetry_sdk::propagation::TraceContextPropagator::new(), ); + irpc_opentelemetry::install().expect("no other propagator is installed"); let exporter = InMemorySpanExporter::default(); let provider = SdkTracerProvider::builder() .with_simple_exporter(exporter.clone()) diff --git a/release.toml b/release.toml index bf6eaefc..c92766fa 100644 --- a/release.toml +++ b/release.toml @@ -1 +1,4 @@ +# irpc and irpc-derive share one version. irpc-opentelemetry has its own +# version, which follows the opentelemetry releases. It sets +# `shared-version = false` in its Cargo.toml. shared-version = true diff --git a/src/lib.rs b/src/lib.rs index 5ba1f9d2..8c24a980 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -220,12 +220,10 @@ use std::{fmt::Debug, future::Future, io, marker::PhantomData, ops::Deref}; /// * `no_rpc` *(optional, no value)*: If set, no implementation of `RemoteService` will be generated and the generated /// code works without the `rpc` feature of `irpc`. /// * `no_spans` *(optional, no value)*: If set, the generated code works without the `spans` feature of `irpc`. -/// * `span_propagation` *(optional, no value)*: If set, enables OpenTelemetry span context propagation -/// across remote connections. When enabled, span context is included in the wire format as -/// `(Option, Message)`, and the generated `RemoteService` implementation -/// will set the parent span from the propagated remote context. Requires the `tracing-opentelemetry` -/// feature to be enabled for actual OpenTelemetry integration; without it, the context is -/// still serialized but has no effect. +/// * `span_propagation` *(optional, no value)*: If set, each request carries the span context +/// of the client, as `(Option, Message)` on the wire. The generated +/// `RemoteService` implementation sets the parent of the request span from it. This needs a +/// [`span_propagation::Propagator`], for example from the `irpc-opentelemetry` crate. /// /// ## Variant attributes /// diff --git a/src/rpc.rs b/src/rpc.rs index 99ad7ae4..ee5129c4 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -290,7 +290,7 @@ pub(crate) fn prepare_write( if S::SPAN_PROPAGATION { // Include span context in wire format - let span_ctx = Some(crate::span_propagation::SpanContextCarrier::from_current()); + let span_ctx = crate::span_propagation::SpanContextCarrier::from_current(); let payload = (span_ctx, msg); if postcard::experimental::serialized_size(&payload)? as u64 > MAX_MESSAGE_SIZE { return Err(e!(WriteError::MaxMessageSizeExceeded)); diff --git a/src/span_propagation.rs b/src/span_propagation.rs index 06fe2832..826d18a8 100644 --- a/src/span_propagation.rs +++ b/src/span_propagation.rs @@ -1,112 +1,108 @@ -//! Span context propagation for remote RPC calls +//! Span context propagation across remote connections. //! -//! This module provides the `SpanContextCarrier` type for propagating trace context -//! across remote boundaries. The type is always available when `rpc` feature is enabled, -//! but actual OpenTelemetry integration requires the `tracing-opentelemetry` feature. +//! A protocol opts in with the `span_propagation` argument of the +//! [`rpc_requests`](crate::rpc_requests) macro. Each request then carries a +//! [`SpanContextCarrier`], a map of text headers, in front of the request. //! -//! The propagated context is scoped to a single request handler via a tokio task-local, -//! installed by the dispatch loop in `Handler::handle_connection`. This isolates concurrent -//! requests from each other and is robust to thread migration across `.await` points. +//! irpc does not know any tracing backend. A [`Propagator`] writes the context +//! of a span into the headers on the client, and sets the parent of a span from +//! the headers on the server. Install one per process with [`set_propagator`]. +//! +//! The propagator does not change the wire format: a protocol with +//! `span_propagation` always sends the `Option`. Without a +//! propagator, its value is `None`. +//! +//! The `irpc-opentelemetry` crate has a propagator for OpenTelemetry. -use std::{collections::HashMap, future::Future}; +use std::{collections::HashMap, future::Future, sync::OnceLock}; +use n0_error::stack_error; use serde::{Deserialize, Serialize}; -#[cfg(feature = "tracing-opentelemetry")] tokio::task_local! { - static SPAN_CONTEXT: opentelemetry::Context; + static SPAN_CONTEXT: SpanContextCarrier; } -/// Carrier for propagating span context across RPC boundaries using W3C Trace Context format. +/// The propagator of this process. +static PROPAGATOR: OnceLock> = OnceLock::new(); + +/// Text headers that carry the context of a span to the remote. /// -/// This type is always available for serialization purposes. When the -/// `tracing-opentelemetry` feature is enabled, it can extract/inject actual -/// OpenTelemetry trace context. Without that feature, it simply serializes as an -/// empty map. +/// On the wire, this is a map from string to string. Propagators decide which +/// headers they use, for example `traceparent` and `tracestate` for W3C trace +/// context. #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct SpanContextCarrier { headers: HashMap, } -#[cfg(feature = "tracing-opentelemetry")] -impl opentelemetry::propagation::Injector for SpanContextCarrier { - fn set(&mut self, key: &str, value: String) { - self.headers.insert(key.to_string(), value); +impl SpanContextCarrier { + /// Returns the value of the header `key`. + pub fn get(&self, key: &str) -> Option<&str> { + self.headers.get(key).map(String::as_str) } -} -#[cfg(feature = "tracing-opentelemetry")] -impl opentelemetry::propagation::Extractor for SpanContextCarrier { - fn get(&self, key: &str) -> Option<&str> { - self.headers.get(key).map(|v| v.as_str()) + /// Sets the header `key` to `value`. + pub fn set(&mut self, key: impl Into, value: impl Into) { + self.headers.insert(key.into(), value.into()); } - fn keys(&self) -> Vec<&str> { - self.headers.keys().map(|k| k.as_str()).collect() + /// Returns the keys of all headers. + pub fn keys(&self) -> impl Iterator { + self.headers.keys().map(String::as_str) } -} -impl SpanContextCarrier { - /// Create a carrier from the current OpenTelemetry context. - /// - /// When `tracing-opentelemetry` feature is enabled, this extracts the current - /// trace context. Without the feature, this returns an empty carrier. - #[cfg(feature = "tracing-opentelemetry")] - pub fn from_current() -> Self { - use opentelemetry::global; - use tracing_opentelemetry::OpenTelemetrySpanExt; + /// Returns the context of the current span, if a propagator is installed. + pub(crate) fn from_current() -> Option { + let propagator = PROPAGATOR.get()?; let mut carrier = Self::default(); - // Get the OTel context from the current tracing span, not from - // opentelemetry::Context::current(). The tracing-opentelemetry layer - // stores OTel spans inside tracing spans, so the thread-local OTel - // context won't have the right span. - let ctx = tracing::Span::current().context(); - global::get_text_map_propagator(|prop| { - prop.inject_context(&ctx, &mut carrier); - }); - carrier + propagator.inject(&tracing::Span::current(), &mut carrier); + Some(carrier) } +} - #[cfg(not(feature = "tracing-opentelemetry"))] - pub fn from_current() -> Self { - Self::default() - } +/// Connects span propagation to a tracing backend. +pub trait Propagator: Send + Sync + 'static { + /// Writes the context of `span` into `carrier`. + fn inject(&self, span: &tracing::Span, carrier: &mut SpanContextCarrier); - /// Extract an OpenTelemetry context from this carrier. - #[cfg(feature = "tracing-opentelemetry")] - pub fn to_context(&self) -> opentelemetry::Context { - use opentelemetry::global; - global::get_text_map_propagator(|prop| { - prop.extract_with_context(&opentelemetry::Context::current(), self) - }) - } + /// Sets the parent of `span` from the context in `carrier`. + fn set_parent(&self, span: &tracing::Span, carrier: &SpanContextCarrier); +} + +/// The error of [`set_propagator`] when a propagator is installed already. +#[stack_error(derive)] +#[error("a span propagator is already installed")] +pub struct PropagatorAlreadySet; + +/// Installs the propagator for this process. +/// +/// # Errors +/// +/// Returns [`PropagatorAlreadySet`] if a propagator is installed already. +pub fn set_propagator(propagator: impl Propagator) -> Result<(), PropagatorAlreadySet> { + PROPAGATOR + .set(Box::new(propagator)) + .map_err(|_| PropagatorAlreadySet) } -/// Run `fut` with `carrier`'s context installed as the per-task scope read by -/// [`set_span_parent_from_remote`]. +/// Runs `fut` with `carrier` in scope for [`set_span_parent_from_remote`]. /// -/// Used by transport implementations (`irpc::rpc`, `irpc::iroh`) to wrap a single -/// request handler. Most users will not call this directly. +/// The server loop calls this for each request. Most users do not call it. pub async fn scope_remote(carrier: Option, fut: F) -> F::Output { - #[cfg(feature = "tracing-opentelemetry")] - if let Some(carrier) = carrier { - return SPAN_CONTEXT.scope(carrier.to_context(), fut).await; + match carrier { + Some(carrier) => SPAN_CONTEXT.scope(carrier, fut).await, + None => fut.await, } - let _ = carrier; - fut.await } -/// Set the parent of a span from the propagated remote context, if one is in scope. +/// Sets the parent of `span` from the span context of the current request. /// -/// Called by the code generated by `rpc_requests(span_propagation)`. Looks up the -/// task-local installed by the dispatch loop; no-op outside that scope. +/// The code from `rpc_requests(span_propagation)` calls this. It does nothing +/// outside of [`scope_remote`] or without a propagator. pub fn set_span_parent_from_remote(span: &tracing::Span) { - #[cfg(feature = "tracing-opentelemetry")] - { - let _ = SPAN_CONTEXT.try_with(|ctx| { - use tracing_opentelemetry::OpenTelemetrySpanExt; - let _ = span.set_parent(ctx.clone()); - }); - } - let _ = span; + let Some(propagator) = PROPAGATOR.get() else { + return; + }; + let _ = SPAN_CONTEXT.try_with(|carrier| propagator.set_parent(span, carrier)); } diff --git a/tests/span_propagator.rs b/tests/span_propagator.rs new file mode 100644 index 00000000..86896632 --- /dev/null +++ b/tests/span_propagator.rs @@ -0,0 +1,84 @@ +//! Checks that a custom [`Propagator`] carries headers from client to server. +//! +//! Lives in its own test binary, because the propagator is global. + +use std::{ + net::{Ipv4Addr, SocketAddr, SocketAddrV4}, + sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }, +}; + +use irpc::{ + Client, WithChannels, + channel::oneshot, + noq::{listen, make_client_endpoint, make_server_endpoint}, + rpc::{Handler, RemoteService}, + rpc_requests, + span_propagation::{Propagator, SpanContextCarrier, set_propagator}, +}; +use n0_future::task::AbortOnDropHandle; +use noq::Endpoint; +use serde::{Deserialize, Serialize}; +use testresult::TestResult; + +fn create_connected_endpoints() -> TestResult<(Endpoint, Endpoint, SocketAddr)> { + let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0).into(); + let (server, cert) = make_server_endpoint(addr)?; + let client = make_client_endpoint(addr, &[cert.as_slice()])?; + let port = server.local_addr()?.port(); + let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into(); + Ok((server, client, server_addr)) +} + +#[rpc_requests(message = EchoMessage, span_propagation)] +#[derive(Debug, Serialize, Deserialize)] +enum EchoProtocol { + #[rpc(tx = oneshot::Sender)] + #[wrap(Echo)] + Echo(String), +} + +/// Writes a counter into the headers, and records what the server reads. +#[derive(Debug, Clone, Default)] +struct TestPropagator { + sent: Arc, + received: Arc>>, +} + +impl Propagator for TestPropagator { + fn inject(&self, _span: &tracing::Span, carrier: &mut SpanContextCarrier) { + let n = self.sent.fetch_add(1, Ordering::SeqCst); + carrier.set("test-request", n.to_string()); + } + + fn set_parent(&self, _span: &tracing::Span, carrier: &SpanContextCarrier) { + let value = carrier.get("test-request").unwrap_or_default().to_string(); + self.received.lock().expect("poisoned").push(value); + } +} + +#[tokio::test] +async fn propagator_carries_headers_to_the_server() -> TestResult<()> { + let propagator = TestPropagator::default(); + set_propagator(propagator.clone())?; + + let (server, client, server_addr) = create_connected_endpoints()?; + let handler: Handler = Arc::new(|request, rx, tx| { + Box::pin(async move { + let EchoMessage::Echo(WithChannels { inner, tx, .. }) = + request.with_remote_channels(rx, tx); + tx.send(inner.0).await.ok(); + Ok(()) + }) + }); + let _server = AbortOnDropHandle::new(tokio::spawn(listen::(server, handler))); + let client = Client::::noq(client, server_addr); + for text in ["a", "b"] { + assert_eq!(client.rpc(Echo(text.into())).await?, text); + } + let received = propagator.received.lock().expect("poisoned").clone(); + assert_eq!(received, ["0", "1"]); + Ok(()) +} From ecfea50fe65a645d593649b1a2c88f3ce9e6bacb Mon Sep 17 00:00:00 2001 From: Frando Date: Fri, 2 Oct 2026 12:10:09 +0200 Subject: [PATCH 02/12] refactor!: find the span propagator through the tracing subscriber The span propagator of irpc is currently a process-wide static, set once with `set_propagator`. A process can only have one, a second call fails, and tests that need different propagators must live in separate test binaries. The setup is also apart from the rest of the tracing setup. This PR removes the static. irpc gets a `PropagatorLayer`, a tracing layer that holds a propagator and records nothing. irpc finds it with `Dispatch::downcast_ref`, the same way `tracing-opentelemetry` finds its own layer. The client looks in the subscriber of the current thread, and the server in the subscriber of the request span. Without the layer, requests send `None` as their span context, as before. The wire format does not change. `rpc` now enables `tracing-subscriber` without default features, for the `Layer` trait. `irpc-opentelemetry` replaces `install()` with `layer()`, which the application adds to its subscriber next to the `tracing-opentelemetry` layer. * removed: `span_propagation::set_propagator` and `PropagatorAlreadySet`. Add a `PropagatorLayer` to the subscriber. * removed: `irpc_opentelemetry::install`. Use `irpc_opentelemetry::layer()` in the subscriber. --- Cargo.toml | 6 +- .../examples/span_propagation.rs | 15 +++-- irpc-opentelemetry/src/lib.rs | 23 ++++--- irpc-opentelemetry/tests/span_propagation.rs | 3 +- .../tests/span_propagation_concurrent.rs | 2 +- src/lib.rs | 3 +- src/span_propagation.rs | 67 +++++++++++-------- tests/span_propagator.rs | 9 +-- 8 files changed, 75 insertions(+), 53 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 9ab1e612..61e45ab2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,7 @@ n0-error = { workspace = true } # used in the endpoint handler code when using rpc tracing = { workspace = true, optional = true } +tracing-subscriber = { version = "0.3.23", default-features = false, optional = true } # used to ser/de messages when using rpc postcard = { workspace = true, features = ["alloc", "use-std"], optional = true } # currently only transport when using rpc @@ -84,7 +85,7 @@ n0-tracing-test = "0.3.0" [features] # enable the remote transport -rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "tokio/io-util", "tokio/rt"] +rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "dep:tracing-subscriber", "tokio/io-util", "tokio/rt"] # enable the iroh transport iroh = ["rpc", "dep:iroh", "dep:portable-atomic", "dep:getrandom"] # add helpers to create noq endpoints, for tests and for rpc on localhost @@ -200,4 +201,7 @@ allowed_external_types = [ "futures_core::stream::Stream", "futures_lite::future::Boxed", "futures_sink::Sink", + # `span_propagation::PropagatorLayer` is a tracing layer + "tracing_core::subscriber::Subscriber", + "tracing_subscriber::layer::Layer", ] diff --git a/irpc-opentelemetry/examples/span_propagation.rs b/irpc-opentelemetry/examples/span_propagation.rs index 0e30abbe..d9e3e136 100644 --- a/irpc-opentelemetry/examples/span_propagation.rs +++ b/irpc-opentelemetry/examples/span_propagation.rs @@ -84,7 +84,6 @@ fn init_tracing( opentelemetry::global::set_text_map_propagator( opentelemetry_sdk::propagation::TraceContextPropagator::new(), ); - irpc_opentelemetry::install()?; // (2) Build the OTLP HTTP/protobuf exporter pointed at the collector // (Jaeger's OTLP ingestor by default). The `reqwest-blocking-client` @@ -126,11 +125,15 @@ fn init_tracing( .with_filter(EnvFilter::new("info")); let fmt_filter = EnvFilter::try_from_default_env() .unwrap_or_else(|_| EnvFilter::new("span_propagation=info")); - let subscriber = Registry::default().with(telemetry).with( - tracing_subscriber::fmt::layer() - .with_target(false) - .with_filter(fmt_filter), - ); + // The irpc layer writes the span context into each request. + let subscriber = Registry::default() + .with(telemetry) + .with(irpc_opentelemetry::layer()) + .with( + tracing_subscriber::fmt::layer() + .with_target(false) + .with_filter(fmt_filter), + ); tracing::subscriber::set_global_default(subscriber)?; Ok(provider) } diff --git a/irpc-opentelemetry/src/lib.rs b/irpc-opentelemetry/src/lib.rs index 90a4d7d3..b2491ec9 100644 --- a/irpc-opentelemetry/src/lib.rs +++ b/irpc-opentelemetry/src/lib.rs @@ -6,13 +6,20 @@ //! propagator of `opentelemetry`, and the server sets the parent of the request //! span from it. //! -//! Install it once at startup, after the `opentelemetry` setup: +//! Add its [`layer`] to the tracing subscriber, next to the +//! `tracing-opentelemetry` layer: //! //! ```no_run +//! use tracing_subscriber::{Registry, layer::SubscriberExt}; +//! //! opentelemetry::global::set_text_map_propagator( //! opentelemetry_sdk::propagation::TraceContextPropagator::new(), //! ); -//! irpc_opentelemetry::install().expect("no other propagator is installed"); +//! let tracer = opentelemetry::global::tracer("app"); +//! let subscriber = Registry::default() +//! .with(tracing_opentelemetry::layer().with_tracer(tracer)) +//! .with(irpc_opentelemetry::layer()); +//! tracing::subscriber::set_global_default(subscriber).expect("no other subscriber is set"); //! ``` //! //! Spans reach OpenTelemetry through the `tracing-opentelemetry` layer. This @@ -20,7 +27,7 @@ //! it re-exports. The application must use the same versions, because both //! crates keep their state in statics. -use irpc::span_propagation::{Propagator, PropagatorAlreadySet, SpanContextCarrier}; +use irpc::span_propagation::{Propagator, PropagatorLayer, SpanContextCarrier}; pub use opentelemetry; use opentelemetry::propagation::{Extractor, Injector}; pub use tracing_opentelemetry; @@ -48,13 +55,9 @@ impl Propagator for OtelPropagator { } } -/// Installs [`OtelPropagator`] as the span propagator of irpc. -/// -/// # Errors -/// -/// Returns [`PropagatorAlreadySet`] if a propagator is installed already. -pub fn install() -> Result<(), PropagatorAlreadySet> { - irpc::span_propagation::set_propagator(OtelPropagator) +/// Returns a tracing layer that makes irpc propagate span context with [`OtelPropagator`]. +pub fn layer() -> PropagatorLayer { + PropagatorLayer::new(OtelPropagator) } /// Writes headers into a [`SpanContextCarrier`]. diff --git a/irpc-opentelemetry/tests/span_propagation.rs b/irpc-opentelemetry/tests/span_propagation.rs index ccbfe0d4..f1a799d2 100644 --- a/irpc-opentelemetry/tests/span_propagation.rs +++ b/irpc-opentelemetry/tests/span_propagation.rs @@ -41,7 +41,6 @@ mod span_propagation { opentelemetry::global::set_text_map_propagator( opentelemetry_sdk::propagation::TraceContextPropagator::new(), ); - irpc_opentelemetry::install().expect("no other propagator is installed"); // Use an in-memory exporter so we can inspect exported spans after the test. let exporter = InMemorySpanExporter::default(); @@ -57,6 +56,8 @@ mod span_propagation { // that impls `LookupSpan` let subscriber = Registry::default() .with(telemetry) + .with(irpc_opentelemetry::layer()) + .with(irpc_opentelemetry::layer()) .with(tracing_subscriber::fmt::layer().with_filter(EnvFilter::from_default_env())); tracing::subscriber::set_global_default(subscriber).expect("global already set"); diff --git a/irpc-opentelemetry/tests/span_propagation_concurrent.rs b/irpc-opentelemetry/tests/span_propagation_concurrent.rs index 8514362c..dc9d8d69 100644 --- a/irpc-opentelemetry/tests/span_propagation_concurrent.rs +++ b/irpc-opentelemetry/tests/span_propagation_concurrent.rs @@ -50,7 +50,6 @@ async fn span_propagation_concurrent() -> n0_error::Result<()> { opentelemetry::global::set_text_map_propagator( opentelemetry_sdk::propagation::TraceContextPropagator::new(), ); - irpc_opentelemetry::install().expect("no other propagator is installed"); let exporter = InMemorySpanExporter::default(); let provider = SdkTracerProvider::builder() .with_simple_exporter(exporter.clone()) @@ -60,6 +59,7 @@ async fn span_propagation_concurrent() -> n0_error::Result<()> { let telemetry = tracing_opentelemetry::layer().with_tracer(tracer); let subscriber = Registry::default() .with(telemetry) + .with(irpc_opentelemetry::layer()) .with(tracing_subscriber::fmt::layer().with_filter(EnvFilter::from_default_env())); tracing::subscriber::set_global_default(subscriber).expect("global already set"); diff --git a/src/lib.rs b/src/lib.rs index 8c24a980..f6550fdc 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -223,7 +223,8 @@ use std::{fmt::Debug, future::Future, io, marker::PhantomData, ops::Deref}; /// * `span_propagation` *(optional, no value)*: If set, each request carries the span context /// of the client, as `(Option, Message)` on the wire. The generated /// `RemoteService` implementation sets the parent of the request span from it. This needs a -/// [`span_propagation::Propagator`], for example from the `irpc-opentelemetry` crate. +/// [`span_propagation::Propagator`] in the tracing subscriber, for example the layer of the +/// `irpc-opentelemetry` crate. /// /// ## Variant attributes /// diff --git a/src/span_propagation.rs b/src/span_propagation.rs index 826d18a8..7a4cbee0 100644 --- a/src/span_propagation.rs +++ b/src/span_propagation.rs @@ -6,7 +6,12 @@ //! //! irpc does not know any tracing backend. A [`Propagator`] writes the context //! of a span into the headers on the client, and sets the parent of a span from -//! the headers on the server. Install one per process with [`set_propagator`]. +//! the headers on the server. +//! +//! irpc finds the propagator through the tracing subscriber: add a +//! [`PropagatorLayer`] to it. The client asks the subscriber of the current +//! thread, and the server asks the subscriber of the request span. Without the +//! layer, irpc does not propagate span context. //! //! The propagator does not change the wire format: a protocol with //! `span_propagation` always sends the `Option`. Without a @@ -14,18 +19,14 @@ //! //! The `irpc-opentelemetry` crate has a propagator for OpenTelemetry. -use std::{collections::HashMap, future::Future, sync::OnceLock}; +use std::{collections::HashMap, future::Future}; -use n0_error::stack_error; use serde::{Deserialize, Serialize}; tokio::task_local! { static SPAN_CONTEXT: SpanContextCarrier; } -/// The propagator of this process. -static PROPAGATOR: OnceLock> = OnceLock::new(); - /// Text headers that carry the context of a span to the remote. /// /// On the wire, this is a map from string to string. Propagators decide which @@ -52,12 +53,14 @@ impl SpanContextCarrier { self.headers.keys().map(String::as_str) } - /// Returns the context of the current span, if a propagator is installed. + /// Returns the context of the current span, if the subscriber has a propagator. pub(crate) fn from_current() -> Option { - let propagator = PROPAGATOR.get()?; - let mut carrier = Self::default(); - propagator.inject(&tracing::Span::current(), &mut carrier); - Some(carrier) + tracing::dispatcher::get_default(|dispatch| { + let propagator = dispatch.downcast_ref::()?; + let mut carrier = Self::default(); + propagator.0.inject(&tracing::Span::current(), &mut carrier); + Some(carrier) + }) } } @@ -70,22 +73,26 @@ pub trait Propagator: Send + Sync + 'static { fn set_parent(&self, span: &tracing::Span, carrier: &SpanContextCarrier); } -/// The error of [`set_propagator`] when a propagator is installed already. -#[stack_error(derive)] -#[error("a span propagator is already installed")] -pub struct PropagatorAlreadySet; - -/// Installs the propagator for this process. +/// A tracing layer that hands out a [`Propagator`] to irpc. /// -/// # Errors -/// -/// Returns [`PropagatorAlreadySet`] if a propagator is installed already. -pub fn set_propagator(propagator: impl Propagator) -> Result<(), PropagatorAlreadySet> { - PROPAGATOR - .set(Box::new(propagator)) - .map_err(|_| PropagatorAlreadySet) +/// The layer records nothing. irpc finds it with [`tracing::Dispatch::downcast_ref`]. +pub struct PropagatorLayer(Box); + +impl PropagatorLayer { + /// Creates a layer for `propagator`. + pub fn new(propagator: impl Propagator) -> Self { + Self(Box::new(propagator)) + } } +impl std::fmt::Debug for PropagatorLayer { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PropagatorLayer").finish_non_exhaustive() + } +} + +impl tracing_subscriber::Layer for PropagatorLayer {} + /// Runs `fut` with `carrier` in scope for [`set_span_parent_from_remote`]. /// /// The server loop calls this for each request. Most users do not call it. @@ -99,10 +106,12 @@ pub async fn scope_remote(carrier: Option, fut: F /// Sets the parent of `span` from the span context of the current request. /// /// The code from `rpc_requests(span_propagation)` calls this. It does nothing -/// outside of [`scope_remote`] or without a propagator. +/// outside of [`scope_remote`], or if the subscriber of `span` has no propagator. pub fn set_span_parent_from_remote(span: &tracing::Span) { - let Some(propagator) = PROPAGATOR.get() else { - return; - }; - let _ = SPAN_CONTEXT.try_with(|carrier| propagator.set_parent(span, carrier)); + span.with_subscriber(|(_id, dispatch)| { + let Some(propagator) = dispatch.downcast_ref::() else { + return; + }; + let _ = SPAN_CONTEXT.try_with(|carrier| propagator.0.set_parent(span, carrier)); + }); } diff --git a/tests/span_propagator.rs b/tests/span_propagator.rs index 86896632..736c9705 100644 --- a/tests/span_propagator.rs +++ b/tests/span_propagator.rs @@ -1,6 +1,4 @@ //! Checks that a custom [`Propagator`] carries headers from client to server. -//! -//! Lives in its own test binary, because the propagator is global. use std::{ net::{Ipv4Addr, SocketAddr, SocketAddrV4}, @@ -16,12 +14,13 @@ use irpc::{ noq::{listen, make_client_endpoint, make_server_endpoint}, rpc::{Handler, RemoteService}, rpc_requests, - span_propagation::{Propagator, SpanContextCarrier, set_propagator}, + span_propagation::{Propagator, PropagatorLayer, SpanContextCarrier}, }; use n0_future::task::AbortOnDropHandle; use noq::Endpoint; use serde::{Deserialize, Serialize}; use testresult::TestResult; +use tracing_subscriber::{Registry, layer::SubscriberExt}; fn create_connected_endpoints() -> TestResult<(Endpoint, Endpoint, SocketAddr)> { let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0).into(); @@ -62,7 +61,9 @@ impl Propagator for TestPropagator { #[tokio::test] async fn propagator_carries_headers_to_the_server() -> TestResult<()> { let propagator = TestPropagator::default(); - set_propagator(propagator.clone())?; + let layer = PropagatorLayer::new(propagator.clone()); + // The test runtime has one thread, so the server task sees this subscriber too. + let _guard = tracing::subscriber::set_default(Registry::default().with(layer)); let (server, client, server_addr) = create_connected_endpoints()?; let handler: Handler = Arc::new(|request, rx, tx| { From 70ac8d95b9fb3e2479d0d28572fdd5bc67c06749 Mon Sep 17 00:00:00 2001 From: Frando Date: Fri, 2 Oct 2026 12:51:18 +0200 Subject: [PATCH 03/12] feat!: put the span propagator behind the `span-propagation` feature With the previous commit, `rpc` enables `tracing-subscriber`, and `PropagatorLayer` puts its `Layer` trait into the public API of irpc. `tracing-subscriber` is 0.x, and most users of irpc do not propagate span context. This PR moves `Propagator`, `PropagatorLayer`, and the task-local of the server behind the new feature `span-propagation`, which is not a default feature. `irpc-opentelemetry` enables it. The wire format stays under `rpc`: a protocol with `span_propagation` always sends `Option`. Without the feature, the client sends `None` and the server ignores the carrier, the same as without a layer. `set_span_parent_from_remote` stays without the feature, because the code of the macro calls it, and does nothing. `tokio/rt`, which only the task-local needs, moves from `rpc` to the new feature. * changed: `span_propagation::Propagator` and `span_propagation::PropagatorLayer` need the feature `span-propagation`. --- .github/workflows/ci.yml | 2 ++ Cargo.toml | 8 +++++--- irpc-opentelemetry/Cargo.toml | 2 +- src/lib.rs | 4 ++-- src/span_propagation.rs | 26 ++++++++++++++++++++++++-- 5 files changed, 34 insertions(+), 8 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f23b1b35..391374ef 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -87,6 +87,8 @@ jobs: run: cargo check --locked --no-default-features --features spans --bins --tests --examples - name: cargo check (irpc, feature rpc) run: cargo check --locked --no-default-features --features rpc --bins --tests --examples + - name: cargo check (irpc, feature span-propagation) + run: cargo check --locked --no-default-features --features span-propagation --bins --tests --examples - name: cargo check (irpc-opentelemetry) run: cargo check --locked -p irpc-opentelemetry --bins --tests --examples diff --git a/Cargo.toml b/Cargo.toml index 61e45ab2..15a9f9da 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -85,7 +85,9 @@ n0-tracing-test = "0.3.0" [features] # enable the remote transport -rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "dep:tracing-subscriber", "tokio/io-util", "tokio/rt"] +rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "tokio/io-util"] +# propagate span context to the remote with a `span_propagation::PropagatorLayer` +span-propagation = ["rpc", "dep:tracing-subscriber", "tokio/rt"] # enable the iroh transport iroh = ["rpc", "dep:iroh", "dep:portable-atomic", "dep:getrandom"] # add helpers to create noq endpoints, for tests and for rpc on localhost @@ -143,7 +145,7 @@ required-features = ["iroh", "spans", "stream", "derive"] [[test]] name = "span_propagator" -required-features = ["noq_endpoint_setup", "derive", "spans"] +required-features = ["noq_endpoint_setup", "derive", "spans", "span-propagation"] [workspace] resolver = "2" @@ -201,7 +203,7 @@ allowed_external_types = [ "futures_core::stream::Stream", "futures_lite::future::Boxed", "futures_sink::Sink", - # `span_propagation::PropagatorLayer` is a tracing layer + # `span_propagation::PropagatorLayer` is a tracing layer, with the `span-propagation` feature "tracing_core::subscriber::Subscriber", "tracing_subscriber::layer::Layer", ] diff --git a/irpc-opentelemetry/Cargo.toml b/irpc-opentelemetry/Cargo.toml index d37a9697..cbe81248 100644 --- a/irpc-opentelemetry/Cargo.toml +++ b/irpc-opentelemetry/Cargo.toml @@ -11,7 +11,7 @@ description = "OpenTelemetry span propagation for irpc" rust-version = "1.91" [dependencies] -irpc = { version = "0.17.0", path = "..", default-features = false, features = ["rpc"] } +irpc = { version = "0.17.0", path = "..", default-features = false, features = ["span-propagation"] } opentelemetry = "0.32" tracing = { workspace = true } tracing-opentelemetry = "0.33" diff --git a/src/lib.rs b/src/lib.rs index f6550fdc..0f15fe9f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -223,8 +223,8 @@ use std::{fmt::Debug, future::Future, io, marker::PhantomData, ops::Deref}; /// * `span_propagation` *(optional, no value)*: If set, each request carries the span context /// of the client, as `(Option, Message)` on the wire. The generated /// `RemoteService` implementation sets the parent of the request span from it. This needs a -/// [`span_propagation::Propagator`] in the tracing subscriber, for example the layer of the -/// `irpc-opentelemetry` crate. +/// propagator in the tracing subscriber, for example the layer of the `irpc-opentelemetry` +/// crate, and the `span-propagation` feature of irpc. /// /// ## Variant attributes /// diff --git a/src/span_propagation.rs b/src/span_propagation.rs index 7a4cbee0..0d784d13 100644 --- a/src/span_propagation.rs +++ b/src/span_propagation.rs @@ -9,10 +9,14 @@ //! the headers on the server. //! //! irpc finds the propagator through the tracing subscriber: add a -//! [`PropagatorLayer`] to it. The client asks the subscriber of the current +//! `PropagatorLayer` to it. The client asks the subscriber of the current //! thread, and the server asks the subscriber of the request span. Without the //! layer, irpc does not propagate span context. //! +//! The propagator and the layer need the `span-propagation` feature. Without +//! it, a protocol with `span_propagation` keeps its wire format, but irpc does +//! not propagate span context. +//! //! The propagator does not change the wire format: a protocol with //! `span_propagation` always sends the `Option`. Without a //! propagator, its value is `None`. @@ -23,6 +27,7 @@ use std::{collections::HashMap, future::Future}; use serde::{Deserialize, Serialize}; +#[cfg(feature = "span-propagation")] tokio::task_local! { static SPAN_CONTEXT: SpanContextCarrier; } @@ -55,6 +60,9 @@ impl SpanContextCarrier { /// Returns the context of the current span, if the subscriber has a propagator. pub(crate) fn from_current() -> Option { + #[cfg(not(feature = "span-propagation"))] + return None; + #[cfg(feature = "span-propagation")] tracing::dispatcher::get_default(|dispatch| { let propagator = dispatch.downcast_ref::()?; let mut carrier = Self::default(); @@ -65,6 +73,7 @@ impl SpanContextCarrier { } /// Connects span propagation to a tracing backend. +#[cfg(feature = "span-propagation")] pub trait Propagator: Send + Sync + 'static { /// Writes the context of `span` into `carrier`. fn inject(&self, span: &tracing::Span, carrier: &mut SpanContextCarrier); @@ -76,8 +85,10 @@ pub trait Propagator: Send + Sync + 'static { /// A tracing layer that hands out a [`Propagator`] to irpc. /// /// The layer records nothing. irpc finds it with [`tracing::Dispatch::downcast_ref`]. +#[cfg(feature = "span-propagation")] pub struct PropagatorLayer(Box); +#[cfg(feature = "span-propagation")] impl PropagatorLayer { /// Creates a layer for `propagator`. pub fn new(propagator: impl Propagator) -> Self { @@ -85,18 +96,25 @@ impl PropagatorLayer { } } +#[cfg(feature = "span-propagation")] impl std::fmt::Debug for PropagatorLayer { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("PropagatorLayer").finish_non_exhaustive() } } +#[cfg(feature = "span-propagation")] impl tracing_subscriber::Layer for PropagatorLayer {} /// Runs `fut` with `carrier` in scope for [`set_span_parent_from_remote`]. /// /// The server loop calls this for each request. Most users do not call it. pub async fn scope_remote(carrier: Option, fut: F) -> F::Output { + #[cfg(not(feature = "span-propagation"))] + let _ = carrier; + #[cfg(not(feature = "span-propagation"))] + return fut.await; + #[cfg(feature = "span-propagation")] match carrier { Some(carrier) => SPAN_CONTEXT.scope(carrier, fut).await, None => fut.await, @@ -106,8 +124,12 @@ pub async fn scope_remote(carrier: Option, fut: F /// Sets the parent of `span` from the span context of the current request. /// /// The code from `rpc_requests(span_propagation)` calls this. It does nothing -/// outside of [`scope_remote`], or if the subscriber of `span` has no propagator. +/// outside of [`scope_remote`], if the subscriber of `span` has no propagator, or +/// without the `span-propagation` feature. pub fn set_span_parent_from_remote(span: &tracing::Span) { + #[cfg(not(feature = "span-propagation"))] + let _ = span; + #[cfg(feature = "span-propagation")] span.with_subscriber(|(_id, dispatch)| { let Some(propagator) = dispatch.downcast_ref::() else { return; From f16aa7d09c923138ccc9bc677f586bd1ef749c36 Mon Sep 17 00:00:00 2001 From: Frando Date: Mon, 5 Oct 2026 10:03:03 +0200 Subject: [PATCH 04/12] cleanups --- irpc-opentelemetry/tests/span_propagation.rs | 1 - src/span_propagation.rs | 14 +++++++------- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/irpc-opentelemetry/tests/span_propagation.rs b/irpc-opentelemetry/tests/span_propagation.rs index f1a799d2..bff633c7 100644 --- a/irpc-opentelemetry/tests/span_propagation.rs +++ b/irpc-opentelemetry/tests/span_propagation.rs @@ -57,7 +57,6 @@ mod span_propagation { let subscriber = Registry::default() .with(telemetry) .with(irpc_opentelemetry::layer()) - .with(irpc_opentelemetry::layer()) .with(tracing_subscriber::fmt::layer().with_filter(EnvFilter::from_default_env())); tracing::subscriber::set_global_default(subscriber).expect("global already set"); diff --git a/src/span_propagation.rs b/src/span_propagation.rs index 0d784d13..73d1ee07 100644 --- a/src/span_propagation.rs +++ b/src/span_propagation.rs @@ -13,13 +13,10 @@ //! thread, and the server asks the subscriber of the request span. Without the //! layer, irpc does not propagate span context. //! -//! The propagator and the layer need the `span-propagation` feature. Without -//! it, a protocol with `span_propagation` keeps its wire format, but irpc does -//! not propagate span context. -//! -//! The propagator does not change the wire format: a protocol with -//! `span_propagation` always sends the `Option`. Without a -//! propagator, its value is `None`. +//! The propagator and the layer need the `span-propagation` feature. Neither +//! the feature nor the layer changes the wire format: a protocol with +//! `span_propagation` always sends an `Option`, which is +//! `None` when irpc does not propagate span context. //! //! The `irpc-opentelemetry` crate has a propagator for OpenTelemetry. @@ -109,6 +106,9 @@ impl tracing_subscriber::Layer for PropagatorLayer {} /// Runs `fut` with `carrier` in scope for [`set_span_parent_from_remote`]. /// /// The server loop calls this for each request. Most users do not call it. +/// +/// The carrier lives in a task-local of `fut`, so concurrent requests do not see +/// each other's context, and the context moves with `fut` across threads. pub async fn scope_remote(carrier: Option, fut: F) -> F::Output { #[cfg(not(feature = "span-propagation"))] let _ = carrier; From b1990b9cf3c920b8126150367d853458a43477e4 Mon Sep 17 00:00:00 2001 From: Frando Date: Mon, 5 Oct 2026 10:09:02 +0200 Subject: [PATCH 05/12] fix(irpc-opentelemetry): keep the local parent if the request has no span context Without context activation in tracing-opentelemetry, an empty carrier made the request span a new root. Only set the parent if the extracted context has a valid span. --- Cargo.lock | 1 + irpc-opentelemetry/Cargo.toml | 3 +- irpc-opentelemetry/src/lib.rs | 12 ++- irpc-opentelemetry/tests/local_parent.rs | 113 +++++++++++++++++++++++ 4 files changed, 126 insertions(+), 3 deletions(-) create mode 100644 irpc-opentelemetry/tests/local_parent.rs diff --git a/Cargo.lock b/Cargo.lock index eb659426..1154936f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1578,6 +1578,7 @@ dependencies = [ "iroh", "irpc", "n0-error", + "n0-future", "opentelemetry", "opentelemetry-otlp", "opentelemetry_sdk", diff --git a/irpc-opentelemetry/Cargo.toml b/irpc-opentelemetry/Cargo.toml index cbe81248..9c0d6d24 100644 --- a/irpc-opentelemetry/Cargo.toml +++ b/irpc-opentelemetry/Cargo.toml @@ -21,8 +21,9 @@ anyhow = { workspace = true } clap = { version = "4.5.41", features = ["derive"] } # Tests and examples need a crypto provider, which iroh's default features supply. iroh = { workspace = true, default-features = true } -irpc = { version = "0.17.0", path = "..", features = ["iroh", "derive", "spans"] } +irpc = { version = "0.17.0", path = "..", features = ["iroh", "derive", "spans", "noq_endpoint_setup"] } n0-error = { workspace = true } +n0-future = { workspace = true } opentelemetry_sdk = { version = "0.32", features = ["testing"] } opentelemetry-otlp = { version = "0.32", default-features = false, features = ["http-proto", "reqwest-blocking-client", "trace"] } rand = "0.10.3" diff --git a/irpc-opentelemetry/src/lib.rs b/irpc-opentelemetry/src/lib.rs index b2491ec9..603fb6a1 100644 --- a/irpc-opentelemetry/src/lib.rs +++ b/irpc-opentelemetry/src/lib.rs @@ -29,7 +29,10 @@ use irpc::span_propagation::{Propagator, PropagatorLayer, SpanContextCarrier}; pub use opentelemetry; -use opentelemetry::propagation::{Extractor, Injector}; +use opentelemetry::{ + propagation::{Extractor, Injector}, + trace::TraceContextExt, +}; pub use tracing_opentelemetry; use tracing_opentelemetry::OpenTelemetrySpanExt; @@ -51,7 +54,12 @@ impl Propagator for OtelPropagator { propagator .extract_with_context(&opentelemetry::Context::current(), &HeadersRef(carrier)) }); - let _ = span.set_parent(context); + // If the headers have no span, `context` is the current context. Without + // context activation in `tracing-opentelemetry`, it has no span either, and + // setting it as parent would make `span` a new root. + if context.span().span_context().is_valid() { + let _ = span.set_parent(context); + } } } diff --git a/irpc-opentelemetry/tests/local_parent.rs b/irpc-opentelemetry/tests/local_parent.rs new file mode 100644 index 00000000..122bf782 --- /dev/null +++ b/irpc-opentelemetry/tests/local_parent.rs @@ -0,0 +1,113 @@ +//! Checks that a request without a client span keeps its local parent on the server. +//! +//! A client with the irpc layer but no active span sends an empty carrier. The +//! server must not set the parent of the request span from it. With context +//! activation off, the current OpenTelemetry context is empty, and setting it as +//! parent would make the request span a new root. + +use std::{ + net::{Ipv4Addr, SocketAddrV4}, + sync::Arc, +}; + +use irpc::{ + Client, WithChannels, + channel::oneshot, + noq::{listen, make_client_endpoint, make_server_endpoint}, + rpc::{Handler, RemoteService}, + rpc_requests, +}; +use n0_error::Result; +use n0_future::task::AbortOnDropHandle; +use opentelemetry::trace::TracerProvider; +use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider}; +use serde::{Deserialize, Serialize}; +use tracing::{Instrument, info_span}; +use tracing_subscriber::{Registry, layer::SubscriberExt}; + +#[rpc_requests(message = EchoMessage, span_propagation)] +#[derive(Debug, Serialize, Deserialize)] +enum EchoProtocol { + #[rpc(tx = oneshot::Sender)] + #[wrap(Echo)] + Echo(String), +} + +#[tokio::test] +async fn request_without_client_span_keeps_local_parent() -> Result<()> { + check_local_parent(true).await +} + +#[tokio::test] +async fn request_without_client_span_keeps_local_parent_without_context_activation() -> Result<()> { + check_local_parent(false).await +} + +async fn check_local_parent(context_activation: bool) -> Result<()> { + opentelemetry::global::set_text_map_propagator( + opentelemetry_sdk::propagation::TraceContextPropagator::new(), + ); + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = Registry::default() + .with( + tracing_opentelemetry::layer() + .with_tracer(provider.tracer("test")) + .with_context_activation(context_activation), + ) + .with(irpc_opentelemetry::layer()); + // The test runtime has one thread, so the server task sees this subscriber too. + let _guard = tracing::subscriber::set_default(subscriber); + + let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0).into(); + let (server, cert) = make_server_endpoint(addr)?; + let client = make_client_endpoint(addr, &[cert.as_slice()])?; + let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server.local_addr()?.port()).into(); + + // Spans are exported when they close, so the handler reports when its spans are closed. + let (done_tx, mut done_rx) = tokio::sync::mpsc::unbounded_channel(); + let handler: Handler = Arc::new(move |request, rx, tx| { + let done_tx = done_tx.clone(); + Box::pin(async move { + async move { + let EchoMessage::Echo(WithChannels { + inner, tx, span, .. + }) = request.with_remote_channels(rx, tx); + let _guard = span.enter(); + tx.send(inner.0).await.ok(); + } + .instrument(info_span!("server-handler")) + .await; + done_tx.send(()).ok(); + Ok(()) + }) + }); + let _server = AbortOnDropHandle::new(tokio::spawn(listen::(server, handler))); + let client = Client::::noq(client, server_addr); + // No span is active here, so the client sends an empty carrier. + assert_eq!(client.rpc(Echo("a".into())).await?, "a"); + done_rx.recv().await.expect("handler is alive"); + + let _ = provider.force_flush(); + let spans = exporter.get_finished_spans().expect("exporter is open"); + let find = |name: &str| { + spans + .iter() + .find(|s| s.name == name) + .unwrap_or_else(|| panic!("no span named {name}")) + }; + let handler_span = find("server-handler"); + let request_span = find("Echo"); + assert_eq!( + request_span.span_context.trace_id(), + handler_span.span_context.trace_id() + ); + assert_eq!( + request_span.parent_span_id, + handler_span.span_context.span_id() + ); + assert!(!request_span.parent_span_is_remote); + Ok(()) +} From 8dccbb302f52468ec566a6682a3a91bf493a4627 Mon Sep 17 00:00:00 2001 From: Frando Date: Mon, 5 Oct 2026 10:18:35 +0200 Subject: [PATCH 06/12] refactor(irpc-opentelemetry): drop re-exports, add README The crate has no opentelemetry types in its API, so the re-exports are not needed. The README and crate docs state the version bound on opentelemetry and tracing-opentelemetry. --- irpc-opentelemetry/Cargo.toml | 3 -- irpc-opentelemetry/README.md | 62 +++++++++++++++++++++++++++++++++++ irpc-opentelemetry/src/lib.rs | 10 +++--- 3 files changed, 66 insertions(+), 9 deletions(-) create mode 100644 irpc-opentelemetry/README.md diff --git a/irpc-opentelemetry/Cargo.toml b/irpc-opentelemetry/Cargo.toml index 9c0d6d24..03cf1910 100644 --- a/irpc-opentelemetry/Cargo.toml +++ b/irpc-opentelemetry/Cargo.toml @@ -34,9 +34,6 @@ tracing-subscriber = { workspace = true, features = ["fmt", "env-filter"] } [package.metadata.cargo_check_external_types] allowed_external_types = [ "irpc::*", - # Re-exported. This crate is 0.x and follows the version of opentelemetry. - "opentelemetry", - "tracing_opentelemetry", ] [package.metadata.release] diff --git a/irpc-opentelemetry/README.md b/irpc-opentelemetry/README.md new file mode 100644 index 00000000..0da2ef86 --- /dev/null +++ b/irpc-opentelemetry/README.md @@ -0,0 +1,62 @@ +# irpc-opentelemetry + +OpenTelemetry span propagation for [irpc](https://github.com/n0-computer/irpc). + +The client writes the context of the current span into each request, and the +server sets the parent of the request span from it, so a trace continues across +irpc calls. + +## Version compatibility + +Your application must use the same semver-compatible versions of +`opentelemetry` and `tracing-opentelemetry` as this crate: + +| irpc-opentelemetry | opentelemetry | tracing-opentelemetry | +|--------------------|---------------|-----------------------| +| 0.1 | 0.32 | 0.33 | + +Both crates keep their state in statics: the global propagator and the +`tracing-opentelemetry` layer. With a second copy of either in your dependency +tree, everything compiles, but irpc silently propagates nothing. To check, +run `cargo tree -i opentelemetry` and `cargo tree -i tracing-opentelemetry`: +each should print one version. If cargo reports that the name is ambiguous, +you have two copies. + +This crate makes a breaking release whenever it moves to a new version of +`opentelemetry` or `tracing-opentelemetry`. + +## Usage + +Enable span propagation on the protocol with +`#[rpc_requests(..., span_propagation)]`, and add the layer of this crate to the +tracing subscriber, next to the `tracing-opentelemetry` layer: + +```rust +use tracing_subscriber::{Registry, layer::SubscriberExt}; + +opentelemetry::global::set_text_map_propagator( + opentelemetry_sdk::propagation::TraceContextPropagator::new(), +); +let tracer = opentelemetry::global::tracer("app"); +let subscriber = Registry::default() + .with(tracing_opentelemetry::layer().with_tracer(tracer)) + .with(irpc_opentelemetry::layer()); +tracing::subscriber::set_global_default(subscriber)?; +``` + +Client and server both need the layer. See +[`examples/span_propagation.rs`](examples/span_propagation.rs) for a full +example that exports to Jaeger. + +## License + +Copyright 2026 N0, INC. + +This project is licensed under either of + +- Apache License, Version 2.0, ([LICENSE-APACHE](../LICENSE-APACHE) or + http://www.apache.org/licenses/LICENSE-2.0) +- MIT license ([LICENSE-MIT](../LICENSE-MIT) or + http://opensource.org/licenses/MIT) + +at your option. diff --git a/irpc-opentelemetry/src/lib.rs b/irpc-opentelemetry/src/lib.rs index 603fb6a1..7815690f 100644 --- a/irpc-opentelemetry/src/lib.rs +++ b/irpc-opentelemetry/src/lib.rs @@ -22,18 +22,16 @@ //! tracing::subscriber::set_global_default(subscriber).expect("no other subscriber is set"); //! ``` //! -//! Spans reach OpenTelemetry through the `tracing-opentelemetry` layer. This -//! crate uses the versions of `opentelemetry` and `tracing-opentelemetry` that -//! it re-exports. The application must use the same versions, because both -//! crates keep their state in statics. +//! Spans reach OpenTelemetry through the `tracing-opentelemetry` layer. The +//! application must use semver-compatible versions of `opentelemetry` and +//! `tracing-opentelemetry` with this crate: both keep their state in statics, +//! and with a second copy of either, irpc silently propagates nothing. use irpc::span_propagation::{Propagator, PropagatorLayer, SpanContextCarrier}; -pub use opentelemetry; use opentelemetry::{ propagation::{Extractor, Injector}, trace::TraceContextExt, }; -pub use tracing_opentelemetry; use tracing_opentelemetry::OpenTelemetrySpanExt; /// A [`Propagator`] that uses the global text map propagator of `opentelemetry`. From 994a0e8d93fe5f0493f2a9bfecee7cd61d5f2911 Mon Sep 17 00:00:00 2001 From: Frando Date: Mon, 5 Oct 2026 10:20:51 +0200 Subject: [PATCH 07/12] fix docs and add readme --- irpc-opentelemetry/README.md | 40 +++++++++++++++++++++-------------- irpc-opentelemetry/src/lib.rs | 12 +++++++---- 2 files changed, 32 insertions(+), 20 deletions(-) diff --git a/irpc-opentelemetry/README.md b/irpc-opentelemetry/README.md index 0da2ef86..5f560029 100644 --- a/irpc-opentelemetry/README.md +++ b/irpc-opentelemetry/README.md @@ -8,22 +8,30 @@ irpc calls. ## Version compatibility -Your application must use the same semver-compatible versions of -`opentelemetry` and `tracing-opentelemetry` as this crate: - -| irpc-opentelemetry | opentelemetry | tracing-opentelemetry | -|--------------------|---------------|-----------------------| -| 0.1 | 0.32 | 0.33 | - -Both crates keep their state in statics: the global propagator and the -`tracing-opentelemetry` layer. With a second copy of either in your dependency -tree, everything compiles, but irpc silently propagates nothing. To check, -run `cargo tree -i opentelemetry` and `cargo tree -i tracing-opentelemetry`: -each should print one version. If cargo reports that the name is ambiguous, -you have two copies. - -This crate makes a breaking release whenever it moves to a new version of -`opentelemetry` or `tracing-opentelemetry`. +Your application must use semver-compatible versions of `irpc`, +`opentelemetry`, and `tracing-opentelemetry` with this crate: + +| irpc-opentelemetry | irpc | opentelemetry | tracing-opentelemetry | +|--------------------|------|---------------|-----------------------| +| 0.1 | 0.17 | 0.32 | 0.33 | + +With a second copy of one of them in your dependency tree, everything +compiles, but no span context reaches the remote: + +- `opentelemetry` keeps the propagator that you register with + `opentelemetry::global::set_text_map_propagator` in a static. A second copy + has its own static, with a propagator that does nothing. +- This crate finds the `tracing-opentelemetry` layer in the subscriber by its + type. The layer type of a second copy is a different type. +- irpc finds the layer of this crate in the subscriber by its type, too. The + layer type of a second copy of irpc is a different type. + +To check, run `cargo tree -i irpc`, `cargo tree -i opentelemetry`, and +`cargo tree -i tracing-opentelemetry`: each should print one version. If cargo +reports that the name is ambiguous, you have two copies. + +This crate makes a breaking release whenever it moves to a new major version +of `irpc`, or a new version of `opentelemetry` or `tracing-opentelemetry`. ## Usage diff --git a/irpc-opentelemetry/src/lib.rs b/irpc-opentelemetry/src/lib.rs index 7815690f..9d1b1ab8 100644 --- a/irpc-opentelemetry/src/lib.rs +++ b/irpc-opentelemetry/src/lib.rs @@ -22,10 +22,14 @@ //! tracing::subscriber::set_global_default(subscriber).expect("no other subscriber is set"); //! ``` //! -//! Spans reach OpenTelemetry through the `tracing-opentelemetry` layer. The -//! application must use semver-compatible versions of `opentelemetry` and -//! `tracing-opentelemetry` with this crate: both keep their state in statics, -//! and with a second copy of either, irpc silently propagates nothing. +//! Spans reach OpenTelemetry through the `tracing-opentelemetry` layer. +//! +//! The application must use semver-compatible versions of `irpc`, +//! `opentelemetry`, and `tracing-opentelemetry` with this crate. With a second +//! copy of one of them, everything compiles, but no span context reaches the +//! remote: `opentelemetry` keeps the registered propagator in a static, and +//! this crate and irpc find their layers in the subscriber by type, so they +//! do not see the static or the layer types of another copy. use irpc::span_propagation::{Propagator, PropagatorLayer, SpanContextCarrier}; use opentelemetry::{ From 57f20ac58e43427882f7028c8c6593bf9251badb Mon Sep 17 00:00:00 2001 From: Frando Date: Mon, 5 Oct 2026 17:02:33 +0200 Subject: [PATCH 08/12] fix: filter the span propagation layer to keep the max level An unfiltered layer enables everything. With per-layer filters on the other layers, adding it raised the max level to TRACE, so every `debug!` and `trace!` reached the other layers. `span_propagation::layer(propagator)` now returns the layer with a `LevelFilter::OFF` per-layer filter. `PropagatorLayer` is private. * removed: `span_propagation::PropagatorLayer`. Use `span_propagation::layer`. * changed: `irpc_opentelemetry::layer` returns `impl Layer`, and needs a subscriber that implements `LookupSpan`. --- Cargo.toml | 7 ++++--- irpc-opentelemetry/Cargo.toml | 5 +++++ irpc-opentelemetry/src/lib.rs | 11 ++++++++--- src/span_propagation.rs | 33 +++++++++++++++++---------------- tests/span_propagator.rs | 22 +++++++++++++++++----- 5 files changed, 51 insertions(+), 27 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 15a9f9da..f3d60d4b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,7 @@ n0-error = { workspace = true } # used in the endpoint handler code when using rpc tracing = { workspace = true, optional = true } -tracing-subscriber = { version = "0.3.23", default-features = false, optional = true } +tracing-subscriber = { version = "0.3.23", default-features = false, features = ["registry"], optional = true } # used to ser/de messages when using rpc postcard = { workspace = true, features = ["alloc", "use-std"], optional = true } # currently only transport when using rpc @@ -86,7 +86,7 @@ n0-tracing-test = "0.3.0" [features] # enable the remote transport rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "tokio/io-util"] -# propagate span context to the remote with a `span_propagation::PropagatorLayer` +# propagate span context to the remote with a `span_propagation::layer` span-propagation = ["rpc", "dep:tracing-subscriber", "tokio/rt"] # enable the iroh transport iroh = ["rpc", "dep:iroh", "dep:portable-atomic", "dep:getrandom"] @@ -203,7 +203,8 @@ allowed_external_types = [ "futures_core::stream::Stream", "futures_lite::future::Boxed", "futures_sink::Sink", - # `span_propagation::PropagatorLayer` is a tracing layer, with the `span-propagation` feature + # `span_propagation::layer` returns a tracing layer, with the `span-propagation` feature "tracing_core::subscriber::Subscriber", "tracing_subscriber::layer::Layer", + "tracing_subscriber::registry::LookupSpan", ] diff --git a/irpc-opentelemetry/Cargo.toml b/irpc-opentelemetry/Cargo.toml index 03cf1910..2f252e3a 100644 --- a/irpc-opentelemetry/Cargo.toml +++ b/irpc-opentelemetry/Cargo.toml @@ -15,6 +15,7 @@ irpc = { version = "0.17.0", path = "..", default-features = false, features = [ opentelemetry = "0.32" tracing = { workspace = true } tracing-opentelemetry = "0.33" +tracing-subscriber = { version = "0.3.23", default-features = false, features = ["registry"] } [dev-dependencies] anyhow = { workspace = true } @@ -34,6 +35,10 @@ tracing-subscriber = { workspace = true, features = ["fmt", "env-filter"] } [package.metadata.cargo_check_external_types] allowed_external_types = [ "irpc::*", + # `layer` returns a tracing layer + "tracing_core::subscriber::Subscriber", + "tracing_subscriber::layer::Layer", + "tracing_subscriber::registry::LookupSpan", ] [package.metadata.release] diff --git a/irpc-opentelemetry/src/lib.rs b/irpc-opentelemetry/src/lib.rs index 9d1b1ab8..a7fecf38 100644 --- a/irpc-opentelemetry/src/lib.rs +++ b/irpc-opentelemetry/src/lib.rs @@ -31,12 +31,14 @@ //! this crate and irpc find their layers in the subscriber by type, so they //! do not see the static or the layer types of another copy. -use irpc::span_propagation::{Propagator, PropagatorLayer, SpanContextCarrier}; +use irpc::span_propagation::{Propagator, SpanContextCarrier}; use opentelemetry::{ propagation::{Extractor, Injector}, trace::TraceContextExt, }; +use tracing::Subscriber; use tracing_opentelemetry::OpenTelemetrySpanExt; +use tracing_subscriber::{Layer, registry::LookupSpan}; /// A [`Propagator`] that uses the global text map propagator of `opentelemetry`. #[derive(Debug, Default, Clone, Copy)] @@ -66,8 +68,11 @@ impl Propagator for OtelPropagator { } /// Returns a tracing layer that makes irpc propagate span context with [`OtelPropagator`]. -pub fn layer() -> PropagatorLayer { - PropagatorLayer::new(OtelPropagator) +pub fn layer() -> impl Layer +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + irpc::span_propagation::layer(OtelPropagator) } /// Writes headers into a [`SpanContextCarrier`]. diff --git a/src/span_propagation.rs b/src/span_propagation.rs index 73d1ee07..1a32247b 100644 --- a/src/span_propagation.rs +++ b/src/span_propagation.rs @@ -8,8 +8,8 @@ //! of a span into the headers on the client, and sets the parent of a span from //! the headers on the server. //! -//! irpc finds the propagator through the tracing subscriber: add a -//! `PropagatorLayer` to it. The client asks the subscriber of the current +//! irpc finds the propagator through the tracing subscriber: add the [`layer`] +//! for it to the subscriber. The client asks the subscriber of the current //! thread, and the server asks the subscriber of the request span. Without the //! layer, irpc does not propagate span context. //! @@ -79,26 +79,27 @@ pub trait Propagator: Send + Sync + 'static { fn set_parent(&self, span: &tracing::Span, carrier: &SpanContextCarrier); } -/// A tracing layer that hands out a [`Propagator`] to irpc. +/// Returns a tracing layer that makes irpc propagate span context with `propagator`. /// /// The layer records nothing. irpc finds it with [`tracing::Dispatch::downcast_ref`]. +/// +/// The layer has a per-layer filter that disables all spans and events. Without +/// it, the layer would enable all levels for the subscriber when other layers +/// use per-layer filters, and every `debug!` and `trace!` would reach them. +/// Another layer must enable the request spans, for example the layer of the +/// tracing backend. #[cfg(feature = "span-propagation")] -pub struct PropagatorLayer(Box); - -#[cfg(feature = "span-propagation")] -impl PropagatorLayer { - /// Creates a layer for `propagator`. - pub fn new(propagator: impl Propagator) -> Self { - Self(Box::new(propagator)) - } +pub fn layer(propagator: impl Propagator) -> impl tracing_subscriber::Layer +where + S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>, +{ + use tracing_subscriber::Layer; + PropagatorLayer(Box::new(propagator)).with_filter(tracing::level_filters::LevelFilter::OFF) } +/// The layer from [`layer`], without its filter. #[cfg(feature = "span-propagation")] -impl std::fmt::Debug for PropagatorLayer { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("PropagatorLayer").finish_non_exhaustive() - } -} +struct PropagatorLayer(Box); #[cfg(feature = "span-propagation")] impl tracing_subscriber::Layer for PropagatorLayer {} diff --git a/tests/span_propagator.rs b/tests/span_propagator.rs index 736c9705..8f83001c 100644 --- a/tests/span_propagator.rs +++ b/tests/span_propagator.rs @@ -1,4 +1,4 @@ -//! Checks that a custom [`Propagator`] carries headers from client to server. +//! Checks a custom [`Propagator`] and its tracing layer. use std::{ net::{Ipv4Addr, SocketAddr, SocketAddrV4}, @@ -14,13 +14,14 @@ use irpc::{ noq::{listen, make_client_endpoint, make_server_endpoint}, rpc::{Handler, RemoteService}, rpc_requests, - span_propagation::{Propagator, PropagatorLayer, SpanContextCarrier}, + span_propagation::{Propagator, SpanContextCarrier}, }; use n0_future::task::AbortOnDropHandle; use noq::Endpoint; use serde::{Deserialize, Serialize}; use testresult::TestResult; -use tracing_subscriber::{Registry, layer::SubscriberExt}; +use tracing::{Subscriber, level_filters::LevelFilter}; +use tracing_subscriber::{Layer, Registry, layer::SubscriberExt}; fn create_connected_endpoints() -> TestResult<(Endpoint, Endpoint, SocketAddr)> { let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0).into(); @@ -61,9 +62,11 @@ impl Propagator for TestPropagator { #[tokio::test] async fn propagator_carries_headers_to_the_server() -> TestResult<()> { let propagator = TestPropagator::default(); - let layer = PropagatorLayer::new(propagator.clone()); + let layer = irpc::span_propagation::layer(propagator.clone()); + // The irpc layer enables no spans, so the filter enables the request spans. // The test runtime has one thread, so the server task sees this subscriber too. - let _guard = tracing::subscriber::set_default(Registry::default().with(layer)); + let subscriber = Registry::default().with(LevelFilter::INFO).with(layer); + let _guard = tracing::subscriber::set_default(subscriber); let (server, client, server_addr) = create_connected_endpoints()?; let handler: Handler = Arc::new(|request, rx, tx| { @@ -83,3 +86,12 @@ async fn propagator_carries_headers_to_the_server() -> TestResult<()> { assert_eq!(received, ["0", "1"]); Ok(()) } + +/// The layer must not raise the max level when other layers use per-layer filters. +#[test] +fn layer_keeps_the_max_level() { + let subscriber = Registry::default() + .with(tracing_subscriber::fmt::layer().with_filter(LevelFilter::INFO)) + .with(irpc::span_propagation::layer(TestPropagator::default())); + assert_eq!(subscriber.max_level_hint(), Some(LevelFilter::INFO)); +} From 35dab1f13e3a284d2a68ba26e5e4294a011fb3ae Mon Sep 17 00:00:00 2001 From: Frando Date: Mon, 12 Oct 2026 10:11:16 +0200 Subject: [PATCH 09/12] feat!: rename the `span-propagation` feature to `tracing-subscriber-03` The layer from `span_propagation::layer` is a layer of tracing-subscriber 0.3, so the feature now names that version. It is not a default feature, and irpc-opentelemetry enables it. If tracing-subscriber 0.4 comes out, irpc can add a `tracing-subscriber-04` feature with its own layer function, and users only depend on the version they enable. Without a layer, the propagator does nothing, so this commit drops the separate `span-propagation` feature instead of keeping both. * changed: the feature `span-propagation` is now `tracing-subscriber-03`. --- .github/workflows/ci.yml | 4 ++-- Cargo.toml | 9 +++++---- irpc-opentelemetry/Cargo.toml | 2 +- src/lib.rs | 2 +- src/span_propagation.rs | 32 +++++++++++++++++--------------- 5 files changed, 26 insertions(+), 23 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 391374ef..7f7fb554 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -87,8 +87,8 @@ jobs: run: cargo check --locked --no-default-features --features spans --bins --tests --examples - name: cargo check (irpc, feature rpc) run: cargo check --locked --no-default-features --features rpc --bins --tests --examples - - name: cargo check (irpc, feature span-propagation) - run: cargo check --locked --no-default-features --features span-propagation --bins --tests --examples + - name: cargo check (irpc, feature tracing-subscriber-03) + run: cargo check --locked --no-default-features --features tracing-subscriber-03 --bins --tests --examples - name: cargo check (irpc-opentelemetry) run: cargo check --locked -p irpc-opentelemetry --bins --tests --examples diff --git a/Cargo.toml b/Cargo.toml index f3d60d4b..34f6c5ea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -86,8 +86,9 @@ n0-tracing-test = "0.3.0" [features] # enable the remote transport rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "tokio/io-util"] -# propagate span context to the remote with a `span_propagation::layer` -span-propagation = ["rpc", "dep:tracing-subscriber", "tokio/rt"] +# propagate span context to the remote with `span_propagation::layer`, a layer of +# tracing-subscriber 0.3. Not a default feature, irpc-opentelemetry enables it. +tracing-subscriber-03 = ["rpc", "dep:tracing-subscriber", "tokio/rt"] # enable the iroh transport iroh = ["rpc", "dep:iroh", "dep:portable-atomic", "dep:getrandom"] # add helpers to create noq endpoints, for tests and for rpc on localhost @@ -145,7 +146,7 @@ required-features = ["iroh", "spans", "stream", "derive"] [[test]] name = "span_propagator" -required-features = ["noq_endpoint_setup", "derive", "spans", "span-propagation"] +required-features = ["noq_endpoint_setup", "derive", "spans", "tracing-subscriber-03"] [workspace] resolver = "2" @@ -203,7 +204,7 @@ allowed_external_types = [ "futures_core::stream::Stream", "futures_lite::future::Boxed", "futures_sink::Sink", - # `span_propagation::layer` returns a tracing layer, with the `span-propagation` feature + # `span_propagation::layer` returns a tracing layer, with the `tracing-subscriber-03` feature "tracing_core::subscriber::Subscriber", "tracing_subscriber::layer::Layer", "tracing_subscriber::registry::LookupSpan", diff --git a/irpc-opentelemetry/Cargo.toml b/irpc-opentelemetry/Cargo.toml index 2f252e3a..1c474dd1 100644 --- a/irpc-opentelemetry/Cargo.toml +++ b/irpc-opentelemetry/Cargo.toml @@ -11,7 +11,7 @@ description = "OpenTelemetry span propagation for irpc" rust-version = "1.91" [dependencies] -irpc = { version = "0.17.0", path = "..", default-features = false, features = ["span-propagation"] } +irpc = { version = "0.17.0", path = "..", default-features = false, features = ["tracing-subscriber-03"] } opentelemetry = "0.32" tracing = { workspace = true } tracing-opentelemetry = "0.33" diff --git a/src/lib.rs b/src/lib.rs index 0f15fe9f..1d2b538a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -224,7 +224,7 @@ use std::{fmt::Debug, future::Future, io, marker::PhantomData, ops::Deref}; /// of the client, as `(Option, Message)` on the wire. The generated /// `RemoteService` implementation sets the parent of the request span from it. This needs a /// propagator in the tracing subscriber, for example the layer of the `irpc-opentelemetry` -/// crate, and the `span-propagation` feature of irpc. +/// crate, and the `tracing-subscriber-03` feature of irpc. /// /// ## Variant attributes /// diff --git a/src/span_propagation.rs b/src/span_propagation.rs index 1a32247b..9e6da692 100644 --- a/src/span_propagation.rs +++ b/src/span_propagation.rs @@ -13,8 +13,10 @@ //! thread, and the server asks the subscriber of the request span. Without the //! layer, irpc does not propagate span context. //! -//! The propagator and the layer need the `span-propagation` feature. Neither -//! the feature nor the layer changes the wire format: a protocol with +//! The propagator and the layer need the `tracing-subscriber-03` feature, +//! because the layer is a layer of `tracing-subscriber` 0.3. A later version +//! of `tracing-subscriber` would get its own feature. Neither the feature nor +//! the layer changes the wire format: a protocol with //! `span_propagation` always sends an `Option`, which is //! `None` when irpc does not propagate span context. //! @@ -24,7 +26,7 @@ use std::{collections::HashMap, future::Future}; use serde::{Deserialize, Serialize}; -#[cfg(feature = "span-propagation")] +#[cfg(feature = "tracing-subscriber-03")] tokio::task_local! { static SPAN_CONTEXT: SpanContextCarrier; } @@ -57,9 +59,9 @@ impl SpanContextCarrier { /// Returns the context of the current span, if the subscriber has a propagator. pub(crate) fn from_current() -> Option { - #[cfg(not(feature = "span-propagation"))] + #[cfg(not(feature = "tracing-subscriber-03"))] return None; - #[cfg(feature = "span-propagation")] + #[cfg(feature = "tracing-subscriber-03")] tracing::dispatcher::get_default(|dispatch| { let propagator = dispatch.downcast_ref::()?; let mut carrier = Self::default(); @@ -70,7 +72,7 @@ impl SpanContextCarrier { } /// Connects span propagation to a tracing backend. -#[cfg(feature = "span-propagation")] +#[cfg(feature = "tracing-subscriber-03")] pub trait Propagator: Send + Sync + 'static { /// Writes the context of `span` into `carrier`. fn inject(&self, span: &tracing::Span, carrier: &mut SpanContextCarrier); @@ -88,7 +90,7 @@ pub trait Propagator: Send + Sync + 'static { /// use per-layer filters, and every `debug!` and `trace!` would reach them. /// Another layer must enable the request spans, for example the layer of the /// tracing backend. -#[cfg(feature = "span-propagation")] +#[cfg(feature = "tracing-subscriber-03")] pub fn layer(propagator: impl Propagator) -> impl tracing_subscriber::Layer where S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>, @@ -98,10 +100,10 @@ where } /// The layer from [`layer`], without its filter. -#[cfg(feature = "span-propagation")] +#[cfg(feature = "tracing-subscriber-03")] struct PropagatorLayer(Box); -#[cfg(feature = "span-propagation")] +#[cfg(feature = "tracing-subscriber-03")] impl tracing_subscriber::Layer for PropagatorLayer {} /// Runs `fut` with `carrier` in scope for [`set_span_parent_from_remote`]. @@ -111,11 +113,11 @@ impl tracing_subscriber::Layer for PropagatorLayer {} /// The carrier lives in a task-local of `fut`, so concurrent requests do not see /// each other's context, and the context moves with `fut` across threads. pub async fn scope_remote(carrier: Option, fut: F) -> F::Output { - #[cfg(not(feature = "span-propagation"))] + #[cfg(not(feature = "tracing-subscriber-03"))] let _ = carrier; - #[cfg(not(feature = "span-propagation"))] + #[cfg(not(feature = "tracing-subscriber-03"))] return fut.await; - #[cfg(feature = "span-propagation")] + #[cfg(feature = "tracing-subscriber-03")] match carrier { Some(carrier) => SPAN_CONTEXT.scope(carrier, fut).await, None => fut.await, @@ -126,11 +128,11 @@ pub async fn scope_remote(carrier: Option, fut: F /// /// The code from `rpc_requests(span_propagation)` calls this. It does nothing /// outside of [`scope_remote`], if the subscriber of `span` has no propagator, or -/// without the `span-propagation` feature. +/// without the `tracing-subscriber-03` feature. pub fn set_span_parent_from_remote(span: &tracing::Span) { - #[cfg(not(feature = "span-propagation"))] + #[cfg(not(feature = "tracing-subscriber-03"))] let _ = span; - #[cfg(feature = "span-propagation")] + #[cfg(feature = "tracing-subscriber-03")] span.with_subscriber(|(_id, dispatch)| { let Some(propagator) = dispatch.downcast_ref::() else { return; From 229cf1248a55f3cad8487779bcbfdf9d904d9b58 Mon Sep 17 00:00:00 2001 From: Frando Date: Mon, 12 Oct 2026 10:56:58 +0200 Subject: [PATCH 10/12] docs: document the feature flags, shorten Cargo.toml comments The README gets a table of all features, and the feature list in the crate docs gets the missing `stream`, `varint-util`, and `tracing-subscriber-03`, and a clearer `spans` entry. The module docs of `span_propagation` no longer link to items that only exist with `tracing-subscriber-03`, so the docs build with default features. Also turns off the default features of tracing-opentelemetry in irpc-opentelemetry, and simplifies the cfg blocks in `scope_remote`. --- Cargo.lock | 2 -- Cargo.toml | 5 ++--- README.md | 17 +++++++++++++++++ irpc-opentelemetry/Cargo.toml | 5 ++--- src/lib.rs | 22 +++++++++++++++------- src/span_propagation.rs | 19 +++++++++++-------- 6 files changed, 47 insertions(+), 23 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1154936f..1921525a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3701,10 +3701,8 @@ checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26" dependencies = [ "js-sys", "opentelemetry", - "smallvec", "tracing", "tracing-core", - "tracing-log", "tracing-subscriber", "web-time", ] diff --git a/Cargo.toml b/Cargo.toml index 34f6c5ea..306c16dd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -86,8 +86,7 @@ n0-tracing-test = "0.3.0" [features] # enable the remote transport rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "tokio/io-util"] -# propagate span context to the remote with `span_propagation::layer`, a layer of -# tracing-subscriber 0.3. Not a default feature, irpc-opentelemetry enables it. +# span propagation layer for tracing-subscriber 0.3 tracing-subscriber-03 = ["rpc", "dep:tracing-subscriber", "tokio/rt"] # enable the iroh transport iroh = ["rpc", "dep:iroh", "dep:portable-atomic", "dep:getrandom"] @@ -204,7 +203,7 @@ allowed_external_types = [ "futures_core::stream::Stream", "futures_lite::future::Boxed", "futures_sink::Sink", - # `span_propagation::layer` returns a tracing layer, with the `tracing-subscriber-03` feature + # `span_propagation::layer`, with `tracing-subscriber-03` "tracing_core::subscriber::Subscriber", "tracing_subscriber::layer::Layer", "tracing_subscriber::registry::LookupSpan", diff --git a/README.md b/README.md index 8171294b..9694b9cd 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,23 @@ A streaming rpc system for iroh See the [module docs](https://docs.rs/irpc/latest/irpc/). +# Feature flags + +| Feature | Default | Description | +|---------|---------|-------------| +| `rpc` | yes | Remote transport over noq. Without it, irpc only has in-process channels. | +| `iroh` | yes | Remote transport over iroh. Enables `rpc`. | +| `spans` | yes | Messages store the current tracing span in [`WithChannels::span`]. | +| `stream` | yes | Convert channels to [`Stream`] and [`Sink`]. | +| `derive` | yes | The `rpc_requests` macro. | +| `noq_endpoint_setup` | no | Helpers to create noq endpoints, for tests and for rpc on localhost. | +| `varint-util` | no | Varint helpers in `irpc::util`, without `rpc`. | +| `tracing-subscriber-03` | no | Span propagation to the remote, with a layer for `tracing-subscriber` 0.3. [`irpc-opentelemetry`](irpc-opentelemetry) enables it. | + +[`WithChannels::span`]: https://docs.rs/irpc/latest/irpc/struct.WithChannels.html#structfield.span +[`Stream`]: https://docs.rs/futures-core/latest/futures_core/stream/trait.Stream.html +[`Sink`]: https://docs.rs/futures-sink/latest/futures_sink/trait.Sink.html + # Docs Properly building docs for this crate is quite complex. For all the gory details, diff --git a/irpc-opentelemetry/Cargo.toml b/irpc-opentelemetry/Cargo.toml index 1c474dd1..71fcdc2d 100644 --- a/irpc-opentelemetry/Cargo.toml +++ b/irpc-opentelemetry/Cargo.toml @@ -14,13 +14,12 @@ rust-version = "1.91" irpc = { version = "0.17.0", path = "..", default-features = false, features = ["tracing-subscriber-03"] } opentelemetry = "0.32" tracing = { workspace = true } -tracing-opentelemetry = "0.33" +tracing-opentelemetry = { version = "0.33", default-features = false } tracing-subscriber = { version = "0.3.23", default-features = false, features = ["registry"] } [dev-dependencies] anyhow = { workspace = true } clap = { version = "4.5.41", features = ["derive"] } -# Tests and examples need a crypto provider, which iroh's default features supply. iroh = { workspace = true, default-features = true } irpc = { version = "0.17.0", path = "..", features = ["iroh", "derive", "spans", "noq_endpoint_setup"] } n0-error = { workspace = true } @@ -42,5 +41,5 @@ allowed_external_types = [ ] [package.metadata.release] -# Released on its own: the version follows the opentelemetry releases, not irpc. +# own version, follows opentelemetry shared-version = false diff --git a/src/lib.rs b/src/lib.rs index 1d2b538a..a4f70d91 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -62,18 +62,25 @@ //! //! ## Features //! -//! - `derive`: Enable the [`rpc_requests`] macro. +//! - `derive`: Enable the [`rpc_requests`] macro. Enabled by default. //! - `rpc`: Enable the rpc features. Enabled by default. //! By disabling this feature, all rpc related dependencies are removed. //! The remaining dependencies are just serde, tokio and tokio-util. //! - `iroh`: Enable the iroh transport in the [`iroh`] module. Enabled by //! default. This feature enables `rpc`. -//! - `spans`: Enable tracing spans for messages. Enabled by default. -//! This is useful even without rpc, to not lose tracing context when message -//! passing. This is frequently done manually. This obviously requires -//! a dependency on tracing. +//! - `spans`: When irpc creates a message, it stores the current tracing span +//! in [`WithChannels::span`]. A handler can enter this span to continue the +//! trace of the caller across the channel, which also helps without `rpc`. +//! For remote requests, the server creates the message, so the span is the +//! current span on the server. The [`rpc_requests`] macro adds a +//! `parent_span()` method to the message enum. Enabled by default. //! - `noq_endpoint_setup`: Easy way to create noq endpoints, in the [`noq`] //! module. This is useful both for testing and for rpc on localhost. +//! - `stream`: Convert channels to `Stream` and `Sink`. Enabled by default. +//! - `varint-util`: Varint helpers in [`util`], without `rpc`. +//! - `tracing-subscriber-03`: Propagate span context to the remote, with a +//! layer for `tracing-subscriber` 0.3, see [`span_propagation`]. Not enabled +//! by default. The `irpc-opentelemetry` crate enables it. //! //! # Example //! @@ -156,6 +163,7 @@ //! framework for any transport with cheap streams such as QUIC. Compared to //! quic-rpc, this crate does not abstract over the stream type and is focused //! on [iroh](https://docs.rs/iroh/latest/iroh/index.html) and our [noq](https://docs.rs/noq/latest/noq/index.html). + #![cfg_attr(irpc_docsrs, feature(doc_cfg))] use std::{fmt::Debug, future::Future, io, marker::PhantomData, ops::Deref}; @@ -306,6 +314,7 @@ use std::{fmt::Debug, future::Future, io, marker::PhantomData, ops::Deref}; /// [`Channels`]: Channels #[cfg(feature = "derive")] pub use irpc_derive::rpc_requests; + #[cfg(feature = "rpc")] use n0_error::AnyError; use n0_error::stack_error; @@ -313,13 +322,12 @@ use serde::{Serialize, de::DeserializeOwned}; use self::{ channel::{ - mpsc, + SendError, mpsc, none::{NoReceiver, NoSender}, oneshot, }, sealed::Sealed, }; -use crate::channel::SendError; pub mod channel; #[cfg(feature = "rpc")] diff --git a/src/span_propagation.rs b/src/span_propagation.rs index 9e6da692..0520ae00 100644 --- a/src/span_propagation.rs +++ b/src/span_propagation.rs @@ -4,11 +4,11 @@ //! [`rpc_requests`](crate::rpc_requests) macro. Each request then carries a //! [`SpanContextCarrier`], a map of text headers, in front of the request. //! -//! irpc does not know any tracing backend. A [`Propagator`] writes the context +//! irpc does not know any tracing backend. A `Propagator` writes the context //! of a span into the headers on the client, and sets the parent of a span from //! the headers on the server. //! -//! irpc finds the propagator through the tracing subscriber: add the [`layer`] +//! irpc finds the propagator through the tracing subscriber: add the `layer` //! for it to the subscriber. The client asks the subscriber of the current //! thread, and the server asks the subscriber of the request span. Without the //! layer, irpc does not propagate span context. @@ -114,13 +114,16 @@ impl tracing_subscriber::Layer for PropagatorLayer {} /// each other's context, and the context moves with `fut` across threads. pub async fn scope_remote(carrier: Option, fut: F) -> F::Output { #[cfg(not(feature = "tracing-subscriber-03"))] - let _ = carrier; - #[cfg(not(feature = "tracing-subscriber-03"))] - return fut.await; + { + let _ = carrier; + fut.await + } #[cfg(feature = "tracing-subscriber-03")] - match carrier { - Some(carrier) => SPAN_CONTEXT.scope(carrier, fut).await, - None => fut.await, + { + match carrier { + Some(carrier) => SPAN_CONTEXT.scope(carrier, fut).await, + None => fut.await, + } } } From a6199bb09a1f0d200af46014f8c4544b470d18cf Mon Sep 17 00:00:00 2001 From: Frando Date: Mon, 12 Oct 2026 11:02:29 +0200 Subject: [PATCH 11/12] test: use Handler::raw in the span propagation tests --- irpc-opentelemetry/tests/local_parent.rs | 11 ++++------- tests/span_propagator.rs | 12 +++++------- 2 files changed, 9 insertions(+), 14 deletions(-) diff --git a/irpc-opentelemetry/tests/local_parent.rs b/irpc-opentelemetry/tests/local_parent.rs index 122bf782..69f3426d 100644 --- a/irpc-opentelemetry/tests/local_parent.rs +++ b/irpc-opentelemetry/tests/local_parent.rs @@ -5,10 +5,7 @@ //! activation off, the current OpenTelemetry context is empty, and setting it as //! parent would make the request span a new root. -use std::{ - net::{Ipv4Addr, SocketAddrV4}, - sync::Arc, -}; +use std::net::{Ipv4Addr, SocketAddrV4}; use irpc::{ Client, WithChannels, @@ -68,9 +65,9 @@ async fn check_local_parent(context_activation: bool) -> Result<()> { // Spans are exported when they close, so the handler reports when its spans are closed. let (done_tx, mut done_rx) = tokio::sync::mpsc::unbounded_channel(); - let handler: Handler = Arc::new(move |request, rx, tx| { + let handler = Handler::raw(move |request: EchoProtocol, rx, tx| { let done_tx = done_tx.clone(); - Box::pin(async move { + async move { async move { let EchoMessage::Echo(WithChannels { inner, tx, span, .. @@ -82,7 +79,7 @@ async fn check_local_parent(context_activation: bool) -> Result<()> { .await; done_tx.send(()).ok(); Ok(()) - }) + } }); let _server = AbortOnDropHandle::new(tokio::spawn(listen::(server, handler))); let client = Client::::noq(client, server_addr); diff --git a/tests/span_propagator.rs b/tests/span_propagator.rs index 8f83001c..dac1b2e1 100644 --- a/tests/span_propagator.rs +++ b/tests/span_propagator.rs @@ -69,13 +69,11 @@ async fn propagator_carries_headers_to_the_server() -> TestResult<()> { let _guard = tracing::subscriber::set_default(subscriber); let (server, client, server_addr) = create_connected_endpoints()?; - let handler: Handler = Arc::new(|request, rx, tx| { - Box::pin(async move { - let EchoMessage::Echo(WithChannels { inner, tx, .. }) = - request.with_remote_channels(rx, tx); - tx.send(inner.0).await.ok(); - Ok(()) - }) + let handler = Handler::raw(|request: EchoProtocol, rx, tx| async move { + let EchoMessage::Echo(WithChannels { inner, tx, .. }) = + request.with_remote_channels(rx, tx); + tx.send(inner.0).await.ok(); + Ok(()) }); let _server = AbortOnDropHandle::new(tokio::spawn(listen::(server, handler))); let client = Client::::noq(client, server_addr); From 73e37ce17cff9f47f3f20c1da6988755edf690d7 Mon Sep 17 00:00:00 2001 From: Frando Date: Mon, 12 Oct 2026 11:21:08 +0200 Subject: [PATCH 12/12] chore: fmt --- src/lib.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/lib.rs b/src/lib.rs index a4f70d91..eb0a00be 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -314,7 +314,6 @@ use std::{fmt::Debug, future::Future, io, marker::PhantomData, ops::Deref}; /// [`Channels`]: Channels #[cfg(feature = "derive")] pub use irpc_derive::rpc_requests; - #[cfg(feature = "rpc")] use n0_error::AnyError; use n0_error::stack_error;