mako_engine/
trace_ctx.rs

1//! W3C trace-context propagation across the outbox boundary.
2//!
3//! End-to-end tracing in `makod` crosses an asynchronous store boundary: an
4//! inbound HTTP request (AS4 push, REST ingest, ERP command) produces outbox
5//! messages that a worker delivers minutes later on a different task. A span
6//! cannot survive that hop — but the W3C `traceparent` string can.
7//!
8//! The transport layer scopes the inbound request's `traceparent` header into
9//! the [`TRACEPARENT`] task-local; every [`OutboxMessage`] created inside
10//! that scope — by `materialise_outbox` or `OutboxMessage::new` — captures it
11//! into its persisted `trace_context` field. Delivery workers then inject it
12//! into outbound HTTP requests (ERP webhook `traceparent` header and the
13//! CloudEvents `traceparent` extension), closing the chain:
14//!
15//! ```text
16//! inbound traceparent ─▶ task-local ─▶ outbox row ─▶ outbound traceparent
17//! ```
18//!
19//! [`OutboxMessage`]: crate::outbox::OutboxMessage
20
21tokio::task_local! {
22    /// The W3C `traceparent` value of the request currently being processed.
23    pub static TRACEPARENT: Option<String>;
24}
25
26/// The `traceparent` of the current task scope, if one was propagated.
27///
28/// Returns `None` outside a [`TRACEPARENT`] scope (workers, tests, CLI).
29#[must_use]
30pub fn current() -> Option<String> {
31    TRACEPARENT
32        .try_with(std::clone::Clone::clone)
33        .ok()
34        .flatten()
35}
36
37#[cfg(test)]
38mod tests {
39    /// A message created inside a TRACEPARENT scope captures it; one created
40    /// outside does not.
41    #[tokio::test]
42    async fn outbox_message_captures_scoped_traceparent() {
43        use crate::ids::{ConversationId, CorrelationId, EventId, ProcessId, StreamId, TenantId};
44        use crate::outbox::OutboxMessage;
45
46        let mk = || {
47            let tenant = TenantId::from_party_id("9900000000001");
48            let process = ProcessId::new();
49            OutboxMessage::new(
50                StreamId::for_process(tenant, &process),
51                process,
52                tenant,
53                CorrelationId::new(),
54                ConversationId::new(),
55                EventId::new(),
56                "APERAK",
57                "9900000000002",
58                serde_json::json!({}),
59            )
60        };
61
62        let tp = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01";
63        let inside = super::TRACEPARENT
64            .scope(Some(tp.to_owned()), async { mk() })
65            .await;
66        assert_eq!(inside.trace_context.as_deref(), Some(tp));
67
68        let outside = mk();
69        assert_eq!(outside.trace_context, None);
70    }
71}
mako_engine/trace_ctx.rs

mako_engine/
trace_ctx.rs
