fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(v1.1.1-routes): outbox-routed sync HTTP + dispatch_mode=async

Browse Source 
Routes gain `dispatch_mode TEXT NOT NULL DEFAULT 'sync'` (migration
0012). Existing routes default to sync so the migration is
non-breaking. `DispatchMode` enum lands in `picloud-shared`.

The user-routes orchestrator handler now branches:
- `dispatch_mode = async` → write outbox row with `reply_to = None`,
  return `202 Accepted` + `{accepted_at, execution_id}`. Dispatcher
  fires the script in the background; retries / dead-letters via
  the framework from commit 5.
- `dispatch_mode = sync` → register an inbox channel
  (`tokio::sync::oneshot`), write outbox row with `reply_to =
  inbox_id`, `.await` on the receiver with a timeout =
  script.timeout_seconds + 2s buffer. Dispatcher hands the result
  back; orchestrator maps `InboxResult` into the HTTP response per
  the design-notes §3 status-code table (422/502/503/504/507/500).

`InboxRegistry` (orchestrator-core/src/inbox.rs) is the in-process
implementation of `InboxResolver`. Lock-free HashMap of pending
oneshot senders keyed by `inbox_id`. Tests cover register/deliver
round-trip, unknown-id is abandoned, dropped-receiver is abandoned,
explicit cancel. Cluster mode (v1.3+) swaps this for
LISTEN/NOTIFY-keyed lookup behind the same trait.

`OutboxWriter` trait lives in `picloud-shared` so orchestrator-core
can write to the outbox without depending on manager-core (which
would invert the dependency arrow). `PostgresOutboxRepo` implements
both `OutboxRepo` (dispatcher surface) and `OutboxWriter`
(orchestrator surface); the picloud binary clones the same concrete
Arc into both trait views.

