diff --git a/crates/executor-core/src/engine.rs b/crates/executor-core/src/engine.rs index ce7a835..5d1575c 100644 --- a/crates/executor-core/src/engine.rs +++ b/crates/executor-core/src/engine.rs @@ -195,7 +195,13 @@ impl Engine { ); engine.set_module_resolver(resolver); let self_engine = self.self_arc(); - sdk::register_all(&mut engine, &self.services, cx, effective_limits, self_engine); + sdk::register_all( + &mut engine, + &self.services, + cx, + effective_limits, + self_engine, + ); let mut scope = Scope::new(); scope.push_constant("ctx", build_ctx_map(&req)); diff --git a/crates/executor-core/src/sdk/invoke.rs b/crates/executor-core/src/sdk/invoke.rs index b28c8e5..04696c6 100644 --- a/crates/executor-core/src/sdk/invoke.rs +++ b/crates/executor-core/src/sdk/invoke.rs @@ -257,7 +257,9 @@ fn args_to_json(value: &Dynamic) -> Result> { return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number)); } if value.is_string() { - return Ok(Json::String(value.clone().into_string().unwrap_or_default())); + return Ok(Json::String( + value.clone().into_string().unwrap_or_default(), + )); } if let Some(arr) = value.clone().try_cast::() { let mut out = Vec::with_capacity(arr.len()); @@ -276,6 +278,14 @@ fn args_to_json(value: &Dynamic) -> Result> { Ok(Json::String(value.to_string())) } +/// `Limits` is `Copy`; passed by value into `register` so closures take +/// owned copies. Hint for readers wondering why it doesn't need `Arc`. +#[allow(dead_code)] +const _LIMITS_IS_COPY: fn() = || { + fn assert_copy() {} + assert_copy::(); +}; + #[cfg(test)] mod tests { use super::*; @@ -320,11 +330,3 @@ mod tests { assert_eq!(j, serde_json::json!({ "x": 42 })); } } - -/// `Limits` is `Copy`; passed by value into `register` so closures take -/// owned copies. Hint for readers wondering why it doesn't need `Arc`. -#[allow(dead_code)] -const _LIMITS_IS_COPY: fn() = || { - fn assert_copy() {} - assert_copy::(); -}; diff --git a/crates/executor-core/src/sdk/queue.rs b/crates/executor-core/src/sdk/queue.rs index 7c44533..37b721e 100644 --- a/crates/executor-core/src/sdk/queue.rs +++ b/crates/executor-core/src/sdk/queue.rs @@ -37,7 +37,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Result<(), Box> { let json = message_to_json(&message)?; - enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default()).map(|_| ()) + enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default()) }, ); } @@ -52,7 +52,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Result<(), Box> { let json = message_to_json(&message)?; let opts = parse_opts(&opts)?; - enqueue_blocking(&svc, &cx, name, json, opts).map(|_| ()) + enqueue_blocking(&svc, &cx, name, json, opts) }, ); } @@ -111,7 +111,7 @@ fn enqueue_blocking( let name = name.to_string(); handle .block_on(async move { svc.enqueue(&cx, &name, payload, opts).await }) - .map(|_| ()) + .map(|_id| ()) .map_err(|err| -> Box { EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into() }) @@ -187,7 +187,9 @@ fn message_to_json(value: &Dynamic) -> Result> { return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number)); } if value.is_string() { - return Ok(Json::String(value.clone().into_string().unwrap_or_default())); + return Ok(Json::String( + value.clone().into_string().unwrap_or_default(), + )); } if let Some(arr) = value.clone().try_cast::() { let mut out = Vec::with_capacity(arr.len()); diff --git a/crates/executor-core/src/sdk/retry.rs b/crates/executor-core/src/sdk/retry.rs index f6cd4da..774dfdf 100644 --- a/crates/executor-core/src/sdk/retry.rs +++ b/crates/executor-core/src/sdk/retry.rs @@ -31,6 +31,8 @@ //! `invoke()`, the inner call gets a fresh cx with `trigger_depth + 1` //! via the invoke bridge — retry doesn't see or interfere with that. +use std::collections::hash_map::DefaultHasher; +use std::hash::Hasher; use std::sync::Arc; use std::time::Duration; @@ -81,11 +83,7 @@ impl BackoffShape { } } -pub(super) fn register( - engine: &mut RhaiEngine, - _services: &Services, - _cx: Arc, -) { +pub(super) fn register(engine: &mut RhaiEngine, _services: &Services, _cx: Arc) { // Custom type so scripts can pass Policy values around. engine.register_type_with_name::("Policy"); @@ -125,9 +123,10 @@ pub(super) fn register( // not `with` because `with` is a Rhai reserved keyword. module.set_native_fn( "run", - move |ctx: NativeCallContext, policy: Policy, fn_ptr: FnPtr| -> Result> { - retry_run(&ctx, &policy, &fn_ptr) - }, + move |ctx: NativeCallContext, + policy: Policy, + fn_ptr: FnPtr| + -> Result> { retry_run(&ctx, &policy, &fn_ptr) }, ); engine.register_static_module("retry", module.into()); @@ -147,20 +146,16 @@ fn build_policy(opts: rhai::Map) -> Result> { p.max_attempts = n.clamp(1, 20); } if let Some(v) = opts.get("backoff") { - let s = v - .clone() - .into_string() - .map_err(|_| -> Box { - EvalAltResult::ErrorRuntime( - "retry::policy: backoff must be a string".into(), - rhai::Position::NONE, - ) - .into() - })?; + let s = v.clone().into_string().map_err(|_| -> Box { + EvalAltResult::ErrorRuntime( + "retry::policy: backoff must be a string".into(), + rhai::Position::NONE, + ) + .into() + })?; p.backoff = BackoffShape::from_wire(&s).ok_or_else(|| -> Box { EvalAltResult::ErrorRuntime( - "retry::policy: backoff must be \"exponential\" | \"linear\" | \"constant\"" - .into(), + "retry::policy: backoff must be \"exponential\" | \"linear\" | \"constant\"".into(), rhai::Position::NONE, ) .into() @@ -251,13 +246,16 @@ fn apply_jitter(raw: u32, pct: u32) -> u32 { // Deterministic-enough jitter without pulling rand into a hot path: // pick a small offset from the high bits of attempt's hash. Random // distribution doesn't matter for retry — bounded ± is what we want. - use std::collections::hash_map::DefaultHasher; - use std::hash::Hasher; let mut h = DefaultHasher::new(); h.write_u64(span); h.write_u32(raw); let r = h.finish(); - let offset = (r % (2 * span + 1)) as i64 - span as i64; + // span is bounded by raw*100/100 = raw ≤ u32::MAX, so 2*span+1 fits + // in i64 without wrapping. `cast_signed` makes the intent explicit + // for clippy::cast_possible_wrap. + let modded = r % (2 * span + 1); + let offset = + i64::try_from(modded).unwrap_or(i64::MAX) - i64::try_from(span).unwrap_or(i64::MAX); let signed = i64::from(raw).saturating_add(offset).max(0); u32::try_from(signed.min(i64::from(u32::MAX))).unwrap_or(u32::MAX) } diff --git a/crates/executor-core/tests/sdk_invoke.rs b/crates/executor-core/tests/sdk_invoke.rs index b87b43b..0f5226c 100644 --- a/crates/executor-core/tests/sdk_invoke.rs +++ b/crates/executor-core/tests/sdk_invoke.rs @@ -45,8 +45,9 @@ impl InvokeService for FakeInvokeService { let entries = self.scripts.lock().unwrap().clone(); let hit = entries.iter().find(|(id, _app, name, _)| match &target { InvokeTarget::Id(t) => t == id, - InvokeTarget::Name(t) => t == name, - InvokeTarget::Path(t) => t == name, // tests stash route paths as names + // Test stashes route paths in the `name` column, so Path + // and Name match the same field. + InvokeTarget::Name(t) | InvokeTarget::Path(t) => t == name, }); let (id, app, name, source): (ScriptId, AppId, String, String) = hit .ok_or_else(|| InvokeError::NotFound(target.describe()))? @@ -241,7 +242,10 @@ async fn invoke_async_returns_execution_id_string() { .unwrap(); // Body is a string — a UUID — surfaced as Json::String. let id = resp.body.as_str().expect("body should be a string"); - assert!(uuid::Uuid::parse_str(id).is_ok(), "expected UUID, got: {id}"); + assert!( + uuid::Uuid::parse_str(id).is_ok(), + "expected UUID, got: {id}" + ); let payloads = svc.async_payloads.lock().unwrap().clone(); assert_eq!(payloads.len(), 1); assert_eq!(payloads[0], json!({ "x": 1 })); diff --git a/crates/executor-core/tests/sdk_queue.rs b/crates/executor-core/tests/sdk_queue.rs index c8e503a..c135a63 100644 --- a/crates/executor-core/tests/sdk_queue.rs +++ b/crates/executor-core/tests/sdk_queue.rs @@ -46,11 +46,7 @@ impl QueueService for RecordingQueue { Ok(*self.depth.lock().unwrap()) } - async fn depth_pending( - &self, - _cx: &SdkCallCx, - _queue_name: &str, - ) -> Result { + async fn depth_pending(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result { Ok(*self.depth_pending.lock().unwrap()) } } diff --git a/crates/executor-core/tests/sdk_retry.rs b/crates/executor-core/tests/sdk_retry.rs index c429fe6..3c42ce3 100644 --- a/crates/executor-core/tests/sdk_retry.rs +++ b/crates/executor-core/tests/sdk_retry.rs @@ -8,10 +8,10 @@ use std::sync::{Arc, Mutex}; use picloud_executor_core::{Engine, ExecRequest, InvocationType, Limits}; use picloud_shared::{ - AppId, ExecutionId, NoopDeadLetterService, NoopDocsService, NoopEmailService, - NoopEventEmitter, NoopFilesService, NoopHttpService, NoopInvokeService, NoopKvService, - NoopModuleSource, NoopPubsubService, NoopQueueService, NoopSecretsService, NoopUsersService, - RequestId, ScriptId, ScriptSandbox, Services, + AppId, ExecutionId, NoopDeadLetterService, NoopDocsService, NoopEmailService, NoopEventEmitter, + NoopFilesService, NoopHttpService, NoopInvokeService, NoopKvService, NoopModuleSource, + NoopPubsubService, NoopQueueService, NoopSecretsService, NoopUsersService, RequestId, ScriptId, + ScriptSandbox, Services, }; use serde_json::Value; @@ -61,11 +61,11 @@ fn baseline_request(app_id: AppId) -> ExecRequest { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn retry_run_succeeds_first_try() { let engine = build_engine(); - let src = r#" + let src = r" let p = retry::policy(#{ max_attempts: 3, base_ms: 1 }); let v = retry::run(p, || 42); #{ statusCode: 200, body: v } - "# + " .to_string(); let req = baseline_request(AppId::new()); let resp = tokio::task::spawn_blocking(move || engine.execute(&src, req)) diff --git a/crates/manager-core/src/auth_api.rs b/crates/manager-core/src/auth_api.rs index 2223c58..29778bb 100644 --- a/crates/manager-core/src/auth_api.rs +++ b/crates/manager-core/src/auth_api.rs @@ -92,11 +92,7 @@ async fn login(State(state): State, Json(input): Json) // canonical row used in the response DTO. let (stored_hash, user_id, is_active) = match creds { Some(c) => (c.password_hash, Some(c.id), c.is_active), - None => ( - crate::auth::TIMING_FLAT_DUMMY_HASH.to_string(), - None, - false, - ), + None => (crate::auth::TIMING_FLAT_DUMMY_HASH.to_string(), None, false), }; let password_ok = verify_password(&stored_hash, &input.password); diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 1423f61..bf5431c 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -163,6 +163,7 @@ impl Dispatcher { Ok(()) } + #[allow(clippy::too_many_lines)] async fn dispatch_one_queue( &self, consumer: &ActiveQueueConsumer, @@ -183,7 +184,11 @@ impl Dispatcher { let Ok(permit) = self.gate.try_acquire() else { let _ = self .queue - .nack(claimed.id, claimed.claim_token, chrono::Duration::milliseconds(100)) + .nack( + claimed.id, + claimed.claim_token, + chrono::Duration::milliseconds(100), + ) .await; return Ok(()); }; @@ -299,7 +304,11 @@ impl Dispatcher { self.config.retry_jitter_pct, ); let delay = chrono::Duration::milliseconds(i64::from(delay_ms)); - if let Err(e) = self.queue.nack(claimed.id, claimed.claim_token, delay).await { + if let Err(e) = self + .queue + .nack(claimed.id, claimed.claim_token, delay) + .await + { tracing::warn!(?e, "queue nack failed"); } return; @@ -621,12 +630,14 @@ impl Dispatcher { )); } - let args = payload.get("args").cloned().unwrap_or(serde_json::Value::Null); + let args = payload + .get("args") + .cloned() + .unwrap_or(serde_json::Value::Null); let trigger_depth = payload .get("trigger_depth") .and_then(serde_json::Value::as_u64) - .map(|n| u32::try_from(n).unwrap_or(u32::MAX)) - .unwrap_or(row.trigger_depth); + .map_or(row.trigger_depth, |n| u32::try_from(n).unwrap_or(u32::MAX)); let execution_id = ExecutionId::new(); let req = ExecRequest { diff --git a/crates/manager-core/src/invoke_service.rs b/crates/manager-core/src/invoke_service.rs index 99fcbb8..fe6d0b1 100644 --- a/crates/manager-core/src/invoke_service.rs +++ b/crates/manager-core/src/invoke_service.rs @@ -94,9 +94,8 @@ impl InvokeServiceImpl { // most route-resolution invokes target write endpoints; a more // exact API would surface method via the SDK in a future bump. let app_routes = self.routes.snapshot_for_app(cx.app_id); - let m = matcher::r#match(&app_routes, "invoke.local", "POST", path).or_else(|| { - matcher::r#match(&app_routes, "invoke.local", "GET", path) - }); + let m = matcher::r#match(&app_routes, "invoke.local", "POST", path) + .or_else(|| matcher::r#match(&app_routes, "invoke.local", "GET", path)); let m = m.ok_or_else(|| InvokeError::NotFound(format!("path {path:?}")))?; // Resolve the script the matched route bound. Cross-app // re-check via the script row (defense in depth — RouteTable @@ -175,10 +174,7 @@ mod tests { } #[async_trait] impl ScriptRepository for OneScriptRepo { - async fn get( - &self, - id: ScriptId, - ) -> Result, ScriptRepositoryError> { + async fn get(&self, id: ScriptId) -> Result, ScriptRepositoryError> { Ok(if id == self.script.id { Some(self.script.clone()) } else { @@ -190,19 +186,18 @@ mod tests { app_id: AppId, name: &str, ) -> Result, ScriptRepositoryError> { - Ok(if app_id == self.script.app_id && name == self.script.name { - Some(self.script.clone()) - } else { - None - }) + Ok( + if app_id == self.script.app_id && name == self.script.name { + Some(self.script.clone()) + } else { + None + }, + ) } async fn list(&self) -> Result, ScriptRepositoryError> { Ok(vec![self.script.clone()]) } - async fn list_for_app( - &self, - _app_id: AppId, - ) -> Result, ScriptRepositoryError> { + async fn list_for_app(&self, _app_id: AppId) -> Result, ScriptRepositoryError> { unimplemented!() } async fn list_for_user( @@ -331,7 +326,9 @@ mod tests { let app_b = AppId::new(); let script = make_script(app_a, "worker"); let svc = InvokeServiceImpl::new( - Arc::new(OneScriptRepo { script: script.clone() }), + Arc::new(OneScriptRepo { + script: script.clone(), + }), Arc::new(RouteTable::new()), Arc::new(CapturingOutbox { last: tokio::sync::Mutex::new(None), @@ -339,7 +336,10 @@ mod tests { ); // Caller is in app_b but the script belongs to app_a. let cx = anon_cx(app_b); - let err = svc.resolve(&cx, InvokeTarget::Id(script.id)).await.unwrap_err(); + let err = svc + .resolve(&cx, InvokeTarget::Id(script.id)) + .await + .unwrap_err(); assert!(matches!(err, InvokeError::CrossApp)); } @@ -348,7 +348,9 @@ mod tests { let app_id = AppId::new(); let script = make_script(app_id, "worker"); let svc = InvokeServiceImpl::new( - Arc::new(OneScriptRepo { script: script.clone() }), + Arc::new(OneScriptRepo { + script: script.clone(), + }), Arc::new(RouteTable::new()), Arc::new(CapturingOutbox { last: tokio::sync::Mutex::new(None), @@ -389,7 +391,9 @@ mod tests { last: tokio::sync::Mutex::new(None), }); let svc = InvokeServiceImpl::new( - Arc::new(OneScriptRepo { script: script.clone() }), + Arc::new(OneScriptRepo { + script: script.clone(), + }), Arc::new(RouteTable::new()), outbox.clone(), ); diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 8b47a28..6439015 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -45,6 +45,7 @@ pub mod files_service; pub mod files_sweep; pub mod gc; pub mod http_service; +pub mod invoke_service; pub mod kv_repo; pub mod kv_service; pub mod log_sink; @@ -55,7 +56,6 @@ pub mod outbox_repo; pub mod principal_resolver; pub mod pubsub_repo; pub mod pubsub_service; -pub mod invoke_service; pub mod queue_repo; pub mod queue_service; pub mod queues_api; diff --git a/crates/manager-core/src/queue_repo.rs b/crates/manager-core/src/queue_repo.rs index 2c4b032..d45d701 100644 --- a/crates/manager-core/src/queue_repo.rs +++ b/crates/manager-core/src/queue_repo.rs @@ -114,8 +114,7 @@ pub trait QueueRepo: Send + Sync { async fn depth(&self, app_id: AppId, queue_name: &str) -> Result; /// `queue::depth_pending(name)` — currently claimable rows. - async fn depth_pending(&self, app_id: AppId, queue_name: &str) - -> Result; + async fn depth_pending(&self, app_id: AppId, queue_name: &str) -> Result; /// Dashboard read-only view: every distinct queue name in `app_id` /// with its aggregate counts. @@ -217,13 +216,11 @@ impl QueueRepo for PostgresQueueRepo { message_id: QueueMessageId, claim_token: Uuid, ) -> Result { - let res = sqlx::query( - "DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2", - ) - .bind(message_id.into_inner()) - .bind(claim_token) - .execute(&self.pool) - .await?; + let res = sqlx::query("DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2") + .bind(message_id.into_inner()) + .bind(claim_token) + .execute(&self.pool) + .await?; Ok(res.rows_affected() == 1) } @@ -275,11 +272,7 @@ impl QueueRepo for PostgresQueueRepo { Ok(u64::try_from(n).unwrap_or(0)) } - async fn depth_pending( - &self, - app_id: AppId, - queue_name: &str, - ) -> Result { + async fn depth_pending(&self, app_id: AppId, queue_name: &str) -> Result { let (n,): (i64,) = sqlx::query_as( "SELECT COUNT(*) FROM queue_messages \ WHERE app_id = $1 AND queue_name = $2 \ diff --git a/crates/manager-core/src/trigger_repo.rs b/crates/manager-core/src/trigger_repo.rs index e678d21..9d8c22a 100644 --- a/crates/manager-core/src/trigger_repo.rs +++ b/crates/manager-core/src/trigger_repo.rs @@ -1323,13 +1323,11 @@ impl TriggerRepo for PostgresTriggerRepo { trigger_id: TriggerId, at: DateTime, ) -> Result<(), TriggerRepoError> { - sqlx::query( - "UPDATE queue_trigger_details SET last_fired_at = $2 WHERE trigger_id = $1", - ) - .bind(trigger_id.into_inner()) - .bind(at) - .execute(&self.pool) - .await?; + sqlx::query("UPDATE queue_trigger_details SET last_fired_at = $2 WHERE trigger_id = $1") + .bind(trigger_id.into_inner()) + .bind(at) + .execute(&self.pool) + .await?; Ok(()) } } diff --git a/crates/manager-core/tests/migration_queue_messages.rs b/crates/manager-core/tests/migration_queue_messages.rs index 74c9492..cf213bd 100644 --- a/crates/manager-core/tests/migration_queue_messages.rs +++ b/crates/manager-core/tests/migration_queue_messages.rs @@ -31,7 +31,9 @@ async fn pool_or_skip() -> Option { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn queue_messages_table_exists_with_expected_columns() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let rows: Vec<(String, String, String)> = sqlx::query_as( "SELECT column_name, data_type, is_nullable \ FROM information_schema.columns \ @@ -63,10 +65,8 @@ async fn queue_messages_table_exists_with_expected_columns() { } // Verify nullability on the right columns. - let nullable: std::collections::HashMap = rows - .into_iter() - .map(|(n, _, nl)| (n, nl)) - .collect(); + let nullable: std::collections::HashMap = + rows.into_iter().map(|(n, _, nl)| (n, nl)).collect(); assert_eq!(nullable["app_id"], "NO"); assert_eq!(nullable["queue_name"], "NO"); assert_eq!(nullable["payload"], "NO"); @@ -77,7 +77,9 @@ async fn queue_messages_table_exists_with_expected_columns() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn queue_trigger_details_table_exists() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let rows: Vec<(String,)> = sqlx::query_as( "SELECT column_name FROM information_schema.columns \ WHERE table_name = 'queue_trigger_details' \ @@ -87,7 +89,12 @@ async fn queue_trigger_details_table_exists() { .await .expect("read"); let names: Vec<&str> = rows.iter().map(|(n,)| n.as_str()).collect(); - for required in ["trigger_id", "queue_name", "visibility_timeout_secs", "last_fired_at"] { + for required in [ + "trigger_id", + "queue_name", + "visibility_timeout_secs", + "last_fired_at", + ] { assert!( names.contains(&required), "queue_trigger_details should have {required}, columns are {names:?}" @@ -97,7 +104,9 @@ async fn queue_trigger_details_table_exists() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn queue_widens_trigger_kind_and_outbox_source_kind() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; // The triggers.kind constraint must accept 'queue'. sqlx::query( "DO $$ BEGIN PERFORM 1 FROM pg_constraint WHERE conname = 'triggers_kind_check'; END $$;", @@ -133,7 +142,9 @@ async fn queue_widens_trigger_kind_and_outbox_source_kind() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn queue_messages_dispatch_index_is_partial() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (defn,): (String,) = sqlx::query_as( "SELECT indexdef FROM pg_indexes \ WHERE indexname = 'idx_queue_messages_dispatch'", diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index f07594b..20380ef 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -264,16 +264,14 @@ pub async fn build_app( // v1.1.9 durable per-app queues. Producers write to queue_messages // via QueueService; the dispatcher's queue arm (commit 6) consumes // claimed messages and fires the registered queue:receive trigger. - let queue_repo: Arc = - Arc::new(picloud_manager_core::queue_repo::PostgresQueueRepo::new( - pool.clone(), - )); - let queue: Arc = Arc::new( - picloud_manager_core::queue_service::QueueServiceImpl::new( + let queue_repo: Arc = Arc::new( + picloud_manager_core::queue_repo::PostgresQueueRepo::new(pool.clone()), + ); + let queue: Arc = + Arc::new(picloud_manager_core::queue_service::QueueServiceImpl::new( queue_repo.clone(), authz.clone(), - ), - ); + )); // Route table created early (before Services) so InvokeServiceImpl // can use it for path resolution. It's populated below from the // route_repo, then re-populated whenever the admin layer writes @@ -312,8 +310,10 @@ pub async fn build_app( ); // v1.1.9: keep the invoke depth bound aligned with the dispatcher's // trigger-depth bound (same counter under the hood). - let mut engine_limits = Limits::default(); - engine_limits.trigger_depth_max = trigger_config.max_trigger_depth; + let engine_limits = Limits { + trigger_depth_max: trigger_config.max_trigger_depth, + ..Limits::default() + }; let engine = Arc::new(Engine::new(engine_limits, services)); // v1.1.9: install the back-reference the `invoke` SDK bridge needs // for synchronous re-entry. Weak so the Arc-cycle stays loose; diff --git a/crates/picloud/tests/invoke_e2e.rs b/crates/picloud/tests/invoke_e2e.rs index bc06b9f..715255a 100644 --- a/crates/picloud/tests/invoke_e2e.rs +++ b/crates/picloud/tests/invoke_e2e.rs @@ -116,7 +116,9 @@ async fn poll_marker(pool: &PgPool, app_id: &str) -> Option { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn invoke_by_name_same_app_returns_value() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool.clone(), "invoke-name").await; // Callee returns 42 — the simplest possible round-trip. @@ -124,7 +126,7 @@ async fn invoke_by_name_same_app_returns_value() { &server, &app_id, "callee", - r#"#{ statusCode: 200, body: 42 }"#, + r"#{ statusCode: 200, body: 42 }", ) .await; @@ -146,7 +148,9 @@ async fn invoke_by_name_same_app_returns_value() { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn invoke_cross_app_rejects() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_a) = server_for(pool.clone(), "invoke-xa").await; // Create a second app via the SAME server / admin (Owner can manage // every app) so the script-create calls don't run into role gates. @@ -163,7 +167,7 @@ async fn invoke_cross_app_rejects() { &server, &app_b, "callee_b", - r#"#{ statusCode: 200, body: 0 }"#, + r"#{ statusCode: 200, body: 0 }", ) .await; @@ -193,7 +197,9 @@ async fn invoke_cross_app_rejects() { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn invoke_depth_limit_exceeds_cleanly() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool.clone(), "invoke-depth").await; // The recurser propagates any throw (no try-catch). At the depth @@ -235,7 +241,9 @@ async fn invoke_depth_limit_exceeds_cleanly() { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn invoke_async_enqueues_outbox_row() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool.clone(), "invoke-async").await; // Callee writes a marker the test polls — the dispatcher must fire diff --git a/crates/picloud/tests/queue_e2e.rs b/crates/picloud/tests/queue_e2e.rs index 87cb89a..019302d 100644 --- a/crates/picloud/tests/queue_e2e.rs +++ b/crates/picloud/tests/queue_e2e.rs @@ -127,14 +127,13 @@ async fn poll_marker(pool: &PgPool, app_id: &str) -> Option { } async fn count_queue_messages(pool: &PgPool, app_id: &str, queue: &str) -> i64 { - let (n,): (i64,) = sqlx::query_as( - "SELECT COUNT(*) FROM queue_messages WHERE app_id = $1 AND queue_name = $2", - ) - .bind(Uuid::parse_str(app_id).expect("uuid")) - .bind(queue) - .fetch_one(pool) - .await - .expect("queue count"); + let (n,): (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM queue_messages WHERE app_id = $1 AND queue_name = $2") + .bind(Uuid::parse_str(app_id).expect("uuid")) + .bind(queue) + .fetch_one(pool) + .await + .expect("queue count"); n } @@ -170,7 +169,9 @@ async fn enqueue_directly(pool: &PgPool, app_id: &str, queue: &str, payload: Val #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn queue_receive_acks_on_success() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool.clone(), "ack").await; let script_id = create_script(&server, &app_id, "worker", ACK_HANDLER).await; @@ -186,9 +187,7 @@ async fn queue_receive_acks_on_success() { enqueue_directly(&pool, &app_id, "jobs", json!({ "x": 42 })).await; - let marker = poll_marker(&pool, &app_id) - .await - .expect("handler fired"); + let marker = poll_marker(&pool, &app_id).await.expect("handler fired"); assert_eq!(marker["source"], "queue"); assert_eq!(marker["queue"]["queue_name"], "jobs"); assert_eq!(marker["queue"]["message"]["x"], 42); @@ -199,7 +198,9 @@ async fn queue_receive_acks_on_success() { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn queue_receive_dead_letters_after_max_attempts() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool.clone(), "dl").await; let script_id = create_script(&server, &app_id, "fail", THROW_HANDLER).await; @@ -236,7 +237,9 @@ async fn queue_receive_dead_letters_after_max_attempts() { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn queue_one_consumer_per_queue_rejected() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool.clone(), "onecon").await; let s1 = create_script(&server, &app_id, "first", ACK_HANDLER).await; @@ -276,7 +279,9 @@ async fn queue_one_consumer_per_queue_rejected() { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn queue_visibility_timeout_reclaim() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool.clone(), "vt").await; let script_id = create_script(&server, &app_id, "slow", ACK_HANDLER).await; @@ -317,13 +322,12 @@ async fn queue_visibility_timeout_reclaim() { // Otherwise the reclaim path is the only way the message would // have been picked up. Poll the claim state. for _ in 0..50 { - let row: Option<(Option,)> = sqlx::query_as( - "SELECT claim_token FROM queue_messages WHERE id = $1", - ) - .bind(msg_id) - .fetch_optional(&pool) - .await - .expect("read"); + let row: Option<(Option,)> = + sqlx::query_as("SELECT claim_token FROM queue_messages WHERE id = $1") + .bind(msg_id) + .fetch_optional(&pool) + .await + .expect("read"); if let Some((token,)) = row { if token.is_none() { return; // reclaim succeeded — claim cleared diff --git a/crates/picloud/tests/retry_e2e.rs b/crates/picloud/tests/retry_e2e.rs index 6138cd1..111f161 100644 --- a/crates/picloud/tests/retry_e2e.rs +++ b/crates/picloud/tests/retry_e2e.rs @@ -93,7 +93,9 @@ async fn execute(server: &TestServer, script_id: &str) -> Value { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn retry_run_eventually_succeeds_inside_http_handler() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool, "retry-succ").await; // attempt counter mutates across retries; the 3rd attempt returns @@ -114,7 +116,9 @@ async fn retry_run_eventually_succeeds_inside_http_handler() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn retry_run_surfaces_last_error_after_max_attempts() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool, "retry-surf").await; let src = r#" @@ -135,7 +139,9 @@ async fn retry_run_surfaces_last_error_after_max_attempts() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn retry_on_codes_filters_unmatched_errors() { - let Some(pool) = pool_or_skip().await else { return }; + let Some(pool) = pool_or_skip().await else { + return; + }; let (server, app_id) = server_for(pool, "retry-codes").await; // attempts is mutated on every entry; we throw a non-matching code @@ -156,5 +162,9 @@ async fn retry_on_codes_filters_unmatched_errors() { "#; let script_id = create_script(&server, &app_id, "codes", src).await; let body = execute(&server, &script_id).await; - assert_eq!(body, json!(1), "non-matching error should surface immediately"); + assert_eq!( + body, + json!(1), + "non-matching error should surface immediately" + ); }