a91b134285
Addresses findings from an independent review of the gap-closing commits: - R1 (regression): add #[serde(default)] to HttpDispatchPayload.method so async-HTTP outbox rows enqueued before the field existed still decode after upgrade instead of dead-lettering on "missing field `method`". Adds a regression test for the missing-key path. - method case: uppercase ctx.request.method at the orchestrator boundary so it honors its documented "uppercased" contract for extension verbs. Route matching is already case-insensitive, so matching is unaffected. - CLI hardening: percent-encode free-form path segments (app slug, topic name, ids, secret name) in the reqwest client so a value containing `/ ? #` can't break out of its URL segment. Adds a `seg()` helper + unit test and applies it uniformly across all path-interpolating methods (new and pre-existing). - S1 (docs): correct the users::email_available enumeration framing — it adds no new vector vs. create's uniqueness error, but is cheaper and unthrottled; there is no built-in throttle/CAPTCHA primitive, so the honest mitigation is a kv-based counter. Updated SDK doc-comments, trait docs, and stdlib-reference. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
104 lines
4.0 KiB
Rust
104 lines
4.0 KiB
Rust
//! `OutboxWriter` — minimal trait the orchestrator-core sync-HTTP path
|
|
//! uses to enqueue rows into the universal trigger outbox. The
|
|
//! manager-core `PostgresOutboxRepo` implements this in addition to
|
|
//! its richer `OutboxRepo` surface; defining it here lets
|
|
//! orchestrator-core depend on the trait without pulling in
|
|
//! manager-core (which would invert the dependency arrow).
|
|
|
|
use async_trait::async_trait;
|
|
use serde::{Deserialize, Serialize};
|
|
use thiserror::Error;
|
|
use uuid::Uuid;
|
|
|
|
use crate::{AdminUserId, AppId, ExecutionId, ScriptId};
|
|
|
|
/// What the orchestrator hands to the outbox when it ingests an HTTP
|
|
/// request. Carries enough for the dispatcher to reconstruct the
|
|
/// `ExecRequest` end-to-end.
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct NewHttpOutbox {
|
|
pub app_id: AppId,
|
|
/// `routes.id` of the matched route. Discriminated against
|
|
/// `triggers.id` by `source_kind = 'http'` on the outbox row.
|
|
pub route_id: Uuid,
|
|
/// Pre-resolved script so the dispatcher doesn't re-look it up.
|
|
pub script_id: ScriptId,
|
|
/// `Some(inbox_id)` for sync HTTP (the orchestrator awaits a
|
|
/// channel keyed on this id). `None` for `dispatch_mode = async`
|
|
/// — dispatcher fires-and-forgets, no reply path.
|
|
pub reply_to: Option<Uuid>,
|
|
/// Serialized `HttpDispatchPayload` (defined below) — everything
|
|
/// the dispatcher needs to reconstruct an `ExecRequest`.
|
|
pub payload: serde_json::Value,
|
|
/// The principal that ingressed the HTTP request (Some when
|
|
/// authenticated, None for public). Forensic only; the script
|
|
/// executes as the route's app principal model, not this.
|
|
pub origin_principal: Option<AdminUserId>,
|
|
/// `0` for direct HTTP ingress; the dispatcher will increment
|
|
/// for any further fan-out triggered by the script.
|
|
pub trigger_depth: u32,
|
|
pub root_execution_id: Option<ExecutionId>,
|
|
}
|
|
|
|
/// The shape the orchestrator serializes into `NewHttpOutbox.payload`
|
|
/// (the JSONB column). Mirrored on the dispatcher side so it can
|
|
/// rebuild an `ExecRequest`.
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct HttpDispatchPayload {
|
|
pub script_name: String,
|
|
pub path: String,
|
|
/// HTTP method (uppercased). `#[serde(default)]` so async-HTTP outbox
|
|
/// rows enqueued before this field existed still decode (as `""`,
|
|
/// matching non-HTTP triggers) instead of dead-lettering on upgrade.
|
|
#[serde(default)]
|
|
pub method: String,
|
|
pub headers: std::collections::BTreeMap<String, String>,
|
|
pub body: serde_json::Value,
|
|
pub params: std::collections::BTreeMap<String, String>,
|
|
pub query: std::collections::BTreeMap<String, String>,
|
|
pub rest: String,
|
|
pub timeout_seconds: u32,
|
|
}
|
|
|
|
#[async_trait]
|
|
pub trait OutboxWriter: Send + Sync {
|
|
/// Insert a sync- or async-HTTP outbox row. Returns the row's id
|
|
/// — the orchestrator stores it locally for forensics and to
|
|
/// correlate `abandoned_executions` rows when the dispatcher's
|
|
/// inbox delivery fails.
|
|
async fn enqueue_http(&self, row: NewHttpOutbox) -> Result<Uuid, OutboxWriterError>;
|
|
}
|
|
|
|
#[derive(Debug, Error)]
|
|
pub enum OutboxWriterError {
|
|
#[error("outbox write failed: {0}")]
|
|
Backend(String),
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
/// Async-HTTP outbox rows enqueued before `method` existed must still
|
|
/// decode after upgrade (it's `#[serde(default)]`), not dead-letter on
|
|
/// a "missing field `method`" error. Regression guard for that field.
|
|
#[test]
|
|
fn http_payload_decodes_without_method_field() {
|
|
let legacy = serde_json::json!({
|
|
"script_name": "todos",
|
|
"path": "/todos",
|
|
// no "method" key — pre-upgrade row
|
|
"headers": {},
|
|
"body": null,
|
|
"params": {},
|
|
"query": {},
|
|
"rest": "",
|
|
"timeout_seconds": 30
|
|
});
|
|
let payload: HttpDispatchPayload =
|
|
serde_json::from_value(legacy).expect("legacy payload must decode");
|
|
assert_eq!(payload.method, "");
|
|
assert_eq!(payload.path, "/todos");
|
|
}
|
|
}
|