@@ -24,5 +24,13 @@
- Configurable per-repo `max_object_size` via `Repo.new/2` (default 128MB unchanged)
- CalVer release automation via CI (`ci/release.sh`, `.anvil.yml` release step)
- Telemetry events for object read/write, ref updates, and receive-pack protocol
- S3 backend parallelism: `list_refs/3` now issues per-ref GETs concurrently
  (`Task.async_stream` with `max_concurrency: 32`), and `put_pack/5` uploads
  the `.pack` and `.idx` files concurrently. Drops protocol-advertisement
  latency on ref-heavy repos from ~25 s → ~1 s. Tunable via
  `list_refs_concurrency`, `list_refs_timeout`, and `put_pack_timeout` in the
  S3 config map. See fangorn/ex_git_objectstore#25.
- Storage backend telemetry: `[:ex_git_objectstore, :storage, :list_refs]`
  and `[:ex_git_objectstore, :storage, :put_pack]` spans (S3 backend only for now)
### Fixed

@@ -25,6 +25,14 @@
  this is fine. For multi-writer scenarios, use an external coordination service
  (e.g., DynamoDB-based locking) to serialize ref updates.
  ## put_pack partial-write semantics
  `put_pack/5` uploads the `.pack` and `.idx` objects concurrently. If one
  upload succeeds and the other fails, the successful object is left in place
  and the function returns `{:error, reason}`. A `.pack` without a matching
  `.idx` is unreachable through any lookup path, so GC/fsck will reclaim it.
  Callers retrying with the same `pack_sha` will overwrite the orphan.
  ## Config
      %{
@@ -36,9 +44,17 @@
          host: "localhost",      # for MinIO
          port: 9000,
          scheme: "http://"
        ],
        # -- Optional tuning for parallelized operations --
        list_refs_concurrency: 32,   # parallel GETs per list_refs call (default 32)
        list_refs_timeout: 30_000,   # per-GET timeout in ms (default 30_000)
        ]
        put_pack_timeout: 60_000     # await timeout for concurrent pack+idx PUTs (default 60_000)
      }
  Tune `list_refs_concurrency` higher for ref-heavy repos if the underlying
  HTTP client pool has the capacity (hackney default pool size is 50).
  ## S3 key layout
      <prefix>/HEAD
@@ -133,9 +149,31 @@
  @impl true
  def put_pack(config, prefix, pack_sha, pack_data, idx_data) do
    with :ok <- s3_put(config, pack_key(prefix, pack_sha, "pack"), pack_data) do
      s3_put(config, pack_key(prefix, pack_sha, "idx"), idx_data)
    end
    pack_size = byte_size(pack_data)
    idx_size = byte_size(idx_data)
    :telemetry.span(
      [:ex_git_objectstore, :storage, :put_pack],
      %{pack_sha: pack_sha, backend: :s3},
      fn ->
        timeout = Map.get(config, :put_pack_timeout, 60_000)
        pack_task =
          Task.async(fn -> s3_put(config, pack_key(prefix, pack_sha, "pack"), pack_data) end)
        idx_task =
          Task.async(fn -> s3_put(config, pack_key(prefix, pack_sha, "idx"), idx_data) end)
        result =
          case Task.await_many([pack_task, idx_task], timeout) do
            [:ok, :ok] -> :ok
            [{:error, reason}, _] -> {:error, reason}
            [_, {:error, reason}] -> {:error, reason}
          end
        {result, %{pack_size: pack_size, idx_size: idx_size}, %{pack_sha: pack_sha, backend: :s3}}
      end
    )
  end
  @impl true
@@ -198,23 +236,48 @@
  @impl true
  def list_refs(config, prefix, ref_prefix) do
    :telemetry.span(
      [:ex_git_objectstore, :storage, :list_refs],
      %{ref_prefix: ref_prefix, backend: :s3},
      fn ->
    full_prefix = "#{prefix}/#{ref_prefix}"
        full_prefix = "#{prefix}/#{ref_prefix}"
        max_concurrency = Map.get(config, :list_refs_concurrency, 32)
        timeout = Map.get(config, :list_refs_timeout, 30_000)
        result =
          case s3_list(config, full_prefix) do
            {:ok, keys} ->
              refs =
                keys
                |> Task.async_stream(
                  fn key -> fetch_ref_from_key(config, key, prefix) end,
                  max_concurrency: max_concurrency,
                  timeout: timeout,
                  on_timeout: :kill_task,
                  ordered: false
                )
                |> Enum.reduce([], fn
                  {:ok, {_name, _sha} = pair}, acc -> [pair | acc]
                  {:ok, nil}, acc -> acc
                  {:exit, _reason}, acc -> acc
                end)
                |> Enum.sort()
              {:ok, refs}
    case s3_list(config, full_prefix) do
      {:ok, keys} ->
        refs =
          keys
          |> Enum.map(fn key ->
            fetch_ref_from_key(config, key, prefix)
          end)
          |> Enum.reject(&is_nil/1)
          |> Enum.sort()
            {:error, _} = err ->
              err
          end
        {:ok, refs}
        ref_count =
          case result do
            {:ok, refs} -> length(refs)
            _ -> 0
          end
      {:error, _} = err ->
        err
    end
        {result, %{ref_count: ref_count}, %{ref_prefix: ref_prefix, backend: :s3}}
      end
    )
  end
  defp fetch_ref_from_key(config, key, prefix) do

