fix(http): use tokio oneshot channel for detecting abort (#1395)

* fix(http): properly handle aborting

closes #1376

* abort early in JS

* avoid using unnecessary mutexes

* fix lint

* update bundle

* simplify

Author: amrbashir

.changes/http-abort.md

@@ -0,0 +1,6 @@
+---
+"http": "patch"
+"http-js": "patch"
+---
+
+Fix cancelling requests using `AbortSignal`.

Cargo.lock

@@ -26,6 +26,7 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 tauri = { workspace = true }
 thiserror = { workspace = true }
+tokio = { version = "1", features = [ "sync", "macros" ] }
 tauri-plugin-fs = { path = "../fs", version = "2.0.0-beta.10" }
 urlpattern = "0.2"
 regex = "1"

plugins/http/Cargo.toml

@@ -86,6 +86,8 @@ export interface ClientOptions {
   proxy?: Proxy;
 }
 
+const ERROR_REQUEST_CANCELLED = "Request canceled";
+
 /**
  * Fetch a resource from the network. It returns a `Promise` that resolves to the
  * `Response` to that `Request`, whether it is successful or not.
@@ -104,6 +106,12 @@ export async function fetch(
   input: URL | Request | string,
   init?: RequestInit & ClientOptions,
 ): Promise<Response> {
+  // abort early here if needed
+  const signal = init?.signal;
+  if (signal?.aborted) {
+    throw new Error(ERROR_REQUEST_CANCELLED);
+  }
+
   const maxRedirections = init?.maxRedirections;
   const connectTimeout = init?.connectTimeout;
   const proxy = init?.proxy;
@@ -115,8 +123,6 @@ export async function fetch(
     delete init.proxy;
   }
 
-  const signal = init?.signal;
-
   const headers = init?.headers
     ? init.headers instanceof Headers
       ? init.headers
@@ -153,6 +159,11 @@ export async function fetch(
     ],
   );
 
+  // abort early here if needed
+  if (signal?.aborted) {
+    throw new Error(ERROR_REQUEST_CANCELLED);
+  }
+
   const rid = await invoke<number>("plugin:http|fetch", {
     clientConfig: {
       method: req.method,
@@ -165,11 +176,17 @@ export async function fetch(
     },
   });
 
-  signal?.addEventListener("abort", () => {
-    void invoke("plugin:http|fetch_cancel", {
-      rid,
-    });
-  });
+  const abort = () => invoke("plugin:http|fetch_cancel", { rid });
+
+  // abort early here if needed
+  if (signal?.aborted) {
+    // we don't care about the result of this proimse
+    // eslint-disable-next-line @typescript-eslint/no-floating-promises
+    abort();
+    throw new Error(ERROR_REQUEST_CANCELLED);
+  }
+
+  signal?.addEventListener("abort", () => abort);
 
   interface FetchSendResponse {
     status: number;
@@ -203,7 +220,6 @@ export async function fetch(
         ? new Uint8Array(body)
         : null,
     {
-      headers: responseHeaders,
       status,
       statusText,
     },

plugins/http/api-iife.js

@@ -11,8 +11,9 @@ use tauri::{
     async_runtime::Mutex,
     command,
     ipc::{CommandScope, GlobalScope},
-    Manager, ResourceId, Runtime, State, Webview,
+    Manager, ResourceId, ResourceTable, Runtime, State, Webview,
 };
+use tokio::sync::oneshot::{channel, Receiver, Sender};
 
 use crate::{
     scope::{Entry, Scope},
@@ -22,20 +23,47 @@ use crate::{
 const HTTP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
 
 struct ReqwestResponse(reqwest::Response);
+impl tauri::Resource for ReqwestResponse {}
 
-type CancelableResponseResult = Result<Result<reqwest::Response>>;
+type CancelableResponseResult = Result<reqwest::Response>;
 type CancelableResponseFuture =
     Pin<Box<dyn Future<Output = CancelableResponseResult> + Send + Sync>>;
 
-struct FetchRequest(Mutex<CancelableResponseFuture>);
-impl FetchRequest {
-    fn new(f: CancelableResponseFuture) -> Self {
-        Self(Mutex::new(f))
+struct FetchRequest {
+    fut: Mutex<CancelableResponseFuture>,
+    abort_tx_rid: ResourceId,
+    abort_rx_rid: ResourceId,
+}
+impl tauri::Resource for FetchRequest {}
+
+struct AbortSender(Sender<()>);
+impl tauri::Resource for AbortRecveiver {}
+
+impl AbortSender {
+    fn abort(self) {
+        let _ = self.0.send(());
     }
 }
 
-impl tauri::Resource for FetchRequest {}
-impl tauri::Resource for ReqwestResponse {}
+struct AbortRecveiver(Receiver<()>);
+impl tauri::Resource for AbortSender {}
+
+trait AddRequest {
+    fn add_request(&mut self, fut: CancelableResponseFuture) -> ResourceId;
+}
+
+impl AddRequest for ResourceTable {
+    fn add_request(&mut self, fut: CancelableResponseFuture) -> ResourceId {
+        let (tx, rx) = channel::<()>();
+        let (tx, rx) = (AbortSender(tx), AbortRecveiver(rx));
+        let req = FetchRequest {
+            fut: Mutex::new(fut),
+            abort_tx_rid: self.add(tx),
+            abort_rx_rid: self.add(rx),
+        };
+        self.add(req)
+    }
+}
 
 #[derive(Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -239,9 +267,9 @@ pub async fn fetch<R: Runtime>(
                     request = request.body(data);
                 }
 
-                let fut = async move { Ok(request.send().await.map_err(Into::into)) };
+                let fut = async move { request.send().await.map_err(Into::into) };
                 let mut resources_table = webview.resources_table();
-                let rid = resources_table.add(FetchRequest::new(Box::pin(fut)));
+                let rid = resources_table.add_request(Box::pin(fut));
 
                 Ok(rid)
             } else {
@@ -260,24 +288,23 @@ pub async fn fetch<R: Runtime>(
                 .header(header::CONTENT_TYPE, data_url.mime_type().to_string())
                 .body(reqwest::Body::from(body))?;
 
-            let fut = async move { Ok(Ok(reqwest::Response::from(response))) };
+            let fut = async move { Ok(reqwest::Response::from(response)) };
             let mut resources_table = webview.resources_table();
-            let rid = resources_table.add(FetchRequest::new(Box::pin(fut)));
+            let rid = resources_table.add_request(Box::pin(fut));
             Ok(rid)
         }
         _ => Err(Error::SchemeNotSupport(scheme.to_string())),
     }
 }
 
 #[command]
-pub async fn fetch_cancel<R: Runtime>(webview: Webview<R>, rid: ResourceId) -> crate::Result<()> {
-    let req = {
-        let resources_table = webview.resources_table();
-        resources_table.get::<FetchRequest>(rid)?
-    };
-    let mut req = req.0.lock().await;
-    *req = Box::pin(async { Err(Error::RequestCanceled) });
-
+pub fn fetch_cancel<R: Runtime>(webview: Webview<R>, rid: ResourceId) -> crate::Result<()> {
+    let mut resources_table = webview.resources_table();
+    let req = resources_table.get::<FetchRequest>(rid)?;
+    let abort_tx = resources_table.take::<AbortSender>(req.abort_tx_rid)?;
+    if let Some(abort_tx) = Arc::into_inner(abort_tx) {
+        abort_tx.abort();
+    }
     Ok(())
 }
 
@@ -286,14 +313,26 @@ pub async fn fetch_send<R: Runtime>(
     webview: Webview<R>,
     rid: ResourceId,
 ) -> crate::Result<FetchResponse> {
-    let req = {
+    let (req, abort_rx) = {
         let mut resources_table = webview.resources_table();
-        resources_table.take::<FetchRequest>(rid)?
+        let req = resources_table.get::<FetchRequest>(rid)?;
+        let abort_rx = resources_table.take::<AbortRecveiver>(req.abort_rx_rid)?;
+        (req, abort_rx)
     };
 
-    let res = match req.0.lock().await.as_mut().await {
-        Ok(Ok(res)) => res,
-        Ok(Err(e)) | Err(e) => return Err(e),
+    let Some(abort_rx) = Arc::into_inner(abort_rx) else {
+        return Err(Error::RequestCanceled);
+    };
+
+    let mut fut = req.fut.lock().await;
+
+    let res = tokio::select! {
+        res = fut.as_mut() => res?,
+        _ = abort_rx.0 => {
+            let mut resources_table = webview.resources_table();
+            resources_table.close(rid)?;
+            return Err(Error::RequestCanceled);
+        }
     };
 
     let status = res.status();