durable_workflow/
workflow_logging.rs

1use crate::{replay_clock, WorkflowContext, WorkflowState};
2use std::{cell::Cell, panic::AssertUnwindSafe};
3
4thread_local! {
5    static RECONSTRUCTING: Cell<bool> = const { Cell::new(false) };
6}
7
8/// The live execution region for workflow observations.
9///
10/// This read-only view does not authorize different durable workflow commands.
11#[derive(Clone, Copy, Debug, PartialEq, Eq)]
12#[non_exhaustive]
13pub enum WorkflowReplayState {
14    /// Reconstructing committed history or a query. Process logs are suppressed.
15    Replaying,
16    /// New execution beyond the committed prefix. A task retry can repeat logs.
17    Executing,
18    /// Outside this workflow invocation, or its state cannot be read safely.
19    Unavailable,
20}
21
22pub(crate) struct ReconstructionGuard(bool);
23
24impl ReconstructionGuard {
25    pub(crate) fn enter() -> Self {
26        Self(RECONSTRUCTING.with(|active| active.replace(true)))
27    }
28}
29
30impl Drop for ReconstructionGuard {
31    fn drop(&mut self) {
32        RECONSTRUCTING.with(|active| active.set(self.0));
33    }
34}
35
36fn replaying(state: &WorkflowState) -> bool {
37    state.observation_terminal
38        || state.command_cursor < state.recorded_commands.len()
39        || state.selection_marker_cursor < state.selection_markers.len()
40        || (state.observation_pending && state.commands.is_empty())
41        || state.cancellation_delivery_intent.is_some()
42        || state.cancellation_scope_opening.is_some()
43        || (state.cancellation_history.delivery.is_some() && !state.cancellation_consumed)
44        || (state.recorded_continue_as_new_sequence.is_some() && !state.continue_as_new_consumed)
45        || state.scope_delivery.as_ref().is_some_and(|replay| {
46            replay.intent.is_some()
47                || replay
48                    .canonical
49                    .deliveries
50                    .keys()
51                    .any(|sequence| !replay.consumed.contains(sequence))
52        })
53}
54
55impl WorkflowContext {
56    /// Observe the current replay boundary without changing commands or history.
57    ///
58    /// Query/offline reconstruction remains `Replaying` after the last recorded
59    /// command. A retained context outside its invocation is `Unavailable`.
60    pub fn replay_state(&self) -> WorkflowReplayState {
61        let Some(query) = replay_clock::observation_mode(self) else {
62            return WorkflowReplayState::Unavailable;
63        };
64        let Ok(state) = self.state.lock() else {
65            return WorkflowReplayState::Unavailable;
66        };
67        if query || RECONSTRUCTING.with(Cell::get) || replaying(&state) {
68            WorkflowReplayState::Replaying
69        } else {
70            WorkflowReplayState::Executing
71        }
72    }
73
74    /// Log new workflow execution through the application's `tracing` subscriber.
75    ///
76    /// Committed replay, queries and discarded invocations stay quiet. Handlers
77    /// must be bounded/nonblocking. This process logger is not an exactly-once
78    /// outbox: logs from an uncommitted task can repeat when the task retries.
79    pub fn logger(&self) -> WorkflowLogger {
80        WorkflowLogger {
81            context: self.clone(),
82        }
83    }
84}
85
86/// Replay-aware host logging with SDK-bound namespace/workflow/run fields.
87#[derive(Clone)]
88pub struct WorkflowLogger {
89    context: WorkflowContext,
90}
91
92impl std::fmt::Debug for WorkflowLogger {
93    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
94        f.debug_struct("WorkflowLogger").finish_non_exhaustive()
95    }
96}
97
98impl WorkflowLogger {
99    /// Emit only in a new execution region with a complete bound run identity.
100    /// Logger failures do not manufacture success, commands or workflow history.
101    pub fn log(&self, level: tracing::Level, message: &str) {
102        if self.context.replay_state() != WorkflowReplayState::Executing {
103            return;
104        }
105        let identity = self.context.state.lock().ok().and_then(|state| {
106            let namespace = state
107                .observation_namespace
108                .as_ref()
109                .filter(|v| !v.is_empty())?;
110            let workflow = state.workflow_id.as_ref().filter(|v| !v.is_empty())?;
111            let run = state.run_id.as_ref().filter(|v| !v.is_empty())?;
112            Some((namespace.clone(), workflow.clone(), run.clone()))
113        });
114        let Some((namespace, workflow, run)) = identity else {
115            return;
116        };
117        // Invoke application code without holding workflow state. The temporary
118        // observation scope also suppresses recursive logs from a subscriber.
119        let _scope = replay_clock::Guard::unavailable();
120        let _ = std::panic::catch_unwind(AssertUnwindSafe(|| {
121            macro_rules! emit {
122                ($level:expr) => {
123                    tracing::event!(target: "durable_workflow::workflow", $level,
124                        namespace = namespace.as_str(),
125                        workflow_instance_id = workflow.as_str(),
126                        workflow_run_id = run.as_str(), message)
127                };
128            }
129            match level {
130                tracing::Level::TRACE => emit!(tracing::Level::TRACE),
131                tracing::Level::DEBUG => emit!(tracing::Level::DEBUG),
132                tracing::Level::INFO => emit!(tracing::Level::INFO),
133                tracing::Level::WARN => emit!(tracing::Level::WARN),
134                tracing::Level::ERROR => emit!(tracing::Level::ERROR),
135            }
136        }));
137    }
138
139    /// Emit a trace observation in new workflow execution.
140    pub fn trace(&self, message: &str) {
141        self.log(tracing::Level::TRACE, message);
142    }
143    /// Emit a debug observation in new workflow execution.
144    pub fn debug(&self, message: &str) {
145        self.log(tracing::Level::DEBUG, message);
146    }
147    /// Emit an informational observation in new workflow execution.
148    pub fn info(&self, message: &str) {
149        self.log(tracing::Level::INFO, message);
150    }
151    /// Emit a warning observation in new workflow execution.
152    pub fn warn(&self, message: &str) {
153        self.log(tracing::Level::WARN, message);
154    }
155    /// Emit an error observation in new workflow execution.
156    pub fn error(&self, message: &str) {
157        self.log(tracing::Level::ERROR, message);
158    }
159}
160
161#[cfg(test)]
162mod tests;
durable_workflow/workflow_logging.rs

durable_workflow/
workflow_logging.rs
