diff --git a/crates/ironclaw_wasm_product_adapters/CLAUDE.md b/crates/ironclaw_wasm_product_adapters/CLAUDE.md index c1bd3b10f67..797c43ffa58 100644 --- a/crates/ironclaw_wasm_product_adapters/CLAUDE.md +++ b/crates/ironclaw_wasm_product_adapters/CLAUDE.md @@ -11,6 +11,7 @@ Owns the thin WASM ProductAdapter host-glue boundary for IronClaw Reborn. - Runtime-specific WIT/request shapes may live here, but shared product-surface DTOs stay in `ironclaw_product_adapters`. - ProductAdapter components may use v1-style minimal WASI p2 for wasm32-wasip2 compatibility: clock/random are allowed; env, args, stdio, preopened directories, inherited network, and DNS lookup must stay disabled. Current slice is parse/render-only: the ProductAdapter WIT `http-egress` import fails closed until host-runtime egress wiring is injected in a follow-up. - Keep dependencies minimal. Avoid workflow/runtime crates unless a concrete host-glue call site requires them and architecture tests are updated deliberately. +- Admission/intake backpressure (the `max_in_flight` permit in `runner.rs`/`runner_immediate_ack.rs`) gates fast intake only — auth, parse, stamp, and `submit_inbound`. An admission/intake permit must NOT be held across an unbounded downstream wait (post-ACK delivery poll, final-reply wait, LLM poll). Release it once the work is durably accepted (`ProductInboundAck::is_durable_outcome`) — before invoking the post-ACK observer — and bound the downstream work with its own mechanism (the delivery-side semaphore/single-flight guard/`max_wait`). Conflating admission with work duration lets a handful of slow turns exhaust every intake slot and silently reject new inbound webhooks under load. Tests: diff --git a/crates/ironclaw_wasm_product_adapters/src/runner_immediate_ack.rs b/crates/ironclaw_wasm_product_adapters/src/runner_immediate_ack.rs index 81f590bb58f..ed43e001a78 100644 --- a/crates/ironclaw_wasm_product_adapters/src/runner_immediate_ack.rs +++ b/crates/ironclaw_wasm_product_adapters/src/runner_immediate_ack.rs @@ -96,15 +96,24 @@ impl NativeProductAdapterRunner { } } tasks.spawn(async move { - let _permit = permit; - // The admission permit bounds the whole tracked post-ACK task, including - // observer follow-up. Otherwise quick workflow acks could release permits - // while long-running observers accumulate without backpressure. + // The admission permit gates only fast intake (auth/parse/stamp/ + // submit), bounded by `workflow_timeout`. It must NOT be held across + // the post-ACK observer delivery poll, which can wait for the run's + // final reply for far longer (Slack's observer polls up to ~120s). + // Holding admission across that unbounded wait would let + // `max_in_flight` slow turns exhaust every intake slot and silently + // reject new inbound webhooks under load. The permit is therefore + // released the moment the workflow durably accepts the inbound + // (`is_durable_outcome`), before observer follow-up runs. On error + // or timeout (no durable acceptance) the permit drops at scope end, + // which keeps correct backpressure over the bounded intake window. + // // Timeout drops the in-flight workflow future. That is the intended // cancellation boundary for this generic async trait call: the // runner does not hold a separate task handle or protocol-specific // resource owner to abort. Workflows that open DB/network resources // must make their own futures cancellation-safe at await points. + let mut permit = Some(permit); let workflow_result = tokio::time::timeout(workflow_timeout, workflow.submit_inbound(workflow_envelope)) .await; @@ -116,6 +125,13 @@ impl NativeProductAdapterRunner { "async webhook workflow dispatch requested retry after protocol ack; event was not retried by protocol transport" ); } + // Release admission before the (potentially long) delivery + // poll once the run is durably submitted. A non-durable ack + // (e.g. retryable rejection) keeps the permit until scope + // end so admission still backpressures un-accepted intake. + if ack.is_durable_outcome() { + drop(permit.take()); + } if let Some(observer) = observer { observer.observe_workflow_ack(envelope, ack).await; } @@ -162,7 +178,7 @@ impl NativeProductAdapterRunner { #[cfg(test)] mod tests { use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; use super::ImmediateAckWorkflowObserver; @@ -314,6 +330,38 @@ mod tests { } } + /// Workflow that returns a *non-durable* retryable rejection via the + /// `Ok(ack)` arm (distinct from `RejectingWorkflow`, which returns `Err`). + /// `ProductInboundAck::Rejected(Retryable)` reports `is_durable_outcome() == + /// false`, so the admission permit must be retained across the observer. + struct NonDurableRejectWorkflow; + + #[async_trait] + impl ironclaw_product_adapters::ProductWorkflow for NonDurableRejectWorkflow { + async fn submit_inbound( + &self, + _envelope: ProductInboundEnvelope, + ) -> Result { + Ok(ProductInboundAck::Rejected( + ironclaw_product_adapters::ProductRejection::retryable( + ironclaw_product_adapters::ProductRejectionKind::PolicyDenied, + "policy temporarily unavailable", + ), + )) + } + + async fn resolve_projection_subscription( + &self, + _envelope: ProductInboundEnvelope, + ) -> Result { + Err(ProductAdapterError::Internal { + detail: ironclaw_product_adapters::redaction::RedactedString::new( + "test stub: resolve_projection_subscription not supported", + ), + }) + } + } + struct RecordingObserver { ack_count: Arc, error_count: Arc, @@ -338,6 +386,34 @@ mod tests { } } + /// Observer that blocks inside `observe_workflow_ack` until released. Models + /// the real `SlackFinalReplyDeliveryObserver`, which polls the run for its + /// final reply for up to `max_wait` (default 120s) inside this callback. + /// + /// Release uses a latched `AtomicBool` polled with `yield_now`, not a + /// `Notify`: a single shared observer is invoked by two tasks, and the + /// second invocation can register *after* the release fires, which + /// `Notify::notify_waiters` would not wake. The latch wakes current and + /// future waiters deterministically. + struct BlockingObserver { + entered: Arc, + released: Arc, + } + + #[async_trait] + impl ImmediateAckWorkflowObserver for BlockingObserver { + async fn observe_workflow_ack( + &self, + _envelope: ProductInboundEnvelope, + _ack: ProductInboundAck, + ) { + self.entered.fetch_add(1, Ordering::SeqCst); + while !self.released.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + } + } + struct BlockingWorkflow { entered: Arc, release: Arc, @@ -398,6 +474,20 @@ mod tests { .expect("webhook auth should verify") } + async fn wait_for_observer_entries( + entered: &AtomicUsize, + expected: usize, + failure_context: &str, + ) { + tokio::time::timeout(Duration::from_secs(1), async { + while entered.load(Ordering::SeqCst) < expected { + tokio::task::yield_now().await; + } + }) + .await + .unwrap_or_else(|_| panic!("timed out waiting for {failure_context}")); + } + #[tokio::test] async fn process_verified_webhook_immediate_ack_dispatches_workflow() { let parse_count = Arc::new(AtomicUsize::new(0)); @@ -510,6 +600,118 @@ mod tests { assert_eq!(parse_count.load(Ordering::SeqCst), 2); } + /// Regression: the admission permit must be released once the workflow has + /// durably accepted the inbound (its `submit_inbound` returned), NOT held + /// across the post-ACK observer delivery poll. The real Slack observer polls + /// the submitted run for its final reply for up to 120s inside + /// `observe_workflow_ack`; if admission stayed pinned for that whole window, + /// only `SLACK_MAX_IN_FLIGHT_WEBHOOKS` slow turns would exhaust all admission + /// slots and new inbound webhooks would be rejected `TooManyInFlight` for up + /// to two minutes — silent backpressure-induced message loss under load. + /// + /// With `max_in_flight = 1`: webhook #1 durably accepts, then its observer + /// blocks (modelling the long delivery poll). Webhook #2 must still be + /// admitted because admission is freed after durable acceptance. + #[tokio::test] + async fn admission_released_after_durable_accept_not_held_across_delivery() { + let parse_count = Arc::new(AtomicUsize::new(0)); + let observer_entered = Arc::new(AtomicUsize::new(0)); + let observer_released = Arc::new(AtomicBool::new(false)); + // AckWorkflow returns a durable NoOp ack immediately; the observer then + // blocks, standing in for the up-to-120s final-reply delivery poll. + let runner = runner_with_workflow(Arc::clone(&parse_count), Arc::new(AckWorkflow), 1); + let observer = Arc::new(BlockingObserver { + entered: Arc::clone(&observer_entered), + released: Arc::clone(&observer_released), + }); + let evidence = verified_evidence(&runner); + + let first = runner + .process_verified_webhook_immediate_ack_with_observer( + b"{}", + &evidence, + Some(Arc::clone(&observer) as Arc), + ) + .await + .expect("first webhook should be accepted for async dispatch"); + assert_eq!(first, WebhookProcessOutcome::AcceptedForAsyncDispatch); + + // Wait until the observer is actively blocking — i.e. the workflow has + // durably accepted and we are now in the delivery phase. + wait_for_observer_entries(&observer_entered, 1, "durable observer to enter").await; + + // The second webhook must be admitted: admission only gates intake, not + // the unbounded delivery wait the observer is currently sitting in. + let second = runner + .process_verified_webhook_immediate_ack_with_observer( + b"{}", + &evidence, + Some(Arc::clone(&observer) as Arc), + ) + .await + .expect("second webhook must be admitted while first is still in delivery"); + assert_eq!(second, WebhookProcessOutcome::AcceptedForAsyncDispatch); + + observer_released.store(true, Ordering::SeqCst); + runner.drain_immediate_ack_tasks().await; + assert_eq!(parse_count.load(Ordering::SeqCst), 2); + } + + /// Counterpart to the durable-accept test: a *non-durable* ack + /// (`Rejected(Retryable)`) must KEEP the admission permit across the observer + /// so admission still backpressures un-accepted intake. This pins the + /// `is_durable_outcome()` guard so a future edit that drops the permit + /// unconditionally would be caught. + #[tokio::test] + async fn admission_retained_across_observer_for_non_durable_ack() { + let parse_count = Arc::new(AtomicUsize::new(0)); + let observer_entered = Arc::new(AtomicUsize::new(0)); + let observer_released = Arc::new(AtomicBool::new(false)); + let runner = runner_with_workflow( + Arc::clone(&parse_count), + Arc::new(NonDurableRejectWorkflow), + 1, + ); + let observer = Arc::new(BlockingObserver { + entered: Arc::clone(&observer_entered), + released: Arc::clone(&observer_released), + }); + let evidence = verified_evidence(&runner); + + let first = runner + .process_verified_webhook_immediate_ack_with_observer( + b"{}", + &evidence, + Some(Arc::clone(&observer) as Arc), + ) + .await + .expect("first webhook should be accepted for async dispatch"); + assert_eq!(first, WebhookProcessOutcome::AcceptedForAsyncDispatch); + + // Wait until the observer is blocking — the workflow returned a + // non-durable ack, so the permit must still be held at this point. + wait_for_observer_entries(&observer_entered, 1, "non-durable observer to enter").await; + + // The permit is retained across the observer for a non-durable ack, so a + // second intake must be rejected while the first is still in delivery. + let err = runner + .process_verified_webhook_immediate_ack_with_observer( + b"{}", + &evidence, + Some(Arc::clone(&observer) as Arc), + ) + .await + .expect_err("second webhook should be rejected while non-durable permit is held"); + assert!(matches!( + err, + RunnerError::TooManyInFlight { max_in_flight: 1 } + )); + + observer_released.store(true, Ordering::SeqCst); + runner.drain_immediate_ack_tasks().await; + assert_eq!(parse_count.load(Ordering::SeqCst), 1); + } + #[tokio::test] async fn process_verified_webhook_immediate_ack_rejects_mismatched_evidence_type() { let parse_count = Arc::new(AtomicUsize::new(0)); diff --git a/docs/plans/2026-06-25-slack-admission-permit.md b/docs/plans/2026-06-25-slack-admission-permit.md new file mode 100644 index 00000000000..4e7d4e0dad1 --- /dev/null +++ b/docs/plans/2026-06-25-slack-admission-permit.md @@ -0,0 +1,101 @@ +# Slack admission permit held across delivery poll — fix plan + +Date: 2026-06-25 +Branch: `fix/reborn-slack-admission-permit` +Owner crate/file: `crates/ironclaw_wasm_product_adapters/src/runner_immediate_ack.rs` + +## Verified bug (file:line evidence) + +The immediate-ACK webhook path acquires an **admission** permit and holds it +across the entire post-ACK task — including the unbounded post-ACK delivery +observer — instead of releasing it once the inbound is durably accepted. + +- Admission permit acquired in `prepare_inbound_envelope` + (`runner.rs:316`, `try_acquire_owned` on `admission: Arc`, + capacity = `max_in_flight`). +- In `process_verified_webhook_immediate_ack_with_observer` + (`runner_immediate_ack.rs`), the permit is moved into the spawned task + (`let _permit = permit;`) and dropped only when the task ends. +- The task awaits `tokio::time::timeout(workflow_timeout, submit_inbound)` + (~2s for Slack via `SLACK_WEBHOOK_WORKFLOW_TIMEOUT`, `slack_host_beta.rs:83`) + and then, on `Ok(Ok(ack))`, awaits `observer.observe_workflow_ack(...)` + **outside** the timeout. +- For Slack the observer is `SlackFinalReplyDeliveryObserver` + (`slack_delivery.rs:1230`) whose `observe_workflow_ack` calls + `deliver_final_reply` (`slack_delivery.rs:1392`), which polls the submitted + run for its final reply for up to `max_wait` — default **120s** + (`slack_delivery.rs:109`). +- Net effect: the admission permit is pinned for up to ~120s per turn even + though `submit_inbound` already returned a durable `Accepted`/`NoOp` ack in + ~2s. With `SLACK_MAX_IN_FLIGHT_WEBHOOKS = 64` (`slack_host_beta.rs:84`), + 64 slow turns exhaust all admission slots; further inbound webhooks are + rejected `TooManyInFlight` (`runner.rs:316-320`) → HTTP 429 + (`slack_serve.rs:337-340`). Slack retries a bounded number of times; under + sustained load the 429s persist for the whole delivery window and Slack + eventually stops retrying → user messages are silently lost. + +The anti-pattern: an **admission/intake** slot conflated with **work duration**, +held across an unbounded downstream wait (the delivery poll). + +`ProductInboundAck::is_durable_outcome()` (`inbound.rs:852`) already exists and +is the right signal for "the run is durably accepted; admission can be freed." + +## Red proof + +New test `admission_released_after_durable_accept_not_held_across_delivery` +(`runner_immediate_ack.rs`): `max_in_flight = 1`, workflow returns a durable ack +immediately, observer blocks (models the 120s poll). On the unpatched base the +second webhook is rejected `TooManyInFlight { max_in_flight: 1 }` → test fails +RED, exactly reproducing the bug. After the fix the second webhook is admitted → +GREEN. + +## Fix (minimal, smallest correct change) + +Decouple admission from delivery **inside the spawned task** in +`runner_immediate_ack.rs`: + +1. Keep the admission permit owned by the task. +2. Run `submit_inbound` under `workflow_timeout` as today. +3. On a **durable** outcome (`ack.is_durable_outcome()`), explicitly + `drop(permit)` to release the admission slot *before* invoking + `observer.observe_workflow_ack(...)`. The run is already durably submitted, so + the long delivery poll no longer consumes an intake slot. +4. On workflow error or timeout (no durable acceptance), the permit drops at the + end of the task as today — error/timeout paths are short and bounded by + `workflow_timeout`, so holding admission across them is correct backpressure. + The observer's `observe_workflow_error` is best-effort and short. + +Admission now gates only fast intake (auth/parse/stamp/submit, bounded by +`workflow_timeout`). The unbounded delivery wait is bounded by its own +mechanism — the delivery-side machinery in `slack_delivery.rs` (the shared +delivery semaphore, single-flight per-run guard, and `max_wait`) — which is +owned by another agent and is **not** modified here. + +### Why this is sufficient (not over-engineered) + +- The run is durable once `submit_inbound` returns `Accepted`/`NoOp` etc.; the + reply is produced by the turn runtime independently of admission. Losing the + delivery poll does not lose the user's message — only the inline push of the + reply, which the delivery side already bounds and guards. +- No new durable-retry subsystem is needed: the existing delivery bounds plus + early admission release remove the silent-drop-under-load failure mode. +- Change is confined to one function in the file lane we own. + +### Cancellation / RAII safety + +`OwnedSemaphorePermit` releases on drop on every path (durable → explicit early +drop; error/timeout/panic → scope-end drop). No `unwrap`/`expect` added. + +## Guardrail + +Add a note to the crate guidance (module doc / nearest `CLAUDE.md`/`AGENTS.md`): +an admission/intake permit must NOT be held across an unbounded downstream wait +(delivery / LLM poll); release it once work is durably accepted and bound the +work with its own mechanism. + +## Test / quality gate + +- `cargo fmt --all` +- `cargo clippy -p ironclaw_wasm_product_adapters --all-targets` (zero warnings) +- `cargo test -p ironclaw_wasm_product_adapters` (red→green + existing suite) +- Build the dependent composition crate to ensure no contract drift.