Merge 8c3932e992 into 4089002880

.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

@@ -1 +1 @@
-if("__TAURI__"in window){var __TAURI_PLUGIN_HTTP__=function(e){"use strict";function t(e,t,r,n){if("function"==typeof t||!t.has(e))throw new TypeError("Cannot read private member from an object whose class did not declare it");return"m"===r?n:"a"===r?n.call(e):n?n.value:t.get(e)}function r(e,t,r,n,s){if("function"==typeof t||!t.has(e))throw new TypeError("Cannot write private member to an object whose class did not declare it");return t.set(e,r),r}var n,s,i,a;"function"==typeof SuppressedError&&SuppressedError;const o="__TAURI_TO_IPC_KEY__";class c{constructor(e){n.set(this,void 0),s.set(this,0),i.set(this,[]),a.set(this,void 0),r(this,n,e||(()=>{})),this.id=function(e,t=!1){return window.__TAURI_INTERNALS__.transformCallback(e,t)}((e=>{const o=e.index;if("end"in e)return void(o==t(this,s,"f")?this.cleanupCallback():r(this,a,o));const c=e.message;if(o==t(this,s,"f")){for(t(this,n,"f").call(this,c),r(this,s,t(this,s,"f")+1);t(this,s,"f")in t(this,i,"f");){const e=t(this,i,"f")[t(this,s,"f")];t(this,n,"f").call(this,e),delete t(this,i,"f")[t(this,s,"f")],r(this,s,t(this,s,"f")+1)}t(this,s,"f")===t(this,a,"f")&&this.cleanupCallback()}else t(this,i,"f")[o]=c}))}cleanupCallback(){Reflect.deleteProperty(window,`_${this.id}`)}set onmessage(e){r(this,n,e)}get onmessage(){return t(this,n,"f")}[(n=new WeakMap,s=new WeakMap,i=new WeakMap,a=new WeakMap,o)](){return`__CHANNEL__:${this.id}`}toJSON(){return this[o]()}}async function d(e,t={},r){return window.__TAURI_INTERNALS__.invoke(e,t,r)}const h="Request cancelled";return e.fetch=async function(e,t){const r=t?.signal;if(r?.aborted)throw new Error(h);const n=t?.maxRedirections,s=t?.connectTimeout,i=t?.proxy,a=t?.danger;t&&(delete t.maxRedirections,delete t.connectTimeout,delete t.proxy,delete t.danger);const o=t?.headers?t.headers instanceof Headers?t.headers:new Headers(t.headers):new Headers,f=new Request(e,t),l=await f.arrayBuffer(),u=0!==l.byteLength?Array.from(new Uint8Array(l)):null;for(const[e,t]of f.headers)o.get(e)||o.set(e,t);const _=(o instanceof Headers?Array.from(o.entries()):Array.isArray(o)?o:Object.entries(o)).map((([e,t])=>[e,"string"==typeof t?t:t.toString()]));if(r?.aborted)throw new Error(h);const w=await d("plugin:http|fetch",{clientConfig:{method:f.method,url:f.url,headers:_,data:u,maxRedirections:n,connectTimeout:s,proxy:i,danger:a}}),p=()=>d("plugin:http|fetch_cancel",{rid:w});if(r?.aborted)throw p(),new Error(h);r?.addEventListener("abort",(()=>{p()}));const{status:y,statusText:m,url:b,headers:g,rid:T}=await d("plugin:http|fetch_send",{rid:w}),R=new ReadableStream({start:e=>{const t=new c;t.onmessage=t=>{if(r?.aborted)return void e.error(h);const n=new Uint8Array(t),s=n[n.byteLength-1],i=n.slice(0,n.byteLength-1);1!=s?e.enqueue(i):e.close()},d("plugin:http|fetch_read_body",{rid:T,streamChannel:t}).catch((t=>{e.error(t)}))}}),A=new Response(204!==y?R:null,{status:y,statusText:m});return Object.defineProperty(A,"url",{value:b}),Object.defineProperty(A,"headers",{value:new Headers(g)}),A},e}({});Object.defineProperty(window.__TAURI__,"http",{value:__TAURI_PLUGIN_HTTP__})}
+if("__TAURI__"in window){var __TAURI_PLUGIN_HTTP__=function(e){"use strict";async function t(e,t={},r){return window.__TAURI_INTERNALS__.invoke(e,t,r)}"function"==typeof SuppressedError&&SuppressedError;const r="Request cancelled";return e.fetch=async function(e,n){const a=n?.signal;if(a?.aborted)throw new Error(r);const o=n?.maxRedirections,s=n?.connectTimeout,i=n?.proxy,d=n?.danger;n&&(delete n.maxRedirections,delete n.connectTimeout,delete n.proxy,delete n.danger);const c=n?.headers?n.headers instanceof Headers?n.headers:new Headers(n.headers):new Headers,u=new Request(e,n),l=await u.arrayBuffer(),_=0!==l.byteLength?Array.from(new Uint8Array(l)):null;for(const[e,t]of u.headers)c.get(e)||c.set(e,t);const f=(c instanceof Headers?Array.from(c.entries()):Array.isArray(c)?c:Object.entries(c)).map((([e,t])=>[e,"string"==typeof t?t:t.toString()]));if(a?.aborted)throw new Error(r);const h=await t("plugin:http|fetch",{clientConfig:{method:u.method,url:u.url,headers:f,data:_,maxRedirections:o,connectTimeout:s,proxy:i,danger:d}}),p=()=>t("plugin:http|fetch_cancel",{rid:h});if(a?.aborted)throw p(),new Error(r);a?.addEventListener("abort",(()=>{p()}));const{status:y,statusText:w,url:b,headers:g,rid:T}=await t("plugin:http|fetch_send",{rid:h}),R=()=>t("plugin:http|fetch_cancel_body",{rid:T}),m=new ReadableStream({start:e=>{if(a?.aborted)return e.error(r),void R();a?.addEventListener("abort",(()=>{e.error(r),R()}))},pull:async e=>(async e=>{let r;try{r=await t("plugin:http|fetch_read_body",{rid:T})}catch(t){return e.error(t),void R()}const n=new Uint8Array(r),a=n[n.byteLength-1],o=n.slice(0,n.byteLength-1);1===a&&e.close(),e.enqueue(o)})(e)}),A=new Response(204!==y?m:null,{status:y,statusText:w});return Object.defineProperty(A,"url",{value:b}),Object.defineProperty(A,"headers",{value:new Headers(g)}),A},e}({});Object.defineProperty(window.__TAURI__,"http",{value:__TAURI_PLUGIN_HTTP__})}

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 = () => {
-    start: (controller) => {
+    return invoke('plugin:http|fetch_cancel_body', { rid: responseRid })
-      const streamChannel = new Channel<ArrayBuffer | number[]>()
+  }
-      streamChannel.onmessage = (res: ArrayBuffer | number[]) => {
+
-        // close early if aborted
+  const readChunk = async (
-        if (signal?.aborted) {
+    controller: ReadableStreamDefaultController<Uint8Array>
-          controller.error(ERROR_REQUEST_CANCELLED)
+  ) => {
+    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 resUint8 = new Uint8Array(res)
+    const dataUint8 = new Uint8Array(data)
-        const lastByte = resUint8[resUint8.byteLength - 1]
+    const lastByte = dataUint8[dataUint8.byteLength - 1]
-        const actualRes = resUint8.slice(0, resUint8.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) {
+    if (lastByte === 1) {
       controller.close()
-          return
     }
 
-        controller.enqueue(actualRes)
+    controller.enqueue(actualData)
   }
 
-      // run a non-blocking body stream fetch
+  const readableStreamBody = new ReadableStream<Uint8Array>({
-      invoke('plugin:http|fetch_read_body', {
+    start: (controller) => {
-        rid: responseRid,
+      // abort early here if needed and drop the body
-        streamChannel
+      if (signal?.aborted) {
-      }).catch((e) => {
+        controller.error(ERROR_REQUEST_CANCELLED)
-        controller.error(e)
+        void dropBody()
-      })
+        return
       }
+
+      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<tauri::ipc::Response> {
-) -> crate::Result<()> {
     let res = {
-        let mut resources_table = webview.resources_table();
+        let resources_table = webview.resources_table();
-        resources_table.take::<ReqwestResponse>(rid)?
+        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;
+
+    let Some(chunk) = res.chunk().await? else {
+        let mut resources_table = webview.resources_table();
+        resources_table.close(rid)?;
+
+        // return a response with a single byte to indicate that the body is empty
+        return Ok(tauri::ipc::Response::new(vec![1]));
+    };
 
-    // 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
+    // append a 0 byte to indicate that the body is not empty
     chunk.push(0);
-        stream_channel.send(tauri::ipc::InvokeResponseBody::Raw(chunk))?;
-    }
 
-    // send 1 to indicate we are done
+    Ok(tauri::ipc::Response::new(chunk))
-    stream_channel.send(tauri::ipc::InvokeResponseBody::Raw(vec![1]))?;
+}
 
+#[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()
 }