req 0.7.2: Measure process ownership, mailbox cleanup, and status-dependent response streaming in Req 0.7.2

# Req Resource Handling: Streams, Process Mailboxes, and Collectables

## Search Known Solution
`search_known_solution` answered `NO_SAFE_MATCH` for `req` with this resource handling query.

## Failure Mode
The naive implementation fails loudly with a `RuntimeError` when delegating `into: :self` streams to background worker processes, and silently with an unwritten file and in-memory body accumulation when using `into: collectable` on non-200 HTTP responses.

## Library Behavior

### 1. Process Mailbox Binding with `into: :self`
When `Req.get(url, into: :self)` is called:
- The adapter sends response chunk messages (`{ref, {:data, chunk}}`) and the completion message (`{ref, :done}`) directly into the requesting process mailbox (`self()`).
- The returned `Req.Response.Async` struct records `pid: self()`.
- If the caller passes `resp.body` to a worker process (such as a `Task` or `GenServer`), attempting to enumerate it raises `RuntimeError: expected to read body chunk in the process <parent_pid> which made the request, got: <worker_pid>`.
- Body chunks cannot be read by another process because they are already queued in the parent process mailbox.

### 2. Mailbox Accumulation and Stream Cancellation
- If a request is made with `into: :self` but the caller never iterates `resp.body`, the chunk and `:done` messages remain trapped in the caller process mailbox indefinitely.
- If the caller halts enumeration early (e.g. using `Enum.take(resp.body, n)`), `Req.Response.Async` catches `:halt` and executes `cancel_fun`, cleanly draining any remaining queued chunk messages for that request `ref` from the mailbox.

### 3. Silent Collectable Bypass on Non-200 Status Codes
- When streaming into a `Collectable` (such as `into: File.stream!("data/download.bin")`), `Req` only invokes the collectable collector if the HTTP response status is exactly `200`.
- For any non-200 status code (such as `404`, `500`, `201`, or `206`), `Req` completely bypasses the `Collectable`, leaves target files untouched, and buffers the full payload in memory as a binary in `response.body`.
- Callers requiring custom streaming across all response status codes must use a callback function (`into: fn {:data, chunk}, acc -> ... end`) rather than a `Collectable`.

{"buildCommand":["mix","deps.compile","--no-deps-check"],"case":{"believed":"An into: :self response body can be enumerated from another process, and a Collectable receives response bytes for every status code.","caseId":"case:sha256:db71b7ef651ef07601ddd8e14167d8f987a50ac72d8e7917a99c3f41078ae9f9","contract":["assert an into: :self body is a Req.Response.Async tied to the requester pid and enumeration in another process raises RuntimeError","assert the requester can consume the same three queued chunks after the worker fails","assert an unconsumed three-chunk stream leaves three data messages and one done message for its ref in the requester mailbox","assert halting after two chunks drains the remaining messages for that response ref","assert a File stream receives a tested 200 body but a tested 404 body remains buffered and does not create the file","assert an into callback receives both chunks from the tested 500 response"],"goal":"Measure process ownership, mailbox cleanup, and status-dependent response streaming in Req 0.7.2","kind":"HOW","packages":["pkg:hex/req@0.7.2","pkg:hex/plug@1.20.3"],"schemaVersion":1,"symbols":["Req.get!","Req.Response.Async","Plug.Conn.send_chunked","Plug.Conn.chunk","Plug.Conn.send_resp"]},"contractCommand":["mix","test","--no-deps-check"],"environment":{"arch":"x64","ecosystem":"hex","executionContext":"elixir","language":"elixir","os":"linux","packageManager":"mix","runtime":"elixir","runtimeVersion":"1","schemaVersion":1},"license":"MIT-0","packages":["pkg:hex/req@0.7.2","pkg:hex/plug@1.20.3"],"schemaVersion":1,"symbols":["Req.get!","Req.Response.Async","Plug.Conn.send_chunked","Plug.Conn.chunk","Plug.Conn.send_resp"],"verifierAdapter":"hex@1"}

defmodule ReqMerge.MixProject do
  use Mix.Project

  def project do
    [
      app: :req_merge,
      version: "0.1.0",
      elixir: "~> 1.15",
      deps: deps()
    ]
  end

  def application, do: [extra_applications: []]

  defp deps do
    # plug is optional in req but required for Req.Test helpers.
    # Omitting it produces a runtime raise instead of a compile error.
    [{:req, "0.7.2"}, {:plug, "1.20.3"}]
  end
end

%{
  "finch": {:hex, :finch, "0.23.0", "e3f9287ac25a8832f848b144c2b57346aac65b205e2e0629a52adfe6507fd837", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.8", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "80e58d3f936f57e3fdf404f83a3642897ae6d9fb642934e46da4d8fe761b99d5"},
  "hpax": {:hex, :hpax, "1.0.4", "777de5d433b0fbdc7c418159c8055910faa8047ffdb3d6b31098d2a46cd7685c", [:mix], [], "hexpm", "afc7cb142ebcc2d01ce7816190b98ce5dd49e799111b24249f3443d730f377ca"},
  "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"},
  "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"},
  "mint": {:hex, :mint, "1.9.3", "3337184d69179695c7a9f1714d92c11e629d36c8c037a21cf490131d3d150554", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "5f7c9342480c069dbbc4eeac3490303c9e01870ff01a7f1d29b6107054fc1e74"},
  "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"},
  "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},
  "plug": {:hex, :plug, "1.20.3", "56c480c633ec2ce10140e236e15233bf576e1d323887d7c96711bd02ab5160db", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "be266aee1b8536ef6409d58cf39a3121319f0ec47cfa1b24024485aa0e76ad76"},
  "plug_crypto": {:hex, :plug_crypto, "2.2.0", "144014737daaf485407f5ed77daeaad74d651b216a28c87543f8cc7043f8efc8", [:mix], [], "hexpm", "83a95744ab1c75876542b6fab135fcc176280e0f301a111c1f757fddcec95d2c"},
  "req": {:hex, :req, "0.7.2", "364eae2e5f5c984f2dac6d71c07f8c8c89ce0bc49c4d746dacb7a306823020de", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:finch, "~> 0.21", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "c9cdfa276b05d8db2a27fda5d233e6858b764d47189d76cbb186e130a871ae0b"},
  "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"},
}

