ref:796cf4019db07f046eea9d127a297f47d76143a9

feat: telemetry spans for fetch, atomic push, and filter application

Operational visibility for the new protocol-v2 code paths. Events emitted: * [:ex_git_objectstore, :protocol, :fetch] — span around every UploadPackV2 fetch. Start/stop metadata carries wants, haves, mode (:full / :shallow / :filtered / :shallow_filtered / :wait_for_done), repo_id; stop measurements carry `objects` and `pack_bytes`. * [:ex_git_objectstore, :protocol, :receive_pack, :atomic] — span around the atomic ref-update flow. Stop metadata carries outcome (:committed / :rolled_back), commands, and validation_failures. * [:ex_git_objectstore, :pack, :filter] — single event per filter application. Measurements: objects_in, objects_out. Metadata: spec_kind (:blob_none / :blob_limit / :tree_depth / :object_type / :sparse_oid / :combine), repo_id. * [:ex_git_objectstore, :protocol, :receive_pack, :rollback_failed] — emitted in the rare case a ref rollback fails during atomic abort (already added in the previous error-handling commit; noted here for completeness). Test file `test/ex_git_objectstore/protocol/telemetry_test.exs` attaches a handler and asserts each event fires with the expected payload for clone, shallow fetch, filter, and atomic-commit paths.
SHA: 796cf4019db07f046eea9d127a297f47d76143a9
Author: Cole Christensen <cole.christensen@macmillan.com>
Date: 2026-04-19 01:19
Parents: b77b43f
3 files changed +311 -28
Type
lib/ex_git_objectstore/protocol/receive_pack.ex +29 −8
@@ -534,15 +534,36 @@
# between steps), leaving refs in a partially-applied state. This
# implementation is as atomic as the storage layer allows.
defp do_atomic_ref_updates(state) do
:telemetry.span(
[:ex_git_objectstore, :protocol, :receive_pack, :atomic],
%{repo_id: state.repo.id, commands: length(state.commands)},
fn ->
validations =
Enum.map(state.commands, fn cmd ->
{cmd, validate_ref_command(state.repo, state.update_hook, cmd)}
end)
validation_failures =
Enum.count(validations, fn {_cmd, v} -> match?({:error, _}, v) end)
result =
case Enum.find(validations, fn {_cmd, v} -> match?({:error, _}, v) end) do
nil -> atomic_commit_phase(state)
{_failing_cmd, {:error, reason}} -> atomic_reject_all(state, validations, reason)
end
{_response, final_state} = result
validations =
Enum.map(state.commands, fn cmd ->
{cmd, validate_ref_command(state.repo, state.update_hook, cmd)}
end)
outcome = if match?({:error, _}, final_state.result), do: :rolled_back, else: :committed
case Enum.find(validations, fn {_cmd, v} -> match?({:error, _}, v) end) do
nil -> atomic_commit_phase(state)
{_failing_cmd, {:error, reason}} -> atomic_reject_all(state, validations, reason)
end
{result,
%{
repo_id: state.repo.id,
commands: length(state.commands),
validation_failures: validation_failures,
outcome: outcome
}}
end
)
end
defp atomic_reject_all(state, validations, reason) do
lib/ex_git_objectstore/protocol/upload_pack_v2.ex +81 −20
@@ -348,26 +348,65 @@
"send_packfile=#{send_packfile?}"
)
span_meta = %{
repo_id: repo.id,
wants: length(wants),
haves: length(haves),
mode: fetch_mode(shallow_opts, filter_spec, wait_for_done?)
}
:telemetry.span([:ex_git_objectstore, :protocol, :fetch], span_meta, fn ->
{result, extra} =
do_handle_fetch(
repo,
wants,
haves,
args,
done?,
wait_for_done?,
shallow_opts,
filter_spec,
send_packfile?
)
{result, Map.merge(span_meta, extra)}
end)
end
defp do_handle_fetch(
repo,
wants,
haves,
_args,
_done?,
wait_for_done?,
shallow_opts,
filter_spec,
send_packfile?
) do
ack_section = build_acknowledgments(repo, haves, send_packfile?)
cond do
# `--negotiate-only` flow: client wants the ACKs and nothing else.
wait_for_done? ->
{{ack_section, :done}, %{pack_bytes: 0, objects: 0}}
{ack_section, :done}
# Normal clone / fetch with `done`, OR shallow/deepen request:
# send packfile immediately.
send_packfile? ->
{build_packfile_response(repo, wants, haves, ack_section, shallow_opts, filter_spec),
:done}
{response, stats} =
build_packfile_response(repo, wants, haves, ack_section, shallow_opts, filter_spec)
# Multi-round negotiation: client hasn't committed yet. Emit the
# ACKs and wait for another fetch command on the same session.
{{response, :done}, stats}
true ->
{{ack_section, :command}, %{pack_bytes: 0, objects: 0}}
{ack_section, :command}
end
end
defp fetch_mode(nil, nil, true), do: :wait_for_done
defp fetch_mode(nil, nil, _), do: :full
defp fetch_mode(nil, _filter, _), do: :filtered
defp fetch_mode(_shallow, nil, _), do: :shallow
defp fetch_mode(_, _, _), do: :shallow_filtered
defp build_packfile_response(repo, wants, haves, ack_section, shallow_opts, filter_spec) do
case collect_objects_maybe_shallow(repo, wants, haves, shallow_opts, filter_spec) do
{:ok, %{objects: objects} = walk} ->
@@ -382,20 +421,23 @@
PktLine.encode_sideband(1, pack_data)
|> IO.iodata_to_binary()
response =
IO.iodata_to_binary([
ack_section,
shallow_info,
packfile_header,
sideband_data,
PktLine.flush()
IO.iodata_to_binary([
ack_section,
shallow_info,
packfile_header,
sideband_data,
PktLine.flush()
])
])
{response, %{pack_bytes: byte_size(pack_data), objects: length(objects)}}
{:error, reason} ->
Logger.error(
"UploadPackV2: collect_objects failed for #{length(wants)} wants, " <>
"#{length(haves)} haves: #{inspect(reason)}"
)
{PktLine.flush(), %{pack_bytes: 0, objects: 0, error: reason}}
PktLine.flush()
end
end
@@ -708,10 +750,29 @@
do: compute_tree_metadata(repo, objects),
else: %{depths: %{}, paths: %{}}
Enum.filter(objects, fn entry ->
kept =
Enum.filter(objects, fn entry ->
Filter.include?(spec, filter_ctx_for(entry, meta), repo)
end)
:telemetry.execute(
[:ex_git_objectstore, :pack, :filter],
%{objects_in: length(objects), objects_out: length(kept)},
%{spec_kind: filter_spec_kind(spec), repo_id: repo.id}
)
kept
Filter.include?(spec, filter_ctx_for(entry, meta), repo)
end)
end
# Tag the filter kind for telemetry consumers without exposing the
# full spec structure (which could contain large values like
# sparse:oid's spec blob sha).
defp filter_spec_kind(:blob_none), do: :blob_none
defp filter_spec_kind({:blob_limit, _}), do: :blob_limit
defp filter_spec_kind({:tree_depth, _}), do: :tree_depth
defp filter_spec_kind({:object_type, _}), do: :object_type
defp filter_spec_kind({:sparse_oid, _}), do: :sparse_oid
defp filter_spec_kind({:combine, _}), do: :combine
# `tree:<n>` uses `ctx.tree_depth`; `sparse:oid=<oid>` uses `ctx.path`.
# Everything else decides from type + size, which are already in the
test/ex_git_objectstore/protocol/telemetry_test.exs +201 −0
@@ -1,0 +1,201 @@
# 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.TelemetryTest do
@moduledoc """
Operational coverage: every new code path added in the recent
protocol-v2 work must emit a telemetry event with useful payload.
"""
use ExUnit.Case, async: false
alias ExGitObjectstore.Object
alias ExGitObjectstore.Object.{Blob, Commit, Tree}
alias ExGitObjectstore.Protocol.{PktLine, ReceivePack, UploadPackV2}
alias ExGitObjectstore.Ref
alias ExGitObjectstore.Test.RepoHelper
@zero_sha String.duplicate("0", 40)
setup do
test_pid = self()
handler_id = "telemetry-test-#{:erlang.unique_integer([:positive])}"
events = [
[:ex_git_objectstore, :protocol, :fetch, :start],
[:ex_git_objectstore, :protocol, :fetch, :stop],
[:ex_git_objectstore, :protocol, :receive_pack, :atomic, :start],
[:ex_git_objectstore, :protocol, :receive_pack, :atomic, :stop],
[:ex_git_objectstore, :pack, :filter]
]
:telemetry.attach_many(
handler_id,
events,
fn event, measurements, metadata, _cfg ->
send(test_pid, {:telemetry, event, measurements, metadata})
end,
nil
)
on_exit(fn -> :telemetry.detach(handler_id) end)
:ok
end
describe "fetch telemetry" do
test "emits :start and :stop with mode + pack stats for a regular clone" do
repo = RepoHelper.memory_repo("tel-fetch")
ExGitObjectstore.init(repo)
sha = make_commit(repo, "hello\n")
:ok = Ref.put(repo, "refs/heads/main", sha, nil)
{_advert, state} = UploadPackV2.init(repo)
request =
PktLine.encode("command=fetch") <>
PktLine.delim() <>
PktLine.encode("want #{sha}") <>
PktLine.encode("done") <>
PktLine.flush()
{_response, _state} = UploadPackV2.feed(state, request)
assert_receive {:telemetry, [:ex_git_objectstore, :protocol, :fetch, :start], _measurements,
%{mode: :full, wants: 1, haves: 0}}
assert_receive {:telemetry, [:ex_git_objectstore, :protocol, :fetch, :stop],
%{duration: duration}, stop_meta}
assert duration > 0
assert stop_meta.objects > 0
assert stop_meta.pack_bytes > 0
assert stop_meta.mode == :full
end
test "reports :shallow mode when deepen args are present" do
repo = RepoHelper.memory_repo("tel-shallow")
ExGitObjectstore.init(repo)
sha = make_commit(repo, "hi\n")
:ok = Ref.put(repo, "refs/heads/main", sha, nil)
{_advert, state} = UploadPackV2.init(repo)
request =
PktLine.encode("command=fetch") <>
PktLine.delim() <>
PktLine.encode("want #{sha}") <>
PktLine.encode("deepen 1") <>
PktLine.encode("done") <>
PktLine.flush()
{_response, _state} = UploadPackV2.feed(state, request)
assert_receive {:telemetry, [:ex_git_objectstore, :protocol, :fetch, :stop], _measurements,
%{mode: :shallow}}
end
end
describe "filter telemetry" do
test "emits :objects_in / :objects_out with spec kind" do
repo = RepoHelper.memory_repo("tel-filter")
ExGitObjectstore.init(repo)
sha = make_commit(repo, "filterable\n")
:ok = Ref.put(repo, "refs/heads/main", sha, nil)
{_advert, state} = UploadPackV2.init(repo)
request =
PktLine.encode("command=fetch") <>
PktLine.delim() <>
PktLine.encode("want #{sha}") <>
PktLine.encode("filter blob:none") <>
PktLine.encode("done") <>
PktLine.flush()
{_response, _state} = UploadPackV2.feed(state, request)
assert_receive {:telemetry, [:ex_git_objectstore, :pack, :filter], measurements, meta}
assert meta.spec_kind == :blob_none
assert measurements.objects_in >= 0
# blob:none always excludes blobs, so there's at least a
# non-zero drop — we don't assert an exact delta but we do
# assert objects_out <= objects_in.
assert measurements.objects_out <= measurements.objects_in
end
end
describe "atomic receive-pack telemetry" do
test "emits :start and :stop with outcome=:committed on success" do
repo = RepoHelper.memory_repo("tel-atomic-ok")
ExGitObjectstore.init(repo)
commit_sha = make_commit(repo, "content\n")
{_advert, state} = ReceivePack.init(repo)
state = %{
state
| 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.
{_report, _final} = ReceivePack.feed(state, empty_pack())
assert_receive {:telemetry, [:ex_git_objectstore, :protocol, :receive_pack, :atomic, :stop],
%{duration: duration},
%{outcome: :committed, commands: 2, validation_failures: 0}}
assert duration > 0
end
end
# --- helpers ---
defp make_commit(repo, content) do
blob = Blob.from_content(content)
{:ok, blob_sha} = Object.write(repo, blob)
tree = Tree.new([%{mode: "100644", name: "f.txt", sha: blob_sha}])
{:ok, tree_sha} = Object.write(repo, tree)
commit = %Commit{
tree: tree_sha,
parents: [],
author: "T <t@t.com> 1000000000 +0000",
committer: "T <t@t.com> 1000000000 +0000",
message: "c\n"
}
{: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