@@ -56,6 +56,22 @@
  * `[:ex_git_objectstore, :protocol, :upload_pack, :start | :stop | :exception]`
    - Measurements: `:system_time` (start), `:duration` (stop)
    - Metadata: `:repo_id`
  ### Storage Backend Operations
  These events are emitted by specific storage backends to measure backend-level
  I/O. Currently only the S3 backend emits them; Filesystem and Memory backends
  may adopt them in the future.
  * `[:ex_git_objectstore, :storage, :list_refs, :start | :stop | :exception]`
    - Start measurements: `:system_time`, `:monotonic_time`
    - Stop measurements: `:monotonic_time`, `:duration`, `:ref_count`
    - Metadata: `:ref_prefix`, `:backend`
  * `[:ex_git_objectstore, :storage, :put_pack, :start | :stop | :exception]`
    - Start measurements: `:system_time`, `:monotonic_time`
    - Stop measurements: `:monotonic_time`, `:duration`, `:pack_size`, `:idx_size`
    - Metadata: `:pack_sha`, `:backend`
  """
  @doc false

@@ -303,5 +303,6 @@
      {:ok, listed_refs} = Repo.storage_call(repo, :list_refs, ["refs/heads/"])
      assert length(listed_refs) == ref_count
      assert listed_refs == Enum.sort(listed_refs), "list_refs result must be sorted"
      listed_set = MapSet.new(listed_refs)
@@ -310,5 +311,104 @@
        assert {ref_name, sha} in listed_set,
               "Expected ref #{ref_name} -> #{sha} to be in list_refs result"
      end
    end
  end
  describe "parallelized operations" do
    test "list_refs returns all refs under concurrency: 1", %{unique_id: unique_id} do
      # Smoke test that the concurrency tuning path works end-to-end.
      # Sets max_concurrency: 1, effectively sequential, and verifies correctness.
      config = Map.put(@minio_config, :list_refs_concurrency, 1)
      repo = Repo.new(unique_id, storage: {S3, config})
      refs =
        for i <- 1..10 do
          sha = :crypto.hash(:sha, "low-conc-#{i}") |> Base.encode16(case: :lower)
          padded = String.pad_leading("#{i}", 2, "0")
          ref_name = "refs/heads/lc-#{padded}"
          :ok = Repo.storage_call(repo, :put_ref, [ref_name, sha, nil])
          {ref_name, sha}
        end
      {:ok, listed} = Repo.storage_call(repo, :list_refs, ["refs/heads/"])
      assert length(listed) == length(refs)
      assert MapSet.new(listed) == MapSet.new(refs)
      assert listed == Enum.sort(listed)
    end
    test "put_pack round-trips both pack and idx concurrently", %{repo: repo} do
      # Larger bytes than the basic put_pack test to make concurrent uploads realistic.
      pack_data = "PACK" <> <<0, 0, 0, 2>> <> :crypto.strong_rand_bytes(128 * 1024)
      idx_data = :crypto.strong_rand_bytes(32 * 1024)
      pack_sha = :crypto.hash(:sha, pack_data) |> Base.encode16(case: :lower)
      assert :ok = Repo.storage_call(repo, :put_pack, [pack_sha, pack_data, idx_data])
      assert {:ok, ^pack_data} = Repo.storage_call(repo, :get_pack, [pack_sha])
      assert {:ok, ^idx_data} = Repo.storage_call(repo, :get_pack_index, [pack_sha])
    end
    test "put_pack propagates error when bucket does not exist", %{unique_id: unique_id} do
      # Use a nonexistent bucket to force a PUT failure on both tasks.
      bad_config = %{@minio_config | bucket: "this-bucket-does-not-exist-#{unique_id}"}
      bad_repo = Repo.new(unique_id, storage: {S3, bad_config})
      result =
        Repo.storage_call(bad_repo, :put_pack, ["deadbeef", "pack", "idx"])
      assert match?({:error, _}, result)
    end
  end
  describe "storage telemetry events" do
    setup %{unique_id: unique_id} = ctx do
      test_pid = self()
      handler_ids =
        Enum.map(
          [
            [:ex_git_objectstore, :storage, :list_refs, :stop],
            [:ex_git_objectstore, :storage, :put_pack, :stop]
          ],
          fn event ->
            id = "tel-#{Enum.join(event, ".")}-#{unique_id}"
            :telemetry.attach(
              id,
              event,
              fn e, measurements, metadata, _ ->
                send(test_pid, {:telemetry, e, measurements, metadata})
              end,
              nil
            )
            id
          end
        )
      on_exit(fn -> Enum.each(handler_ids, &:telemetry.detach/1) end)
      ctx
    end
    test "list_refs emits start/stop with backend and ref_count", %{repo: repo} do
      :ok = Repo.storage_call(repo, :put_ref, ["refs/heads/main", String.duplicate("a", 40), nil])
      {:ok, _} = Repo.storage_call(repo, :list_refs, ["refs/heads/"])
      assert_received {:telemetry, [:ex_git_objectstore, :storage, :list_refs, :stop],
                       %{duration: _, ref_count: 1}, %{ref_prefix: "refs/heads/", backend: :s3}}
    end
    test "put_pack emits start/stop with sizes and backend", %{repo: repo} do
      pack_data = "PACK" <> <<0, 0, 0, 2>> <> "body"
      idx_data = "idx"
      :ok = Repo.storage_call(repo, :put_pack, ["tel-pack", pack_data, idx_data])
      pack_size = byte_size(pack_data)
      idx_size = byte_size(idx_data)
      assert_received {:telemetry, [:ex_git_objectstore, :storage, :put_pack, :stop],
                       %{duration: _, pack_size: ^pack_size, idx_size: ^idx_size},
                       %{pack_sha: "tel-pack", backend: :s3}}
    end
  end