diff --git a/crates/openshell-supervisor-middleware/src/http_hooks/pipeline.rs b/crates/openshell-supervisor-middleware/src/http_hooks/pipeline.rs index 82aaf13cac..45627cd7df 100644 --- a/crates/openshell-supervisor-middleware/src/http_hooks/pipeline.rs +++ b/crates/openshell-supervisor-middleware/src/http_hooks/pipeline.rs @@ -41,10 +41,10 @@ use openshell_core::proto::{ use crate::headers::{self, HeaderAuthority}; use crate::{ - ChainRunner, DescribedChainEntry, HttpHookVersion, MAX_MIDDLEWARE_CHAIN_TIMEOUT, - MAX_MIDDLEWARE_PAYLOAD_BYTES, MiddlewareDenial, MiddlewareDiagnosticPolicy, - MiddlewareInvocation, MiddlewareSessionPermit, NamespacedFinding, OnError, - TransformedBodyPolicy, middleware_denial_reason, safe_reason, + ChainRunner, DescribedChainEntry, HttpHookVersion, HttpRequestFailureKind, + MAX_MIDDLEWARE_CHAIN_TIMEOUT, MAX_MIDDLEWARE_PAYLOAD_BYTES, MiddlewareDenial, + MiddlewareDiagnosticPolicy, MiddlewareInvocation, MiddlewareSessionPermit, NamespacedFinding, + OnError, TransformedBodyPolicy, middleware_denial_reason, safe_reason, }; /// Largest normalized body chunk a stage sends or receives. @@ -197,6 +197,8 @@ pub enum HttpStageOutcome { /// Audit record for one stage and phase. Carries no service-provided text. #[derive(Debug, Clone, PartialEq, Eq)] pub struct HttpStageInvocation { + /// Failure cause retained without service text or status metadata. + pub failure_kind: Option, pub config_name: String, pub implementation: String, pub outcome: HttpStageOutcome, @@ -213,6 +215,8 @@ pub struct HttpStageInvocation { /// Findings, metadata, and invocations for one exchange. #[derive(Debug, Default)] pub struct HttpStageDiagnostics { + /// Terminal cause, including failures with no failed stage invocation. + pub failure_kind: Option, pub findings: Vec, pub metadata: BTreeMap>, pub invocations: Vec, @@ -221,6 +225,7 @@ pub struct HttpStageDiagnostics { impl HttpStageDiagnostics { /// Append `other`, keeping chain order. pub fn extend(&mut self, other: Self) { + self.failure_kind = other.failure_kind.or(self.failure_kind); self.findings.extend(other.findings); self.metadata.extend(other.metadata); self.invocations.extend(other.invocations); @@ -248,6 +253,7 @@ impl HttpStageDiagnostics { .iter_mut() .find(|(existing, _)| existing.name == invocation.config_name) { + existing.failure_kind = invocation.failure_kind.or(existing.failure_kind); existing.failed |= invocation.failed; existing.transformed |= invocation.transformed; if denied { @@ -258,6 +264,7 @@ impl HttpStageDiagnostics { } applied.push(( MiddlewareInvocation { + failure_kind: invocation.failure_kind, name: invocation.config_name.clone(), implementation: invocation.implementation.clone(), decision: if denied { @@ -286,6 +293,8 @@ impl HttpStageDiagnostics { /// The exchange failed closed or was denied. #[derive(Debug)] pub struct HttpMiddlewareFailure { + /// Cause preserved independently of sandbox failure codes and diagnostics. + pub kind: Option, /// Platform-owned reason: `middleware_denied:[:]` for a /// denial, `middleware_failed: ` for a failure, and /// `middleware_cancelled: ` when the input producer went away. @@ -1558,13 +1567,17 @@ async fn run_checkpoint( .ok() .flatten() .is_none_or(|previous| previous != body); - if replaced && let Some(reason) = transformed_body_denial(body_policy, &body) { + if replaced && let Some((reason, kind)) = transformed_body_denial(body_policy, &body) { return Err(TaskError::Failed(HttpMiddlewareFailure { + kind, reason, denial: None, failure: None, end_reason: MiddlewareSessionEndReason::PolicyDenial, - diagnostics: Box::default(), + diagnostics: Box::new(HttpStageDiagnostics { + failure_kind: kind, + ..Default::default() + }), })); } let (header_mutations, output_body_bytes) = start.expect("a started stage output"); @@ -2009,15 +2022,21 @@ async fn drive_buffered( } /// The body policy's verdict on a replaced body, or `None` when it passes. -fn transformed_body_denial(policy: TransformedBodyPolicy<'_>, body: &[u8]) -> Option { +fn transformed_body_denial( + policy: TransformedBodyPolicy<'_>, + body: &[u8], +) -> Option<(String, Option)> { let TransformedBodyPolicy::Reevaluate(validate) = policy else { return None; }; match validate(body) { - Ok(reason) => reason, - Err(error) => Some(format!( - "transformed_body_policy_evaluation_failed: {}", - safe_reason(&error.to_string()) + Ok(reason) => reason.map(|reason| (reason, None)), + Err(error) => Some(( + format!( + "transformed_body_policy_evaluation_failed: {}", + safe_reason(&error.to_string()) + ), + Some(HttpRequestFailureKind::PolicyEvaluation), )), } } @@ -2179,7 +2198,16 @@ async fn drive_stream( held.send_replace(false); } StageResult::OutputChunk(chunk) if started => { - if chunk.data.is_empty() || chunk.data.len() > chunk_limit { + if chunk.data.len() > chunk_limit { + return Err(entry_failure_with_code( + entry, + "stream_output_chunk_invalid", + HttpFailureCode::LifecycleViolation, + HttpRequestFailureKind::Capacity, + ) + .into()); + } + if chunk.data.is_empty() { return Err(entry_failure(entry, "stream_output_chunk_invalid").into()); } output_bytes = output_bytes.saturating_add(chunk.data.len()); @@ -2266,19 +2294,27 @@ pub(super) fn classify_result( /// Classify an RPC status. Every status fails the exchange closed; the reason /// is platform-owned for external services. fn status_failure(entry: &DescribedChainEntry, status: &tonic::Status) -> HttpMiddlewareFailure { - match status.code() { - tonic::Code::DeadlineExceeded => entry_failure(entry, "middleware_timeout"), - // The service does not implement the HTTP hook RPC its binding - // selected. - tonic::Code::Unimplemented => entry_failure(entry, "middleware_unimplemented"), - // The reason may carry an in-process service's status text, so the - // sandbox sees only that the middleware failed. - _ => entry_failure_with_code( - entry, - &diagnostic_policy(entry).error_reason(status), + let (reason, code) = match status.code() { + tonic::Code::DeadlineExceeded => { + ("middleware_timeout".to_string(), HttpFailureCode::Timeout) + } + // The service does not implement the HTTP hook RPC its binding selected. + tonic::Code::Unimplemented => ( + "middleware_unimplemented".to_string(), + HttpFailureCode::LifecycleViolation, + ), + // In-process status text stays in diagnostics, never the sandbox code. + _ => ( + diagnostic_policy(entry).error_reason(status), HttpFailureCode::Unavailable, ), - } + }; + entry_failure_with_code( + entry, + &reason, + code, + HttpRequestFailureKind::RemoteStatus(status.code()), + ) } fn diagnostic_policy(entry: &DescribedChainEntry) -> MiddlewareDiagnosticPolicy { @@ -2298,6 +2334,7 @@ fn mutation_failure( entry, &diagnostic_policy(entry).header_mutation_error_reason(error), HttpFailureCode::HeaderMutation(error.code()), + HttpRequestFailureKind::HeaderMutation, ) } @@ -2413,6 +2450,7 @@ pub(super) fn denial( .invocations .push(invocation(entry, HttpStageOutcome::Deny, reason_code)); HttpMiddlewareFailure { + kind: None, reason: middleware_denial_reason(&denial.config_name, denial.reason_code.as_deref()), denial: Some(denial), failure: None, @@ -2423,7 +2461,12 @@ pub(super) fn denial( /// The stage failed and the exchange fails closed. pub fn entry_failure(entry: &DescribedChainEntry, reason: &str) -> HttpMiddlewareFailure { - entry_failure_with_code(entry, reason, HttpFailureCode::of_reason(reason)) + entry_failure_with_code( + entry, + reason, + HttpFailureCode::of_reason(reason), + HttpRequestFailureKind::of_platform_reason(reason), + ) } /// The stage failed with `reason`, which the sandbox sees as `code`. @@ -2431,14 +2474,20 @@ fn entry_failure_with_code( entry: &DescribedChainEntry, reason: &str, code: HttpFailureCode, + kind: HttpRequestFailureKind, ) -> HttpMiddlewareFailure { - let mut diagnostics = HttpStageDiagnostics::default(); + let mut diagnostics = HttpStageDiagnostics { + failure_kind: Some(kind), + ..Default::default() + }; diagnostics.invocations.push(HttpStageInvocation { + failure_kind: Some(kind), failed: true, failure_reason: Some(reason.to_string()), ..invocation(entry, HttpStageOutcome::FailClosed, None) }); HttpMiddlewareFailure { + kind: Some(kind), reason: format!("middleware_failed: {reason}"), denial: None, failure: Some(Box::new(MiddlewareFailureDetail { @@ -2452,22 +2501,32 @@ fn entry_failure_with_code( /// The exchange failed for a reason no single stage owns. pub fn platform_failure(reason: &str) -> HttpMiddlewareFailure { + let kind = HttpRequestFailureKind::InvalidResult; HttpMiddlewareFailure { + kind: Some(kind), reason: format!("middleware_failed: {reason}"), denial: None, failure: None, end_reason: MiddlewareSessionEndReason::MiddlewareFailure, - diagnostics: Box::default(), + diagnostics: Box::new(HttpStageDiagnostics { + failure_kind: Some(kind), + ..Default::default() + }), } } fn cancelled(reason: &str) -> HttpMiddlewareFailure { + let kind = HttpRequestFailureKind::Cancelled; HttpMiddlewareFailure { + kind: Some(kind), reason: format!("middleware_cancelled: {reason}"), denial: None, failure: None, end_reason: MiddlewareSessionEndReason::Cancellation, - diagnostics: Box::default(), + diagnostics: Box::new(HttpStageDiagnostics { + failure_kind: Some(kind), + ..Default::default() + }), } } @@ -2477,6 +2536,7 @@ pub fn invocation( reason_code: Option, ) -> HttpStageInvocation { HttpStageInvocation { + failure_kind: None, config_name: entry.entry.name.clone(), implementation: entry.entry.implementation.clone(), outcome, @@ -2491,6 +2551,7 @@ pub fn invocation( pub fn fail_open_invocation(entry: &DescribedChainEntry, reason: &str) -> HttpStageInvocation { HttpStageInvocation { + failure_kind: Some(HttpRequestFailureKind::of_platform_reason(reason)), failed: true, failure_reason: Some(reason.to_string()), ..invocation(entry, HttpStageOutcome::FailOpen, None) diff --git a/crates/openshell-supervisor-middleware/src/http_hooks/request.rs b/crates/openshell-supervisor-middleware/src/http_hooks/request.rs index 1b945a733c..2eb1beb724 100644 --- a/crates/openshell-supervisor-middleware/src/http_hooks/request.rs +++ b/crates/openshell-supervisor-middleware/src/http_hooks/request.rs @@ -267,7 +267,13 @@ fn capacity_exhausted(headers: Vec) -> HttpRequestPreflightOutcome { allowed: false, reason: "middleware_failed: session_capacity_exhausted".into(), session_capacity_exhausted: true, - ..outcome(headers, HttpStageDiagnostics::default()) + ..outcome( + headers, + HttpStageDiagnostics { + failure_kind: Some(crate::HttpRequestFailureKind::Capacity), + ..Default::default() + }, + ) } } diff --git a/crates/openshell-supervisor-middleware/src/http_hooks/tests.rs b/crates/openshell-supervisor-middleware/src/http_hooks/tests.rs index 5b2f652398..dbb639b6da 100644 --- a/crates/openshell-supervisor-middleware/src/http_hooks/tests.rs +++ b/crates/openshell-supervisor-middleware/src/http_hooks/tests.rs @@ -150,6 +150,7 @@ enum Preflight { Inspect(HttpBodyMode), Deny(&'static str), Fail(tonic::Code), + Stall, } /// What a scripted stage does with the body it selected. @@ -172,6 +173,7 @@ struct ScriptedStage { inspect_content_type: Option<&'static str>, output_content_type: Option<&'static str>, output_after_input: bool, + output_chunk_size: Option, ended: Arc>>, /// Heads this stage received in `Begin`. begun: Arc>>>, @@ -192,6 +194,7 @@ impl ScriptedStage { inspect_content_type: None, output_content_type: None, output_after_input: false, + output_chunk_size: None, ended: Arc::default(), begun: Arc::default(), preflights: Arc::default(), @@ -274,6 +277,7 @@ impl InProcessMiddleware for ScriptedStage { } let inspect_content_type = self.inspect_content_type; let output_after_input = self.output_after_input; + let output_chunk_size = self.output_chunk_size; let ended = Arc::clone(&self.ended); let begun = Arc::clone(&self.begun); let preflights = Arc::clone(&self.preflights); @@ -345,6 +349,7 @@ impl InProcessMiddleware for ScriptedStage { })) } Preflight::Fail(code) => Err(tonic::Status::new(*code, "scripted")), + Preflight::Stall => futures::future::pending().await, } } Some(http_request_hook_event::Event::Begin(started)) => { @@ -393,7 +398,10 @@ impl InProcessMiddleware for ScriptedStage { } result(http_request_hook_result::Result::OutputChunk( HttpOutputChunk { - data: chunk.data.to_ascii_uppercase(), + data: output_chunk_size.map_or_else( + || chunk.data.to_ascii_uppercase(), + |size| vec![b'x'; size], + ), }, )) } @@ -1038,13 +1046,27 @@ async fn failures_name_the_failed_stage_and_a_platform_code() { HttpFailureCode::Unavailable, ), ] { - let (runner, described) = - scripted_chain(vec![scripted("example/a", preflight, Body::Unchanged)]).await; + let Preflight::Fail(status) = preflight else { + unreachable!() + }; + let (runner, described) = scripted_chain(vec![scripted( + "example/a", + Preflight::Fail(status), + Body::Unchanged, + )]) + .await; let preflight = runner .preflight_described_http_request(described, request_input(b"hello")) .await .expect("preflight"); assert!(!preflight.allowed, "{case}"); + let kind = Some(crate::HttpRequestFailureKind::RemoteStatus(status)); + assert_eq!(preflight.diagnostics.failure_kind, kind, "{case}"); + assert_eq!( + preflight.diagnostics.applied()[0].failure_kind, + kind, + "{case}" + ); assert_eq!( preflight.failure, Some(MiddlewareFailureDetail { @@ -1067,6 +1089,8 @@ async fn failures_name_the_failed_stage_and_a_platform_code() { .expect("preflight"); assert!(denied.denial.is_some()); assert!(denied.failure.is_none()); + assert_eq!(denied.diagnostics.failure_kind, None); + assert_eq!(denied.diagnostics.applied()[0].failure_kind, None); } #[tokio::test] @@ -1696,3 +1720,152 @@ async fn stages_cannot_change_headers_that_reference_credentials() { ); } } + +#[tokio::test(start_paused = true)] +async fn local_deadlines_preserve_their_scope() { + use crate::HttpRequestFailureKind; + for (timeout, expected) in [ + ( + std::time::Duration::from_millis(5), + HttpRequestFailureKind::StageTimeout, + ), + ( + MAX_MIDDLEWARE_CHAIN_TIMEOUT * 2, + HttpRequestFailureKind::ChainTimeout, + ), + ] { + let (runner, mut described) = scripted_chain(vec![scripted( + "example/a", + Preflight::Stall, + Body::Unchanged, + )]) + .await; + described[0].timeout = timeout; + let outcome = runner + .preflight_described_http_request(described, request_input(b"hello")) + .await + .expect("preflight"); + assert!(!outcome.allowed); + assert_eq!(outcome.diagnostics.failure_kind, Some(expected)); + assert_eq!( + outcome.diagnostics.applied()[0].failure_kind, + Some(expected) + ); + assert_eq!( + outcome.failure.expect("sandbox detail").code, + HttpFailureCode::Timeout + ); + } +} + +#[tokio::test] +async fn stage_causes_survive_diagnostic_merging_and_compaction() { + use crate::HttpRequestFailureKind; + let (_, described) = scripted_chain(vec![scripted( + "example/a", + Preflight::Continue, + Body::Unchanged, + )]) + .await; + for (reason, expected) in [ + ( + "middleware_body_timeout", + HttpRequestFailureKind::WholeBodyTimeout, + ), + ( + "middleware_stall_timeout", + HttpRequestFailureKind::StreamIdleTimeout, + ), + ( + "diagnostics_reason_over_capacity", + HttpRequestFailureKind::Capacity, + ), + ( + "preflight_result_unknown", + HttpRequestFailureKind::InvalidResult, + ), + ("binding_not_described", HttpRequestFailureKind::Unavailable), + ("middleware_stream_closed", HttpRequestFailureKind::Io), + ] { + let failure = entry_failure(&described[0], reason); + let mut diagnostics = crate::HttpStageDiagnostics::default(); + diagnostics.invocations.push(super::pipeline::invocation( + &described[0], + HttpStageOutcome::Duplex, + None, + )); + diagnostics.extend(*failure.diagnostics); + assert_eq!(failure.kind, Some(expected)); + assert_eq!(diagnostics.failure_kind, Some(expected)); + let applied = diagnostics.applied(); + assert_eq!(applied.len(), 1); + assert!(applied[0].failed); + assert_eq!(applied[0].failure_kind, Some(expected)); + } +} + +#[tokio::test] +async fn duplex_chunk_capacity_is_distinct_from_invalid_output() { + use crate::HttpRequestFailureKind; + for (size, expected) in [ + (1024, None), + (1025, Some(HttpRequestFailureKind::Capacity)), + (0, Some(HttpRequestFailureKind::InvalidResult)), + ] { + let stage = Arc::new(ScriptedStage { + output_chunk_size: Some(size), + ..ScriptedStage::new( + "example/a", + Preflight::Inspect(HttpBodyMode::Duplex), + Body::Uppercase, + ) + }); + let (runner, described) = scripted_chain(vec![stage]).await; + let preflight = runner + .preflight_described_http_request(described, request_input(b"a")) + .await + .expect("preflight"); + let (input, inputs) = mpsc::channel(4); + input + .send(HttpBodyInput::Chunk(b"a".to_vec())) + .await + .unwrap(); + input + .send(HttpBodyInput::End { + trailers: Vec::new(), + }) + .await + .unwrap(); + drop(input); + let (output, mut outputs) = mpsc::channel(4); + let run = preflight.pipeline.unwrap().run_with_body_policy( + inputs, + output, + TransformedBodyPolicy::NotPolicyRelevant, + ); + let drain = async { while outputs.recv().await.is_some() {} }; + let (finish, ()) = tokio::join!(run, drain); + match expected { + None => assert_eq!( + finish.expect("exact chunk limit").diagnostics.failure_kind, + None + ), + Some(expected) => { + let failure = finish.expect_err("invalid chunk"); + assert_eq!(failure.kind, Some(expected)); + assert_eq!( + failure.diagnostics.applied()[0].failure_kind, + Some(expected) + ); + assert_eq!( + failure.reason, + "middleware_failed: stream_output_chunk_invalid" + ); + assert_eq!( + failure.failure.unwrap().code, + HttpFailureCode::LifecycleViolation + ); + } + } + } +} diff --git a/crates/openshell-supervisor-middleware/src/lib.rs b/crates/openshell-supervisor-middleware/src/lib.rs index d8edfb9cff..7dae05d16b 100644 --- a/crates/openshell-supervisor-middleware/src/lib.rs +++ b/crates/openshell-supervisor-middleware/src/lib.rs @@ -8,6 +8,8 @@ mod diagnostics; pub mod headers; mod http_hooks; mod remote; +mod request_failure; +pub use request_failure::HttpRequestFailureKind; mod response; mod websocket; @@ -679,6 +681,8 @@ pub struct HttpRequestInput { #[derive(Debug, Clone)] pub struct ChainOutcome { + /// Failure cause, independent of diagnostic text and explicit denials. + pub failure_kind: Option, pub allowed: bool, pub reason: String, pub body: Vec, @@ -710,6 +714,8 @@ pub struct NamespacedFinding { #[derive(Debug, Clone, PartialEq, Eq)] pub struct MiddlewareInvocation { + /// Failure cause, independent of diagnostic text and explicit denials. + pub failure_kind: Option, pub name: String, pub implementation: String, pub decision: Decision, @@ -732,11 +738,13 @@ enum OnErrorAction { fn apply_on_error( entry: &DescribedChainEntry, reason: &str, + failure_kind: HttpRequestFailureKind, applied: &mut Vec, ) -> OnErrorAction { match entry.entry.on_error { OnError::FailOpen => { applied.push(MiddlewareInvocation { + failure_kind: Some(failure_kind), name: entry.entry.name.clone(), implementation: entry.entry.implementation.clone(), decision: Decision::Allow, @@ -747,6 +755,7 @@ fn apply_on_error( } OnError::FailClosed => { applied.push(MiddlewareInvocation { + failure_kind: Some(failure_kind), name: entry.entry.name.clone(), implementation: entry.entry.implementation.clone(), decision: Decision::Deny, @@ -2115,6 +2124,7 @@ impl ChainRunner { .find(|entry| entry.http_hook_version() == Some(HttpHookVersion::Current)) { return Ok(ChainOutcome { + failure_kind: Some(HttpRequestFailureKind::InvalidResult), allowed: false, reason: format!("middleware_failed: {MIDDLEWARE_HOOK_VERSIONS_MIXED}"), body: input.body, @@ -2122,6 +2132,7 @@ impl ChainRunner { findings: Vec::new(), metadata: BTreeMap::new(), applied: vec![MiddlewareInvocation { + failure_kind: Some(HttpRequestFailureKind::InvalidResult), name: entry.entry.name.clone(), implementation: entry.entry.implementation.clone(), decision: Decision::Deny, @@ -2180,10 +2191,18 @@ impl ChainRunner { for entry in entries { let Some(_binding) = entry.binding.as_ref() else { - match apply_on_error(entry, "binding_not_described", &mut applied) { + match apply_on_error( + entry, + "binding_not_described", + HttpRequestFailureKind::of_platform_reason("binding_not_described"), + &mut applied, + ) { OnErrorAction::FailOpen => continue, OnErrorAction::FailClosed(reason) => { return Ok(ChainOutcome { + failure_kind: applied + .last() + .and_then(|invocation| invocation.failure_kind), allowed: false, reason, body, @@ -2197,10 +2216,18 @@ impl ChainRunner { } }; if body.len() > entry.max_payload_bytes { - match apply_on_error(entry, "request_body_over_capacity", &mut applied) { + match apply_on_error( + entry, + "request_body_over_capacity", + HttpRequestFailureKind::of_platform_reason("request_body_over_capacity"), + &mut applied, + ) { OnErrorAction::FailOpen => continue, OnErrorAction::FailClosed(reason) => { return Ok(ChainOutcome { + failure_kind: applied + .last() + .and_then(|invocation| invocation.failure_kind), allowed: false, reason, body, @@ -2223,10 +2250,18 @@ impl ChainRunner { &entry.entry.implementation, ); if let Err(reason) = validate_request_view(request) { - match apply_on_error(entry, reason, &mut applied) { + match apply_on_error( + entry, + reason, + HttpRequestFailureKind::of_platform_reason(reason), + &mut applied, + ) { OnErrorAction::FailOpen => continue, OnErrorAction::FailClosed(reason) => { return Ok(ChainOutcome { + failure_kind: applied + .last() + .and_then(|invocation| invocation.failure_kind), allowed: false, reason, body, @@ -2244,10 +2279,18 @@ impl ChainRunner { }; let remaining = chain_deadline.saturating_duration_since(tokio::time::Instant::now()); if remaining.is_zero() { - match apply_on_error(entry, "middleware_chain_timeout", &mut applied) { + match apply_on_error( + entry, + "middleware_chain_timeout", + HttpRequestFailureKind::of_platform_reason("middleware_chain_timeout"), + &mut applied, + ) { OnErrorAction::FailOpen => continue, OnErrorAction::FailClosed(reason) => { return Ok(ChainOutcome { + failure_kind: applied + .last() + .and_then(|invocation| invocation.failure_kind), allowed: false, reason, body, @@ -2260,24 +2303,41 @@ impl ChainRunner { } } } - let mut result = match call_with_timeout( + let mut result = match tokio::time::timeout( entry.timeout.min(remaining), - "EvaluateHttpRequest", service.service.evaluate_http_request(request), ) .await { - Ok(result) => result.into_inner(), - Err(err) => { + Ok(Ok(result)) => result.into_inner(), + error => { + let (err, failure_kind) = match error { + Ok(Err(err)) => { + let kind = HttpRequestFailureKind::RemoteStatus(err.code()); + (err, kind) + } + Err(_) => ( + tonic::Status::deadline_exceeded("EvaluateHttpRequest timed out"), + if remaining <= entry.timeout { + HttpRequestFailureKind::ChainTimeout + } else { + HttpRequestFailureKind::StageTimeout + }, + ), + Ok(Ok(_)) => unreachable!("successful result handled above"), + }; let reason = if err.code() == tonic::Code::DeadlineExceeded { "middleware_timeout".to_string() } else { service.diagnostic_policy.error_reason(&err) }; - match apply_on_error(entry, &reason, &mut applied) { + match apply_on_error(entry, &reason, failure_kind, &mut applied) { OnErrorAction::FailOpen => continue, OnErrorAction::FailClosed(reason) => { return Ok(ChainOutcome { + failure_kind: applied + .last() + .and_then(|invocation| invocation.failure_kind), allowed: false, reason, body, @@ -2293,10 +2353,18 @@ impl ChainRunner { }; if let Err(reason) = validate_response_envelope(&result) { - match apply_on_error(entry, reason, &mut applied) { + match apply_on_error( + entry, + reason, + HttpRequestFailureKind::of_platform_reason(reason), + &mut applied, + ) { OnErrorAction::FailOpen => continue, OnErrorAction::FailClosed(reason) => { return Ok(ChainOutcome { + failure_kind: applied + .last() + .and_then(|invocation| invocation.failure_kind), allowed: false, reason, body, @@ -2317,10 +2385,18 @@ impl ChainRunner { let decision = match Decision::try_from(result.decision) { Ok(decision @ (Decision::Allow | Decision::Deny)) => decision, Ok(Decision::Unspecified) | Err(_) => { - match apply_on_error(entry, "invalid_response_decision", &mut applied) { + match apply_on_error( + entry, + "invalid_response_decision", + HttpRequestFailureKind::of_platform_reason("invalid_response_decision"), + &mut applied, + ) { OnErrorAction::FailOpen => continue, OnErrorAction::FailClosed(reason) => { return Ok(ChainOutcome { + failure_kind: applied + .last() + .and_then(|invocation| invocation.failure_kind), allowed: false, reason, body, @@ -2355,6 +2431,7 @@ impl ChainRunner { ); } applied.push(MiddlewareInvocation { + failure_kind: None, name: entry.entry.name.clone(), implementation: entry.entry.implementation.clone(), decision, @@ -2362,6 +2439,7 @@ impl ChainRunner { failed: false, }); return Ok(ChainOutcome { + failure_kind: None, allowed: false, reason: middleware_denial_reason( &denial.config_name, @@ -2377,10 +2455,18 @@ impl ChainRunner { } if result.has_body && result.body.len() > entry.max_payload_bytes { - match apply_on_error(entry, "response_body_over_capacity", &mut applied) { + match apply_on_error( + entry, + "response_body_over_capacity", + HttpRequestFailureKind::of_platform_reason("response_body_over_capacity"), + &mut applied, + ) { OnErrorAction::FailOpen => continue, OnErrorAction::FailClosed(reason) => { return Ok(ChainOutcome { + failure_kind: applied + .last() + .and_then(|invocation| invocation.failure_kind), allowed: false, reason, body, @@ -2411,10 +2497,18 @@ impl ChainRunner { let reason = service .diagnostic_policy .header_mutation_error_reason(&error); - match apply_on_error(entry, &reason, &mut applied) { + match apply_on_error( + entry, + &reason, + HttpRequestFailureKind::HeaderMutation, + &mut applied, + ) { OnErrorAction::FailOpen => continue, OnErrorAction::FailClosed(reason) => { return Ok(ChainOutcome { + failure_kind: applied + .last() + .and_then(|invocation| invocation.failure_kind), allowed: false, reason, body, @@ -2454,6 +2548,7 @@ impl ChainRunner { ); } applied.push(MiddlewareInvocation { + failure_kind: None, name: entry.entry.name.clone(), implementation: entry.entry.implementation.clone(), decision, @@ -2468,15 +2563,19 @@ impl ChainRunner { if body_transformed && let TransformedBodyPolicy::Reevaluate(validate) = transformed_body_policy { - let denied = match validate(&body) { - Ok(reason) => reason, - Err(error) => Some(format!( - "transformed_body_policy_evaluation_failed: {}", - safe_reason(&error.to_string()) - )), + let (denied, failure_kind) = match validate(&body) { + Ok(reason) => (reason, None), + Err(error) => ( + Some(format!( + "transformed_body_policy_evaluation_failed: {}", + safe_reason(&error.to_string()) + )), + Some(HttpRequestFailureKind::PolicyEvaluation), + ), }; if let Some(reason) = denied { return Ok(ChainOutcome { + failure_kind, allowed: false, reason, body, @@ -2491,6 +2590,7 @@ impl ChainRunner { } Ok(ChainOutcome { + failure_kind: None, allowed: true, reason: String::new(), body, @@ -3171,6 +3271,10 @@ mod tests { .await .expect("evaluate"); assert!(!outcome.allowed); + assert_eq!( + outcome.failure_kind, + Some(HttpRequestFailureKind::Unavailable) + ); assert!(outcome.reason.starts_with("middleware_failed:")); } @@ -3608,6 +3712,10 @@ mod tests { .expect("policy evaluator failure should be a chain outcome"); assert!(!outcome.allowed); + assert_eq!( + outcome.failure_kind, + Some(HttpRequestFailureKind::PolicyEvaluation) + ); assert!( outcome .reason @@ -4673,6 +4781,10 @@ mod tests { .await .expect("operator timeout outcome"); assert!(!outcome.allowed); + assert_eq!( + outcome.failure_kind, + Some(HttpRequestFailureKind::StageTimeout) + ); assert_eq!(outcome.reason, "middleware_failed: middleware_timeout"); } diff --git a/crates/openshell-supervisor-middleware/src/request_failure.rs b/crates/openshell-supervisor-middleware/src/request_failure.rs new file mode 100644 index 0000000000..dd363714e3 --- /dev/null +++ b/crates/openshell-supervisor-middleware/src/request_failure.rs @@ -0,0 +1,103 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Platform-owned failure causes, independent of service diagnostic text. + +/// Retains why HTTP request processing failed across middleware and relay layers. +/// +/// Explicit policy denials have no failure kind. Remote status codes are retained +/// without status messages or metadata, which may contain workload data. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum HttpRequestFailureKind { + StageTimeout, + ChainTimeout, + RequestBodyTimeout, + WholeBodyTimeout, + StreamIdleTimeout, + RemoteStatus(tonic::Code), + Unavailable, + InvalidResult, + Capacity, + HeaderMutation, + Io, + Cancelled, + PolicyEvaluation, +} + +impl HttpRequestFailureKind { + pub(crate) fn of_platform_reason(reason: &str) -> Self { + match reason { + "middleware_timeout" => Self::StageTimeout, + "middleware_chain_timeout" => Self::ChainTimeout, + "middleware_body_timeout" => Self::WholeBodyTimeout, + "middleware_stall_timeout" => Self::StreamIdleTimeout, + "binding_not_described" => Self::Unavailable, + "head_input_over_capacity" + | "buffered_input_over_capacity" + | "buffered_output_over_capacity" + | "request_output_over_capacity" + | "session_capacity_exhausted" + | "diagnostics_reason_over_capacity" + | "diagnostics_findings_over_capacity" + | "diagnostics_metadata_over_capacity" + | "request_body_over_capacity" + | "request_config_over_capacity" + | "request_context_over_capacity" + | "request_target_over_capacity" + | "request_header_count_over_capacity" + | "request_header_bytes_over_capacity" + | "response_body_over_capacity" + | "response_reason_over_capacity" + | "header_mutation_count_over_capacity" + | "header_mutation_bytes_over_capacity" + | "response_findings_over_capacity" + | "response_finding_over_capacity" + | "response_metadata_count_over_capacity" + | "response_metadata_bytes_over_capacity" + | "response_envelope_over_capacity" => Self::Capacity, + "middleware_stream_closed" | "middleware_result_stream_closed" => Self::Io, + _ => Self::InvalidResult, + } + } + + /// Returns a fixed category safe for operator diagnostics and span attributes. + #[must_use] + pub const fn category(self) -> &'static str { + match self { + Self::StageTimeout + | Self::ChainTimeout + | Self::RequestBodyTimeout + | Self::WholeBodyTimeout + | Self::StreamIdleTimeout => "timeout", + Self::RemoteStatus(_) => "remote_status", + Self::Unavailable => "unavailable", + Self::InvalidResult => "protocol", + Self::Capacity => "capacity", + Self::HeaderMutation => "header_mutation", + Self::Io => "io", + Self::Cancelled => "cancelled", + Self::PolicyEvaluation => "policy_evaluation", + } + } + + /// Identifies the platform deadline; a remote `DEADLINE_EXCEEDED` is a status. + #[must_use] + pub const fn deadline_scope(self) -> Option<&'static str> { + match self { + Self::StageTimeout => Some("stage"), + Self::ChainTimeout => Some("chain"), + Self::RequestBodyTimeout => Some("request_body"), + Self::WholeBodyTimeout => Some("whole_body"), + Self::StreamIdleTimeout => Some("stream_idle"), + _ => None, + } + } +} + +impl std::fmt::Display for HttpRequestFailureKind { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(self.category()) + } +} + +impl std::error::Error for HttpRequestFailureKind {} diff --git a/crates/openshell-supervisor-network/src/l7/middleware.rs b/crates/openshell-supervisor-network/src/l7/middleware.rs index d44120b9dd..e9f229c0ba 100644 --- a/crates/openshell-supervisor-network/src/l7/middleware.rs +++ b/crates/openshell-supervisor-network/src/l7/middleware.rs @@ -14,8 +14,8 @@ use openshell_ocsf::{ }; use openshell_supervisor_middleware::{ ChainHttpHookVersion, HttpBodyInput, HttpBodyOutput, HttpFailureCode, HttpMiddlewareFailure, - HttpPipelineFinish, HttpRequestPipeline, HttpRequestPreflightInput, HttpStageDiagnostics, - MiddlewareFailureDetail, + HttpPipelineFinish, HttpRequestFailureKind, HttpRequestPipeline, HttpRequestPreflightInput, + HttpStageDiagnostics, MiddlewareFailureDetail, }; use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; @@ -1071,6 +1071,7 @@ async fn apply_pipeline_middleware_chain, events: RequestEvents, @@ -1481,11 +1492,13 @@ impl RequestBodyStream { if failure.end_reason == MiddlewareSessionEndReason::Cancellation && upstream_failed.load(Ordering::Acquire) => { + let mut diagnostics = *failure.diagnostics; + diagnostics.failure_kind = Some(HttpRequestFailureKind::Io); self.events.emit( false, "middleware_cancelled: upstream_write_failed", None, - *failure.diagnostics, + diagnostics, ); return Err(miette!("upstream stopped accepting the request body")); } @@ -1493,11 +1506,15 @@ impl RequestBodyStream { }; let failure = match &error { BodyRunError::Middleware(failure) => crate::l7::rest::LiveRequestFailure::Middleware { + kind: failure.kind, + delivery: self.delivery.state(), reason: failure.reason.clone(), denial: failure.denial.clone(), failure: failure.failure.as_deref().cloned(), }, - BodyRunError::ClientTimeout => crate::l7::rest::LiveRequestFailure::ClientTimeout, + BodyRunError::ClientTimeout => crate::l7::rest::LiveRequestFailure::ClientTimeout { + delivery: self.delivery.state(), + }, BodyRunError::Client(_) => { return self.events.body_failed(error).map(|_| unreachable!()); } @@ -1512,7 +1529,10 @@ impl RequestBodyStream { false, &format!("middleware_cancelled: {reason}"), None, - HttpStageDiagnostics::default(), + HttpStageDiagnostics { + failure_kind: Some(HttpRequestFailureKind::Cancelled), + ..Default::default() + }, ); } } @@ -1528,7 +1548,6 @@ pub(super) fn request_middleware_events( denial: Option<&openshell_supervisor_middleware::MiddlewareDenial>, diagnostics: &HttpStageDiagnostics, ) -> Vec { - let applied = diagnostics.applied(); let req = crate::l7::provider::L7Request { action: action.to_string(), target: target.to_string(), @@ -1539,19 +1558,29 @@ pub(super) fn request_middleware_events( middleware_events( ctx, &req, - &openshell_supervisor_middleware::ChainOutcome { - allowed, - reason: reason.to_string(), - body: Vec::new(), - header_mutations: Vec::new(), - findings: diagnostics.findings.clone(), - metadata: diagnostics.metadata.clone(), - applied, - denial: denial.cloned(), - }, + &request_middleware_outcome(allowed, reason, denial, diagnostics), ) } +fn request_middleware_outcome( + allowed: bool, + reason: &str, + denial: Option<&openshell_supervisor_middleware::MiddlewareDenial>, + diagnostics: &HttpStageDiagnostics, +) -> openshell_supervisor_middleware::ChainOutcome { + openshell_supervisor_middleware::ChainOutcome { + failure_kind: diagnostics.failure_kind, + allowed, + reason: reason.to_string(), + body: Vec::new(), + header_mutations: Vec::new(), + findings: diagnostics.findings.clone(), + metadata: diagnostics.metadata.clone(), + applied: diagnostics.applied(), + denial: denial.cloned(), + } +} + /// The sandbox stopped sending a request body that DUPLEX request /// middleware was processing; the request was aborted. pub(super) fn request_client_timeout_event( @@ -1594,7 +1623,10 @@ pub(super) fn mixed_protocol_events( target: &str, chain: &[openshell_supervisor_middleware::DescribedChainEntry], ) -> Vec { - let mut diagnostics = HttpStageDiagnostics::default(); + let mut diagnostics = HttpStageDiagnostics { + failure_kind: Some(HttpRequestFailureKind::InvalidResult), + ..Default::default() + }; if let Some(entry) = chain.iter().find(|entry| { entry.http_hook_version() == Some(openshell_supervisor_middleware::HttpHookVersion::Current) }) { @@ -1603,6 +1635,7 @@ pub(super) fn mixed_protocol_events( .push(openshell_supervisor_middleware::HttpStageInvocation { config_name: entry.config_name().to_string(), implementation: entry.implementation().to_string(), + failure_kind: Some(HttpRequestFailureKind::InvalidResult), outcome: openshell_supervisor_middleware::HttpStageOutcome::FailClosed, input_bytes: 0, output_bytes: None, @@ -2701,3 +2734,39 @@ mod tests { } } } + +#[cfg(test)] +mod failure_cause_tests { + use super::*; + #[test] + fn request_terminal_cause_survives_successful_stage_records() { + for kind in [ + HttpRequestFailureKind::RequestBodyTimeout, + HttpRequestFailureKind::Cancelled, + ] { + let diagnostics = HttpStageDiagnostics { + failure_kind: Some(kind), + invocations: vec![openshell_supervisor_middleware::HttpStageInvocation { + failure_kind: None, + config_name: "guard".into(), + implementation: "test/guard".into(), + outcome: openshell_supervisor_middleware::HttpStageOutcome::Duplex, + input_bytes: 0, + output_bytes: None, + transformed: false, + failed: false, + reason_code: None, + failure_reason: None, + }], + ..Default::default() + }; + let outcome = + request_middleware_outcome(false, "unchanged diagnostic", None, &diagnostics); + assert_eq!(outcome.failure_kind, Some(kind)); + assert!(!outcome.allowed); + assert!(outcome.denial.is_none()); + assert!(!outcome.applied[0].failed); + assert_eq!(outcome.applied[0].failure_kind, None); + } + } +} diff --git a/crates/openshell-supervisor-network/src/l7/mod.rs b/crates/openshell-supervisor-network/src/l7/mod.rs index 3f26114998..4e753cc02e 100644 --- a/crates/openshell-supervisor-network/src/l7/mod.rs +++ b/crates/openshell-supervisor-network/src/l7/mod.rs @@ -16,6 +16,7 @@ pub(crate) mod middleware; pub mod path; pub mod provider; pub mod relay; +pub(crate) mod request_delivery; pub mod rest; pub mod tls; pub(crate) mod token_grant_injection; diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index 9be6870910..6199a99ee7 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -957,7 +957,7 @@ async fn reject_live_request_body( .await?; } } - crate::l7::rest::LiveRequestFailure::ClientTimeout => { + crate::l7::rest::LiveRequestFailure::ClientTimeout { .. } => { crate::l7::rest::send_request_timeout_response( &rejected, &ctx.policy_name, @@ -7974,6 +7974,7 @@ network_policies: body_length: crate::l7::provider::BodyLength::None, }; let outcome = ChainOutcome { + failure_kind: None, allowed: true, reason: String::new(), // The transformed body still holds the raw secret; emission must never @@ -7992,6 +7993,7 @@ network_policies: }], metadata: BTreeMap::new(), applied: vec![MiddlewareInvocation { + failure_kind: None, name: "regex-redactor".into(), implementation: "openshell/regex".into(), decision: openshell_core::proto::Decision::Allow, @@ -8067,6 +8069,7 @@ network_policies: ); let denied_outcome = ChainOutcome { + failure_kind: None, allowed: false, reason: "middleware_denied:content-guard:content_match".into(), body: Vec::new(), @@ -8074,6 +8077,7 @@ network_policies: findings: Vec::new(), metadata: BTreeMap::new(), applied: vec![MiddlewareInvocation { + failure_kind: None, name: "content-guard".into(), implementation: "example/content-guard".into(), decision: openshell_core::proto::Decision::Deny, diff --git a/crates/openshell-supervisor-network/src/l7/request_delivery.rs b/crates/openshell-supervisor-network/src/l7/request_delivery.rs new file mode 100644 index 0000000000..1af7feb562 --- /dev/null +++ b/crates/openshell-supervisor-network/src/l7/request_delivery.rs @@ -0,0 +1,44 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Observes local upstream writes without deciding whether a request is allowed. + +use std::sync::{ + Arc, + atomic::{AtomicU8, Ordering}, +}; + +/// Local forwarding progress; completion does not prove upstream processing. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RequestDeliveryState { + NotStarted, + Started, + Completed, +} + +/// Shared by the body runner and its writer so cancellation retains progress. +#[derive(Clone, Debug, Default)] +pub struct RequestDelivery(Arc); + +impl RequestDelivery { + /// Call before awaiting a write, because an error can follow a partial write. + pub(crate) fn start(&self) { + let _ = self + .0 + .compare_exchange(0, 1, Ordering::AcqRel, Ordering::Acquire); + } + + /// Call only after every local body/framing write succeeds. + pub(crate) fn complete(&self) { + self.0.store(2, Ordering::Release); + } + + pub(crate) fn state(&self) -> RequestDeliveryState { + match self.0.load(Ordering::Acquire) { + 0 => RequestDeliveryState::NotStarted, + 1 => RequestDeliveryState::Started, + // Only start and complete can write this private atomic. + _ => RequestDeliveryState::Completed, + } + } +} diff --git a/crates/openshell-supervisor-network/src/l7/rest/request_body.rs b/crates/openshell-supervisor-network/src/l7/rest/request_body.rs index ab07329575..9e50bcda9d 100644 --- a/crates/openshell-supervisor-network/src/l7/rest/request_body.rs +++ b/crates/openshell-supervisor-network/src/l7/rest/request_body.rs @@ -11,8 +11,9 @@ use std::sync::atomic::{AtomicBool, Ordering}; +use crate::l7::request_delivery::{RequestDelivery, RequestDeliveryState}; use futures::FutureExt as _; -use openshell_supervisor_middleware::{HttpBodyOutput, HttpPipelineFinish}; +use openshell_supervisor_middleware::{HttpBodyOutput, HttpPipelineFinish, HttpRequestFailureKind}; use super::*; @@ -541,28 +542,38 @@ fn commit_live_head( pub enum LiveRequestFailure { /// Request middleware rejected or failed the body. #[error("{reason}")] + #[diagnostic(help("local upstream delivery: {delivery:?}"))] Middleware { + #[source] + kind: Option, + delivery: RequestDeliveryState, reason: String, denial: Option, failure: Option, }, /// The sandbox stopped sending the request body. #[error("request body client progress timeout")] - ClientTimeout, + #[diagnostic(help("local upstream delivery: {delivery:?}"))] + ClientTimeout { delivery: RequestDeliveryState }, } /// The upstream stopped accepting the request body. It may have answered /// before it stopped reading, for example with 413. #[derive(Debug, thiserror::Error, miette::Diagnostic)] #[error("upstream stopped accepting the request body: {detail}")] +#[diagnostic(help("local upstream delivery: {delivery:?}"))] pub struct UpstreamWriteFailed { + #[source] + kind: HttpRequestFailureKind, detail: String, + delivery: RequestDeliveryState, } /// Upstream write half that records whether a write failed. struct UpstreamWrites<'a, W> { inner: &'a mut W, failed: &'a AtomicBool, + delivery: &'a RequestDelivery, } impl UpstreamWrites<'_, W> { @@ -584,6 +595,9 @@ impl AsyncWrite for UpstreamWrites<'_, W> { buf: &[u8], ) -> std::task::Poll> { let this = self.get_mut(); + if !buf.is_empty() { + this.delivery.start(); + } let result = std::pin::Pin::new(&mut *this.inner).poll_write(cx, buf); this.record(result) } @@ -761,18 +775,26 @@ where U: AsyncWrite + Unpin, { let failed = AtomicBool::new(false); + let delivery = body.delivery.clone(); let mut upstream = UpstreamWrites { inner: upstream, failed: &failed, + delivery: &delivery, }; let uploaded = upload_live_body(head, body, client, &mut upstream, options, &failed).await; match uploaded { Err(error) if failed.load(Ordering::Acquire) => { Err(miette::Report::new(UpstreamWriteFailed { + kind: HttpRequestFailureKind::Io, detail: error.to_string(), + delivery: delivery.state(), })) } - uploaded => uploaded, + Ok(()) => { + delivery.complete(); + Ok(()) + } + Err(error) => Err(error), } } @@ -964,15 +986,66 @@ mod tests { let (mut upstream, peer) = tokio::io::duplex(16); drop(peer); let failed = AtomicBool::new(false); + let delivery = RequestDelivery::default(); let mut writes = UpstreamWrites { inner: &mut upstream, failed: &failed, + delivery: &delivery, }; writes .write_all(b"body") .await .expect_err("the upstream closed"); assert!(failed.load(Ordering::Acquire)); + assert_eq!(delivery.state(), RequestDeliveryState::Started); + } + + #[tokio::test] + async fn partial_upstream_write_retains_started_delivery() { + use std::{ + pin::Pin, + task::{Context, Poll}, + }; + struct PartialWriter(Vec); + impl AsyncWrite for PartialWriter { + fn poll_write( + mut self: Pin<&mut Self>, + _: &mut Context<'_>, + bytes: &[u8], + ) -> Poll> { + if self.0.is_empty() { + self.0.extend_from_slice(&bytes[..2]); + Poll::Ready(Ok(2)) + } else { + Poll::Ready(Err(std::io::ErrorKind::BrokenPipe.into())) + } + } + fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + fn poll_shutdown( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + } + let delivery = RequestDelivery::default(); + let failed = AtomicBool::new(false); + let mut upstream = PartialWriter(Vec::new()); + assert_eq!(delivery.state(), RequestDeliveryState::NotStarted); + let mut writes = UpstreamWrites { + inner: &mut upstream, + failed: &failed, + delivery: &delivery, + }; + writes + .write_all(b"POST / HTTP/1.1\r\n\r\n") + .await + .expect_err("partial write"); + assert_eq!(upstream.0, b"PO"); + assert!(failed.load(Ordering::Acquire)); + assert_eq!(delivery.state(), RequestDeliveryState::Started); } #[tokio::test] diff --git a/docs/extensibility/supervisor-middleware/operations.mdx b/docs/extensibility/supervisor-middleware/operations.mdx index 4d10855c9d..f7b656f8dd 100644 --- a/docs/extensibility/supervisor-middleware/operations.mdx +++ b/docs/extensibility/supervisor-middleware/operations.mdx @@ -161,6 +161,12 @@ Upgrade the gateway before deploying the service. Recreate sandboxes whose super The [content guard example](https://github.com/NVIDIA/OpenShell/tree/main/examples/supervisor-middleware-content-guard) implements both HTTP operations. +### Failure Attribution in the Runtime + +HTTP request outcomes retain a platform-owned failure kind separately from diagnostic text and sandbox-facing failure codes. Runtime callers can distinguish local stage, chain, whole-body, stream-idle and client-progress deadlines from remote gRPC status codes, including a service-reported deadline. An explicit service or policy denial has no platform failure kind. Diagnostic sanitization and enforcement behavior stay the same. + +The live request relay also records whether upstream writing was attempted and whether local forwarding completed. A write that fails after partial progress remains started. Local completion does not prove upstream receipt or processing, and does not authorize a retry. + ## WebSocket WebSocket middleware checks the text messages the sandbox sends over a WebSocket connection.