@@ -398,17 +398,22 @@
        GitDaemon.git!(client, ["add", "a.txt"])
        GitDaemon.git!(client, ["commit", "-m", "first"])
        {out, _code} =
        {out, code} =
          GitDaemon.git_at(client, [
            "push",
            "git://127.0.0.1:#{port}/repo",
            "main:refs/heads/main"
          ])
        refute code == 0, "hook-rejected push should exit non-zero:\n#{out}"
        assert out =~ "hook_rejected"
        assert out =~ "CODEOWNERS approval required",
               "expected detail slice in rejection message, got:\n#{out}"
        assert {:error, _} = Ref.get(repo, "refs/heads/main"),
               "ref was updated despite hook rejection"
      after
        stop.()
      end

@@ -331,7 +331,18 @@
  end
  describe "partial clone (--filter)" do
    # Each filter test asserts directly on the pack contents via
    # `git cat-file --batch-check --batch-all-objects` in the cloned
    # repo. `--no-checkout` is used so that post-clone lazy-fetches
    # don't pollute the local store — after clone the only objects
    # present are exactly what the server sent. That's what lets us
    # prove the filter was honoured instead of relying on git's
    # `remote.origin.promisor=true` config (which git sets whenever
    # the server merely advertised `filter`, regardless of pack
    # contents).
    @tag :tmp_dir
    test "--filter=blob:none produces a pack with zero blobs",
         %{tmp_dir: tmp_dir} do
    test "--filter=blob:none produces a blobless pack", %{tmp_dir: tmp_dir} do
      repo = make_linear_repo("filter-blob", 3)
      {port, stop} = GitDaemon.start_upload_pack(repo)
@@ -345,25 +356,28 @@
            "protocol.version=2",
            "clone",
            "--filter=blob:none",
            "--no-checkout",
            "--no-local",
            "git://127.0.0.1:#{port}/repo",
            dest
          ])
        assert code == 0, "partial clone failed:\n#{out}"
        assert_promisor_configured(dest)
        counts = count_local_objects_by_type(dest)
        # Git considers a blobless clone "partial" when the repo-local
        # config has `remote.origin.promisor=true`. With a real filter
        # server the config gets set automatically; here we assert the
        # clone claims partial state.
        {cfg, _} = GitDaemon.git_at(dest, ["config", "--get", "remote.origin.promisor"])
        assert String.trim(cfg) == "true"
        assert counts.blob == 0, "blob:none must produce zero blobs, got #{counts.blob}"
        assert counts.commit == 3, "expected 3 commits in pack, got #{counts.commit}"
        assert counts.tree >= 3, "expected ≥ 3 trees in pack, got #{counts.tree}"
      after
        stop.()
      end
    end
    @tag :tmp_dir
    test "--filter=tree:0 produces a treeless pack", %{tmp_dir: tmp_dir} do
    test "--filter=tree:0 produces a pack with no trees and no blobs",
         %{tmp_dir: tmp_dir} do
      repo = make_linear_repo("filter-tree", 3)
      {port, stop} = GitDaemon.start_upload_pack(repo)
@@ -377,12 +391,26 @@
            "protocol.version=2",
            "clone",
            "--filter=tree:0",
            "--no-checkout",
            "--no-local",
            "git://127.0.0.1:#{port}/repo",
            dest
          ])
        assert code == 0, "tree:0 clone failed:\n#{out}"
        assert_promisor_configured(dest)
        counts = count_local_objects_by_type(dest)
        # tree:0 = include objects with tree-depth < 0, which per git's
        # "only the root tree per commit" semantics means no trees at
        # all in our emission (our walker emits depth-0 root trees
        # only when depth < limit, and limit 0 rules everything out).
        # Every commit must still be present.
        assert counts.tree == 0, "tree:0 must produce zero trees, got #{counts.tree}"
        assert counts.blob == 0, "tree:0 implies zero blobs, got #{counts.blob}"
        assert counts.commit == 3, "expected 3 commits, got #{counts.commit}"
      after
        stop.()
      end
