feat(stats): measure the download rate from received bytes, not verified pieces

lib/elixir_torrent/application.ex

@@ -13,6 +13,7 @@ defmodule ElixirTorrentApplication do
       Peer.UtPex.RecentCache,
       Peer.DialStats,
       Peer.Holepunch.Store,
+      Torrent.ReceivedBytes,
       Torrents,
       PeerDiscovery,
       Acceptor,

lib/elixir_torrent/peer/connection_manager.ex

@@ -478,13 +478,18 @@ defmodule Peer.ConnectionManager do
     end
   end
 
+  # `Torrent.Model` reports the download rate in bytes per millisecond, usually as
+  # a float. The guard used to accept only integers, so the real (float) readout
+  # never matched and every downloading torrent counted as "low speed" whatever it
+  # was doing: escalated discovery below 50 peers and the snub path at 12+ stayed
+  # on for torrents running at MB/s.
   defp low_download_speed?(hash) do
     case Torrent.get(hash, :speed) do
-      %{download: speed} when is_integer(speed) -> speed < @low_speed_bytes_per_sec
+      %{download: per_ms} when is_number(per_ms) -> per_ms * 1_000 < @low_speed_bytes_per_sec
       _ -> true
     end
   catch
-    _ -> true
+    :exit, _ -> true
   end
 
   @spec seeding?(Torrent.hash()) :: boolean()

lib/elixir_torrent/peer/controller/state.ex

@@ -1353,6 +1353,9 @@ defmodule Peer.Controller.State do
   @spec count_block(t(), Torrent.length()) :: t()
   defp count_block(%__MODULE__{} = state, length) do
     now = System.monotonic_time(:millisecond)
