From bd0b5ba45fd24864e07ddf0879c116b23808bd57 Mon Sep 17 00:00:00 2001 From: Shiju Date: Mon, 2 Nov 2026 19:54:49 +0530 Subject: [PATCH] fix(middleware): preserve HTTP request failure causes Retain typed local deadline and remote failure causes through HTTP hook diagnostics and request handling. Track local upstream write progress through the existing live writer, including partial failures. Preserve response selection, diagnostic text and request retries. Refs #4144 Build on PR #4359. Caller and event exposure remains with the companion tracing work. Signed-off-by: Shiju --- .../src/http_hooks/pipeline.rs | 115 ++++++++--- .../src/http_hooks/request.rs | 8 +- .../src/http_hooks/tests.rs | 179 +++++++++++++++++- .../src/lib.rs | 150 +++++++++++++-- .../src/request_failure.rs | 103 ++++++++++ .../src/l7/middleware.rs | 111 +++++++++-- .../src/l7/mod.rs | 1 + .../src/l7/relay.rs | 6 +- .../src/l7/request_delivery.rs | 44 +++++ .../src/l7/rest/request_body.rs | 79 +++++++- .../supervisor-middleware/operations.mdx | 6 + 11 files changed, 727 insertions(+), 75 deletions(-) create mode 100644 crates/openshell-supervisor-middleware/src/request_failure.rs create mode 100644 crates/openshell-supervisor-network/src/l7/request_delivery.rs 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.