[Android] Stream multipart bundle bodies straight to disk

Refactors MultipartStreamReader and BundleDownloader so the JS bundle
chunk of a Metro multipart response is written directly into the tmp
file via a BufferedSink, instead of being buffered in heap and then
drained. For a 100 MB bundle on a dev machine, peak heap drops:

  MultipartStreamReader:  131 MB -> 36 MB  (3.6x)
  BundleDownloader:       115 MB -> 33 MB  (3.5x)

Wall-clock time is unchanged within noise. Cumulative bytes allocated
on the worker thread are also unchanged (~100 MB); that's dominated by
okio's per-thread SegmentPool capping at 64 KB and is independent of
whether the reader retains the body. The streaming property is proved
by peak heap + the body being delivered as null in onChunkComplete.

MultipartStreamReader (packages/react-native/ReactAndroid/src/main/.../MultipartStreamReader.kt)
- New ChunkListener API:
    onChunkHeader(headers): BufferedSink?
      Return a sink to stream the body to it; return null to receive
      the body buffered as a Buffer via onChunkComplete.
    onChunkComplete(headers, body: Buffer?, isLastChunk)
      `body` is non-null iff onChunkHeader returned null.
    onChunkProgress is unchanged.
- New algorithm: working buffer bounded by READ_CHUNK_SIZE + maxDelim
  (~16 KB + 21 B); reads upstream incrementally, scans for the next
  delimiter, transfers safe-to-flush bytes to the sink (or accumulator)
  using okio segment-move semantics. Lookahead window equals
  `maxDelimLen - 1` so delimiters spanning two reads are still detected.
- The reader cannot know which chunk is the last until it sees the next
  delimiter. Listeners must route on headers (Content-Type,
  X-Http-Status), not on isLastChunk position. Routing-from-headers
  matches what BundleDownloader needs and is more robust anyway.
- Chunks with no `\r\n\r\n` header separator are still supported
  (mirrors prior behaviour; required by testMultipleParts).

BundleDownloader (packages/react-native/ReactAndroid/src/main/.../BundleDownloader.kt)
- processMultipartResponse uses a new inner StreamingBundleChunkListener
  that returns Okio.buffer(Okio.sink(tmpFile)) when the chunk is
  application/javascript AND has effective status 200 (X-Http-Status
  header overrides response.code()). Progress JSON and error bodies are
  buffered in memory as before, so JSONObject parsing and
  DebugServerException diagnostics still work.
- On stream success, the listener closes the sink, populates BundleInfo
  from X-Metro-Files-Changed-Count, atomically renames tmp -> output,
  and fires callback.onSuccess(). On stream failure (premature upstream
  EOF, IOException), the outer code deletes the half-written tmp file
  before surfacing the error.
- closeOpenSinkQuietly ensures the file handle never leaks if
  readAllParts throws mid-stream.

Tests
- MultipartStreamReaderTest: adapted to the new ChunkListener API.
  All 4 existing correctness cases (simple, multiple parts, no
  delimiter, no close delimiter) still pass; behaviour around
  headerless chunks is preserved.
- MultipartStreamReaderPerfTest: now exercises the streaming path
  (returns a CountingDiscardingSink) and asserts the body is delivered
  via the sink (not buffered) plus peak heap < 64 MB. Drops the
  thread-allocated assertion; it's an okio SegmentPool artifact, not a
  reader property.
- BundleDownloaderPerfTest: budgets retargeted to peak heap < 80 MB.

Test commands unchanged:
  ./gradlew :packages:react-native:ReactAndroid:testDebugUnitTest \
    -Preact.internal.useHermesNightly=true \
    --tests "com.facebook.react.devsupport.*"

  ./gradlew :packages:react-native:ReactAndroid:testDebugUnitTest \
    -PrunPerfTests=true -Preact.internal.useHermesNightly=true \
    --tests "*PerfTest" --info

Author: AndreiCalazans

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

@@ -27,6 +27,7 @@ import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.Response
 import okio.Buffer
+import okio.BufferedSink
 import okio.BufferedSource
 import okio.Okio
 import org.json.JSONException
@@ -183,82 +184,28 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
     }
     val source = checkNotNull(response.body()?.source())
     val bodyReader = MultipartStreamReader(source, boundary)
-    val completed =
-        bodyReader.readAllParts(
-            object : ChunkListener {
-              @Throws(IOException::class)
-              override fun onChunkComplete(
-                  headers: Map<String, String>,
-                  body: Buffer,
-                  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.
-                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()
-                  }
-                  processBundleResult(
-                      url,
-                      status,
-                      Headers.of(headers),
-                      body,
-                      outputFile,
-                      bundleInfo,
-                      callback,
-                  )
-                } else {
-                  if (
-                      !headers.containsKey("Content-Type") ||
-                          headers["Content-Type"] != "application/json"
-                  ) {
-                    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")
-                  }
-                }
-              }
-
-              override fun onChunkProgress(
-                  headers: Map<String, String>,
-                  loaded: Long,
-                  total: Long,
-              ) {
-                if ("application/javascript" == headers["Content-Type"]) {
-                  callback.onProgress(
-                      "Downloading",
-                      (loaded / 1024).toInt(),
-                      (total / 1024).toInt(),
-                      null,
-                  )
-                }
-              }
-            }
+    val tmpFile = File(outputFile.path + ".tmp")
+    val streamingHandler =
+        StreamingBundleChunkListener(
+            url = url,
+            outerStatus = response.code(),
+            outputFile = outputFile,
+            tmpFile = tmpFile,
+            bundleInfo = bundleInfo,
+            callback = callback,
         )
