@@ -1,0 +1,273 @@
# Copyright 2026 Cole Christensen
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
defmodule ExGitObjectstore.Protocol.ReceivePackStreamingTest do
# Regression coverage for the streaming receive path. The previous
# implementation buffered the entire pack as one binary (`<>` per chunk)
# and re-hashed it on every feed; on a real-sized push (the openvswitch
# mirror, ~106 MB / 134k objects) that exhausted the container memory
# cap and pinned a CPU for tens of minutes before failing.
#
# Two invariants we want to lock in here:
#
# 1. Resident memory growth during ingestion is roughly proportional
# to total pack size, NOT pack_size × chunk_count. The iolist body
# drops the K² binary-copy cost; this test asserts the pack-acc
# heap does not balloon.
#
# 2. Per-chunk CPU is bounded. The throttled SHA-1 check makes the
# whole receive O(N) total, where pre-fix it was O(N²/chunk_size)
# — a 5 MB pack delivered in 4 KB chunks should not take seconds.
use ExUnit.Case, async: false
alias ExGitObjectstore.{Object, Ref}
alias ExGitObjectstore.Object.{Blob, Commit, Tree}
alias ExGitObjectstore.Pack.Writer
alias ExGitObjectstore.Protocol.{PktLine, ReceivePack}
alias ExGitObjectstore.Test.RepoHelper
@zero_sha String.duplicate("0", 40)
setup do
repo = RepoHelper.memory_repo("streaming-#{:erlang.unique_integer([:positive])}")
ExGitObjectstore.init(repo)
%{repo: repo}
end
describe "streaming ingestion" do
test "5 MB pack delivered in 4 KB chunks completes and lands every object", %{repo: repo} do
{commit_sha, pack_data, expected_shas} = build_chunky_repo_pack(80, 65 * 1024)
assert byte_size(pack_data) >= 5 * 1024 * 1024,
"expected ≥5 MB pack to exercise the streaming path; got #{byte_size(pack_data)}"
{_advert, state} = ReceivePack.init(repo)
commands_blob = commands_blob_for("refs/heads/main", commit_sha)
{<<>>, state} = ReceivePack.feed(state, commands_blob)
assert state.phase == :pack
# Deliver the pack in small chunks. With the pre-fix implementation
# this fed K² binary-copy cost into the receive path; with the iolist
# body it's K × O(chunk_size).
state = feed_in_chunks(state, pack_data, 4 * 1024)
# Above the @check_interval (4 MB) threshold so the throttled check
# has fired at least once. flush/1 covers any tail bytes that didn't
# cross the next 4 MB boundary.
{final_response, final_state} =
if ReceivePack.done?(state), do: {<<>>, state}, else: ReceivePack.flush(state)
assert ReceivePack.done?(final_state),
"pack should be marked done after flush"
# Report-status confirms the pack and ref made it through.
response_blob = if final_response == <<>>, do: <<>>, else: final_response
{:ok, packets, _} = PktLine.decode(response_blob)
data_lines = for {:data, d} <- packets, do: d
assert "unpack ok" in data_lines
assert "ok refs/heads/main" in data_lines
assert {:ok, ^commit_sha} = Ref.get(repo, "refs/heads/main")
# Spot-check a few of the blob shas — if any one is missing the pack
# parser dropped objects on the floor.
for sha <- Enum.take(expected_shas, 5) do
assert {:ok, _} = Object.read(repo, sha)
end
end
test "process heap during a 5 MB chunked feed stays bounded", %{repo: repo} do
# The pre-fix implementation grew the pack_buffer binary by `<>` on
# every chunk, which BEAM copies on certain growth patterns — so a
# 5 MB pack delivered in 4 KB chunks could push the receive process
# heap well beyond the pack size. The iolist body is O(1) per chunk.
# We assert that peak BEAM heap allocated to the receive process
# stays within ~3× the pack size — a generous bound that still flags
# any regression to quadratic copying.
{commit_sha, pack_data, _shas} = build_chunky_repo_pack(80, 65 * 1024)
pack_bytes = byte_size(pack_data)
{_advert, state} = ReceivePack.init(repo)
commands_blob = commands_blob_for("refs/heads/main", commit_sha)
{<<>>, state} = ReceivePack.feed(state, commands_blob)
# Sample heap size across the chunked feed. We sample after every
# ~64 chunks to bound sampling overhead; the peak is what we care
# about.
chunks = chunkify(pack_data, 4 * 1024)
{state, peak_heap_words} =
chunks
|> Enum.with_index()
|> Enum.reduce({state, 0}, fn {chunk, i}, {st, peak} ->
{_resp, st2} = ReceivePack.feed(st, chunk)
peak2 =
if rem(i, 64) == 0 do
{:total_heap_size, words} = Process.info(self(), :total_heap_size)
max(peak, words)
else
peak
end
{st2, peak2}
end)
{_resp, state} =
if ReceivePack.done?(state), do: {<<>>, state}, else: ReceivePack.flush(state)
assert ReceivePack.done?(state)
# `total_heap_size` is in words (8 bytes on 64-bit). Convert and
# compare against pack size. The bound is loose because the test
# process also owns chunk binaries, the iolist body, and ExUnit
# state — but a quadratic regression would push this 10×+.
peak_bytes = peak_heap_words * :erlang.system_info(:wordsize)
assert peak_bytes < pack_bytes * 3,
"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.
{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)
{response, state} = ReceivePack.feed(state, pack_data)
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 partial buffer reports incomplete_pack rather than hanging",
%{repo: repo} do
# 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)
{_advert, state} = ReceivePack.init(repo)
commands_blob = commands_blob_for("refs/heads/main", String.duplicate("a", 40))
{<<>>, state} = ReceivePack.feed(state, commands_blob)
{<<>>, state} = ReceivePack.feed(state, truncated)
{report, final_state} = ReceivePack.flush(state)
assert ReceivePack.done?(final_state)
assert match?({:error, :incomplete_pack}, final_state.result)
{:ok, packets, _} = PktLine.decode(report)
data_lines = for {:data, d} <- packets, do: d
assert Enum.any?(data_lines, &String.starts_with?(&1, "unpack "))
end
end
# ---- helpers ----
defp commands_blob_for(ref, new_sha) do
PktLine.encode("#{@zero_sha} #{new_sha} #{ref}") <> PktLine.flush()
end
defp feed_in_chunks(state, data, chunk_size) do
data
|> chunkify(chunk_size)
|> Enum.reduce(state, fn chunk, acc ->
{_response, new_state} = ReceivePack.feed(acc, chunk)
new_state
end)
end
defp chunkify(<<>>, _), do: []
defp chunkify(bin, n) when byte_size(bin) <= n, do: [bin]
defp chunkify(bin, n) do
<<head::binary-size(n), rest::binary>> = bin
[head | chunkify(rest, n)]
end
# Build a repo with `n_blobs` blobs of `blob_size` bytes each, plus a
# tree referencing all of them and a single commit. Returns
# {commit_sha, pack_binary, [blob_shas]}. Used to synthesize packs of
# specific sizes for the streaming-path tests; the blob contents are
# cheap pseudo-random so zlib doesn't compress them down to zero.
defp build_chunky_repo_pack(n_blobs, blob_size) do
# `:crypto.strong_rand_bytes` produces incompressible content, so the
# resulting pack size is predictable: roughly n_blobs * blob_size plus a
# small per-object header overhead. Avoids fighting zlib on synthetic
# data when we want a specific pack size to exercise streaming code.
blobs =
for _i <- 1..n_blobs do
content = :crypto.strong_rand_bytes(blob_size)
blob = Blob.from_content(content)
{:ok, sha} = Object.write(memory_repo_for_object_writes(), blob)
{sha, content, blob}
end
repo = memory_repo_for_object_writes()
entries =
Enum.with_index(blobs, fn {sha, _content, _}, i ->
%{mode: "100644", name: "f#{i}", sha: sha}
end)
tree = Tree.new(entries)
{:ok, tree_sha} = Object.write(repo, tree)
commit = %Commit{
tree: tree_sha,
parents: [],
author: "Streaming Test <st@t.com> 1000000000 +0000",
committer: "Streaming Test <st@t.com> 1000000000 +0000",
message: "streaming test\n"
}
{:ok, commit_sha} = Object.write(repo, commit)
pack_objects =
[
{:tree, Tree.encode_content(tree), tree_sha},
{:commit, Commit.encode_content(commit), commit_sha}
| for({sha, content, _} <- blobs, do: {:blob, content, sha})
]
{pack_data, _} = Writer.generate(pack_objects)
{commit_sha, pack_data, Enum.map(blobs, fn {sha, _, _} -> sha end)}
end
# The blobs/tree/commit need a writable repo to produce SHAs. We don't
# care which — just need somewhere `Object.write` can land. A fresh
# memory repo per call is fine since SHA derives from content.
defp memory_repo_for_object_writes do
repo = RepoHelper.memory_repo("seed-#{:erlang.unique_integer([:positive])}")
ExGitObjectstore.init(repo)
repo
end
end