The dispatcher's HTTP arm (commit 5 had a stub) now decodes the
`HttpDispatchPayload` off the outbox row, looks up the script,
synthesizes an `ExecRequest`, and runs it through the executor.
Outcome routing reuses the same path as KV triggers — sync HTTP
flows through the inbox, async dispatch gets dropped after
success (or DL'd on exhaustion).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-01 22:12:55 +02:00
parent 6a2971ac70
commit 77b2cb58bb
14 changed files with 767 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

344
crates/orchestrator-core/src/api.rs
View File

@@ -17,13 +17,15 @@ use axum::{
use chrono::Utc;
use picloud_executor_core::{ExecError, ExecRequest, ExecResponse, InvocationType};
use picloud_shared::{
AppId, ExecutionId, ExecutionLog, ExecutionLogSink, ExecutionStatus, Principal, RequestId,
ScriptId,
AppId, DispatchMode, ExecutionId, ExecutionLog, ExecutionLogSink, ExecutionStatus,
HttpDispatchPayload, InboxFailureKind, InboxResult, NewHttpOutbox, OutboxWriter, Principal,
RequestId, ScriptId,
};
use serde_json::Value as Json_;
use uuid::Uuid;
use crate::client::ExecutorClient;
use crate::inbox::InboxRegistry;
use crate::resolver::{ResolverError, ScriptResolver};
use crate::routing::{AppDomainTable, RouteTable};
@@ -39,6 +41,14 @@ pub struct DataPlaneState<E, R> {
/// Routing table for user-defined paths, partitioned per app.
/// Shared with the manager (admin router writes; this side reads).
pub routes: Arc<RouteTable>,
/// NATS-style inbox registry (v1.1.1). Used by sync HTTP via
/// outbox to await the dispatcher's delivery on a oneshot
/// channel.
pub inbox: Arc<InboxRegistry>,
/// Writer for the universal trigger outbox (v1.1.1). The sync
/// HTTP path inserts a row with `reply_to = inbox_id`; the async
/// path inserts with `reply_to = None` and returns 202.
pub outbox: Arc<dyn OutboxWriter>,
}
impl<E, R> Clone for DataPlaneState<E, R> {
@@ -49,6 +59,8 @@ impl<E, R> Clone for DataPlaneState<E, R> {
log_sink: self.log_sink.clone(),
app_domains: self.app_domains.clone(),
routes: self.routes.clone(),
inbox: self.inbox.clone(),
outbox: self.outbox.clone(),
}
}
}
@@ -202,50 +214,312 @@ where
Err(e) => return Err(ApiError::BadRequest(format!("body read failed: {e}"))),
};
let mut req = build_exec_request(
matched.matched.script_id,
&script.name,
&headers,
&body_bytes,
app_id,
principal,
)?;
req.path = path;
req.params = matched.params;
req.query = parse_query_string(&query_str);
req.rest = matched.rest.unwrap_or_default();
req.sandbox_overrides = script.sandbox;
let body_json: Json_ = if body_bytes.is_empty() {
Json_::Null
} else {
serde_json::from_slice(&body_bytes)
.map_err(|e| ApiError::BadRequest(format!("invalid JSON body: {e}")))?
};
let header_map: BTreeMap<String, String> = headers
.iter()
.filter_map(|(k, v)| {
v.to_str()
.ok()
.map(|s| (k.as_str().to_string(), s.to_string()))
})
.collect();
let query = parse_query_string(&query_str);
let rest = matched.rest.clone().unwrap_or_default();
let request_id = req.request_id;
let request_path = req.path.clone();
let request_headers = req.headers.clone();
let request_body = req.body.clone();
match matched.matched.dispatch_mode {
DispatchMode::Async => {
handle_async_route(
&state,
app_id,
matched.matched.route_id,
matched.matched.script_id,
&script.name,
path,
method,
header_map,
body_json,
matched.params,
query,
rest,
script.timeout_seconds,
principal,
)
.await
}
DispatchMode::Sync => {
handle_sync_route(
&state,
app_id,
matched.matched.route_id,
matched.matched.script_id,
&script.name,
path,
method,
header_map,
body_json,
matched.params,
query,
rest,
script.timeout_seconds,
principal,
)
.await
}
}
}
let timeout = Duration::from_secs(u64::from(script.timeout_seconds));
#[allow(clippy::too_many_arguments)]
async fn handle_async_route<E, R>(
state: &DataPlaneState<E, R>,
app_id: AppId,
route_id: Uuid,
script_id: ScriptId,
script_name: &str,
path: String,
method: String,
headers: BTreeMap<String, String>,
body: Json_,
params: BTreeMap<String, String>,
query: BTreeMap<String, String>,
rest: String,
timeout_seconds: u32,
principal: Option<Principal>,
) -> Result<Response, ApiError>
where
E: ExecutorClient + 'static,
R: ScriptResolver + 'static,
{
let payload = HttpDispatchPayload {
script_name: script_name.to_string(),
path,
method,
headers,
body,
params,
query,
rest,
timeout_seconds,
};
let payload_value = serde_json::to_value(&payload)
.map_err(|e| ApiError::BadRequest(format!("payload serialize: {e}")))?;
let execution_id = ExecutionId::new();
state
.outbox
.enqueue_http(NewHttpOutbox {
app_id,
route_id,
script_id,
reply_to: None,
payload: payload_value,
origin_principal: principal.map(|p| p.user_id),
trigger_depth: 0,
root_execution_id: Some(execution_id),
})
.await
.map_err(|e| ApiError::OutboxWrite(e.to_string()))?;
Ok((
StatusCode::ACCEPTED,
Json(serde_json::json!({
"accepted_at": Utc::now().to_rfc3339(),
"execution_id": execution_id.to_string(),
})),
)
.into_response())
}
#[allow(clippy::too_many_arguments)]
async fn handle_sync_route<E, R>(
state: &DataPlaneState<E, R>,
app_id: AppId,
route_id: Uuid,
script_id: ScriptId,
script_name: &str,
path: String,
method: String,
headers: BTreeMap<String, String>,
body: Json_,
params: BTreeMap<String, String>,
query: BTreeMap<String, String>,
rest: String,
timeout_seconds: u32,
principal: Option<Principal>,
) -> Result<Response, ApiError>
where
E: ExecutorClient + 'static,
R: ScriptResolver + 'static,
{
let payload = HttpDispatchPayload {
script_name: script_name.to_string(),
path: path.clone(),
method,
headers: headers.clone(),
body: body.clone(),
params,
query,
rest,
timeout_seconds,
};
let payload_value = serde_json::to_value(&payload)
.map_err(|e| ApiError::BadRequest(format!("payload serialize: {e}")))?;
// Register the inbox before writing the outbox row so the
// dispatcher can't race-deliver before the orchestrator is
// listening.
let (inbox_id, rx) = state.inbox.register();
let execution_id = ExecutionId::new();
let outbox_id = state
.outbox
.enqueue_http(NewHttpOutbox {
app_id,
route_id,
script_id,
reply_to: Some(inbox_id),
payload: payload_value,
origin_principal: principal.map(|p| p.user_id),
trigger_depth: 0,
root_execution_id: Some(execution_id),
})
.await
.map_err(|e| {
// Failed outbox write — abandon the inbox so the dispatcher
// can never deliver to a stale entry.
state.inbox.cancel(inbox_id);
ApiError::OutboxWrite(e.to_string())
})?;
// Wait for the dispatcher's delivery. Outer timeout = script
// wall-clock + a small buffer to cover dispatcher latency.
let wait_budget = Duration::from_secs(u64::from(timeout_seconds)) + Duration::from_secs(2);
let request_id = RequestId::new();
let started = Utc::now();
let outcome = state.executor.execute(&script.source, req, timeout).await;
let result = tokio::time::timeout(wait_budget, rx).await;
let finished = Utc::now();
let log = build_execution_log(
script.app_id,
matched.matched.script_id,
// Tear down the receiver if it's still alive. `inbox.cancel` is a
// no-op when the dispatcher already delivered.
let _ = state.inbox.cancel(inbox_id);
let response = match result {
Ok(Ok(InboxResult::Success(summary))) => http_response_from_summary(summary),
Ok(Ok(InboxResult::Failure { kind, message })) => failure_to_response(kind, &message),
Ok(Err(_recv)) => {
// Channel was closed without a value — dispatcher dropped
// the sender. Treat as platform failure.
tracing::warn!(
outbox_id = %outbox_id,
"inbox channel closed without delivery"
);
failure_to_response(
InboxFailureKind::Platform,
"dispatcher closed inbox without delivery",
)
}
Err(_elapsed) => {
// Outer timeout — either the script was too slow or the
// dispatcher is wedged. Returns 504 by default.
failure_to_response(InboxFailureKind::Timeout, "request timed out")
}
};
let log = build_inbox_execution_log(
app_id,
script_id,
request_id,
request_path,
request_headers,
request_body,
&outcome,
path,
headers,
body,
response.status().as_u16(),
started,
finished,
);
if let Err(e) = state.log_sink.record(log).await {
tracing::warn!(
error = %e,
script_id = %matched.matched.script_id,
%script_id,
"failed to persist execution log"
);
}
Ok(exec_response_to_http(outcome?))
Ok(response)
}
fn http_response_from_summary(summary: picloud_shared::ExecResponseSummary) -> Response {
let status =
StatusCode::from_u16(summary.status_code).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
let mut http_headers = HeaderMap::new();
for (k, v) in summary.headers {
if let (Ok(name), Ok(value)) = (k.parse::<HeaderName>(), v.parse::<HeaderValue>()) {
http_headers.insert(name, value);
}
}
http_headers
.entry(axum::http::header::CONTENT_TYPE)
.or_insert_with(|| HeaderValue::from_static("application/json"));
(status, http_headers, Json(summary.body)).into_response()
}
/// Map `InboxFailureKind` onto the design-notes §3 status-code table.
fn failure_to_response(kind: InboxFailureKind, message: &str) -> Response {
let status = match kind {
InboxFailureKind::Validation => StatusCode::UNPROCESSABLE_ENTITY,
InboxFailureKind::Runtime => StatusCode::BAD_GATEWAY,
InboxFailureKind::Overloaded => StatusCode::SERVICE_UNAVAILABLE,
InboxFailureKind::Timeout => StatusCode::GATEWAY_TIMEOUT,
InboxFailureKind::OperationBudget => StatusCode::INSUFFICIENT_STORAGE,
InboxFailureKind::Platform => StatusCode::INTERNAL_SERVER_ERROR,
};
let body = Json(serde_json::json!({ "error": message }));
if matches!(kind, InboxFailureKind::Overloaded) {
return (status, [(axum::http::header::RETRY_AFTER, "1")], body).into_response();
}
(status, body).into_response()
}
#[allow(clippy::too_many_arguments)]
fn build_inbox_execution_log(
app_id: AppId,
script_id: ScriptId,
request_id: RequestId,
request_path: String,
request_headers: BTreeMap<String, String>,
request_body: Json_,
response_code: u16,
started: chrono::DateTime<Utc>,
finished: chrono::DateTime<Utc>,
) -> ExecutionLog {
let duration_ms = u64::try_from(
finished
.signed_duration_since(started)
.num_milliseconds()
.max(0),
)
.unwrap_or(0);
let status = if (200..400).contains(&response_code) {
ExecutionStatus::Success
} else {
ExecutionStatus::Error
};
ExecutionLog {
id: Uuid::new_v4(),
app_id,
script_id,
request_id,
request_path,
request_headers,
request_body,
response_code: Some(response_code),
response_body: None,
script_logs: Json_::Array(vec![]),
duration_ms,
status,
created_at: started,
}
}
fn parse_query_string(s: &str) -> BTreeMap<String, String> {
@@ -421,6 +695,9 @@ pub enum ApiError {
#[error("execution error: {0}")]
Exec(#[from] ExecError),
#[error("outbox write failed: {0}")]
OutboxWrite(String),
}
impl IntoResponse for ApiError {
@@ -444,6 +721,13 @@ impl IntoResponse for ApiError {
let (status, message) = match &self {
E::NotFound(_) => (StatusCode::NOT_FOUND, self.to_string()),
E::BadRequest(_) => (StatusCode::BAD_REQUEST, self.to_string()),
E::OutboxWrite(e) => {
tracing::error!(error = %e, "outbox write failed");
(
StatusCode::INTERNAL_SERVER_ERROR,
"internal error".to_string(),
)
}
E::Resolver(e) => {
tracing::error!(error = %e, "resolver failure");
(

139
crates/orchestrator-core/src/inbox.rs Normal file
View File

@@ -0,0 +1,139 @@
//! In-process `InboxRegistry` — the NATS-style request/reply
//! implementation for sync HTTP via the trigger outbox (design notes
//! §3).
//!
//! Workflow:
//! 1. Orchestrator allocates an `inbox_id`, calls
//! `registry.register()` to get a oneshot receiver.
//! 2. Orchestrator writes an outbox row with `reply_to = inbox_id`.
//! 3. Dispatcher picks the row, runs the script, calls
//! `registry.deliver(inbox_id, result)`.
//! 4. Orchestrator's `.await` on the receiver fires; it maps the
//! `InboxResult` back into an HTTP response.
//!
//! `Delivered` means the receiver was alive when delivery hit. If the
//! orchestrator timed out and dropped the receiver before delivery,
//! `Abandoned` comes back — the dispatcher writes an
//! `abandoned_executions` row (design notes §3 #9).
//!
//! Cluster mode (v1.3+) swaps this for a Postgres `LISTEN/NOTIFY`-
//! based resolver; the `InboxResolver` trait stays the same.
use std::collections::HashMap;
use std::sync::Mutex;
use async_trait::async_trait;
use picloud_shared::{InboxDeliveryOutcome, InboxResolver, InboxResult};
use tokio::sync::oneshot;
use uuid::Uuid;
pub struct InboxRegistry {
inner: Mutex<HashMap<Uuid, oneshot::Sender<InboxResult>>>,
}
impl InboxRegistry {
#[must_use]
pub fn new() -> Self {
Self {
inner: Mutex::new(HashMap::new()),
}
}
/// Allocate a new inbox id and register the sender side. The
/// caller awaits the returned `Receiver`; the dispatcher delivers
/// the outcome via `deliver(id, …)`.
#[must_use]
pub fn register(&self) -> (Uuid, oneshot::Receiver<InboxResult>) {
let id = Uuid::new_v4();
let (tx, rx) = oneshot::channel();
if let Ok(mut g) = self.inner.lock() {
g.insert(id, tx);
}
(id, rx)
}
/// Cancel a pending inbox (orchestrator timed out and gave up).
/// Drops the sender so any future `deliver` returns `Abandoned`.
/// Returns `true` if the receiver was still registered.
pub fn cancel(&self, id: Uuid) -> bool {
self.inner
.lock()
.map(|mut g| g.remove(&id).is_some())
.unwrap_or(false)
}
}
impl Default for InboxRegistry {
fn default() -> Self {
Self::new()
}
}
#[async_trait]
impl InboxResolver for InboxRegistry {
async fn deliver(&self, inbox_id: Uuid, result: InboxResult) -> InboxDeliveryOutcome {
let Ok(mut g) = self.inner.lock() else {
return InboxDeliveryOutcome::Abandoned;
};
let Some(tx) = g.remove(&inbox_id) else {
return InboxDeliveryOutcome::Abandoned;
};
// `send` returns Err iff the receiver was dropped — exactly
// the abandoned-execution case.
if tx.send(result).is_err() {
InboxDeliveryOutcome::Abandoned
} else {
InboxDeliveryOutcome::Delivered
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use picloud_shared::ExecResponseSummary;
use std::collections::BTreeMap;
fn ok_result() -> InboxResult {
InboxResult::Success(ExecResponseSummary {
status_code: 200,
headers: BTreeMap::new(),
body: serde_json::json!({ "ok": true }),
})
}
#[tokio::test]
async fn register_then_deliver_resolves_receiver() {
let reg = InboxRegistry::new();
let (id, rx) = reg.register();
let outcome = reg.deliver(id, ok_result()).await;
assert_eq!(outcome, InboxDeliveryOutcome::Delivered);
let received = rx.await.expect("receiver should fire");
assert!(matches!(received, InboxResult::Success(_)));
}
#[tokio::test]
async fn deliver_to_unknown_id_is_abandoned() {
let reg = InboxRegistry::new();
let outcome = reg.deliver(Uuid::new_v4(), ok_result()).await;
assert_eq!(outcome, InboxDeliveryOutcome::Abandoned);
}
#[tokio::test]
async fn dropping_receiver_then_delivering_is_abandoned() {
let reg = InboxRegistry::new();
let (id, rx) = reg.register();
drop(rx);
let outcome = reg.deliver(id, ok_result()).await;
assert_eq!(outcome, InboxDeliveryOutcome::Abandoned);
}
#[tokio::test]
async fn cancel_removes_sender() {
let reg = InboxRegistry::new();
let (id, _rx) = reg.register();
assert!(reg.cancel(id));
let outcome = reg.deliver(id, ok_result()).await;
assert_eq!(outcome, InboxDeliveryOutcome::Abandoned);
}
}

2
crates/orchestrator-core/src/lib.rs
View File

@@ -11,10 +11,12 @@
pub mod api;
pub mod client;
pub mod gate;
pub mod inbox;
pub mod resolver;
pub mod routing;
pub use api::{data_plane_router, user_routes_router, DataPlaneState};
pub use client::{ExecutorClient, LocalExecutorClient, RemoteExecutorClient};
pub use gate::{AcquireError, ExecutionGate};
pub use inbox::InboxRegistry;
pub use resolver::{ResolverError, ScriptResolver};

9
crates/orchestrator-core/src/routing/matcher.rs
View File

@@ -38,6 +38,11 @@ pub struct MatchResult {
pub struct Matched {
pub route_id: uuid::Uuid,
pub script_id: picloud_shared::ScriptId,
/// Per-route dispatch mode (v1.1.1). Forwarded to the
/// orchestrator's HTTP handler so it can pick the sync or async
/// path. Defaults to `Sync` for older routes that predate the
/// column.
pub dispatch_mode: picloud_shared::DispatchMode,
}
/// A single route ready for matching. `app_id` is carried so the
@@ -51,6 +56,7 @@ pub struct CompiledRoute {
pub host: HostPattern,
pub path: PathPattern,
pub method: Option<String>,
pub dispatch_mode: picloud_shared::DispatchMode,
}
/// Find the best matching route for the request. Returns `None` if no
@@ -180,6 +186,7 @@ fn match_within_bucket(
matched: Matched {
route_id: route.route_id,
script_id: route.script_id,
dispatch_mode: route.dispatch_mode,
},
params: BTreeMap::new(),
rest: None,
@@ -230,6 +237,7 @@ fn match_within_bucket(
matched: Matched {
route_id: route.route_id,
script_id: route.script_id,
dispatch_mode: route.dispatch_mode,
},
params,
rest,
@@ -312,6 +320,7 @@ mod tests {
host,
path: parse_path(path_kind, raw).unwrap(),
method: None,
dispatch_mode: picloud_shared::DispatchMode::Sync,
}
}