+    # Byte-granular source for the torrent's download readout (see
+    # `Torrent.ReceivedBytes`); `downloaded` only moves per verified piece.
+    :ok = Torrent.ReceivedBytes.add(state.hash, length)
 
     %__MODULE__{
       state

lib/elixir_torrent/torrent/model.ex

@@ -6,15 +6,13 @@ defmodule Torrent.Model do
   use GenServer
   use Via
 
-  alias Torrent.{Bitfield, Session}
+  alias Torrent.{Bitfield, ReceivedBytes, Session}
 
   require Logger
 
   @timeout_detect_the_speed 5 * 1_000
-  # Averaging window for the download readout, and the point past which a torrent
-  # that has not completed a single piece is called stopped — see `download_rate/1`.
-  @rate_window 60 * 1_000
-  @rate_window_max 10 * 60 * 1_000
+  # Sliding window for the download readout — see `download_rate/1`.
+  @rate_window 20 * 1_000
   # Mid-download resume checkpoint (BEP-adjacent): persist bitfield + counters to
   # `.term` so a restart loads Session.apply/2 and Resume runs :verify_saved on
   # only the pieces we claim — not a blind re-download from peers. Without this,
@@ -232,89 +230,49 @@ defmodule Torrent.Model do
   defp detected_the_speed(current, old),
     do: (current - old) / @timeout_detect_the_speed
 
-  # Download rate in the same Kb/s units, but averaged over a window long enough to
-  # contain several pieces instead of differenced over the 5 s tick.
+  # Download rate in the same units (bytes/ms ≈ KB/s), from the byte-granular
+  # `Torrent.ReceivedBytes` counter that every `Peer.Controller` bumps per accepted
+  # block, differenced over a sliding @rate_window of 5 s samples.
   #
-  # `downloaded` advances only when a whole piece completes and verifies, so a 5 s
-  # difference is quantized to piece size: at 55 KB/s with 1 MiB pieces one lands
-  # every ~19 s, so three ticks in four read exactly 0. That is what reported
-  # 0 B/s for torrents demonstrably progressing (#53b).
-  #
-  # Two narrower fixes were tried live and both failed, which is why the window is
-  # the shape it is:
-  #   * An EMA over the quantized samples. Any time constant short enough to track
-  #     a fast torrent still collapses between a slow torrent's pieces — observed
-  #     decaying to 1e-39 — and one long enough for the slow torrent is uselessly
-  #     laggy for the fast one.
-  #   * Timing each arrival against the previous one. A piece completing inside a
-  #     single tick makes the measured interval ~5 s, which is a real burst rate
-  #     (209 KB/s on a 1 MiB piece) but a bad basis for deciding the torrent has
-  #     stalled, so the readout alternated between the burst and 0.
-  #
-  # Averaging `delta / elapsed` over a fixed @rate_window sidesteps both: the window
-  # spans enough pieces that quantization averages out, and it is the same
-  # measurement used by hand when auditing this node (sum of `left` deltas over
-  # ≥100 s). While a window is still open the last published average is held rather
-  # than a burst rate, clamped by `piece_length / elapsed` — were the torrent still
-  # going that fast, the next piece would already have landed. A torrent that
-  # completes nothing for @rate_window_max is called stopped; that is generous on
-  # purpose, since a genuinely slow torrent can need minutes per piece.
-  defp download_rate(%Torrent{} = torrent) do
+  # It used to be derived from `downloaded`, which advances only when a whole piece
+  # completes and verifies, so every short window was quantized to piece size: at
+  # 64 KB/s with 1 MiB pieces the readout swung 12-157 KB/s, and a torrent slower
+  # than one piece per window read 0 while bytes were arriving (#53b). That number
+  # also gated `Peer.ConnectionManager`'s escalation and snubbing, so the error was
+  # not only cosmetic. Counting received bytes is what other clients show as the
+  # payload download rate, and the window only has to smooth TCP burstiness, not
+  # span pieces.
+  defp download_rate(%Torrent{left: 0} = torrent) do
+    Process.delete(progress_key(torrent))
+    0.0
+  end
+
+  defp download_rate(%Torrent{hash: hash} = torrent) do
     now = System.monotonic_time(:millisecond)
+    sample = {now, ReceivedBytes.total(hash)}
     key = progress_key(torrent)
-    {start_at, start_downloaded} = rate_window(key, now, torrent.downloaded)
-    elapsed = max(now - start_at, 1)
-    delta = torrent.downloaded - start_downloaded
-
-    case rate_verdict(torrent, delta, elapsed) do
-      {:publish, rate} ->
-        Process.put(key, {now, torrent.downloaded})
-        rate
-
-      :hold ->
-        torrent.speed.download
-
-      {:ceiling, ceiling} ->
-        min(torrent.speed.download, ceiling)
-    end
+    samples = window_samples([sample | Process.get(key, [])], now)
+    Process.put(key, samples)
+    rate_over(samples)
   end
 
-  defp rate_verdict(%Torrent{left: 0}, _delta, _elapsed), do: {:publish, 0.0}
-
-  defp rate_verdict(_torrent, delta, elapsed) when delta > 0 and elapsed >= @rate_window,
-    do: {:publish, delta / elapsed}
-
-  # Bytes have landed inside this window, so the torrent is demonstrably moving:
-  # hold the last published average until the window matures. Deliberately no
-  # ceiling here — progress is proof, and applying one anyway is what made the
-  # readout sawtooth from a true 100 KB/s down to 11 as the window aged, since the
-  # ceiling is `piece_length / elapsed` and `elapsed` grows all window long.
-  defp rate_verdict(_torrent, delta, _elapsed) when delta > 0, do: :hold
+  # Newest first. Keep every sample inside the window plus the first one older than
+  # it, so the span covers the whole window once the torrent has been up that long.
+  defp window_samples([newest | rest], now) do
+    {inside, outside} = Enum.split_while(rest, fn {at, _} -> now - at < @rate_window end)
+    [newest | inside] ++ Enum.take(outside, 1)
+  end
 
-  defp rate_verdict(_torrent, 0, elapsed) when elapsed >= @rate_window_max,
-    do: {:publish, 0.0}
+  defp rate_over([{now, total} | _] = samples) do
+    {oldest_at, oldest_total} = List.last(samples)
 
-  # Nothing at all has arrived yet. Now the ceiling is meaningful: were the torrent
-  # still running faster than this, the first piece of the window would have landed.
-  defp rate_verdict(torrent, 0, elapsed),
-    do: {:ceiling, do_piece_length(torrent) / elapsed}
+    if now > oldest_at,
+      do: max(total - oldest_total, 0) / (now - oldest_at),
+      else: 0.0
+  end
 
   defp progress_key(%Torrent{hash: hash}), do: {:speed_rate_window, hash}
 
-  # The first tick has to open the window, otherwise `elapsed` would be recomputed
-  # from `now` every tick and could never grow.
-  defp rate_window(key, now, downloaded) do
-    case Process.get(key) do
-      nil ->
-        window = {now, downloaded}
-        Process.put(key, window)
-        window
-
-      stored ->
-        stored
-    end
-  end
-
   defp message_for_next_detection(torrent) do
     message = {:detected_the_speed, torrent.downloaded, torrent.uploaded}
     Process.send_after(self(), message, @timeout_detect_the_speed)

lib/elixir_torrent/torrent/received_bytes.ex

@@ -0,0 +1,63 @@
+defmodule Torrent.ReceivedBytes do
+  @moduledoc """
+  Per-torrent count of piece payload bytes received from peers.
+
+  The download readout used to be derived from `Torrent.Model.downloaded`, which
+  advances only when a whole piece completes and verifies: at 64 KB/s with 1 MiB
+  pieces it moves once every ~16 s, so any short window reads 0 or a burst, and a
+  long one lags (PLAN #53b). Every `Peer.Controller` already knows the exact size
+  of each block it accepts; this table adds those sizes up per info-hash so the
+  Model can difference a byte-granular counter instead.
+
+  `:ets.update_counter/4` on a `write_concurrency` table is an atomic in-place
+  increment with no message passing, so the hot receive path pays one ETS write
+  per 16 KiB block (~130/s at 2 MB/s). The table is owned by this process, not by
+  a torrent, so it survives a Model or peer restart; a missing table (a test that
+  did not start the application) makes `add/2` a no-op.
+  """
+
+  use GenServer
+
+  @table :elixir_torrent_received_bytes
+
+  @spec start_link(keyword()) :: GenServer.on_start()
+  def start_link(opts \\ []), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+
+  @impl GenServer
+  def init(_opts) do
+    if :ets.info(@table) == :undefined do
+      :ets.new(@table, [:set, :public, :named_table, write_concurrency: true])
+    end
+
+    {:ok, nil}
+  end
+
+  @doc "Adds `bytes` received for `hash`."
+  @spec add(Torrent.hash(), non_neg_integer()) :: :ok
+  def add(hash, bytes) when is_binary(hash) and is_integer(bytes) and bytes >= 0 do
+    _ = :ets.update_counter(@table, hash, bytes, {hash, 0})
+    :ok
+  rescue
+    ArgumentError -> :ok
+  end
+
+  @doc "Total bytes received for `hash` since the counter was created."
+  @spec total(Torrent.hash()) :: non_neg_integer()
+  def total(hash) when is_binary(hash) do
+    case :ets.lookup(@table, hash) do
+      [{^hash, n}] -> n
+      [] -> 0
+    end
+  rescue
+    ArgumentError -> 0
+  end
+
+  @doc "Forgets the counter for `hash` (the torrent was removed)."
+  @spec delete(Torrent.hash()) :: :ok
+  def delete(hash) when is_binary(hash) do
+    _ = :ets.delete(@table, hash)
+    :ok
+  rescue
+    ArgumentError -> :ok
+  end
+end

lib/elixir_torrent/torrents.ex

@@ -124,7 +124,8 @@ defmodule Torrents do
          download_root <-
            if(delete_data?, do: Torrent.Removal.download_root(hash), else: File.cwd!()),
          :ok <- stop_pid(pid),
-         :ok <- Torrent.Session.delete(hash) do
+         :ok <- Torrent.Session.delete(hash),
+         :ok <- Torrent.ReceivedBytes.delete(hash) do
       maybe_delete_data(paths, download_root, delete_data?)
     end
   end

test/connection_manager_test.exs

@@ -364,15 +364,30 @@ defmodule Peer.ConnectionManagerTest do
       assert discovery_counts(announce_pid) == {1, 1}
     end
 
+    # Torrent.Model reports bytes per millisecond, as a float. The gate used to
+    # accept only integers, so a real readout always counted as "low speed".
     test "does not escalate when download speed clears the low threshold" do
+      fast_per_ms = (@low_speed_bytes_per_sec + 8_000) / 1_000
+
       {_hash, manager_pid, announce_pid} =
-        setup_escalation_scenario(31, download_speed: @low_speed_bytes_per_sec)
+        setup_escalation_scenario(31, download_speed: fast_per_ms)
 
       tick(manager_pid)
 
       assert discovery_counts(announce_pid) == {0, 0}
     end
 
+    test "escalates when a real (float, per-ms) readout is below the threshold" do
+      slow_per_ms = (@low_speed_bytes_per_sec - 8_000) / 1_000
+
+      {_hash, manager_pid, announce_pid} =
+        setup_escalation_scenario(31, download_speed: slow_per_ms)
+
+      tick(manager_pid)
+
+      assert discovery_counts(announce_pid) == {1, 1}
+    end
+
     test "does not escalate once connected reaches target even with dead throughput" do
       {_hash, manager_pid, announce_pid} = setup_escalation_scenario(50)