Return 5XX for requests that land in a reused session that crashes (#540)

Author: ulyssa

cli/tests/integration/reusable_sessions.rs

@@ -24,3 +24,42 @@ viceroy_test!(check_hostcalls_implemented, |is_component| {
 
     Ok(())
 });
+
+viceroy_test!(check_crash_causes_5xx, |is_component| {
+    let test = Test::using_fixture("reusable-sessions.wasm")
+        .adapt_component(is_component)
+        .via_hyper();
+
+    let reqs = (0..5)
+        .into_iter()
+        .map(|n| {
+            let body = if n == 4 {
+                "crash!".to_owned()
+            } else {
+                n.to_string()
+            };
+            Request::post("/").body(body).unwrap()
+        })
+        .collect();
+
+    let mut resps = test.against_many(reqs).await?;
+    let errs = resps.split_off(4);
+
+    assert_eq!(resps.len(), 4);
+    assert_eq!(errs.len(), 1);
+
+    for (n, resp) in resps.into_iter().enumerate() {
+        let exp = format!("Response #{}", n + 1);
+        assert_eq!(resp.status(), StatusCode::OK);
+        assert_eq!(resp.into_body().read_into_string().await?, exp);
+    }
+
+    for resp in errs.into_iter() {
+        assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
+        let body = resp.into_body().read_into_string().await?;
+        let needle = "error while executing at wasm backtrace";
+        assert!(body.contains(needle), "missing expected string: {body:?}");
+    }
+
+    Ok(())
+});

cli/tests/integration/unknown_import_behavior.rs

@@ -33,8 +33,14 @@ async fn trap_behavior_function_called() -> TestResult {
     // key parts that should be relatively stable across invocations and wasmtime
     // versions. Fundamentally though, we're still making assertions based on pretty-printed errors,
     // so beware of trivial breakages.
-    assert!(body.contains("error while executing at wasm backtrace"));
-    assert!(body.contains("unknown_import::main::"));
+    assert!(
+        body.contains("error while executing at wasm backtrace"),
+        "missing expected string: {body:?}"
+    );
+    assert!(
+        body.contains("unknown_import::main::"),
+        "missing expected string: {body:?}"
+    );
 
     Ok(())
 }

cli/tests/trap-test/src/main.rs

@@ -37,11 +37,8 @@ async fn fatal_error_traps_impl(adapt_core_wasm: bool) -> TestResult {
     // Trap error supplied by the Guest.
 
     let body = resp.into_body().read_into_string().await?;
-
-    assert_eq!(
-        body,
-        "Fatal error: [A fatal error occurred in the test-only implementation of header_values_get]"
-    );
+    let needle = "Fatal error: [A fatal error occurred in the test-only implementation of header_values_get]";
+    assert!(body.contains(needle), "body missing expected string: {body}");
 
     Ok(())
 }

src/execute.rs

@@ -596,12 +596,7 @@ impl ExecuteCtx {
                     "There was an error handling the request {}",
                     e.to_string()
                 );
-                #[allow(unused_mut)]
-                let mut response = Response::builder()
-                    .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
-                    .body(Body::empty())
-                    .unwrap();
-                (response, Some(e))
+                (anyhow_response(&e), Some(e))
             }
             Err(e) => panic!("failed to run guest: {}", e),
         }
