@@ -64,27 +64,25 @@
  @typedoc """
  Streaming pack accumulator.
  `body` is an iolist of every byte received *except the final 20-byte
  trailer*. Iolist append is O(1) per chunk — no copy — vs. the previous
  binary `<>` which forced a copy on every feed and made the receive path
  `feed/2` is absorb-only: it appends bytes to `body` (an iolist, O(1)
  per chunk) and updates a 20-byte `lookahead`. It does NOT verify or
  decide completeness — that's `flush/1`'s job. Per-chunk cost is O(1)
  amortized, total CPU during streaming is O(N).
  O(N²) on memory traffic alone.
  This is the design we landed on after a misstep: an earlier version
  (PR #27) ran a SHA-1 over the whole buffer per chunk to opportunistically
  detect completeness. That was O(N²/chunk_size) total — a 100 MB pack
  delivered in 32 KB chunks did roughly 320 GB of SHA-1 work and was the
  bottleneck this code was supposed to fix. We then "fixed" it by
  `lookahead` always holds the most recent (up to) 20 bytes. As new data
  arrives, the old lookahead bytes graduate into `body` and the newest 20
  bytes become the new lookahead. When the pack is complete, the lookahead
  IS the trailer.
  throttling the SHA check to once per 4 MB, which is exactly the
  scheduling-around-cost cheat `docs/PERFORMANCE.md` warns against:
  `bytes_since_check` is what makes the receive path O(N) on CPU instead of
  O(N²). The previous implementation re-hashed the entire growing buffer
  on every chunk arrival to test for completeness — for a 100 MB pack
  delivered in 32 KB chunks, that's 100 MB × 3000 chunks of SHA-1 work,
  roughly 5 minutes single-threaded. We instead defer the SHA-1 check
  until a non-trivial amount of new data has arrived (`@check_interval`),
  bringing total verify work to O(N²/check_interval) which at the chosen
      "Performance fixes that schedule around the cost are cheats until
       proven otherwise."
  The honest fix is to not check during streaming at all. `Pack.Reader.parse/2`
  already verifies the SHA trailer as part of parsing — calling it once
  at flush time runs a single O(N) hash and parse, no per-chunk work.
  4 MB interval is a fraction of a second. The trade-off is up to
  `check_interval` extra bytes buffered after the actual pack end before
  we notice — `flush/1` exists for callers that have an out-of-band EOF
  signal (e.g. the SSH layer) to force the final check.
  `body_size` is `iolist_size(body)` kept incrementally so we can enforce
  `@max_pack_size` without traversing the iolist.
@@ -92,8 +90,7 @@
  @type pack_acc :: %{
          body: iolist(),
          body_size: non_neg_integer(),
          lookahead: binary()
          lookahead: binary(),
          bytes_since_check: non_neg_integer()
        }
  @type state :: %__MODULE__{
@@ -123,18 +120,11 @@
  ]
  @trailer_size 20
  # 4 MB between SHA-1 completeness checks. Empirically this is small enough
  # that the worst-case overshoot (pack ends just past a check boundary,
  # caller doesn't call flush/1) costs at most one extra hash, while still
  # large enough that streaming a 100 MB pack does ~25 hash passes total
  # rather than one per 32 KB chunk.
  @check_interval 4 * 1024 * 1024
  defp empty_pack_acc do
    %{
      body: [],
      body_size: 0,
      lookahead: <<>>,
      bytes_since_check: 0
      lookahead: <<>>
    }
  end
@@ -179,9 +169,9 @@
      {:ok, commands, client_caps, rest} ->
        # SSH can splice the pack's first bytes onto the same packet that
        # ended the command list. Seed the streaming accumulator with that
        # `rest` so it's absorbed exactly like data arriving via a
        # `rest` so it's hashed/buffered exactly like data arriving via a
        # subsequent feed/2.
        state = %{
        new_state = %{
          state
          | commands: commands,
            client_caps: client_caps,
@@ -190,8 +180,13 @@
            pack_acc: absorb(empty_pack_acc(), rest)
        }
        # If this push is delete-only, no pack will arrive — process
        # the ref updates eagerly. Otherwise wait for flush/1.
        if delete_only?(commands) do
          process_ref_updates(new_state)
        else
        # Check if the pack is already complete in the buffer
        maybe_process_pack(state)
          {<<>>, new_state}
        end
      {:need_more, _} ->
        # Buffer partial data for the next feed call
@@ -212,7 +207,8 @@
      report = build_error_report(:pack_too_large, state.commands)
      {report, %{state | phase: :done, result: {:error, :pack_too_large}}}
    else
      # Absorb-only: never verify, never transition. The caller signals
      # end-of-stream via flush/1 once it knows no more bytes are coming.
      state = %{state | pack_acc: new_acc}
      maybe_process_pack(state)
      {<<>>, %{state | pack_acc: new_acc}}
    end
  end
@@ -222,48 +218,43 @@
  end
  @doc """
  Force a final completeness check on the buffered pack.
  Signal end-of-stream and finalize the receive.
  `feed/2` only verifies the pack's SHA-1 trailer every `@check_interval`
  bytes — that's what keeps a 100 MB push from doing 3000 full-buffer hash
  passes. The trade-off is that the *last* `<= @check_interval` bytes
  might arrive without crossing a boundary, so the receive sits in
  `:pack` phase even after the pack is fully delivered.
  `feed/2` is intentionally absorb-only — it does no SHA verification and
  does not detect completeness. That keeps per-chunk cost at O(1) iolist
  append, total streaming cost at O(N), regardless of chunk count. The
  caller — which knows when no more bytes are coming — is responsible
  for calling `flush/1` to trigger the actual SHA-1 verification (via
  `Pack.Reader.parse/2`'s built-in checksum check), parse the entries,
  store objects, and apply ref updates.
  Both transports we ship signal end-of-stream naturally:
    * **HTTP receive-pack**: the request body has a known length; flush
      after `read_full_body` returns.
    * **SSH receive-pack**: the client sends `{:eof, channel_id}` when
      it's done; flush in that handler (`Anvil.SSH.CLI`).
  If the buffered bytes don't form a valid pack (truncated, corrupted,
  Transports that have an out-of-band end-of-stream signal (the SSH layer's
  `{:eof, channel_id}` message; an HTTP request body's natural end) call
  `flush/1` once they know no more data is coming. It bypasses the
  throttle and runs a final hash, transitioning to `:done` if the bytes
  buffered so far form a valid pack.
  bad SHA), flush returns a structured error report rather than leaving
  the state machine wedged in `:pack`.
  Returns `{response, state}` matching `feed/2`'s shape.
  """
  @spec flush(state()) :: {binary(), state()}
  def flush(%__MODULE__{phase: :pack, commands: commands} = state) do
    if delete_only?(commands) do
    all_deletes? = Enum.all?(commands, fn cmd -> cmd.new_sha == @zero_sha end)
      process_ref_updates(state)
    else
      process_pack_and_refs(state)
    cond do
      all_deletes? or commands == [] ->
        process_ref_updates(state)
      true ->
        case pack_check(state.pack_acc, :force) do
          {:complete, _acc} ->
            process_pack_and_refs(state)
          {:incomplete, new_acc} ->
            # No more data is coming and the buffered bytes don't form a
            # valid pack. Surface this rather than sitting in :pack forever.
            report = build_error_report(:incomplete_pack, state.commands)
            {report,
             %{state | pack_acc: new_acc, phase: :done, result: {:error, :incomplete_pack}}}
        end
    end
  end
  def flush(%__MODULE__{} = state), do: {<<>>, state}
  defp delete_only?([]), do: true
  defp delete_only?(commands), do: Enum.all?(commands, fn cmd -> cmd.new_sha == @zero_sha end)
  @doc """
  Check if the protocol exchange is complete.
  """
@@ -277,6 +268,6 @@
  # the trailing 20-byte window graduate into the iolist body; the rest
  # stays in `lookahead` until more data arrives. Per-chunk cost: O(1)
  # iolist cons + a single 20-byte slice. No SHA work happens here — that
  # is throttled by `bytes_since_check` and only fires in pack_check/2.
  # only runs once at flush time via `Pack.Reader.parse/2`.
  defp absorb(acc, data) when is_binary(data) do
    combined = acc.lookahead <> data
@@ -294,8 +285,7 @@
          acc
          | body: [acc.body, graduate],
            body_size: acc.body_size + graduate_size,
            lookahead: new_lookahead,
            bytes_since_check: acc.bytes_since_check + graduate_size
            lookahead: new_lookahead
        }
    end
  end
@@ -461,67 +451,6 @@
    end
  end
  defp maybe_process_pack(%{commands: commands} = state) do
    # If all commands are deletes (new_sha == 0000...), no pack is sent
    all_deletes? = Enum.all?(commands, fn cmd -> cmd.new_sha == @zero_sha end)
    if all_deletes? or commands == [] do
      process_ref_updates(state)
    else
      case pack_check(state.pack_acc, :throttled) do
        {:complete, _acc} ->
          process_pack_and_refs(state)
        {:incomplete, new_acc} ->
          {<<>>, %{state | pack_acc: new_acc}}
      end
    end
  end
  # A complete pack is: 12-byte header + entries + 20-byte SHA-1 trailer.
  # The accumulator buffers everything except the trailing 20 bytes; the
  # pack is complete iff that trailer equals SHA-1 of everything before
  # it.
  #
  # Mode `:throttled` only attempts the SHA when at least @check_interval
  # bytes have arrived since the last check. This is what stops the receive
  # path from re-hashing the entire growing buffer on every 32 KB chunk.
  # Mode `:force` always attempts the SHA — used by `flush/1` when the
  # transport (e.g. SSH) signals that no more bytes are coming and we need
  # to make a final determination.
  defp pack_check(nil, _mode), do: {:incomplete, nil}
  defp pack_check(%{lookahead: l} = acc, _mode) when byte_size(l) < @trailer_size,
    do: {:incomplete, acc}
  defp pack_check(%{body_size: n} = acc, _mode) when n < 12, do: {:incomplete, acc}
  defp pack_check(%{body_size: n, bytes_since_check: c} = acc, :throttled)
       when n > @check_interval and c < @check_interval,
       do: {:incomplete, acc}
  # Bodies smaller than the throttle window are cheap to hash and frequent
  # in tests/HTTP-one-shot pushes — bypass the throttle for those so callers
  # don't need to know about flush/1 just to make small pushes finalize.
  defp pack_check(%{body: body, lookahead: lookahead} = acc, _mode) do
    # SHA-1 collision probability against random partial pack data is 1 in
    # 2^160, so an opportunistic match means we have the actual trailer.
    # The `Pack.Reader.parse/2` downstream pass catches anything that
    # somehow slipped past (with a structured error rather than a panic),
    # so we don't bother with the prior magic-bytes ("PACK") prefix check
    # — verifying the magic would require materializing the iolist for
    # 8 bits of additional confidence.
    candidate = :crypto.hash(:sha, body)
    new_acc = %{acc | bytes_since_check: 0}
    if candidate == lookahead do
      {:complete, new_acc}
    else
      {:incomplete, new_acc}
    end
  end
  defp process_pack_and_refs(state) do
    :telemetry.span(
      [:ex_git_objectstore, :protocol, :receive_pack],
@@ -530,9 +459,13 @@
        # Materialize the iolist body once for the existing single-shot
        # Pack.Reader. This is the only point in the receive path where we
        # hold a binary the size of the pack — and only briefly, before the
        # parse pass replaces it with parsed entries. A future change can
        # convert Pack.Reader to a streaming consumer to drop even this
        # peak (see issue #153 for the longer-term plan).
        # parse pass replaces it with parsed entries. Pack.Reader.parse/2
        # internally verifies the trailing SHA-1 checksum, so we don't run
        # one ourselves: any truncation, corruption, or trailer mismatch
        # surfaces as a structured `{:error, reason}` from the parser. A
        # future change can convert Pack.Reader to a streaming consumer
        # to drop the peak buffer too — that's the remaining gap on
        # issue #153.
        pack_data = pack_acc_to_binary(state.pack_acc)
        result =
@@ -549,6 +482,8 @@
      end
    )
  end
  defp pack_acc_to_binary(nil), do: <<>>
  defp pack_acc_to_binary(%{body: body, lookahead: lookahead}) do
    :erlang.iolist_to_binary([body, lookahead])

@@ -73,7 +73,7 @@
      {_advert, state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, commit_sha, "refs/heads/main"}])
      {response, state} = ReceivePack.feed(state, commands <> pack_data)
      {response, state} = rp_feed_and_flush(state, commands <> pack_data)
      assert ReceivePack.done?(state)
      assert_report_ok(response, ["refs/heads/main"])
@@ -115,7 +115,7 @@
      {_advert, state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, commit_sha, "refs/heads/main"}])
      {response, state} = rp_feed_and_flush(state, commands <> pack_data)
      {response, state} = ReceivePack.feed(state, commands <> pack_data)
      assert ReceivePack.done?(state)
      assert_report_ok(response, ["refs/heads/main"])
@@ -150,7 +150,7 @@
      {_advert, state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, commit_sha, "refs/heads/main"}])
      {response, state} = ReceivePack.feed(state, commands <> pack_data)
      {response, state} = rp_feed_and_flush(state, commands <> pack_data)
      assert ReceivePack.done?(state)
      assert_report_ok(response, ["refs/heads/main"])
@@ -179,7 +179,7 @@
      repo = make_filesystem_repo(base, "target_incr")
      {_advert, state} = ReceivePack.init(repo)
      commands1 = build_receive_commands([{@zero_sha, c1_sha, "refs/heads/main"}])
      {response1, _state} = rp_feed_and_flush(state, commands1 <> pack1)
      {response1, _state} = ReceivePack.feed(state, commands1 <> pack1)
      assert_report_ok(response1, ["refs/heads/main"])
      # Second commit (modifies the file — thin pack references first blob as delta base)
@@ -194,7 +194,7 @@
      # Push incremental update
      {_advert2, state2} = ReceivePack.init(repo)
      commands2 = build_receive_commands([{c1_sha, c2_sha, "refs/heads/main"}])
      {response2, state2} = ReceivePack.feed(state2, commands2 <> pack2)
      {response2, state2} = rp_feed_and_flush(state2, commands2 <> pack2)
      assert ReceivePack.done?(state2)
      assert_report_ok(response2, ["refs/heads/main"])
@@ -239,7 +239,7 @@
          {@zero_sha, feature_sha, "refs/heads/feature"}
        ])
      {response, state} = ReceivePack.feed(state, commands <> pack_data)
      {response, state} = rp_feed_and_flush(state, commands <> pack_data)
      assert ReceivePack.done?(state)
      assert_report_ok(response, ["refs/heads/main", "refs/heads/feature"])
@@ -265,7 +265,7 @@
      repo = make_filesystem_repo(base, "target_force")
      {_advert, state} = ReceivePack.init(repo)
      cmd1 = build_receive_commands([{@zero_sha, c1_sha, "refs/heads/main"}])
      {r1, _} = rp_feed_and_flush(state, cmd1 <> pack1)
      {r1, _} = ReceivePack.feed(state, cmd1 <> pack1)
      assert_report_ok(r1, ["refs/heads/main"])
      # Amend commit (rewrites history — different SHA for same ref)
@@ -281,7 +281,7 @@
      # Force push: old_sha=c1, new_sha=c2 (non-fast-forward)
      {_advert2, state2} = ReceivePack.init(repo)
      cmd2 = build_receive_commands([{c1_sha, c2_sha, "refs/heads/main"}])
      {r2, state2} = ReceivePack.feed(state2, cmd2 <> pack2)
      {r2, state2} = rp_feed_and_flush(state2, cmd2 <> pack2)
      assert ReceivePack.done?(state2)
      assert_report_ok(r2, ["refs/heads/main"])
@@ -321,7 +321,7 @@
      repo = make_filesystem_repo(base, "target_merge")
      {_advert, state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, merge_sha, "refs/heads/main"}])
      {response, state} = rp_feed_and_flush(state, commands <> pack_data)
      {response, state} = ReceivePack.feed(state, commands <> pack_data)
      assert ReceivePack.done?(state)
      assert_report_ok(response, ["refs/heads/main"])
@@ -358,7 +358,7 @@
          {@zero_sha, tag_sha, "refs/tags/v1.0"}
        ])
      {response, state} = ReceivePack.feed(state, commands <> pack_data)
      {response, state} = rp_feed_and_flush(state, commands <> pack_data)
      assert ReceivePack.done?(state)
      assert_report_ok(response, ["refs/heads/main", "refs/tags/v1.0"])
@@ -387,7 +387,7 @@
      repo = make_filesystem_repo(base, "upload_single")
      {_advert, state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, commit_sha, "refs/heads/main"}])
      {r, _} = ReceivePack.feed(state, commands <> pack_data)
      {r, _} = rp_feed_and_flush(state, commands <> pack_data)
      assert_report_ok(r, ["refs/heads/main"])
      # Now fetch via UploadPack
@@ -428,7 +428,7 @@
      repo = make_filesystem_repo(base, "upload_multi")
      {_advert, state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, commit_sha, "refs/heads/main"}])
      {r, _} = rp_feed_and_flush(state, commands <> pack_data)
      {r, _} = ReceivePack.feed(state, commands <> pack_data)
      assert_report_ok(r, ["refs/heads/main"])
      # Fetch via UploadPack
@@ -484,7 +484,7 @@
      repo = make_filesystem_repo(base, "roundtrip")
      {_advert, rp_state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, commit_sha, "refs/heads/main"}])
      {r, _} = ReceivePack.feed(rp_state, commands <> pack_data)
      {r, _} = rp_feed_and_flush(rp_state, commands <> pack_data)
      assert_report_ok(r, ["refs/heads/main"])
      # Fetch back via UploadPack
@@ -541,7 +541,7 @@
          {@zero_sha, feature_sha, "refs/heads/feature"}
        ])
      {r, _} = rp_feed_and_flush(rp_state, commands <> pack_data)
      {r, _} = ReceivePack.feed(rp_state, commands <> pack_data)
      assert_report_ok(r, ["refs/heads/main", "refs/heads/feature"])
      # Fetch feature branch via UploadPack
@@ -569,7 +569,7 @@
  # ============================================================================
  describe "adversarial packs" do
    test "corrupted pack checksum is rejected (stays incomplete)", ctx do
    test "corrupted pack checksum is rejected with structured error", ctx do
      %{work_dir: work_dir, base: base} = ctx
      File.write!(Path.join(work_dir, "file.txt"), "content\n")
@@ -589,15 +589,21 @@
      {_advert, state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, commit_sha, "refs/heads/main"}])
      {response, state} = ReceivePack.feed(state, commands <> corrupted)
      {response, state} = rp_feed_and_flush(state, commands <> corrupted)
      # Bad checksum surfaces from Pack.Reader.parse as a structured error
      # at flush time. The state machine transitions to :done with an
      # error result — it does NOT sit in :pack waiting for more data
      # Corrupted checksum means pack appears incomplete — state machine waits for more data
      refute ReceivePack.done?(state)
      assert response == <<>>
      assert state.phase == :pack
      # (the previous SHA-verify-per-chunk trick made it look like that,
      # which was the cheat).
      assert ReceivePack.done?(state)
      assert match?({:error, _}, state.result)
      # Report-status carries the unpack failure to the client.
      assert byte_size(response) > 0
      assert response =~ "unpack "
    end
    test "truncated pack data is rejected with structured error", ctx do
    test "truncated pack data is rejected (stays incomplete)", ctx do
      %{work_dir: work_dir, base: base} = ctx
      File.write!(Path.join(work_dir, "file.txt"), "content\n")
@@ -615,10 +621,15 @@
      {_advert, state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, commit_sha, "refs/heads/main"}])
      {response, state} = ReceivePack.feed(state, commands <> truncated)
      {response, state} = rp_feed_and_flush(state, commands <> truncated)
      # Truncated pack: Pack.Reader.parse fails (either the trailer SHA
      # mismatches or an entry parse runs off the end). flush surfaces
      # Truncated pack won't have valid checksum — stays incomplete
      # that as a structured error and transitions the state machine to
      # :done so we don't hang on a closed channel.
      assert ReceivePack.done?(state)
      assert match?({:error, _}, state.result)
      assert byte_size(response) > 0
      assert response =~ "unpack "
      refute ReceivePack.done?(state)
      assert response == <<>>
    end
  end
@@ -649,13 +660,13 @@
          {@zero_sha, commit_sha, "refs/heads/to-delete"}
        ])
      {r1, _} = rp_feed_and_flush(state, cmd1 <> pack_data)
      {r1, _} = ReceivePack.feed(state, cmd1 <> pack_data)
      assert_report_ok(r1, ["refs/heads/main", "refs/heads/to-delete"])
      # Now delete the branch (no pack needed for delete-only)
      {_advert2, state2} = ReceivePack.init(repo)
      cmd2 = build_receive_commands([{commit_sha, @zero_sha, "refs/heads/to-delete"}])
      {r2, state2} = ReceivePack.feed(state2, cmd2)
      {r2, state2} = rp_feed_and_flush(state2, cmd2)
      assert ReceivePack.done?(state2)
      assert_report_ok(r2, ["refs/heads/to-delete"])
@@ -677,7 +688,7 @@
      repo = make_filesystem_repo(base, "advert_check")
      {_advert, state} = ReceivePack.init(repo)
      commands = build_receive_commands([{@zero_sha, commit_sha, "refs/heads/main"}])
      {r, _} = rp_feed_and_flush(state, commands <> pack_data)
      {r, _} = ReceivePack.feed(state, commands <> pack_data)
      assert_report_ok(r, ["refs/heads/main"])
      # Now check UploadPack advertisement
@@ -781,6 +792,17 @@
      assert "ok #{ref}" in data_lines,
             "Expected 'ok #{ref}' in report, got: #{inspect(data_lines)}"
    end
  end
  # ReceivePack.feed/2 is now absorb-only: it never finalizes the state
  # machine. Tests that fed a complete pack and expected `done?` true
  # immediately go through this helper, which mirrors what real SSH/HTTP
  # callers do once they observe end-of-stream (SSH client EOF, end of
  # HTTP request body).
  defp rp_feed_and_flush(state, data) do
    {feed_resp, state} = ReceivePack.feed(state, data)
    {flush_resp, state} = ReceivePack.flush(state)
    {feed_resp <> flush_resp, state}
  end
  defp git_index_pack!(base, pack_data) do

@@ -72,6 +72,16 @@
    header <> checksum
  end
  # `ReceivePack.feed/2` is now absorb-only and does not finalize the
  # state machine. Tests that fed a complete-looking blob now go through
  # this helper, which mirrors what the SSH/HTTP transports do once
  # they observe end-of-stream.
  defp feed_and_flush(state, data) do
    {feed_resp, state} = ReceivePack.feed(state, data)
    {flush_resp, state} = ReceivePack.flush(state)
    {feed_resp <> flush_resp, state}
  end
  defp assert_ref_ok(response, ref) do
    assert response =~ "ok #{ref}"
  end
@@ -94,7 +104,7 @@
      {_advert, state} = ReceivePack.init(repo, update_hook: update_hook)
      data = push_command(@zero_sha, sha, "refs/heads/main")
      {response, _state} = feed_and_flush(state, data <> empty_pack())
      {response, _state} = ReceivePack.feed(state, data <> empty_pack())
      assert_received {:update_called, "refs/heads/main"}
      assert_ref_ok(response, "refs/heads/main")
@@ -119,7 +129,7 @@
          {@zero_sha, sha, "refs/heads/blocked"}
        ])
      {response, _state} = feed_and_flush(state, data <> empty_pack())
      {response, _state} = ReceivePack.feed(state, data <> empty_pack())
      assert_ref_ok(response, "refs/heads/allowed")
      assert_ref_ng(response, "refs/heads/blocked")
@@ -141,7 +151,7 @@
      {_advert, state} = ReceivePack.init(repo, update_hook: update_hook)
      data = push_command(@zero_sha, sha, "refs/heads/main")
      {_response, _state} = ReceivePack.feed(state, data <> empty_pack())
      {_response, _state} = feed_and_flush(state, data <> empty_pack())
      assert_received {:update_args, "refs/heads/main", @zero_sha, ^sha}
    end
@@ -151,7 +161,7 @@
      {_advert, state} = ReceivePack.init(repo)
      data = push_command(@zero_sha, sha, "refs/heads/main")
      {response, _state} = ReceivePack.feed(state, data <> empty_pack())
      {response, _state} = feed_and_flush(state, data <> empty_pack())
      assert_ref_ok(response, "refs/heads/main")
    end
@@ -171,7 +181,7 @@
      {_advert, state} = ReceivePack.init(repo, post_receive_hook: post_receive_hook)
      data = push_command(@zero_sha, sha, "refs/heads/main")
      {_response, _state} = ReceivePack.feed(state, data <> empty_pack())
      {_response, _state} = feed_and_flush(state, data <> empty_pack())
      assert_received {:post_receive, changes}
      assert [%{ref: "refs/heads/main", old_sha: @zero_sha, new_sha: ^sha, status: :ok}] = changes
@@ -202,7 +212,7 @@
          {@zero_sha, sha, "refs/heads/blocked"}
        ])
      {_response, _state} = ReceivePack.feed(state, data <> empty_pack())
      {_response, _state} = feed_and_flush(state, data <> empty_pack())
      assert_received {:post_receive, changes}
      assert length(changes) == 2
@@ -223,7 +233,7 @@
      {_advert, state} = ReceivePack.init(repo, post_receive_hook: post_receive_hook)
      data = push_command(@zero_sha, sha, "refs/heads/main")
      {response, state} = ReceivePack.feed(state, data <> empty_pack())
      {response, state} = feed_and_flush(state, data <> empty_pack())
      # Push should still succeed even though post_receive failed
      assert_ref_ok(response, "refs/heads/main")
@@ -250,7 +260,7 @@
        )
      data = push_command(@zero_sha, sha, "refs/heads/main")
      {_response, _state} = feed_and_flush(state, data <> empty_pack())
      {_response, _state} = ReceivePack.feed(state, data <> empty_pack())
      refute_received :post_receive_called
    end
@@ -260,7 +270,7 @@
      {_advert, state} = ReceivePack.init(repo)
      data = push_command(@zero_sha, sha, "refs/heads/main")
      {response, _state} = feed_and_flush(state, data <> empty_pack())
      {response, _state} = ReceivePack.feed(state, data <> empty_pack())
      assert_ref_ok(response, "refs/heads/main")
    end

@@ -140,34 +140,41 @@
             "process heap peaked at #{div(peak_bytes, 1024)} KB during a #{div(pack_bytes, 1024)} KB receive — looks like the binary `<>` regression is back"
    end
    test "small pack still finalizes via feed alone (no flush needed)", %{repo: repo} do
      # Small packs (< @check_interval) need to finalize without the caller
      # having to call flush/1 — that's what keeps HTTP one-shot and tiny
      # SSH pushes working unchanged.
    test "small pack also requires flush — feed is absorb-only", %{repo: repo} do
      # The previous design auto-finalized small packs from feed/2 by
      # SHA-verifying on every chunk. That was the cheat. The honest
      # contract is that feed/2 NEVER finalizes — the caller signals
      # end-of-stream by calling flush/1 once it knows no more bytes are
      # coming. This test pins that contract for both small and large
      # packs.
      {commit_sha, pack_data, _shas} = build_chunky_repo_pack(2, 32)
      assert byte_size(pack_data) < 4 * 1024 * 1024
      {_advert, state} = ReceivePack.init(repo)
      commands_blob = commands_blob_for("refs/heads/main", commit_sha)
      {<<>>, state} = ReceivePack.feed(state, commands_blob)
      {<<>>, state} = ReceivePack.feed(state, pack_data)
      {response, state} = ReceivePack.feed(state, pack_data)
      # feed alone leaves us in :pack — explicitly NOT done.
      refute ReceivePack.done?(state)
      {response, state} = ReceivePack.flush(state)
      assert ReceivePack.done?(state)
      {:ok, packets, _} = PktLine.decode(response)
      data_lines = for {:data, d} <- packets, do: d
      assert "unpack ok" in data_lines
      assert {:ok, ^commit_sha} = Ref.get(repo, "refs/heads/main")
    end
    test "flush on a truncated buffer reports an unpack error rather than hanging",
    test "flush on a partial buffer reports incomplete_pack rather than hanging",
         %{repo: repo} do
      # If the transport gives up halfway, flush/1 hands the buffered
      # bytes to Pack.Reader.parse/2 which detects the truncated trailer
      # (or malformed body, depending on where the cut landed) and
      # returns a structured error. ReceivePack surfaces that as a done
      # state with `result: {:error, _}` and an `unpack <reason>`
      # report-status line, rather than sitting in :pack forever.
      # If the transport gives up halfway, flush/1 must produce a definite
      # result (error) rather than leaving the state machine in :pack
      # forever. Prior to flush/1 the SHA-verify-per-chunk trick covered
      # this implicitly by never matching, so the state stayed in :pack
      # until the channel closed — flush makes that surface as a structured
      # error report.
      {_commit_sha, pack_data, _shas} = build_chunky_repo_pack(3, 64)
      truncated = binary_part(pack_data, 0, byte_size(pack_data) - 30)
@@ -180,7 +187,7 @@
      {report, final_state} = ReceivePack.flush(state)
      assert ReceivePack.done?(final_state)
      assert match?({:error, :incomplete_pack}, final_state.result)
      assert match?({:error, _}, final_state.result)
      {:ok, packets, _} = PktLine.decode(report)
      data_lines = for {:data, d} <- packets, do: d

@@ -176,7 +176,9 @@
      {pack_data, _} = Writer.generate(objects)
      {feed_resp, state} = ReceivePack.feed(state, commands <> pack_data)
      {flush_resp, state} = ReceivePack.flush(state)
      response = feed_resp <> flush_resp
      {response, state} = ReceivePack.feed(state, commands <> pack_data)
      assert ReceivePack.done?(state)
      # Client sent report-status capability, so we should get a report
@@ -224,6 +226,8 @@
      {pack_data, _} = Writer.generate(objects)
      # Feed commands + pack
      {response, state} = ReceivePack.feed(state, commands <> pack_data)
      {feed_resp, state} = ReceivePack.feed(state, commands <> pack_data)
      {flush_resp, state} = ReceivePack.flush(state)
      response = feed_resp <> flush_resp
      assert ReceivePack.done?(state)
@@ -328,9 +332,11 @@
      refute ReceivePack.done?(state)
      {r3, state} = ReceivePack.feed(state, pack_part2)
      assert r3 == <<>>
      {flush_resp, state} = ReceivePack.flush(state)
      assert ReceivePack.done?(state)
      {:ok, packets, _} = PktLine.decode(r3)
      {:ok, packets, _} = PktLine.decode(flush_resp)
      data_lines = for {:data, d} <- packets, do: d
      assert "unpack ok" in data_lines
      assert "ok refs/heads/main" in data_lines
@@ -380,9 +386,11 @@
      assert state.phase == :commands
      {r2, state} = ReceivePack.feed(state, part2)
      assert r2 == <<>>
      {flush_resp, state} = ReceivePack.flush(state)
      assert ReceivePack.done?(state)
      {:ok, packets, _} = PktLine.decode(r2)
      {:ok, packets, _} = PktLine.decode(flush_resp)
      data_lines = for {:data, d} <- packets, do: d
      assert "unpack ok" in data_lines
    end
@@ -435,10 +443,12 @@
      # Second feed: remaining data including flush + pack
      {response2, state} = ReceivePack.feed(state, second_chunk)
      assert response2 == <<>>
      {flush_resp, state} = ReceivePack.flush(state)
      assert ReceivePack.done?(state)
      # Should get proper report-status
      {:ok, packets, _} = PktLine.decode(response2)
      {:ok, packets, _} = PktLine.decode(flush_resp)
      data_lines = for {:data, d} <- packets, do: d
      assert "unpack ok" in data_lines
      assert "ok refs/heads/main" in data_lines

@@ -141,7 +141,8 @@
      checksum = :crypto.hash(:sha, header)
      pack = header <> checksum
      {_response, _state} = ReceivePack.feed(state, cmd <> pack)
      {_feed_resp, state} = ReceivePack.feed(state, cmd <> pack)
      {_flush_resp, _state} = ReceivePack.flush(state)
      assert_received {:telemetry_event, [:ex_git_objectstore, :protocol, :receive_pack, :start],
                       %{system_time: _}, %{repo_id: "test-repo", command_count: 1}}

@@ -253,7 +253,14 @@
  defp drive_receive_recv(client, state) do
    case :gen_tcp.recv(client, 0, 15_000) do
      {:ok, data} -> drive_receive_feed(client, state, data)
      {:error, _} -> :ok
      {:ok, data} ->
        drive_receive_feed(client, state, data)
      {:error, _} ->
        # Client closed its side — that's end-of-stream. Flush so the
        # state machine actually finalizes (the absorb-only feed/2
        # contract leaves it in :pack until somebody calls flush).
        _ = drive_receive_flush(client, state)
        :ok
    end
  end
@@ -264,6 +271,16 @@
    drive_receive(client, new_state)
  end
  # When the TCP recv loop returns an error/timeout (i.e. the client
  # closed its side), we treat that as end-of-stream and flush. Real
  # SSH/HTTP layers do the same on their respective EOF signals; the
  # test daemon just polls until socket close.
  defp drive_receive_flush(client, state) do
    {response, new_state} = ReceivePack.flush(state)
    if byte_size(response) > 0, do: :ok = :gen_tcp.send(client, response)
    new_state
  end
  # --- shared ---
  defp read_one_pkt(client) do
@@ -371,17 +388,11 @@
  end
  defp drive_receive_state(body, state) do
    # HTTP delivers the whole request body in one shot — feed once,
    # then flush since the body has a known length.
    {first, state} = ReceivePack.feed(state, body)
    if ReceivePack.done?(state) do
    {second, state} = ReceivePack.flush(state)
    {first <> second, state}
      {first, state}
    else
      # Some state machines accept the pack across multiple feeds; for
      # HTTP the whole body arrives at once, so a second empty feed is
      # enough to let maybe_process_pack run.
      {second, state} = ReceivePack.feed(state, <<>>)
      {first <> second, state}
    end
  end
  # --- HTTP/1.1 mini-parser ---

@@ -91,7 +91,9 @@
    repo = build_repo(conn, repo_id)
    {:ok, conn, body} = read_full_body(conn)
    {_advert, state} = ReceivePack.init(repo)
    {response, _state} = ReceivePack.feed(state, body)
    {feed_resp, state} = ReceivePack.feed(state, body)
    {flush_resp, _state} = ReceivePack.flush(state)
    response = feed_resp <> flush_resp
    conn
    |> put_resp_content_type("application/x-git-receive-pack-result")