dcontext_dactor/
lib.rs

1//! # dcontext-dactor
2//!
3//! Automatic dcontext propagation through [dactor](https://github.com/microsoft/dactor)
4//! actor messages.
5//!
6//! This crate bridges [`dcontext`] (distributed context propagation) with
7//! [`dactor`] (actor framework) by providing interceptors that automatically
8//! carry context across actor message boundaries — **no boilerplate needed
9//! in handlers**.
10//!
11//! ## How It Works
12//!
13//! Context propagation is a **two-stage interceptor pipeline**:
14//!
15//! 1. **Outbound interceptor** (sender side) — [`ContextOutboundInterceptor`]
16//!    captures the current dcontext and attaches it as message headers.
17//!    - Local targets → [`ContextSnapshotHeader`] (preserves local-only values)
18//!    - Remote targets → [`ContextHeader`] (serialized wire bytes)
19//!
20//! 2. **Inbound interceptor** (receiver side) — [`ContextInboundInterceptor`]
21//!    performs two actions:
22//!    - `on_receive`: normalizes headers — deserializes wire bytes into a
23//!      [`ContextSnapshotHeader`] if needed.
24//!    - `wrap_handler`: wraps the handler future with
25//!      [`dcontext::async_ctx::with_context`], restoring the propagated
26//!      snapshot into the async task-local scope automatically.
27//!
28//! This uses dactor 0.3's `wrap_handler` feature to wrap the handler future
29//! with a task-local context scope. `dcontext::async_ctx::get_context` /
30//! `dcontext::async_ctx::set_context` work transparently inside the handler.
31//!
32//! ## Error Handling
33//!
34//! Both interceptors accept an [`ErrorPolicy`] that controls behavior when
35//! serialization or deserialization fails:
36//!
37//! - [`ErrorPolicy::LogAndContinue`] (default) — log a warning, deliver the
38//!   message without context.
39//! - [`ErrorPolicy::Reject`] — reject the message via
40//!   [`Disposition::Reject`](dactor::Disposition::Reject).
41//!
42//! ## Local vs. Remote
43//!
44//! - **Local** (same process): Context is propagated via
45//!   [`ContextSnapshot`](dcontext::ContextSnapshot), preserving local-only
46//!   values that cannot be serialized. **No serialization is performed.**
47//!
48//! - **Remote** (cross-process): Context is serialized to bytes via
49//!   [`dcontext::async_ctx::serialize_context`] and transmitted as a wire
50//!   header.
51//!   Local-only values are excluded.
52//!
53//! ## Quick Start
54//!
55//! ```ignore
56//! use dcontext_dactor::{ContextOutboundInterceptor, ContextInboundInterceptor};
57//!
58//! // Register interceptors on your runtime — that's it!
59//! runtime.add_outbound_interceptor(Box::new(ContextOutboundInterceptor::default()));
60//! runtime.add_inbound_interceptor(Box::new(ContextInboundInterceptor::default()));
61//!
62//! // Handlers automatically have dcontext available — no boilerplate
63//! #[async_trait]
64//! impl Handler<MyMessage> for MyActor {
65//!     async fn handle(&mut self, msg: MyMessage, ctx: &mut ActorContext) -> () {
66//!         // dcontext is automatically restored by the interceptor's wrap_handler
67//!         let rid: RequestId = dcontext::async_ctx::get_context("request_id").unwrap();
68//!         // ... handle message with context available ...
69//!     }
70//! }
71//! ```
72//!
73//! ## Manual Extraction
74//!
75//! For advanced use cases (e.g., spawning sub-tasks that need context),
76//! [`extract_context`] is still available to pull the snapshot from headers.
77//! [`with_propagated_context`] is retained for backward compatibility but is
78//! no longer needed when using the interceptor pipeline.
79
80mod header;
81mod inbound;
82mod outbound;
83mod propagation;
84
85pub use header::{ContextHeader, ContextSnapshotHeader};
86pub use inbound::ContextInboundInterceptor;
87pub use outbound::ContextOutboundInterceptor;
88pub use propagation::extract_context;
89
90#[allow(deprecated)]
91pub use propagation::with_propagated_context;
92
93/// Register dcontext header deserializers with a dactor [`HeaderRegistry`](dactor::HeaderRegistry).
94///
95/// Call this at startup before processing remote messages so that
96/// [`ContextHeader`] can be reconstructed from wire bytes during remote
97/// actor communication.
98///
99/// # Example
100///
101/// ```ignore
102/// use dactor::HeaderRegistry;
103/// use dcontext_dactor::register_context_headers;
104///
105/// let mut registry = HeaderRegistry::new();
106/// register_context_headers(&mut registry);
107/// // pass registry to your transport / runtime
108/// ```
109pub fn register_context_headers(registry: &mut dactor::HeaderRegistry) {
110    registry.register("dcontext.wire", |bytes: &[u8]| {
111        Some(Box::new(ContextHeader::from_bytes(bytes.to_vec())))
112    });
113}
114
115/// Controls how interceptors behave when serialization or deserialization fails.
116///
117/// Passed to [`ContextOutboundInterceptor::new`] and
118/// [`ContextInboundInterceptor::new`] to configure error handling.
119#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
120pub enum ErrorPolicy {
121    /// Log a warning and deliver the message without propagated context.
122    /// This is the default.
123    #[default]
124    LogAndContinue,
125    /// Reject the message via [`Disposition::Reject`](dactor::Disposition::Reject).
126    Reject,
127}
128
129#[cfg(test)]
130mod tests;
dcontext_dactor/lib.rs

dcontext_dactor/
lib.rs