@@ -393,6 +421,7 @@
    # pack that contains only commit objects; git considers this a
    # partial clone and sets promisor=true.
    @tag :tmp_dir
    test "--filter=object:type=commit yields a commits-only pack", %{tmp_dir: tmp_dir} do
    test "--filter=object:type=commit yields a pack with only commits",
         %{tmp_dir: tmp_dir} do
      repo = make_linear_repo("filter-obj-type", 3)
      {port, stop} = GitDaemon.start_upload_pack(repo)
@@ -406,6 +435,7 @@
            "protocol.version=2",
            "clone",
            "--filter=object:type=commit",
            "--no-checkout",
            "--no-local",
            "git://127.0.0.1:#{port}/repo",
            dest
@@ -413,17 +443,24 @@
        assert code == 0, "object:type=commit clone failed:\n#{out}"
        assert_promisor_configured(dest)
        counts = count_local_objects_by_type(dest)
        assert counts.commit == 3, "expected 3 commits, got #{counts.commit}"
        {cfg, _} = GitDaemon.git_at(dest, ["config", "--get", "remote.origin.promisor"])
        assert String.trim(cfg) == "true"
        assert counts.tree == 0, "object:type=commit must exclude trees, got #{counts.tree}"
        assert counts.blob == 0, "object:type=commit must exclude blobs, got #{counts.blob}"
        assert counts.tag == 0, "object:type=commit must exclude tags, got #{counts.tag}"
      after
        stop.()
      end
    end
    # `combine:blob:none+tree:1` applies both filters: `blob:none`
    # excludes every blob; `tree:1` keeps only trees at depth < 1,
    # i.e. only the root tree per commit (sub-trees and their entries
    # are at depth ≥ 1).
    # `combine:blob:none+tree:1` applies both filters: no blobs, and
    # only root trees + their direct entries (which, with blobs
    # already excluded, means just the root tree per commit).
    @tag :tmp_dir
    test "--filter=combine:… composes sub-filters", %{tmp_dir: tmp_dir} do
    test "--filter=combine:… applies every sub-filter", %{tmp_dir: tmp_dir} do
      repo = make_linear_repo("filter-combine", 3)
      {port, stop} = GitDaemon.start_upload_pack(repo)
@@ -437,15 +474,31 @@
            "protocol.version=2",
            "clone",
            "--filter=combine:blob:none+tree:1",
            "--no-checkout",
            "--no-local",
            "git://127.0.0.1:#{port}/repo",
            dest
          ])
        assert code == 0, "combine filter clone failed:\n#{out}"
        assert_promisor_configured(dest)
        {cfg, _} = GitDaemon.git_at(dest, ["config", "--get", "remote.origin.promisor"])
        assert String.trim(cfg) == "true"
        counts = count_local_objects_by_type(dest)
        assert counts.commit == 3, "expected 3 commits, got #{counts.commit}"
        assert counts.blob == 0,
               "combine:blob:none+tree:1 must exclude blobs, got #{counts.blob}"
        # Trees: make_linear_repo builds flat commits with one root
        # tree each (no sub-trees), so tree:1 keeps the 3 root trees
        # and nothing below (there is nothing below). The distinct
        # root-tree count is ≤ 3 (may be ≤ due to tree reuse, but
        # for this fixture each commit has a different blob-sha so
        # each tree is distinct).
        assert counts.tree > 0 and counts.tree <= 3,
               "expected 1..3 trees (only root level), got #{counts.tree}"
      after
        stop.()
      end
@@ -507,6 +560,7 @@
            "protocol.version=2",
            "clone",
            "--filter=sparse:oid=#{sparse_sha}",
            "--no-checkout",
            "--no-local",
            "git://127.0.0.1:#{port}/repo",
            dest
