fix: Android OOM by eliminating intermediate buffer copies (#54854)

Summary:
During JS bundle downloads from Metro, the multipart stream reader was copying each chunk into a new Buffer before passing it to listeners. For large bundles, this resulted in elevated peak memory usage due to duplicating chunk data (Okio read buffer + intermediate Buffer copy, plus downstream buffering), which can exceed emulator heap limits for large bundles.
Example: #52818
Repro: #52797

### Changes
- Multipart parsing: pass a **bounded** `BufferedSource` per part (prevents reading past the part into the next boundary) and **drain unread bytes** after callbacks so listeners don’t need to fully consume the body.
- BundleDownloader: keep streaming download behavior while restoring **atomic writes** (`.tmp` + rename) to avoid partial bundles on interruption.
- Make Content-Type checks tolerant of parameters, parse `X-Http-Status` safely.

## Changelog:

[ANDROID] [FIXED] - Reduced memory usage during JS bundle downloads by eliminating intermediate buffer copies

Pull Request resolved: #54854

Test Plan:
# From OSS contributor
- [x] Verified multipart bundle downloads work correctly with progress callbacks displayed
- [x] Verified non-multipart fallback path still functions
- [x] Verified error responses are handled correctly
- [x] Tested with large bundles (repro above) and confirmed reduced memory pressure and no crashes

# E2E With Playground app (D102154809) and profiling

 {F1988731912}

### Memory regression test (Android)

Built a scripted A/B harness for `BundleDownloader.downloadBundleFromURL` using an RNTester playground native module that exposes the downloader to JS and to `adb shell am broadcast`. Per iteration: force-stop the app, relaunch, capture `VmHWM` from `/proc/<pid>/status` as the pre-download high-water mark, fire the broadcast, wait for a `DOWNLOAD_COMPLETE` logcat sentinel, capture `VmHWM` and `dumpsys meminfo` again. Compared this commit against its parent (`abf4662924`) on a Samsung Galaxy S22 (`SM-S901B`, 256 MB default Java heap).

**OOM regression test.** With a 200 MB synthetic JS bundle and the default 256 MB Java heap, the parent commit deterministically `OutOfMemoryError`s in `okio.Segment.<init>` from `MultipartStreamReader.readAllParts`. With this commit applied, the same download completes successfully. To get a numerical comparison at all I had to add `android:largeHeap="true"` and shrink the test bundle to 80 MB so the parent variant could finish.

**Steady-state memory** — 80 MB bundle, 5 iterations per variant, `largeHeap=true`:

| Variant | Avg post-download `VmHWM` | Avg delta from pre | Worst-case delta |
|---|---|---|---|---|
| After | 395 MB | +84 MB | +91 MB |
| Before | 394 MB | +103 MB | **+196 MB** |

`progressEvents = 0` for every parent-commit run confirms the parent code was the variant under test: its `headers["Content-Type"] == "application/javascript"` exact-match filter never fires against Metro's `application/javascript; charset=utf-8`. This patch tolerates the `; charset=…` parameter and emits ~150 events per download.

**Per-iteration data**

```
phase        iter  pre_vmhwm_kb  post_vmhwm_kb  delta_kb  duration_ms  progress_events
with-fix     1     313344        396796         83452     4386         149
with-fix     2     310344        388944         78600     4252         136
with-fix     3     313428        397092         83664     4602         150
with-fix     4     306804        398272         91468     4358         150
with-fix     5     309196        391764         82568     4410         149
without-fix  1     312332        393780         81448     3922         0
without-fix  2     196304        392668         196364    4621         0
without-fix  3     317616        393640         76024     4486         0
without-fix  4     314836        397812         82976     4362         0
without-fix  5     313224        392808         79584     3795         0
```

**Existing tests.** `buck2 test fbsource//xplat/js/react-native-github/packages/react-native/ReactAndroid/src/test/java/com/facebook/react/devsupport:devsupport_MultipartStreamReaderTestAndroid` passes, including the new `testListenerDoesNotNeedToFullyReadBody` and `testHeaderNamesAreCaseInsensitive` cases that cover the bounded `BufferedSource` listener contract and case-insensitive header lookup.

Reviewed By: GijsWeterings

Differential Revision: D93102028

Pulled By: robhogan

fbshipit-source-id: ec63d31cc6d72d2d4d852578072d810d3a54218d

Author: DorianMazur

packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/BundleDownloader.kt

@@ -26,7 +26,6 @@ import okhttp3.Headers
 import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.Response
-import okio.Buffer
 import okio.BufferedSource
 import okio.Okio
 import org.json.JSONException
@@ -121,11 +120,7 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
 
                   val url = resp.request().url().toString()
                   // Make sure the result is a multipart response and parse the boundary.
-                  var contentType = resp.header("content-type")
-                  if (contentType == null) {
-                    // fallback to empty string for nullability
-                    contentType = ""
-                  }
+                  val contentType = resp.header("content-type") ?: ""
                   val regex = Pattern.compile("multipart/mixed;.*boundary=\"([^\"]+)\"")
                   val match = regex.matcher(contentType)
                   if (contentType.isNotEmpty() && match.find()) {
@@ -135,18 +130,28 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
                     // In case the server doesn't support multipart/mixed responses, fallback to
                     // normal
                     // download.
-                    resp.body().use { body ->
-                      if (body != null) {
+                    val body = resp.body()
+                    if (body != null) {
+                      body.use {
                         processBundleResult(
                             url,
                             resp.code(),
                             resp.headers(),
-                            body.source(),
+                            it.source(),
                             outputFile,
                             bundleInfo,
                             callback,
                         )
                       }
+                    } else {
+                      callback.onFailure(
+                          makeGeneric(
+                              url,
+                              "Development server response body was empty.",
+                              "URL: $url",
+                              null,
+                          )
+                      )
                     }
                   }
                 }
@@ -164,7 +169,8 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
       bundleInfo: BundleInfo?,
       callback: DevBundleDownloadListener,
   ) {
-    if (response.body() == null) {
+    val responseBody = response.body()
+    if (responseBody == null) {
       callback.onFailure(
           DebugServerException(
               ("""
@@ -181,27 +187,25 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
       )
       return
     }
-    val source = checkNotNull(response.body()?.source())
+
+    val source = responseBody.source()
     val bodyReader = MultipartStreamReader(source, boundary)
     val completed =
         bodyReader.readAllParts(
             object : ChunkListener {
               @Throws(IOException::class)
               override fun onChunkComplete(
                   headers: Map<String, String>,
-                  body: Buffer,
+                  body: BufferedSource,
                   isLastChunk: Boolean,
               ) {
                 // This will get executed for every chunk of the multipart response. The last chunk
                 // (isLastChunk = true) will be the JS bundle, the other ones will be progress
-                // events
-                // encoded as JSON.
+                // events encoded as JSON.
                 if (isLastChunk) {
                   // The http status code for each separate chunk is in the X-Http-Status header.
-                  var status = response.code()
-                  if (headers.containsKey("X-Http-Status")) {
-                    status = headers.getOrDefault("X-Http-Status", "0").toInt()
-                  }
+                  val status = headers["X-Http-Status"]?.toIntOrNull() ?: response.code()
+
                   processBundleResult(
                       url,
                       status,
@@ -211,34 +215,26 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
                       bundleInfo,
                       callback,
                   )
-                } else {
-                  if (
-                      !headers.containsKey("Content-Type") ||
-                          headers["Content-Type"] != "application/json"
-                  ) {
-                    return
-                  }
+                  return
+                }
 
-                  try {
-                    val progress = JSONObject(body.readUtf8())
-                    val status =
-                        if (progress.has("status")) progress.getString("status") else "Bundling"
-                    var done: Int? = null
-                    if (progress.has("done")) {
-                      done = progress.getInt("done")
-                    }
-                    var total: Int? = null
-                    if (progress.has("total")) {
-                      total = progress.getInt("total")
-                    }
-                    var percent: Int? = null
-                    if (progress.has("percent")) {
-                      percent = progress.getInt("percent")
-                    }
-                    callback.onProgress(status, done, total, percent)
-                  } catch (e: JSONException) {
-                    FLog.e(ReactConstants.TAG, "Error parsing progress JSON. $e")
-                  }
+                val contentType = headers["Content-Type"] ?: return
+                if (!isJsonContentType(contentType)) {
+                  return
+                }
+
+                try {
+                  // Body is already bounded to this part; safe to read fully.
+                  val progress = JSONObject(body.readUtf8())
+                  val status =
+                      if (progress.has("status")) progress.getString("status") else "Bundling"
+                  val done: Int? = if (progress.has("done")) progress.getInt("done") else null
+                  val total: Int? = if (progress.has("total")) progress.getInt("total") else null
+                  val percent: Int? =
+                      if (progress.has("percent")) progress.getInt("percent") else null
+                  callback.onProgress(status, done, total, percent)
+                } catch (e: JSONException) {
+                  FLog.e(ReactConstants.TAG, "Error parsing progress JSON.", e)
                 }
               }
 
@@ -247,7 +243,8 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
                   loaded: Long,
                   total: Long,
               ) {
-                if ("application/javascript" == headers["Content-Type"]) {
+                val contentType = headers["Content-Type"] ?: return
+                if (isJavaScriptContentType(contentType)) {
                   callback.onProgress(
                       "Downloading",
                       (loaded / 1024).toInt(),
@@ -258,6 +255,7 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
               }
             }
         )
+
     if (!completed) {
       callback.onFailure(
           DebugServerException(
@@ -331,7 +329,7 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
 
     @Throws(IOException::class)
     private fun storePlainJSInFile(body: BufferedSource, outputFile: File): Boolean {
-      Okio.sink(outputFile).use { it -> body.readAll(it) }
+      Okio.sink(outputFile).use { sink -> body.readAll(sink) }
       return true
     }
 
@@ -348,5 +346,10 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
         }
       }
     }
+
+    private fun isJsonContentType(value: String): Boolean = value.startsWith("application/json")
+
+    private fun isJavaScriptContentType(value: String): Boolean =
+        value.startsWith("application/javascript")
   }
 }

packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/MultipartStreamReader.kt

@@ -10,10 +10,14 @@
 package com.facebook.react.devsupport
 
 import java.io.IOException
+import java.util.TreeMap
 import kotlin.math.max
 import okio.Buffer
 import okio.BufferedSource
 import okio.ByteString
+import okio.Okio
+import okio.Source
+import okio.Timeout
 
 /** Utility class to parse the body of a response of type multipart/mixed. */
 internal class MultipartStreamReader(
@@ -25,7 +29,7 @@ internal class MultipartStreamReader(
   interface ChunkListener {
     /** Invoked when a chunk of a multipart response is fully downloaded. */
     @Throws(IOException::class)
-    fun onChunkComplete(headers: Map<String, String>, body: Buffer, isLastChunk: Boolean)
+    fun onChunkComplete(headers: Map<String, String>, body: BufferedSource, isLastChunk: Boolean)
 
     /** Invoked as bytes of the current chunk are read. */
     @Throws(IOException::class)
@@ -44,12 +48,16 @@ internal class MultipartStreamReader(
     val closeDelimiter: ByteString = ByteString.encodeUtf8("$CRLF--$boundary--$CRLF")
     val headersDelimiter: ByteString = ByteString.encodeUtf8(CRLF + CRLF)
 
-    val bufferLen = 4 * 1024
+    // Buffer size for reading from network. Increased from 4KB to 16KB for better I/O
+    // throughput and fewer syscalls. For a 2MB bundle: ~128 reads instead of ~512.
+    // Memory impact is negligible (12KB increase) while I/O overhead is significantly reduced.
+    val bufferLen = 16 * 1024
     var chunkStart: Long = 0
     var bytesSeen: Long = 0
     val content = Buffer()
+
     var currentHeaders: Map<String, String>? = null
-    var currentHeadersLength: Long = 0
+    var currentBodyStartIndexInContent: Long = -1
 
     while (true) {
       var isCloseDelimiter = false
@@ -58,6 +66,7 @@ internal class MultipartStreamReader(
       // to allow for the edge case when the delimiter is cut by read call.
       val searchStart =
           max((bytesSeen - closeDelimiter.size()).toDouble(), chunkStart.toDouble()).toLong()
+
       var indexOfDelimiter = content.indexOf(delimiter, searchStart)
       if (indexOfDelimiter == -1L) {
         isCloseDelimiter = true
@@ -68,16 +77,17 @@ internal class MultipartStreamReader(
         bytesSeen = content.size()
 
         if (currentHeaders == null) {
-          val indexOfHeaders = content.indexOf(headersDelimiter, searchStart)
-          if (indexOfHeaders >= 0) {
-            source.read(content, indexOfHeaders)
+          val indexOfHeadersDelimiter = content.indexOf(headersDelimiter, searchStart)
+          if (indexOfHeadersDelimiter >= 0) {
             val headers = Buffer()
-            content.copyTo(headers, searchStart, indexOfHeaders - searchStart)
-            currentHeadersLength = headers.size() + headersDelimiter.size()
+            content.copyTo(headers, searchStart, indexOfHeadersDelimiter - searchStart)
             currentHeaders = parseHeaders(headers)
+            currentBodyStartIndexInContent =
+                indexOfHeadersDelimiter + headersDelimiter.size().toLong()
           }
         } else {
-          emitProgress(currentHeaders, content.size() - currentHeadersLength, false, listener)
+          val loaded = max(0L, content.size() - currentBodyStartIndexInContent)
+          emitProgress(currentHeaders, loaded, false, listener)
         }
 
         val bytesRead = source.read(content, bufferLen.toLong())
@@ -92,26 +102,30 @@ internal class MultipartStreamReader(
 
       // Ignore preamble
       if (chunkStart > 0) {
-        val chunk = Buffer()
+        if (currentHeaders != null && currentBodyStartIndexInContent >= 0) {
+          val loadedFinal = max(0L, chunkEnd - currentBodyStartIndexInContent)
+          emitProgress(currentHeaders, loadedFinal, true, listener)
+        }
         content.skip(chunkStart)
-        content.read(chunk, length)
-        emitProgress(currentHeaders, chunk.size() - currentHeadersLength, true, listener)
-        emitChunk(chunk, isCloseDelimiter, listener)
+        emitChunk(content, length, isCloseDelimiter, listener)
+
         currentHeaders = null
-        currentHeadersLength = 0
+        currentBodyStartIndexInContent = -1
       } else {
         content.skip(chunkEnd)
       }
       if (isCloseDelimiter) {
         return true
       }
+
       chunkStart = delimiter.size().toLong()
       bytesSeen = chunkStart
     }
   }
 
   private fun parseHeaders(data: Buffer): Map<String, String> {
-    val headers: MutableMap<String, String> = mutableMapOf()
+    // Header names are case-insensitive
+    val headers: MutableMap<String, String> = TreeMap(String.CASE_INSENSITIVE_ORDER)
     val text = data.readUtf8()
     val lines = text.split(CRLF.toRegex()).dropLastWhile { it.isEmpty() }.toTypedArray()
     for (line in lines) {
@@ -126,20 +140,81 @@ internal class MultipartStreamReader(
     return headers
   }
 
+  /**
+   * Emits a chunk to the listener. The `body` passed to the listener is bounded to the chunk body
+   * bytes, so the listener cannot accidentally read into the next boundary.
+   *
+   * Also drains any unread body bytes after the callback to keep parsing in sync.
+   */
   @Throws(IOException::class)
-  private fun emitChunk(chunk: Buffer, done: Boolean, listener: ChunkListener) {
+  private fun emitChunk(
+      content: Buffer,
+      chunkLength: Long,
+      done: Boolean,
+      listener: ChunkListener,
+  ) {
     val marker: ByteString = ByteString.encodeUtf8(CRLF + CRLF)
-    val indexOfMarker = chunk.indexOf(marker)
-    if (indexOfMarker == -1L) {
-      listener.onChunkComplete(emptyMap(), chunk, done)
-    } else {
-      val headers = Buffer()
-      val body = Buffer()
-      chunk.read(headers, indexOfMarker)
-      chunk.skip(marker.size().toLong())
-      chunk.readAll(body)
-      listener.onChunkComplete(parseHeaders(headers), body, done)
+    val indexOfMarker = content.indexOf(marker, 0)
+
+    if (indexOfMarker == -1L || indexOfMarker >= chunkLength) {
+      // No headers marker found inside the chunk. Treat the entire chunk as body.
+      val bodyLength = chunkLength
+      val body = Okio.buffer(FixedLengthSource(content, bodyLength))
+      try {
+        listener.onChunkComplete(emptyMap(), body, done)
+      } finally {
+        drainFully(body)
+      }
+      return
     }
+
+    // Headers exist.
+    val headersBuf = Buffer()
+    content.read(headersBuf, indexOfMarker)
+    content.skip(marker.size().toLong())
+    val headers = parseHeaders(headersBuf)
+
+    val maxBodyLength = chunkLength - indexOfMarker - marker.size().toLong()
+    val body = Okio.buffer(FixedLengthSource(content, maxBodyLength))
+    try {
+      listener.onChunkComplete(headers, body, done)
+    } finally {
+      drainFully(body)
+    }
+  }
+
+  private fun drainFully(body: BufferedSource) {
+    // Drain remaining bytes from this part body (if listener didn't).
+    // Use small reusable buffer to avoid unbounded memory.
+    val tmp = Buffer()
+    try {
+      while (true) {
+        val r = body.read(tmp, 8 * 1024L)
+        if (r == -1L) break
+        tmp.clear()
+      }
+    } catch (_: IOException) {
+      // Best-effort drain; parsing will likely fail upstream anyway.
+    }
+  }
+
+  private class FixedLengthSource(
+      private val upstream: Buffer,
+      private var remaining: Long,
+  ) : Source {
+    override fun read(sink: Buffer, byteCount: Long): Long {
+      if (byteCount == 0L) return 0L
+      if (remaining == 0L) return -1L
+      val toRead = minOf(byteCount, remaining)
+      val read = upstream.read(sink, toRead)
+      if (read == -1L) return -1L
+      remaining -= read
+      return read
+    }
+
+    override fun timeout(): Timeout = Timeout.NONE
+
+    override fun close() = Unit
   }
 
   @Throws(IOException::class)