defmodule ReqResourceHandlingTest do
  use ExUnit.Case, async: false

  # Plug helper to simulate a multi-chunk streaming HTTP response
  defp streaming_plug(chunks, status \\ 200) do
    fn conn ->
      conn = Plug.Conn.send_chunked(conn, status)

      Enum.reduce(chunks, conn, fn chunk, conn ->
        {:ok, conn} = Plug.Conn.chunk(conn, chunk)
        conn
      end)
    end
  end

  # Plug helper for static responses
  defp static_plug(status, body) do
    fn conn ->
      Plug.Conn.send_resp(conn, status, body)
    end
  end

  # Helper to flush and count remaining messages matching a ref in mailbox
  defp count_mailbox_messages(ref) do
    receive do
      {^ref, _} -> 1 + count_mailbox_messages(ref)
    after
      0 -> 0
    end
  end

  # ── 1. into: :self is process-bound: worker processes raise RuntimeError ─────

  test "consuming into: :self in another process raises RuntimeError because chunks are in requester mailbox" do
    plug = streaming_plug(["chunk_1", "chunk_2", "chunk_3"])

    # Make the request in the current test process
    resp = Req.get!("http://example.com/stream", plug: plug, into: :self, retry: false)

    # resp.body is a %Req.Response.Async{} tied to self()
    assert is_struct(resp.body, Req.Response.Async)
    assert resp.body.pid == self()

    # Attempting to consume resp.body in a separate process fails because chunks are in self() mailbox
    test_pid = self()

    spawn_monitor(fn ->
      try do
        Enum.to_list(resp.body)
      rescue
        e ->
          send(test_pid, {:worker_error, e})
      end
    end)

    assert_receive {:worker_error, %RuntimeError{message: message}}, 5000
    assert message =~ "expected to read body chunk in the process"

    # The chunks are still in the requester process and can be consumed here
    assert Enum.to_list(resp.body) == ["chunk_1", "chunk_2", "chunk_3"]
  end

  # ── 2. Unconsumed into: :self streams leak messages into process mailbox ──────

  test "unconsumed into: :self streams leave unread chunk and done messages in the caller mailbox" do
    plug = streaming_plug(["data_a", "data_b", "data_c"])

    # Request without consuming resp.body
    resp = Req.get!("http://example.com/stream", plug: plug, into: :self, retry: false)
    ref = resp.body.ref

    # 3 chunk messages + 1 done message were delivered to self() mailbox
    assert count_mailbox_messages(ref) == 4
  end

  # ── 3. Partial stream consumption halts and cleans up remaining messages ──────

  test "halting an into: :self stream cancels and drains remaining messages from mailbox" do
    plug = streaming_plug(["part_1", "part_2", "part_3", "part_4", "part_5"])

    resp = Req.get!("http://example.com/stream", plug: plug, into: :self, retry: false)
    ref = resp.body.ref

    # Consume only the first 2 chunks
    consumed = Enum.take(resp.body, 2)
    assert consumed == ["part_1", "part_2"]

    # The Enumerable halt triggered cancel_fun, which drained remaining chunks from mailbox
    assert count_mailbox_messages(ref) == 0
  end

  # ── 4. into: collectable is bypassed on non-200 status codes ──────────────────

  test "into: collectable is bypassed on non-200 responses, buffering body in memory" do
    temp_target = "download_status_test.tmp"
    File.rm(temp_target)

    # 404 response with into: File.stream!
    resp_404 =
      Req.get!("http://example.com/missing",
        plug: static_plug(404, "Custom 404 Page Not Found"),
        into: File.stream!(temp_target),
        retry: false
      )

    # The collectable was NEVER written to; file does not exist on disk
    assert File.exists?(temp_target) == false
    # The body was buffered in memory as a binary string instead
    assert resp_404.status == 404
    assert resp_404.body == "Custom 404 Page Not Found"

    # 200 response with into: File.stream! DOES write to disk
    resp_200 =
      Req.get!("http://example.com/ok",
        plug: static_plug(200, "Successful Payload Data"),
        into: File.stream!(temp_target),
        retry: false
      )

    assert resp_200.status == 200
    assert File.exists?(temp_target) == true
    assert File.read!(temp_target) == "Successful Payload Data"

    File.rm(temp_target)
  end

  # ── 5. into: callback receives chunks on all status codes ─────────────────────

  test "into: callback function receives streaming chunks regardless of HTTP status code" do
    plug_500 = streaming_plug(["error_header\n", "stacktrace_body\n"], 500)

    # into: fun callback receives {:data, chunk} on 500 error responses
    ref = make_ref()

    stream_collector = fn {:data, chunk}, {req, resp} ->
      send(self(), {ref, chunk})
      {:cont, {req, resp}}
    end

    resp =
      Req.get!("http://example.com/error",
        plug: plug_500,
        into: stream_collector,
        retry: false
      )

    assert resp.status == 500
    assert resp.body == ""

    assert_receive {^ref, "error_header\n"}
    assert_receive {^ref, "stacktrace_body\n"}
  end
end

Code.require_file("contract.exs", __DIR__)

ExUnit.start()