@@ -514,8 +568,18 @@
        assert code == 0, "sparse filter clone failed:\n#{out}"
        assert_promisor_configured(dest)
        # The pack must contain app_sha (under src/) but must NOT
        # contain test_sha (under test/). Trees and commits are
        # always included regardless of sparse:oid.
        shas = local_object_shas(dest)
        assert app_sha in shas,
               "sparse:oid clone missing the matching blob src/app.ex (#{app_sha})"
        {cfg, _} = GitDaemon.git_at(dest, ["config", "--get", "remote.origin.promisor"])
        assert String.trim(cfg) == "true"
        refute test_sha in shas,
               "sparse:oid clone included the excluded blob test/app_test.ex (#{test_sha})"
      after
        stop.()
      end
@@ -689,5 +753,42 @@
  defp walk_back(repo, sha, steps) do
    {:ok, %Commit{parents: [parent | _]}} = ExGitObjectstore.ObjectResolver.read(repo, sha)
    walk_back(repo, parent, steps - 1)
  end
  # Enumerate every object in the clone's local store and count by
  # git object type. Requires `--no-checkout` at clone time so
  # post-clone lazy-fetches don't pollute the count.
  defp count_local_objects_by_type(clone_dir) do
    {out, 0} =
      GitDaemon.git_at(clone_dir, ["cat-file", "--batch-check", "--batch-all-objects"])
    out
    |> String.split("\n", trim: true)
    |> Enum.reduce(%{blob: 0, tree: 0, commit: 0, tag: 0}, fn line, acc ->
      case String.split(line, " ", parts: 3) do
        [_sha, "blob", _size] -> Map.update!(acc, :blob, &(&1 + 1))
        [_sha, "tree", _size] -> Map.update!(acc, :tree, &(&1 + 1))
        [_sha, "commit", _size] -> Map.update!(acc, :commit, &(&1 + 1))
        [_sha, "tag", _size] -> Map.update!(acc, :tag, &(&1 + 1))
        _ -> acc
      end
    end)
  end
  defp local_object_shas(clone_dir) do
    {out, 0} =
      GitDaemon.git_at(clone_dir, ["cat-file", "--batch-check", "--batch-all-objects"])
    out
    |> String.split("\n", trim: true)
    |> Enum.map(fn line -> line |> String.split(" ", parts: 2) |> List.first() end)
    |> MapSet.new()
  end
  defp assert_promisor_configured(clone_dir) do
    {cfg, _} = GitDaemon.git_at(clone_dir, ["config", "--get", "remote.origin.promisor"])
    assert String.trim(cfg) == "true",
           "expected remote.origin.promisor=true in clone (indicates filter was honoured)"
  end
end

