From 01c92711f8f7715b50a0c1951b855eb30b2d4b7d Mon Sep 17 00:00:00 2001 From: Roland Rodriguez Date: Sun, 11 Oct 2026 18:27:30 -0600 Subject: [PATCH] fix(runtime): honor hook limits and preserve mutation completion Keep committed hooks, audit events, and webhook dispatch independent of the HTTP receiver. Await mutation outcomes without a reply deadline and return an explicit unknown-outcome status when the actor reply is lost. Fixes #259 Fixes #260 --- CHANGELOG.md | 21 + Cargo.lock | 4 +- crates/schema-forge-acton/Cargo.toml | 2 +- crates/schema-forge-acton/src/actor.rs | 24 +- crates/schema-forge-acton/src/error.rs | 6 + crates/schema-forge-acton/src/events.rs | 4 +- .../src/hooks/dispatch_actor.rs | 89 ++- crates/schema-forge-acton/src/hooks/mod.rs | 8 +- .../src/hooks/tonic_dispatcher.rs | 11 +- crates/schema-forge-acton/src/messages.rs | 75 +- .../src/routes/create_intents.rs | 44 +- .../schema-forge-acton/src/routes/entities.rs | 662 ++++++++++-------- .../tests/hook_dispatch_limits.rs | 111 +++ .../tests/mutation_completion.rs | 437 ++++++++++++ .../tests/tonic_dispatcher_integration.rs | 46 ++ crates/schema-forge-cli/Cargo.toml | 4 +- docs/hooks-reference.md | 15 +- 17 files changed, 1200 insertions(+), 363 deletions(-) create mode 100644 crates/schema-forge-acton/tests/hook_dispatch_limits.rs create mode 100644 crates/schema-forge-acton/tests/mutation_completion.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 013f325b..c506053c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,27 @@ is pre-1.0; breaking changes bump the **minor** version per ## [Unreleased] +## [0.52.4] - 2026-10-11 + +### Fixed + +- Honor `schema_forge.hooks.default_timeout_ms` for gRPC bindings without a + per-binding timeout, and enforce `max_concurrent_async` for background hooks. + Zero concurrency is treated as one to keep queued hooks making progress. +- Await entity mutation replies without the former five-second deadline. + Slow creates, updates, patches, and deletes now acknowledge their actual + storage result. A lost actor reply returns 504 `mutation_outcome_unknown`, + directing callers to reconcile before retrying. +- Run committed mutation hooks, audit events, and webhook dispatch from actor + completion, including after HTTP cancellation. Resolve runtime webhook + subscriptions directly through storage to avoid actor self-deadlocks. + Create-intent audit completion also survives cancelled requests. + +### Upgrade notes + +- Restart the service to apply hook concurrency configuration. Post-commit + dispatch remains process-local; this release does not add durable delivery. + ## [0.52.3] - 2026-10-11 ### Fixed diff --git a/Cargo.lock b/Cargo.lock index 4f975bce..3883f423 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6743,7 +6743,7 @@ dependencies = [ [[package]] name = "schema-forge-acton" -version = "0.50.2" +version = "0.50.3" dependencies = [ "acton-service", "arc-swap", @@ -6829,7 +6829,7 @@ dependencies = [ [[package]] name = "schema-forge-cli" -version = "0.52.3" +version = "0.52.4" dependencies = [ "acton-service", "assert_cmd", diff --git a/crates/schema-forge-acton/Cargo.toml b/crates/schema-forge-acton/Cargo.toml index 9592e6e1..678d12df 100644 --- a/crates/schema-forge-acton/Cargo.toml +++ b/crates/schema-forge-acton/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "schema-forge-acton" -version = "0.50.2" +version = "0.50.3" edition = "2021" [dependencies] diff --git a/crates/schema-forge-acton/src/actor.rs b/crates/schema-forge-acton/src/actor.rs index d6a02614..1585b7bf 100644 --- a/crates/schema-forge-acton/src/actor.rs +++ b/crates/schema-forge-acton/src/actor.rs @@ -451,7 +451,7 @@ fn configure_backend_operations(actor: &mut ManagedActor) { let backend = actor.model.backend.clone(); let request = ctx.message().clone(); Reply::pending(async move { - let result = match backend { + let result = match backend.clone() { Some(backend) => match &request.membership { Some(membership) => { backend @@ -462,7 +462,7 @@ fn configure_backend_operations(actor: &mut ManagedActor) { }, None => Err(schema_forge_backend::create_intent::CreateIntentError::Unsupported), }; - request.reply.send(result).await; + request.reply.send_with_backend(result, backend).await; }) }); @@ -472,7 +472,7 @@ fn configure_backend_operations(actor: &mut ManagedActor) { let membership = ctx.message().membership.clone(); let reply = ctx.message().reply.clone(); Reply::pending(async move { - let result = match backend { + let result = match backend.clone() { Some(b) => match &membership { Some(membership) => b.create_with_membership(&entity, membership).await, None => b.create(&entity).await, @@ -482,7 +482,7 @@ fn configure_backend_operations(actor: &mut ManagedActor) { Err(no_backend_error()) } }; - reply.send(result).await; + reply.send_with_backend(result, backend).await; }) }); @@ -523,7 +523,7 @@ fn configure_backend_operations(actor: &mut ManagedActor) { let backend = actor.model.backend.clone(); let request = ctx.message().clone(); Reply::pending(async move { - let result = match backend { + let result = match backend.clone() { Some(b) => b.update_if(&request.entity, &request.expected).await, None => Err( schema_forge_backend::conditional::ConditionalMutationError::Backend( @@ -531,7 +531,7 @@ fn configure_backend_operations(actor: &mut ManagedActor) { ), ), }; - request.reply.send(result).await; + request.reply.send_with_backend(result, backend).await; }) }); @@ -539,7 +539,7 @@ fn configure_backend_operations(actor: &mut ManagedActor) { let backend = actor.model.backend.clone(); let request = ctx.message().clone(); Reply::pending(async move { - let result = match backend { + let result = match backend.clone() { Some(b) => { b.delete_if(&request.schema, &request.id, &request.expected) .await @@ -550,7 +550,7 @@ fn configure_backend_operations(actor: &mut ManagedActor) { ), ), }; - request.reply.send(result).await; + request.reply.send_with_backend(result, backend).await; }) }); @@ -581,14 +581,14 @@ fn configure_backend_operations(actor: &mut ManagedActor) { let entity = ctx.message().entity.clone(); let reply = ctx.message().reply.clone(); Reply::pending(async move { - let result = match backend { + let result = match backend.clone() { Some(b) => b.update(&entity).await, None => { warn!("UpdateEntity received but no backend is configured"); Err(no_backend_error()) } }; - reply.send(result).await; + reply.send_with_backend(result, backend).await; }) }); @@ -598,14 +598,14 @@ fn configure_backend_operations(actor: &mut ManagedActor) { let id = ctx.message().id.clone(); let reply = ctx.message().reply.clone(); Reply::pending(async move { - let result = match backend { + let result = match backend.clone() { Some(b) => b.delete(&schema, &id).await, None => { warn!("DeleteEntity received but no backend is configured"); Err(no_backend_error()) } }; - reply.send(result).await; + reply.send_with_backend(result, backend).await; }) }); diff --git a/crates/schema-forge-acton/src/error.rs b/crates/schema-forge-acton/src/error.rs index f6556021..0e80b86e 100644 --- a/crates/schema-forge-acton/src/error.rs +++ b/crates/schema-forge-acton/src/error.rs @@ -67,6 +67,9 @@ pub enum ForgeError { invite_id: String, accept_url: String, }, + /// The mutation reply was lost after submission. Its outcome is unknown. + /// Maps to 504; callers must reconcile before retrying. + MutationOutcomeUnknown, /// Internal error. Maps to 500. Internal { message: String }, } @@ -141,6 +144,7 @@ impl fmt::Display for ForgeError { Self::InviteDeliveryFailed { .. } => { write!(f, "Invitation created, but email delivery failed. Share the accept link to complete onboarding.") } + Self::MutationOutcomeUnknown => write!(f, "Mutation outcome is unknown. Reconcile the record or create intent before retrying."), Self::Internal { message } => { write!(f, "internal error: {message}") } @@ -173,6 +177,7 @@ impl ForgeError { Self::RateLimited { .. } => StatusCode::TOO_MANY_REQUESTS, Self::HookAborted { .. } => StatusCode::UNPROCESSABLE_ENTITY, Self::HookUnavailable { .. } => StatusCode::SERVICE_UNAVAILABLE, + Self::MutationOutcomeUnknown => StatusCode::GATEWAY_TIMEOUT, Self::Internal { .. } => StatusCode::INTERNAL_SERVER_ERROR, } } @@ -199,6 +204,7 @@ impl ForgeError { Self::HookAborted { .. } => "hook_aborted", Self::HookUnavailable { .. } => "hook_unavailable", Self::InviteDeliveryFailed { .. } => "invite_delivery_failed", + Self::MutationOutcomeUnknown => "mutation_outcome_unknown", Self::Internal { .. } => "internal_error", } } diff --git a/crates/schema-forge-acton/src/events.rs b/crates/schema-forge-acton/src/events.rs index 3f1c553b..bd67f38c 100644 --- a/crates/schema-forge-acton/src/events.rs +++ b/crates/schema-forge-acton/src/events.rs @@ -260,7 +260,7 @@ pub(crate) fn configure_actor(actor: &mut ManagedActor) { Ok(stored) => entity = stored, Err(error) => { tracing::warn!(%error, "committed create snapshot unavailable; no event will be published"); - envelope.mutation.reply.send(result).await; + envelope.mutation.reply.send_with_backend(result, Some(backend.clone())).await; return; } } @@ -285,7 +285,7 @@ pub(crate) fn configure_actor(actor: &mut ManagedActor) { } } } - envelope.mutation.reply.send(result).await; + envelope.mutation.reply.send_with_backend(result, Some(backend.clone())).await; } else { tracing::error!("event mutation received without backend"); } diff --git a/crates/schema-forge-acton/src/hooks/dispatch_actor.rs b/crates/schema-forge-acton/src/hooks/dispatch_actor.rs index ca2658c0..82ec2e1f 100644 --- a/crates/schema-forge-acton/src/hooks/dispatch_actor.rs +++ b/crates/schema-forge-acton/src/hooks/dispatch_actor.rs @@ -11,16 +11,17 @@ //! The fix is a tiny dedicated actor whose only job is to receive a //! [`DispatchHook`] message and run [`run_after_hook`] under acton's //! scheduler. The actor uses `act_on` (read-only state, concurrent -//! handler execution) so a slow hook endpoint can't block other -//! dispatches — the runtime's high-water mark is the natural -//! backpressure mechanism. +//! handler execution). Configured semaphore admission bounds active calls; +//! the runtime also bounds batches and buffers queued messages. A slow batch +//! can apply backpressure to subsequent dispatches. //! //! Per-schema or per-endpoint pooling is a potential future //! optimization; one shared actor is sufficient for the current //! workload and removes the deadlock entirely because the actor never //! holds a lock on the trigger entity row. -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; +use tokio::sync::{oneshot, Mutex, Semaphore}; use acton_service::prelude::*; use sync_wrapper::SyncFuture; @@ -49,29 +50,77 @@ pub struct HookDispatchActor; // `InitForge` at boot and which therefore opts out. impl ActorExtension for HookDispatchActor { fn configure(actor: &mut ManagedActor) { - actor.act_on::(|_actor, ctx| { + // Configuration is immutable for an actor incarnation. Keep the unit + // actor's public construction compatible and initialize admission from + // the first dispatch, including after an actor restart. + let limit = Arc::new(OnceLock::>::new()); + let deferred_limit = limit.clone(); + actor.act_on::(move |_actor, ctx| { let msg = ctx.message().clone(); - // The acton-reactive `FutureBox` requires `Send + Sync`, - // but the `HookDispatcher` trait returns `BoxFuture<'_, T>` - // which is `Send` only (the `async_trait` default). Wrap - // the async block in `SyncFuture` to satisfy the bound; - // this is sound because `Future::poll` already requires - // exclusive access to the future via `Pin<&mut Self>`. + let semaphore = limit + .get_or_init(|| { + Arc::new(Semaphore::new( + msg.config + .max_concurrent_async + .clamp(1, Semaphore::MAX_PERMITS), + )) + }) + .clone(); + Reply::pending(SyncFuture::new(run_dispatch(msg, semaphore))) + }); + actor.act_on::(move |_actor, ctx| { + let message = ctx.message().clone(); + let semaphore = deferred_limit + .get_or_init(|| { + Arc::new(Semaphore::new( + message + .config + .max_concurrent_async + .clamp(1, Semaphore::MAX_PERMITS), + )) + }) + .clone(); Reply::pending(SyncFuture::new(async move { - let Some(dispatcher) = msg.dispatcher else { - warn!( - schema = %msg.invocation.schema, - event = ?msg.invocation.event, - "DispatchHook received but no dispatcher is configured" - ); - return; - }; - run_after_hook(dispatcher.as_ref(), &msg.config, msg.invocation).await; + let receiver = message.receiver.lock().await.take(); + if let Some(receiver) = receiver { + if let Ok(dispatch) = receiver.await { + run_dispatch(dispatch, semaphore).await; + } + } })) }); } } +async fn run_dispatch(message: DispatchHook, semaphore: Arc) { + let Some(dispatcher) = message.dispatcher else { + warn!(schema = %message.invocation.schema, "hook dispatch received without a dispatcher"); + return; + }; + let Ok(_permit) = semaphore.acquire().await else { + warn!("after-hook admission closed"); + return; + }; + run_after_hook(dispatcher.as_ref(), &message.config, message.invocation).await; +} + +/// Reserve hook queue admission before submitting a mutation. The actor waits +/// for a commit payload, or exits if validation/submission/storage fails. +/// The receiver has one owner; the wrapper only satisfies message cloning. +#[derive(Clone, Debug)] +pub(crate) struct DeferredHook { + pub config: HooksConfig, + receiver: Arc>>>, +} +impl DeferredHook { + pub(crate) fn new(config: HooksConfig, receiver: oneshot::Receiver) -> Self { + Self { + config, + receiver: Arc::new(Mutex::new(Some(receiver))), + } + } +} + // --------------------------------------------------------------------------- // DispatchHook message // --------------------------------------------------------------------------- diff --git a/crates/schema-forge-acton/src/hooks/mod.rs b/crates/schema-forge-acton/src/hooks/mod.rs index ec11e8a2..ad373916 100644 --- a/crates/schema-forge-acton/src/hooks/mod.rs +++ b/crates/schema-forge-acton/src/hooks/mod.rs @@ -27,10 +27,10 @@ //! operate on the already-defaulted/computed/validated field set and a //! `@require` rejection short-circuits the write before any hook //! round-trip. -//! 2. **Detached (`after_*`)**: the route handler queues a -//! [`dispatch_actor::DispatchHook`] message on the -//! [`dispatch_actor::HookDispatchActor`] and returns immediately to -//! the caller. The actor runs the dispatch concurrently under +//! 2. **Detached (`after_*`)**: the route reserves bounded queue space on +//! [`dispatch_actor::HookDispatchActor`] before submitting the mutation. +//! Actor-owned completion supplies the hook payload after storage succeeds, +//! including when the HTTP receiver has disconnected. The actor runs the dispatch under //! acton-reactive supervision, so in-flight hook calls participate in //! graceful shutdown and the runtime's backpressure budget. Errors //! are logged. diff --git a/crates/schema-forge-acton/src/hooks/tonic_dispatcher.rs b/crates/schema-forge-acton/src/hooks/tonic_dispatcher.rs index 1737ca28..28ad8d4f 100644 --- a/crates/schema-forge-acton/src/hooks/tonic_dispatcher.rs +++ b/crates/schema-forge-acton/src/hooks/tonic_dispatcher.rs @@ -44,10 +44,7 @@ use tonic::{Request, Status}; use tracing::{debug, warn}; use super::credential::HookCredentialSource; -use super::{ - HookBinding, HookDispatcher, HookError, HookInvocation, HookOutcome, HooksConfig, - DEFAULT_HOOK_TIMEOUT_MS, -}; +use super::{HookBinding, HookDispatcher, HookError, HookInvocation, HookOutcome, HooksConfig}; /// gRPC metadata key carrying the bearer credential. /// @@ -126,6 +123,7 @@ pub struct TonicHookDispatcher { /// Whether `http://` endpoints may be dialed. See /// [`HooksConfig::allow_plaintext`]. allow_plaintext: bool, + default_timeout_ms: u32, } impl TonicHookDispatcher { @@ -213,6 +211,7 @@ impl TonicHookDispatcher { channels: Mutex::new(HashMap::new()), tls, allow_plaintext: hooks.allow_plaintext, + default_timeout_ms: hooks.default_timeout_ms, }) } @@ -338,7 +337,7 @@ impl HookDispatcher for TonicHookDispatcher { endpoint = %binding.endpoint, "tonic dispatch (before)" ); - let timeout = binding.timeout_ms.unwrap_or(DEFAULT_HOOK_TIMEOUT_MS); + let timeout = binding.timeout_ms.unwrap_or(self.default_timeout_ms); let response = self.invoke(binding, invocation, timeout).await?; Ok(decode_outcome(&response)) } @@ -354,7 +353,7 @@ impl HookDispatcher for TonicHookDispatcher { endpoint = %binding.endpoint, "tonic dispatch (after)" ); - let timeout = binding.timeout_ms.unwrap_or(DEFAULT_HOOK_TIMEOUT_MS); + let timeout = binding.timeout_ms.unwrap_or(self.default_timeout_ms); let _ = self.invoke(binding, invocation, timeout).await?; Ok(()) } diff --git a/crates/schema-forge-acton/src/messages.rs b/crates/schema-forge-acton/src/messages.rs index c65b1870..b29efca1 100644 --- a/crates/schema-forge-acton/src/messages.rs +++ b/crates/schema-forge-acton/src/messages.rs @@ -23,23 +23,69 @@ use tokio::sync::{oneshot, Mutex}; // ReplyChannel — Clone-safe oneshot wrapper // --------------------------------------------------------------------------- +type Completion = Box< + dyn FnOnce( + &T, + Option>, + ) -> std::pin::Pin + Send + Sync>> + + Send + + Sync, +>; +struct PendingReply { + sender: oneshot::Sender, + completion: Option>, +} /// A single-use reply channel that satisfies `Clone` (required by `ActonMessage`). /// /// Wraps a `tokio::sync::oneshot::Sender` in `Arc>>` so that /// `Clone` is trivially implemented (via `Arc::clone`). Only the first call to /// [`send`](ReplyChannel::send) delivers the value; subsequent calls are no-ops. -pub struct ReplyChannel(Arc>>>); +pub struct ReplyChannel(Arc>>>); impl ReplyChannel { /// Create a new `ReplyChannel` from a `oneshot::Sender`. pub fn new(sender: oneshot::Sender) -> Self { - Self(Arc::new(Mutex::new(Some(sender)))) + Self(Arc::new(Mutex::new(Some(PendingReply { + sender, + completion: None, + })))) } /// Send a value through the channel. Only the first call delivers; subsequent /// calls are silently ignored (the sender has already been consumed). pub async fn send(self, value: T) { - if let Some(sender) = self.0.lock().await.take() { + self.send_with_backend(value, None).await; + } + + /// Attach actor-owned post-commit work without changing mutation messages. + pub(crate) fn with_completion(sender: oneshot::Sender, completion: F) -> Self + where + F: FnOnce(&T, Option>) -> Fut + + Send + + Sync + + 'static, + Fut: std::future::Future + Send + Sync + 'static, + { + Self(Arc::new(Mutex::new(Some(PendingReply { + sender, + completion: Some(Box::new(move |value, backend| { + Box::pin(completion(value, backend)) + })), + })))) + } + + /// Run completion even if the HTTP receiver was dropped. Release the + /// single-use channel lock before awaiting any application work. + pub(crate) async fn send_with_backend( + self, + value: T, + backend: Option>, + ) { + let pending = self.0.lock().await.take(); + if let Some(PendingReply { sender, completion }) = pending { + if let Some(completion) = completion { + completion(&value, backend).await; + } let _ = sender.send(value); } } @@ -398,3 +444,26 @@ pub struct ProcessCreateIntent { >, >, } + +#[cfg(test)] +mod completion_tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[tokio::test] + async fn completion_runs_once_even_when_the_receiver_is_gone() { + let completed = Arc::new(AtomicUsize::new(0)); + let observed = completed.clone(); + let (tx, rx) = oneshot::channel(); + let reply = ReplyChannel::with_completion(tx, move |value: &usize, _| { + let value = *value; + async move { + observed.fetch_add(value, Ordering::SeqCst); + } + }); + drop(rx); + let second = reply.clone(); + tokio::join!(reply.send(1), second.send(1)); + assert_eq!(completed.load(Ordering::SeqCst), 1); + } +} diff --git a/crates/schema-forge-acton/src/routes/create_intents.rs b/crates/schema-forge-acton/src/routes/create_intents.rs index 52daa9f7..5a64694b 100644 --- a/crates/schema-forge-acton/src/routes/create_intents.rs +++ b/crates/schema-forge-acton/src/routes/create_intents.rs @@ -89,38 +89,58 @@ pub(super) async fn process( state: &AppState, request: CreateIntentRequest, ) -> Result { - process_with_claims(state, request, None, None).await + process_with_claims(state, request, None, None, None).await } async fn process_with_claims( state: &AppState, request: CreateIntentRequest, claims: Option<&Claims>, membership: Option, + headers: Option<&HeaderMap>, ) -> Result { let forge = state .actor::() .ok_or_else(|| error(CreateIntentError::Unsupported))?; let (tx, rx) = oneshot::channel(); let notification = entities::mutation_notification(claims, None); + let audit = match &request { + CreateIntentRequest::Commit { scope, entity, .. } => Some(( + scope.schema.name.to_string(), + entity.fields.keys().cloned().collect::>(), + )), + _ => None, + }; + let audit_state = state.clone(); + let audit_claims = claims.cloned(); + let audit_headers = headers.cloned().unwrap_or_default(); + let reply = ReplyChannel::with_completion( + tx, + move |result: &Result, _| { + let receipt = result + .as_ref() + .ok() + .filter(|receipt| receipt.created) + .cloned(); + sync_wrapper::SyncFuture::new(async move { + if let (Some(receipt), Some((schema, changed_fields))) = (receipt, audit) { + super::audit::log_forge_event(&audit_state, audit_claims.as_ref(), &audit_headers, + "forge.entity.created", acton_service::audit::AuditSeverity::Informational, + serde_json::json!({"schema": schema, "intent_id": receipt.id.as_str(), "changed_fields": changed_fields, "entity_id": receipt.entity_id.as_ref().map(|id| id.as_str())})).await; + } + }) + }, + ); entities::send_mutation( &forge, ProcessCreateIntent { request, membership, - reply: ReplyChannel::new(tx), + reply, }, ¬ification, ) .await?; - tokio::time::timeout(Duration::from_secs(5), rx) - .await - .map_err(|_| ForgeError::BackendUnavailable { - message: "Create outcome is uncertain. Reconcile this intent before retrying.".into(), - })? - .map_err(|_| ForgeError::BackendUnavailable { - message: "Create outcome is uncertain. Reconcile this intent before retrying.".into(), - })? - .map_err(error) + entities::await_mutation(rx).await?.map_err(error) } async fn scope( @@ -344,6 +364,7 @@ pub(super) async fn commit( entity: Entity, claims: Option<&Claims>, membership: Option, + headers: &HeaderMap, ) -> Result { process_with_claims( state, @@ -355,6 +376,7 @@ pub(super) async fn commit( }, claims, membership, + Some(headers), ) .await } diff --git a/crates/schema-forge-acton/src/routes/entities.rs b/crates/schema-forge-acton/src/routes/entities.rs index ec317d3b..0af564c1 100644 --- a/crates/schema-forge-acton/src/routes/entities.rs +++ b/crates/schema-forge-acton/src/routes/entities.rs @@ -48,7 +48,7 @@ use std::sync::Arc; #[cfg(feature = "sse")] use crate::events::Notification; #[cfg(not(feature = "sse"))] -#[derive(Default)] +#[derive(Clone, Default)] pub(super) struct Notification; pub(super) fn mutation_notification( @@ -142,6 +142,160 @@ pub(super) async fn ask_forge(rx: oneshot::Receiver) -> Result(rx: oneshot::Receiver) -> Result { + rx.await.map_err(|_| ForgeError::MutationOutcomeUnknown) +} + +/// Actor-owned commit effects. Do not ask ForgeActor from this completion: +/// its scheduler may be draining a batch or a serialized SSE mutation. +struct MutationCompletion { + after_hook: Option>, + state: AppState, + schema: SchemaDefinition, + claims: Option, + headers: HeaderMap, + dispatcher: Option>, + hooks: HooksConfig, + operation: &'static str, + before: Option>, + notification: Notification, +} + +impl MutationCompletion { + /// Acquire bounded hook queue admission in the HTTP route, before storage + /// starts. Commit completion only fills this one-shot payload, so a full + /// hook mailbox can never make ForgeActor wait on hooks calling it back. + async fn prepare(mut self) -> Self { + let event = if self.operation == "delete" { + HookEvent::AfterDelete + } else { + HookEvent::AfterChange + }; + if self.dispatcher.is_some() + && self.hooks.enabled + && self + .hooks + .binding_for(self.schema.name.as_str(), event) + .is_some() + { + if let Some(actor) = self.state.actor::() { + let (sender, receiver) = oneshot::channel(); + self.after_hook = Some(sender); + actor + .send(crate::hooks::dispatch_actor::DeferredHook::new( + self.hooks.clone(), + receiver, + )) + .await; + } else { + tracing::warn!(schema = %self.schema.name, "HookDispatchActor not registered; after-hook dispatch dropped"); + } + } + self + } + + async fn finish(self, entity: Entity, backend: Option>) { + let deleting = self.operation == "delete"; + let event_type = if deleting { + "deleted" + } else if self.operation == "create" { + "created" + } else { + "updated" + }; + if let (Some(dispatcher), Some(sender)) = (self.dispatcher, self.after_hook) { + complete_after_hook( + sender, + AfterHookCtx { + dispatcher, + hooks_config: self.hooks, + schema: self.schema.clone(), + event: if deleting { + HookEvent::AfterDelete + } else { + HookEvent::AfterChange + }, + operation: self.operation.into(), + user_id: self.claims.as_ref().map(|c| c.sub.clone()), + }, + &entity, + self.before.as_ref(), + ); + } + let changed_fields = match &self.before { + Some(before) => changed_field_names(before, &entity.fields), + None if deleting => Vec::new(), + None => entity.fields.keys().cloned().collect(), + }; + let audit_name = match self.operation { + "create" => "forge.entity.created", + "patch" => "forge.entity.patched", + "delete" => "forge.entity.deleted", + _ => "forge.entity.updated", + }; + super::audit::log_forge_event( + &self.state, + self.claims.as_ref(), + &self.headers, + audit_name, + if deleting { + acton_service::audit::AuditSeverity::Warning + } else { + acton_service::audit::AuditSeverity::Informational + }, + serde_json::json!({ + "schema": self.schema.name.as_str(), + "entity_id": entity.id.as_str(), + "changed_fields": changed_fields, + "user": self.claims.as_ref().map(|c| &c.sub), + }), + ) + .await; + let event = notification_event(&self.notification, || { + let subject = self.claims.as_ref().map(|c| c.sub.as_str()); + match event_type { + "created" => { + crate::webhook::WebhookEvent::from_create(&self.schema, &entity, subject) + } + "deleted" => crate::webhook::WebhookEvent::from_delete( + self.schema.name.as_str(), + entity.id.as_str(), + subject, + ), + _ => crate::webhook::WebhookEvent::from_update(&self.schema, &entity, subject), + } + }); + dispatch_webhook( + &self.state, + &self.schema, + event, + event_type, + backend.as_deref(), + ) + .await; + } +} + +fn mutation_reply( + sender: oneshot::Sender, + completion: MutationCompletion, + snapshot: F, +) -> ReplyChannel +where + F: FnOnce(&T) -> Option + Send + Sync + 'static, +{ + ReplyChannel::with_completion(sender, move |result, backend| { + let entity = snapshot(result); + sync_wrapper::SyncFuture::new(async move { + if let Some(entity) = entity { + completion.finish(entity, backend).await; + } + }) + }) +} + /// Crate-internal accessor for the compiled Cedar policy store, reused by the /// export route ([`crate::routes::export`]) which needs the same store to run /// the distinct `Export{Schema}` authorization check. @@ -330,8 +484,9 @@ async fn persist_entity_update( forge: &acton_service::prelude::ActorHandle, entity: Entity, expected: Option, - notification: &Notification, + completion: MutationCompletion, ) -> Result<(Entity, Option), ForgeError> { + let notification = completion.notification.clone(); if let Some(expected) = expected { let (tx, rx) = oneshot::channel(); send_mutation( @@ -339,12 +494,19 @@ async fn persist_entity_update( UpdateEntityIf { entity, expected, - reply: ReplyChannel::new(tx), + reply: mutation_reply( + tx, + completion, + |result: &Result< + schema_forge_backend::conditional::VersionedEntity, + ConditionalMutationError, + >| result.as_ref().ok().map(|v| v.entity.clone()), + ), }, - notification, + ¬ification, ) .await?; - let result = ask_forge(rx).await?.map_err(conditional_error)?; + let result = await_mutation(rx).await?.map_err(conditional_error)?; Ok((result.entity, Some(result.revision))) } else { let (tx, rx) = oneshot::channel(); @@ -352,12 +514,18 @@ async fn persist_entity_update( forge, UpdateEntity { entity, - reply: ReplyChannel::new(tx), + reply: mutation_reply( + tx, + completion, + |result: &Result| { + result.as_ref().ok().cloned() + }, + ), }, - notification, + ¬ification, ) .await?; - Ok((ask_forge(rx).await?.map_err(ForgeError::from)?, None)) + Ok((await_mutation(rx).await?.map_err(ForgeError::from)?, None)) } } @@ -481,7 +649,7 @@ async fn apply_read_hook( apply_before_hook(ctx, fields).await } -/// Owned parameter bundle for [`fire_after_hook`]. +/// Owned parameter bundle for [`complete_after_hook`]. /// /// Everything is owned and `'static`-friendly because the dispatch is /// handed off to [`HookDispatchActor`] for asynchronous execution under @@ -495,16 +663,10 @@ struct AfterHookCtx { user_id: Option, } -/// Queue an `after_*` hook on the [`HookDispatchActor`] for detached -/// dispatch. Honors the global `enabled` flag and per-schema binding -/// presence inside the actor, so this helper itself is unconditional -/// once a dispatcher exists. -/// -/// Detached, not "fire-and-forget": the dispatch runs under acton -/// supervision so in-flight hook calls participate in graceful shutdown -/// and the runtime's concurrency budget. -async fn fire_after_hook( - state: &AppState, +/// Fill a hook reservation after successful storage completion. Sending a +/// oneshot payload cannot wait for actor mailbox capacity. +fn complete_after_hook( + sender: oneshot::Sender, ctx: AfterHookCtx, entity: &Entity, before: Option<&BTreeMap>, @@ -522,28 +684,15 @@ async fn fire_after_hook( if let Some(before) = before { invocation = invocation.with_changes(before); } - match state.actor::() { - Some(actor) => { - actor - .send(DispatchHook { - invocation, - dispatcher: Some(ctx.dispatcher), - config: ctx.hooks_config, - }) - .await; - } - None => { - // Defensive fallback: if the dispatch actor wasn't wired - // into the service (e.g. a test that forgot to register - // it), log and drop. We deliberately do **not** spawn here - // — doing so would reintroduce the orphaned-task anti- - // pattern that issue #11 was filed against. - tracing::warn!( - schema = %invocation.schema, - event = ?invocation.event, - "HookDispatchActor not registered; after-hook dispatch dropped" - ); - } + if sender + .send(DispatchHook { + invocation, + dispatcher: Some(ctx.dispatcher), + config: ctx.hooks_config, + }) + .is_err() + { + tracing::warn!("reserved after-hook receiver unavailable after commit"); } } @@ -559,6 +708,7 @@ async fn dispatch_webhook( schema_def: &SchemaDefinition, event: crate::webhook::WebhookEvent, event_type: &str, + backend: Option<&dyn crate::state::DynForgeBackend>, ) { let webhook_config = &state.config().custom.schema_forge.webhooks; let dispatcher = match crate::webhook::get_dispatcher(webhook_config) { @@ -591,58 +741,14 @@ async fn dispatch_webhook( }); } - // 2. Runtime subscriptions via actor query - let forge = match state.actor::() { - Some(f) => f, - None => { - if !subs.is_empty() { - dispatcher.dispatch(event, subs); - } - return; - } - }; - - // Query WebhookSubscription entities for this schema - let (tx, rx) = oneshot::channel(); - let ws_schema = SchemaName::new("WebhookSubscription"); - if let Ok(ws_name) = ws_schema { - // First get the WebhookSubscription schema def to get its ID - let (stx, srx) = oneshot::channel(); - forge - .send(crate::messages::GetSchema { - name: ws_name.as_str().to_string(), - reply: ReplyChannel::new(stx), - }) - .await; - - if let Ok(Some(ws_def)) = ask_forge(srx).await { - use schema_forge_core::query::{FieldPath, Filter, Query as CoreQuery}; - let query = CoreQuery::new(ws_def.id.clone()).with_filter(Filter::and(vec![ - Filter::eq( - FieldPath::parse("target_schema").unwrap(), - DynamicValue::Text(schema_def.name.as_str().to_string()), - ), - Filter::eq( - FieldPath::parse("active").unwrap(), - DynamicValue::Boolean(true), - ), - ])); - - forge - .send(QueryEntities { - query, - reply: ReplyChannel::new(tx), - }) - .await; - - if let Ok(Ok(result)) = ask_forge(rx).await { - for entity in &result.entities { - if let Some(sub) = resolve_runtime_subscription(entity, event_type) { - subs.push(sub); - } - } - } - } + match runtime_webhook_subscriptions(state, schema_def, backend).await { + Ok(result) => subs.extend( + result + .entities + .iter() + .filter_map(|entity| resolve_runtime_subscription(entity, event_type)), + ), + Err(error) => tracing::warn!(%error, "runtime webhook subscriptions unavailable"), } if !subs.is_empty() { @@ -650,6 +756,63 @@ async fn dispatch_webhook( } } +/// Committed writes query storage directly to avoid a back-edge into their +/// actor. A no-op PATCH runs in the route and may use the actor safely. +async fn runtime_webhook_subscriptions( + state: &AppState, + schema: &SchemaDefinition, + backend: Option<&dyn crate::state::DynForgeBackend>, +) -> Result { + let name = SchemaName::new("WebhookSubscription").map_err(|error| ForgeError::Internal { + message: error.to_string(), + })?; + let forge = state.actor::(); + let definition = if let Some(backend) = backend { + backend + .load_schema_metadata(&name) + .await + .map_err(ForgeError::from)? + } else if let Some(forge) = &forge { + let (tx, rx) = oneshot::channel(); + forge + .send(GetSchema { + name: name.to_string(), + reply: ReplyChannel::new(tx), + }) + .await; + ask_forge(rx).await? + } else { + None + }; + let Some(definition) = definition else { + return Ok(schema_forge_backend::entity::QueryResult::new(vec![], None)); + }; + let query = schema_forge_core::query::Query::new(definition.id).with_filter(Filter::and(vec![ + Filter::eq( + FieldPath::parse("target_schema").unwrap(), + DynamicValue::Text(schema.name.to_string()), + ), + Filter::eq( + FieldPath::parse("active").unwrap(), + DynamicValue::Boolean(true), + ), + ])); + if let Some(backend) = backend { + return backend.query(&query).await.map_err(ForgeError::from); + } + let forge = forge.ok_or_else(|| ForgeError::Internal { + message: "ForgeActor not registered".into(), + })?; + let (tx, rx) = oneshot::channel(); + forge + .send(QueryEntities { + query, + reply: ReplyChannel::new(tx), + }) + .await; + ask_forge(rx).await?.map_err(ForgeError::from) +} + /// Convert a WebhookSubscription entity to a ResolvedSubscription. fn resolve_runtime_subscription( entity: &Entity, @@ -3108,13 +3271,15 @@ pub async fn create_entity( super::creator_membership::prepare(&state, &schema_def, &entity, claims.as_ref()).await?; if let Some(intent) = intent { - let changed_fields: Vec<_> = entity.fields.keys().cloned().collect(); - let receipt = - super::create_intents::commit(&state, intent, entity, claims.as_ref(), membership) - .await?; - if receipt.created { - super::audit::log_forge_event(&state, claims.as_ref(), &headers, "forge.entity.created", acton_service::audit::AuditSeverity::Informational, serde_json::json!({"schema":schema,"intent_id":receipt.id.as_str(),"changed_fields":changed_fields,"entity_id":receipt.entity_id.as_ref().map(|id|id.as_str())})).await; - } + let receipt = super::create_intents::commit( + &state, + intent, + entity, + claims.as_ref(), + membership, + &headers, + ) + .await?; return super::create_intents::result(state, schema, claims, headers, &receipt).await; } @@ -3126,35 +3291,32 @@ pub async fn create_entity( CreateEntity { entity, membership, - reply: ReplyChannel::new(tx), + reply: mutation_reply( + tx, + MutationCompletion { + after_hook: None, + state: state.clone(), + schema: schema_def.clone(), + claims: claims.clone(), + headers, + dispatcher: hook_dispatcher, + hooks: hooks_config, + operation: "create", + before: None, + notification: notification.clone(), + } + .prepare() + .await, + |result: &Result| { + result.as_ref().ok().cloned() + }, + ), }, ¬ification, ) .await?; - let mut created = ask_forge(rx).await?.map_err(ForgeError::from)?; - - // after_change hook — handed off to HookDispatchActor for - // detached dispatch under acton supervision. - if let Some(dispatcher) = hook_dispatcher.clone() { - fire_after_hook( - &state, - AfterHookCtx { - dispatcher, - hooks_config: hooks_config.clone(), - schema: schema_def.clone(), - event: HookEvent::AfterChange, - operation: "create".to_string(), - user_id: claims.as_ref().map(|c| c.sub.clone()), - }, - &created, - None, - ) - .await; - } + let mut created = await_mutation(rx).await?.map_err(ForgeError::from)?; - let changed_fields: Vec<_> = created.fields.keys().cloned().collect(); - - // Filter read-restricted fields from response filter_entity_fields( &policy_store, &mut created, @@ -3163,32 +3325,6 @@ pub async fn create_entity( FieldFilterDirection::Read, ); - // Audit: entity created - super::audit::log_forge_event( - &state, - claims.as_ref(), - &headers, - "forge.entity.created", - acton_service::audit::AuditSeverity::Informational, - serde_json::json!({ - "schema": schema, - "entity_id": created.id.as_str(), - "changed_fields": changed_fields, - "user": claims.as_ref().map(|c| &c.sub), - }), - ) - .await; - - // Webhook: fire notifications - let webhook_event = notification_event(¬ification, || { - crate::webhook::WebhookEvent::from_create( - &schema_def, - &created, - claims.as_ref().map(|c| c.sub.as_str()), - ) - }); - dispatch_webhook(&state, &schema_def, webhook_event, "created").await; - Ok(( StatusCode::CREATED, Json(entity_to_response(&created, &schema_def)), @@ -4074,30 +4210,22 @@ pub async fn update_entity( check_field_constraints(&schema_def, &entity.fields)?; let notification = mutation_notification(claims.as_ref(), None); + let completion = MutationCompletion { + after_hook: None, + state: state.clone(), + schema: schema_def.clone(), + claims: claims.clone(), + headers, + dispatcher: hook_dispatcher, + hooks: hooks_config, + operation: "update", + before: Some(existing.fields.clone()), + notification, + } + .prepare() + .await; let (mut updated, revision) = - persist_entity_update(&forge, entity, expected, ¬ification).await?; - let changed_fields = changed_field_names(&existing.fields, &updated.fields); - - // after_change hook — handed off to HookDispatchActor for - // detached dispatch under acton supervision. - if let Some(dispatcher) = hook_dispatcher.clone() { - fire_after_hook( - &state, - AfterHookCtx { - dispatcher, - hooks_config: hooks_config.clone(), - schema: schema_def.clone(), - event: HookEvent::AfterChange, - operation: "update".to_string(), - user_id: claims.as_ref().map(|c| c.sub.clone()), - }, - &updated, - Some(&existing.fields), - ) - .await; - } - - // Filter read-restricted fields from response + persist_entity_update(&forge, entity, expected, completion).await?; filter_entity_fields( &policy_store, &mut updated, @@ -4106,32 +4234,6 @@ pub async fn update_entity( FieldFilterDirection::Read, ); - // Audit: entity updated - super::audit::log_forge_event( - &state, - claims.as_ref(), - &headers, - "forge.entity.updated", - acton_service::audit::AuditSeverity::Informational, - serde_json::json!({ - "schema": schema, - "entity_id": updated.id.as_str(), - "changed_fields": changed_fields, - "user": claims.as_ref().map(|c| &c.sub), - }), - ) - .await; - - // Webhook: fire notifications - let webhook_event = notification_event(¬ification, || { - crate::webhook::WebhookEvent::from_update( - &schema_def, - &updated, - claims.as_ref().map(|c| c.sub.as_str()), - ) - }); - dispatch_webhook(&state, &schema_def, webhook_event, "updated").await; - Ok(( revision_headers(revision.as_ref())?, Json(entity_to_response(&updated, &schema_def)), @@ -4421,35 +4523,28 @@ pub async fn patch_entity( // A conditional no-op still reaches storage to compare the authorized // baseline atomically, and advances the revision on acceptance. let notification = mutation_notification(claims.as_ref(), None); + let completion = MutationCompletion { + after_hook: None, + state: state.clone(), + schema: schema_def.clone(), + claims: claims.clone(), + headers, + dispatcher: hook_dispatcher, + hooks: hooks_config, + operation: "patch", + before: Some(baseline_fields), + notification, + } + .prepare() + .await; let (mut updated, revision) = if delta.is_empty() && expected.is_none() { + completion.finish(existing.clone(), None).await; (existing, None) } else { let entity = Entity::with_id(entity_id, schema_name, delta); check_field_constraints(&schema_def, &entity.fields)?; - persist_entity_update(&forge, entity, expected, ¬ification).await? + persist_entity_update(&forge, entity, expected, completion).await? }; - - let changed_fields = changed_field_names(&baseline_fields, &updated.fields); - - // after_change hook — dispatched via HookDispatchActor - if let Some(dispatcher) = hook_dispatcher.clone() { - fire_after_hook( - &state, - AfterHookCtx { - dispatcher, - hooks_config: hooks_config.clone(), - schema: schema_def.clone(), - event: HookEvent::AfterChange, - operation: "patch".to_string(), - user_id: claims.as_ref().map(|c| c.sub.clone()), - }, - &updated, - Some(&baseline_fields), - ) - .await; - } - - // Filter read-restricted fields from response filter_entity_fields( &policy_store, &mut updated, @@ -4458,32 +4553,6 @@ pub async fn patch_entity( FieldFilterDirection::Read, ); - // Audit: entity patched - super::audit::log_forge_event( - &state, - claims.as_ref(), - &headers, - "forge.entity.patched", - acton_service::audit::AuditSeverity::Informational, - serde_json::json!({ - "schema": schema, - "entity_id": updated.id.as_str(), - "changed_fields": changed_fields, - "user": claims.as_ref().map(|c| &c.sub), - }), - ) - .await; - - // Webhook: fire notifications - let webhook_event = notification_event(¬ification, || { - crate::webhook::WebhookEvent::from_update( - &schema_def, - &updated, - claims.as_ref().map(|c| c.sub.as_str()), - ) - }); - dispatch_webhook(&state, &schema_def, webhook_event, "updated").await; - Ok(( revision_headers(revision.as_ref())?, Json(entity_to_response(&updated, &schema_def)), @@ -4605,9 +4674,8 @@ pub async fn delete_entity( }; // Hooks see the same snapshot that passed record authorization. let notification = mutation_notification(claims.as_ref(), Some(&existing)); - let pre_delete_snapshot = Some(existing); - if let (Some(ref dispatcher), Some(snapshot)) = (&hook_dispatcher, &pre_delete_snapshot) { - let mut fields = snapshot.fields.clone(); + if let Some(ref dispatcher) = hook_dispatcher { + let mut fields = existing.fields.clone(); apply_before_hook( BeforeHookCtx { dispatcher: dispatcher.as_ref(), @@ -4623,6 +4691,20 @@ pub async fn delete_entity( .await?; } + let completion = MutationCompletion { + after_hook: None, + state: state.clone(), + schema: schema_def, + claims: claims.clone(), + headers, + dispatcher: hook_dispatcher, + hooks: hooks_config, + operation: "delete", + before: None, + notification: notification.clone(), + } + .prepare() + .await; if let Some(expected) = expected { let (tx, rx) = oneshot::channel(); send_mutation( @@ -4631,12 +4713,18 @@ pub async fn delete_entity( schema: schema_name, id: entity_id, expected, - reply: ReplyChannel::new(tx), + reply: mutation_reply( + tx, + completion, + move |result: &Result<(), ConditionalMutationError>| { + result.as_ref().ok().map(|_| existing) + }, + ), }, ¬ification, ) .await?; - ask_forge(rx).await?.map_err(conditional_error)?; + await_mutation(rx).await?.map_err(conditional_error)?; } else { let (tx, rx) = oneshot::channel(); send_mutation( @@ -4644,58 +4732,20 @@ pub async fn delete_entity( DeleteEntity { schema: schema_name, id: entity_id, - reply: ReplyChannel::new(tx), + reply: mutation_reply( + tx, + completion, + move |result: &Result<(), schema_forge_backend::BackendError>| { + result.as_ref().ok().map(|_| existing) + }, + ), }, ¬ification, ) .await?; - ask_forge(rx).await?.map_err(ForgeError::from)?; + await_mutation(rx).await?.map_err(ForgeError::from)?; } - // after_delete hook — handed off to HookDispatchActor for - // detached dispatch under acton supervision. - if let (Some(dispatcher), Some(snapshot)) = (hook_dispatcher.clone(), pre_delete_snapshot) { - fire_after_hook( - &state, - AfterHookCtx { - dispatcher, - hooks_config: hooks_config.clone(), - schema: schema_def.clone(), - event: HookEvent::AfterDelete, - operation: "delete".to_string(), - user_id: claims.as_ref().map(|c| c.sub.clone()), - }, - &snapshot, - None, - ) - .await; - } - - // Audit: entity deleted - super::audit::log_forge_event( - &state, - claims.as_ref(), - &headers, - "forge.entity.deleted", - acton_service::audit::AuditSeverity::Warning, - serde_json::json!({ - "schema": schema, - "entity_id": id, - "user": claims.as_ref().map(|c| &c.sub), - }), - ) - .await; - - // Webhook: fire notifications - let webhook_event = notification_event(¬ification, || { - crate::webhook::WebhookEvent::from_delete( - &schema, - &id, - claims.as_ref().map(|c| c.sub.as_str()), - ) - }); - dispatch_webhook(&state, &schema_def, webhook_event, "deleted").await; - Ok(StatusCode::NO_CONTENT) } @@ -4706,6 +4756,20 @@ mod tests { Cardinality, FieldDefinition, FieldModifier, FieldName, SchemaId, TextConstraints, }; + #[tokio::test] + async fn a_lost_mutation_reply_reports_an_unknown_outcome() { + let (sender, receiver) = oneshot::channel::<()>(); + drop(sender); + assert_eq!( + await_mutation(receiver).await, + Err(ForgeError::MutationOutcomeUnknown) + ); + assert_eq!( + ForgeError::MutationOutcomeUnknown.into_response().status(), + StatusCode::GATEWAY_TIMEOUT + ); + } + #[test] fn revision_condition_requires_exact_single_token_and_matching_baseline() { let revision = EntityRevision::fresh(); diff --git a/crates/schema-forge-acton/tests/hook_dispatch_limits.rs b/crates/schema-forge-acton/tests/hook_dispatch_limits.rs new file mode 100644 index 00000000..2350d1db --- /dev/null +++ b/crates/schema-forge-acton/tests/hook_dispatch_limits.rs @@ -0,0 +1,111 @@ +//! Background admission follows the deployment's hook configuration. +use acton_service::prelude::{ActonApp, ActorExtension, ActorHandleInterface}; +use schema_forge_acton::{ + hooks::{ + DispatchHook, HookBinding, HookDispatcher, HookError, HookInvocation, HookOutcome, + HooksConfig, + }, + HookDispatchActor, +}; +use schema_forge_core::types::HookEvent; +use std::{sync::Arc, time::Duration}; +use tokio::sync::{mpsc, Semaphore}; + +#[derive(Debug)] +struct BlockingDispatcher { + started: mpsc::UnboundedSender<()>, + completed: mpsc::UnboundedSender<()>, + release: Semaphore, +} +#[async_trait::async_trait] +impl HookDispatcher for BlockingDispatcher { + async fn call_before( + &self, + _: &HookBinding, + _: HookInvocation, + ) -> Result { + Ok(HookOutcome::default()) + } + async fn call_after(&self, _: &HookBinding, _: HookInvocation) -> Result<(), HookError> { + self.started.send(()).unwrap(); + self.release.acquire().await.unwrap().forget(); + self.completed.send(()).unwrap(); + Ok(()) + } +} + +async fn check_limit(configured: usize, expected: usize) { + let mut runtime = ActonApp::launch_async().await; + let mut builder = runtime.new_actor::(); + HookDispatchActor::configure(&mut builder); + let actor = builder.handle().clone(); + let (started, mut starts) = mpsc::unbounded_channel(); + let (completed, mut finishes) = mpsc::unbounded_channel(); + let dispatcher = Arc::new(BlockingDispatcher { + started, + completed, + release: Semaphore::new(0), + }); + let config = HooksConfig { + enabled: true, + max_concurrent_async: configured, + bindings: vec![HookBinding { + schema: "Note".into(), + event: HookEvent::AfterChange, + endpoint: "http://unused".into(), + timeout_ms: None, + required: false, + descriptor_path: None, + }], + ..Default::default() + }; + for _ in 0..6 { + actor + .send(DispatchHook { + invocation: HookInvocation { + schema: "Note".into(), + event: HookEvent::AfterChange, + operation: "create".into(), + user_id: None, + entity_id: None, + fields: Default::default(), + changed_fields: vec![], + previous: Default::default(), + }, + dispatcher: Some(dispatcher.clone()), + config: config.clone(), + }) + .await; + } + let actor = builder.start().await; + for _ in 0..expected { + tokio::time::timeout(Duration::from_secs(5), starts.recv()) + .await + .unwrap() + .unwrap(); + } + assert!( + tokio::time::timeout(Duration::from_millis(50), starts.recv()) + .await + .is_err() + ); + dispatcher.release.add_permits(6); + for _ in 0..6 { + tokio::time::timeout(Duration::from_secs(5), finishes.recv()) + .await + .unwrap() + .unwrap(); + } + actor.stop().await.unwrap(); + runtime.shutdown_all().await.unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn configured_async_limit_bounds_calls_and_queued_work_finishes() { + check_limit(2, 2).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn zero_async_limit_still_makes_progress_serially() { + check_limit(0, 1).await; +} diff --git a/crates/schema-forge-acton/tests/mutation_completion.rs b/crates/schema-forge-acton/tests/mutation_completion.rs new file mode 100644 index 00000000..c232e115 --- /dev/null +++ b/crates/schema-forge-acton/tests/mutation_completion.rs @@ -0,0 +1,437 @@ +//! Deliberately gated storage verifies HTTP acknowledgement and actor-owned effects. +//! This fixture models delayed completion, not database transaction semantics. +mod support; +use acton_service::{ + config::Config, middleware::Claims, prelude::ActorHandleInterface, + service_builder::ServiceBuilder, +}; +use axum::{ + body::Body, + http::{Request, StatusCode}, + Extension, Router, +}; +use schema_forge_acton::{ + config::SchemaForgeConfig, + hooks::{HookBinding, HookDispatcher, HookError, HookInvocation, HookOutcome, HooksConfig}, + messages::{InitForge, ReplyChannel}, + routes::forge_routes, + ForgeActor, HookDispatchActor, +}; +#[cfg(feature = "sse")] +use schema_forge_acton::{ + events::{ConfigureEvents, EventsRuntime}, + events_config::EventsConfig, +}; +use schema_forge_backend::{BackendError, Entity, EntityStore, QueryResult, SchemaBackend}; +use schema_forge_core::{ + migration::MigrationStep, + query::{AggregateQuery, AggregateResult, Query}, + types::{DynamicValue, EntityId, HookEvent, SchemaDefinition, SchemaName}, +}; +use std::{ + collections::{BTreeMap, HashMap}, + sync::Arc, + time::Duration, +}; +use tokio::sync::{mpsc, oneshot, Mutex, Semaphore}; +use tower::ServiceExt; + +struct GatedBackend { + reads: support::FixtureBackend, + row: Mutex, + entered: mpsc::UnboundedSender<()>, + release: Semaphore, + subscriptions: SchemaDefinition, + queried: mpsc::UnboundedSender<()>, +} +impl GatedBackend { + async fn gate(&self) { + self.entered.send(()).unwrap(); + self.release.acquire().await.unwrap().forget(); + } +} +impl SchemaBackend for GatedBackend { + async fn apply_migration( + &self, + name: &SchemaName, + steps: &[MigrationStep], + ) -> Result<(), BackendError> { + self.reads.apply_migration(name, steps).await + } + async fn store_schema_metadata(&self, schema: &SchemaDefinition) -> Result<(), BackendError> { + self.reads.store_schema_metadata(schema).await + } + async fn load_schema_metadata( + &self, + name: &SchemaName, + ) -> Result, BackendError> { + if name == &self.subscriptions.name { + return Ok(Some(self.subscriptions.clone())); + } + self.reads.load_schema_metadata(name).await + } + async fn list_schema_metadata(&self) -> Result, BackendError> { + self.reads.list_schema_metadata().await + } +} +impl EntityStore for GatedBackend { + async fn create(&self, entity: &Entity) -> Result { + self.gate().await; + if entity.fields.get("title") == Some(&DynamicValue::Text("fail".into())) { + return Err(BackendError::Internal { + message: "deliberate failed write".into(), + }); + } + *self.row.lock().await = entity.clone(); + Ok(entity.clone()) + } + async fn get(&self, _: &SchemaName, _: &EntityId) -> Result { + Ok(self.row.lock().await.clone()) + } + async fn update(&self, entity: &Entity) -> Result { + self.gate().await; + let mut row = self.row.lock().await; + row.fields.extend(entity.fields.clone()); + Ok(row.clone()) + } + async fn delete(&self, _: &SchemaName, _: &EntityId) -> Result<(), BackendError> { + self.gate().await; + Ok(()) + } + async fn query(&self, query: &Query) -> Result { + if query.schema == self.subscriptions.id { + self.queried.send(()).unwrap(); + return Ok(QueryResult::new(vec![], None)); + } + self.reads.query(query).await + } + async fn count(&self, query: &Query) -> Result { + self.reads.count(query).await + } + async fn aggregate( + &self, + query: &AggregateQuery, + ) -> Result, BackendError> { + self.reads.aggregate(query).await + } +} +#[derive(Debug)] +struct CaptureHooks(mpsc::UnboundedSender); +#[async_trait::async_trait] +impl HookDispatcher for CaptureHooks { + async fn call_before( + &self, + _: &HookBinding, + _: HookInvocation, + ) -> Result { + Ok(HookOutcome::default()) + } + async fn call_after( + &self, + _: &HookBinding, + invocation: HookInvocation, + ) -> Result<(), HookError> { + self.0.send(invocation).unwrap(); + Ok(()) + } +} +struct Fixture { + router: Router, + backend: Arc, + entered: mpsc::UnboundedReceiver<()>, + hooks: mpsc::UnboundedReceiver, + path: String, + queries: mpsc::UnboundedReceiver<()>, + audit: tokio::net::UdpSocket, +} +async fn fixture() -> Fixture { + let schema = schema_forge_dsl::parse( + r#" + @access(read: ["writer"], write: ["writer"], delete: ["writer"]) + @webhook(events: ["created", "updated", "deleted"]) + @hook(after_change) """capture changes""" + @hook(after_delete) """capture deletes""" + schema Note { title: text required } + "#, + ) + .unwrap() + .remove(0); + let row = Entity::new( + schema.name.clone(), + BTreeMap::from([("title".into(), DynamicValue::Text("before".into()))]), + ); + let path = format!("/schemas/Note/entities/{}", row.id); + let (started, entered) = mpsc::unbounded_channel(); + let (queried, queries) = mpsc::unbounded_channel(); + let backend = Arc::new(GatedBackend { + reads: support::FixtureBackend::new(schema.clone(), vec![row.clone()]), + row: Mutex::new(row), + entered: started, + release: Semaphore::new(0), + subscriptions: schema_forge_dsl::parse( + "schema WebhookSubscription { target_schema: text active: boolean }", + ) + .unwrap() + .remove(0), + queried, + }); + let (hook_tx, hooks) = mpsc::unbounded_channel(); + let mut custom = SchemaForgeConfig::default(); + custom.schema_forge.webhooks.enabled = true; + custom.schema_forge.hooks = HooksConfig { + enabled: true, + bindings: [HookEvent::AfterChange, HookEvent::AfterDelete] + .into_iter() + .map(|event| HookBinding { + schema: "Note".into(), + event, + endpoint: "http://unused".into(), + timeout_ms: None, + required: false, + descriptor_path: None, + }) + .collect(), + ..Default::default() + }; + let audit = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let service = ServiceBuilder::new() + .with_config(Config { + custom, + audit: Some(acton_service::audit::config::AuditConfig { + syslog: acton_service::audit::config::SyslogConfig { + transport: "udp".into(), + address: audit.local_addr().unwrap().to_string(), + ..Default::default() + }, + ..Default::default() + }), + ..Config::default() + }) + .with_actor::() + .with_actor::() + .build(); + let actor = service.state().actor::().unwrap(); + let (tx, rx) = oneshot::channel(); + actor + .send(InitForge { + registry: HashMap::from([("Note".into(), schema)]), + backend: backend.clone(), + tenant_config: None, + record_access_policy: None, + hook_dispatcher: Some(Arc::new(CaptureHooks(hook_tx))), + storage_registry: Default::default(), + policy_store: None, + custom_policies_dir: None, + reply: ReplyChannel::new(tx), + }) + .await; + rx.await.unwrap(); + // Exercise the serialized commit handlers as well as the ordinary actor + // handlers when the runtime's SSE feature is enabled. + #[cfg(feature = "sse")] + { + let user = schema_forge_dsl::parse(schema_forge_core::system_schemas::USER_SCHEMA) + .unwrap() + .remove(0); + let store = Arc::new(schema_forge_backend::EntityAuthStore::new( + backend.clone(), + user, + Arc::new(|_| Some(10)), + )); + let (tx, rx) = oneshot::channel(); + actor + .send(ConfigureEvents { + runtime: Arc::new( + EventsRuntime::new( + EventsConfig { + enabled: true, + ..Default::default() + }, + store, + backend.clone(), + ) + .unwrap(), + ), + reply: ReplyChannel::new(tx), + }) + .await; + rx.await.unwrap(); + } + let caller = Claims { + sub: "user:writer".into(), + roles: vec!["writer".into()], + perms: vec![], + exp: 9_999_999_999, + iat: None, + jti: None, + iss: None, + aud: None, + email: None, + username: None, + custom: HashMap::new(), + }; + let router = forge_routes() + .layer(Extension(caller)) + .with_state(service.state().clone()); + Fixture { + router, + backend, + entered, + hooks, + path, + queries, + audit, + } +} +fn request(method: &str, path: &str, title: &str) -> Request { + Request::builder() + .method(method) + .uri(path) + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({"fields":{"title":title}}).to_string(), + )) + .unwrap() +} +async fn next(receiver: &mut mpsc::UnboundedReceiver) -> T { + tokio::time::timeout(Duration::from_secs(10), receiver.recv()) + .await + .unwrap() + .unwrap() +} +async fn committed_audit(socket: &tokio::net::UdpSocket, method: &str) { + let expected = match method { + "POST" => "forge.entity.created", + "PUT" => "forge.entity.updated", + "PATCH" => "forge.entity.patched", + _ => "forge.entity.deleted", + }; + tokio::time::timeout(Duration::from_secs(10), async { + let mut buffer = [0; 8192]; + loop { + let length = socket.recv(&mut buffer).await.unwrap(); + let event = std::str::from_utf8(&buffer[..length]).unwrap(); + if event.contains(expected) { + assert!(event.contains("user:writer")); + assert!(event.contains("Note")); + break; + } + } + }) + .await + .unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn slow_create_and_updates_acknowledge_commits_after_five_seconds() { + for method in ["POST", "PUT", "PATCH", "DELETE"] { + let mut fixture = fixture().await; + let path = if method == "POST" { + "/schemas/Note/entities" + } else { + &fixture.path + }; + let work = tokio::spawn( + fixture + .router + .clone() + .oneshot(request(method, path, "after")), + ); + next(&mut fixture.entered).await; + // Deliberately exceed the former reply deadline after storage starts. + tokio::time::sleep(Duration::from_millis(5200)).await; + fixture.backend.release.add_permits(1); + let response = work.await.unwrap().unwrap(); + assert_eq!( + response.status(), + match method { + "POST" => StatusCode::CREATED, + "DELETE" => StatusCode::NO_CONTENT, + _ => StatusCode::OK, + } + ); + let hook = next(&mut fixture.hooks).await; + next(&mut fixture.queries).await; + committed_audit(&fixture.audit, method).await; + assert_eq!( + hook.operation, + match method { + "POST" => "create", + "PUT" => "update", + "PATCH" => "patch", + _ => "delete", + } + ); + if method != "DELETE" { + assert_eq!( + fixture.backend.row.lock().await.fields["title"], + DynamicValue::Text("after".into()) + ); + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn committed_effects_survive_dropped_http_requests_for_every_mutation() { + for method in ["POST", "PUT", "PATCH", "DELETE"] { + let mut fixture = fixture().await; + let path = if method == "POST" { + "/schemas/Note/entities" + } else { + &fixture.path + }; + let work = tokio::spawn( + fixture + .router + .clone() + .oneshot(request(method, path, "after")), + ); + next(&mut fixture.entered).await; + work.abort(); + assert!(work.await.unwrap_err().is_cancelled()); + fixture.backend.release.add_permits(1); + let hook = next(&mut fixture.hooks).await; + next(&mut fixture.queries).await; + committed_audit(&fixture.audit, method).await; + assert_eq!( + hook.fields["title"], + DynamicValue::Text( + if method == "DELETE" { + "before" + } else { + "after" + } + .into() + ) + ); + if method == "PUT" || method == "PATCH" { + assert_eq!(hook.changed_fields, ["title"]); + assert_eq!(hook.previous["title"], DynamicValue::Text("before".into())); + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_storage_does_not_dispatch_success_effects() { + let mut fixture = fixture().await; + let work = tokio::spawn(fixture.router.oneshot(request( + "POST", + "/schemas/Note/entities", + "fail", + ))); + next(&mut fixture.entered).await; + fixture.backend.release.add_permits(1); + assert_eq!( + work.await.unwrap().unwrap().status(), + StatusCode::INTERNAL_SERVER_ERROR + ); + assert!( + tokio::time::timeout(Duration::from_millis(50), fixture.hooks.recv()) + .await + .is_err() + ); + assert_eq!( + fixture.backend.row.lock().await.fields["title"], + DynamicValue::Text("before".into()) + ); +} diff --git a/crates/schema-forge-acton/tests/tonic_dispatcher_integration.rs b/crates/schema-forge-acton/tests/tonic_dispatcher_integration.rs index e3857652..2ce92478 100644 --- a/crates/schema-forge-acton/tests/tonic_dispatcher_integration.rs +++ b/crates/schema-forge-acton/tests/tonic_dispatcher_integration.rs @@ -55,6 +55,7 @@ enum Behavior { Abort(String), /// Pass-through (no modifications). PassThrough, + Delay(Duration), } #[tonic::async_trait] @@ -85,6 +86,10 @@ impl TranslationHooks for TestService { tags: Vec::new(), }, Behavior::PassThrough => TranslationBeforeChangeResponse::default(), + Behavior::Delay(delay) => { + tokio::time::sleep(*delay).await; + TranslationBeforeChangeResponse::default() + } }; Ok(Response::new(resp)) } @@ -93,6 +98,9 @@ impl TranslationHooks for TestService { &self, request: Request, ) -> Result, Status> { + if let Behavior::Delay(delay) = self.behavior { + tokio::time::sleep(delay).await; + } *self.after_count.lock().await += 1; self.after_requests.lock().await.push(request.into_inner()); Ok(Response::new(TranslationAfterChangeResponse::default())) @@ -453,3 +461,41 @@ async fn unreachable_endpoint_yields_unavailable() { "unexpected error: {err:?}" ); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn configured_default_timeout_and_binding_override_apply_to_both_hook_directions() { + let (addr, _svc, shutdown) = spawn_server(Behavior::Delay(Duration::from_millis(100))).await; + let mut before = binding(addr, HookEvent::BeforeChange); + let mut after = binding(addr, HookEvent::AfterChange); + before.timeout_ms = None; + after.timeout_ms = None; + let cfg = HooksConfig { + enabled: true, + allow_plaintext: true, + default_timeout_ms: 20, + bindings: vec![before.clone(), after.clone()], + ..Default::default() + }; + let dispatcher = TonicHookDispatcher::new(&cfg, TonicDispatcherConfig::default()).unwrap(); + // Tonic's transport deadline and the outer timer race. Either reports + // failure; the old hard-coded 30s default would let both calls succeed. + assert!(dispatcher + .call_before(&before, invocation(HookEvent::BeforeChange)) + .await + .is_err()); + assert!(dispatcher + .call_after(&after, invocation(HookEvent::AfterChange)) + .await + .is_err()); + before.timeout_ms = Some(1000); + after.timeout_ms = Some(1000); + dispatcher + .call_before(&before, invocation(HookEvent::BeforeChange)) + .await + .unwrap(); + dispatcher + .call_after(&after, invocation(HookEvent::AfterChange)) + .await + .unwrap(); + shutdown.send(()).unwrap(); +} diff --git a/crates/schema-forge-cli/Cargo.toml b/crates/schema-forge-cli/Cargo.toml index ce8bac32..ce5cbc10 100644 --- a/crates/schema-forge-cli/Cargo.toml +++ b/crates/schema-forge-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "schema-forge-cli" -version = "0.52.3" +version = "0.52.4" edition = "2021" [[bin]] @@ -13,7 +13,7 @@ schema-forge-dsl = { path = "../schema-forge-dsl" } schema-forge-backend = { version = "0.23.0", path = "../schema-forge-backend", optional = true } schema-forge-surrealdb = { path = "../schema-forge-surrealdb", optional = true } schema-forge-postgres = { path = "../schema-forge-postgres", optional = true } -schema-forge-acton = { version = "0.50.2", path = "../schema-forge-acton", optional = true } +schema-forge-acton = { version = "0.50.3", path = "../schema-forge-acton", optional = true } clap = { version = "4", features = ["derive", "env", "string"] } clap_complete = "4" serde = { version = "1", features = ["derive"] } diff --git a/docs/hooks-reference.md b/docs/hooks-reference.md index c72c9569..1e365a5c 100644 --- a/docs/hooks-reference.md +++ b/docs/hooks-reference.md @@ -40,6 +40,7 @@ example before reading anything else, jump to - 4.1 [Configuration](#41-configuration) - 4.2 [Observing dispatch](#42-observing-dispatch) - 4.3 [Failure modes](#43-failure-modes) + - 4.4 [Mutation completion and client disconnects](#44-mutation-completion-and-client-disconnects) 5. [Evolving Hooks Over Time](#5-evolving-hooks-over-time) - 5.1 [Schema migration diffs](#51-schema-migration-diffs) - 5.2 [`hooks list` and `hooks diff`](#52-hooks-list-and-hooks-diff) @@ -603,7 +604,7 @@ Top-level fields: |---|---|---| | `enabled` | `false` | Global kill-switch. When `false`, all hook annotations are ignored at runtime. Set this to `false` in local dev to run without hook services. | | `default_timeout_ms` | `30000` | Per-call timeout applied to any binding that does not set its own. Bumped from 5s in v0.13 (issue #11) so handlers that walk related rows or call back into SchemaForge have realistic headroom. | -| `max_concurrent_async` | `100` | Upper bound on background after-hook dispatches. | +| `max_concurrent_async` | `100` | Upper bound on active background after-hook dispatches. Queued work waits for admission; zero is treated as one. The actor runtime may impose a tighter limit. Restart to apply changes. | | `allow_plaintext` | `false` | Permit `http://` endpoints. Off by default because a hook payload carries entity fields, the triggering user's subject, and a replayable bearer credential. See [3.5](#35-securing-the-hook-transport). | | `client_identity` | *unset* | Client certificate SchemaForge presents to mutual-TLS hook services, and the anchors it verifies them against: `cert_path`, `key_path`, optional `root_ca_path` and `exclusive_roots`. Resolved at startup, so a bad key fails the boot rather than the first hooked write. | | `bindings` | `[]` | List of per-(schema, event) bindings. | @@ -690,6 +691,18 @@ Fire-and-forget events (`after_change`, `after_delete`) never affect the HTTP response. All failures are logged at `ERROR` and the client sees the operation as successful. +### 4.4 Mutation completion and client disconnects + +Entity mutations wait for the actor's storage result without a reply deadline. +Once submitted, a write can still commit if the HTTP client or a proxy stops +waiting. Its after hook, audit event, and webhook dispatch run from actor +completion, independently of the HTTP receiver. Delivery remains process-local; +a process crash can still lose background work. + +If the actor reply channel closes, the API returns HTTP 504 with error code +`mutation_outcome_unknown`. Reconcile the record or the same create intent before +retrying. This response does not establish whether storage committed the write. + --- ## 5. Evolving Hooks Over Time