diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b21272..8c194ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ ## Unreleased +- Workflow code can use `ctx.logger()` to send new-execution observations to + the application's `tracing` subscriber with namespace/workflow/run identity. + Committed replay, query reconstruction and discarded invocations stay quiet. + `ctx.replay_state()` exposes the live consumed-history boundary. Logs from + uncommitted task attempts may repeat on retry; handlers must be nonblocking. + ## 4.16.0 - Completion payload planning reads the optional advertised diff --git a/README.md b/README.md index 8107ad7..e81ec08 100644 --- a/README.md +++ b/README.md @@ -495,6 +495,41 @@ Keep workflow command order deterministic: See the runnable [`workflow_testing`](examples/workflow_testing.rs) example and [`workflow_testing` tests](tests/workflow_testing.rs). +## Workflow logging + +Use `ctx.logger()` for observations written from workflow code. It sends events +to the application's existing `tracing` subscriber with the target +`durable_workflow::workflow` and SDK-bound `namespace`, `workflow_instance_id` +and `workflow_run_id` fields. No global subscriber is installed. + +```rust,no_run +# use durable_workflow::{Client, Worker, json}; +# fn configure(client: Client) { +let mut worker = Worker::new(client, "orders"); +worker.register_workflow("order", |ctx, _input| async move { + ctx.logger().info("processing order"); + let result = ctx.activity("charge", json!([])).await?; + ctx.logger().info("charge completed"); + Ok(result) +}); +# } +``` + +Ordinary `tracing` calls execute again when a worker reconstructs a workflow. +The workflow logger suppresses the committed prefix, query/offline replay and +completed histories, including after a cold worker replacement. It emits when +execution reaches new code beyond that prefix. `ctx.replay_state()` exposes +that live boundary as `WorkflowReplayState`; a retained context outside its +invocation is `Unavailable`. Use the logger for observations and keep durable +command order independent of the observation state. + +This is process logging. A log from an uncommitted task attempt can repeat if +that attempt crashes or retries. It is not an exactly-once outbox. Keep host +handlers bounded and nonblocking, for example with an application-owned logging +queue, so they do not stall the worker. Unwinding subscriber panics are isolated +from the workflow result; aborting processes and blocking handlers remain the +application's responsibility. + ## Inspect a run's activities Use `Client::get_run_activities(workflow_id, run_id)` next to the selected run's diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index a36cb57..bb2b18d 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -72,6 +72,12 @@ { "glob": "src/replay_clock.rs" }, + { + "glob": "src/workflow_logging.rs" + }, + { + "glob": "src/nexus.rs" + }, { "glob": "src/replay_clock/condition.rs" }, diff --git a/src/lib.rs b/src/lib.rs index 85f647b..d6547cc 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -36,6 +36,8 @@ mod update_outcome; mod worker_capacity; mod worker_session; mod workflow_hydration; +mod workflow_logging; +pub use workflow_logging::{WorkflowLogger, WorkflowReplayState}; pub use activity_error::ActivityError; pub use activity_retry::{ActivityAttemptLimit, ActivityRetryInfo}; @@ -10242,6 +10244,7 @@ impl Worker { self.allow_cancellation_scope_authoring, self.allow_cancellation_scope_delivery, )?; + workflow_state.observation_namespace = Some(self.client.namespace.clone()); workflow_state.history_budget = history_budget; workflow_state.workflow_command_identity = workflow_command_identity; if let Some(canonical) = cancellation_history { @@ -12366,6 +12369,11 @@ fn is_authored_command_open_event(event: &HistoryEvent) -> bool { #[derive(Debug)] struct WorkflowState { + observation_namespace: Option, + observation_terminal: bool, + // A pending recorded call consumes its descriptor before yielding. Cursor + // exhaustion alone therefore cannot make the surrounding poll live. + observation_pending: bool, workflow_id: Option, run_id: Option, task_queue: String, @@ -12663,6 +12671,19 @@ impl WorkflowState { }) }); Ok(Self { + observation_namespace: None, + observation_terminal: history.iter().any(|event| { + matches!( + event.event_type.as_str(), + "WorkflowCompleted" + | "WorkflowFailed" + | "WorkflowCancelled" + | "WorkflowTerminated" + | "WorkflowTimedOut" + | "WorkflowContinuedAsNew" + ) + }), + observation_pending: false, workflow_command_identity: String::new(), workflow_stream_command_counter: 0, workflow_id, @@ -14734,6 +14755,7 @@ impl Future for SelectCall { // on, so an otherwise commandless replay remains legitimately // pending instead of failing as an untracked yield. state.matched_recorded_pending = true; + state.observation_pending = true; } marker }; @@ -15812,6 +15834,7 @@ impl ActivityCall { return Poll::Ready(outcome.map_err(Error::ActivityFailed)); } state.matched_recorded_pending = true; + state.observation_pending = true; self.scheduled = true; return Poll::Pending; } @@ -16010,6 +16033,7 @@ impl Future for TimerCall { return Poll::Ready(Ok(())); } state.matched_recorded_pending = true; + state.observation_pending = true; self.scheduled = true; self.matched_pending = true; return Poll::Pending; @@ -16592,6 +16616,7 @@ impl ChildWorkflowCall { return Poll::Ready(outcome.map_err(Error::ChildWorkflowFailed)); } state.matched_recorded_pending = true; + state.observation_pending = true; self.scheduled = true; self.matched_pending = true; return Poll::Pending; @@ -16840,6 +16865,7 @@ impl SignalCall { } state.matched_recorded_pending = true; + state.observation_pending = true; self.opened_wait = true; self.matched_pending = true; return Poll::Pending; diff --git a/src/nexus.rs b/src/nexus.rs index aea03ac..1a0065b 100644 --- a/src/nexus.rs +++ b/src/nexus.rs @@ -518,6 +518,7 @@ impl Future for NexusServiceCall { ); } state.matched_recorded_pending = true; + state.observation_pending = true; self.suspended = true; return Poll::Pending; } diff --git a/src/replay_clock.rs b/src/replay_clock.rs index 0f641a0..eadfd64 100644 --- a/src/replay_clock.rs +++ b/src/replay_clock.rs @@ -54,6 +54,11 @@ impl Drop for Guard { } } +pub(super) fn observation_mode(context: &WorkflowContext) -> Option { + let active = ACTIVE.with(|active| active.borrow().as_ref().and_then(Weak::upgrade))?; + Arc::ptr_eq(&active, &context.state).then(|| QUERY_REPLAY.with(Cell::get)) +} + fn unavailable() -> Error { Error::ReplayClockUnavailable } @@ -445,11 +450,15 @@ impl WorkflowState { } pub(super) fn consume_unproven_boundary(&mut self, path: &[ParallelGroupMetadata]) { + if path.is_empty() { + self.observation_pending = false; + } self.replay_clock.unavailable(); self.unavailable_cancellation_boundary(path); } pub(super) fn consume_replay_group(&mut self, sequences: &[u64], failed: Option) { + self.observation_pending = false; // Core's query group-send branch supplies no aggregate time. Do not // transfer executor authority into reconstructed query execution. self.replay_clock.time = if QUERY_REPLAY.with(Cell::get) { @@ -477,16 +486,19 @@ impl WorkflowState { } pub(super) fn consume_replay_selection(&mut self, group: &str) { + self.observation_pending = false; self.replay_clock.unavailable(); self.advance_cancellation_selection(group); } pub(super) fn consume_replay_receipt(&mut self, group: &str, sequence: u64) { + self.observation_pending = false; self.replay_clock.unavailable(); self.advance_cancellation_receipt(group, sequence); } pub(super) fn consume_replay_handle(&mut self, handle: &DurableOperationHandle, failed: bool) { + self.observation_pending = false; self.replay_clock.unavailable(); self.advance_cancellation_handle(handle, failed); } @@ -501,6 +513,7 @@ impl WorkflowState { } pub(super) fn observe_replay_sequence(&mut self, sequence: u64) { + self.observation_pending = false; self.replay_clock.time = self .replay_clock .boundaries diff --git a/src/testing/environment.rs b/src/testing/environment.rs index 2a5b028..eb8ca58 100644 --- a/src/testing/environment.rs +++ b/src/testing/environment.rs @@ -119,6 +119,7 @@ impl WorkflowEnvironment { /// Newly emitted durable commands are divergences, including commands added /// after the recorded history ended. No activity callbacks are invoked. pub fn replay(&self, history: &WorkflowHistory) -> Result { + let _observations = crate::workflow_logging::ReconstructionGuard::enter(); self.worker.validate_registration()?; let mut worker = self.worker.clone(); if let Some(queue) = &history.task_queue { diff --git a/src/workflow_logging.rs b/src/workflow_logging.rs new file mode 100644 index 0000000..800bed9 --- /dev/null +++ b/src/workflow_logging.rs @@ -0,0 +1,162 @@ +use crate::{replay_clock, WorkflowContext, WorkflowState}; +use std::{cell::Cell, panic::AssertUnwindSafe}; + +thread_local! { + static RECONSTRUCTING: Cell = const { Cell::new(false) }; +} + +/// The live execution region for workflow observations. +/// +/// This read-only view does not authorize different durable workflow commands. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum WorkflowReplayState { + /// Reconstructing committed history or a query. Process logs are suppressed. + Replaying, + /// New execution beyond the committed prefix. A task retry can repeat logs. + Executing, + /// Outside this workflow invocation, or its state cannot be read safely. + Unavailable, +} + +pub(crate) struct ReconstructionGuard(bool); + +impl ReconstructionGuard { + pub(crate) fn enter() -> Self { + Self(RECONSTRUCTING.with(|active| active.replace(true))) + } +} + +impl Drop for ReconstructionGuard { + fn drop(&mut self) { + RECONSTRUCTING.with(|active| active.set(self.0)); + } +} + +fn replaying(state: &WorkflowState) -> bool { + state.observation_terminal + || state.command_cursor < state.recorded_commands.len() + || state.selection_marker_cursor < state.selection_markers.len() + || (state.observation_pending && state.commands.is_empty()) + || state.cancellation_delivery_intent.is_some() + || state.cancellation_scope_opening.is_some() + || (state.cancellation_history.delivery.is_some() && !state.cancellation_consumed) + || (state.recorded_continue_as_new_sequence.is_some() && !state.continue_as_new_consumed) + || state.scope_delivery.as_ref().is_some_and(|replay| { + replay.intent.is_some() + || replay + .canonical + .deliveries + .keys() + .any(|sequence| !replay.consumed.contains(sequence)) + }) +} + +impl WorkflowContext { + /// Observe the current replay boundary without changing commands or history. + /// + /// Query/offline reconstruction remains `Replaying` after the last recorded + /// command. A retained context outside its invocation is `Unavailable`. + pub fn replay_state(&self) -> WorkflowReplayState { + let Some(query) = replay_clock::observation_mode(self) else { + return WorkflowReplayState::Unavailable; + }; + let Ok(state) = self.state.lock() else { + return WorkflowReplayState::Unavailable; + }; + if query || RECONSTRUCTING.with(Cell::get) || replaying(&state) { + WorkflowReplayState::Replaying + } else { + WorkflowReplayState::Executing + } + } + + /// Log new workflow execution through the application's `tracing` subscriber. + /// + /// Committed replay, queries and discarded invocations stay quiet. Handlers + /// must be bounded/nonblocking. This process logger is not an exactly-once + /// outbox: logs from an uncommitted task can repeat when the task retries. + pub fn logger(&self) -> WorkflowLogger { + WorkflowLogger { + context: self.clone(), + } + } +} + +/// Replay-aware host logging with SDK-bound namespace/workflow/run fields. +#[derive(Clone)] +pub struct WorkflowLogger { + context: WorkflowContext, +} + +impl std::fmt::Debug for WorkflowLogger { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WorkflowLogger").finish_non_exhaustive() + } +} + +impl WorkflowLogger { + /// Emit only in a new execution region with a complete bound run identity. + /// Logger failures do not manufacture success, commands or workflow history. + pub fn log(&self, level: tracing::Level, message: &str) { + if self.context.replay_state() != WorkflowReplayState::Executing { + return; + } + let identity = self.context.state.lock().ok().and_then(|state| { + let namespace = state + .observation_namespace + .as_ref() + .filter(|v| !v.is_empty())?; + let workflow = state.workflow_id.as_ref().filter(|v| !v.is_empty())?; + let run = state.run_id.as_ref().filter(|v| !v.is_empty())?; + Some((namespace.clone(), workflow.clone(), run.clone())) + }); + let Some((namespace, workflow, run)) = identity else { + return; + }; + // Invoke application code without holding workflow state. The temporary + // observation scope also suppresses recursive logs from a subscriber. + let _scope = replay_clock::Guard::unavailable(); + let _ = std::panic::catch_unwind(AssertUnwindSafe(|| { + macro_rules! emit { + ($level:expr) => { + tracing::event!(target: "durable_workflow::workflow", $level, + namespace = namespace.as_str(), + workflow_instance_id = workflow.as_str(), + workflow_run_id = run.as_str(), message) + }; + } + match level { + tracing::Level::TRACE => emit!(tracing::Level::TRACE), + tracing::Level::DEBUG => emit!(tracing::Level::DEBUG), + tracing::Level::INFO => emit!(tracing::Level::INFO), + tracing::Level::WARN => emit!(tracing::Level::WARN), + tracing::Level::ERROR => emit!(tracing::Level::ERROR), + } + })); + } + + /// Emit a trace observation in new workflow execution. + pub fn trace(&self, message: &str) { + self.log(tracing::Level::TRACE, message); + } + /// Emit a debug observation in new workflow execution. + pub fn debug(&self, message: &str) { + self.log(tracing::Level::DEBUG, message); + } + /// Emit an informational observation in new workflow execution. + pub fn info(&self, message: &str) { + self.log(tracing::Level::INFO, message); + } + /// Emit a warning observation in new workflow execution. + pub fn warn(&self, message: &str) { + self.log(tracing::Level::WARN, message); + } + /// Emit an error observation in new workflow execution. + pub fn error(&self, message: &str) { + self.log(tracing::Level::ERROR, message); + } +} + +#[cfg(test)] +mod tests; diff --git a/src/workflow_logging/tests.rs b/src/workflow_logging/tests.rs new file mode 100644 index 0000000..7b8fa04 --- /dev/null +++ b/src/workflow_logging/tests.rs @@ -0,0 +1,794 @@ +use super::*; +use crate::{ + encode_value_envelope, json, + testing::{WorkflowEnvironment, WorkflowHistory}, + AvroValue, Client, Error, HistoryEvent, Value, Worker, WorkflowFuture, WorkflowTask, + DEFAULT_CODEC, +}; +use std::{ + collections::BTreeMap, + future::Future, + sync::{Arc, Mutex}, + task::Poll, + time::Duration, +}; + +#[derive(Debug, PartialEq)] +struct Record { + level: tracing::Level, + fields: BTreeMap, +} + +#[derive(Default)] +struct Log { + records: Mutex>, + panic: bool, + on_event: Option>, +} + +struct Fields(BTreeMap); +impl tracing::field::Visit for Fields { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.0.insert(field.name().into(), value.into()); + } + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0.insert(field.name().into(), format!("{value:?}")); + } +} + +impl tracing::Subscriber for Log { + fn enabled(&self, _: &tracing::Metadata<'_>) -> bool { + true + } + fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id { + tracing::span::Id::from_u64(1) + } + fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {} + fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {} + fn event(&self, event: &tracing::Event<'_>) { + if event.metadata().target() != "durable_workflow::workflow" { + return; + } + assert!(!self.panic, "application logger failed"); + if let Some(callback) = &self.on_event { + callback(); + } + let mut fields = Fields(BTreeMap::new()); + event.record(&mut fields); + self.records.lock().unwrap().push(Record { + level: *event.metadata().level(), + fields: fields.0, + }); + } + fn enter(&self, _: &tracing::span::Id) {} + fn exit(&self, _: &tracing::span::Id) {} +} + +fn event(kind: &str, payload: Value) -> HistoryEvent { + serde_json::from_value(json!({"event_type":kind,"payload":payload})).unwrap() +} + +fn timers(count: u64) -> Vec { + (1..=count).flat_map(|sequence| { + let payload = json!({"sequence":sequence,"timer_id":format!("timer-{sequence}"),"delay_seconds":1}); + [event("TimerScheduled",payload.clone()),event("TimerFired",payload)] + }).collect() +} + +fn context(history: Vec) -> WorkflowContext { + let mut state = WorkflowState::new_with_identity( + history, + Some("order".into()), + Some("run-1".into()), + "tests".into(), + DEFAULT_CODEC.into(), + None, + ) + .unwrap(); + state.observation_namespace = Some("shipping".into()); + WorkflowContext { + state: Arc::new(Mutex::new(state)), + cancellation_scope_id: "root".into(), + } +} + +fn task(history: Vec) -> WorkflowTask { + serde_json::from_value(json!({ + "task_id":"task-1","workflow_type":"logged","workflow_id":"order","run_id":"run-1", + "payload_codec":DEFAULT_CODEC,"arguments":encode_value_envelope(&json!([]),DEFAULT_CODEC).unwrap(), + "history_events":history + })).unwrap() +} + +fn worker(logging: bool) -> Worker { + let client = Client::builder("http://127.0.0.1:9") + .namespace("shipping") + .build() + .unwrap(); + let mut worker = Worker::new(client, "tests"); + worker.register_workflow("logged", move |ctx, _| { + if logging { + ctx.logger().info("factory"); + } + async move { + for step in 0..3 { + if logging { + ctx.logger().info(&format!("step-{step}")); + } + ctx.sleep(Duration::from_secs(1)).await?; + } + if logging { + ctx.logger().info("done"); + } + Ok(json!("complete")) + } + }); + worker +} + +fn messages(log: &Log) -> Vec { + log.records + .lock() + .unwrap() + .iter() + .map(|r| r.fields["message"].clone()) + .collect() +} + +fn poll(ctx: &WorkflowContext, future: &mut WorkflowFuture) -> Poll> { + Worker::poll_workflow_future(ctx, future).unwrap() +} + +#[test] +fn multistep_cold_worker_replacement_logs_only_new_regions_and_preserves_commands() { + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + for prefix in 0..=3 { + // A new Worker and invocation have no process-local replay memory. + let input = task(timers(prefix)); + let plain = worker(false).execute_workflow_task(input.clone()).unwrap(); + let logged = worker(true).execute_workflow_task(input).unwrap(); + assert_eq!(plain, logged); + } + }); + assert_eq!( + messages(&log), + ["factory", "step-0", "step-1", "step-2", "done"] + ); + for record in log.records.lock().unwrap().iter() { + assert_eq!(record.fields.len(), 4); + assert_eq!(record.fields["namespace"], "shipping"); + assert_eq!(record.fields["workflow_instance_id"], "order"); + assert_eq!(record.fields["workflow_run_id"], "run-1"); + } +} + +#[test] +fn uncommitted_attempt_retry_can_repeat_process_logs() { + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + for _ in 0..2 { + worker(true).execute_workflow_task(task(vec![])).unwrap(); + } + }); + assert_eq!(messages(&log), ["factory", "step-0", "factory", "step-0"]); +} + +#[test] +fn live_accessor_tracks_consumption_without_history_budget_or_time_heuristics() { + let ctx = context(timers(1)); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Unavailable); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + assert_eq!(inner.replay_state(), WorkflowReplayState::Replaying); + inner.sleep(Duration::from_secs(1)).await?; + assert_eq!(inner.replay_state(), WorkflowReplayState::Executing); + Ok(AvroValue::Null) + }); + assert!(matches!(poll(&ctx, &mut future), Poll::Ready(Ok(_)))); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Unavailable); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[test] +fn pending_recorded_future_does_not_turn_cursor_exhaustion_into_live_execution() { + let ctx = context(vec![event( + "TimerScheduled", + json!({"sequence":1,"timer_id":"timer-1","delay_seconds":1}), + )]); + let inner = ctx.clone(); + let log = Arc::new(Log::default()); + let mut future: WorkflowFuture = Box::pin(async move { + let mut timer = Box::pin(inner.sleep(Duration::from_secs(1))); + futures_util::future::poll_fn(|cx| { + assert!(timer.as_mut().poll(cx).is_pending()); + assert_eq!(inner.replay_state(), WorkflowReplayState::Replaying); + inner.logger().info("pending is replay"); + Poll::Pending + }) + .await + }); + tracing::subscriber::with_default(log.clone(), || { + assert!(poll(&ctx, &mut future).is_pending()); + }); + assert!(messages(&log).is_empty()); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[test] +fn query_and_offline_reconstruction_stay_quiet_after_consuming_the_last_command() { + let ctx = context(vec![]); + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + { + let _query = replay_clock::Guard::query(&ctx); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Replaying); + ctx.logger().warn("query"); + } + let environment = WorkflowEnvironment::from_worker(worker(true)); + let history = WorkflowHistory::from_history( + "logged", + json!([]), + json!({ + "workflow_id":"order","run_id":"run-1","events":timers(3) + }), + ) + .unwrap(); + // Missing a terminal record makes this a divergent offline replay, but + // reconstruction must stay silent even while discovering that result. + let _ = environment.replay(&history); + { + let _live = replay_clock::Guard::enter(&ctx); + ctx.logger().info("live restored"); + } + }); + assert_eq!(messages(&log), ["live restored"]); +} + +#[test] +fn every_terminal_history_stays_quiet_with_an_exhausted_command_cursor() { + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + for kind in [ + "WorkflowCompleted", + "WorkflowFailed", + "WorkflowCancelled", + "WorkflowTerminated", + "WorkflowTimedOut", + ] { + let ctx = context(vec![event(kind, json!({}))]); + let _live = replay_clock::Guard::enter(&ctx); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Replaying); + ctx.logger().error(kind); + } + }); + assert!(messages(&log).is_empty()); +} + +#[test] +fn retained_context_and_future_drop_cannot_emit_outside_their_invocation() { + struct OnDrop(WorkflowLogger); + impl Drop for OnDrop { + fn drop(&mut self) { + self.0.info("discarded future"); + } + } + let ctx = context(vec![]); + let retained = ctx.logger(); + let inner = ctx.clone(); + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + let mut future: WorkflowFuture = Box::pin(async move { + let _on_drop = OnDrop(inner.logger()); + inner.logger().info("live"); + futures_util::future::pending().await + }); + assert!(poll(&ctx, &mut future).is_pending()); + drop(future); + retained.info("after disposal"); + }); + assert_eq!(messages(&log), ["live"]); +} + +#[test] +fn nested_invocations_restore_identity_and_all_levels_use_host_tracing() { + let a = context(vec![]); + let b = context(vec![]); + b.state.lock().unwrap().run_id = Some("run-2".into()); + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + let _outer = replay_clock::Guard::enter(&a); + a.logger().trace("trace"); + { + let _inner = replay_clock::Guard::enter(&b); + a.logger().error("wrong invocation"); + b.logger().debug("debug"); + } + a.logger().info("info"); + a.logger().warn("warn"); + a.logger().error("error"); + }); + assert_eq!(messages(&log), ["trace", "debug", "info", "warn", "error"]); + let records = log.records.lock().unwrap(); + assert_eq!( + records.iter().map(|r| r.level).collect::>(), + [ + tracing::Level::TRACE, + tracing::Level::DEBUG, + tracing::Level::INFO, + tracing::Level::WARN, + tracing::Level::ERROR + ] + ); + assert_eq!(records[1].fields["workflow_run_id"], "run-2"); + assert_eq!(records[2].fields["workflow_run_id"], "run-1"); +} + +#[test] +fn missing_run_identity_is_silent_instead_of_inventing_identity() { + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + for field in 0..3 { + let ctx = context(vec![]); + { + let mut state = ctx.state.lock().unwrap(); + match field { + 0 => state.observation_namespace = None, + 1 => state.workflow_id = None, + _ => state.run_id = None, + } + } + let _live = replay_clock::Guard::enter(&ctx); + ctx.logger().info("unbound"); + } + }); + assert!(messages(&log).is_empty()); +} + +#[test] +fn host_logger_panics_preserve_commands_results_and_restore_invocation_state() { + let log = Arc::new(Log { + panic: true, + ..Log::default() + }); + tracing::subscriber::with_default(log, || { + for prefix in 0..=3 { + let input = task(timers(prefix)); + assert_eq!( + worker(true).execute_workflow_task(input.clone()).unwrap(), + worker(false).execute_workflow_task(input).unwrap() + ); + } + let ctx = context(vec![]); + let _live = replay_clock::Guard::enter(&ctx); + ctx.logger().info("panic"); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Executing); + assert!(ctx.state.lock().is_ok()); + }); +} + +#[test] +fn logging_does_not_turn_application_failure_into_success() { + let mut worker = worker(false); + worker.register_workflow("fails", |ctx, _| async move { + ctx.logger().error("failed"); + Err(Error::WorkerLoop("business failure".into())) + }); + let mut input = task(vec![]); + input.workflow_type = "fails".into(); + let log = Arc::new(Log::default()); + let commands = tracing::subscriber::with_default(log.clone(), || { + worker.execute_workflow_task(input).unwrap() + }); + assert_eq!(messages(&log), ["failed"]); + assert_eq!(commands.len(), 1); + assert_eq!(commands[0]["type"], "fail_workflow"); +} + +#[test] +fn synchronous_recorded_side_effect_and_version_markers_use_the_live_cursor() { + let ctx = context(vec![ + event( + "SideEffectRecorded", + json!({"sequence":1,"result":encode_value_envelope(&json!(7),DEFAULT_CODEC).unwrap()}), + ), + event( + "VersionMarkerRecorded", + json!({"sequence":2,"change_id":"shipping","version":1,"min_supported":-1,"max_supported":1}), + ), + ]); + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + let _live = replay_clock::Guard::enter(&ctx); + ctx.logger().info("committed prefix"); + let value: i32 = ctx + .side_effect(|| panic!("recorded callback must not run")) + .unwrap(); + assert_eq!(value, 7); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Replaying); + assert_eq!(ctx.get_version("shipping", -1, 1).unwrap(), 1); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Executing); + ctx.logger().info("after markers"); + }); + assert_eq!(messages(&log), ["after markers"]); + assert!(ctx.take_commands().unwrap().is_empty()); + + let ctx = context(timers(1)); + let _live = replay_clock::Guard::enter(&ctx); + assert_eq!(ctx.get_version("added-later", -1, 1).unwrap(), -1); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Replaying); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[test] +fn nested_parallel_completion_releases_the_observation_boundary() { + use crate::ParallelOperation; + let operations = || { + vec![ + ParallelOperation::timer(Duration::from_secs(1)), + ParallelOperation::group(vec![ParallelOperation::timer(Duration::from_secs(1))]), + ] + }; + let mut history = vec![]; + for (index, descriptor) in crate::parallel_descriptors(operations(), 1) + .unwrap() + .iter() + .enumerate() + { + let sequence = index as u64 + 1; + let mut payload = + json!({"sequence":sequence,"timer_id":format!("timer-{sequence}"),"delay_seconds":1}); + crate::apply_parallel_group_path(payload.as_object_mut().unwrap(), &descriptor.group_path); + history.extend([ + event("TimerScheduled", payload.clone()), + event("TimerFired", payload), + ]); + } + let ctx = context(history); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + assert_eq!(inner.replay_state(), WorkflowReplayState::Replaying); + inner.parallel(operations()).await?; + assert_eq!(inner.replay_state(), WorkflowReplayState::Executing); + Ok(AvroValue::Null) + }); + assert!(matches!(poll(&ctx, &mut future), Poll::Ready(Ok(_)))); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[test] +fn a_completed_parallel_member_does_not_release_a_pending_aggregate_boundary() { + use crate::ParallelOperation; + let operations = || { + vec![ + ParallelOperation::timer(Duration::from_secs(1)), + ParallelOperation::timer(Duration::from_secs(1)), + ] + }; + let mut history = vec![]; + for (index, descriptor) in crate::parallel_descriptors(operations(), 1) + .unwrap() + .iter() + .enumerate() + { + let sequence = index as u64 + 1; + let mut payload = + json!({"sequence":sequence,"timer_id":format!("timer-{sequence}"),"delay_seconds":1}); + crate::apply_parallel_group_path(payload.as_object_mut().unwrap(), &descriptor.group_path); + history.push(event("TimerScheduled", payload.clone())); + if index == 1 { + history.push(event("TimerFired", payload)); + } + } + let ctx = context(history); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + let mut call = Box::pin(inner.parallel(operations())); + futures_util::future::poll_fn(|cx| { + assert!(call.as_mut().poll(cx).is_pending()); + assert_eq!(inner.replay_state(), WorkflowReplayState::Replaying); + inner.logger().info("pending aggregate"); + Poll::Pending + }) + .await + }); + let log = Arc::new(Log::default()); + tracing::subscriber::with_default( + log.clone(), + || assert!(poll(&ctx, &mut future).is_pending()), + ); + assert!(messages(&log).is_empty()); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +fn selection(ctx: &WorkflowContext) -> crate::SelectCall { + use crate::{ActivityOptions, ParallelOperation}; + ctx.select_keyed(vec![ + ( + "slow", + ParallelOperation::activity_with_options( + "slow-activity", + ActivityOptions::new().task_queue("default"), + json!([]), + ), + ), + ( + "fast", + ParallelOperation::activity_with_options( + "fast-activity", + ActivityOptions::new().task_queue("default"), + json!([]), + ), + ), + ]) +} + +#[test] +fn selection_marker_releases_live_logging_even_with_an_unfinished_loser() { + let fixture: Value = serde_json::from_str(include_str!( + "../../tests/fixtures/durable_selection_runtime_history.json" + )) + .unwrap(); + let mut history: Vec = + serde_json::from_value(fixture["history"].clone()).unwrap(); + history.retain(|e| !(e.event_type == "ActivityCompleted" && e.payload["sequence"] == 1)); + let ctx = context(history); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + inner.logger().info("before selection"); + selection(&inner).await?; + assert_eq!(inner.replay_state(), WorkflowReplayState::Executing); + inner.logger().info("selected"); + Ok(AvroValue::Null) + }); + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + assert!(matches!(poll(&ctx, &mut future), Poll::Ready(Ok(_)))) + }); + assert_eq!(messages(&log), ["selected"]); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[test] +fn terminal_members_without_a_committed_selection_marker_remain_replay() { + let fixture: Value = serde_json::from_str(include_str!( + "../../tests/fixtures/durable_selection_runtime_history.json" + )) + .unwrap(); + let mut history: Vec = + serde_json::from_value(fixture["history"].clone()).unwrap(); + history.retain(|e| e.event_type != "SelectionResolved"); + let ctx = context(history); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + let mut call = Box::pin(selection(&inner)); + futures_util::future::poll_fn(|cx| { + assert!(call.as_mut().poll(cx).is_pending()); + assert_eq!(inner.replay_state(), WorkflowReplayState::Replaying); + inner.logger().info("awaiting marker"); + Poll::Pending + }) + .await + }); + let log = Arc::new(Log::default()); + tracing::subscriber::with_default( + log.clone(), + || assert!(poll(&ctx, &mut future).is_pending()), + ); + assert!(messages(&log).is_empty()); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[test] +fn canonical_cancellation_is_replayed_until_its_authored_delivery_is_consumed() { + let fixture: Value = serde_json::from_str(include_str!( + "../../tests/fixtures/replay-regressions/cooperative-cold-activity-delivery.json" + )) + .unwrap(); + let state = WorkflowState::new_with_identity( + serde_json::from_value(fixture["history"].clone()).unwrap(), + Some("order".into()), + Some("regression-corpus-run".into()), + "tests".into(), + DEFAULT_CODEC.into(), + None, + ) + .unwrap(); + let ctx = WorkflowContext { + state: Arc::new(Mutex::new(state)), + cancellation_scope_id: "root".into(), + }; + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + assert_eq!(inner.replay_state(), WorkflowReplayState::Replaying); + assert!(inner.activity("prepare", json!([])).await.is_err()); + assert_eq!(inner.replay_state(), WorkflowReplayState::Executing); + Ok(AvroValue::Null) + }); + assert!(matches!(poll(&ctx, &mut future), Poll::Ready(Ok(_)))); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[test] +fn continued_history_stays_quiet_after_its_continuation_is_consumed() { + let ctx = context(vec![event( + "WorkflowContinuedAsNew", + json!({"sequence":1,"continued_to_run_id":"run-2"}), + )]); + let _live = replay_clock::Guard::enter(&ctx); + assert!(ctx.continue_as_new(json!([])).is_err()); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Replaying); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[tokio::test] +async fn actual_instance_state_query_reconstruction_never_emits_live_workflow_logs() { + let mut worker = worker(false); + worker.register_replayed_workflow( + "stateful", + || 0_u8, + |ctx, _, state| async move { + ctx.logger().info("entry"); + ctx.sleep(Duration::from_secs(1)).await?; + state.update(|n| *n = 1)?; + ctx.logger().info("after last history boundary"); + Ok(Value::Null) + }, + ); + worker.register_replayed_query::("stateful", "current", |_, state, _| async move { + Ok(json!(*state)) + }); + let query: crate::QueryTask = serde_json::from_value(json!({ + "query_task_id":"query-1","workflow_type":"stateful","query_name":"current", + "workflow_id":"order","run_id":"run-1","payload_codec":DEFAULT_CODEC, + "workflow_arguments":encode_value_envelope(&json!([]),DEFAULT_CODEC).unwrap(), + "query_arguments":encode_value_envelope(&json!([]),DEFAULT_CODEC).unwrap(), + "history_events":timers(1),"run_status":"running" + })) + .unwrap(); + let log = Arc::new(Log::default()); + let _subscriber = tracing::subscriber::set_default(log.clone()); + for _ in 0..2 { + assert_eq!( + worker + .execute_query_task(query.clone()) + .await + .unwrap() + .into_json() + .unwrap(), + json!(1) + ); + } + assert!(messages(&log).is_empty()); +} + +#[test] +fn subscriber_runs_without_a_workflow_lock_and_recursive_workflow_logs_are_silent() { + let ctx = context(vec![]); + let callback_ctx = ctx.clone(); + let log = Arc::new(Log { + on_event: Some(Box::new(move || { + assert!(callback_ctx.state.try_lock().is_ok()); + assert_eq!( + callback_ctx.replay_state(), + WorkflowReplayState::Unavailable + ); + callback_ctx.logger().warn("recursive"); + })), + ..Log::default() + }); + tracing::subscriber::with_default(log.clone(), || { + let _live = replay_clock::Guard::enter(&ctx); + ctx.logger().info("outer"); + assert_eq!(ctx.replay_state(), WorkflowReplayState::Executing); + }); + assert_eq!(messages(&log), ["outer"]); +} + +fn nexus_history(wait_for: &str, completed: bool) -> Vec { + let payload = json!({"sequence":1,"service_call_id":"call-1","endpoint_name":"billing", + "service_name":"payments","operation_name":"charge","operation_mode":"sync","wait_for":wait_for}); + let mut history = vec![event("ServiceCallStarted", payload.clone())]; + if completed { + let mut terminal = payload; + terminal["response_payload"] = + encode_value_envelope(&json!("paid"), DEFAULT_CODEC).unwrap(); + history.push(event("ServiceCallCompleted", terminal)); + } + history +} + +#[test] +fn recorded_nexus_without_outcome_stays_quiet_across_pending_polls() { + let ctx = context(nexus_history("completed", false)); + let inner = ctx.clone(); + let states = Arc::new(Mutex::new(vec![])); + let observed = states.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + let mut call = + Box::pin(inner.call_nexus_service("billing", "payments", "charge", json!([]))); + futures_util::future::poll_fn(|cx| { + for _ in 0..2 { + assert!(call.as_mut().poll(cx).is_pending()); + observed.lock().unwrap().push(inner.replay_state()); + inner.logger().info("pending Nexus replay"); + } + Poll::Pending + }) + .await + }); + let log = Arc::new(Log::default()); + tracing::subscriber::with_default( + log.clone(), + || assert!(poll(&ctx, &mut future).is_pending()), + ); + assert_eq!(*states.lock().unwrap(), [WorkflowReplayState::Replaying; 2]); + assert!(messages(&log).is_empty()); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[test] +fn nexus_terminal_result_and_accepted_only_admission_release_the_correct_boundary() { + use crate::{NexusServiceOptions, NexusWaitFor}; + let log = Arc::new(Log::default()); + tracing::subscriber::with_default(log.clone(), || { + for accepted_only in [false, true] { + let ctx = context(nexus_history( + if accepted_only { + "accepted" + } else { + "completed" + }, + !accepted_only, + )); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + assert_eq!(inner.replay_state(), WorkflowReplayState::Replaying); + inner.logger().info("recorded Nexus prefix"); + let result = inner + .call_nexus_service_with_options( + "billing", + "payments", + "charge", + NexusServiceOptions { + wait_for: if accepted_only { + NexusWaitFor::Accepted + } else { + NexusWaitFor::Completed + }, + ..Default::default() + }, + json!([]), + ) + .await?; + assert_eq!(result.service_call_id, "call-1"); + assert_eq!( + result.status, + if accepted_only { + "started" + } else { + "completed" + } + ); + assert_eq!( + result.result, + if accepted_only { + AvroValue::Null + } else { + AvroValue::String("paid".into()) + } + ); + assert_eq!(inner.replay_state(), WorkflowReplayState::Executing); + inner.logger().info(if accepted_only { + "admitted" + } else { + "completed" + }); + Ok(AvroValue::Null) + }); + assert!(matches!(poll(&ctx, &mut future), Poll::Ready(Ok(_)))); + assert!(ctx.take_commands().unwrap().is_empty()); + } + }); + assert_eq!(messages(&log), ["completed", "admitted"]); +} diff --git a/tests/fixtures/replay-regressions/replay-aware-pending-nexus-logging.json b/tests/fixtures/replay-regressions/replay-aware-pending-nexus-logging.json new file mode 100644 index 0000000..5950547 --- /dev/null +++ b/tests/fixtures/replay-regressions/replay-aware-pending-nexus-logging.json @@ -0,0 +1,19 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "rust-replay-aware-pending-nexus-logging", + "protocol_version": "1.20", + "bindings": ["rust"], + "workflow": {"type": "corpus.replay-aware-nexus-pending", "input": [], "payload_codec": "avro"}, + "history": [ + {"event_type": "ServiceCallStarted", "payload": { + "sequence": 1, "service_call_id": "call-1", "endpoint_name": "billing", "service_name": "payments", + "operation_name": "charge", "operation_mode": "sync", "wait_for": "completed", "status": "started" + }} + ], + "expected": { + "waiting_for_history": true, + "workflow_logs": [], + "task_failure": {"failure": {"type": "WorkflowTaskWaitingForHistory", "message": "Workflow task waiting for scheduled history."}} + } +} diff --git a/tests/fixtures/replay-regressions/replay-aware-workflow-logging.json b/tests/fixtures/replay-regressions/replay-aware-workflow-logging.json new file mode 100644 index 0000000..b1b7c14 --- /dev/null +++ b/tests/fixtures/replay-regressions/replay-aware-workflow-logging.json @@ -0,0 +1,14 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "rust-replay-aware-workflow-logging", + "protocol_version": "1.20", + "bindings": ["rust"], + "workflow": {"type": "corpus.replay-aware-logging", "input": [], "payload_codec": "avro"}, + "history": [ + {"event_type": "TimerScheduled", "payload": {"sequence": 1, "timer_id": "timer-1", "delay_seconds": 1}}, + {"event_type": "TimerFired", "payload": {"sequence": 1, "timer_id": "timer-1", "delay_seconds": 1}} + ], + "command_sequence": [{"type": "complete_workflow", "result": "complete"}], + "expected": {"type": "complete_workflow", "result": "complete", "workflow_logs": ["fresh"]} +} diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs index 7d75a63..c4c0669 100644 --- a/tests/replay_regression_corpus.rs +++ b/tests/replay_regression_corpus.rs @@ -30,6 +30,9 @@ mod business_clock; #[path = "replay_regression_corpus/redrive_successor_strictness.rs"] mod redrive_successor_strictness; +#[path = "replay_regression_corpus/workflow_logging.rs"] +mod workflow_logging; + #[derive(Clone, Debug, Deserialize, Serialize)] struct TypedReplayContract { message: String, @@ -528,6 +531,8 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< | "corpus.child-policy-author" | "corpus.activity-policy-author" | "corpus.business-replay-clock" + | "corpus.replay-aware-logging" + | "corpus.replay-aware-nexus-pending" | "corpus.business-replay-clock-child" | "corpus.business-replay-clock-signal" | "corpus.business-replay-clock-condition" @@ -670,6 +675,10 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< .map_err(|error| format!("{fixture_id} sticky cache configuration failed: {error}"))?; } match workflow_type { + "corpus.replay-aware-logging" => workflow_logging::register(&mut worker, workflow_type), + "corpus.replay-aware-nexus-pending" => { + workflow_logging::register_nexus_pending(&mut worker, workflow_type) + } "corpus.business-replay-clock" => business_clock::register(&mut worker, workflow_type), "corpus.business-replay-clock-signal" => { business_clock::register_signal(&mut worker, workflow_type) @@ -1357,7 +1366,13 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< } else { None }; - let replay = worker.run_once().await; + let logs = Arc::new(workflow_logging::Capture::default()); + let replay = if workflow_logging::is_fixture(workflow_type) { + use tracing::instrument::WithSubscriber; + worker.run_once().with_subscriber(logs.clone()).await + } else { + worker.run_once().await + }; if fixture["worker_task"]["hydration_authority_refused"] == true { let requests = server.requests.lock().unwrap(); if requests.iter().any(|request| { @@ -1401,11 +1416,14 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< let failure = server.request_body(&failure_path).ok_or_else(|| { format!("{fixture_id} did not acknowledge its pending history") })?; - let observed = json!({ + let mut observed = json!({ "command_sequence": [], "waiting_for_history": true, "task_failure": failure, }); + if workflow_logging::is_fixture(workflow_type) { + observed["workflow_logs"] = logs.messages(); + } if let Some(expected_commands) = fixture.get("command_sequence") { fixture_matches( expected_commands, @@ -1494,6 +1512,9 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< json!({"history_page_size": poll_history_page_size}), ), ]); + if workflow_logging::is_fixture(workflow_type) { + observed.insert("workflow_logs".into(), logs.messages()); + } if let Some(worker_registration) = worker_registration { observed.insert("worker_registration".to_string(), worker_registration); } diff --git a/tests/replay_regression_corpus/workflow_logging.rs b/tests/replay_regression_corpus/workflow_logging.rs new file mode 100644 index 0000000..16b55ce --- /dev/null +++ b/tests/replay_regression_corpus/workflow_logging.rs @@ -0,0 +1,93 @@ +use super::*; + +// The official consumer also runs against the SDK before this API existed. +// Its fallback is ordinary host logging, which repeats the committed prefix. +#[allow(dead_code)] +trait LegacyWorkflowLogging { + fn logger(&self) -> LegacyLogger; +} +impl LegacyWorkflowLogging for WorkflowContext { + fn logger(&self) -> LegacyLogger { + LegacyLogger + } +} +#[allow(dead_code)] +struct LegacyLogger; +#[allow(dead_code)] +impl LegacyLogger { + fn info(&self, message: &str) { + tracing::event!(target:"durable_workflow::workflow",tracing::Level::INFO,message); + } +} + +pub(super) fn register(worker: &mut Worker, workflow_type: &str) { + worker.register_workflow(workflow_type, |ctx, _| async move { + ctx.logger().info("committed"); + ctx.sleep(Duration::from_secs(1)).await?; + ctx.logger().info("fresh"); + Ok(json!("complete")) + }); +} + +pub(super) fn is_fixture(workflow_type: &str) -> bool { + matches!( + workflow_type, + "corpus.replay-aware-logging" | "corpus.replay-aware-nexus-pending" + ) +} + +pub(super) fn register_nexus_pending(worker: &mut Worker, workflow_type: &str) { + worker.register_workflow(workflow_type, |ctx, _| async move { + ctx.logger().info("committed"); + let mut call = Box::pin(ctx.call_nexus_service("billing", "payments", "charge", json!([]))); + std::future::poll_fn(|cx| { + assert!(std::future::Future::poll(call.as_mut(), cx).is_pending()); + ctx.logger().info("pending Nexus replay"); + std::task::Poll::Pending + }) + .await + }); +} + +#[derive(Default)] +pub(super) struct Capture(Mutex>); +struct Message(Option); +impl tracing::field::Visit for Message { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + if field.name() == "message" { + self.0 = Some(value.into()); + } + } + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + self.0 = Some(format!("{value:?}")); + } + } +} +impl tracing::Subscriber for Capture { + fn enabled(&self, _: &tracing::Metadata<'_>) -> bool { + true + } + fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id { + tracing::span::Id::from_u64(1) + } + fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {} + fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {} + fn event(&self, event: &tracing::Event<'_>) { + if event.metadata().target() != "durable_workflow::workflow" { + return; + } + let mut message = Message(None); + event.record(&mut message); + if let Some(message) = message.0 { + self.0.lock().unwrap().push(message); + } + } + fn enter(&self, _: &tracing::span::Id) {} + fn exit(&self, _: &tracing::span::Id) {} +} +impl Capture { + pub(super) fn messages(&self) -> Value { + json!(*self.0.lock().unwrap()) + } +}