diff --git a/crates/executor-core/tests/sdk_pubsub.rs b/crates/executor-core/tests/sdk_pubsub.rs index c0eac03..6aa6e31 100644 --- a/crates/executor-core/tests/sdk_pubsub.rs +++ b/crates/executor-core/tests/sdk_pubsub.rs @@ -19,6 +19,7 @@ use serde_json::{json, Value}; #[derive(Default)] struct RecordingPubsub { last: Mutex>, + count: std::sync::atomic::AtomicUsize, } #[async_trait] @@ -32,6 +33,7 @@ impl PubsubService for RecordingPubsub { if topic.trim().is_empty() { return Err(PubsubError::EmptyTopic); } + self.count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); *self.last.lock().unwrap() = Some((topic.to_string(), message)); Ok(()) } @@ -163,3 +165,40 @@ async fn publish_empty_topic_throws() { .expect("spawn_blocking should not panic"); assert!(res.is_err(), "empty topic should throw"); } + +/// `PICLOUD_MAX_EMISSIONS_PER_EXECUTION` bounds a one-request outbox-amplification +/// DoS. Only the counter itself was unit-tested (`emit_budget`); this drives it +/// through the REAL `pubsub::publish_durable` SDK surface — i.e. it pins that the +/// charge_emission call is actually wired into publish, not just that the counter +/// works. Uses the DEFAULT budget (1000) so it needs no env mutation: a loop of +/// 1001 publishes must error, and no more than the budget may have reached the +/// service. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn publish_beyond_the_emission_budget_is_refused() { + let svc = Arc::new(RecordingPubsub::default()); + let engine = make_engine(svc.clone()); + // 1001 durable publishes in one execution — one past the default ceiling. + let src = r#" + let n = 0; + while n < 1001 { pubsub::publish_durable("t", #{ i: n }); n += 1; } + "# + .to_string(); + let req = baseline_request(AppId::new()); + let res = tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .expect("spawn_blocking should not panic"); + + let err = res.expect_err("exceeding the per-execution emission budget must error"); + assert!( + format!("{err:?}").to_lowercase().contains("emission"), + "the error must name the emission budget, got: {err:?}" + ); + // The rail actually stopped the amplification: the service saw at most the + // budget, not all 1001. + let seen = svc.count.load(std::sync::atomic::Ordering::SeqCst); + assert!( + seen <= 1000, + "the service must not have received more than the 1000-emission budget, saw {seen}" + ); + assert!(seen > 0, "some publishes should have landed before the cap"); +}