diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 053b3b5..96b0c52 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -384,6 +384,39 @@ impl Dispatcher { return Ok(()); } + // Fire-time `enabled` re-check (§4.3). The queue arm does not flow + // through `dispatch_one`'s unified `!active` gate; its only other + // `enabled` guard is the per-tick `list_active_queue_consumers` + // snapshot, which is stale for messages already claimed in this + // tick. A script disabled after the list query but before this + // claimed message executes must NOT run (`script` here is a fresh + // read from line ~331, so `enabled` is current). Release the claim + // (nack) rather than ack/dead-letter: the message stays queued and + // is processed when the script is re-enabled, and the next tick + // won't re-claim it because the list filters `s.enabled`. Without + // this nack the message would sit claimed indefinitely — + // `reclaim_visibility_timeouts` only reclaims for enabled triggers. + if !script.enabled { + tracing::info!( + script_id = %consumer.script_id, + trigger_id = %consumer.trigger_id, + "queue consumer script disabled at fire time; releasing claim" + ); + if let Err(e) = self + .queue + .nack( + claimed.id, + claimed.claim_token, + chrono::Duration::seconds(1), + ) + .await + { + tracing::warn!(?e, "queue nack on disabled consumer failed"); + } + drop(permit); + return Ok(()); + } + let principal = self .principals .resolve(consumer.registered_by_principal) @@ -743,6 +776,22 @@ impl Dispatcher { DispatcherError::ResolveTrigger(format!("script {} not found", trigger.script_id)) })?; + // Audit 2026-06-11 H-F1 sibling — same-app guard mirroring + // build_http_request / build_invoke_request / dispatch_one_queue. + // `build_exec_request` stamps `ExecRequest.app_id = row.app_id` + // while sourcing the body from `trigger.script_id`; without this + // check a hand-edited outbox/trigger row (or a partial restore, or + // a script re-pointed across apps) could run one app's script under + // another app's `SdkCallCx.app_id` — the cross-app isolation + // boundary. Not reachable via the trigger-create or `apply` paths + // (both resolve the script within the app's own scope), so this is + // the runtime backstop the other arms already carry. + if script.app_id != row.app_id { + return Err(DispatcherError::ResolveTrigger( + "trigger outbox target belongs to a different app".into(), + )); + } + Ok(ResolvedTrigger { trigger_kind: trigger.kind, is_dead_letter_handler: matches!(trigger.kind, TriggerKind::DeadLetter), @@ -1535,4 +1584,837 @@ mod tests { ); assert_eq!(failure_kind_to_status(InboxFailureKind::Platform), 500); } + + // ---------------------------------------------------------------- + // Queue-arm fire-time `enabled` gate (§4.3). + // + // `dispatch_one_queue` re-reads the script fresh after claiming a + // message and, if it has been disabled since the per-tick + // `list_active_queue_consumers` snapshot, releases the claim (nack) + // without resolving a principal or executing. This test proves that + // gate end-to-end with in-memory stubs. + // + // Regression property: the principal resolver returns a valid + // `Principal` and the executor records `executed = true` before + // erroring, so DELETING the `if !script.enabled` gate makes the flow + // fall through to resolve → execute, flipping `executed` and failing + // `assert!(!executed)`. (Verified by temporarily removing the gate.) + // ---------------------------------------------------------------- + mod queue_enabled_gate { + use super::*; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + use async_trait::async_trait; + use chrono::Utc; + use picloud_executor_core::{ExecError, ExecRequest, ExecResponse}; + use picloud_orchestrator_core::{ExecutionGate, ExecutorClient}; + use picloud_shared::{ + AdminUserId, AppId, InstanceRole, Principal, Script, ScriptId, ScriptSandbox, + }; + use uuid::Uuid; + + use crate::abandoned_repo::{AbandonedRepo, NewAbandonedExecution}; + use crate::dead_letter_repo::{DeadLetterRepo, NewDeadLetter}; + use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow}; + use crate::principal_resolver::{PrincipalResolver, PrincipalResolverError}; + use crate::queue_repo::{ClaimedMessage, NewQueueMessage, QueueRepo, QueueStats}; + use crate::repo::{NewScript, ScriptPatch, ScriptRepository, ScriptRepositoryError}; + use crate::trigger_config::BackoffShape; + use crate::trigger_repo::{ActiveQueueConsumer, TriggerRepo}; + + // ---- ScriptRepository: only `get` is exercised. ---- + pub(super) struct DisabledScriptRepo { + pub(super) script: Script, + } + + #[async_trait] + impl ScriptRepository for DisabledScriptRepo { + async fn get(&self, _id: ScriptId) -> Result, ScriptRepositoryError> { + Ok(Some(self.script.clone())) + } + async fn get_by_name( + &self, + _app_id: AppId, + _name: &str, + ) -> Result, ScriptRepositoryError> { + unimplemented!("not used by this test") + } + async fn list(&self) -> Result, ScriptRepositoryError> { + unimplemented!("not used by this test") + } + async fn list_for_app( + &self, + _app_id: AppId, + ) -> Result, ScriptRepositoryError> { + unimplemented!("not used by this test") + } + async fn list_for_user( + &self, + _user_id: AdminUserId, + ) -> Result, ScriptRepositoryError> { + unimplemented!("not used by this test") + } + async fn create(&self, _input: NewScript) -> Result { + unimplemented!("not used by this test") + } + async fn update( + &self, + _id: ScriptId, + _patch: ScriptPatch, + ) -> Result { + unimplemented!("not used by this test") + } + async fn delete(&self, _id: ScriptId) -> Result<(), ScriptRepositoryError> { + unimplemented!("not used by this test") + } + async fn count_routes_for_script( + &self, + _script_id: ScriptId, + ) -> Result { + unimplemented!("not used by this test") + } + async fn count_triggers_for_script( + &self, + _script_id: ScriptId, + ) -> Result { + unimplemented!("not used by this test") + } + async fn list_imports( + &self, + _script_id: ScriptId, + ) -> Result, ScriptRepositoryError> { + unimplemented!("not used by this test") + } + } + + // ---- QueueRepo: `claim` returns the message, `nack` records. ---- + struct ClaimNackQueue { + claimed: ClaimedMessage, + nacked: Arc, + } + + #[async_trait] + impl QueueRepo for ClaimNackQueue { + async fn enqueue( + &self, + _msg: NewQueueMessage, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn claim( + &self, + _app_id: AppId, + _queue_name: &str, + ) -> Result, crate::queue_repo::QueueRepoError> { + Ok(Some(self.claimed.clone())) + } + async fn ack( + &self, + _message_id: picloud_shared::QueueMessageId, + _claim_token: Uuid, + ) -> Result { + unimplemented!("not used by this test") + } + async fn nack( + &self, + _message_id: picloud_shared::QueueMessageId, + _claim_token: Uuid, + _retry_delay: chrono::Duration, + ) -> Result { + self.nacked.store(true, Ordering::SeqCst); + Ok(true) + } + async fn reclaim_visibility_timeouts( + &self, + ) -> Result { + unimplemented!("not used by this test") + } + async fn depth( + &self, + _app_id: AppId, + _queue_name: &str, + ) -> Result { + unimplemented!("not used by this test") + } + async fn depth_pending( + &self, + _app_id: AppId, + _queue_name: &str, + ) -> Result { + unimplemented!("not used by this test") + } + async fn list_for_app( + &self, + _app_id: AppId, + ) -> Result, crate::queue_repo::QueueRepoError> { + unimplemented!("not used by this test") + } + #[allow(clippy::too_many_arguments)] + async fn dead_letter( + &self, + _message_id: picloud_shared::QueueMessageId, + _claim_token: Uuid, + _app_id: AppId, + _queue_name: &str, + _trigger_id: Option, + _script_id: Option, + _attempt: u32, + _first_attempt_at: chrono::DateTime, + _last_error: &str, + ) -> Result + { + unimplemented!("not used by this test") + } + } + + // ---- PrincipalResolver: returns a valid Principal (regression). ---- + pub(super) struct OkPrincipals; + + #[async_trait] + impl PrincipalResolver for OkPrincipals { + async fn resolve( + &self, + user_id: AdminUserId, + ) -> Result { + Ok(Principal { + user_id, + instance_role: InstanceRole::Owner, + scopes: None, + app_binding: None, + }) + } + } + + // ---- ExecutorClient: records execution then errors (regression). ---- + pub(super) struct RecordingExecutor { + pub(super) executed: Arc, + } + + #[async_trait] + impl ExecutorClient for RecordingExecutor { + async fn execute( + &self, + _source: &str, + _req: ExecRequest, + _timeout: std::time::Duration, + ) -> Result { + self.executed.store(true, Ordering::SeqCst); + Err(ExecError::Runtime("stub".into())) + } + // Intentionally NOT overriding `execute_with_identity`: the + // default impl forwards to `execute`, which is what gate + // removal would reach. + } + + // ---- Remaining deps: never exercised by the disabled path. ---- + pub(super) struct UnusedTriggers; + + #[async_trait] + impl TriggerRepo for UnusedTriggers { + async fn create_kv_trigger( + &self, + _app_id: AppId, + _req: crate::trigger_repo::CreateKvTrigger, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn create_docs_trigger( + &self, + _app_id: AppId, + _req: crate::trigger_repo::CreateDocsTrigger, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn create_dead_letter_trigger( + &self, + _app_id: AppId, + _req: crate::trigger_repo::CreateDeadLetterTrigger, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn create_cron_trigger( + &self, + _app_id: AppId, + _req: crate::trigger_repo::CreateCronTrigger, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn create_files_trigger( + &self, + _app_id: AppId, + _req: crate::trigger_repo::CreateFilesTrigger, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn create_pubsub_trigger( + &self, + _app_id: AppId, + _req: crate::trigger_repo::CreatePubsubTrigger, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn create_email_trigger( + &self, + _app_id: AppId, + _req: crate::trigger_repo::CreateEmailTrigger, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn email_inbound_target( + &self, + _trigger_id: picloud_shared::TriggerId, + ) -> Result< + Option, + crate::trigger_repo::TriggerRepoError, + > { + unimplemented!("not used by this test") + } + async fn list_for_app( + &self, + _app_id: AppId, + ) -> Result, crate::trigger_repo::TriggerRepoError> + { + unimplemented!("not used by this test") + } + async fn get( + &self, + _id: picloud_shared::TriggerId, + ) -> Result, crate::trigger_repo::TriggerRepoError> + { + unimplemented!("not used by this test") + } + async fn delete( + &self, + _id: picloud_shared::TriggerId, + ) -> Result { + unimplemented!("not used by this test") + } + async fn list_matching_kv( + &self, + _app_id: AppId, + _collection: &str, + _op: picloud_shared::KvEventOp, + ) -> Result< + Vec, + crate::trigger_repo::TriggerRepoError, + > { + unimplemented!("not used by this test") + } + async fn list_matching_docs( + &self, + _app_id: AppId, + _collection: &str, + _op: picloud_shared::DocsEventOp, + ) -> Result< + Vec, + crate::trigger_repo::TriggerRepoError, + > { + unimplemented!("not used by this test") + } + async fn list_matching_files( + &self, + _app_id: AppId, + _collection: &str, + _op: picloud_shared::FilesEventOp, + ) -> Result< + Vec, + crate::trigger_repo::TriggerRepoError, + > { + unimplemented!("not used by this test") + } + async fn list_matching_dead_letter( + &self, + _app_id: AppId, + _source: &str, + _trigger_id: Option, + _script_id: Option, + ) -> Result< + Vec, + crate::trigger_repo::TriggerRepoError, + > { + unimplemented!("not used by this test") + } + async fn create_queue_trigger( + &self, + _app_id: AppId, + _req: crate::trigger_repo::CreateQueueTrigger, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn list_active_queue_consumers( + &self, + ) -> Result, crate::trigger_repo::TriggerRepoError> + { + unimplemented!("not used by this test") + } + async fn touch_queue_trigger_last_fired_at( + &self, + _trigger_id: picloud_shared::TriggerId, + _at: chrono::DateTime, + ) -> Result<(), crate::trigger_repo::TriggerRepoError> { + unimplemented!("not used by this test") + } + } + + pub(super) struct UnusedDeadLetters; + + #[async_trait] + impl DeadLetterRepo for UnusedDeadLetters { + async fn insert( + &self, + _row: NewDeadLetter, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn get( + &self, + _id: picloud_shared::DeadLetterId, + ) -> Result< + Option, + crate::dead_letter_repo::DeadLetterRepoError, + > { + unimplemented!("not used by this test") + } + async fn list_for_app( + &self, + _app_id: AppId, + _unresolved_only: bool, + _limit: i64, + _offset: i64, + ) -> Result< + Vec, + crate::dead_letter_repo::DeadLetterRepoError, + > { + unimplemented!("not used by this test") + } + async fn unresolved_count( + &self, + _app_id: AppId, + ) -> Result { + unimplemented!("not used by this test") + } + async fn resolve( + &self, + _id: picloud_shared::DeadLetterId, + _reason: &str, + ) -> Result<(), crate::dead_letter_repo::DeadLetterRepoError> { + unimplemented!("not used by this test") + } + async fn gc( + &self, + _older_than: chrono::DateTime, + _limit: i64, + ) -> Result { + unimplemented!("not used by this test") + } + } + + struct UnusedOutbox; + + #[async_trait] + impl OutboxRepo for UnusedOutbox { + async fn insert( + &self, + _row: NewOutboxRow, + ) -> Result { + unimplemented!("not used by this test") + } + async fn claim_due( + &self, + _claimed_by: &str, + _limit: i64, + ) -> Result, crate::outbox_repo::OutboxRepoError> { + unimplemented!("not used by this test") + } + async fn delete(&self, _id: Uuid) -> Result<(), crate::outbox_repo::OutboxRepoError> { + unimplemented!("not used by this test") + } + async fn reschedule( + &self, + _id: Uuid, + _attempt_count: u32, + _next_attempt_at: chrono::DateTime, + ) -> Result<(), crate::outbox_repo::OutboxRepoError> { + unimplemented!("not used by this test") + } + } + + pub(super) struct UnusedAbandoned; + + #[async_trait] + impl AbandonedRepo for UnusedAbandoned { + async fn insert( + &self, + _row: NewAbandonedExecution, + ) -> Result { + unimplemented!("not used by this test") + } + async fn gc( + &self, + _older_than: chrono::DateTime, + _limit: i64, + ) -> Result { + unimplemented!("not used by this test") + } + } + + pub(super) struct UnusedLogSink; + + #[async_trait] + impl ExecutionLogSink for UnusedLogSink { + async fn record( + &self, + _log: picloud_shared::ExecutionLog, + ) -> Result<(), picloud_shared::LogSinkError> { + unimplemented!("not used by this test") + } + } + + pub(super) struct UnusedInbox; + + #[async_trait] + impl InboxResolver for UnusedInbox { + async fn deliver( + &self, + _inbox_id: Uuid, + _result: picloud_shared::InboxResult, + ) -> picloud_shared::InboxDeliveryOutcome { + unimplemented!("not used by this test") + } + } + + pub(super) fn disabled_script(app_id: AppId) -> Script { + Script { + id: ScriptId::new(), + app_id, + name: "worker".into(), + description: None, + version: 1, + source: "0".into(), + kind: picloud_shared::ScriptKind::Endpoint, + timeout_seconds: 30, + memory_limit_mb: 64, + sandbox: ScriptSandbox::default(), + enabled: false, + created_at: Utc::now(), + updated_at: Utc::now(), + } + } + + #[tokio::test] + async fn disabled_queue_consumer_releases_claim_without_executing() { + let app_id = AppId::new(); + let script = disabled_script(app_id); + + let claimed = ClaimedMessage { + id: picloud_shared::QueueMessageId::new(), + app_id, + queue_name: "jobs".into(), + payload: serde_json::Value::Null, + enqueued_at: Utc::now(), + attempt: 1, + max_attempts: 5, + claim_token: Uuid::new_v4(), + }; + + let consumer = ActiveQueueConsumer { + trigger_id: picloud_shared::TriggerId::new(), + app_id, + script_id: script.id, + queue_name: "jobs".into(), + visibility_timeout_secs: 30, + retry_max_attempts: 5, + retry_backoff: BackoffShape::Exponential, + retry_base_ms: 1000, + registered_by_principal: AdminUserId::new(), + }; + + let nacked = Arc::new(AtomicBool::new(false)); + let executed = Arc::new(AtomicBool::new(false)); + + let dispatcher = Dispatcher { + outbox: Arc::new(UnusedOutbox), + triggers: Arc::new(UnusedTriggers), + scripts: Arc::new(DisabledScriptRepo { + script: script.clone(), + }), + 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(ClaimNackQueue { + claimed, + nacked: nacked.clone(), + }), + config: TriggerConfig::from_env(), + instance_id: "test-instance".into(), + }; + + let result = dispatcher.dispatch_one_queue(&consumer).await; + + // 1. The disabled path returns Ok(()). + assert!(result.is_ok(), "dispatch_one_queue returned {result:?}"); + // 2. The claim was released via nack. + assert!( + nacked.load(Ordering::SeqCst), + "expected nack to release the claim for a disabled consumer" + ); + // 3. The executor was never reached. If the `if !script.enabled` + // gate is deleted, the flow falls through to resolve → + // execute and this flips to true. + assert!( + !executed.load(Ordering::SeqCst), + "executor ran for a disabled queue consumer; fire-time gate missing" + ); + } + } + + // ---------------------------------------------------------------- + // OUTBOX (async-HTTP) arm fire-time `enabled` gate (§4.3). + // + // `dispatch_one` builds an `ExecRequest` from an HTTP outbox row via + // `build_http_request`, which sets `resolved.active = script.enabled` + // but does NOT itself reject a disabled script. The unified gate at + // `if !resolved.active` then drops the row (delete) without executing. + // This test proves that gate end-to-end for an HTTP-source row whose + // target script was disabled after the row was enqueued. + // + // Regression property (mirrors the queue test): `RecordingExecutor` + // flips `executed = true` before erroring, so DELETING the + // `if !resolved.active` gate makes the flow fall through to execute, + // flipping `executed` and failing `assert!(!executed)`. + // ---------------------------------------------------------------- + mod outbox_enabled_gate { + use super::*; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + + use async_trait::async_trait; + use chrono::Utc; + use picloud_orchestrator_core::ExecutionGate; + use picloud_shared::AppId; + use uuid::Uuid; + + use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow, OutboxSourceKind}; + + // Reuse the stub trait impls + helpers from the queue test so the + // two gate tests share one set of in-memory deps. + use super::queue_enabled_gate::{ + disabled_script, DisabledScriptRepo, OkPrincipals, RecordingExecutor, UnusedAbandoned, + UnusedDeadLetters, UnusedInbox, UnusedLogSink, UnusedTriggers, + }; + + // QueueRepo is never reached on the outbox arm — a panic stub keeps + // the surface honest without pulling the queue test's claim stub. + struct UnusedQueue; + + #[async_trait] + impl crate::queue_repo::QueueRepo for UnusedQueue { + async fn enqueue( + &self, + _msg: crate::queue_repo::NewQueueMessage, + ) -> Result + { + unimplemented!("not used by this test") + } + async fn claim( + &self, + _app_id: AppId, + _queue_name: &str, + ) -> Result, crate::queue_repo::QueueRepoError> + { + unimplemented!("not used by this test") + } + async fn ack( + &self, + _message_id: picloud_shared::QueueMessageId, + _claim_token: Uuid, + ) -> Result { + unimplemented!("not used by this test") + } + async fn nack( + &self, + _message_id: picloud_shared::QueueMessageId, + _claim_token: Uuid, + _retry_delay: chrono::Duration, + ) -> Result { + unimplemented!("not used by this test") + } + async fn reclaim_visibility_timeouts( + &self, + ) -> Result { + unimplemented!("not used by this test") + } + async fn depth( + &self, + _app_id: AppId, + _queue_name: &str, + ) -> Result { + unimplemented!("not used by this test") + } + async fn depth_pending( + &self, + _app_id: AppId, + _queue_name: &str, + ) -> Result { + unimplemented!("not used by this test") + } + async fn list_for_app( + &self, + _app_id: AppId, + ) -> Result< + Vec<(String, crate::queue_repo::QueueStats)>, + crate::queue_repo::QueueRepoError, + > { + unimplemented!("not used by this test") + } + #[allow(clippy::too_many_arguments)] + async fn dead_letter( + &self, + _message_id: picloud_shared::QueueMessageId, + _claim_token: Uuid, + _app_id: AppId, + _queue_name: &str, + _trigger_id: Option, + _script_id: Option, + _attempt: u32, + _first_attempt_at: chrono::DateTime, + _last_error: &str, + ) -> Result + { + unimplemented!("not used by this test") + } + } + + // OutboxRepo whose `delete` records the id it was called with; the + // other methods are never reached on the disabled-drop path. + struct RecordingOutbox { + deleted: Arc>>, + } + + #[async_trait] + impl OutboxRepo for RecordingOutbox { + async fn insert( + &self, + _row: NewOutboxRow, + ) -> Result { + unimplemented!("not used by this test") + } + async fn claim_due( + &self, + _claimed_by: &str, + _limit: i64, + ) -> Result, crate::outbox_repo::OutboxRepoError> { + unimplemented!("not used by this test") + } + async fn delete(&self, id: Uuid) -> Result<(), crate::outbox_repo::OutboxRepoError> { + *self.deleted.lock().unwrap() = Some(id); + Ok(()) + } + async fn reschedule( + &self, + _id: Uuid, + _attempt_count: u32, + _next_attempt_at: chrono::DateTime, + ) -> Result<(), crate::outbox_repo::OutboxRepoError> { + unimplemented!("not used by this test") + } + } + + #[tokio::test] + async fn disabled_http_outbox_row_is_dropped_without_executing() { + let app_id = AppId::new(); + let script = disabled_script(app_id); + + // Minimal valid HttpDispatchPayload (see shared::outbox_writer). + let payload = serde_json::json!({ + "script_name": "worker", + "path": "/hook", + "method": "POST", + "headers": {}, + "body": null, + "params": {}, + "query": {}, + "rest": "", + "timeout_seconds": 30, + }); + + let row_id = Uuid::new_v4(); + let row = OutboxRow { + id: row_id, + // Same app_id as the script so the same-app guard passes. + app_id, + source_kind: OutboxSourceKind::Http, + trigger_id: None, + script_id: Some(script.id), + reply_to: None, + payload, + origin_principal: None, + trigger_depth: 0, + 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: script.clone(), + }), + 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), + config: TriggerConfig::from_env(), + instance_id: "test-instance".into(), + }; + + let result = dispatcher.dispatch_one(row).await; + + // 1. The disabled-drop path returns Ok(()). + assert!(result.is_ok(), "dispatch_one returned {result:?}"); + // 2. The outbox row was deleted with its own id. + assert_eq!( + *deleted.lock().unwrap(), + Some(row_id), + "expected the disabled HTTP outbox row to be deleted" + ); + // 3. The executor was never reached. If the `if !resolved.active` + // gate at dispatch_one is deleted, the flow falls through to + // execute and this flips to true. + assert!( + !executed.load(Ordering::SeqCst), + "executor ran for a disabled HTTP outbox row; fire-time gate missing" + ); + } + } }