From 2247c4d80440e6b188a379c0b0b0eedd99982fc0 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 6 Jun 2026 19:57:53 +0200 Subject: [PATCH] =?UTF-8?q?feat(v1.1.9):=20dispatcher=20Invoke=20arm=20?= =?UTF-8?q?=E2=80=94=20invoke=5Fasync=20runs=20once?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the commit 2 placeholder with the real OutboxSourceKind::Invoke handler. build_invoke_request(row) hydrates an (ResolvedTrigger, ExecRequest) pair from a payload InvokeService::enqueue_async wrote. Validates: - script_id present + parses as UUID - script exists - script.app_id == row.app_id (defensive — re-check the cross-app boundary; same-app was already verified at enqueue time) Synthesized ResolvedTrigger carries retry_max_attempts = 1 so the existing handle_failure path skips the retry loop. invoke_async runs exactly once; on throw the row is deleted + a dead_letters row written (handle_failure already does this when attempt >= max_attempts), so a caller who wants retry semantics wraps the call site in retry::with. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/manager-core/src/dispatcher.rs | 116 ++++++++++++++++++++++---- 1 file changed, 100 insertions(+), 16 deletions(-) diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 835965b..1423f61 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -372,22 +372,18 @@ impl Dispatcher { return Ok(()); } }, - OutboxSourceKind::Invoke => { - // Wired in commit 10 (invoke_async() — function composition). - // Until then a malformed row is impossible to write (no - // producer), but defensively drop the row so a manual - // INSERT doesn't loop the dispatcher. - tracing::warn!( - outbox_id = %row.id, - "OutboxSourceKind::Invoke not wired yet; dropping row" - ); - self.outbox - .delete(row.id) - .await - .map_err(|e| DispatcherError::Outbox(e.to_string()))?; - drop(permit); - return Ok(()); - } + OutboxSourceKind::Invoke => match self.build_invoke_request(&row).await { + Ok(pair) => pair, + Err(err) => { + tracing::warn!(outbox_id = %row.id, ?err, "invoke exec build failed; dropping"); + self.outbox + .delete(row.id) + .await + .map_err(|e| DispatcherError::Outbox(e.to_string()))?; + drop(permit); + return Ok(()); + } + }, OutboxSourceKind::Kv | OutboxSourceKind::Docs | OutboxSourceKind::DeadLetter @@ -587,6 +583,94 @@ impl Dispatcher { Ok((resolved, req)) } + /// v1.1.9. Build an `(ResolvedTrigger, ExecRequest)` from an + /// invoke_async outbox row. The payload was written by + /// `InvokeService::enqueue_async`. No retry policy — invoke_async + /// runs once; on failure the row is deleted and a dead_letters row + /// is written (so the misfire is observable). + async fn build_invoke_request( + &self, + row: &OutboxRow, + ) -> Result<(ResolvedTrigger, ExecRequest), DispatcherError> { + // Extract the args + script_id from the payload (InvokeService + // populates these). Defensive: dead_letter the row if any + // required field is missing. + let payload = &row.payload; + let Some(script_id_str) = payload.get("script_id").and_then(|v| v.as_str()) else { + return Err(DispatcherError::ResolveTrigger( + "invoke row missing script_id".into(), + )); + }; + let script_id = picloud_shared::ScriptId::from( + uuid::Uuid::parse_str(script_id_str) + .map_err(|e| DispatcherError::ResolveTrigger(format!("script_id: {e}")))?, + ); + let script = self + .scripts + .get(script_id) + .await + .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))? + .ok_or_else(|| { + DispatcherError::ResolveTrigger(format!("script {script_id} not found")) + })?; + // Same-app guard — the script could have been re-assigned or + // the row could have been hand-edited. Reject mismatches. + if script.app_id != row.app_id { + return Err(DispatcherError::ResolveTrigger( + "invoke target belongs to a different app".into(), + )); + } + + 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); + + let execution_id = ExecutionId::new(); + let req = ExecRequest { + execution_id, + request_id: RequestId::new(), + script_id: script.id, + script_name: script.name.clone(), + invocation_type: InvocationType::Function, + path: "/invoke_async".into(), + headers: std::collections::BTreeMap::new(), + body: args, + params: std::collections::BTreeMap::new(), + query: std::collections::BTreeMap::new(), + rest: String::new(), + sandbox_overrides: script.sandbox, + app_id: row.app_id, + // invoke_async runs with no principal — same as HTTP outbox. + principal: None, + trigger_depth, + root_execution_id: row.root_execution_id.unwrap_or(execution_id), + is_dead_letter_handler: false, + event: None, + }; + + // Build a synthetic ResolvedTrigger so the dispatch_one outer + // loop can carry on through the same code path. Retry knobs + // bound to 1 attempt: invoke_async runs ONCE (callers wrap in + // retry::with for retries). + let resolved = ResolvedTrigger { + trigger_kind: TriggerKind::Cron, // placeholder; not used downstream + is_dead_letter_handler: false, + script_id: script.id, + script_source: script.source, + script_name: script.name, + script_updated_at: script.updated_at, + sandbox_overrides: script.sandbox, + registered_by_principal: picloud_shared::AdminUserId::from(uuid::Uuid::nil()), + retry_max_attempts: 1, + retry_backoff: BackoffShape::Constant, + retry_base_ms: 0, + }; + Ok((resolved, req)) + } + async fn handle_success( &self, row: &OutboxRow,