From a09c346e832c5ca7f7f5bf25a29161af1b12c12e Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Wed, 15 Jul 2026 20:13:00 +0200 Subject: [PATCH] test(dispatcher): pin trigger_depth ceiling in the async outbox path An outbox row whose trigger_depth exceeds max_trigger_depth is dropped without executing the handler. Mirrors the disabled-http drop harness; mutation-verified (weakening the depth gate fails it). Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/manager-core/src/dispatcher.rs | 71 +++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 8f37912..07af521 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -2785,5 +2785,76 @@ mod tests { "executor ran for a disabled HTTP outbox row; fire-time gate missing" ); } + + /// `trigger_depth` bounds chain DEPTH. The sync `invoke()` path is pinned + /// (sdk/invoke.rs), but the ASYNC dispatcher enforcement — `dispatch_one` + /// dropping a row past `max_trigger_depth` — was not. Without it, a trigger + /// whose handler writes back to the collection it watches is an infinite + /// outbox self-amplification loop (write→trigger→write→…) from one request. + #[tokio::test] + async fn outbox_row_past_max_depth_is_dropped_without_executing() { + let app_id = AppId::new(); + // ENABLED script — so what stops execution is the DEPTH gate, not the + // disabled-drop path. (The depth check runs before script resolution, + // so the repo is never even consulted, but keep it honest.) + let mut script = disabled_script(app_id); + script.enabled = true; + + let config = TriggerConfig::from_env(); + let row_id = Uuid::new_v4(); + let row = OutboxRow { + id: row_id, + app_id, + source_kind: OutboxSourceKind::Kv, + trigger_id: Some(TriggerId::new()), + script_id: Some(script.id), + reply_to: None, + payload: serde_json::json!({}), + origin_principal: None, + // One past the ceiling — must be dropped, not dispatched. + trigger_depth: config.max_trigger_depth + 1, + root_execution_id: None, + attempt_count: 0, + next_attempt_at: Utc::now(), + created_at: Utc::now(), + }; + + let deleted = Arc::new(Mutex::new(None)); + let executed = Arc::new(AtomicBool::new(false)); + let dispatcher = Dispatcher { + outbox: Arc::new(RecordingOutbox { + deleted: deleted.clone(), + }), + triggers: Arc::new(UnusedTriggers), + scripts: Arc::new(DisabledScriptRepo { script }), + dead_letters: Arc::new(UnusedDeadLetters), + abandoned: Arc::new(UnusedAbandoned), + principals: Arc::new(OkPrincipals), + executor: Arc::new(RecordingExecutor { + executed: executed.clone(), + }), + gate: Arc::new(ExecutionGate::new(1)), + log_sink: Arc::new(UnusedLogSink), + inbox: Arc::new(UnusedInbox), + queue: Arc::new(UnusedQueue), + group_queue: Arc::new(NoopGroupQueue), + config, + instance_id: "test-instance".into(), + }; + + let result = dispatcher.dispatch_one(row).await; + + assert!(result.is_ok(), "dispatch_one returned {result:?}"); + assert_eq!( + *deleted.lock().unwrap(), + Some(row_id), + "a depth-exceeded row must be deleted (not left to re-amplify)" + ); + assert!( + !executed.load(Ordering::SeqCst), + "executor ran for a depth-exceeded row — the depth gate is missing, \ + so a self-writing trigger would loop unbounded" + ); + } } }