feat(interceptors): per-interceptor wall-clock timeout (§9.4 M5)
A [[interceptors]] marker can set timeout_ms (migration 0075, CHECK > 0); a
runaway guard (loop {}) is interrupted and the op DENIED (fail closed) within
budget rather than hanging the write path. The effective deadline is
min(caller-remaining, now + timeout) — a hook can only tighten, never extend,
its caller's deadline. NULL uses PICLOUD_INTERCEPTOR_TIMEOUT_MS (default 5s).
Wiring: run_resolved_blocking splits into a core + a _with_timeout variant that
computes the effective deadline from engine::ambient_deadline() and runs via
execute_ast_with_deadline; run_one_hook passes the marker's timeout_ms (or the
env default). timeout_ms threaded end to end — manifest, plan, BundleInterceptor,
the reconcile diff (part of the mutable body: a timeout change is an Update),
insert_interceptor_tx, resolve_chain/list_for_owner/list_on_app_chain +
SealedInterceptor/InterceptorMarker, and interceptor_service → ResolvedInterceptor.
Schema snapshot re-blessed. Pinned by a journey: a loop{} guard with
timeout_ms=100 is denied and its write does not persist.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -134,6 +134,7 @@ Environment variables consumed by the `picloud` binary:
|
|||||||
| `PICLOUD_DB_MAX_CONNECTIONS` | `32` | Postgres pool size. Matched to `PICLOUD_MAX_CONCURRENT_EXECUTIONS` so the data plane can't starve background workers. |
|
| `PICLOUD_DB_MAX_CONNECTIONS` | `32` | Postgres pool size. Matched to `PICLOUD_MAX_CONCURRENT_EXECUTIONS` so the data plane can't starve background workers. |
|
||||||
| `PICLOUD_SESSION_TTL_HOURS` | `24` | Sliding-window admin session lifetime (idle timeout). |
|
| `PICLOUD_SESSION_TTL_HOURS` | `24` | Sliding-window admin session lifetime (idle timeout). |
|
||||||
| `PICLOUD_SESSION_ABSOLUTE_TTL_HOURS` | `720` (30 days) | Absolute hard cap on an admin session's lifetime (audit 2026-07-11 C1). The sliding `touch` bump is clamped at `login_time + this`, so even a continuously-used or stolen-but-warm admin token self-expires. Mirrors the data-plane app-user session cap. |
|
| `PICLOUD_SESSION_ABSOLUTE_TTL_HOURS` | `720` (30 days) | Absolute hard cap on an admin session's lifetime (audit 2026-07-11 C1). The sliding `touch` bump is clamped at `login_time + this`, so even a continuously-used or stolen-but-warm admin token self-expires. Mirrors the data-plane app-user session cap. |
|
||||||
|
| `PICLOUD_INTERCEPTOR_TIMEOUT_MS` | `5000` (5s) | §9.4 M5 default per-interceptor wall-clock timeout when a `[[interceptors]]` marker sets no `timeout_ms`. Bounds ONE hook run so a runaway guard (`loop {}`) is denied (fail closed) rather than hanging the guarded write. The effective deadline is always tightened to at most the caller's remaining budget — a hook can never EXTEND its caller's deadline. |
|
||||||
| `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` | `600` (10 min) | How long a dispatcher may hold an OUTBOX claim before the reclaim task returns the row to the queue. **Without this a crash or restart mid-dispatch stranded every in-flight row permanently** — the dispatcher claims a row and only clears the claim on success (delete) or failure (reschedule), and `claim_due` selects `claimed_at IS NULL`, so a process that died in between left rows nothing would ever pick up again. Every other claim-based store (queue, group queue, workflow steps) already had a reclaimer; the outbox was the gap, and since it is the universal trigger path, the loss covered kv/docs/files/cron/pubsub/email/`invoke_async`/dead-letter alike. The default is deliberately generous: a script may run 300s, so a claim older than twice that is abandoned rather than slow. A reclaim does **not** bump `attempt_count` (the handler never ran — same reasoning as the transient queue `release`). Runs on the existing `PICLOUD_QUEUE_RECLAIM_INTERVAL_MS` ticker. |
|
| `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` | `600` (10 min) | How long a dispatcher may hold an OUTBOX claim before the reclaim task returns the row to the queue. **Without this a crash or restart mid-dispatch stranded every in-flight row permanently** — the dispatcher claims a row and only clears the claim on success (delete) or failure (reschedule), and `claim_due` selects `claimed_at IS NULL`, so a process that died in between left rows nothing would ever pick up again. Every other claim-based store (queue, group queue, workflow steps) already had a reclaimer; the outbox was the gap, and since it is the universal trigger path, the loss covered kv/docs/files/cron/pubsub/email/`invoke_async`/dead-letter alike. The default is deliberately generous: a script may run 300s, so a claim older than twice that is abandoned rather than slow. A reclaim does **not** bump `attempt_count` (the handler never ran — same reasoning as the transient queue `release`). Runs on the existing `PICLOUD_QUEUE_RECLAIM_INTERVAL_MS` ticker. |
|
||||||
| `PICLOUD_REALTIME_BROADCAST_CAPACITY` | `64` | Per-channel SSE broadcast buffer depth (a slow consumer sees oldest events dropped). |
|
| `PICLOUD_REALTIME_BROADCAST_CAPACITY` | `64` | Per-channel SSE broadcast buffer depth (a slow consumer sees oldest events dropped). |
|
||||||
| `PICLOUD_REALTIME_MAX_CHANNELS` | `100000` | Max live SSE channels per map (per-app and per-group). A subscribe that would open a NEW channel past this is refused with 503 (audit 2026-07-11 B6), so a client naming unbounded distinct topics can't grow the broadcaster maps to OOM. |
|
| `PICLOUD_REALTIME_MAX_CHANNELS` | `100000` | Max live SSE channels per map (per-app and per-group). A subscribe that would open a NEW channel past this is refused with 503 (audit 2026-07-11 B6), so a client naming unbounded distinct topics can't grow the broadcaster maps to OOM. |
|
||||||
|
|||||||
@@ -44,6 +44,14 @@ fn current_deadline() -> Option<Instant> {
|
|||||||
CURRENT_DEADLINE.with(Cell::get)
|
CURRENT_DEADLINE.with(Cell::get)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The current-thread execution deadline, if one is installed. Public so the
|
||||||
|
/// interceptor hook can compute a per-hook timeout as `min(ambient, now + t)` —
|
||||||
|
/// a hook must never be able to EXTEND its caller's deadline (§9.4 M5).
|
||||||
|
#[must_use]
|
||||||
|
pub fn ambient_deadline() -> Option<Instant> {
|
||||||
|
current_deadline()
|
||||||
|
}
|
||||||
|
|
||||||
use chrono::{DateTime, Utc};
|
use chrono::{DateTime, Utc};
|
||||||
use picloud_shared::{
|
use picloud_shared::{
|
||||||
ScriptId, ScriptValidator, SdkCallCx, Services, TriggerEvent, ValidatedScript, ValidationError,
|
ScriptId, ScriptValidator, SdkCallCx, Services, TriggerEvent, ValidatedScript, ValidationError,
|
||||||
|
|||||||
@@ -23,7 +23,8 @@
|
|||||||
//! execution, makes any nested op the interceptor performs bypass interception.
|
//! execution, makes any nested op the interceptor performs bypass interception.
|
||||||
|
|
||||||
use std::cell::{Cell, RefCell};
|
use std::cell::{Cell, RefCell};
|
||||||
use std::sync::Arc;
|
use std::sync::{Arc, LazyLock};
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use picloud_shared::{InterceptorChain, InterceptorEntry, InterceptorService, ScriptId, SdkCallCx};
|
use picloud_shared::{InterceptorChain, InterceptorEntry, InterceptorService, ScriptId, SdkCallCx};
|
||||||
use rhai::EvalAltResult;
|
use rhai::EvalAltResult;
|
||||||
@@ -33,7 +34,18 @@ use tokio::runtime::Handle as TokioHandle;
|
|||||||
use crate::engine::Engine;
|
use crate::engine::Engine;
|
||||||
use crate::sandbox::Limits;
|
use crate::sandbox::Limits;
|
||||||
use crate::sdk::bridge::runtime_err;
|
use crate::sdk::bridge::runtime_err;
|
||||||
use crate::sdk::invoke::run_resolved_blocking;
|
use crate::sdk::invoke::run_resolved_blocking_with_timeout;
|
||||||
|
|
||||||
|
/// Default per-interceptor wall-clock timeout when a marker sets no `timeout_ms`
|
||||||
|
/// (§9.4 M5). Read once from `PICLOUD_INTERCEPTOR_TIMEOUT_MS`; a runaway guard
|
||||||
|
/// (`loop {}`) is denied once this elapses rather than hanging the write path.
|
||||||
|
static DEFAULT_TIMEOUT_MS: LazyLock<u32> = LazyLock::new(|| {
|
||||||
|
std::env::var("PICLOUD_INTERCEPTOR_TIMEOUT_MS")
|
||||||
|
.ok()
|
||||||
|
.and_then(|s| s.parse::<u32>().ok())
|
||||||
|
.filter(|&v| v > 0)
|
||||||
|
.unwrap_or(5000)
|
||||||
|
});
|
||||||
|
|
||||||
/// Clone-cheap bundle of the §9.4 interceptor dependencies threaded into every
|
/// Clone-cheap bundle of the §9.4 interceptor dependencies threaded into every
|
||||||
/// SDK service `register` fn (M1). Carries the resolver, the `invoke()` re-entry
|
/// SDK service `register` fn (M1). Carries the resolver, the `invoke()` re-entry
|
||||||
@@ -288,17 +300,23 @@ fn run_one_hook(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Run the sealed script with the operation context as its request body, under
|
// Run the sealed script with the operation context as its request body, under
|
||||||
// the re-entrancy guard (its own writes bypass interception) and the identity
|
// the re-entrancy guard (its own writes bypass interception), the identity
|
||||||
// cycle guard (its own script id is on the stack).
|
// cycle guard (its own script id is on the stack), and a per-hook timeout
|
||||||
|
// (§9.4 M5 — the marker's `timeout_ms` or the env default, tightened to at
|
||||||
|
// most the caller's remaining deadline). A runaway guard is denied, not hung.
|
||||||
|
let timeout = Duration::from_millis(u64::from(
|
||||||
|
resolved.timeout_ms.unwrap_or(*DEFAULT_TIMEOUT_MS),
|
||||||
|
));
|
||||||
let ret = {
|
let ret = {
|
||||||
let _reentry = ReentryGuard::enter();
|
let _reentry = ReentryGuard::enter();
|
||||||
let _cycle = CycleGuard::enter(script.script_id);
|
let _cycle = CycleGuard::enter(script.script_id);
|
||||||
run_resolved_blocking(
|
run_resolved_blocking_with_timeout(
|
||||||
self_engine,
|
self_engine,
|
||||||
cx,
|
cx,
|
||||||
script,
|
script,
|
||||||
payload.clone(),
|
payload.clone(),
|
||||||
&format!("{service}::{op} interceptor `{}`", script.name),
|
&format!("{service}::{op} interceptor `{}`", script.name),
|
||||||
|
Some(timeout),
|
||||||
)?
|
)?
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
@@ -179,12 +179,53 @@ fn invoke_blocking(
|
|||||||
/// response `body` JSON. Shared by `invoke()` and the §9.4 interceptor hook —
|
/// response `body` JSON. Shared by `invoke()` and the §9.4 interceptor hook —
|
||||||
/// the caller is responsible for the depth check (both do it before resolving).
|
/// the caller is responsible for the depth check (both do it before resolving).
|
||||||
/// `label` prefixes any compile/execute error.
|
/// `label` prefixes any compile/execute error.
|
||||||
|
/// Run a resolved script through the `invoke()` re-entry core, inheriting the
|
||||||
|
/// caller's ambient execution deadline.
|
||||||
pub(super) fn run_resolved_blocking(
|
pub(super) fn run_resolved_blocking(
|
||||||
self_engine: &Arc<Engine>,
|
self_engine: &Arc<Engine>,
|
||||||
cx: &Arc<SdkCallCx>,
|
cx: &Arc<SdkCallCx>,
|
||||||
resolved: &picloud_shared::ResolvedScript,
|
resolved: &picloud_shared::ResolvedScript,
|
||||||
body_json: Json,
|
body_json: Json,
|
||||||
label: &str,
|
label: &str,
|
||||||
|
) -> Result<Json, Box<EvalAltResult>> {
|
||||||
|
run_resolved_core(
|
||||||
|
self_engine,
|
||||||
|
cx,
|
||||||
|
resolved,
|
||||||
|
body_json,
|
||||||
|
label,
|
||||||
|
crate::engine::ambient_deadline(),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Like [`run_resolved_blocking`] but bounds the run by a per-hook `timeout`
|
||||||
|
/// (§9.4 M5). The effective deadline is `min(ambient, now + timeout)` — a hook
|
||||||
|
/// can only ever TIGHTEN, never extend, its caller's deadline.
|
||||||
|
pub(super) fn run_resolved_blocking_with_timeout(
|
||||||
|
self_engine: &Arc<Engine>,
|
||||||
|
cx: &Arc<SdkCallCx>,
|
||||||
|
resolved: &picloud_shared::ResolvedScript,
|
||||||
|
body_json: Json,
|
||||||
|
label: &str,
|
||||||
|
timeout: Option<std::time::Duration>,
|
||||||
|
) -> Result<Json, Box<EvalAltResult>> {
|
||||||
|
let ambient = crate::engine::ambient_deadline();
|
||||||
|
let own = timeout.map(|d| std::time::Instant::now() + d);
|
||||||
|
let effective = match (ambient, own) {
|
||||||
|
(Some(a), Some(o)) => Some(a.min(o)),
|
||||||
|
(a, None) => a,
|
||||||
|
(None, o) => o,
|
||||||
|
};
|
||||||
|
run_resolved_core(self_engine, cx, resolved, body_json, label, effective)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn run_resolved_core(
|
||||||
|
self_engine: &Arc<Engine>,
|
||||||
|
cx: &Arc<SdkCallCx>,
|
||||||
|
resolved: &picloud_shared::ResolvedScript,
|
||||||
|
body_json: Json,
|
||||||
|
label: &str,
|
||||||
|
deadline: Option<std::time::Instant>,
|
||||||
) -> Result<Json, Box<EvalAltResult>> {
|
) -> Result<Json, Box<EvalAltResult>> {
|
||||||
let req = ExecRequest {
|
let req = ExecRequest {
|
||||||
execution_id: ExecutionId::new(),
|
execution_id: ExecutionId::new(),
|
||||||
@@ -218,7 +259,7 @@ pub(super) fn run_resolved_blocking(
|
|||||||
EvalAltResult::ErrorRuntime(format!("{label}: {e}").into(), rhai::Position::NONE).into()
|
EvalAltResult::ErrorRuntime(format!("{label}: {e}").into(), rhai::Position::NONE).into()
|
||||||
})?;
|
})?;
|
||||||
let resp = self_engine
|
let resp = self_engine
|
||||||
.execute_ast(&ast, req)
|
.execute_ast_with_deadline(&ast, req, deadline)
|
||||||
.map_err(|e| -> Box<EvalAltResult> {
|
.map_err(|e| -> Box<EvalAltResult> {
|
||||||
EvalAltResult::ErrorRuntime(format!("{label}: {e}").into(), rhai::Position::NONE).into()
|
EvalAltResult::ErrorRuntime(format!("{label}: {e}").into(), rhai::Position::NONE).into()
|
||||||
})?;
|
})?;
|
||||||
|
|||||||
10
crates/manager-core/migrations/0075_interceptor_timeout.sql
Normal file
10
crates/manager-core/migrations/0075_interceptor_timeout.sql
Normal file
@@ -0,0 +1,10 @@
|
|||||||
|
-- §9.4 Service Interceptors M5 — per-interceptor wall-clock timeout.
|
||||||
|
--
|
||||||
|
-- An optional per-marker `timeout_ms` bounds how long ONE interceptor hook may
|
||||||
|
-- run before it is denied (a runaway guard, e.g. `loop {}`, must not hang the
|
||||||
|
-- guarded write path). NULL means "use the instance default"
|
||||||
|
-- (`PICLOUD_INTERCEPTOR_TIMEOUT_MS`). The effective deadline is always tightened
|
||||||
|
-- to at most the caller's remaining budget — a hook can never EXTEND it.
|
||||||
|
ALTER TABLE interceptors
|
||||||
|
ADD COLUMN timeout_ms INTEGER
|
||||||
|
CHECK (timeout_ms IS NULL OR timeout_ms > 0);
|
||||||
@@ -126,6 +126,10 @@ pub struct BundleInterceptor {
|
|||||||
#[serde(default = "default_before_phase")]
|
#[serde(default = "default_before_phase")]
|
||||||
pub phase: String,
|
pub phase: String,
|
||||||
pub script: String,
|
pub script: String,
|
||||||
|
/// §9.4 M5: per-hook timeout in ms (`None` = instance default). `default`
|
||||||
|
/// so a pre-M5 CLI (which omits it) still deserializes.
|
||||||
|
#[serde(default)]
|
||||||
|
pub timeout_ms: Option<i32>,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn default_before_phase() -> String {
|
fn default_before_phase() -> String {
|
||||||
@@ -828,8 +832,8 @@ pub struct CurrentState {
|
|||||||
/// v1.2 Workflows owned directly by this node (app-owned; empty for a group).
|
/// v1.2 Workflows owned directly by this node (app-owned; empty for a group).
|
||||||
pub workflows: Vec<crate::workflow_repo::Workflow>,
|
pub workflows: Vec<crate::workflow_repo::Workflow>,
|
||||||
/// §9.4 interceptor markers declared directly at this node, as
|
/// §9.4 interceptor markers declared directly at this node, as
|
||||||
/// `(service, op, phase, script)` tuples.
|
/// `(service, op, phase, script, timeout_ms)` tuples.
|
||||||
pub interceptors: Vec<(String, String, String, String)>,
|
pub interceptors: Vec<(String, String, String, String, Option<i32>)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// One row of the read-only extension-point report (§5.5).
|
/// One row of the read-only extension-point report (§5.5).
|
||||||
@@ -1620,6 +1624,7 @@ impl ApplyService {
|
|||||||
&bi.op,
|
&bi.op,
|
||||||
&bi.phase,
|
&bi.phase,
|
||||||
&bi.script,
|
&bi.script,
|
||||||
|
bi.timeout_ms,
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||||||
@@ -4263,7 +4268,7 @@ impl ApplyService {
|
|||||||
.await
|
.await
|
||||||
.map_err(|e| ApplyError::Backend(e.to_string()))?
|
.map_err(|e| ApplyError::Backend(e.to_string()))?
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|m| (m.service, m.op, m.phase, m.script))
|
.map(|m| (m.service, m.op, m.phase, m.script, m.timeout_ms))
|
||||||
.collect();
|
.collect();
|
||||||
Ok(CurrentState {
|
Ok(CurrentState {
|
||||||
scripts,
|
scripts,
|
||||||
@@ -4352,24 +4357,30 @@ fn compute_diff_with_names(
|
|||||||
/// declaring is a `Delete` (applied only under `--prune`).
|
/// declaring is a `Delete` (applied only under `--prune`).
|
||||||
fn diff_interceptors(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
|
fn diff_interceptors(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
|
||||||
let key = |service: &str, op: &str, phase: &str| format!("{service}/{op}/{phase}");
|
let key = |service: &str, op: &str, phase: &str| format!("{service}/{op}/{phase}");
|
||||||
let live: HashMap<String, &str> = current
|
// The reconcile identity is `(service, op, phase)`; the script AND the
|
||||||
|
// timeout are the mutable body, so a change to either is an Update.
|
||||||
|
let live: HashMap<String, (&str, Option<i32>)> = current
|
||||||
.interceptors
|
.interceptors
|
||||||
.iter()
|
.iter()
|
||||||
.map(|(s, o, p, script)| (key(s, o, p), script.as_str()))
|
.map(|(s, o, p, script, timeout)| (key(s, o, p), (script.as_str(), *timeout)))
|
||||||
.collect();
|
.collect();
|
||||||
let mut out = Vec::new();
|
let mut out = Vec::new();
|
||||||
for bi in &bundle.interceptors {
|
for bi in &bundle.interceptors {
|
||||||
let k = key(&bi.service, &bi.op, &bi.phase);
|
let k = key(&bi.service, &bi.op, &bi.phase);
|
||||||
match live.get(&k) {
|
match live.get(&k) {
|
||||||
Some(cur) if *cur == bi.script => out.push(ResourceChange {
|
Some((cur_script, cur_timeout))
|
||||||
op: Op::NoOp,
|
if *cur_script == bi.script && *cur_timeout == bi.timeout_ms =>
|
||||||
key: k,
|
{
|
||||||
detail: None,
|
out.push(ResourceChange {
|
||||||
}),
|
op: Op::NoOp,
|
||||||
|
key: k,
|
||||||
|
detail: None,
|
||||||
|
});
|
||||||
|
}
|
||||||
Some(_) => out.push(ResourceChange {
|
Some(_) => out.push(ResourceChange {
|
||||||
op: Op::Update,
|
op: Op::Update,
|
||||||
key: k,
|
key: k,
|
||||||
detail: Some("interceptor script changed".into()),
|
detail: Some("interceptor script or timeout changed".into()),
|
||||||
}),
|
}),
|
||||||
None => out.push(ResourceChange {
|
None => out.push(ResourceChange {
|
||||||
op: Op::Create,
|
op: Op::Create,
|
||||||
@@ -4378,7 +4389,7 @@ fn diff_interceptors(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceCha
|
|||||||
}),
|
}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
for (s, o, p, _) in ¤t.interceptors {
|
for (s, o, p, _, _) in ¤t.interceptors {
|
||||||
let present = bundle
|
let present = bundle
|
||||||
.interceptors
|
.interceptors
|
||||||
.iter()
|
.iter()
|
||||||
|
|||||||
@@ -23,6 +23,8 @@ pub struct InterceptorMarker {
|
|||||||
/// `before` (allow/deny + transform) or `after` (observe; §9.4 M3).
|
/// `before` (allow/deny + transform) or `after` (observe; §9.4 M3).
|
||||||
pub phase: String,
|
pub phase: String,
|
||||||
pub script: String,
|
pub script: String,
|
||||||
|
/// §9.4 M5: per-hook timeout, or `None` for the instance default.
|
||||||
|
pub timeout_ms: Option<i32>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// List the markers declared **directly at** `owner` (not inherited), ordered
|
/// List the markers declared **directly at** `owner` (not inherited), ordered
|
||||||
@@ -31,10 +33,10 @@ pub async fn list_for_owner(
|
|||||||
pool: &PgPool,
|
pool: &PgPool,
|
||||||
owner: ScriptOwner,
|
owner: ScriptOwner,
|
||||||
) -> Result<Vec<InterceptorMarker>, sqlx::Error> {
|
) -> Result<Vec<InterceptorMarker>, sqlx::Error> {
|
||||||
let rows: Vec<(String, String, String, String)> = match owner {
|
let rows: Vec<(String, String, String, String, Option<i32>)> = match owner {
|
||||||
ScriptOwner::App(a) => {
|
ScriptOwner::App(a) => {
|
||||||
sqlx::query_as(
|
sqlx::query_as(
|
||||||
"SELECT service, op, phase, interceptor_script FROM interceptors \
|
"SELECT service, op, phase, interceptor_script, timeout_ms FROM interceptors \
|
||||||
WHERE app_id = $1 ORDER BY service, op, phase",
|
WHERE app_id = $1 ORDER BY service, op, phase",
|
||||||
)
|
)
|
||||||
.bind(a.into_inner())
|
.bind(a.into_inner())
|
||||||
@@ -43,7 +45,7 @@ pub async fn list_for_owner(
|
|||||||
}
|
}
|
||||||
ScriptOwner::Group(g) => {
|
ScriptOwner::Group(g) => {
|
||||||
sqlx::query_as(
|
sqlx::query_as(
|
||||||
"SELECT service, op, phase, interceptor_script FROM interceptors \
|
"SELECT service, op, phase, interceptor_script, timeout_ms FROM interceptors \
|
||||||
WHERE group_id = $1 ORDER BY service, op, phase",
|
WHERE group_id = $1 ORDER BY service, op, phase",
|
||||||
)
|
)
|
||||||
.bind(g.into_inner())
|
.bind(g.into_inner())
|
||||||
@@ -53,12 +55,15 @@ pub async fn list_for_owner(
|
|||||||
};
|
};
|
||||||
Ok(rows
|
Ok(rows
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|(service, op, phase, script)| InterceptorMarker {
|
.map(
|
||||||
service,
|
|(service, op, phase, script, timeout_ms)| InterceptorMarker {
|
||||||
op,
|
service,
|
||||||
phase,
|
op,
|
||||||
script,
|
phase,
|
||||||
})
|
script,
|
||||||
|
timeout_ms,
|
||||||
|
},
|
||||||
|
)
|
||||||
.collect())
|
.collect())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -69,10 +74,10 @@ pub async fn list_on_app_chain(
|
|||||||
pool: &PgPool,
|
pool: &PgPool,
|
||||||
app_id: AppId,
|
app_id: AppId,
|
||||||
) -> Result<Vec<InterceptorMarker>, sqlx::Error> {
|
) -> Result<Vec<InterceptorMarker>, sqlx::Error> {
|
||||||
let rows: Vec<(String, String, String, String)> = sqlx::query_as(&format!(
|
let rows: Vec<(String, String, String, String, Option<i32>)> = sqlx::query_as(&format!(
|
||||||
"{CHAIN_LEVELS_CTE} \
|
"{CHAIN_LEVELS_CTE} \
|
||||||
SELECT DISTINCT ON (i.service, i.op, i.phase) \
|
SELECT DISTINCT ON (i.service, i.op, i.phase) \
|
||||||
i.service, i.op, i.phase, i.interceptor_script \
|
i.service, i.op, i.phase, i.interceptor_script, i.timeout_ms \
|
||||||
FROM chain c \
|
FROM chain c \
|
||||||
JOIN interceptors i ON (i.app_id = c.app_owner OR i.group_id = c.group_owner) \
|
JOIN interceptors i ON (i.app_id = c.app_owner OR i.group_id = c.group_owner) \
|
||||||
ORDER BY i.service, i.op, i.phase, c.depth ASC",
|
ORDER BY i.service, i.op, i.phase, c.depth ASC",
|
||||||
@@ -82,12 +87,15 @@ pub async fn list_on_app_chain(
|
|||||||
.await?;
|
.await?;
|
||||||
Ok(rows
|
Ok(rows
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|(service, op, phase, script)| InterceptorMarker {
|
.map(
|
||||||
service,
|
|(service, op, phase, script, timeout_ms)| InterceptorMarker {
|
||||||
op,
|
service,
|
||||||
phase,
|
op,
|
||||||
script,
|
phase,
|
||||||
})
|
script,
|
||||||
|
timeout_ms,
|
||||||
|
},
|
||||||
|
)
|
||||||
.collect())
|
.collect())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -100,6 +108,8 @@ pub struct SealedInterceptor {
|
|||||||
pub marker_app: Option<Uuid>,
|
pub marker_app: Option<Uuid>,
|
||||||
pub marker_group: Option<Uuid>,
|
pub marker_group: Option<Uuid>,
|
||||||
pub script_name: String,
|
pub script_name: String,
|
||||||
|
/// §9.4 M5: per-hook wall-clock timeout, or `None` for the instance default.
|
||||||
|
pub timeout_ms: Option<i32>,
|
||||||
/// `(script_id, source, updated_at)` of the sealed script, or `None` if it
|
/// `(script_id, source, updated_at)` of the sealed script, or `None` if it
|
||||||
/// is missing/disabled at the declaring owner.
|
/// is missing/disabled at the declaring owner.
|
||||||
pub script: Option<(Uuid, String, DateTime<Utc>)>,
|
pub script: Option<(Uuid, String, DateTime<Utc>)>,
|
||||||
@@ -157,6 +167,7 @@ pub async fn resolve_chain(
|
|||||||
Option<Uuid>, // marker_group
|
Option<Uuid>, // marker_group
|
||||||
String, // script_name
|
String, // script_name
|
||||||
String, // phase
|
String, // phase
|
||||||
|
Option<i32>, // timeout_ms
|
||||||
i32, // depth (0 = app, larger = ancestor)
|
i32, // depth (0 = app, larger = ancestor)
|
||||||
Option<Uuid>, // script id
|
Option<Uuid>, // script id
|
||||||
Option<String>, // script source
|
Option<String>, // script source
|
||||||
@@ -166,12 +177,12 @@ pub async fn resolve_chain(
|
|||||||
markers AS ( \
|
markers AS ( \
|
||||||
SELECT i.interceptor_script AS script_name, \
|
SELECT i.interceptor_script AS script_name, \
|
||||||
i.app_id AS marker_app, i.group_id AS marker_group, \
|
i.app_id AS marker_app, i.group_id AS marker_group, \
|
||||||
i.phase AS phase, c.depth AS depth \
|
i.phase AS phase, i.timeout_ms AS timeout_ms, c.depth AS depth \
|
||||||
FROM chain c \
|
FROM chain c \
|
||||||
JOIN interceptors i ON (i.app_id = c.app_owner OR i.group_id = c.group_owner) \
|
JOIN interceptors i ON (i.app_id = c.app_owner OR i.group_id = c.group_owner) \
|
||||||
WHERE i.service = $2 AND i.op = $3 \
|
WHERE i.service = $2 AND i.op = $3 \
|
||||||
) \
|
) \
|
||||||
SELECT m.marker_app, m.marker_group, m.script_name, m.phase, m.depth, \
|
SELECT m.marker_app, m.marker_group, m.script_name, m.phase, m.timeout_ms, m.depth, \
|
||||||
s.id, s.source, s.updated_at \
|
s.id, s.source, s.updated_at \
|
||||||
FROM markers m \
|
FROM markers m \
|
||||||
LEFT JOIN scripts s ON ( \
|
LEFT JOIN scripts s ON ( \
|
||||||
@@ -191,11 +202,12 @@ pub async fn resolve_chain(
|
|||||||
// depth = ancestor runs first); `after` app→ancestor = depth ASC.
|
// depth = ancestor runs first); `after` app→ancestor = depth ASC.
|
||||||
let mut before: Vec<(i32, SealedInterceptor)> = Vec::new();
|
let mut before: Vec<(i32, SealedInterceptor)> = Vec::new();
|
||||||
let mut after: Vec<(i32, SealedInterceptor)> = Vec::new();
|
let mut after: Vec<(i32, SealedInterceptor)> = Vec::new();
|
||||||
for (marker_app, marker_group, script_name, phase, depth, sid, src, upd) in rows {
|
for (marker_app, marker_group, script_name, phase, timeout_ms, depth, sid, src, upd) in rows {
|
||||||
let sealed = SealedInterceptor {
|
let sealed = SealedInterceptor {
|
||||||
marker_app,
|
marker_app,
|
||||||
marker_group,
|
marker_group,
|
||||||
script_name,
|
script_name,
|
||||||
|
timeout_ms,
|
||||||
script: match (sid, src, upd) {
|
script: match (sid, src, upd) {
|
||||||
(Some(id), Some(source), Some(updated_at)) => Some((id, source, updated_at)),
|
(Some(id), Some(source), Some(updated_at)) => Some((id, source, updated_at)),
|
||||||
_ => None,
|
_ => None,
|
||||||
@@ -225,37 +237,43 @@ pub async fn insert_interceptor_tx(
|
|||||||
op: &str,
|
op: &str,
|
||||||
phase: &str,
|
phase: &str,
|
||||||
script: &str,
|
script: &str,
|
||||||
|
timeout_ms: Option<i32>,
|
||||||
) -> Result<(), sqlx::Error> {
|
) -> Result<(), sqlx::Error> {
|
||||||
match owner {
|
match owner {
|
||||||
ScriptOwner::App(a) => {
|
ScriptOwner::App(a) => {
|
||||||
sqlx::query(
|
sqlx::query(
|
||||||
// The conflict arbiter includes `phase` to match the
|
// The conflict arbiter includes `phase` to match the
|
||||||
// per-(owner, service, op, phase) index (§9.4 M3).
|
// per-(owner, service, op, phase) index (§9.4 M3). A re-apply
|
||||||
"INSERT INTO interceptors (app_id, service, op, phase, interceptor_script) \
|
// that changes only the script or timeout updates in place.
|
||||||
VALUES ($1, $2, $3, $4, $5) \
|
"INSERT INTO interceptors (app_id, service, op, phase, interceptor_script, timeout_ms) \
|
||||||
|
VALUES ($1, $2, $3, $4, $5, $6) \
|
||||||
ON CONFLICT (app_id, service, op, phase) WHERE app_id IS NOT NULL \
|
ON CONFLICT (app_id, service, op, phase) WHERE app_id IS NOT NULL \
|
||||||
DO UPDATE SET interceptor_script = EXCLUDED.interceptor_script, updated_at = NOW()",
|
DO UPDATE SET interceptor_script = EXCLUDED.interceptor_script, \
|
||||||
|
timeout_ms = EXCLUDED.timeout_ms, updated_at = NOW()",
|
||||||
)
|
)
|
||||||
.bind(a.into_inner())
|
.bind(a.into_inner())
|
||||||
.bind(service)
|
.bind(service)
|
||||||
.bind(op)
|
.bind(op)
|
||||||
.bind(phase)
|
.bind(phase)
|
||||||
.bind(script)
|
.bind(script)
|
||||||
|
.bind(timeout_ms)
|
||||||
.execute(&mut **tx)
|
.execute(&mut **tx)
|
||||||
.await?;
|
.await?;
|
||||||
}
|
}
|
||||||
ScriptOwner::Group(g) => {
|
ScriptOwner::Group(g) => {
|
||||||
sqlx::query(
|
sqlx::query(
|
||||||
"INSERT INTO interceptors (group_id, service, op, phase, interceptor_script) \
|
"INSERT INTO interceptors (group_id, service, op, phase, interceptor_script, timeout_ms) \
|
||||||
VALUES ($1, $2, $3, $4, $5) \
|
VALUES ($1, $2, $3, $4, $5, $6) \
|
||||||
ON CONFLICT (group_id, service, op, phase) WHERE group_id IS NOT NULL \
|
ON CONFLICT (group_id, service, op, phase) WHERE group_id IS NOT NULL \
|
||||||
DO UPDATE SET interceptor_script = EXCLUDED.interceptor_script, updated_at = NOW()",
|
DO UPDATE SET interceptor_script = EXCLUDED.interceptor_script, \
|
||||||
|
timeout_ms = EXCLUDED.timeout_ms, updated_at = NOW()",
|
||||||
)
|
)
|
||||||
.bind(g.into_inner())
|
.bind(g.into_inner())
|
||||||
.bind(service)
|
.bind(service)
|
||||||
.bind(op)
|
.bind(op)
|
||||||
.bind(phase)
|
.bind(phase)
|
||||||
.bind(script)
|
.bind(script)
|
||||||
|
.bind(timeout_ms)
|
||||||
.execute(&mut **tx)
|
.execute(&mut **tx)
|
||||||
.await?;
|
.await?;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -33,6 +33,9 @@ impl InterceptorServiceImpl {
|
|||||||
/// (`cx.app_id`); only the `owner` is the sealing owner.
|
/// (`cx.app_id`); only the `owner` is the sealing owner.
|
||||||
fn map_entry(app_id: picloud_shared::AppId, sealed: SealedInterceptor) -> InterceptorEntry {
|
fn map_entry(app_id: picloud_shared::AppId, sealed: SealedInterceptor) -> InterceptorEntry {
|
||||||
let owner = sealed.sealing_owner();
|
let owner = sealed.sealing_owner();
|
||||||
|
// A negative/oversized DB value is impossible (CHECK > 0), but clamp
|
||||||
|
// defensively to `None` (→ instance default) rather than panic.
|
||||||
|
let timeout_ms = sealed.timeout_ms.and_then(|v| u32::try_from(v).ok());
|
||||||
match sealed.script {
|
match sealed.script {
|
||||||
Some((sid, source, updated_at)) => InterceptorEntry::Run(ResolvedInterceptor {
|
Some((sid, source, updated_at)) => InterceptorEntry::Run(ResolvedInterceptor {
|
||||||
script: Box::new(ResolvedScript {
|
script: Box::new(ResolvedScript {
|
||||||
@@ -43,7 +46,7 @@ impl InterceptorServiceImpl {
|
|||||||
updated_at,
|
updated_at,
|
||||||
name: sealed.script_name,
|
name: sealed.script_name,
|
||||||
}),
|
}),
|
||||||
timeout_ms: None,
|
timeout_ms,
|
||||||
}),
|
}),
|
||||||
None => InterceptorEntry::Dangling(sealed.script_name),
|
None => InterceptorEntry::Dangling(sealed.script_name),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -318,6 +318,7 @@ table: interceptors
|
|||||||
created_at: timestamp with time zone NOT NULL default=now()
|
created_at: timestamp with time zone NOT NULL default=now()
|
||||||
updated_at: timestamp with time zone NOT NULL default=now()
|
updated_at: timestamp with time zone NOT NULL default=now()
|
||||||
phase: text NOT NULL default='before'::text
|
phase: text NOT NULL default='before'::text
|
||||||
|
timeout_ms: integer NULL
|
||||||
|
|
||||||
table: kv_entries
|
table: kv_entries
|
||||||
app_id: uuid NOT NULL
|
app_id: uuid NOT NULL
|
||||||
@@ -954,6 +955,7 @@ constraints on groups:
|
|||||||
constraints on interceptors:
|
constraints on interceptors:
|
||||||
[CHECK] interceptors_owner_exactly_one: CHECK (((group_id IS NULL) <> (app_id IS NULL)))
|
[CHECK] interceptors_owner_exactly_one: CHECK (((group_id IS NULL) <> (app_id IS NULL)))
|
||||||
[CHECK] interceptors_phase_check: CHECK ((phase = ANY (ARRAY['before'::text, 'after'::text])))
|
[CHECK] interceptors_phase_check: CHECK ((phase = ANY (ARRAY['before'::text, 'after'::text])))
|
||||||
|
[CHECK] interceptors_timeout_ms_check: CHECK (((timeout_ms IS NULL) OR (timeout_ms > 0)))
|
||||||
[FOREIGN KEY] interceptors_app_id_fkey: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE
|
[FOREIGN KEY] interceptors_app_id_fkey: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE
|
||||||
[FOREIGN KEY] interceptors_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE
|
[FOREIGN KEY] interceptors_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE
|
||||||
[PRIMARY KEY] interceptors_pkey: PRIMARY KEY (id)
|
[PRIMARY KEY] interceptors_pkey: PRIMARY KEY (id)
|
||||||
@@ -1148,3 +1150,4 @@ constraints on workflows:
|
|||||||
0072: execution source workflow
|
0072: execution source workflow
|
||||||
0073: interceptors
|
0073: interceptors
|
||||||
0074: interceptor phase
|
0074: interceptor phase
|
||||||
|
0075: interceptor timeout
|
||||||
|
|||||||
@@ -263,7 +263,13 @@ pub fn build_bundle(manifest: &Manifest, base_dir: &Path) -> Result<Value> {
|
|||||||
.iter()
|
.iter()
|
||||||
.flat_map(|i| {
|
.flat_map(|i| {
|
||||||
i.ops.iter().map(move |op| {
|
i.ops.iter().map(move |op| {
|
||||||
json!({ "service": i.service, "op": op, "phase": i.phase, "script": i.script })
|
json!({
|
||||||
|
"service": i.service,
|
||||||
|
"op": op,
|
||||||
|
"phase": i.phase,
|
||||||
|
"script": i.script,
|
||||||
|
"timeout_ms": i.timeout_ms,
|
||||||
|
})
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
.collect::<Vec<_>>(),
|
.collect::<Vec<_>>(),
|
||||||
|
|||||||
@@ -88,6 +88,11 @@ pub struct ManifestInterceptor {
|
|||||||
/// cannot roll back the write).
|
/// cannot roll back the write).
|
||||||
#[serde(default = "default_interceptor_phase")]
|
#[serde(default = "default_interceptor_phase")]
|
||||||
pub phase: String,
|
pub phase: String,
|
||||||
|
/// §9.4 M5: per-hook wall-clock timeout in ms. Omit for the instance
|
||||||
|
/// default (`PICLOUD_INTERCEPTOR_TIMEOUT_MS`). Only ever tightened to the
|
||||||
|
/// caller's remaining budget — a hook cannot extend its caller's deadline.
|
||||||
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||||
|
pub timeout_ms: Option<u32>,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn default_interceptor_service() -> String {
|
fn default_interceptor_service() -> String {
|
||||||
|
|||||||
@@ -727,3 +727,76 @@ fn an_after_hook_sees_the_result_and_cannot_roll_back() {
|
|||||||
"the after-hook must see result==true and deny, but the delete must have persisted"
|
"the after-hook must see result==true and deny, but the delete must have persisted"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// --- §9.4 M5: per-interceptor timeout -------------------------------------
|
||||||
|
|
||||||
|
/// A guard that never returns — it must be interrupted by its per-hook timeout
|
||||||
|
/// rather than hanging the guarded write.
|
||||||
|
const SLOW_GUARD: &str = r#"
|
||||||
|
loop {}
|
||||||
|
"#;
|
||||||
|
|
||||||
|
/// A writer whose single guarded set triggers the runaway guard.
|
||||||
|
const SLOW_WRITER: &str = r#"
|
||||||
|
let denied = false;
|
||||||
|
try { kv::collection("c").set("k", 1); } catch(e) { denied = true; }
|
||||||
|
#{ denied: denied }
|
||||||
|
"#;
|
||||||
|
|
||||||
|
/// M5: a `timeout_ms` on the marker bounds the hook — a runaway `loop {}` guard
|
||||||
|
/// is interrupted and the op is DENIED (fail closed) within the budget, rather
|
||||||
|
/// than hanging the write path. The write must not persist.
|
||||||
|
#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
|
||||||
|
#[test]
|
||||||
|
fn a_runaway_interceptor_is_denied_by_its_timeout() {
|
||||||
|
let Some(fx) = common::fixture_or_skip() else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let env = common::admin_env(fx);
|
||||||
|
let app = common::unique_slug("to-app");
|
||||||
|
let _a = AppGuard::new(&env.url, &env.token, &app);
|
||||||
|
common::pic_as(&env)
|
||||||
|
.args(["apps", "create", &app])
|
||||||
|
.assert()
|
||||||
|
.success();
|
||||||
|
|
||||||
|
let dir = manifest_dir();
|
||||||
|
fs::write(dir.path().join("scripts/guard.rhai"), SLOW_GUARD).unwrap();
|
||||||
|
fs::write(dir.path().join("scripts/writer.rhai"), SLOW_WRITER).unwrap();
|
||||||
|
fs::write(
|
||||||
|
dir.path().join("picloud.toml"),
|
||||||
|
format!(
|
||||||
|
"[app]\nslug = \"{app}\"\nname = \"TO\"\n\n\
|
||||||
|
[[scripts]]\nname = \"guard\"\nfile = \"scripts/guard.rhai\"\n\n\
|
||||||
|
[[scripts]]\nname = \"writer\"\nfile = \"scripts/writer.rhai\"\n\n\
|
||||||
|
[[interceptors]]\nscript = \"guard\"\nops = [\"set\"]\ntimeout_ms = 100\n"
|
||||||
|
),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
common::pic_as(&env)
|
||||||
|
.args(["apply", "--file"])
|
||||||
|
.arg(dir.path().join("picloud.toml"))
|
||||||
|
.assert()
|
||||||
|
.success();
|
||||||
|
|
||||||
|
let body = invoke_body(&env, &app_script_id(&env, &app, "writer"));
|
||||||
|
assert_eq!(
|
||||||
|
body,
|
||||||
|
serde_json::json!({ "denied": true }),
|
||||||
|
"a runaway interceptor must be timed out and the op denied (fail closed)"
|
||||||
|
);
|
||||||
|
|
||||||
|
// The write must not have landed.
|
||||||
|
let k = String::from_utf8(
|
||||||
|
common::pic_as(&env)
|
||||||
|
.args(["kv", "get", "--app", &app, "--collection", "c", "k"])
|
||||||
|
.output()
|
||||||
|
.unwrap()
|
||||||
|
.stdout,
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
assert!(
|
||||||
|
k.trim() == "null" || k.trim() == "()" || k.trim().is_empty(),
|
||||||
|
"the timed-out write must NOT persist, got: {k}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user