@@ -275,54 +275,96 @@
  end
  describe "concurrency" do
    # Two concurrent clients on the same daemon port — one pushes, the
    # other fetches. This doesn't verify snapshot isolation (the server
    # has no transactional snapshot semantics on top of the storage
    # backend), but it does ensure the daemon correctly serves
    # independent connections without head-of-line blocking or shared
    # mutable state leaking between them.
    # Two concurrent clones against the SAME upload-pack daemon port.
    # This forces a real accept-loop race: the daemon must spawn two
    # independent handler processes for two overlapping connections,
    # each with its own UploadPackV2 state. Shared mutable state
    # between handlers (or head-of-line blocking in the listener)
    # would show up here as a slow second clone or corrupted packs.
    @tag :tmp_dir
    test "two concurrent clones on the same upload-pack port",
         %{tmp_dir: tmp_dir} do
      repo = fresh_repo("concurrent-reads", [{"f.txt", "v1\n"}, {"f.txt", "v2\n"}])
      {port, stop} = GitDaemon.start_upload_pack(repo)
      try do
        url = "git://127.0.0.1:#{port}/repo"
        dest_a = Path.join(tmp_dir, "clone-a")
        dest_b = Path.join(tmp_dir, "clone-b")
    test "concurrent fetch and push on the same port", %{tmp_dir: tmp_dir} do
      repo =
        results =
          [fn -> GitDaemon.git_clone(url, dest_a) end, fn -> GitDaemon.git_clone(url, dest_b) end]
          |> Task.async_stream(& &1.(), timeout: 30_000, max_concurrency: 2)
          |> Enum.map(fn {:ok, r} -> r end)
        for {out, code} <- results do
          assert code == 0, "clone failed (exit #{code}):\n#{out}"
        end
        fresh_repo("concurrent", [
          {"f.txt", "v1\n"},
          {"f.txt", "v2\n"}
        ])
        # Both clones must see the same HEAD and the same objects —
        # server state did not get mutated by the parallel handling.
        {sha_a, _} = GitDaemon.git_at(dest_a, ["rev-parse", "HEAD"])
        {sha_b, _} = GitDaemon.git_at(dest_b, ["rev-parse", "HEAD"])
        assert String.trim(sha_a) == String.trim(sha_b)
      after
        stop.()
      end
    end
    # Overlap a fetch and a push on the SAME repo (shared storage).
    # We start both daemons bound to the same in-memory repo and run
    # them together. The test does not require linearisability — our
    # backends aren't multi-key transactional — but it does require
    # that neither operation crashes, leaves partial state, or
    # produces corrupt output.
    @tag :tmp_dir
    test "fetch and push against the same shared repo run to completion",
         %{tmp_dir: tmp_dir} do
      repo = fresh_repo("concurrent-rw", [{"f.txt", "v1\n"}, {"f.txt", "v2\n"}])
      {up_port, stop_up} = GitDaemon.start_upload_pack(repo)
      {rp_port, stop_rp} = GitDaemon.start_receive_pack(repo)
      try do
        # Prepare a push client.
        push_client = GitDaemon.init_client_dir(tmp_dir, "pusher")
        File.write!(Path.join(push_client, "new.txt"), "new\n")
        GitDaemon.git!(push_client, ["add", "new.txt"])
        GitDaemon.git!(push_client, ["commit", "-m", "pushed"])
        {push_sha, _} = GitDaemon.git_at(push_client, ["rev-parse", "HEAD"])
        push_sha = String.trim(push_sha)
        fetch_url = "git://127.0.0.1:#{up_port}/repo"
        push_url = "git://127.0.0.1:#{rp_port}/repo"
        dest = Path.join(tmp_dir, "reader")
        # Run both in parallel. `Task.async_stream` forces them onto
        # separate processes.
        results =
          [
            fn -> GitDaemon.git_clone(fetch_url, dest) end,
            fn ->
              dest = Path.join(tmp_dir, "clone")
              GitDaemon.git_clone("git://127.0.0.1:#{up_port}/repo", dest)
            end,
            fn ->
              GitDaemon.git_at(push_client, [
                "-c",
                "protocol.version=2",
                "push",
                push_url,
                "git://127.0.0.1:#{rp_port}/repo",
                "main:refs/heads/pushed"
              ])
            end
          ]
          |> Task.async_stream(& &1.(), timeout: 30_000, max_concurrency: 2)
          |> Task.async_stream(& &1.(), timeout: 30_000)
          |> Enum.map(fn {:ok, result} -> result end)
          |> Enum.map(fn {:ok, r} -> r end)
        for {out, code} <- results do
          assert code == 0, "operation failed (exit #{code}):\n#{out}"
        end
        # Ref actually landed on the server — push wasn't dropped.
        assert {:ok, ^push_sha} = Ref.get(repo, "refs/heads/pushed")
        # Clone finished with a well-formed repo (fsck-clean).
        {_fsck_out, fsck_code} = GitDaemon.git_at(dest, ["fsck", "--no-dangling"])
        assert fsck_code == 0, "clone is corrupt after concurrent push"
      after
        stop_up.()
        stop_rp.()

