fix(http): fix aborting a streaming response

.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/build.rs

@@ -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/guest-js/index.ts

@@ -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.
@@ -229,37 +229,53 @@ export async function fetch(
     rid
   })
 
-  const readableStreamBody = new ReadableStream({
+  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()
+    }
+
+    controller.enqueue(actualData)
+  }
+
+  const readableStreamBody = 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)
+      // abort early here if needed and drop the body
+      if (signal?.aborted) {
+        controller.error(ERROR_REQUEST_CANCELLED)
+        void dropBody()
+        return
       }
 
-      // run a non-blocking body stream fetch
-      invoke('plugin:http|fetch_read_body', {
-        rid: responseRid,
-        streamChannel
-      }).catch((e) => {
-        controller.error(e)
+      signal?.addEventListener('abort', () => {
+        controller.error(ERROR_REQUEST_CANCELLED)
+        void dropBody()
       })
-    }
+    },
+    pull: async (controller) => readChunk(controller)
   })
 
   const res = new Response(status !== 204 ? readableStreamBody : null, {

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

@@ -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/reference.md

@@ -17,8 +17,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
 
@@ -84,6 +85,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/default.toml

@@ -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/schemas/schema.json

@@ -318,6 +318,16 @@
           "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"
+        },
+        {
+          "description": "Denies the fetch_cancel_body command without any pre-configured scope.",
+          "type": "string",
+          "const": "deny-fetch-cancel-body"
+        },
         {
           "description": "Enables the fetch_read_body command without any pre-configured scope.",
           "type": "string",

plugins/http/src/commands.rs

@@ -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/lib.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()
 }