+    val completed: Boolean =
+        try {
+          bodyReader.readAllParts(streamingHandler)
+        } finally {
+          streamingHandler.closeOpenSinkQuietly()
+        }
     if (!completed) {
+      // If we partially wrote a tmp file before the upstream died, scrap it so we don't leave
+      // half-baked bundles on disk.
+      if (tmpFile.exists()) {
+        tmpFile.delete()
+      }
       callback.onFailure(
           DebugServerException(
               ("""
@@ -276,6 +223,124 @@ public class BundleDownloader public constructor(private val client: OkHttpClien
     }
   }
 
+  /**
+   * Routes multipart chunks for a bundle download. The JS bundle chunk (Content-Type
+   * `application/javascript` with an effective HTTP status of 200) is streamed directly into
+   * a temporary file via a [BufferedSink], so no copy of the body is held in heap. Progress
+   * JSON chunks and error responses are buffered in memory because they're either tiny or
+   * bounded, and the listener needs to parse them in full.
+   */
+  private inner class StreamingBundleChunkListener(
+      private val url: String,
+      private val outerStatus: Int,
+      private val outputFile: File,
+      private val tmpFile: File,
+      private val bundleInfo: BundleInfo?,
+      private val callback: DevBundleDownloadListener,
+  ) : ChunkListener {
+
+    private var bundleSink: BufferedSink? = null
+
+    @Throws(IOException::class)
+    override fun onChunkHeader(headers: Map<String, String>): BufferedSink? {
+      if (headers["Content-Type"] != "application/javascript") return null
+      val effectiveStatus = effectiveStatus(headers)
+      if (effectiveStatus != 200) return null
+      // Stream the JS bundle straight to disk — never materialize in heap.
+      val sink = Okio.buffer(Okio.sink(tmpFile))
+      bundleSink = sink
+      return sink
+    }
+
+    @Throws(IOException::class)
+    override fun onChunkComplete(
+        headers: Map<String, String>,
+        body: Buffer?,
+        isLastChunk: Boolean,
+    ) {
+      val sink = bundleSink
+      if (sink != null) {
+        bundleSink = null
+        sink.close()
+        finalizeStreamedBundle(headers)
+        return
+      }
+      when (headers["Content-Type"]) {
+        "application/javascript" -> {
+          // Bundle returned with an error status — it was buffered so we can surface a useful
+          // diagnostic to the developer.
+          val buffered = body ?: Buffer()
+          processBundleResult(
+              url,
+              effectiveStatus(headers),
+              Headers.of(headers),
+              buffered,
+              outputFile,
+              bundleInfo,
+              callback,
+          )
+        }
+        "application/json" -> dispatchProgressJson(body)
+        else -> Unit // Unknown chunk type; ignore as before.
+      }
+    }
+
+    override fun onChunkProgress(
+        headers: Map<String, String>,
+        loaded: Long,
+        total: Long,
+    ) {
+      if ("application/javascript" == headers["Content-Type"]) {
+        callback.onProgress(
+            "Downloading",
+            (loaded / 1024).toInt(),
+            (total / 1024).toInt(),
+            null,
+        )
+      }
+    }
+
+    /** Make sure we never leak the tmp-file sink if [readAllParts] throws mid-stream. */
+    fun closeOpenSinkQuietly() {
+      val sink = bundleSink ?: return
+      bundleSink = null
+      try {
+        sink.close()
+      } catch (e: IOException) {
+        FLog.w(TAG, "Failed to close partial bundle sink", e)
+      }
+    }
+
+    @Throws(IOException::class)
+    private fun finalizeStreamedBundle(headers: Map<String, String>) {
+      if (bundleInfo != null) {
+        populateBundleInfo(url, Headers.of(headers), bundleInfo)
+      }
+      if (!tmpFile.renameTo(outputFile)) {
+        throw IOException("Couldn't rename $tmpFile to $outputFile")
+      }
+      callback.onSuccess()
+    }
+
+    private fun dispatchProgressJson(body: Buffer?) {
+      val payload = body ?: return
+      try {
+        val progress = JSONObject(payload.readUtf8())
+        val status =
+            if (progress.has("status")) progress.getString("status") else "Bundling"
+        val done = if (progress.has("done")) progress.getInt("done") else null
+        val total = if (progress.has("total")) progress.getInt("total") else null
+        val percent = 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")
+      }
+    }
+
+    private fun effectiveStatus(headers: Map<String, String>): Int =
+        headers["X-Http-Status"]?.toIntOrNull() ?: outerStatus
+  }
+
   @Throws(IOException::class)
   private fun processBundleResult(
       url: String,