From d7bf36f7e84a5086d8b869451507c1b1b5a9cf27 Mon Sep 17 00:00:00 2001 From: Kilerd Date: Mon, 14 Sep 2026 15:51:25 +0900 Subject: [PATCH] refactor(message)!: expose spawned task handles --- MIGRATION.md | 41 +++++++++ README.md | 24 +++++- examples/message/src/main.rs | 25 ++++-- gotcha/src/message.rs | 124 ++++++++++++++------------- gotcha/src/message/tests.rs | 158 +++++++++++++++++++++++++++++++++++ 5 files changed, 304 insertions(+), 68 deletions(-) create mode 100644 gotcha/src/message/tests.rs diff --git a/MIGRATION.md b/MIGRATION.md index de84e59..644f13f 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -1,5 +1,6 @@ # Migration Guide +- [Unreleased: message task handles](#unreleased-message-task-handles) - [Unreleased: static OpenAPI descriptors](#unreleased-static-openapi-descriptors) - [Unreleased: owned scheduled tasks](#unreleased-owned-scheduled-tasks) - [Unreleased: shared startup](#unreleased-shared-startup) @@ -16,6 +17,46 @@ --- +# Unreleased: message task handles + +`Message` and `Messager` remain available in the same module and root exports, without a feature +flag. They provide a typed command convention and application context, rather than a queue or +transport. `send` still executes directly in the caller's future; its output (including application +errors), panic, and cancellation remain at that boundary. + +`Messager::spawn` now returns `tokio::task::JoinHandle` and accepts any message output, +including `Result`. Previously it only accepted `Output = ()` and discarded the handle. + +```rust,no_run +# use gotcha::{Message, Messager}; +# async fn example>>( +# messager: Messager<(), ()>, command: M, +# ) -> Result<(), Box> { +let task = messager.spawn(command); +let output = task.await?; // Task panic/cancellation is a JoinError. +let value = output?; // Application errors stay in the message's Result. +# let _ = value; +# Ok(()) +# } +``` + +Existing calls with a semicolon still compile. Code requiring a `()` return, such as a function +ending in `messager.spawn(command)` without a semicolon or a stored function pointer, must adapt +to the new return type. To explicitly keep the previous detached behavior, use +`drop(messager.spawn(command));`. + +Retaining a handle allows joining or aborting; it does **not** enable automatic cancellation on +drop. Dropping it, including when its owning future is cancelled, detaches the task and loses its +result. To cancel, call `task.abort()` and then `task.await` to observe completion. Abort requires +the task to yield and can race with normal completion; already performed side effects remain. + +These tasks are not registered with the application's shutdown or scheduler ownership. Await +`messager.send(command)` inside a scheduled execution when the command should share that execution's +lifetime. Commands that spawn further tasks must manage them separately. No queue, retry, or extra +task has been introduced: `spawn` returns the handle from its existing `tokio::spawn` call. + +--- + # Unreleased: static OpenAPI descriptors `#[api]` keeps its existing syntax and OpenAPI behavior. Its generated `Operable` now stores diff --git a/README.md b/README.md index bf9fb3c..5321cec 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ An enhanced web framework built on top of Axum, providing additional features an - 🔌 **WebSocket & SSE** - Real-time endpoints, re-exported and ready - 📁 **Static Files** - Serve static content effortlessly - ⏰ **Task Scheduling** - Cron and interval-based background tasks -- 💌 **Message System** - Built-in inter-service communication +- 💌 **Typed Commands** - Reusable messages with application state and configuration - ⚙️ **Smart Configuration** - Environment-based config with variable resolution - 🏗️ **Two APIs** - Choose between simple builder API or advanced trait-based API @@ -415,6 +415,28 @@ Configuration supports: A single underscore stays part of the field name, so snake_case fields are addressable, and typed fields (numbers, booleans) parse the value rather than rejecting it. +### Messages + +`Message` groups a command's input, output, and handler. `Messager` carries the +application context and can be extracted as `State>`. This optional convention lets +HTTP handlers, scheduled tasks, and other commands reuse the same operations. Ordinary async +service methods remain equally valid; no message feature flag is required. + +`messager.send(command).await` runs directly in the calling future. There is no queue, transport, +retry, or new task. Results and panics propagate to the caller; dropping the future cancels its +in-flight handler without undoing completed side effects. + +`messager.spawn(command)` starts a Tokio task and returns `JoinHandle`. Await it +to observe the result or a `JoinError` for panic/cancellation. If the command returns `Result`, +the application error stays in the inner result. Call `abort()` and then await the handle to request +cancellation and wait for completion. Dropping the handle detaches the task, even when caused by +cancelling the future that owns it; application shutdown does not automatically wait for it. + +To include a command in a scheduled execution's shutdown boundary, await `send` inside that +execution. See the [message example](examples/message/src/main.rs) for nested commands and joining +a spawned result, and the [migration guide](MIGRATION.md#unreleased-message-task-handles) for the +changed `spawn` return type. + ### Task Scheduling Requires the `task` feature. Registration collects work; execution starts after all application diff --git a/examples/message/src/main.rs b/examples/message/src/main.rs index 40fbea6..8e64aac 100644 --- a/examples/message/src/main.rs +++ b/examples/message/src/main.rs @@ -1,7 +1,8 @@ -//! Demonstrates the message system: a handler extracts `State>`, -//! sends a message that reads the application state, and that message in turn -//! spawns a fire-and-forget background message. +//! Demonstrates typed commands with shared application context. The HTTP handler +//! spawns a greeting command and joins its result; that command directly awaits +//! another message. Plain async service methods can also organize this work. +use gotcha::axum::http::StatusCode; use gotcha::prelude::*; use gotcha::{async_trait, Message, Messager}; @@ -10,7 +11,7 @@ struct AppState { greeting: String, } -/// Produces a greeting from the app state and kicks off a background message. +/// Produces a greeting from the app state and awaits a nested message. struct Greet { name: String, } @@ -19,24 +20,30 @@ struct Greet { impl Message for Greet { type Output = String; async fn handle(self, messager: Messager) -> String { - messager.spawn(LogGreeting(self.name.clone())); + messager.send(LogGreeting(self.name.clone())).await; format!("{}, {}!", messager.state().greeting, self.name) } } -/// A fire-and-forget background message. +/// A message executed within the greeting command's future. struct LogGreeting(String); #[async_trait] impl Message for LogGreeting { type Output = (); async fn handle(self, _messager: Messager) { - println!("[background] greeted {}", self.0); + println!("greeted {}", self.0); } } -async fn hello(State(messager): State>) -> impl Responder { - messager.send(Greet { name: "world".to_string() }).await +async fn hello(State(messager): State>) -> Result { + let task = messager.spawn(Greet { name: "world".to_string() }); + // Awaiting observes completion and panics. Dropping this handle would detach + // the task, including if the HTTP handler's future were cancelled. + task.await.map_err(|error| { + gotcha::tracing::error!(%error, "Greeting task failed"); + StatusCode::INTERNAL_SERVER_ERROR + }) } #[tokio::main] diff --git a/gotcha/src/message.rs b/gotcha/src/message.rs index 0e74cd3..447360f 100644 --- a/gotcha/src/message.rs +++ b/gotcha/src/message.rs @@ -1,38 +1,67 @@ -//! # Message Module +//! Typed commands with access to the application context. //! -//! A small message-dispatch system for background/async work with access to the -//! application state. +//! A [`Message`] groups a command's input, output, and handler. [`Messager`] passes +//! the application state/config to that handler and to any further messages it +//! sends. This convention lets HTTP handlers and scheduled tasks reuse commands; +//! ordinary async service methods remain an equally valid way to organize work. +//! Messages are always available, without a feature flag. //! -//! Implement [`Message`] for a unit of work, then dispatch it with a [`Messager`]: -//! `send` awaits the result, `spawn` runs it fire-and-forget. The handler receives -//! the `Messager`, so it can read the application state/config -//! ([`Messager::state`] / [`Messager::context`]) and dispatch further messages. +//! There is no queue, worker pool, retry, or transport between services: +//! [`send`](Messager::send) directly awaits the handler in the calling future; +//! [`spawn`](Messager::spawn) starts a Tokio task and returns its [`JoinHandle`]. +//! Both clone the context, using the state/config types' own `Clone` behavior. //! //! The `Messager` is extractable in handlers as `State>`, because it //! implements `FromRef>` (the context the framework injects as //! the axum state). //! -//! ```ignore -//! use gotcha::prelude::*; -//! use gotcha::{async_trait, Message, Messager}; +//! ```rust,no_run +//! use gotcha::{async_trait, ConfigWrapper, GotchaContext, Message, Messager, State}; +//! +//! #[derive(Clone)] +//! struct AppState { greeting: String } //! //! struct Greet { name: String } //! //! #[async_trait] -//! impl Message for Greet { +//! impl Message for Greet { //! type Output = String; -//! async fn handle(self, messager: Messager) -> String { +//! async fn handle(self, messager: Messager) -> String { //! format!("{}, {}!", messager.state().greeting, self.name) //! } //! } //! -//! async fn hello(State(messager): State>) -> impl Responder { +//! async fn hello(State(messager): State>) -> String { //! messager.send(Greet { name: "world".into() }).await //! } +//! +//! # async fn example() -> Result<(), tokio::task::JoinError> { +//! let messager = Messager::new(GotchaContext { +//! state: AppState { greeting: "Hello".into() }, +//! config: ConfigWrapper::<()>::default(), +//! }); +//! let task = messager.spawn(Greet { name: "world".into() }); +//! assert_eq!(task.await?, "Hello, world!"); +//! # Ok(()) +//! # } //! ``` +//! +//! # Task lifetime +//! +//! Managing spawned messages is the caller's responsibility; they are not +//! registered with application shutdown or the task scheduler. Dropping their +//! handle (including by cancelling a future that owns it) detaches the task; it +//! does not cancel it. Retain the handle to join or abort the task. The runtime +//! must remain alive for the task to finish. +//! +//! To keep a message inside a scheduled execution's existing cancellation and +//! shutdown boundary, await `messager.send(message)` in that execution. A message +//! that spawns further tasks must manage those tasks separately. Cancellation +//! drops futures; it does not roll back side effects already performed. use async_trait::async_trait; use axum::extract::FromRef; +use tokio::task::JoinHandle; use crate::GotchaContext; @@ -46,7 +75,8 @@ where S: Clone + Send + Sync + 'static, C: Clone + Send + Sync + 'static, { - /// The value produced by handling this message. + /// The value produced by handling this message, including any application + /// error when this is a `Result`. type Output: Send + 'static; /// Handle the message, producing its output. @@ -95,15 +125,32 @@ where &self.context.state } - /// Dispatch a message and await its output. + /// Execute a message directly in the calling future and return its output. + /// + /// Nothing executes until this future is polled. There is no queue or new + /// task: application errors are returned unchanged, panics unwind through + /// the caller, and dropping this future drops the in-flight handler future. pub async fn send>(&self, message: M) -> M::Output { message.handle(self.clone()).await } - /// Dispatch a message as a detached background task (fire-and-forget). - pub fn spawn>(&self, message: M) { + /// Spawn a Tokio task and return its handle, without waiting for completion. + /// + /// Awaiting the handle returns `Result`. + /// If the output itself is a `Result`, application errors remain in that + /// inner result; a panic or task cancellation produces the outer `JoinError`. + /// + /// Dropping the handle detaches the task. To request cancellation, call + /// [`JoinHandle::abort`], then await the handle to observe completion. Abort + /// takes effect when the task yields and may race with normal completion. + /// Application shutdown does not automatically cancel or join this task. + /// + /// # Panics + /// + /// Panics if called outside a Tokio runtime. + pub fn spawn>(&self, message: M) -> JoinHandle { let messager = self.clone(); - tokio::spawn(async move { message.handle(messager).await }); + tokio::spawn(async move { message.handle(messager).await }) } } @@ -118,43 +165,4 @@ where } #[cfg(test)] -mod tests { - use super::*; - use crate::{ConfigWrapper, EmptyConfig}; - - #[derive(Clone, Default)] - struct AppState { - greeting: String, - } - - struct Greet { - name: String, - } - - #[async_trait] - impl Message for Greet { - type Output = String; - async fn handle(self, messager: Messager) -> String { - format!("{}, {}!", messager.state().greeting, self.name) - } - } - - #[test] - fn send_dispatches_and_reads_state() { - let context = GotchaContext { - config: ConfigWrapper { - server: Default::default(), - app: EmptyConfig::default(), - }, - state: AppState { greeting: "Hello".to_string() }, - }; - let messager = Messager::new(context); - - let output = tokio::runtime::Builder::new_current_thread() - .build() - .unwrap() - .block_on(messager.send(Greet { name: "world".to_string() })); - - assert_eq!(output, "Hello, world!"); - } -} +mod tests; diff --git a/gotcha/src/message/tests.rs b/gotcha/src/message/tests.rs new file mode 100644 index 0000000..4c3a7a8 --- /dev/null +++ b/gotcha/src/message/tests.rs @@ -0,0 +1,158 @@ +use super::*; +use crate::{ConfigWrapper, EmptyConfig}; +use futures_util::{task::noop_waker_ref, FutureExt}; +use std::future::Future; +use std::panic::AssertUnwindSafe; +use std::sync::Arc; +use std::task::Context; +use std::time::Duration; +use tokio::sync::Notify; + +#[derive(Clone, Default)] +struct AppState { + greeting: String, +} + +struct Greet { + name: String, +} + +#[async_trait] +impl Message for Greet { + type Output = String; + async fn handle(self, messager: Messager) -> String { + format!("{}, {}!", messager.state().greeting, self.name) + } +} + +#[test] +fn send_dispatches_and_reads_state() { + let context = GotchaContext { + config: ConfigWrapper { + server: Default::default(), + app: EmptyConfig::default(), + }, + state: AppState { greeting: "Hello".to_string() }, + }; + let messager = Messager::new(context); + + let output = tokio::runtime::Builder::new_current_thread() + .build() + .unwrap() + .block_on(messager.send(Greet { name: "world".to_string() })); + + assert_eq!(output, "Hello, world!"); +} + +fn messager() -> Messager<(), ()> { + Messager::new(GotchaContext { + config: ConfigWrapper::default(), + state: (), + }) +} + +struct Outcome(Result); + +#[async_trait] +impl Message<(), ()> for Outcome { + type Output = Result; + + async fn handle(self, _: Messager<(), ()>) -> Self::Output { + self.0 + } +} + +struct Panics; + +#[async_trait] +impl Message<(), ()> for Panics { + type Output = (); + + async fn handle(self, _: Messager<(), ()>) { + panic!("message failed"); + } +} + +#[tokio::test] +async fn results_and_panics_reach_the_caller_through_the_correct_boundary() { + let messager = messager(); + for output in [Ok(42), Err("rejected")] { + assert_eq!(messager.send(Outcome(output)).await, output); + assert_eq!(messager.spawn(Outcome(output)).await.unwrap(), output); + } + + let panic = AssertUnwindSafe(messager.send(Panics)).catch_unwind().await.unwrap_err(); + assert_eq!(panic.downcast_ref::<&str>(), Some(&"message failed")); + let error = messager.spawn(Panics).await.unwrap_err(); + assert!(error.is_panic()); + assert_eq!(error.into_panic().downcast_ref::<&str>(), Some(&"message failed")); +} + +#[derive(Clone, Default)] +struct WaitForRelease { + started: Arc, + release: Arc, + dropped: Arc, +} + +struct DropSignal(Arc); + +impl Drop for DropSignal { + fn drop(&mut self) { + self.0.notify_one(); + } +} + +#[async_trait] +impl Message<(), ()> for WaitForRelease { + type Output = u32; + + async fn handle(self, _: Messager<(), ()>) -> u32 { + let _resource = DropSignal(self.dropped); + self.started.notify_one(); + self.release.notified().await; + 42 + } +} + +#[test] +fn send_is_lazy_and_dropping_its_future_drops_the_handler_without_a_runtime() { + let messager = messager(); + let message = WaitForRelease::default(); + let mut future = Box::pin(messager.send(message.clone())); + assert!(message.started.notified().now_or_never().is_none()); + + // Poll directly without a runtime: execution must remain in the calling future. + assert!(future.as_mut().poll(&mut Context::from_waker(noop_waker_ref())).is_pending()); + assert!(message.started.notified().now_or_never().is_some()); + assert!(message.dropped.notified().now_or_never().is_none()); + + drop(future); + assert!(message.dropped.notified().now_or_never().is_some()); +} + +#[tokio::test(start_paused = true)] +async fn spawn_handle_can_abort_and_join_the_in_flight_message() { + let message = WaitForRelease::default(); + let handle = messager().spawn(message.clone()); + tokio::time::timeout(Duration::from_secs(1), message.started.notified()).await.unwrap(); + assert!(message.dropped.notified().now_or_never().is_none()); + + handle.abort(); + let error = tokio::time::timeout(Duration::from_secs(1), handle).await.unwrap().unwrap_err(); + assert!(error.is_cancelled()); + assert!(message.dropped.notified().now_or_never().is_some()); +} + +#[tokio::test(start_paused = true)] +async fn dropping_the_spawn_handle_detaches_the_message() { + let message = WaitForRelease::default(); + let handle = messager().spawn(message.clone()); + tokio::time::timeout(Duration::from_secs(1), message.started.notified()).await.unwrap(); + + drop(handle); + tokio::task::yield_now().await; + assert!(message.dropped.notified().now_or_never().is_none()); + message.release.notify_one(); + tokio::time::timeout(Duration::from_secs(1), message.dropped.notified()).await.unwrap(); +}