fix(runtime): preserve committed mutation effects and honor hook configuration

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

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]

crates/schema-forge-acton/src/actor.rs

@@ -451,7 +451,7 @@ fn configure_backend_operations(actor: &mut ManagedActor<Idle, ForgeActor>) {
         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<Idle, ForgeActor>) {
                 },
                 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<Idle, ForgeActor>) {
         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<Idle, ForgeActor>) {
                     Err(no_backend_error())
                 }
             };
-            reply.send(result).await;
+            reply.send_with_backend(result, backend).await;
         })
     });
 
@@ -523,23 +523,23 @@ fn configure_backend_operations(actor: &mut ManagedActor<Idle, ForgeActor>) {
         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(
                         no_backend_error(),
                     ),
                 ),
             };
-            request.reply.send(result).await;
+            request.reply.send_with_backend(result, backend).await;
         })
     });
 
     actor.act_on::<crate::messages::DeleteEntityIf>(|actor, ctx| {
         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<Idle, ForgeActor>) {
                     ),
                 ),
             };
-            request.reply.send(result).await;
+            request.reply.send_with_backend(result, backend).await;
         })
     });
 
@@ -581,14 +581,14 @@ fn configure_backend_operations(actor: &mut ManagedActor<Idle, ForgeActor>) {
         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<Idle, ForgeActor>) {
         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;
         })
     });
 

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",
         }
     }

crates/schema-forge-acton/src/events.rs

@@ -260,7 +260,7 @@ pub(crate) fn configure_actor(actor: &mut ManagedActor<Idle, ForgeActor>) {
                                         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<Idle, ForgeActor>) {
                                 }
                             }
                         }
-                        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");
                     }

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<Idle, Self>) {
-        actor.act_on::<DispatchHook>(|_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::<Arc<Semaphore>>::new());
+        let deferred_limit = limit.clone();
+        actor.act_on::<DispatchHook>(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::<DeferredHook>(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<Semaphore>) {
+    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<Mutex<Option<oneshot::Receiver<DispatchHook>>>>,
+}
+impl DeferredHook {
+    pub(crate) fn new(config: HooksConfig, receiver: oneshot::Receiver<DispatchHook>) -> Self {
+        Self {
+            config,
+            receiver: Arc::new(Mutex::new(Some(receiver))),
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // DispatchHook message
 // ---------------------------------------------------------------------------

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.

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(())
     }