@@ -16,17 +16,22 @@
  @moduledoc """
  Operational coverage: every new code path added in the recent
  protocol-v2 work must emit a telemetry event with useful payload.
  Fetch / filter coverage drives the UploadPackV2 state machine
  directly (no network). Atomic receive-pack coverage drives a
  real `git push --atomic` subprocess against a TCP daemon so the
  telemetry path executes exactly as it would in production.
  """
  use ExUnit.Case, async: false
  alias ExGitObjectstore.Object
  alias ExGitObjectstore.Object.{Blob, Commit, Tree}
  alias ExGitObjectstore.Protocol.{PktLine, ReceivePack, UploadPackV2}
  alias ExGitObjectstore.Protocol.{PktLine, UploadPackV2}
  alias ExGitObjectstore.Ref
  alias ExGitObjectstore.Test.RepoHelper
  alias ExGitObjectstore.Test.{GitDaemon, RepoHelper}
  @zero_sha String.duplicate("0", 40)
  @moduletag timeout: :timer.minutes(1)
  setup do
    test_pid = self()
@@ -139,33 +144,44 @@
  end
  describe "atomic receive-pack telemetry" do
    @tag :integration
    @tag :tmp_dir
    test "emits :start and :stop with outcome=:committed on real `git push --atomic`",
    test "emits :start and :stop with outcome=:committed on success" do
         %{tmp_dir: tmp_dir} do
      repo = RepoHelper.memory_repo("tel-atomic-ok")
      ExGitObjectstore.init(repo)
      commit_sha = make_commit(repo, "content\n")
      {port, stop} = GitDaemon.start_receive_pack(repo)
      try do
        client = GitDaemon.init_client_dir(tmp_dir)
        File.write!(Path.join(client, "a.txt"), "hi\n")
        GitDaemon.git!(client, ["add", "a.txt"])
        GitDaemon.git!(client, ["commit", "-m", "first"])
      {_advert, state} = ReceivePack.init(repo)
      state = %{
        state
        {_out, code} =
          GitDaemon.git_at(client, [
            "push",
            "--atomic",
            "git://127.0.0.1:#{port}/repo",
            "main:refs/heads/a",
            "main:refs/heads/b"
          ])
        assert code == 0
        | client_caps: MapSet.new(["report-status", "atomic"]),
          commands: [
            %{ref: "refs/heads/a", old_sha: @zero_sha, new_sha: commit_sha},
            %{ref: "refs/heads/b", old_sha: @zero_sha, new_sha: commit_sha}
          ],
          phase: :pack
      }
      # Drive through the pack-stage to trigger ref-update processing.
      # With no pack bytes and all-creates, ReceivePack treats it as
      # the "all-creates, pack optional" path and runs ref updates.
        assert_receive {:telemetry,
                        [:ex_git_objectstore, :protocol, :receive_pack, :atomic, :stop],
                        %{duration: duration},
                        %{outcome: :committed, commands: 2, validation_failures: 0}}
      {_report, _final} = ReceivePack.feed(state, empty_pack())
      assert_receive {:telemetry, [:ex_git_objectstore, :protocol, :receive_pack, :atomic, :stop],
                      %{duration: duration},
        assert duration > 0
                      %{outcome: :committed, commands: 2, validation_failures: 0}}
        # And the refs actually landed.
        {:ok, _} = Ref.get(repo, "refs/heads/a")
        {:ok, _} = Ref.get(repo, "refs/heads/b")
      after
      assert duration > 0
        stop.()
      end
    end
  end
@@ -188,14 +204,5 @@
    {:ok, sha} = Object.write(repo, commit)
    sha
  end
  # Empty pack (header + checksum) — enough for ReceivePack's
  # `check_pack_complete/1` to declare the pack phase done so the
  # state machine runs ref updates.
  defp empty_pack do
    header = <<"PACK", 2::unsigned-big-32, 0::unsigned-big-32>>
    checksum = :crypto.hash(:sha, header)
    <<header::binary, checksum::binary>>
  end
end