ref:1bd54cd6cefabac7d52ef7088fda9ed02bc2f1ec

fix: add buffering to UploadPackV2 for split SSH data

SSH can split protocol data across multiple messages. The v2 state machine was dispatching commands on each data message without waiting for a complete command (terminated by flush). This caused "expected packfile" errors when the fetch command arrived in multiple fragments. Added a buffer field to accumulate partial data and only dispatch when a flush-terminated command is complete. Mirrors the existing negotiate_buffer approach in UploadPack (v1). Also fixes alias ordering (credo --strict) and adds comprehensive unit tests for the buffering behavior. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
SHA: 1bd54cd6cefabac7d52ef7088fda9ed02bc2f1ec
Author: Cole Christensen <cole.christensen@macmillan.com>
Date: 2026-03-14 00:39
Parents: 55327c2
2 files changed +341 -8
Type
lib/ex_git_objectstore/protocol/upload_pack_v2.ex +27 −8
@@ -30,8 +30,8 @@
This is a pure functional state machine — no processes.
"""
alias ExGitObjectstore.{ObjectResolver, Ref, Repo}
alias ExGitObjectstore.Object
alias ExGitObjectstore.Object.{Blob, Commit, Tag, Tree}
alias ExGitObjectstore.{ObjectResolver, Ref, Repo}
alias ExGitObjectstore.Pack.Writer
alias ExGitObjectstore.Protocol.PktLine
@@ -40,10 +40,11 @@
@type state :: %__MODULE__{
repo: Repo.t(),
phase: :command | :done,
phase: :command | :done
buffer: binary()
}
defstruct [:repo, phase: :command]
defstruct [:repo, phase: :command, buffer: <<>>]
@doc """
Create a new v2 upload-pack state machine and generate the capability advertisement.
@@ -59,19 +60,30 @@
@doc """
Feed a v2 command from the client into the state machine.
Returns `{response_data, new_state}`.
SSH can split protocol data across multiple messages, so we buffer
incomplete data until a complete command (terminated by a flush packet)
arrives before dispatching.
"""
@spec feed(state(), binary()) :: {binary(), state()}
def feed(%__MODULE__{phase: :command} = state, data) do
# Prepend any buffered partial data from previous feed calls.
full_data = state.buffer <> data
case parse_command(data) do
case parse_command(full_data) do
{:ls_refs, args} ->
response = handle_ls_refs(state.repo, args)
{response, state}
{response, %{state | buffer: <<>>}}
{:fetch, args} ->
response = handle_fetch(state.repo, args)
{response, %{state | phase: :done, buffer: <<>>}}
{:incomplete, _rest} ->
# Not enough data yet — buffer and wait for more
{<<>>, %{state | buffer: full_data}}
{response, %{state | phase: :done}}
{:error, _} ->
{PktLine.flush(), %{state | phase: :done}}
{PktLine.flush(), %{state | phase: :done, buffer: <<>>}}
end
end
@@ -106,7 +118,14 @@
defp parse_command(data) do
case PktLine.decode(data) do
{:ok, packets, _rest} ->
# A complete v2 command is terminated by a flush packet.
dispatch_command(packets)
# If we decoded packets but there's no flush, the data is
# incomplete (split across SSH messages) — buffer it.
if :flush in packets do
dispatch_command(packets)
else
{:incomplete, data}
end
{:error, _} = err ->
err
test/ex_git_objectstore/protocol/upload_pack_v2_test.exs +314 −0
@@ -1,0 +1,314 @@
# 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.UploadPackV2Test do
use ExUnit.Case, async: true
alias ExGitObjectstore.{Object, Ref}
alias ExGitObjectstore.Object.{Blob, Commit, Tree}
alias ExGitObjectstore.Pack.Reader
alias ExGitObjectstore.Protocol.{PktLine, UploadPackV2}
alias ExGitObjectstore.Test.RepoHelper
defp create_commit(repo, content, message, parents \\ []) do
blob = Blob.from_content(content)
{:ok, blob_sha} = Object.write(repo, blob)
tree = Tree.new([%{mode: "100644", name: "file.txt", sha: blob_sha}])
{:ok, tree_sha} = Object.write(repo, tree)
commit = %Commit{
tree: tree_sha,
parents: parents,
author: "Test <t@t.com> 1000000000 +0000",
committer: "Test <t@t.com> 1000000000 +0000",
message: message
}
{:ok, commit_sha} = Object.write(repo, commit)
{commit_sha, blob_sha, tree_sha}
end
describe "init" do
test "advertises v2 capabilities" do
repo = RepoHelper.memory_repo()
ExGitObjectstore.init(repo)
{advert, state} = UploadPackV2.init(repo)
assert state.phase == :command
assert state.buffer == <<>>
{:ok, packets, _rest} = PktLine.decode(advert)
data_lines = for {:data, d} <- packets, do: d
assert "version 2" in data_lines
assert "ls-refs" in data_lines
assert :flush in packets
end
end
describe "ls-refs" do
test "returns refs for a repo with commits" do
repo = RepoHelper.memory_repo()
ExGitObjectstore.init(repo)
{commit_sha, _, _} = create_commit(repo, "hello", "init")
:ok = Ref.put(repo, "refs/heads/main", commit_sha, nil)
{_advert, state} = UploadPackV2.init(repo)
# Build a ls-refs command
ls_refs_cmd =
IO.iodata_to_binary([
PktLine.encode("command=ls-refs"),
PktLine.delim(),
PktLine.encode("ref-prefix refs/heads/"),
PktLine.flush()
])
{response, new_state} = UploadPackV2.feed(state, ls_refs_cmd)
# Should still be in :command phase (ls-refs doesn't end the session)
assert new_state.phase == :command
assert new_state.buffer == <<>>
assert byte_size(response) > 0
{:ok, packets, _} = PktLine.decode(response)
data_lines = for {:data, d} <- packets, do: d
assert Enum.any?(data_lines, &String.contains?(&1, commit_sha))
end
end
describe "fetch" do
test "returns packfile for a simple clone" do
repo = RepoHelper.memory_repo()
ExGitObjectstore.init(repo)
{commit_sha, _, _} = create_commit(repo, "hello", "init")
:ok = Ref.put(repo, "refs/heads/main", commit_sha, nil)
{_advert, state} = UploadPackV2.init(repo)
# Build a fetch command
fetch_cmd =
IO.iodata_to_binary([
PktLine.encode("command=fetch"),
PktLine.delim(),
PktLine.encode("want #{commit_sha}"),
PktLine.encode("done"),
PktLine.flush()
])
{response, new_state} = UploadPackV2.feed(state, fetch_cmd)
assert new_state.phase == :done
assert byte_size(response) > 0
# Response should contain "packfile" section
{:ok, packets, _} = PktLine.decode(response)
data_lines = for {:data, d} <- packets, do: d
assert "packfile" in data_lines
end
end
describe "buffering split data" do
test "handles ls-refs command split across two feed calls" do
repo = RepoHelper.memory_repo()
ExGitObjectstore.init(repo)
{commit_sha, _, _} = create_commit(repo, "hello", "init")
:ok = Ref.put(repo, "refs/heads/main", commit_sha, nil)
{_advert, state} = UploadPackV2.init(repo)
# Build a complete ls-refs command
full_cmd =
IO.iodata_to_binary([
PktLine.encode("command=ls-refs"),
PktLine.delim(),
PktLine.encode("ref-prefix refs/heads/"),
PktLine.flush()
])
# Split at an arbitrary point mid-command
split_point = div(byte_size(full_cmd), 2)
<<part1::binary-size(split_point), part2::binary>> = full_cmd
# First feed: should buffer, return empty response
{response1, state1} = UploadPackV2.feed(state, part1)
assert response1 == <<>>
assert state1.phase == :command
assert state1.buffer == part1
# Second feed: should complete the command
{response2, state2} = UploadPackV2.feed(state1, part2)
assert state2.phase == :command
assert state2.buffer == <<>>
assert byte_size(response2) > 0
{:ok, packets, _} = PktLine.decode(response2)
data_lines = for {:data, d} <- packets, do: d
assert Enum.any?(data_lines, &String.contains?(&1, commit_sha))
end
test "handles fetch command split across three feed calls" do
repo = RepoHelper.memory_repo()
ExGitObjectstore.init(repo)
{commit_sha, _, _} = create_commit(repo, "content", "commit")
:ok = Ref.put(repo, "refs/heads/main", commit_sha, nil)
{_advert, state} = UploadPackV2.init(repo)
# Build a complete fetch command
full_cmd =
IO.iodata_to_binary([
PktLine.encode("command=fetch"),
PktLine.delim(),
PktLine.encode("want #{commit_sha}"),
PktLine.encode("done"),
PktLine.flush()
])
# Split into three parts
size = byte_size(full_cmd)
s1 = div(size, 3)
s2 = div(size * 2, 3)
<<p1::binary-size(s1), rest::binary>> = full_cmd
<<p2::binary-size(s2 - s1), p3::binary>> = rest
# Feed part 1
{r1, st1} = UploadPackV2.feed(state, p1)
assert r1 == <<>>
assert st1.phase == :command
# Feed part 2
{r2, st2} = UploadPackV2.feed(st1, p2)
assert r2 == <<>>
assert st2.phase == :command
# Feed part 3 — should complete
{r3, st3} = UploadPackV2.feed(st2, p3)
assert st3.phase == :done
assert byte_size(r3) > 0
{:ok, packets, _} = PktLine.decode(r3)
data_lines = for {:data, d} <- packets, do: d
assert "packfile" in data_lines
end
test "handles split at exact pkt-line boundary" do
repo = RepoHelper.memory_repo()
ExGitObjectstore.init(repo)
{commit_sha, _, _} = create_commit(repo, "test", "init")
:ok = Ref.put(repo, "refs/heads/main", commit_sha, nil)
{_advert, state} = UploadPackV2.init(repo)
# Split exactly between the command line and the delim
command_line = PktLine.encode("command=ls-refs")
rest_of_cmd = IO.iodata_to_binary([PktLine.delim(), PktLine.flush()])
{r1, st1} = UploadPackV2.feed(state, command_line)
assert r1 == <<>>
assert st1.phase == :command
{r2, st2} = UploadPackV2.feed(st1, rest_of_cmd)
assert st2.phase == :command
assert byte_size(r2) > 0
end
end
describe "full v2 conversation" do
test "ls-refs followed by fetch" do
repo = RepoHelper.memory_repo()
ExGitObjectstore.init(repo)
{commit_sha, _, _} = create_commit(repo, "hello world", "initial")
:ok = Ref.put(repo, "refs/heads/main", commit_sha, nil)
{_advert, state} = UploadPackV2.init(repo)
# Step 1: ls-refs
ls_refs =
IO.iodata_to_binary([
PktLine.encode("command=ls-refs"),
PktLine.delim(),
PktLine.encode("ref-prefix refs/heads/"),
PktLine.flush()
])
{ls_response, state} = UploadPackV2.feed(state, ls_refs)
assert state.phase == :command
{:ok, ls_packets, _} = PktLine.decode(ls_response)
data_lines = for {:data, d} <- ls_packets, do: d
assert Enum.any?(data_lines, &String.contains?(&1, commit_sha))
# Step 2: fetch
fetch =
IO.iodata_to_binary([
PktLine.encode("command=fetch"),
PktLine.delim(),
PktLine.encode("want #{commit_sha}"),
PktLine.encode("done"),
PktLine.flush()
])
{fetch_response, state} = UploadPackV2.feed(state, fetch)
assert state.phase == :done
assert byte_size(fetch_response) > 0
# Verify packfile in response
{:ok, fetch_packets, _} = PktLine.decode(fetch_response)
data_lines = for {:data, d} <- fetch_packets, do: d
assert "packfile" in data_lines
# Extract and verify pack data
pack_data = extract_pack_data(fetch_response)
{:ok, objects} = Reader.parse(pack_data)
types = Enum.map(objects, fn obj -> obj.type end)
assert :commit in types
assert :tree in types
assert :blob in types
end
end
# Extract raw pack data from sideband-encoded response
defp extract_pack_data(response) do
{:ok, packets, _} = PktLine.decode(response)
# Find the packfile marker, then collect sideband-1 data
{_, after_packfile} =
Enum.split_while(packets, fn
{:data, "packfile"} -> false
_ -> true
end)
after_packfile
|> tl()
|> Enum.flat_map(fn
{:data, data} ->
case PktLine.decode_sideband(data) do
{:pack, pack} -> [pack]
_ -> []
end
_ ->
[]
end)
|> IO.iodata_to_binary()
end
end