@@ -678,7 +673,15 @@ impl ExecuteCtx {
 
                 // Ensure the downstream response channel is closed, whether or not a response was
                 // sent during execution.
-                store.data_mut().session.close_downstream_response_sender();
+                let resp = outcome
+                    .as_ref()
+                    .err()
+                    .map(exec_err_to_response)
+                    .unwrap_or_default();
+                store
+                    .data_mut()
+                    .session
+                    .close_downstream_response_sender(resp);
 
                 let request_duration = Instant::now().duration_since(start_timestamp);
 
@@ -745,7 +748,12 @@ impl ExecuteCtx {
 
                 // Ensure the downstream response channel is closed, whether or not a response was
                 // sent during execution.
-                store.data_mut().close_downstream_response_sender();
+                let resp = outcome
+                    .as_ref()
+                    .err()
+                    .map(exec_err_to_response)
+                    .unwrap_or_default();
+                store.data_mut().close_downstream_response_sender(resp);
 
                 let request_duration = Instant::now().duration_since(start_timestamp);
 
@@ -830,7 +838,9 @@ impl ExecuteCtx {
 
         // Ensure the downstream response channel is closed, whether or not a response was
         // sent during execution.
-        store.data_mut().close_downstream_response_sender();
+        store
+            .data_mut()
+            .close_downstream_response_sender(Response::default());
 
         // We don't do anything with any response on the receiver, but
         // it's important to keep it alive until after the program has
@@ -986,17 +996,24 @@ fn write_profile(store: &mut wasmtime::Store<WasmCtx>, guest_profile_path: Optio
 }
 
 fn guest_result_to_response(resp: Response<Body>, err: Option<anyhow::Error>) -> Response<Body> {
-    if let Some(err) = err {
-        let body = err.root_cause().to_string();
-        Response::builder()
-            .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
-            .body(Body::from(body.as_bytes()))
-            .unwrap()
+    err.as_ref().map(anyhow_response).unwrap_or(resp)
+}
+
+fn exec_err_to_response(err: &ExecutionError) -> Response<Body> {
+    if let ExecutionError::WasmTrap(e) = err {
+        anyhow_response(e)
     } else {
-        resp
+        panic!("failed to run guest: {err}")
     }
 }
 
+fn anyhow_response(err: &anyhow::Error) -> Response<Body> {
+    Response::builder()
+        .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
+        .body(Body::from(format!("{err:?}").into_bytes()))
+        .unwrap()
+}
+
 impl Drop for ExecuteCtx {
     fn drop(&mut self) {
         if let Some(join_handle) = self.epoch_increment_thread.take() {

src/linking.rs

@@ -2,9 +2,10 @@
 
 use {
     crate::{
-        config::ExperimentalModule, execute::ExecuteCtx, logging::LogEndpoint, session::Session,
-        wiggle_abi, Error,
+        body::Body, config::ExperimentalModule, execute::ExecuteCtx, logging::LogEndpoint,
+        session::Session, wiggle_abi, Error,
     },
+    hyper::Response,
     std::collections::HashSet,
     wasmtime::{GuestProfiler, Linker, Store, StoreLimits, StoreLimitsBuilder, UpdateDeadline},
     wasmtime_wasi::preview1::WasiP1Ctx,
@@ -134,8 +135,8 @@ impl ComponentCtx {
         &self.limiter
     }
 
-    pub fn close_downstream_response_sender(&mut self) {
-        self.session.close_downstream_response_sender()
+    pub fn close_downstream_response_sender(&mut self, resp: Response<Body>) {
+        self.session.close_downstream_response_sender(resp)
     }
 
     /// Initialize a new [`Store`][store], given an [`ExecuteCtx`][ctx].
@@ -232,8 +233,8 @@ impl WasmCtx {
 }
 
 impl WasmCtx {
-    pub fn close_downstream_response_sender(&mut self) {
-        self.session.close_downstream_response_sender()
+    pub fn close_downstream_response_sender(&mut self, resp: Response<Body>) {
+        self.session.close_downstream_response_sender(resp)
     }
 }
 

src/session.rs

@@ -267,9 +267,10 @@ impl Session {
         self.downstream_resp.redirect_to_pushpin(redirect_info)
     }
 
-    /// Close the downstream response sender, potentially without sending any response.
-    pub fn close_downstream_response_sender(&mut self) {
-        self.downstream_resp.close()
+    /// Ensure the downstream response sender is closed, and send the provided response if it
+    /// isn't.
+    pub fn close_downstream_response_sender(&mut self, resp: Response<Body>) {
+        let _ = self.downstream_resp.send(resp);
     }
 
     // ----- Bodies API -----

src/session/downstream.rs

@@ -85,6 +85,7 @@ impl DownstreamResponseState {
     }
 
     /// Close the `DownstreamResponse`, potentially without sending any response.
+    #[allow(unused)]
     pub fn close(&mut self) {
         *self = DownstreamResponseState::Closed;
     }

test-fixtures/Cargo.toml

@@ -11,9 +11,9 @@ publish = false
 anyhow = "1.0.86"
 base64 = "0.21.2"
 bitflags = "1.3.2"
-fastly = "0.11.5"
-fastly-shared = "0.11.5"
-fastly-sys = "0.11.5"
+fastly = "0.11.7"
+fastly-shared = "0.11.7"
+fastly-sys = "0.11.7"
 hex-literal = "0.4.1"
 bytes = "1.0.0"
 http = "1.1.0"

test-fixtures/src/bin/reusable-sessions.rs

@@ -3,45 +3,7 @@
 use fastly::{Request, Response};
 use fastly::handle::{BodyHandle, RequestHandle};
 use fastly_shared::{FastlyStatus, INVALID_REQUEST_HANDLE};
-
-pub type RequestPromiseHandle = u32;
-
-#[derive(Default)]
-#[repr(C)]
-pub struct NextRequestOptions {
-    pub reserved: u64,
-}
-
-bitflags::bitflags! {
-    /// Request options.
-    #[derive(Default)]
-    #[repr(transparent)]
-    pub struct NextRequestOptionsMask: u32 {
-        const RESERVED = 1 << 0;
-    }
-}
-
-#[link(wasm_import_module = "fastly_http_downstream")]
-extern "C" {
-    #[link_name = "next_request"]
-    pub fn next_request(
-        options_mask: NextRequestOptionsMask,
-        options: *const NextRequestOptions,
-        handle_out: *mut RequestPromiseHandle,
-    ) -> FastlyStatus;
-
-    #[link_name = "next_request_wait"]
-    pub fn next_request_wait(
-        handle: RequestPromiseHandle,
-        req_handle_out: *mut fastly_sys::RequestHandle,
-        body_handle_out: *mut fastly_sys::BodyHandle,
-    ) -> FastlyStatus;
-
-    #[link_name = "next_request_abandon"]
-    pub fn next_request_abandon(
-        handle: RequestPromiseHandle,
-    ) -> FastlyStatus;
-}
+use fastly_sys::fastly_http_downstream::*;
 
 fn is_ready(handle: u32) -> bool {
     let mut ready_out: u32 = 0;