fix(http): fix aborting a streaming response (#2562)

Author: amrbashir

.changes/http-stream-cancel.md

@@ -0,0 +1,7 @@
+---
+"http": "patch"
+"http-js": "patch"
+---
+
+Fix aborting a request in the middle of a streaming response.
+

plugins/http/api-iife.js

@@ -6,7 +6,13 @@
 #[allow(dead_code)]
 mod scope;
 
-const COMMANDS: &[&str] = &["fetch", "fetch_cancel", "fetch_send", "fetch_read_body"];
+const COMMANDS: &[&str] = &[
+    "fetch",
+    "fetch_cancel",
+    "fetch_send",
+    "fetch_read_body",
+    "fetch_cancel_body",
+];
 
 /// HTTP scope entry.
 #[derive(schemars::JsonSchema)]

plugins/http/build.rs

@@ -26,7 +26,7 @@
  * @module
  */
 
-import { Channel, invoke } from '@tauri-apps/api/core'
+import { invoke } from '@tauri-apps/api/core'
 
 /**
  * Configuration of a proxy that a Client should pass requests to.
@@ -126,7 +126,7 @@ export async function fetch(
   input: URL | Request | string,
   init?: RequestInit & ClientOptions
 ): Promise<Response> {
-  // abort early here if needed
+  // Optimistically check for abort signal and avoid doing any work
   const signal = init?.signal
   if (signal?.aborted) {
     throw new Error(ERROR_REQUEST_CANCELLED)
@@ -181,7 +181,7 @@ export async function fetch(
     ]
   )
 
-  // abort early here if needed
+  // Optimistically check for abort signal and avoid doing any work on the Rust side
   if (signal?.aborted) {
     throw new Error(ERROR_REQUEST_CANCELLED)
   }
@@ -201,7 +201,8 @@ export async function fetch(
 
   const abort = () => invoke('plugin:http|fetch_cancel', { rid })
 
-  // abort early here if needed
+  // Optimistically check for abort signal
+  // and avoid doing any work after doing intial work on the Rust side
   if (signal?.aborted) {
     // we don't care about the result of this proimse
     // eslint-disable-next-line @typescript-eslint/no-floating-promises
@@ -229,41 +230,52 @@ export async function fetch(
     rid
   })
 
+  const dropBody = () => {
+    return invoke('plugin:http|fetch_cancel_body', { rid: responseRid })
+  }
+
+  const readChunk = async (
+    controller: ReadableStreamDefaultController<Uint8Array>
+  ) => {
+    let data: ArrayBuffer
+    try {
+      data = await invoke('plugin:http|fetch_read_body', {
+        rid: responseRid
+      })
+    } catch (e) {
+      // close the stream if an error occurs
+      // and drop the body on Rust side
+      controller.error(e)
+      void dropBody()
+      return
+    }
+
+    const dataUint8 = new Uint8Array(data)
+    const lastByte = dataUint8[dataUint8.byteLength - 1]
+    const actualData = dataUint8.slice(0, dataUint8.byteLength - 1)
+
+    // close when the signal to close (last byte is 1) is sent from the IPC.
+    if (lastByte === 1) {
+      controller.close()
+      return
+    }
+
+    controller.enqueue(actualData)
+  }
+
   // no body for 101, 103, 204, 205 and 304
   // see https://fetch.spec.whatwg.org/#null-body-status
   const body = [101, 103, 204, 205, 304].includes(status)
     ? null
-    : new ReadableStream({
+    : new ReadableStream<Uint8Array>({
         start: (controller) => {
-          const streamChannel = new Channel<ArrayBuffer | number[]>()
-          streamChannel.onmessage = (res: ArrayBuffer | number[]) => {
-            // close early if aborted
-            if (signal?.aborted) {
-              controller.error(ERROR_REQUEST_CANCELLED)
-              return
-            }
-
-            const resUint8 = new Uint8Array(res)
-            const lastByte = resUint8[resUint8.byteLength - 1]
-            const actualRes = resUint8.slice(0, resUint8.byteLength - 1)
-
-            // close when the signal to close (last byte is 1) is sent from the IPC.
-            if (lastByte == 1) {
-              controller.close()
-              return
-            }
-
-            controller.enqueue(actualRes)
-          }
-
-          // run a non-blocking body stream fetch
-          invoke('plugin:http|fetch_read_body', {
-            rid: responseRid,
-            streamChannel
-          }).catch((e) => {
-            controller.error(e)
+          // listen for abort events to cancel reading
+          signal?.addEventListener('abort', () => {
+            controller.error(ERROR_REQUEST_CANCELLED)
+            void dropBody()
           })
-        }
+        },
+        pull: (controller) => readChunk(controller)
       })
 
   const res = new Response(body, {

plugins/http/guest-js/index.ts

@@ -0,0 +1,13 @@
+# Automatically generated - DO NOT EDIT!
+
+"$schema" = "../../schemas/schema.json"
+
+[[permission]]
+identifier = "allow-fetch-cancel-body"
+description = "Enables the fetch_cancel_body command without any pre-configured scope."
+commands.allow = ["fetch_cancel_body"]
+
+[[permission]]
+identifier = "deny-fetch-cancel-body"
+description = "Denies the fetch_cancel_body command without any pre-configured scope."
+commands.deny = ["fetch_cancel_body"]

plugins/http/permissions/autogenerated/commands/fetch_cancel_body.toml

@@ -15,8 +15,9 @@ All fetch operations are enabled.
 
 - `allow-fetch`
 - `allow-fetch-cancel`
-- `allow-fetch-read-body`
 - `allow-fetch-send`
+- `allow-fetch-read-body`
+- `allow-fetch-cancel-body`
 
 ## Permission Table
 
@@ -82,6 +83,32 @@ Denies the fetch_cancel command without any pre-configured scope.
 <tr>
 <td>
 
+`http:allow-fetch-cancel-body`
+
+</td>
+<td>
+
+Enables the fetch_cancel_body command without any pre-configured scope.
+
+</td>
+</tr>
+
+<tr>
+<td>
+
+`http:deny-fetch-cancel-body`
+
+</td>
+<td>
+
+Denies the fetch_cancel_body command without any pre-configured scope.
+
+</td>
+</tr>
+
+<tr>
+<td>
+
 `http:allow-fetch-read-body`
 
 </td>

plugins/http/permissions/autogenerated/reference.md

@@ -17,6 +17,7 @@ All fetch operations are enabled.
 permissions = [
   "allow-fetch",
   "allow-fetch-cancel",
-  "allow-fetch-read-body",
   "allow-fetch-send",
+  "allow-fetch-read-body",
+  "allow-fetch-cancel-body",
 ]

plugins/http/permissions/default.toml

@@ -318,6 +318,18 @@
           "const": "deny-fetch-cancel",
           "markdownDescription": "Denies the fetch_cancel command without any pre-configured scope."
         },
+        {
+          "description": "Enables the fetch_cancel_body command without any pre-configured scope.",
+          "type": "string",
+          "const": "allow-fetch-cancel-body",
+          "markdownDescription": "Enables the fetch_cancel_body command without any pre-configured scope."
+        },
+        {
+          "description": "Denies the fetch_cancel_body command without any pre-configured scope.",
+          "type": "string",
+          "const": "deny-fetch-cancel-body",
+          "markdownDescription": "Denies the fetch_cancel_body command without any pre-configured scope."
+        },
         {
           "description": "Enables the fetch_read_body command without any pre-configured scope.",
           "type": "string",
@@ -343,10 +355,10 @@
           "markdownDescription": "Denies the fetch_send command without any pre-configured scope."
         },
         {
-          "description": "This permission set configures what kind of\nfetch operations are available from the http plugin.\n\nThis enables all fetch operations but does not\nallow explicitly any origins to be fetched. This needs to\nbe manually configured before usage.\n\n#### Granted Permissions\n\nAll fetch operations are enabled.\n\n\n#### This default permission set includes:\n\n- `allow-fetch`\n- `allow-fetch-cancel`\n- `allow-fetch-read-body`\n- `allow-fetch-send`",
+          "description": "This permission set configures what kind of\nfetch operations are available from the http plugin.\n\nThis enables all fetch operations but does not\nallow explicitly any origins to be fetched. This needs to\nbe manually configured before usage.\n\n#### Granted Permissions\n\nAll fetch operations are enabled.\n\n\n#### This default permission set includes:\n\n- `allow-fetch`\n- `allow-fetch-cancel`\n- `allow-fetch-send`\n- `allow-fetch-read-body`\n- `allow-fetch-cancel-body`",
           "type": "string",
           "const": "default",
-          "markdownDescription": "This permission set configures what kind of\nfetch operations are available from the http plugin.\n\nThis enables all fetch operations but does not\nallow explicitly any origins to be fetched. This needs to\nbe manually configured before usage.\n\n#### Granted Permissions\n\nAll fetch operations are enabled.\n\n\n#### This default permission set includes:\n\n- `allow-fetch`\n- `allow-fetch-cancel`\n- `allow-fetch-read-body`\n- `allow-fetch-send`"
+          "markdownDescription": "This permission set configures what kind of\nfetch operations are available from the http plugin.\n\nThis enables all fetch operations but does not\nallow explicitly any origins to be fetched. This needs to\nbe manually configured before usage.\n\n#### Granted Permissions\n\nAll fetch operations are enabled.\n\n\n#### This default permission set includes:\n\n- `allow-fetch`\n- `allow-fetch-cancel`\n- `allow-fetch-send`\n- `allow-fetch-read-body`\n- `allow-fetch-cancel-body`"
         }
       ]
     }

plugins/http/permissions/schemas/schema.json

@@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize};
 use tauri::{
     async_runtime::Mutex,
     command,
-    ipc::{Channel, CommandScope, GlobalScope},
+    ipc::{CommandScope, GlobalScope},
     Manager, ResourceId, ResourceTable, Runtime, State, Webview,
 };
 use tokio::sync::oneshot::{channel, Receiver, Sender};
@@ -415,26 +415,42 @@ pub async fn fetch_send<R: Runtime>(
 pub async fn fetch_read_body<R: Runtime>(
     webview: Webview<R>,
     rid: ResourceId,
-    stream_channel: Channel<tauri::ipc::InvokeResponseBody>,
-) -> crate::Result<()> {
+) -> crate::Result<tauri::ipc::Response> {
     let res = {
-        let mut resources_table = webview.resources_table();
-        resources_table.take::<ReqwestResponse>(rid)?
+        let resources_table = webview.resources_table();
+        resources_table.get::<ReqwestResponse>(rid)?
     };
 
-    let mut res = Arc::into_inner(res).unwrap().0;
+    // SAFETY: we can access the inner value mutably
+    // because we are the only ones with a reference to it
+    // and we don't want to use `Arc::into_inner` because we want to keep the value in the table
+    // for potential future calls to `fetch_cancel_body`
+    let res_ptr = Arc::as_ptr(&res) as *mut ReqwestResponse;
+    let res = unsafe { &mut *res_ptr };
+    let res = &mut res.0;
 
-    // send response through IPC channel
-    while let Some(chunk) = res.chunk().await? {
-        let mut chunk = chunk.to_vec();
-        // append 0 to indicate we are not done yet
-        chunk.push(0);
-        stream_channel.send(tauri::ipc::InvokeResponseBody::Raw(chunk))?;
-    }
+    let Some(chunk) = res.chunk().await? else {
+        let mut resources_table = webview.resources_table();
+        resources_table.close(rid)?;
 
-    // send 1 to indicate we are done
-    stream_channel.send(tauri::ipc::InvokeResponseBody::Raw(vec![1]))?;
+        // return a response with a single byte to indicate that the body is empty
+        return Ok(tauri::ipc::Response::new(vec![1]));
+    };
+
+    let mut chunk = chunk.to_vec();
+    // append a 0 byte to indicate that the body is not empty
+    chunk.push(0);
 
+    Ok(tauri::ipc::Response::new(chunk))
+}
+
+#[command]
+pub async fn fetch_cancel_body<R: Runtime>(
+    webview: Webview<R>,
+    rid: ResourceId,
+) -> crate::Result<()> {
+    let mut resources_table = webview.resources_table();
+    resources_table.close(rid)?;
     Ok(())
 }
 

plugins/http/src/commands.rs

@@ -84,7 +84,8 @@ pub fn init<R: Runtime>() -> TauriPlugin<R> {
             commands::fetch,
             commands::fetch_cancel,
             commands::fetch_send,
-            commands::fetch_read_body
+            commands::fetch_read_body,
+            commands::fetch_cancel_body,
         ])
         .build()
 }