try to reject Promise on panic!()

Because of a limitation with wasm_bindgen (rustwasm/wasm-bindgen#2724),
a Rust panic in asynchronous code isn't able to reject the Promise it
happened in.

This change makes the main worker Promise cancellable, with a panic hook
or by rewriting the wasm binary, it's possible to reject the Promise and
handle the error.

The worker concurrency model makes rejecting the Promise not a guarantee
in cases the worker has a lot of traffic that cause a panic. See comment
in source code.

Usage (for now), add to your worker:
```rust
fn start() {
    std::panic::set_hook(Box::new(|i| {
        __workers_rs_cancel();
        worker::console_error_panic_hook::hook(i);
    }));
}
```

For example, previously:
```
POST https://example.com/ - Exception Thrown @ 8/10/2023, 12:26:59 PM
✘ [ERROR]   Error: The script will never generate a response.
```

with this change:
```
POST https://example.com/ - Exception Thrown @ 8/10/2023, 12:27:05 PM
  (error) panicked at 'oops', src/lib.rs:43:5

Stack:

Error
    at entry.js:746:17
    at logError (entry.js:304:14)
    at __wbg_new_abda76e883ba8a5f (entry.js:745:10)
    at console_error_panic_hook::hook::hb5e89e21e6f36fcb (wasm://wasm/0351507a:wasm-function[174]:0x1bca6)
    at test_coredump_worker::start_start_glue::{{closure}}::hc242a0946799f25b (wasm://wasm/0351507a:wasm-function[854]:0x2d662)
    at std::panicking::rust_panic_with_hook::h85b7f6628c291e12 (wasm://wasm/0351507a:wasm-function[269]:0x2234c)
    at std::panicking::begin_panic_handler::{{closure}}::h1e17bad04a5713a4 (wasm://wasm/0351507a:wasm-function[323]:0x24aa6)
    at std::sys_common::backtrace::__rust_end_short_backtrace::h63adae5c31458c4b (wasm://wasm/0351507a:wasm-function[875]:0x2d72e)
    at rust_begin_unwind (wasm://wasm/0351507a:wasm-function[491]:0x29a2d)
    at core::panicking::panic_fmt::hf5c4cd929d4aaa9e (wasm://wasm/0351507a:wasm-function[601]:0x2ba1d)
```

worker-macros/src/event.rs

@@ -53,7 +53,9 @@ pub fn expand_macro(attr: TokenStream, item: TokenStream) -> TokenStream {
             let error_handling = match respond_with_errors {
                 true => {
                     quote! {
-                        ::worker::Response::error(e.to_string(), 500).unwrap().into()
+                        Ok(::worker::worker_sys::web_sys::Response::from(
+                            ::worker::Response::error(e.to_string(), 500).unwrap()
+                        ).into())
                     }
                 }
                 false => {
@@ -64,20 +66,48 @@ pub fn expand_macro(attr: TokenStream, item: TokenStream) -> TokenStream {
             // create a new "main" function that takes the worker_sys::Request, and calls the
             // original attributed function, passing in a converted worker::Request
             let wrapper_fn = quote! {
-                pub async fn #wrapper_fn_ident(
+                pub fn #wrapper_fn_ident(
                     req: ::worker::worker_sys::web_sys::Request,
                     env: ::worker::Env,
                     ctx: ::worker::worker_sys::Context
-                ) -> ::worker::worker_sys::web_sys::Response {
+                ) -> ::worker::js_sys::Promise {
                     let ctx = worker::Context::new(ctx);
-                    // get the worker::Result<worker::Response> by calling the original fn
-                    match #input_fn_ident(::worker::Request::from(req), env, ctx).await.map(::worker::worker_sys::web_sys::Response::from) {
-                        Ok(res) => res,
-                        Err(e) => {
-                            ::worker::console_error!("{}", &e);
-                            #error_handling
+
+                    let promise = ::worker::wasm_bindgen_futures::future_to_promise(async move {
+                        // get the worker::Result<worker::Response> by calling the original fn
+                        match #input_fn_ident(::worker::Request::from(req), env, ctx).await.map(::worker::worker_sys::web_sys::Response::from) {
+                            Ok(res) => Ok(res.into()),
+                            Err(e) => {
+                                ::worker::console_error!("{}", &e);
+                                #error_handling
+                            }
                         }
-                    }
+                    });
+
+                    // Wrap the user promise into our cancellable promise
+                    // with an AbortController.
+                    let abort_controller = Box::new(::worker::AbortController::default());
+                    let promise = ::worker::cancellable_promise::make(abort_controller.signal(), promise);
+
+                    // Save the AbortController.
+                    *ABORT_CONTROLLER.lock().unwrap() = Some(abort_controller);
+
+                    // Remove the AbortController once the Promise terminates.
+                    let promise = {
+                        let clean_abort_controller = ::worker::wasm_bindgen::closure::Closure::new(move || {
+                            if let Ok(mut abort_controller) = ABORT_CONTROLLER.lock() {
+                                *abort_controller = None;
+                            }
+                        });
+                        // prevent the closure of being dropped before JS tries
+                        // to call it.
+                        let promise = promise.finally(&clean_abort_controller);
+                        clean_abort_controller.forget();
+
+                        promise
+                    };
+
+                    promise
                 }
             };
             let wasm_bindgen_code =
@@ -87,9 +117,38 @@ pub fn expand_macro(attr: TokenStream, item: TokenStream) -> TokenStream {
             let output = quote! {
                 #input_fn
 
+                use std::sync::Mutex;
+                ::worker::lazy_static::lazy_static! {
+                    // Keep the last request's AbortController, allowing the
+                    // request to be aborted later if the code panics.
+                    // Panics here cause the worker to hang,
+                    // see https://github.com/rustwasm/wasm-bindgen/issues/2724.
+                    //
+                    // Only keeping the last request's AbortController may lead
+                    // to some requests hanging, but it requires a lot of traffic
+                    // that cause a panic.
+                    //
+                    // Note that while the worker is able to concurrently process
+                    // request, we only keep the latest request's AbortController.
+                    // This is to avoid cancelling a request from another request's
+                    // context which leads to the error
+                    // `Error: Cannot perform I/O on behalf of a different request...`.
+                    static ref ABORT_CONTROLLER: Mutex<Option<Box<worker::AbortController>>> = Mutex::new(None);
+                }
+
+                #[no_mangle]
+                pub extern "C" fn __workers_rs_cancel() {
+                    if let Ok(controller) = ABORT_CONTROLLER.lock() {
+                        if let Some(controller) = controller.as_ref() {
+                            controller.abort();
+                        }
+                    }
+                }
+
                 mod _worker_fetch {
                     use ::worker::{wasm_bindgen, wasm_bindgen_futures};
                     use super::#input_fn_ident;
+                    use super::ABORT_CONTROLLER;
                     #wasm_bindgen_code
                 }
             };

worker/Cargo.toml

@@ -32,6 +32,8 @@ wasm-streams = "0.3.0"
 worker-kv = "0.6.0"
 worker-macros = { path = "../worker-macros", version = "0.0.10" }
 worker-sys = { path = "../worker-sys", version = "0.0.10" }
+lazy_static = "1.4.0"
+console_error_panic_hook = "0.1.7"
 
 [dependencies.web-sys]
 version = "0.3.63"

worker/src/abort.rs

@@ -16,7 +16,7 @@ impl AbortController {
     }
 
     /// Aborts any operation using a [AbortSignal] created from this controller.
-    pub fn abort(self) {
+    pub fn abort(&self) {
         self.inner.abort()
     }
 
@@ -78,3 +78,6 @@ impl Deref for AbortSignal {
         &self.inner
     }
 }
+
+unsafe impl Sync for AbortController {}
+unsafe impl Send for AbortController {}

worker/src/cancellable_promise.rs

@@ -0,0 +1,40 @@
+use crate::js_sys::Promise;
+use crate::wasm_bindgen::prelude::*;
+use crate::AbortSignal;
+
+pub fn make(signal: AbortSignal, p: Promise) -> Promise {
+    Promise::new(&mut |resolve, reject| {
+        let msg = "Request has been aborted";
+
+        if signal.aborted() {
+            reject
+                .call1(&JsValue::undefined(), &msg.into())
+                .unwrap_throw();
+        }
+
+        {
+            let reject = reject.clone();
+            let on_abort = Closure::<dyn FnMut(JsValue)>::new(move |_| {
+                reject
+                    .call1(&JsValue::undefined(), &msg.into())
+                    .unwrap_throw();
+            });
+            signal.set_onabort(Some(&on_abort.as_ref().unchecked_ref()));
+            // prevent the closure of being dropped before JS tries
+            // to call it.
+            on_abort.forget();
+        }
+
+        // Listen for the initial promise completion
+        {
+            let resolve2 = Closure::new(move |val| {
+                resolve.call1(&JsValue::undefined(), &val).unwrap_throw();
+            });
+            let reject2 = Closure::new(move |val| {
+                reject.call1(&JsValue::undefined(), &val).unwrap_throw();
+            });
+            p.then2(&resolve2, &reject2);
+        }
+        ()
+    })
+}

worker/src/lib.rs

@@ -7,7 +7,11 @@ use std::result::Result as StdResult;
 #[doc(hidden)]
 pub use async_trait;
 #[doc(hidden)]
+pub use console_error_panic_hook;
+#[doc(hidden)]
 pub use js_sys;
+#[doc(hidden)]
+pub use lazy_static;
 pub use url::Url;
 #[doc(hidden)]
 pub use wasm_bindgen;
@@ -56,6 +60,8 @@ mod cf;
 mod context;
 mod cors;
 // Require pub module for macro export
+#[doc(hidden)]
+pub mod cancellable_promise;
 #[cfg(feature = "d1")]
 pub mod d1;
 mod date;