[fetch] Support streaming fetch requests. (emscripten-core#25383)

Add a new `FETCH_STREAMING` setting that enables using the DOM fetch API
to stream data when paired with `EMSCRIPTEN_FETCH_STREAM_DATA`. When
`EMSCRIPTEN_FETCH_STREAM_DATA` is not used we will default to a regular
XMLHttpRequest. To keep the emscripten Fetch.js code the same I've
implemented a polyfill of XMLHttpRequest using Fetch. To support
streaming I wired up the fetch code to use the old onprogress code that
old versions of Firefox supported.

Most of the current API is supported with some notable exceptions:
 - synchronous requests
 - overriding the mime type

I also changed a few of the tests to support both sync and async so I
could test them with the fetch backend.

Author: brendandahl

ChangeLog.md

@@ -24,6 +24,8 @@ See docs/process.md for more on how version tagging works.
   `--proxy-to-worker` flag) was removed due to lack of usage.  If you were
   depending on this feature but missed the PSA, please let us know about your
   use case. (#25645, #25440)
+- The fetch library now supports streaming data requests when
+  `-sFETCH_STREAMING` is enabled.
 
 4.0.20 - 11/18/25
 -----------------

site/source/docs/tools_reference/settings_reference.rst

@@ -2785,6 +2785,26 @@ If nonzero, enables emscripten_fetch API.
 
 Default value: false
 
+.. _fetch_streaming:
+
+FETCH_STREAMING
+===============
+
+Enables streaming fetched data when the fetch attribute
+EMSCRIPTEN_FETCH_STREAM_DATA is used. For streaming requests, the DOM Fetch
+API is used otherwise XMLHttpRequest is used.
+Both modes generally support the same API, but there are some key
+differences:
+
+ - XHR supports synchronous requests
+ - XHR supports overriding mime types
+ - Fetch supports streaming data using the 'onprogress' callback
+
+If set to a value of 2, only the DOM Fetch backend will be used. This should
+only be used in testing.
+
+Default value: 0
+
 .. _wasmfs:
 
 WASMFS

src/Fetch.js

@@ -4,6 +4,243 @@
  * SPDX-License-Identifier: MIT
  */
 
+#if FETCH_STREAMING
+/**
+ * A class that mimics the XMLHttpRequest API using the modern Fetch API.
+ * This implementation is specifically tailored to only handle 'arraybuffer'
+ * responses.
+ */
+// TODO Use a regular class name when #5840 is fixed.
+var FetchXHR = class {
+  // --- Public XHR Properties ---
+
+  // Event Handlers
+  onload = null;
+  onerror = null;
+  onprogress = null;
+  onreadystatechange = null;
+  ontimeout = null;
+
+  // Request Configuration
+  responseType = 'arraybuffer';
+  withCredentials = false;
+  timeout = 0; // Standard XHR timeout property
+
+  // Response / State Properties
+  readyState = 0; // 0: UNSENT
+  response = null;
+  responseURL = '';
+  status = 0;
+  statusText = '';
+
+  // --- Internal Properties ---
+  _method = '';
+  _url = '';
+  _headers = {};
+  _abortController = null;
+  _aborted = false;
+  _responseHeaders = null;
+
+  // --- Private state management ---
+  _changeReadyState(state) {
+    this.readyState = state;
+    this.onreadystatechange?.();
+  }
+
+  // --- Public XHR Methods ---
+
+  /**
+   * Initializes a request.
+   * @param {string} method The HTTP request method (e.g., 'GET', 'POST').
+   * @param {string} url The URL to send the request to.
+   * @param {boolean} [async=true] This parameter is ignored as Fetch is always async.
+   * @param {string|null} [user=null] The username for basic authentication.
+   * @param {string|null} [password=null] The password for basic authentication.
+   */
+  open(method, url, async = true, user = null, password = null) {
+    if (this.readyState !== 0 && this.readyState !== 4) {
+      console.warn("FetchXHR.open() called while a request is in progress.");
+      this.abort();
+    }
+
+    // Reset internal state for the new request
+    this._method = method;
+    this._url = url;
+    this._headers = {};
+    this._responseHeaders = null;
+
+    // The async parameter is part of the XHR API but is an error here because
+    // the Fetch API is inherently asynchronous and does not support synchronous requests.
+    if (!async) {
+      throw new Error("FetchXHR does not support synchronous requests.");
+    }
+
+    // Handle Basic Authentication if user/password are provided.
+    // This creates a base64-encoded string and sets the Authorization header.
+    if (user) {
+      const credentials = btoa(`${user}:${password || ''}`);
+      this._headers['Authorization'] = `Basic ${credentials}`;
+    }
+
+    this._changeReadyState(1); // 1: OPENED
+  }
+
+  /**
+   * Sets the value of an HTTP request header.
+   * @param {string} header The name of the header.
+   * @param {string} value The value of the header.
+   */
+  setRequestHeader(header, value) {
+    if (this.readyState !== 1) {
+      throw new Error('setRequestHeader can only be called when state is OPENED.');
+    }
+    this._headers[header] = value;
+  }
+
+  /**
+   * This method is not effectively implemented because Fetch API relies on the
+   * server's Content-Type header and does not support overriding the MIME type
+   * on the client side in the same way as XHR.
+   * @param {string} mimetype The MIME type to use.
+   */
+  overrideMimeType(mimetype) {
+    throw new Error("overrideMimeType is not supported by the Fetch API and has no effect.");
+  }
+
+  /**
+   * Returns a string containing all the response headers, separated by CRLF.
+   * @returns {string} The response headers.
+   */
+  getAllResponseHeaders() {
+    if (!this._responseHeaders) {
+      return '';
+    }
+
+    let headersString = '';
+    // The Headers object is iterable.
+    for (const [key, value] of this._responseHeaders.entries()) {
+      headersString += `${key}: ${value}\r\n`;
+    }
+    return headersString;
+  }
+
+  /**
+   * Sends the request.
+   * @param body The body of the request.
+   */
+  async send(body = null) {
+    if (this.readyState !== 1) {
+      throw new Error('send() can only be called when state is OPENED.');
+    }
+
+    this._abortController = new AbortController();
+    const signal = this._abortController.signal;
+
+    // Handle timeout
+    let timeoutID;
+    if (this.timeout > 0) {
+      timeoutID = setTimeout(
+        () => this._abortController.abort(new DOMException('The user aborted a request.', 'TimeoutError')),
+        this.timeout
+      );
+    }
+
+    const fetchOptions = {
+      method: this._method,
+      headers: this._headers,
+      body: body,
+      signal: signal,
+      credentials: this.withCredentials ? 'include' : 'same-origin',
+    };
+
+    try {
+      const response = await fetch(this._url, fetchOptions);
+
+      // Populate response properties once headers are received
+      this.status = response.status;
+      this.statusText = response.statusText;
+      this.responseURL = response.url;
+      this._responseHeaders = response.headers;
+      this._changeReadyState(2); // 2: HEADERS_RECEIVED
+
+      // Start processing the body
+      this._changeReadyState(3); // 3: LOADING
+
+      if (!response.body) {
+        throw new Error("Response has no body to read.");
+      }
+
+      const reader = response.body.getReader();
+      const contentLength = +response.headers.get('Content-Length');
+
+      let receivedLength = 0;
+      const chunks = [];
+
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) {
+          break;
+        }
+
+        chunks.push(value);
+        receivedLength += value.length;
+
+        if (this.onprogress) {
+          // Convert to ArrayBuffer as requested by responseType.
+          this.response = value.buffer;
+          const progressEvent = {
+            lengthComputable: contentLength > 0,
+            loaded: receivedLength,
+            total: contentLength
+          };
+          this.onprogress(progressEvent);
+        }
+      }
+
+      // Combine chunks into a single Uint8Array.
+      const allChunks = new Uint8Array(receivedLength);
+      let position = 0;
+      for (const chunk of chunks) {
+        allChunks.set(chunk, position);
+        position += chunk.length;
+      }
+
+      // Convert to ArrayBuffer as requested by responseType
+      this.response = allChunks.buffer;
+    } catch (error) {
+      this.statusText = error.message;
+
+      if (error.name === 'AbortError') {
+        // Do nothing.
+      } else if (error.name === 'TimeoutError') {
+        this.ontimeout?.();
+      } else {
+        // This is a network error
+        this.onerror?.();
+      }
+    } finally {
+      clearTimeout(timeoutID);
+      if (!this._aborted) {
+        this._changeReadyState(4); // 4: DONE
+        // The XHR 'load' event fires for successful HTTP statuses (2xx) as well as
+        // unsuccessful ones (4xx, 5xx). The 'error' event is for network failures.
+        this.onload?.();
+      }
+    }
+  }
+
+  /**
+   * Aborts the request if it has already been sent.
+   */
+  abort() {
+    this._aborted = true;
+    this.status = 0;
+    this._changeReadyState(4); // 4: DONE
+    this._abortController?.abort();
+  }
+}
+#endif
+
 var Fetch = {
   // HandleAllocator for XHR request object
   // xhrs: undefined,
@@ -267,7 +504,18 @@ function fetchXHR(fetch, onsuccess, onerror, onprogress, onreadystatechange) {
   var userNameStr = userName ? UTF8ToString(userName) : undefined;
   var passwordStr = password ? UTF8ToString(password) : undefined;
 
+#if FETCH_STREAMING == 1
+  if (fetchAttrStreamData) {
+    var xhr = new FetchXHR();
+  } else {
+    var xhr = new XMLHttpRequest();
+  }
+#elif FETCH_STREAMING == 2
+  // This setting forces using FetchXHR for all requests. Used only in testing.
+  var xhr = new FetchXHR();
+#else
   var xhr = new XMLHttpRequest();
+#endif
   xhr.withCredentials = !!{{{ makeGetValue('fetch_attr', C_STRUCTS.emscripten_fetch_attr_t.withCredentials, 'u8') }}};;
 #if FETCH_DEBUG
   dbg(`fetch: xhr.timeout: ${xhr.timeout}, xhr.withCredentials: ${xhr.withCredentials}`);
@@ -276,8 +524,8 @@ function fetchXHR(fetch, onsuccess, onerror, onprogress, onreadystatechange) {
   xhr.open(requestMethod, url_, !fetchAttrSynchronous, userNameStr, passwordStr);
   if (!fetchAttrSynchronous) xhr.timeout = timeoutMsecs; // XHR timeout field is only accessible in async XHRs, and must be set after .open() but before .send().
   xhr.url_ = url_; // Save the url for debugging purposes (and for comparing to the responseURL that server side advertised)
-#if ASSERTIONS
-  assert(!fetchAttrStreamData, 'streaming uses moz-chunked-arraybuffer which is no longer supported; TODO: rewrite using fetch()');
+#if ASSERTIONS && !FETCH_STREAMING
+  assert(!fetchAttrStreamData, 'Streaming is only supported when FETCH_STREAMING is enabled.');
 #endif
   xhr.responseType = 'arraybuffer';
 

src/lib/libfetch.js

@@ -22,6 +22,9 @@ var LibraryFetch = {
   $fetchCacheData: fetchCacheData,
 #endif
   $fetchXHR: fetchXHR,
+#if FETCH_STREAMING
+  $FetchXHR: FetchXHR,
+#endif
 
   emscripten_start_fetch: startFetch,
   emscripten_start_fetch__deps: [
@@ -38,7 +41,10 @@ var LibraryFetch = {
     '$fetchLoadCachedData',
     '$fetchDeleteCachedData',
 #endif
-  ]
+#if FETCH_STREAMING
+    '$FetchXHR',
+#endif
+  ],
 };
 
 addToLibrary(LibraryFetch);

src/settings.js

@@ -1823,6 +1823,21 @@ var FETCH_DEBUG = false;
 // [link]
 var FETCH = false;
 
+// Enables streaming fetched data when the fetch attribute
+// EMSCRIPTEN_FETCH_STREAM_DATA is used. For streaming requests, the DOM Fetch
+// API is used otherwise XMLHttpRequest is used.
+// Both modes generally support the same API, but there are some key
+// differences:
+//
+//  - XHR supports synchronous requests
+//  - XHR supports overriding mime types
+//  - Fetch supports streaming data using the 'onprogress' callback
+//
+// If set to a value of 2, only the DOM Fetch backend will be used. This should
+// only be used in testing.
+// [link]
+var FETCH_STREAMING = 0;
+
 // ATTENTION [WIP]: Experimental feature. Please use at your own risk.
 // This will eventually replace the current JS file system implementation.
 // If set to 1, uses new filesystem implementation.

test/decorators.py

@@ -404,6 +404,21 @@ def metafunc(self, with_wasm64, *args, **kwargs):
   return metafunc
 
 
+def also_with_fetch_streaming(f):
+  assert callable(f)
+
+  @wraps(f)
+  def metafunc(self, with_fetch, *args, **kwargs):
+    if with_fetch:
+      self.set_setting('FETCH_STREAMING', '2')
+      self.cflags += ['-DSKIP_SYNC_FETCH_TESTS']
+    f(self, *args, **kwargs)
+
+  parameterize(metafunc, {'': (False,),
+                          'fetch_backend': (True,)})
+  return metafunc
+
+
 def also_with_wasm2js(func):
   assert callable(func)
 

test/fetch/test_fetch_redirect.c

@@ -69,11 +69,13 @@ void start_next_async_fetch() {
     async_method_idx++;
     if (async_method_idx >= num_methods) {
       // All async tests done, now run sync tests
+#ifndef SKIP_SYNC_FETCH_TESTS
       for (int m = 0; m < num_methods; ++m) {
         for (int i = 0; i < num_codes; ++i) {
           fetchSyncTest(redirect_codes[i], methods[m]);
         }
       }
+#endif
       exit(0);
     }
   }