fix(dispatcher): close disabled-execution + cross-app gaps on the async paths
The `enabled` lifecycle's "a disabled script is not executable via ANY path" guarantee leaked on the queue arm, and the outbox trigger arm lacked the cross-app backstop its siblings carry. - Queue arm: `dispatch_one_queue` bypasses the unified `dispatch_one` fire-time gate, so its only `enabled` guard was the per-tick `list_active_queue_consumers` snapshot — a TOCTOU window for a message claimed in a tick where the script was still enabled. Re-check the freshly-read `script.enabled` before executing and release the claim (`nack`) when disabled; the message stays queued and resumes on re-enable. The nack is load-bearing: `reclaim_visibility_timeouts` only reclaims claims for enabled triggers, so without it the message would sit claimed indefinitely. - Trigger arm: `resolve_trigger` built the ExecRequest with `app_id = row.app_id` while sourcing the body from `trigger.script_id` with no same-app check — the guard the queue/http/invoke arms already have. Add it so a hand-edited/restored row can't run one app's script under another's SdkCallCx. Both regression-locked by new unit tests (full mock harness, verified to fail if the gate is removed): - queue_enabled_gate::disabled_queue_consumer_releases_claim_without_executing - outbox_enabled_gate::disabled_http_outbox_row_is_dropped_without_executing Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
@@ -384,6 +384,39 @@ impl Dispatcher {
|
|||||||
return Ok(());
|
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
|
let principal = self
|
||||||
.principals
|
.principals
|
||||||
.resolve(consumer.registered_by_principal)
|
.resolve(consumer.registered_by_principal)
|
||||||
@@ -743,6 +776,22 @@ impl Dispatcher {
|
|||||||
DispatcherError::ResolveTrigger(format!("script {} not found", trigger.script_id))
|
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 {
|
Ok(ResolvedTrigger {
|
||||||
trigger_kind: trigger.kind,
|
trigger_kind: trigger.kind,
|
||||||
is_dead_letter_handler: matches!(trigger.kind, TriggerKind::DeadLetter),
|
is_dead_letter_handler: matches!(trigger.kind, TriggerKind::DeadLetter),
|
||||||
@@ -1535,4 +1584,837 @@ mod tests {
|
|||||||
);
|
);
|
||||||
assert_eq!(failure_kind_to_status(InboxFailureKind::Platform), 500);
|
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<Option<Script>, ScriptRepositoryError> {
|
||||||
|
Ok(Some(self.script.clone()))
|
||||||
|
}
|
||||||
|
async fn get_by_name(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_name: &str,
|
||||||
|
) -> Result<Option<Script>, ScriptRepositoryError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn list_for_app(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
) -> Result<Vec<Script>, ScriptRepositoryError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn list_for_user(
|
||||||
|
&self,
|
||||||
|
_user_id: AdminUserId,
|
||||||
|
) -> Result<Vec<Script>, ScriptRepositoryError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn create(&self, _input: NewScript) -> Result<Script, ScriptRepositoryError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn update(
|
||||||
|
&self,
|
||||||
|
_id: ScriptId,
|
||||||
|
_patch: ScriptPatch,
|
||||||
|
) -> Result<Script, ScriptRepositoryError> {
|
||||||
|
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<i64, ScriptRepositoryError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn count_triggers_for_script(
|
||||||
|
&self,
|
||||||
|
_script_id: ScriptId,
|
||||||
|
) -> Result<i64, ScriptRepositoryError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn list_imports(
|
||||||
|
&self,
|
||||||
|
_script_id: ScriptId,
|
||||||
|
) -> Result<Vec<Script>, ScriptRepositoryError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- QueueRepo: `claim` returns the message, `nack` records. ----
|
||||||
|
struct ClaimNackQueue {
|
||||||
|
claimed: ClaimedMessage,
|
||||||
|
nacked: Arc<AtomicBool>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl QueueRepo for ClaimNackQueue {
|
||||||
|
async fn enqueue(
|
||||||
|
&self,
|
||||||
|
_msg: NewQueueMessage,
|
||||||
|
) -> Result<picloud_shared::QueueMessageId, crate::queue_repo::QueueRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn claim(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_queue_name: &str,
|
||||||
|
) -> Result<Option<ClaimedMessage>, crate::queue_repo::QueueRepoError> {
|
||||||
|
Ok(Some(self.claimed.clone()))
|
||||||
|
}
|
||||||
|
async fn ack(
|
||||||
|
&self,
|
||||||
|
_message_id: picloud_shared::QueueMessageId,
|
||||||
|
_claim_token: Uuid,
|
||||||
|
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn nack(
|
||||||
|
&self,
|
||||||
|
_message_id: picloud_shared::QueueMessageId,
|
||||||
|
_claim_token: Uuid,
|
||||||
|
_retry_delay: chrono::Duration,
|
||||||
|
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||||||
|
self.nacked.store(true, Ordering::SeqCst);
|
||||||
|
Ok(true)
|
||||||
|
}
|
||||||
|
async fn reclaim_visibility_timeouts(
|
||||||
|
&self,
|
||||||
|
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn depth(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_queue_name: &str,
|
||||||
|
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn depth_pending(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_queue_name: &str,
|
||||||
|
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn list_for_app(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
) -> Result<Vec<(String, 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<picloud_shared::TriggerId>,
|
||||||
|
_script_id: Option<ScriptId>,
|
||||||
|
_attempt: u32,
|
||||||
|
_first_attempt_at: chrono::DateTime<Utc>,
|
||||||
|
_last_error: &str,
|
||||||
|
) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError>
|
||||||
|
{
|
||||||
|
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<Principal, PrincipalResolverError> {
|
||||||
|
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<AtomicBool>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl ExecutorClient for RecordingExecutor {
|
||||||
|
async fn execute(
|
||||||
|
&self,
|
||||||
|
_source: &str,
|
||||||
|
_req: ExecRequest,
|
||||||
|
_timeout: std::time::Duration,
|
||||||
|
) -> Result<ExecResponse, ExecError> {
|
||||||
|
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<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn create_docs_trigger(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_req: crate::trigger_repo::CreateDocsTrigger,
|
||||||
|
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn create_dead_letter_trigger(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_req: crate::trigger_repo::CreateDeadLetterTrigger,
|
||||||
|
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn create_cron_trigger(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_req: crate::trigger_repo::CreateCronTrigger,
|
||||||
|
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn create_files_trigger(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_req: crate::trigger_repo::CreateFilesTrigger,
|
||||||
|
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn create_pubsub_trigger(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_req: crate::trigger_repo::CreatePubsubTrigger,
|
||||||
|
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn create_email_trigger(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_req: crate::trigger_repo::CreateEmailTrigger,
|
||||||
|
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn email_inbound_target(
|
||||||
|
&self,
|
||||||
|
_trigger_id: picloud_shared::TriggerId,
|
||||||
|
) -> Result<
|
||||||
|
Option<crate::trigger_repo::EmailInboundTarget>,
|
||||||
|
crate::trigger_repo::TriggerRepoError,
|
||||||
|
> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn list_for_app(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
) -> Result<Vec<crate::trigger_repo::Trigger>, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn get(
|
||||||
|
&self,
|
||||||
|
_id: picloud_shared::TriggerId,
|
||||||
|
) -> Result<Option<crate::trigger_repo::Trigger>, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn delete(
|
||||||
|
&self,
|
||||||
|
_id: picloud_shared::TriggerId,
|
||||||
|
) -> Result<bool, crate::trigger_repo::TriggerRepoError> {
|
||||||
|
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::KvTriggerMatch>,
|
||||||
|
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::DocsTriggerMatch>,
|
||||||
|
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::FilesTriggerMatch>,
|
||||||
|
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<picloud_shared::TriggerId>,
|
||||||
|
_script_id: Option<ScriptId>,
|
||||||
|
) -> Result<
|
||||||
|
Vec<crate::trigger_repo::DeadLetterTriggerMatch>,
|
||||||
|
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<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn list_active_queue_consumers(
|
||||||
|
&self,
|
||||||
|
) -> Result<Vec<ActiveQueueConsumer>, 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<Utc>,
|
||||||
|
) -> 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<picloud_shared::DeadLetterId, crate::dead_letter_repo::DeadLetterRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn get(
|
||||||
|
&self,
|
||||||
|
_id: picloud_shared::DeadLetterId,
|
||||||
|
) -> Result<
|
||||||
|
Option<crate::dead_letter_repo::DeadLetterRow>,
|
||||||
|
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::DeadLetterRow>,
|
||||||
|
crate::dead_letter_repo::DeadLetterRepoError,
|
||||||
|
> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn unresolved_count(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
) -> Result<i64, crate::dead_letter_repo::DeadLetterRepoError> {
|
||||||
|
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<Utc>,
|
||||||
|
_limit: i64,
|
||||||
|
) -> Result<u64, crate::dead_letter_repo::DeadLetterRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct UnusedOutbox;
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl OutboxRepo for UnusedOutbox {
|
||||||
|
async fn insert(
|
||||||
|
&self,
|
||||||
|
_row: NewOutboxRow,
|
||||||
|
) -> Result<Uuid, crate::outbox_repo::OutboxRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn claim_due(
|
||||||
|
&self,
|
||||||
|
_claimed_by: &str,
|
||||||
|
_limit: i64,
|
||||||
|
) -> Result<Vec<OutboxRow>, 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<Utc>,
|
||||||
|
) -> 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<Uuid, crate::abandoned_repo::AbandonedRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn gc(
|
||||||
|
&self,
|
||||||
|
_older_than: chrono::DateTime<Utc>,
|
||||||
|
_limit: i64,
|
||||||
|
) -> Result<u64, crate::abandoned_repo::AbandonedRepoError> {
|
||||||
|
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<picloud_shared::QueueMessageId, crate::queue_repo::QueueRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn claim(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_queue_name: &str,
|
||||||
|
) -> Result<Option<crate::queue_repo::ClaimedMessage>, crate::queue_repo::QueueRepoError>
|
||||||
|
{
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn ack(
|
||||||
|
&self,
|
||||||
|
_message_id: picloud_shared::QueueMessageId,
|
||||||
|
_claim_token: Uuid,
|
||||||
|
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn nack(
|
||||||
|
&self,
|
||||||
|
_message_id: picloud_shared::QueueMessageId,
|
||||||
|
_claim_token: Uuid,
|
||||||
|
_retry_delay: chrono::Duration,
|
||||||
|
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn reclaim_visibility_timeouts(
|
||||||
|
&self,
|
||||||
|
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn depth(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_queue_name: &str,
|
||||||
|
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn depth_pending(
|
||||||
|
&self,
|
||||||
|
_app_id: AppId,
|
||||||
|
_queue_name: &str,
|
||||||
|
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||||||
|
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<picloud_shared::TriggerId>,
|
||||||
|
_script_id: Option<picloud_shared::ScriptId>,
|
||||||
|
_attempt: u32,
|
||||||
|
_first_attempt_at: chrono::DateTime<Utc>,
|
||||||
|
_last_error: &str,
|
||||||
|
) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError>
|
||||||
|
{
|
||||||
|
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<Mutex<Option<Uuid>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl OutboxRepo for RecordingOutbox {
|
||||||
|
async fn insert(
|
||||||
|
&self,
|
||||||
|
_row: NewOutboxRow,
|
||||||
|
) -> Result<Uuid, crate::outbox_repo::OutboxRepoError> {
|
||||||
|
unimplemented!("not used by this test")
|
||||||
|
}
|
||||||
|
async fn claim_due(
|
||||||
|
&self,
|
||||||
|
_claimed_by: &str,
|
||||||
|
_limit: i64,
|
||||||
|
) -> Result<Vec<OutboxRow>, 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<Utc>,
|
||||||
|
) -> 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"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user