diff --git a/.gitignore b/.gitignore index bb58252..b8fb460 100644 --- a/.gitignore +++ b/.gitignore @@ -22,6 +22,10 @@ erl_crash.dump # Ignore package tarball (built via "mix hex.build"). elixir_torrent-*.tar +# Sobelow SARIF report (built via "mix sobelow --out"); CI regenerates it per run, +# a local CI-parity run would otherwise leave it in the tree. +sobelow.sarif + # Local session state, DHT node id persistence, and downloaded data (engine default layout). /.elixir_torrent/ diff --git a/CHANGELOG.md b/CHANGELOG.md index ae541c4..d1c9591 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,128 @@ # Changelog +## 0.6.6 - 2026-08-27 + +A swarm-health release. Under CGNAT, where a torrent runs on a handful of peers, +three separate mechanisms were removing peers we could not afford to lose or +parking them on work they were not allowed to do — torrents sat at 99% for hours +with unchoked idle peers and unclaimed blocks side by side. + +### Fixed + +- Peers are no longer banned for correctly answering a request we withdrew. + A cancel is not atomic: BEP 6 obliges the peer to answer every request exactly + once, so a block cancelled, choked or repinned away still arrives one RTT + later. Both the piece and the reject landed in the "not outstanding" branch, + which disconnected the peer and blacklisted its ID across every torrent — 46 + bans from `handle_piece` and 29 from `handle_reject` in five minutes, hitting + the Fast clients hardest precisely because they are the ones obliged to + reject. Withdrawn blocks are now remembered in a bounded per-block *count* + (one answer per request, so arrival order stops mattering) evicted by + generation rather than cleared wholesale. An answer matching neither set is + wasteful, not malicious: it ends the connection at 512 blocks without banning. +- Fast-extension messages sent without the extension advertised, and frames that + stall half-way, no longer ban the peer. 71 of 98 disconnects in one + three-minute window were mainstream qBittorrent and Transmission builds. + `have_all`/`have_none` now go through the normal bitfield handler — a seeder + recorded as having nothing is a peer we can never request from — and the + advisory messages are logged and ignored. A truncated frame is a congested + path, so the connection still drops, but the ID is not blacklisted. +- Peer bans expire. The blacklist was a `MapSet` that only grew, so one bad + frame excluded a peer for the whole session across every torrent: 352 IDs in + ten minutes while eight of nine torrents could not exceed three connections. + Bans now last 30 minutes with a cap and a background sweep, and each records + the rule that fired. +- The dial layer no longer writes off a candidate pool a CGNAT host cannot + refill. A failure row was cleared only by `mark_productive/3`, which asks + whether an endpoint was *useful* rather than whether it is *reachable* — a + leecher we keep choked never delivers bytes and so carried its failure history + forever; registration now clears it, inbound peers included. Retention was + also refreshed on every failure, so a row re-dialled once per 30 minutes never + aged out and reached fail counts above 100; escalation now counts failures + within a 10-minute streak window. Sticky blocks became last-resort under + `min_count` pressure (productive, then soft before sticky, then v6 before v4, + then fewest failures) instead of refusing resurrection absolutely — 1151 of + 1162 active blocks did, leaving a 50-endpoint request returning 3-8. +- Endgame now applies to pieces that were already in flight when the torrent + crossed the threshold. A worker read the mode once at `State.download/3`, so + exactly the pieces endgame exists for ran without redundancy: one 1 MiB piece + held a torrent at 99.939% for over an hour with all 27 remaining blocks in + flight to a single peer that had logged 297 request timeouts on it, while + three unchoked peers holding the piece had nothing they were allowed to ask + for. `:reconcile_pump` upgrades active workers level-triggered; the transition + is idempotent and one-way and re-queues in-flight subpieces so endgame *adds* + sources for a block rather than taking it from the peer already fetching it. +- A pin is released from an unchoked peer that delivers nothing. The staleness + test required `choke_me`, but a choked peer holds no requests at all — the + harmful case was the one it skipped, an unchoked peer sitting on a full + 64-request pipeline nobody else may touch. Two such peers re-requested their + 64 blocks 821 and 622 times in five minutes while two pieces with every block + unclaimed had no peer. Zero bytes now releases the pin either way, on a longer + threshold when unchoked (60 s, more than a block timeout) than when choked. +- A peer is no longer repinned off the piece it is fetching. "Drained" ignored + *who* had claimed the blocks, so a peer that had claimed the rest of its piece + made it look finished, was moved away, and the move cancelled the very + requests that drained it — oscillating at the 2 s tick, 17 wire requests per + block received. Draining now also requires no in-flight requests from that + peer. Measured over seven minutes live: requests per block 17:1 → 1.1:1, + re-request factor 5.66× → 1.08×, swarm 28 → 73 peers, 1.79 → 2.47 MB/s. +- Two endgame defects that parked torrents just short of completion: the + drained-pin probe branched on the *torrent's* mode rather than the worker's, + holding a pin for work that worker could never hand out, and + `endgame_preferred_index/2` destructured the peer key backwards + (`{_hash, peer_id}` against `Peer.make_key/2`'s `{id, hash}`), hashing the + torrent hash and so funnelling the entire swarm onto one index. +- Outside endgame, a peer may leave a piece whose blocks are all claimed. + `piece_has_waiting?/2` counts blocks in flight to *other* peers, which a + normal-mode worker can never hand out — 27 of 37 peers were pinned to 4 + claimed pieces while 8 pieces with 49-64 free blocks had no peer at all. +- A piece worker whose holders have all disconnected is released. The abort + check also required the swarm to be empty, so on any torrent with peers such a + worker held one of the `@max_parallel_pieces` slots forever — 7 of 12 slots + live, capping a torrent with 26 unchoked peers at 5 pieces in flight. +- A piece failing its SHA-1 check now blames the peer that supplied it. The + worker previously just re-requested every block and could pick the same peer + again; one torrent sat at 99.84% for hours re-downloading one index. Blocks + now remember their source, a peer stops being asked for an index it has + corrupted, and is dropped after `@max_hash_failures` *distinct* ruined pieces. +- Writes to a peer that has stopped reading are bounded. A TCP socket defaults + to `send_timeout: :infinity`, so a sender blocked inside `:prim_inet.send/4` + never returned and its mailbox only grew — one held 20863 messages. Accepted + and dialled sockets now take a 30 s send timeout and close on it. +- The upload delivery task no longer crashes when a peer cannot take a block. + Only `:noproc` was tolerated, so a peer shutting down mid-call produced one + crash report per in-flight block — 333 in fifteen minutes. BEP 3 permits + simply not answering a request, so this is now a cancellation with one debug + line; any other exit still crashes. +- One pending pump wake per torrent. Every trigger — the 2 s reconcile tick, + each peer handoff, every `requests_are_dealt` closure — started its own + self-rescheduling `{:next_piece}` chain, and they accumulated: 99 → ~2600 + discovery dial cycles per minute over twelve minutes with no change in swarm + size, until the tracker answered 403. +- A peer disconnect logs why it ended. `Peer.Endpoints` monitors a supervisor + with `auto_shutdown: :any_significant`, which exits with a bare `:shutdown` + whatever the child's reason was, so every disconnect read `reason=:shutdown` — + useless for the one question worth asking, whether the peer left or we dropped + it. A protocol error now also logs the rejected wire message, and the + piece-bounds check logs the block alongside the torrent's geometry. +- The background DHT metadata-lookup dedup table has a permanent owner. It was + created by whichever `Magnet.Fetcher` ran first and died with that torrent, so + every later background task crashed with `ArgumentError` on insert; a + supervised GenServer now owns it. + +### Performance + +- Dial batches overlap instead of serialising on their slowest endpoint. One + endpoint can hold a slot for the whole connect + handshake budget (measured + successes at 16 s, 42 s and 42 s, worst case near 55 s), and until it resolved + the manager would start nothing new for that torrent. That hurt worst where it + mattered most: a starved torrent with an all-IPv4 queue is capped to a + four-endpoint probe batch, so it made four attempts per minute against a + ~1% CGNAT success rate. Batches are now bounded by endpoints in flight (40) + and concurrent batches (3), with in-flight endpoints excluded from selection + so two batches cannot dial the same peer. + + ## 0.6.5 - 2026-08-18 ### Added diff --git a/PROTOCOL.md b/PROTOCOL.md index 7f41318..f8b835c 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -32,7 +32,7 @@ partial, and what is still missing — the same table the maintainer works from. | [BEP 15](https://www.bittorrent.org/beps/bep_0015.html) | UDP tracker protocol | **Full** | Connect, announce, scrape, error packets; 60 s connection_id cache; full 15×2ⁿ s long-announce ladder reconnects after expiry, while scrape and under-target fast-fail use intentionally shorter ladders; compact IPv4/IPv6 peers | | [BEP 16](https://www.bittorrent.org/beps/bep_0016.html) | Superseeding | **Full** | Automatically enters initial-seed mode when a live download completes with no confirmed remote seed: one rare fabricated `have` per peer, assignment rotation on propagation, hidden-piece rejection, and normal seeding restored for a complete remote bitfield or after restart | | [BEP 19](https://www.bittorrent.org/beps/bep_0019.html) | WebSeed — HTTP/FTP seeding (GetRight-style) | **Partial** | HTTP/HTTPS `Range` fetches for v1/hybrid torrents share the peer verify/write path, with corrupt mirrors disabled per session; FTP, GetRight gap scheduling, and pure-v2 mapping are not implemented; BEP 17 is not planned | -| [BEP 20](https://www.bittorrent.org/beps/bep_0020.html) | Peer ID conventions | **Full** | Version-derived prefix (`ET0-6-5` for package 0.6.5); one runtime-generated 20-byte identity per application instance | +| [BEP 20](https://www.bittorrent.org/beps/bep_0020.html) | Peer ID conventions | **Full** | Version-derived prefix (`ET0-6-6` for package 0.6.6); one runtime-generated 20-byte identity per application instance | | [BEP 23](https://www.bittorrent.org/beps/bep_0023.html) | Compact peer lists | **Full** | Compact IPv4 peers and dictionary-model IP literals; malformed values are ignored, while legacy dictionary hostnames are intentionally not DNS-resolved; combined with BEP 7 for `peers6` | | [BEP 24](https://www.bittorrent.org/beps/bep_0024.html) | Tracker returns external IP | **Partial** | Decodes `external ip` from HTTP responses; not used for listen-address selection | | [BEP 29](https://www.bittorrent.org/beps/bep_0029.html) | Micro Transport Protocol (uTP) | **Substantially Full** | SYN/STATE/DATA/FIN, LEDBAT, cumulative and selective ACK with SACK fast-loss recovery, owner-buffer-aware receive windows, type-preserving retransmission, symmetric FIN close, and dead zero-window probing; TCP-first dial with uTP fallback over DHT's shared UDP socket | diff --git a/README.md b/README.md index d1bcfbb..7f604af 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # ElixirTorrent -[![GitHub release](https://img.shields.io/badge/release-0.6.5-181717?logo=github)](https://github.com/daniboybye/ElixirTorrent/releases/tag/0.6.5) [![Changelog](https://img.shields.io/badge/changelog-blue)](https://hexdocs.pm/elixir_torrent/changelog.html) [![Hex.pm](https://img.shields.io/hexpm/v/elixir_torrent.svg)](https://hex.pm/packages/elixir_torrent/0.6.5) [![HexDocs](https://img.shields.io/badge/hexdocs-0.6.5-8E44AD)](https://hexdocs.pm/elixir_torrent/0.6.5) [![Hex.pm Downloads](https://img.shields.io/hexpm/dt/elixir_torrent.svg)](https://hex.pm/packages/elixir_torrent) [![License](https://img.shields.io/hexpm/l/elixir_torrent.svg)](https://github.com/daniboybye/ElixirTorrent/blob/master/LICENSE) +[![GitHub release](https://img.shields.io/badge/release-0.6.6-181717?logo=github)](https://github.com/daniboybye/ElixirTorrent/releases/tag/0.6.6) [![Changelog](https://img.shields.io/badge/changelog-blue)](https://hexdocs.pm/elixir_torrent/changelog.html) [![Hex.pm](https://img.shields.io/hexpm/v/elixir_torrent.svg)](https://hex.pm/packages/elixir_torrent/0.6.6) [![HexDocs](https://img.shields.io/badge/hexdocs-0.6.6-8E44AD)](https://hexdocs.pm/elixir_torrent/0.6.6) [![Hex.pm Downloads](https://img.shields.io/hexpm/dt/elixir_torrent.svg)](https://hex.pm/packages/elixir_torrent) [![License](https://img.shields.io/hexpm/l/elixir_torrent.svg)](https://github.com/daniboybye/ElixirTorrent/blob/master/LICENSE) [![build](https://img.shields.io/github/actions/workflow/status/daniboybye/ElixirTorrent/build-and-publish.yml?branch=master&label=build&logo=github)](https://github.com/daniboybye/ElixirTorrent/actions/workflows/build-and-publish.yml) [![codecov](https://codecov.io/gh/daniboybye/ElixirTorrent/branch/master/graph/badge.svg)](https://codecov.io/gh/daniboybye/ElixirTorrent) [![BEPs](https://img.shields.io/badge/BEPs-23%20implemented-E8A33D)](PROTOCOL.md) [![Last commit](https://img.shields.io/github/last-commit/daniboybye/ElixirTorrent/master)](https://github.com/daniboybye/ElixirTorrent/commits/master) @@ -56,7 +56,7 @@ Full per-BEP status, including the known gaps: **[PROTOCOL.md](PROTOCOL.md)**. ```elixir def deps do [ - {:elixir_torrent, "~> 0.6.5"} + {:elixir_torrent, "~> 0.6.6"} ] end ``` @@ -173,7 +173,7 @@ Full reference: [`hexdocs.pm/elixir_torrent/ElixirTorrent.html`](https://hexdocs | `stop_all_and_serialize/0` | Graceful stop + persist for every torrent | | `remove/2` | Stop and drop from session; optional `delete_data: true` | | `get/2` | Low-level field access (prefer `stats/2`) | -| `version/0` | Version-derived client peer ID prefix (`ET0-6-5`, BEP 20) | +| `version/0` | Version-derived client peer ID prefix (`ET0-6-6`, BEP 20) | ## ElixirTorrent Web (desktop app) diff --git a/lib/elixir_torrent/acceptor.ex b/lib/elixir_torrent/acceptor.ex index 76353f6..7f76cb0 100644 --- a/lib/elixir_torrent/acceptor.ex +++ b/lib/elixir_torrent/acceptor.ex @@ -23,8 +23,28 @@ defmodule Acceptor do defdelegate handshakes(peers, hash), to: Handshakes - @tcp_performance [nodelay: true, recbuf: 262_144, sndbuf: 262_144] - @tcp_connect_fallback [nodelay: true] + # `:gen_tcp.send/2` defaults to `send_timeout: :infinity`, so a peer that stops + # reading blocks its `Peer.Sender` inside `:prim_inet.send/4` forever while + # wire casts keep arriving — an unbounded mailbox on a remote-controlled + # trigger. Live: one sender sat in that call with a 20 863-message mailbox and + # 3 MB of heap, still climbing. With a bound the write fails instead, and + # `Peer.Sender.do_send/2`'s existing `{:error, _} -> stop` path tears the peer + # down. `send_timeout_close` drops the socket too: a timed-out write may be + # partial, so the wire stream is no longer trustworthy. 30s is far beyond any + # healthy write and well under the 100s peer inactivity timeout. + @send_timeout_ms 30_000 + @tcp_performance [ + nodelay: true, + recbuf: 262_144, + sndbuf: 262_144, + send_timeout: @send_timeout_ms, + send_timeout_close: true + ] + @tcp_connect_fallback [ + nodelay: true, + send_timeout: @send_timeout_ms, + send_timeout_close: true + ] @spec socket_options() :: list() def socket_options, do: [:binary, active: false, reuseaddr: true] diff --git a/lib/elixir_torrent/acceptor/blacklist.ex b/lib/elixir_torrent/acceptor/blacklist.ex index 58e99b5..911043b 100644 --- a/lib/elixir_torrent/acceptor/blacklist.ex +++ b/lib/elixir_torrent/acceptor/blacklist.ex @@ -1,25 +1,77 @@ defmodule Acceptor.BlackList do @moduledoc """ - GenServer holding peer IDs rejected after failed handshakes (BEP 3 peer churn guard). + Peer IDs to refuse for a while after they broke the wire protocol (BEP 3 peer + churn guard). + + Entries expire. This used to be a `MapSet` that only grew, so one bad frame + excluded a peer for the rest of the session, across every torrent — and on a + CGNAT host, where a torrent runs on 1-8 peers, that is the swarm. A live + ten-minute debug run banned 352 peer IDs, mostly mainstream clients, while + eight of nine torrents could not get past three connections. + + A ban is a hedge against a peer that will misbehave again, and re-testing it + costs one connection, so it does not need to outlive the failure by much. """ use GenServer, start: {GenServer, :start_link, [__MODULE__, nil, [name: __MODULE__]]} + @ttl_ms 30 * 60 * 1_000 + @sweep_ms 60_000 + # Bounds memory on a long session. Well past what a healthy run reaches, so + # hitting it means something is banning indiscriminately. + @max_entries 2_000 + @spec put(Peer.id()) :: :ok def put(peer_id), do: GenServer.cast(__MODULE__, peer_id) @spec member?(Peer.id()) :: boolean() def member?(peer_id), do: GenServer.call(__MODULE__, peer_id) - @spec init(term()) :: {:ok, MapSet.t(Peer.id())} - def init(_), do: {:ok, MapSet.new()} + @spec init(term()) :: {:ok, %{optional(Peer.id()) => integer()}} + def init(_) do + schedule_sweep() + {:ok, %{}} + end + + @spec handle_call(Peer.id(), GenServer.from(), %{optional(Peer.id()) => integer()}) :: + {:reply, boolean(), %{optional(Peer.id()) => integer()}} + def handle_call(peer_id, _, state) do + case Map.fetch(state, peer_id) do + {:ok, expires_at} -> + if System.monotonic_time(:millisecond) < expires_at do + {:reply, true, state} + else + {:reply, false, Map.delete(state, peer_id)} + end + + :error -> + {:reply, false, state} + end + end + + @spec handle_cast(Peer.id(), %{optional(Peer.id()) => integer()}) :: + {:noreply, %{optional(Peer.id()) => integer()}} + def handle_cast(peer_id, state) do + state + |> Map.put(peer_id, System.monotonic_time(:millisecond) + @ttl_ms) + |> cap() + |> then(&{:noreply, &1}) + end + + @spec handle_info(:sweep, %{optional(Peer.id()) => integer()}) :: + {:noreply, %{optional(Peer.id()) => integer()}} + def handle_info(:sweep, state) do + now = System.monotonic_time(:millisecond) + schedule_sweep() + {:noreply, Map.reject(state, fn {_id, expires_at} -> expires_at <= now end)} + end + + defp cap(state) when map_size(state) <= @max_entries, do: state - @spec handle_call(Peer.id(), GenServer.from(), MapSet.t(Peer.id())) :: - {:reply, boolean(), MapSet.t(Peer.id())} - def handle_call(peer_id, _, state), - do: {:reply, MapSet.member?(state, peer_id), state} + defp cap(state) do + {oldest, _} = Enum.min_by(state, fn {_id, expires_at} -> expires_at end) + Map.delete(state, oldest) + end - @spec handle_cast(Peer.id(), MapSet.t(Peer.id())) :: {:noreply, MapSet.t(Peer.id())} - def handle_cast(peer_id, state), - do: {:noreply, MapSet.put(state, peer_id)} + defp schedule_sweep, do: Process.send_after(self(), :sweep, @sweep_ms) end diff --git a/lib/elixir_torrent/application.ex b/lib/elixir_torrent/application.ex index 112fc43..c52f0f4 100644 --- a/lib/elixir_torrent/application.ex +++ b/lib/elixir_torrent/application.ex @@ -21,6 +21,7 @@ defmodule ElixirTorrentApplication do NAT.PortMapper, Magnet.Fetcher.Supervisor, Magnet.Fetcher.ConnectionLimit, + Magnet.Fetcher.DhtBackgroundStore, Magnet.Bootstrap.Supervisor ] |> Supervisor.start_link(strategy: :one_for_one) diff --git a/lib/elixir_torrent/magnet/fetcher.ex b/lib/elixir_torrent/magnet/fetcher.ex index e7a548b..1d5d15f 100644 --- a/lib/elixir_torrent/magnet/fetcher.ex +++ b/lib/elixir_torrent/magnet/fetcher.ex @@ -612,8 +612,13 @@ defmodule Magnet.Fetcher do Logger.debug("[magnet_fetch] dht_background_peers hash=#{hash_hex} count=#{length(peers)}") end + # This task detaches from the round that spawned it and sleeps for seconds, + # so the table can legitimately be gone by now (application shutdown). + ensure_dht_bg_table() :ets.insert(@dht_bg_table, {hash, {:done, peers}}) :ok + rescue + ArgumentError -> :ok end @doc false diff --git a/lib/elixir_torrent/magnet/fetcher/dht_background_store.ex b/lib/elixir_torrent/magnet/fetcher/dht_background_store.ex new file mode 100644 index 0000000..1a1b3e9 --- /dev/null +++ b/lib/elixir_torrent/magnet/fetcher/dht_background_store.ex @@ -0,0 +1,33 @@ +defmodule Magnet.Fetcher.DhtBackgroundStore do + @moduledoc """ + Permanent owner of the background DHT lookup table. + + ETS tables die with the process that created them. The deep `get_peers` retry + runs in a detached task that outlives the fetch round which started it, so + letting whichever caller happened to touch the table first own it meant the + task's result insert crashed with `ArgumentError` as soon as that caller was + gone — losing the peers it had just spent ~13s finding. + """ + + use GenServer + + @table :magnet_fetcher_dht_background + + @spec table() :: atom() + def table, do: @table + + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts \\ []) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + @impl GenServer + @spec init(keyword()) :: {:ok, nil} + def init(_opts) do + if :ets.info(@table) == :undefined do + :ets.new(@table, [:named_table, :public, read_concurrency: true]) + end + + {:ok, nil} + end +end diff --git a/lib/elixir_torrent/peer/connection_manager.ex b/lib/elixir_torrent/peer/connection_manager.ex index 8c2fe46..0be23d0 100644 --- a/lib/elixir_torrent/peer/connection_manager.ex +++ b/lib/elixir_torrent/peer/connection_manager.ex @@ -22,6 +22,15 @@ defmodule Peer.ConnectionManager do @swarm_cap 60 @default_batch 40 @escalated_batch 50 + # A dial batch resolves only when its slowest endpoint does, and one endpoint + # can hold a slot for the full connect + handshake budget (~55s). Blocking the + # next batch on that made acquisition crawl exactly where it hurts: a starved + # torrent whose queue is all IPv4 gets a 4-endpoint probe batch, so it managed + # four dials per minute against a ~1% CGNAT success rate. Overlapping batches + # decouple the rate from the slowest peer; the caps below keep the socket cost + # bounded (a batch already runs at most 20 concurrent connects internally). + @max_in_flight_dials 40 + @max_dial_batches 3 @normal_interval_ms 3_000 @escalated_interval_ms 1_000 @low_speed_bytes_per_sec 32_768 @@ -118,7 +127,19 @@ defmodule Peer.ConnectionManager do def init(hash) do send_after(self(), :tick, @normal_interval_ms) # nil = never snubbed; monotonic ms can be negative so 0 is not a safe sentinel. - {:ok, %{hash: hash, queue: %{}, dialing?: false, dial_task: nil, last_snub_ms: nil}} + # `dialing?` is a hard stop on starting anything new; `in_flight`/`batches` + # are the running cost the caps are enforced against. + {:ok, + %{ + hash: hash, + queue: %{}, + dialing?: false, + dial_task: nil, + in_flight: MapSet.new(), + batches: 0, + dial_tasks: [], + last_snub_ms: nil + }} end @impl GenServer @@ -172,28 +193,46 @@ defmodule Peer.ConnectionManager do @impl GenServer def handle_info({:dial_done, selected_keys, results}, %{hash: hash} = state) do - {_ok, _failures, failed_peers} = results - failed_keys = MapSet.new(Enum.map(failed_peers, fn {p, _} -> {p.ip, p.port} end)) - - keys_to_drop = Enum.reject(selected_keys, &MapSet.member?(failed_keys, &1)) - queue = Map.drop(state.queue, keys_to_drop) connected = Swarm.count(hash) record_failures(hash, results, connected) - state = %{state | queue: queue, dialing?: false, dial_task: nil} + state = release_dial_batch(state, selected_keys, results) + maybe_replenish_discovery(state, connected) maybe_dial(state, connected) end + # An endpoint that failed stays queued so backoff ordering decides when to retry + # it; the rest resolved and leave the queue with the batch. Freeing the slots is + # what lets the next batch start, so it happens whatever the results were. + defp release_dial_batch(state, selected_keys, {_ok, _failures, failed_peers}) do + failed_keys = MapSet.new(failed_peers, fn {p, _} -> {p.ip, p.port} end) + keys_to_drop = Enum.reject(selected_keys, &MapSet.member?(failed_keys, &1)) + batches = max(state.batches - 1, 0) + + %{ + state + | queue: Map.drop(state.queue, keys_to_drop), + in_flight: MapSet.difference(state.in_flight, MapSet.new(selected_keys)), + batches: batches, + dial_tasks: Enum.filter(state.dial_tasks, &Process.alive?/1), + dialing?: false, + dial_task: if(batches == 0, do: nil, else: state.dial_task) + } + end + @impl GenServer - def terminate(_reason, %{dial_task: pid}) when is_pid(pid) do - if Process.alive?(pid), do: Process.exit(pid, :shutdown) + def terminate(_reason, state) do + for pid <- [state[:dial_task] | Map.get(state, :dial_tasks, [])], + is_pid(pid), + Process.alive?(pid) do + Process.exit(pid, :shutdown) + end + :ok end - def terminate(_reason, _state), do: :ok - defp handle_tick(state, hash, connected) do cond do connected >= @swarm_cap and low_download_speed?(hash) and downloading?(hash) -> @@ -265,7 +304,13 @@ defmodule Peer.ConnectionManager do defp maybe_dial(state, connected) when connected >= @swarm_cap, do: {:noreply, state} defp maybe_dial(%{hash: hash} = state, connected) do - dial_batch(state, batch_size(hash, connected)) + headroom = @max_in_flight_dials - MapSet.size(state.in_flight) + + if headroom <= 0 or state.batches >= @max_dial_batches do + {:noreply, state} + else + dial_batch(state, min(batch_size(hash, connected), headroom)) + end end defp maybe_dial_or_noreply(state, connected) do @@ -277,29 +322,49 @@ defmodule Peer.ConnectionManager do end defp dial_batch(%{hash: hash, queue: queue} = state, batch) do - peers = - hash - |> prioritize_dial_queue(DialQueue.peers(queue)) - |> Handshakes.select_peers_to_dial(hash, batch) + case select_dial_peers(state, hash, batch) do + [] -> + if map_size(queue) == 0 do + PeerDiscovery.Announce.replenish_candidates(hash) + end - if peers == [] do - if map_size(queue) == 0 do - PeerDiscovery.Announce.replenish_candidates(hash) - end + {:noreply, state} - {:noreply, state} - else - selected_keys = Enum.map(peers, fn p -> {p.ip, p.port} end) - parent = self() + peers -> + {:noreply, launch_dial_batch(state, hash, peers)} + end + end - {:ok, dial_task} = - Task.start(fn -> - results = Handshakes.dial_peers(peers, hash) - send(parent, {:dial_done, selected_keys, results}) - end) + defp select_dial_peers(%{queue: queue, in_flight: in_flight}, hash, batch) do + hash + |> prioritize_dial_queue(DialQueue.peers(queue)) + # An endpoint stays in the queue until its dial resolves, so overlapping + # batches would otherwise pick the same one twice. + |> Enum.reject(&MapSet.member?(in_flight, {&1.ip, &1.port})) + |> Handshakes.select_peers_to_dial(hash, batch) + end - {:noreply, %{state | dialing?: true, dial_task: dial_task}} - end + defp launch_dial_batch(state, hash, peers) do + selected_keys = Enum.map(peers, fn p -> {p.ip, p.port} end) + parent = self() + + {:ok, dial_task} = + Task.start(fn -> + results = Handshakes.dial_peers(peers, hash) + send(parent, {:dial_done, selected_keys, results}) + end) + + in_flight = MapSet.union(state.in_flight, MapSet.new(selected_keys)) + batches = state.batches + 1 + + %{ + state + | in_flight: in_flight, + batches: batches, + dial_tasks: [dial_task | state.dial_tasks], + dialing?: MapSet.size(in_flight) >= @max_in_flight_dials or batches >= @max_dial_batches, + dial_task: dial_task + } end defp record_failures(hash, {ok_count, failures, failed_peers}, connected) do diff --git a/lib/elixir_torrent/peer/controller.ex b/lib/elixir_torrent/peer/controller.ex index e0f115d..a4f3249 100644 --- a/lib/elixir_torrent/peer/controller.ex +++ b/lib/elixir_torrent/peer/controller.ex @@ -106,6 +106,15 @@ defmodule Peer.Controller do :exit, _ -> :error end + @doc """ + Reports that a piece this peer supplied on its own failed its hash check. + + Repeated failures disconnect the peer — see `Peer.Controller.State.hash_check_failed/2`. + """ + @spec hash_check_failed(Peer.key(), Torrent.index()) :: :ok + def hash_check_failed(key, index), + do: GenServer.cast(via(key), {:hash_check_failed, [index]}) + @spec handle_choke(Peer.key()) :: :ok def handle_choke(key), do: GenServer.cast(via(key), {:handle_choke, []}) @@ -190,6 +199,12 @@ defmodule Peer.Controller do Downloads.response(hash, index, key_to_id(key), begin, block) GenServer.cast(via(key), {:handle_piece, [index, begin, byte_size(block)]}) else + require Logger + + Logger.debug( + "[peer_wire] peer=#{Peer.log_id(key_to_id(key))} hash=#{Torrent.hex_encoded_hash(hash)} rejected=piece_bounds index=#{index} begin=#{begin} bytes=#{byte_size(block)} pieces=#{inspect(Torrent.get(hash, :pieces_count))} piece_len=#{inspect(Torrent.Model.piece_length(hash, index))}" + ) + GenServer.stop(via(key), {:shutdown, :protocol_error}) end end @@ -349,14 +364,23 @@ defmodule Peer.Controller do end @spec terminate(term(), State.t()) :: :ok - def terminate({:shutdown, :protocol_error}, state) do + def terminate({:shutdown, :protocol_error} = reason, state) do + note_disconnect_reason(state, reason) State.notify_hash_request_disconnect(state, :protocol_error) Torrent.Superseed.release(state.hash, state.id) Torrent.PiecesStatistic.remove_peer(state.hash, state.bitfield, state.pieces_count) + + require Logger + + Logger.debug( + "[peer_wire] banned peer=#{Peer.log_id(state.id)} hash=#{Torrent.hex_encoded_hash(state.hash)} cause=controller_protocol_error" + ) + Acceptor.malicious_peer(state.id) end def terminate(reason, %State{} = state) do + note_disconnect_reason(state, reason) State.record_pex_recent_disconnect(state, reason) State.notify_hash_request_disconnect(state, reason) # :shutdown / {:shutdown,_} are normal OTP stop paths (app teardown, swarm @@ -377,6 +401,20 @@ defmodule Peer.Controller do :ok end + # Our supervisor is auto_shutdown, so it exits with a bare `:shutdown` and + # Peer.Endpoints — which monitors it, not us — cannot see this reason unless we + # hand it over first. terminate/2 runs before the supervisor exits, so the note + # is in place by the time the :DOWN lands. + @spec note_disconnect_reason(State.t(), term()) :: :ok + defp note_disconnect_reason(%State{socket: nil}, _reason), do: :ok + + defp note_disconnect_reason(%State{} = state, reason) do + case Peer.Transport.safe_peername(state.socket) do + {:ok, {ip, port}} -> Peer.Endpoints.note_disconnect_reason(state.hash, ip, port, reason) + _ -> :ok + end + end + @doc false @spec quiet_disconnect_reason?(term()) :: boolean() def quiet_disconnect_reason?(:normal), do: true @@ -781,24 +819,40 @@ defmodule Peer.Controller do {:noreply, state} end - def handle_cast({message, _}, %State{hash: hash, fast_extension: nil} = state) - when message in [ - :handle_suggest_piece, - :handle_have_all, - :handle_have_none, - :handle_allowed_fast, - :handle_reject - ] do - if Magnet.Bootstrap.active?(hash) do - {:noreply, state} - else - {:stop, {:shutdown, :protocol_error}, state} - end + # BEP 6 sends these only once both sides advertised the Fast extension, and we + # do advertise it, so a peer sending one without it is at odds with the spec. + # Tearing the connection down was still the wrong answer, because terminate/2 + # also blacklists the peer ID for every torrent: 71 of 98 disconnects in one + # three-minute window, all of them qBittorrent and Transmission builds that do + # support Fast, on a host whose torrents run on 1-8 peers. + # + # `have all` / `have none` are self-describing — they carry what a full or empty + # bitfield carries — so they fall through to the normal handler. What is left is + # advisory (`suggest piece`, `allowed fast`) or self-healing (`reject`, which the + # request timeout already covers), so ignoring it costs nothing. + def handle_cast({message, _}, %State{fast_extension: nil} = state) + when message in [:handle_suggest_piece, :handle_allowed_fast, :handle_reject] do + require Logger + + Logger.debug( + "[peer_wire] peer=#{Peer.log_id(state.id)} hash=#{Torrent.hex_encoded_hash(state.hash)} ignored=#{message} reason=fast_not_negotiated" + ) + + {:noreply, state} end def handle_cast({fun, args}, state) do case apply(State, fun, [state | args]) do {:error, reason, state} -> + # Which wire message reached this verdict, not just that something did: + # a :protocol_error here ends the connection and blacklists the peer, and + # the reason alone is shared by a dozen rules. + require Logger + + Logger.debug( + "[peer_wire] peer=#{Peer.log_id(state.id)} hash=#{Torrent.hex_encoded_hash(state.hash)} rejected=#{fun} reason=#{inspect(reason)}" + ) + {:stop, {:shutdown, reason}, state} state -> diff --git a/lib/elixir_torrent/peer/controller/state.ex b/lib/elixir_torrent/peer/controller/state.ex index 480939e..cd0e416 100644 --- a/lib/elixir_torrent/peer/controller/state.ex +++ b/lib/elixir_torrent/peer/controller/state.ex @@ -36,6 +36,26 @@ defmodule Peer.Controller.State do ut_metadata_requests: %{window_started_at: nil, count: 0}, hash_requests: %{}, requests: MapSet.new(), + # Blocks we asked for and then withdrew (wire cancel, repin, or a choke that + # drops the peer's queue). A cancel is not atomic: the block may already be + # on the wire, and BEP 6 peers must answer every request exactly once, so a + # withdrawn block legitimately comes back as a piece or a reject one RTT + # later. Kept so those answers are recognised instead of read as a protocol + # violation. Bounded — see put_withdrawn/4. + # + # A count per block, not a set: the same block is often requested, withdrawn + # and requested again, so several answers can be owed for it at once, and + # BEP 6's exactly-one-answer rule makes the count exact. It also makes the + # bookkeeping independent of which answer arrives first. + withdrawn: %{}, + # Previous generation of `withdrawn`. Eviction has to drop the oldest + # entries, and a map has no order to drop by; retiring a full generation + # keeps at least @max_withdrawn recent withdrawals alive at all times. + withdrawn_prev: %{}, + # Blocks that matched neither `requests` nor `withdrawn`. Never fatal on its + # own (our window is finite), but a peer pushing data we never asked for is + # spending our bandwidth, so it is capped. + unsolicited_blocks: 0, # Inbound block requests accepted for asynchronous disk reads but not yet # sent to the peer. BEP 6 requires a choke to explicitly reject every # queued request outside the peer-specific allowed-fast set. @@ -68,6 +88,9 @@ defmodule Peer.Controller.State do # piece still reflected in `downloaded_bytes`. pin_downloaded_bytes: 0, superseed_piece: nil, + # Piece indices this peer supplied single-handedly that then failed their + # hash check. We stop asking this peer for them. See `hash_check_failed/2`. + hash_failures: MapSet.new(), bitfield: nil, interested: false, choke: true, @@ -104,6 +127,9 @@ defmodule Peer.Controller.State do ut_metadata_requests: %{window_started_at: integer() | nil, count: non_neg_integer()}, hash_requests: %{Peer.HashTransfer.ref() => map()}, requests: MapSet.t(subpiece()), + withdrawn: %{subpiece() => pos_integer()}, + withdrawn_prev: %{subpiece() => pos_integer()}, + unsolicited_blocks: non_neg_integer(), upload_requests: MapSet.t(subpiece()), pending_requests: non_neg_integer(), rank: non_neg_integer(), @@ -113,6 +139,7 @@ defmodule Peer.Controller.State do pinned_at: non_neg_integer(), pin_downloaded_bytes: non_neg_integer(), superseed_piece: Torrent.index() | :all | nil, + hash_failures: MapSet.t(Torrent.index()), bitfield: bitfield(), interested: boolean(), choke: boolean(), @@ -127,6 +154,12 @@ defmodule Peer.Controller.State do # far below what mainstream clients accept (they advertise reqq 250-500); when # a peer advertises a smaller reqq we honor it instead of overflowing its queue. @max_unanswered_requests 64 + # A withdrawn block stays recognisable for as long as it can plausibly still be + # in flight: at most one full pipeline per withdrawal, and a peer can be + # re-pinned before the previous round's answers land, so allow a few rounds. + @max_withdrawn 4 * @max_unanswered_requests + # Roughly 8 MiB of unrequested 16 KiB blocks before we give up on the peer. + @max_unsolicited_blocks 512 @request_pipeline_depth 64 @max_pending_hash_requests 8 # One metadata response can carry 16 KiB. 128 requests/s still permits about @@ -139,10 +172,57 @@ defmodule Peer.Controller.State do # the first active index (BEP-3 endgame needs multi-source redundancy). @stale_pin_ms 20_000 @stale_pin_ms_endgame 15_000 + # Same idea for a peer that is *not* choking us and still delivers nothing. + # That case is the damaging one: a choked peer holds no requests (handle_choke + # clears them), while an unchoked one keeps a full pipeline, so every block it + # sits on is unavailable to anybody else and is merely re-timed-out every + # @timeout_request. Live, two such peers held one whole piece each — 61 and 64 + # blocks, 0 and 48 KiB delivered — while two pieces with all 64 blocks free had + # no peer at all, and the torrent stopped dead at 99%. The threshold is longer + # than the choked one and longer than a block timeout, so a merely slow peer + # gets a full request cycle to prove itself first; this is the same 60 s idea + # other clients call snubbing. + @snubbed_pin_ms 60_000 + # How many *distinct* pieces this peer may supply single-handedly that then + # fail their SHA-1 before we drop the connection. A peer with one or two bad + # pieces on disk is otherwise perfectly good — a live run had one serve 99.84% + # of a 1.3 GB torrent correctly — so the per-index skip below carries the fix + # and disconnecting is reserved for a peer whose whole copy looks wrong. + @max_hash_failures 3 @spec key(t()) :: Peer.key() def key(state), do: make_key(state.hash, state.id) + @doc false + @spec max_hash_failures() :: pos_integer() + def max_hash_failures, do: @max_hash_failures + + @doc """ + Records a piece this peer supplied alone that failed its hash check. + + BEP 3 verifies pieces, not blocks, so a peer serving corrupt data is only + detectable after a whole piece is assembled. Nothing else in the download path + remembers where those bytes came from, so without this the piece picker hands + the same index straight back to the same peer and the torrent re-downloads it + forever. The index is remembered and never requested from this peer again; + only a peer that ruins `@max_hash_failures` different pieces is disconnected. + """ + @spec hash_check_failed(t(), Torrent.index()) :: t() | {:error, :corrupt_pieces, t()} + def hash_check_failed(%__MODULE__{} = state, index) do + failures = MapSet.put(state.hash_failures, index) + state = %__MODULE__{state | hash_failures: failures} + + Logger.warning( + "[peer_download] peer=#{Peer.log_id(state.id)} hash=#{Torrent.hex_encoded_hash(state.hash)} corrupt_piece index=#{index} bad_pieces=#{MapSet.size(failures)}/#{@max_hash_failures}" + ) + + if MapSet.size(failures) >= @max_hash_failures do + {:error, :corrupt_pieces, state} + else + state + end + end + @doc false @spec ut_metadata_request_limit() :: pos_integer() def ut_metadata_request_limit, do: @ut_metadata_request_limit @@ -703,12 +783,12 @@ defmodule Peer.Controller.State do @spec cancel(t(), Torrent.index(), Torrent.begin(), Torrent.length()) :: t() def cancel(state, index, begin, length) do - if member_request?(state, index, begin, length) do - Sender.cancel(key(state), index, begin, length) - end + member? = member_request?(state, index, begin, length) + if member?, do: Sender.cancel(key(state), index, begin, length) state |> delete_request(index, begin, length) + |> then(&if member?, do: put_withdrawn(&1, index, begin, length), else: &1) |> make_request() end @@ -837,8 +917,11 @@ defmodule Peer.Controller.State do # Peer choked us → they will drop any queued requests. Reset both the # in-flight set and the pending-ack counter so we can re-fill the - # pipeline from zero on the next unchoke. + # pipeline from zero on the next unchoke. A Fast peer owes us a reject for + # each dropped request (BEP 6), and a block already on the wire still + # arrives, so the set moves to `withdrawn` rather than vanishing. %__MODULE__{state | choke_me: true, requests: MapSet.new(), pending_requests: 0} + |> withdraw_all(state.requests) end @spec handle_unchoke(t()) :: t() @@ -1172,21 +1255,36 @@ defmodule Peer.Controller.State do @spec handle_piece(t(), Torrent.index(), Torrent.begin(), Torrent.length()) :: t() | {:error, :protocol_error, t()} def handle_piece(state, index, begin, length) do - if member_request?(state, index, begin, length) do - now = System.monotonic_time(:millisecond) + case classify_answer(state, index, begin, length) do + :requested -> + count_block(state, length) + |> delete_request(index, begin, length) + |> make_request() - state - |> Map.update!(:rank, &(&1 + length)) - |> Map.update!(:downloaded_bytes, &(&1 + length)) - |> Map.update!(:pin_downloaded_bytes, &(&1 + length)) - |> Map.put(:last_block_at, now) - |> delete_request(index, begin, length) - |> make_request() - else - {:error, :protocol_error, state} + :withdrawn -> + # The block was already on the wire when we cancelled. The payload is + # stored either way (the controller hands it to the piece worker before + # this cast), so count it and keep the peer. + count_block(state, length) + |> drop_withdrawn(index, begin, length) + |> make_request() + + :unsolicited -> + note_unsolicited(state, "piece index=#{index} begin=#{begin} len=#{length}") end end + @spec count_block(t(), Torrent.length()) :: t() + defp count_block(%__MODULE__{} = state, length) do + %__MODULE__{ + state + | rank: state.rank + length, + downloaded_bytes: state.downloaded_bytes + length, + pin_downloaded_bytes: state.pin_downloaded_bytes + length, + last_block_at: System.monotonic_time(:millisecond) + } + end + # DHT (BEP 5 § BitTorrent Protocol Extension) @spec handle_port(t(), non_neg_integer()) :: t() def handle_port(%__MODULE__{hash: hash} = state, dht_port) @@ -1291,14 +1389,23 @@ defmodule Peer.Controller.State do end defp do_handle_reject(state, index, begin, length) do - if member_request?(state, index, begin, length) do - Downloads.reject(state.hash, index, state.id, begin, length) + case classify_answer(state, index, begin, length) do + :requested -> + Downloads.reject(state.hash, index, state.id, begin, length) - state - |> delete_request(index, begin, length) - |> make_request() - else - {:error, :protocol_error, state} + state + |> delete_request(index, begin, length) + |> make_request() + + # BEP 6 obliges the peer to reject what it drops, so a cancel or a choke + # earns exactly this message back. The block was already handed back to + # its piece worker when we withdrew it; nothing left to do but keep the + # peer. + :withdrawn -> + drop_withdrawn(state, index, begin, length) + + :unsolicited -> + note_unsolicited(state, "reject index=#{index} begin=#{begin} len=#{length}") end end @@ -1581,16 +1688,19 @@ defmodule Peer.Controller.State do defp do_make_request(state), do: state - @spec download_request_skip_reason(t(), Torrent.index()) :: :queue_full | :choked | nil + @spec download_request_skip_reason(t(), Torrent.index()) :: + :queue_full | :choked | :corrupt_source | nil defp download_request_skip_reason(state, index) do cond do + MapSet.member?(state.hash_failures, index) -> :corrupt_source full_requests_queue?(state) -> :queue_full state.choke_me and not FastExtension.download?(state.fast_extension, index) -> :choked true -> nil end end - @spec log_download_request_skip(t(), Torrent.index(), :queue_full | :choked) :: t() + @spec log_download_request_skip(t(), Torrent.index(), :queue_full | :choked | :corrupt_source) :: + t() defp log_download_request_skip(state, index, :queue_full) do log_download(state, "request_skip queue_full index=#{index}", :debug) state @@ -1601,6 +1711,14 @@ defmodule Peer.Controller.State do state end + # This peer already served this piece with a bad hash. Drop the pin as well as + # the request, otherwise it stays pinned to an index it will never be asked + # for and stops contributing entirely. + defp log_download_request_skip(state, index, :corrupt_source) do + log_download(state, "request_skip corrupt_source index=#{index}", :debug) + clear_pin(state) + end + @spec apply_download_request(t(), Torrent.index()) :: t() defp apply_download_request(state, index) do pid = self() @@ -1906,6 +2024,77 @@ defmodule Peer.Controller.State do MapSet.member?(state.requests, subpiece(index, begin, length)) end + # Which of our own requests does this piece/reject answer? Treating "not + # outstanding" as a protocol error banned well-behaved peers for the RTT after + # every cancel, choke and repin — the ban is torrent-wide, so an aggressive + # re-pinning scheduler was quietly emptying the swarm. + @spec classify_answer(t(), Torrent.index(), Torrent.begin(), Torrent.length()) :: + :requested | :withdrawn | :unsolicited + defp classify_answer(state, index, begin, length) do + subpiece = subpiece(index, begin, length) + + cond do + MapSet.member?(state.requests, subpiece) -> :requested + Map.has_key?(state.withdrawn, subpiece) -> :withdrawn + Map.has_key?(state.withdrawn_prev, subpiece) -> :withdrawn + true -> :unsolicited + end + end + + @spec put_withdrawn(t(), Torrent.index(), Torrent.begin(), Torrent.length()) :: t() + defp put_withdrawn(%__MODULE__{} = state, index, begin, length) do + withdrawn = Map.update(state.withdrawn, subpiece(index, begin, length), 1, &(&1 + 1)) + + if map_size(withdrawn) > @max_withdrawn do + %__MODULE__{state | withdrawn: %{}, withdrawn_prev: withdrawn} + else + %__MODULE__{state | withdrawn: withdrawn} + end + end + + @spec withdraw_all(t(), MapSet.t(subpiece())) :: t() + defp withdraw_all(state, subpieces) do + Enum.reduce(subpieces, state, fn {index, begin, length}, acc -> + put_withdrawn(acc, index, begin, length) + end) + end + + @spec drop_withdrawn(t(), Torrent.index(), Torrent.begin(), Torrent.length()) :: t() + defp drop_withdrawn(%__MODULE__{} = state, index, begin, length) do + subpiece = subpiece(index, begin, length) + + if Map.has_key?(state.withdrawn, subpiece) do + %__MODULE__{state | withdrawn: decrement_owed(state.withdrawn, subpiece)} + else + %__MODULE__{state | withdrawn_prev: decrement_owed(state.withdrawn_prev, subpiece)} + end + end + + @spec decrement_owed(%{subpiece() => pos_integer()}, subpiece()) :: %{ + subpiece() => pos_integer() + } + defp decrement_owed(owed, subpiece) do + case owed do + %{^subpiece => 1} -> Map.delete(owed, subpiece) + %{^subpiece => n} -> Map.put(owed, subpiece, n - 1) + _ -> owed + end + end + + @spec note_unsolicited(t(), String.t()) :: t() | {:error, :unsolicited_blocks, t()} + defp note_unsolicited(%__MODULE__{} = state, what) do + state = %__MODULE__{state | unsolicited_blocks: state.unsolicited_blocks + 1} + log_download(state, "unsolicited #{what} total=#{state.unsolicited_blocks}", :debug) + + if state.unsolicited_blocks > @max_unsolicited_blocks do + # Wasteful, not malicious: drop the connection but leave the peer ID + # dialable, unlike :protocol_error. + {:error, :unsolicited_blocks, state} + else + state + end + end + @spec full_requests_queue?(t()) :: boolean() # Counts in-flight wire requests (`requests`) plus piece-worker :ok acks not # yet processed into `requests` (`pending_requests`). Before pending existed, @@ -1959,12 +2148,17 @@ defmodule Peer.Controller.State do @doc false @spec stale_useless_pin?(t()) :: boolean() def stale_useless_pin?(%__MODULE__{status: idx} = state) when is_integer(idx) do - state.choke_me and state.pin_downloaded_bytes == 0 and - pin_age_ms(state) >= stale_pin_threshold_ms(state.hash) + state.pin_downloaded_bytes == 0 and pin_age_ms(state) >= useless_pin_threshold_ms(state) end def stale_useless_pin?(_), do: false + @spec useless_pin_threshold_ms(t()) :: non_neg_integer() + defp useless_pin_threshold_ms(%__MODULE__{choke_me: true} = state), + do: stale_pin_threshold_ms(state.hash) + + defp useless_pin_threshold_ms(%__MODULE__{}), do: @snubbed_pin_ms + # Flush wire cancels + piece-worker rejects before repin or disconnect. # Mirrors handle_choke/1 local cleanup but also sends cancels — we are # actively switching pieces, not being choked by the remote peer. @@ -1975,7 +2169,7 @@ defmodule Peer.Controller.State do Downloads.reject(state.hash, index, state.id, begin, length) end) - %{state | requests: MapSet.new(), pending_requests: 0} + withdraw_all(%{state | requests: MapSet.new(), pending_requests: 0}, state.requests) end @spec apply_pin(t(), Torrent.index()) :: t() diff --git a/lib/elixir_torrent/peer/dial_backoff.ex b/lib/elixir_torrent/peer/dial_backoff.ex index 2db1960..bd81c13 100644 --- a/lib/elixir_torrent/peer/dial_backoff.ex +++ b/lib/elixir_torrent/peer/dial_backoff.ex @@ -36,6 +36,13 @@ defmodule Peer.DialBackoff do # can see it and escalate. Must be > @default_ttl_ms so the counter survives # a normal transient block. When retention expires, the whole row is swept. @fail_count_retention_ms 30 * 60 * 1_000 + # Escalation asks "is this endpoint dead", so it must count failures *close + # together*, not failures ever. Retention is refreshed on every failure, so a + # row re-dialled at least once per @fail_count_retention_ms never aged out and + # its counter only climbed: live rows reached fail_count 107-109, which is a + # permanent sticky block on an endpoint that merely fails intermittently. + # Failures further apart than this start a fresh streak. + @fail_streak_window_ms 10 * 60 * 1_000 # Hard cap on ETS rows — under heavy dial churn the table grows ~2k/30min today; # evict the oldest retention_until rows when we exceed this ceiling so memory # stays bounded on long sessions without weakening active blocks. @@ -76,8 +83,7 @@ defmodule Peer.DialBackoff do def filter(peers, hash, min_count \\ 0) when is_list(peers) and is_integer(min_count) do now = System.monotonic_time(:millisecond) {allowed, blocked} = partition_by_block(peers, hash, now) - soft_blocked = reject_sticky_blocked(blocked, hash, now) - apply_min_count_resurrection(allowed, soft_blocked, hash, min_count) + apply_min_count_resurrection(allowed, blocked, hash, min_count) catch :exit, _ -> peers end @@ -89,22 +95,14 @@ defmodule Peer.DialBackoff do end) end - # Sticky blocks (churn / hard failures / escalated) stay blocked even under target. - @spec reject_sticky_blocked([Peer.t()], Torrent.hash(), integer()) :: [Peer.t()] - defp reject_sticky_blocked(blocked, hash, now) do - Enum.reject(blocked, fn %Peer{ip: ip, port: port} -> - sticky_blocked?(hash, ip, port, now) - end) - end - @spec apply_min_count_resurrection([Peer.t()], [Peer.t()], Torrent.hash(), non_neg_integer()) :: [Peer.t()] - defp apply_min_count_resurrection(allowed, soft_blocked, hash, min_count) do - if min_count <= 0 or length(allowed) >= min_count or soft_blocked == [] do + defp apply_min_count_resurrection(allowed, blocked, hash, min_count) do + if min_count <= 0 or length(allowed) >= min_count or blocked == [] do allowed else - need = min(min_count - length(allowed), length(soft_blocked)) - allowed ++ take_blocked_for_min_count(soft_blocked, hash, need) + need = min(min_count - length(allowed), length(blocked)) + allowed ++ take_blocked_for_min_count(blocked, hash, need) end end @@ -119,6 +117,24 @@ defmodule Peer.DialBackoff do :exit, _ -> :ok end + @doc """ + Forget an endpoint's failure history — we have a live connection to it. + + This table answers "is this endpoint reachable", and a registered peer settles + that question: TCP connect and the BEP 3 handshake both completed. Only + `mark_productive/3` used to clear a row, which instead answers "was it useful", + and the two come apart badly on a leecher with nothing to trade: peers keep us + choked, so a reachable endpoint never delivers bytes and carries its failure + history for the whole session. Live, 1151 of 1162 active blocks were sticky and + every torrent was down to 3-8 dialable endpoints out of 37-53 known. + """ + @spec record_success(Torrent.hash(), :inet.ip_address(), :inet.port_number()) :: :ok + def record_success(hash, ip, port) do + GenServer.cast(__MODULE__, {:record_success, hash, ip, port}) + catch + :exit, _ -> :ok + end + @spec productive?(Torrent.hash(), :inet.ip_address(), :inet.port_number()) :: boolean() def productive?(hash, ip, port) do now = System.monotonic_time(:millisecond) @@ -194,7 +210,8 @@ defmodule Peer.DialBackoff do defp blocked?(hash, ip, port, now) do case :ets.lookup(@table, key(hash, ip, port)) do - [{_, blocked_until, _retention, _sticky, _fail_count}] when is_integer(blocked_until) -> + [{_, blocked_until, _retention, _sticky, _fail_count, _last_at}] + when is_integer(blocked_until) -> now < blocked_until _ -> @@ -204,7 +221,8 @@ defmodule Peer.DialBackoff do defp sticky_blocked?(hash, ip, port, now) do case :ets.lookup(@table, key(hash, ip, port)) do - [{_, blocked_until, _retention, true, _fail_count}] when is_integer(blocked_until) -> + [{_, blocked_until, _retention, true, _fail_count, _last_at}] + when is_integer(blocked_until) -> now < blocked_until _ -> @@ -212,6 +230,14 @@ defmodule Peer.DialBackoff do end end + @spec fail_count(Torrent.hash(), :inet.ip_address(), :inet.port_number()) :: non_neg_integer() + defp fail_count(hash, ip, port) do + case :ets.lookup(@table, key(hash, ip, port)) do + [{_, _, _, _, n, _}] when is_integer(n) -> n + _ -> 0 + end + end + @impl GenServer def init(_) do :ets.new(@table, [:named_table, :public, :set, read_concurrency: true]) @@ -231,6 +257,11 @@ defmodule Peer.DialBackoff do {:noreply, state} end + def handle_cast({:record_success, hash, ip, port}, state) do + :ets.delete(@table, key(hash, ip, port)) + {:noreply, state} + end + def handle_cast({:record, hash, ip, port, ttl_ms, reason}, state) do now = System.monotonic_time(:millisecond) key = key(hash, ip, port) @@ -250,21 +281,26 @@ defmodule Peer.DialBackoff do ) :: {boolean(), pos_integer()} defp insert_failure_record(key, _ip, _port, ttl_ms, reason, now) do productive? = productive_at?(key, now) - prev_fail_count = lookup_fail_count(key) - fail_count = prev_fail_count + 1 + fail_count = streak_count(key, now) + 1 {final_ttl, sticky?} = escalate(reason, fail_count, ttl_ms, productive?) blocked_until = now + final_ttl retention_until = max(blocked_until, now + @fail_count_retention_ms) - true = :ets.insert(@table, {key, blocked_until, retention_until, sticky?, fail_count}) + true = :ets.insert(@table, {key, blocked_until, retention_until, sticky?, fail_count, now}) {sticky?, fail_count} end - @spec lookup_fail_count(term()) :: non_neg_integer() - defp lookup_fail_count(key) do + # Failures more than @fail_streak_window_ms apart are separate incidents, not + # mounting evidence that the endpoint is dead. + @spec streak_count(term(), integer()) :: non_neg_integer() + defp streak_count(key, now) do case :ets.lookup(@table, key) do - [{_, _, _, _, n}] when is_integer(n) -> n - _ -> 0 + [{_, _, _, _, n, last_at}] + when is_integer(n) and is_integer(last_at) and now - last_at <= @fail_streak_window_ms -> + n + + _ -> + 0 end end @@ -295,7 +331,7 @@ defmodule Peer.DialBackoff do # Sweep by retention_until (the 3rd tuple element), not blocked_until — the # fail_count needs to outlive the block so the next re-dial can escalate. :ets.select_delete(@table, [ - {{:"$1", :"$2", :"$3", :"$4", :"$5"}, [{:<, :"$3", now}], [true]} + {{:"$1", :"$2", :"$3", :"$4", :"$5", :"$6"}, [{:<, :"$3", now}], [true]} ]) :ets.select_delete(@productive_table, [ @@ -318,9 +354,9 @@ defmodule Peer.DialBackoff do @table |> :ets.tab2list() - |> Enum.sort_by(fn {_, _, retention, _, _} -> retention end) + |> Enum.sort_by(fn {_, _, retention, _, _, _} -> retention end) |> Enum.take(drop) - |> Enum.each(fn {key, _, _, _, _} -> :ets.delete(@table, key) end) + |> Enum.each(fn {key, _, _, _, _, _} -> :ets.delete(@table, key) end) end end @@ -364,24 +400,36 @@ defmodule Peer.DialBackoff do end end - # When min_count pressure resurrects soft-blocked peers: productive first - # (known byte-deliverers), then v6 before v4. Under CGNAT outbound v6 yield - # is ~10–20× better than v4; matching a v4-heavy allowed slice would amplify - # dead v4 candidates and drain the v6 dial budget. + # Priority when min_count pressure makes us re-dial blocked endpoints: + # + # 1. productive — it already delivered bytes to us, so it is scarce and proven + # 2. soft-blocked before sticky — sticky means we deliberately wrote it off + # 3. v6 before v4 — outbound v6 yield here is ~10-20× v4 under CGNAT, so a + # v4-heavy slice would burn the batch on timeouts + # 4. fewest failures first — least evidence against it + # + # Sticky used to be excluded outright, which is right while anything else is + # dialable and starves the torrent once nothing is. Every block a CGNAT host + # records is sticky in practice — churn, hard failures and the escalated + # write-off all are — so live 1151 of 1162 active blocks refused resurrection, + # leaving eight of nine torrents asking for a 50-endpoint batch and getting + # 3-8, with the scarce v6 pool blocked 2-6 of 5-8. A SYN to a written-off + # endpoint costs one timeout; not dialling costs the torrent. @spec take_blocked_for_min_count([Peer.t()], Torrent.hash(), non_neg_integer()) :: [Peer.t()] defp take_blocked_for_min_count(_blocked, _hash, 0), do: [] defp take_blocked_for_min_count(blocked, hash, need) do - {productive, rest} = - Enum.split_with(blocked, fn %Peer{ip: ip, port: port} -> - productive?(hash, ip, port) - end) - - {v6, v4} = Enum.split_with(rest, fn peer -> peer_family(peer) == :inet6 end) + now = System.monotonic_time(:millisecond) - productive - |> Kernel.++(v6) - |> Kernel.++(v4) + blocked + |> Enum.sort_by(fn %Peer{ip: ip, port: port} = peer -> + { + if(productive?(hash, ip, port), do: 0, else: 1), + if(sticky_blocked?(hash, ip, port, now), do: 1, else: 0), + if(peer_family(peer) == :inet6, do: 0, else: 1), + fail_count(hash, ip, port) + } + end) |> Enum.take(need) end diff --git a/lib/elixir_torrent/peer/endpoints.ex b/lib/elixir_torrent/peer/endpoints.ex index 4ff8dda..9e175f1 100644 --- a/lib/elixir_torrent/peer/endpoints.ex +++ b/lib/elixir_torrent/peer/endpoints.ex @@ -43,6 +43,30 @@ defmodule Peer.Endpoints do :exit, _ -> :ok end + @doc """ + Record why a peer connection ended, so the disconnect log can name the cause. + + What this module monitors is the per-peer supervisor, and that supervisor is + `auto_shutdown: :any_significant` — it exits with a bare `:shutdown` as soon as + either child stops, whatever the child's own reason was. So every disconnect + logged `reason=:shutdown`: true of the supervisor, and useless for telling a + peer that closed the socket on us apart from one we dropped ourselves. + + The controller reports its reason from `terminate/2`, which runs before the + supervisor exits, so the note is already here when the `:DOWN` arrives. + """ + @spec note_disconnect_reason( + Torrent.hash(), + :inet.ip_address(), + :inet.port_number(), + term() + ) :: :ok + def note_disconnect_reason(hash, ip, port, reason) do + GenServer.cast(__MODULE__, {:note_disconnect_reason, endpoint_key(hash, ip, port), reason}) + catch + :exit, _ -> :ok + end + @spec registered?(Torrent.hash(), :inet.ip_address(), :inet.port_number()) :: boolean() def registered?(hash, ip, port) do GenServer.call(__MODULE__, {:registered?, hash, ip, port}) @@ -75,7 +99,12 @@ defmodule Peer.Endpoints do def init(_) do table = :ets.new(@table, [:set, :protected, read_concurrency: true]) peer_ids = :ets.new(@peer_id_table, [:set, :protected, read_concurrency: true]) - {:ok, %{table: table, peer_ids: peer_ids, monitors: %{}}} + {:ok, %{table: table, peer_ids: peer_ids, monitors: %{}, reasons: %{}}} + end + + @impl GenServer + def handle_cast({:note_disconnect_reason, key, reason}, state) do + {:noreply, %{state | reasons: Map.put(state.reasons, key, reason)}} end @impl GenServer @@ -118,6 +147,14 @@ defmodule Peer.Endpoints do ref = Process.monitor(pid) :ets.insert(table, {key, pid}) now = System.monotonic_time(:millisecond) + # A note left by a previous connection to this endpoint describes that one, + # not the connection starting now. + state = %{state | reasons: Map.delete(state.reasons, key)} + # Registration is the proof of reachability the dial backoff is asking for: + # connect and the BEP 3 handshake both completed. Inbound peers count too — + # if they reached us, our earlier dial timeouts were about their NAT, not a + # dead host. + Peer.DialBackoff.record_success(hash, ip, port) {:reply, :ok, %{state | monitors: Map.put(monitors, ref, {key, now})}} end @@ -180,8 +217,9 @@ defmodule Peer.Endpoints do :ets.delete(state.table, key) drop_peer_id(state.peer_ids, key) maybe_backoff_churn(key, connected_at) - log_disconnect(key, reason) - {:noreply, %{state | monitors: monitors}} + {noted, reasons} = Map.pop(state.reasons, key) + log_disconnect(key, noted || reason) + {:noreply, %{state | monitors: monitors, reasons: reasons}} end end diff --git a/lib/elixir_torrent/peer/sender.ex b/lib/elixir_torrent/peer/sender.ex index 76faf02..074b20e 100644 --- a/lib/elixir_torrent/peer/sender.ex +++ b/lib/elixir_torrent/peer/sender.ex @@ -416,8 +416,15 @@ defmodule Peer.Sender do # this message is stale and dropped. Cheaper than Process.cancel_timer/1 on every # completed frame, and just as race-free: a cancel racing the timer's own send # would need this same staleness check anyway. + # Dropping the connection is the defence here; a ban is not. A frame that stops + # half-way is what a congested path looks like — a uTP window that collapsed, a + # peer swapping — and on this host that is routine, not a slow-loris. Banning + # the peer ID for it excluded mainstream clients from every torrent at once. def handle_info({:frame_stall, ref}, %__MODULE__{frame_stall_ref: ref, key: key} = state) do - Acceptor.malicious_peer(Peer.key_to_id(key)) + Logger.debug( + "[peer_sender] frame_stalled peer=#{Peer.log_key(key)} hash=#{Torrent.hex_encoded_hash(Peer.key_to_hash(key))} buffered=#{byte_size(state.buffer)}" + ) + {:stop, {:shutdown, :frame_stalled}, state} end @@ -468,7 +475,7 @@ defmodule Peer.Sender do drain_inbound(state) :protocol_error -> - Acceptor.malicious_peer(Peer.key_to_id(key)) + ban(key, :unparsable_message) {:stop, {:shutdown, :protocol_error}, %{state | buffer: rest}} end @@ -477,11 +484,23 @@ defmodule Peer.Sender do {:noreply, state, @timeout} :protocol_error -> - Acceptor.malicious_peer(Peer.key_to_id(key)) + ban(key, :malformed_frame) {:stop, {:shutdown, :protocol_error}, state} end end + # Naming the rule matters: a ban keeps this peer out of every torrent, so a + # bare "protocol_error" leaves no way to tell a genuinely broken client from a + # rule of ours that is too strict. + @spec ban(Peer.key(), atom()) :: :ok + defp ban(key, cause) do + Logger.debug( + "[peer_sender] banned peer=#{Peer.log_key(key)} hash=#{Torrent.hex_encoded_hash(Peer.key_to_hash(key))} cause=#{cause}" + ) + + Acceptor.malicious_peer(Peer.key_to_id(key)) + end + # Starts the stall watchdog the first time `buffer` holds an incomplete frame, and # leaves it alone on every later call for that SAME frame -- only drain_inbound's # `:ok` branch (real progress) is allowed to replace it. An empty buffer means diff --git a/lib/elixir_torrent/torrent/controller.ex b/lib/elixir_torrent/torrent/controller.ex index 2a4cefb..e4d26bc 100644 --- a/lib/elixir_torrent/torrent/controller.ex +++ b/lib/elixir_torrent/torrent/controller.ex @@ -46,6 +46,14 @@ defmodule Torrent.Controller do # Rate-limit "download waiting" — reconcile_pump + peer kicks can fire # {:next_piece} many times per second while conn=0; one line/min is enough. @download_waiting_log_interval_sec 60 + # A {:next_piece} that cannot place a piece reschedules itself, and every + # independent trigger — the 2 s reconcile tick, peer handoff kicks, each + # piece's requests_are_dealt closure — starts another such chain. Nothing + # deduplicated them, so the chains simply accumulated: a live debug build went + # from 99 to ~2600 discovery dial cycles per minute over twelve minutes with no + # change in swarm size, and the tracker started answering 403. One pending wake + # per controller is enough, because every wake runs the same full pump attempt. + @next_piece_timer_key :next_piece_timer @spec start_link(Torrent.hash()) :: GenServer.on_start() def start_link(hash), @@ -89,7 +97,7 @@ defmodule Torrent.Controller do "[resume] controller_start hash=#{Torrent.hex_encoded_hash(hash)} scheduling download" ) - send_after(self(), {:next_piece, :random}, 500) + schedule_next_piece(:random, 500) send_after(self(), :unchoke, 1_000) # Kick off the periodic pump reconciler — see @reconcile_interval. send_after(self(), :reconcile_pump, @reconcile_interval) @@ -118,13 +126,17 @@ defmodule Torrent.Controller do end reconcile_pump_kick(hash, active_count, effective_max, connected, unchoked) + reconcile_endgame_mode(hash, active) reconcile_refresh_interest(hash, active, active_count, connected) send_after(self(), :reconcile_pump, @reconcile_interval) {:noreply, hash} end - def handle_info({:next_piece, strategy} = msg, hash) do + def handle_info({:next_piece, strategy}, hash) do + cancel_next_piece_timer() + strategy = collapse_queued_next_piece(strategy) + cond do Model.downloaded?(hash) -> mark_complete(hash) @@ -140,7 +152,7 @@ defmodule Torrent.Controller do Model.set_peer_status(hash, :connecting_to_peers) PeerDiscovery.connecting_to_peers(hash) - send_after(self(), msg, 20_000) + schedule_next_piece(strategy, 20_000) end {:noreply, hash} @@ -167,7 +179,7 @@ defmodule Torrent.Controller do effective_max = effective_max_parallel(unchoked) if length(active) >= effective_max do - send_after(self(), {:next_piece, strategy}, 500) + schedule_next_piece(strategy, 500) else pick_and_start_piece(hash, strategy, active, connected) end @@ -214,7 +226,7 @@ defmodule Torrent.Controller do mark_complete(hash) else fallback_when_no_piece(hash, active, connected) - send_after(self(), {:next_piece, strategy}, @next_piece_timeout) + schedule_next_piece(strategy, @next_piece_timeout) end end @@ -247,7 +259,7 @@ defmodule Torrent.Controller do Model.set_peer_status(hash, nil) PeerDiscovery.connecting_to_peers(hash) - send_after(self(), {:next_piece, strategy}, @next_piece_timeout) + schedule_next_piece(strategy, @next_piece_timeout) end end @@ -261,7 +273,7 @@ defmodule Torrent.Controller do "[reconcile_pump] hash=#{Torrent.hex_encoded_hash(hash)} active=#{active_count} max=#{effective_max} unchoked=#{unchoked} peers=#{connected} kick=next_piece" ) - send(self(), {:next_piece, :rare}) + schedule_next_piece(:rare, 0) true -> Logger.debug( @@ -270,6 +282,18 @@ defmodule Torrent.Controller do end end + # Crossing into endgame is a property of the torrent, but the decision that + # matters — may two peers hold the same block? — lives in each piece worker, + # which read the mode when it started. Pieces already in flight at the + # transition therefore stayed in normal mode and kept one-block-one-peer + # exclusivity for the rest of the download, which is what left a torrent at + # 99.94% with its last piece fully claimed by a single timing-out peer. + defp reconcile_endgame_mode(hash, active) do + if Model.get(hash, :mode) == :endgame do + Enum.each(active, &Downloads.piece_enter_endgame(hash, &1)) + end + end + defp reconcile_refresh_interest(hash, active, active_count, connected) do # Endgame (or multiple in-flight pieces): refresh interest on every active # index, not only Model.peer_status. A single controller status made @@ -300,7 +324,7 @@ defmodule Torrent.Controller do # closure to fire (that closure only wakes on the last subpiece being handed # out, which never happens if the peer stalls). A small delay lets the just- # started piece begin taking requests before we contend again. - send_after(self(), {:next_piece, strategy}, @post_start_pump_delay) + schedule_next_piece(strategy, @post_start_pump_delay) end defp seeder_has_piece?(hash, index) do @@ -378,6 +402,57 @@ defmodule Torrent.Controller do |> Enum.each(&Downloads.abort_idle_piece(hash, &1, force: true)) end + # An immediate kick stays a plain `send`: `send_after(_, 0)` hands delivery to + # the timer service, which loses the ordering guarantee that a self-send is + # already queued before the next message this process handles. Bursts of these + # are collapsed on the receiving side instead. + defp schedule_next_piece(strategy, 0) do + send(self(), {:next_piece, strategy}) + :ok + end + + defp schedule_next_piece(strategy, delay) do + case Process.get(@next_piece_timer_key) do + ref when is_reference(ref) -> + if Process.read_timer(ref), do: :ok, else: put_next_piece_timer(strategy, delay) + + _ -> + put_next_piece_timer(strategy, delay) + end + end + + defp put_next_piece_timer(strategy, delay) do + Process.put(@next_piece_timer_key, send_after(self(), {:next_piece, strategy}, delay)) + :ok + end + + # We are about to run the pump, so any wake still armed for it is redundant. + # Dropping only the stored reference would leave that timer to fire later and + # start a second chain — which is precisely how the chains accumulated. + defp cancel_next_piece_timer do + case Process.delete(@next_piece_timer_key) do + ref when is_reference(ref) -> Process.cancel_timer(ref) + _ -> :ok + end + + :ok + end + + # Immediate `send`s (Controller.kick/1, the reconcile kick) bypass the timer, + # so a burst can still be sitting in the mailbox. Handling one is equivalent to + # handling all of them; `:rare` wins over `:random` because rarest-first is the + # strategy the reconcile path asks for. + defp collapse_queued_next_piece(strategy) do + receive do + {:next_piece, queued} -> collapse_queued_next_piece(merge_strategy(strategy, queued)) + after + 0 -> strategy + end + end + + defp merge_strategy(:rare, _), do: :rare + defp merge_strategy(_, queued), do: queued + defp maybe_log_download_waiting(hash, connected, downloaded, left, status) do key = {:download_waiting_logged, hash} now = System.monotonic_time(:second) diff --git a/lib/elixir_torrent/torrent/downloads.ex b/lib/elixir_torrent/torrent/downloads.ex index 53ed851..85c1cdb 100644 --- a/lib/elixir_torrent/torrent/downloads.ex +++ b/lib/elixir_torrent/torrent/downloads.ex @@ -56,6 +56,23 @@ defmodule Torrent.Downloads do # be re-pinned to a fresh active piece. defdelegate piece_has_waiting?(hash, index), to: Piece, as: :has_waiting? + # Blocks nobody has claimed yet. `piece_has_waiting?/2` also counts blocks + # in flight to other peers, which is right for endgame but wrong when + # deciding whether *this* peer still has work here. + defdelegate piece_has_unclaimed?(hash, index), to: Piece, as: :has_unclaimed? + + # Whether a piece still has anything for one specific peer: unclaimed blocks, + # or blocks that peer is already fetching. + defdelegate piece_serves_peer?(hash, index, peer_id), to: Piece, as: :serves_peer? + + # Upgrade a running worker to endgame; see `Piece.enter_endgame/2`. + defdelegate piece_enter_endgame(hash, index), to: Piece, as: :enter_endgame + + # Distinguishes "no worker yet" from "worker with nothing left to hand out", + # which `piece_has_waiting?/2` collapses into `false`. + @spec piece_whereis(Torrent.hash(), Torrent.index()) :: pid() | nil + defdelegate piece_whereis(hash, index), to: Piece, as: :whereis + @spec piece_has_in_flight?(Torrent.hash(), Torrent.index()) :: boolean() def piece_has_in_flight?(hash, index) do case Piece.whereis(hash, index) do diff --git a/lib/elixir_torrent/torrent/downloads/piece.ex b/lib/elixir_torrent/torrent/downloads/piece.ex index 3470edf..c535d56 100644 --- a/lib/elixir_torrent/torrent/downloads/piece.ex +++ b/lib/elixir_torrent/torrent/downloads/piece.ex @@ -73,6 +73,42 @@ defmodule Torrent.Downloads.Piece do :exit, _ -> false end + # Same probe restricted to blocks nobody has claimed yet — see + # `handle_call(:has_unclaimed?, ...)`. + @spec has_unclaimed?(Torrent.hash(), Torrent.index()) :: boolean() + def has_unclaimed?(hash, index) do + case GenServer.whereis(key(index, hash)) do + nil -> false + pid -> GenServer.call(pid, :has_unclaimed?, 1_000) + end + catch + :exit, _ -> false + end + + # Same probe as `has_unclaimed?/2` but also true while this peer's own requests + # are in flight here — see `handle_call({:serves_peer?, _}, ...)`. + @spec serves_peer?(Torrent.hash(), Torrent.index(), Peer.id()) :: boolean() + def serves_peer?(hash, index, peer_id) do + case GenServer.whereis(key(index, hash)) do + nil -> false + pid -> GenServer.call(pid, {:serves_peer?, peer_id}, 1_000) + end + catch + :exit, _ -> false + end + + # Level-triggered mode upgrade from the torrent controller: a worker reads the + # torrent's mode once (State.download/3), so pieces already in flight when the + # torrent crosses into endgame would otherwise never get redundant sources. + # Idempotent, and one-way — endgame is never revoked. + @spec enter_endgame(Torrent.hash(), Torrent.index()) :: :ok + def enter_endgame(hash, index) do + case GenServer.whereis(key(index, hash)) do + nil -> :ok + pid -> GenServer.cast(pid, {:enter_endgame, []}) + end + end + @spec whereis(Torrent.hash(), Torrent.index()) :: pid() | nil def whereis(hash, index), do: GenServer.whereis(key(index, hash)) @@ -179,6 +215,23 @@ defmodule Torrent.Downloads.Piece do {:reply, state.requests != [], state} end + # Strictly "are there blocks left to hand out". Unlike :has_waiting? this + # ignores in-flight requests, because a peer cannot be given a block that + # another peer already holds — outside endgame, where duplicating them is + # the whole point. + def handle_call(:has_unclaimed?, _from, state) do + {:reply, state.waiting != [], state} + end + + # "Is this peer still working here?" — unclaimed blocks it could be handed, or + # blocks it is already fetching. See `Swarm.pin_drained?/4` for why the second + # half matters: a peer holding every in-flight request on a piece is the reason + # that piece has nothing unclaimed left. + def handle_call({:serves_peer?, peer_id}, _from, state) do + serves? = state.waiting != [] or Enum.any?(state.requests, &(&1.peer_id == peer_id)) + {:reply, serves?, state} + end + # Sync ack for Downloads.request/4 — see request/4 above. def handle_call({:request, [peer_id, callback]}, _from, state) do new_state = State.request(state, peer_id, callback) @@ -218,6 +271,7 @@ defmodule Torrent.Downloads.Piece do {:stop, :normal, state} else Logger.warning("[piece_download] hash=#{hash_hex} index=#{state.index} verify_failed") + report_corrupt_source(state, hash_hex) fire_dealt(state) {:stop, {:shutdown, :wrong_subpiece}, state} end @@ -225,6 +279,25 @@ defmodule Torrent.Downloads.Piece do defp finish_if_complete(state), do: {:noreply, state} + # A piece that fails its SHA-1 was assembled from bad bytes. When one peer + # supplied every block, it is provably the source, so tell its controller — + # otherwise nothing stops us re-requesting the same piece from the same peer + # forever. With several contributors the failure cannot be attributed and the + # piece is simply retried. + defp report_corrupt_source(%State{} = state, hash_hex) do + case State.sole_contributor(state) do + nil -> + :ok + + peer_id -> + Logger.warning( + "[piece_download] hash=#{hash_hex} index=#{state.index} corrupt_source peer=#{Peer.log_id(peer_id)}" + ) + + Peer.Controller.hash_check_failed({peer_id, state.hash}, state.index) + end + end + # Best-effort invocation of the controller's pump-wake closure. It is # idempotent from the controller's perspective (posts {:next_piece, :rare}; # the controller's handler is a capacity check). Guard against nil (worker diff --git a/lib/elixir_torrent/torrent/downloads/piece/state.ex b/lib/elixir_torrent/torrent/downloads/piece/state.ex index fdb51e6..714d709 100644 --- a/lib/elixir_torrent/torrent/downloads/piece/state.ex +++ b/lib/elixir_torrent/torrent/downloads/piece/state.ex @@ -24,7 +24,12 @@ defmodule Torrent.Downloads.Piece.State do :timer, :mode, monitoring: %{}, - requests: [] + requests: [], + # Blocks accepted into this piece, counted per supplying peer. A piece that + # fails its hash check was assembled from bad bytes, and when one peer + # supplied all of them BEP 3 gives us a provable culprit — see + # `sole_contributor/1`. + contributors: %{} ] @type timer :: reference() | nil @@ -37,7 +42,8 @@ defmodule Torrent.Downloads.Piece.State do timer: timer(), mode: Piece.mode(), monitoring: map(), - requests: list(Request.t()) + requests: list(Request.t()), + contributors: %{optional(Peer.id()) => pos_integer()} } @subpiece_length Piece.max_length() @@ -107,6 +113,37 @@ defmodule Torrent.Downloads.Piece.State do } end + # A worker captures the torrent's mode once, in `download/3`, so a piece that + # was started before the torrent crossed into endgame keeps running in normal + # mode — where a block belongs to exactly one peer — for the rest of its life. + # That removes redundancy from precisely the pieces endgame exists for: live, + # the last piece of a torrent sat with all 64 blocks claimed by a single peer + # that timed out 297 times over an hour, while three unchoked peers holding + # the piece had nothing to ask for. Re-queueing the in-flight subpieces is + # what opens them to other peers; the requests themselves stay, and + # `do_request/3` caps redundancy at @endgame_redundancy per block. + @spec enter_endgame(t()) :: t() + def enter_endgame(%__MODULE__{mode: :endgame} = state), do: state + + def enter_endgame(%__MODULE__{} = state) do + Logger.debug( + "[piece_download] hash=#{Torrent.hex_encoded_hash(state.hash)} index=#{state.index} mode=endgame in_flight=#{length(state.requests)} waiting=#{length(state.waiting)}" + ) + + # Endgame runs without the piece-level stall/orphan timer, and whichever of + # the two is pending must also be flushed: delivered late it would abort a + # worker that is now making progress. + cancel_timer(state.timer, :idle_orphan_check) + cancel_timer(state.timer, :timeout) + + %__MODULE__{ + state + | mode: :endgame, + timer: nil, + waiting: Enum.uniq(state.waiting ++ Enum.map(state.requests, & &1.subpiece)) + } + end + @spec make_subpieces(waiting(), Torrent.length(), Torrent.length() | 0) :: waiting() defp make_subpieces(acc, len, pos) when pos + @subpiece_length >= len do [{pos, len - pos} | acc] @@ -234,7 +271,8 @@ defmodule Torrent.Downloads.Piece.State do state = %__MODULE__{ state | requests: requests, - waiting: List.delete(state.waiting, subpiece) + waiting: List.delete(state.waiting, subpiece), + contributors: Map.update(state.contributors, peer_id, 1, &(&1 + 1)) } with %__MODULE__{mode: :endgame, waiting: []} <- state do @@ -243,6 +281,20 @@ defmodule Torrent.Downloads.Piece.State do end end + @doc """ + The peer that supplied every accepted block of this piece, if there was only one. + + Returns `nil` when several peers contributed, because then a failed hash check + cannot be pinned on any single one of them. + """ + @spec sole_contributor(t()) :: Peer.id() | nil + def sole_contributor(%__MODULE__{contributors: contributors}) do + case Map.keys(contributors) do + [peer_id] -> peer_id + _ -> nil + end + end + @spec valid_subpiece?(t(), Torrent.begin(), Torrent.length()) :: boolean() defp valid_subpiece?(state, begin, length) do piece_len = Model.piece_length(state.hash, state.index) @@ -297,9 +349,15 @@ defmodule Torrent.Downloads.Piece.State do end @doc false + # "Can this piece still be sourced?" is a question about *this index*, not about + # the torrent. The old test also required `Swarm.count(hash) == 0`, so on any + # torrent with peers an idle worker whose holders had all disconnected was never + # aborted: it kept one of the @max_parallel_pieces slots with `monitoring: 0` + # and 64 unclaimed blocks forever. Observed live at 7 of 12 slots held that way, + # capping a 26-unchoked-peer torrent at 5 pieces in flight. @spec orphan_no_sources?(t()) :: boolean() - def orphan_no_sources?(%__MODULE__{hash: hash, monitoring: monitoring}) do - map_size(monitoring) == 0 and Swarm.count(hash) == 0 + def orphan_no_sources?(%__MODULE__{hash: hash, index: index, monitoring: monitoring}) do + map_size(monitoring) == 0 and not Swarm.any_has_piece?(hash, index) end @spec maybe_abort_orphan(t()) :: t() | {:abort, t()} diff --git a/lib/elixir_torrent/torrent/swarm.ex b/lib/elixir_torrent/torrent/swarm.ex index 4c5def0..7b8f8e7 100644 --- a/lib/elixir_torrent/torrent/swarm.ex +++ b/lib/elixir_torrent/torrent/swarm.ex @@ -86,8 +86,8 @@ defmodule Torrent.Swarm do # subpieces left (drained) → free to move. This is the fix: without # it, a peer would sit pinned to a drained piece until its worker # died, losing all its bandwidth to the pump. - # * pinned to another active piece with waiting subpieces BUT this peer - # is choked and has delivered zero bytes on the pin for longer than + # * pinned to another active piece that still has something for it BUT this + # peer is choked and has delivered zero bytes on the pin for longer than # @stale_pin_ms → free to move (endgame monopoly fix). In endgame, a # stable hash spreads useless pins across ALL remaining indices so one # choked piece does not hoard the whole swarm. @@ -102,28 +102,40 @@ defmodule Torrent.Swarm do true other -> - may_leave_pin?(hash, key, index, other, active_indices) + endgame? = Model.get(hash, :mode) == :endgame + + target_accepts_repin?(hash, index, endgame?) and + may_leave_pin?(hash, key, index, other, active_indices) end catch :exit, _ -> false end - defp may_leave_pin?(hash, key, index, other, active_indices) do - drained? = not Downloads.piece_has_waiting?(hash, other) - endgame? = Model.get(hash, :mode) == :endgame - useless? = useless_pin_may_switch?(hash, key, index, other, active_indices) + # Only gates *moving* a peer, never its first pin: `interested` is sent from + # `check_interested/1`, which needs an integer pin, so refusing a peer that has + # none would leave it permanently un-interested and therefore never unchoked. + # + # Moving one is different. Pinning a peer to a piece whose blocks are all + # already claimed parks its bandwidth until something else moves it. Live: 27 + # of 37 peers pinned to 4 such pieces while 8 active pieces with 49-64 free + # blocks had no peer at all. A piece with no worker yet is accepted; it is + # about to start. + defp target_accepts_repin?(hash, index, endgame?) do + case Downloads.piece_whereis(hash, index) do + nil -> true + _pid -> endgame? or Downloads.piece_has_unclaimed?(hash, index) + end + end + defp may_leave_pin?(hash, key, index, other, active_indices) do cond do other not in active_indices -> true - drained? and not endgame? -> - true - - drained? and endgame? and useless? -> - true + pin_drained?(hash, other, Peer.key_to_id(key)) -> + endgame_target_allows?(hash, key, index, active_indices) - useless? -> + useless_pin_may_switch?(hash, key, index, other, active_indices) -> true true -> @@ -131,25 +143,76 @@ defmodule Torrent.Swarm do end end + # Where a freed peer is allowed to land. Outside endgame nothing extra is + # needed: `target_accepts_repin?/3` requires the target to still hold + # unclaimed blocks, so a piece stops accepting peers once they have claimed + # everything it has. Endgame accepts *any* active index by design, which + # leaves the count unbounded — and `reconcile_refresh_interest/4` walks every + # active index in order, so without a gate the whole freed set lands on + # whichever index it reaches first. The stable hash spreads them across the + # remaining indices instead, which is what endgame wants anyway: redundant + # sources on all of them rather than a crowd on one. + defp endgame_target_allows?(hash, key, index, active_indices) do + case Model.get(hash, :mode) do + :endgame when length(active_indices) > 1 -> + endgame_preferred_index(key, active_indices) == index + + _ -> + true + end + end + + # "Nothing here for this peer any more." Outside endgame that means every + # block has been claimed *by someone else*, so this peer cannot be handed one + # and holding it here only wastes it. Endgame deliberately re-requests + # in-flight blocks from several peers, so there the pin stays useful until the + # piece is genuinely finished. + # + # The peer's own in-flight requests have to count as work, or the rule eats + # itself: a peer that has claimed every remaining block makes the piece look + # drained, gets moved off, and the move cancels the very requests that drained + # it — so the blocks return to `waiting`, the piece looks workable again, and + # the next reconcile tick pins the peer straight back. Live, that oscillated at + # the 2 s tick: one peer re-sent the same block 79 times in ten minutes and + # delivered none of it, and the swarm as a whole sent 17 requests per block + # received. + # + # The question is asked of the piece worker rather than split on the torrent's + # mode, because the two do not agree: a torrent enters `:endgame` while the + # workers it already started keep running in normal mode, and the endgame + # branch used to probe `piece_has_waiting?/2`, which counts *other* peers' + # in-flight blocks. On a normal-mode piece that is precisely the wrong answer — + # it cannot hand this peer one of those blocks — so the pin was held forever. + # Live on a torrent parked at 99.18%: 8 of 10 peers pinned to one piece with no + # unclaimed blocks left, three of them unchoked and idle, while six pieces sat + # with all 64 blocks unclaimed and no peer at all. + # + # `waiting` already encodes the mode, which is why one probe serves both: a + # normal-mode worker drops a block from `waiting` when it hands it out, while + # an endgame worker keeps it there until it is *delivered*, so redundant + # re-requests stay possible and the pin stays useful until the piece is + # genuinely finished — the endgame behaviour the split was there to protect. + defp pin_drained?(hash, index, peer_id), + do: not Downloads.piece_serves_peer?(hash, index, peer_id) + # A peer choked with zero bytes on its current pin for long enough is not # contributing to that piece. In endgame, only re-pin to another active # index when a stable hash says this peer "owns" that index — otherwise # reconcile's multi-interest pass would leave every peer on the last index. defp useless_pin_may_switch?(hash, key, index, _other, active_indices) do Peer.Controller.stale_useless_pin?(key) and - case Model.get(hash, :mode) do - :endgame when length(active_indices) > 1 -> - endgame_preferred_index(key, active_indices) == index - - _ -> - true - end + endgame_target_allows?(hash, key, index, active_indices) end - defp endgame_preferred_index({_hash, peer_id}, active_indices) do + # Spreads peers across the remaining indices by hashing the *peer*. The key is + # `{id, hash}` (`Peer.make_key/2`), and this destructured it as `{_hash, + # peer_id}` — so every peer on a torrent hashed the same value and got the same + # answer, funnelling the whole swarm onto one index instead of spreading it. + # That is the inverse of the intent, and it is what held 8 of 10 peers on a + # single piece while a torrent sat at 99.18%. + defp endgame_preferred_index(key, active_indices) do sorted = Enum.sort(active_indices) - bucket = rem(:erlang.phash2(peer_id, length(sorted)), length(sorted)) - Enum.at(sorted, bucket) + Enum.at(sorted, :erlang.phash2(Peer.key_to_id(key), length(sorted))) end @spec seed(Torrent.hash()) :: :ok diff --git a/lib/elixir_torrent/torrent/uploader.ex b/lib/elixir_torrent/torrent/uploader.ex index 2a073de..b850389 100644 --- a/lib/elixir_torrent/torrent/uploader.ex +++ b/lib/elixir_torrent/torrent/uploader.ex @@ -7,6 +7,8 @@ defmodule Torrent.Uploader do alias Torrent.{FileHandle, Model} + require Logger + @spec child_spec(Torrent.hash()) :: Supervisor.child_spec() def child_spec(hash) do %{ @@ -42,14 +44,48 @@ defmodule Torrent.Uploader do ) end + # The peer this block is for can disappear at any point: it disconnects, trips + # a protocol error, or its supervisor shuts it down. The callback is a + # `GenServer.call` into that peer, so it then exits with the peer's reason. + # Sending a block to a peer that is gone is moot rather than a fault — before + # this only `:noproc` was tolerated, so one peer dying with `:protocol_error` + # while it had requests outstanding logged 87 `[error]` task crashes at once. + # Anything that is not the peer going away is still allowed to crash the task. @spec deliver_callback((iodata() -> any()), iodata()) :: any() defp deliver_callback(callback, block) do callback.(block) catch - :exit, :noproc -> :cancelled - :exit, {:noproc, _call} -> :cancelled + :exit, reason -> + if undeliverable?(reason), do: cancel_upload(reason), else: exit(reason) end + # Reasons that mean "this block is never reaching that peer". Anything else + # still crashes the task, so a genuine fault in the delivery path is not + # swallowed. A timeout counts: the peer's controller could not accept the + # block within the call timeout, which in practice is a peer whose socket has + # stopped draining. BEP 3 lets us simply not answer a request — it will ask + # again if it still wants the block. + @spec undeliverable?(term()) :: boolean() + defp undeliverable?(:noproc), do: true + defp undeliverable?(:normal), do: true + defp undeliverable?(:shutdown), do: true + defp undeliverable?(:timeout), do: true + defp undeliverable?({:shutdown, _reason}), do: true + # `GenServer.call/3` wraps the callee's exit reason as `{reason, call_info}`. + defp undeliverable?({reason, _call}), do: undeliverable?(reason) + defp undeliverable?(_), do: false + + # One debug line rather than a task crash dump. A slow peer with a deep + # request queue produced 333 `[error]` reports in fifteen minutes — all the + # same fact, repeated once per in-flight block. + defp cancel_upload(reason) do + Logger.debug("[peer_upload] deliver_cancelled reason=#{inspect(unwrap_reason(reason))}") + :cancelled + end + + defp unwrap_reason({reason, _call}), do: reason + defp unwrap_reason(reason), do: reason + @spec cancel( Torrent.hash(), Peer.id(), diff --git a/mix.exs b/mix.exs index 2a449b9..c46acf6 100644 --- a/mix.exs +++ b/mix.exs @@ -1,7 +1,7 @@ defmodule ElixirTorrent.MixProject do use Mix.Project - @version "0.6.5" + @version "0.6.6" def project do [ diff --git a/test/acceptor_dial_handshake_coverage_batch_test.exs b/test/acceptor_dial_handshake_coverage_batch_test.exs index b279e88..a53f0af 100644 --- a/test/acceptor_dial_handshake_coverage_batch_test.exs +++ b/test/acceptor_dial_handshake_coverage_batch_test.exs @@ -688,6 +688,28 @@ defmodule AcceptorDialHandshakeCoverageBatchTest do assert byte_size(Acceptor.key()) == 4 end + test "peer TCP sockets bound their writes, inbound and outbound alike" do + # Without send_timeout a peer that stops reading blocks Peer.Sender inside + # :gen_tcp.send forever and its mailbox grows unbounded. + {:ok, listen} = :gen_tcp.listen(0, Acceptor.tcp_socket_options(:inet)) + on_exit(fn -> :gen_tcp.close(listen) end) + + {:ok, port} = :inet.port(listen) + {:ok, outbound} = :gen_tcp.connect({127, 0, 0, 1}, port, [:binary, active: false]) + on_exit(fn -> :gen_tcp.close(outbound) end) + + {:ok, accepted} = :gen_tcp.accept(listen, 2_000) + on_exit(fn -> :gen_tcp.close(accepted) end) + + :ok = Acceptor.apply_tcp_performance(outbound) + + for socket <- [accepted, outbound] do + assert {:ok, opts} = :inet.getopts(socket, [:send_timeout, :send_timeout_close]) + assert Keyword.fetch!(opts, :send_timeout) > 0 + assert Keyword.fetch!(opts, :send_timeout_close) + end + end + test "compute_all_global_ips and format_ip cover runtime snapshot" do ips = Acceptor.compute_all_global_ips() assert Map.has_key?(ips, :inet6_all) diff --git a/test/connection_manager_test.exs b/test/connection_manager_test.exs index 37072ca..97ad445 100644 --- a/test/connection_manager_test.exs +++ b/test/connection_manager_test.exs @@ -851,6 +851,50 @@ defmodule Peer.ConnectionManagerTest do assert log =~ "ok=1" end + test "an in-flight batch does not block the next one, and the caps do" do + hash = :crypto.strong_rand_bytes(20) + pid = start_isolated_manager(hash) + on_exit(fn -> TestSupport.Sync.safe_stop(pid, 1_000) end) + + # A batch resolves only when its slowest endpoint does, up to ~55s. Serialising + # on that starved torrents whose queue yields only a few dialable endpoints. + peers = for n <- 1..8, do: %Peer{ip: {10, 0, 1, n}, port: 7000 + n} + queue = Enum.reduce(peers, %{}, &DialQueue.offer(&2, [&1], :discovery)) + + in_flight = MapSet.new(Enum.map(peers, &{&1.ip, &1.port})) + + # Every candidate is already being dialled by the batch in flight, so the + # second batch has nothing left to pick and must not redial them. + :sys.replace_state(pid, fn state -> + %{state | queue: queue, in_flight: in_flight, batches: 1, dialing?: false} + end) + + :ok = GenServer.cast(pid, :dial_now) + state = :sys.get_state(pid) + assert state.batches == 1 + assert MapSet.equal?(state.in_flight, in_flight) + + # At the batch cap nothing starts even with free candidates. + :sys.replace_state(pid, fn state -> + %{state | queue: queue, in_flight: MapSet.new(), batches: 3, dialing?: false} + end) + + :ok = GenServer.cast(pid, :dial_now) + state = :sys.get_state(pid) + assert state.batches == 3 + assert MapSet.size(state.in_flight) == 0 + + # Resolving a batch gives its slot back (empty queue, so nothing refills it). + :sys.replace_state(pid, fn state -> + %{state | queue: %{}, in_flight: in_flight, batches: 3} + end) + + send(pid, {:dial_done, [{{10, 0, 1, 1}, 7001}], {0, %{timeout: 1}, []}}) + state = :sys.get_state(pid) + assert state.batches == 2 + refute MapSet.member?(state.in_flight, {{10, 0, 1, 1}, 7001}) + end + test "dial_now while already dialing does not start a second batch" do hash = :crypto.strong_rand_bytes(20) pid = start_isolated_manager(hash) diff --git a/test/corrupt_piece_source_test.exs b/test/corrupt_piece_source_test.exs new file mode 100644 index 0000000..0c549a3 --- /dev/null +++ b/test/corrupt_piece_source_test.exs @@ -0,0 +1,188 @@ +defmodule CorruptPieceSourceTest do + @moduledoc """ + A peer that serves data failing the BEP 3 piece hash must be attributable and, + after repeated offences, dropped. Without that the piece picker hands the same + index straight back to the same peer and the torrent re-downloads it forever. + """ + + use ExUnit.Case, async: true + + alias Peer.Controller.State, as: ControllerState + alias Torrent.Downloads.Piece + + @piece_len 4 * Piece.max_length() + + defp piece_state(hash) do + %Piece.State{index: 7, hash: hash, waiting: []} + end + + defp with_blocks(%Piece.State{} = state, blocks) do + Enum.reduce(blocks, state, fn {peer_id, count}, %Piece.State{} = acc -> + %Piece.State{acc | contributors: Map.put(acc.contributors, peer_id, count)} + end) + end + + describe "attributing a failed piece" do + setup do + {:ok, hash: :crypto.strong_rand_bytes(20)} + end + + test "a piece assembled from one peer names that peer", %{hash: hash} do + state = with_blocks(piece_state(hash), [{"peer-a", 64}]) + + assert Piece.State.sole_contributor(state) == "peer-a" + end + + test "a piece assembled from several peers names none of them", %{hash: hash} do + state = with_blocks(piece_state(hash), [{"peer-a", 60}, {"peer-b", 4}]) + + assert Piece.State.sole_contributor(state) == nil + end + + test "a piece with no accepted blocks names no one", %{hash: hash} do + assert Piece.State.sole_contributor(piece_state(hash)) == nil + end + end + + describe "recording block sources" do + setup do + hash = :crypto.strong_rand_bytes(20) + dir = Path.join(System.tmp_dir!(), "corrupt_src_#{System.unique_integer([:positive])}") + File.mkdir_p!(dir) + on_exit(fn -> File.rm_rf(dir) end) + + torrent = %Torrent{ + hash: hash, + metadata: %{ + "info" => %{ + "name" => "t.bin", + "piece length" => @piece_len, + "length" => @piece_len, + "pieces" => :crypto.strong_rand_bytes(20) + } + }, + left: @piece_len, + last_index: 0, + last_piece_length: @piece_len, + download_dir: dir + } + + {:ok, model} = Torrent.Model.start_link(torrent) + :ok = Torrent.PiecesStatistic.init(torrent) + files = start_supervised!({Torrent.FileHandle, hash}) + + on_exit(fn -> + TestSupport.Sync.safe_stop(files, 1_000) + TestSupport.Sync.safe_stop(model, 1_000) + end) + + {:ok, hash: hash} + end + + test "an accepted block credits the peer that sent it", %{hash: hash} do + len = Piece.max_length() + block = :binary.copy(<<1>>, len) + + state = + %Piece.State{index: 0, hash: hash, waiting: [{0, len}, {len, len}], mode: nil} + |> Piece.State.response("peer-a", 0, block) + + assert state.contributors == %{"peer-a" => 1} + assert Piece.State.sole_contributor(state) == "peer-a" + + state = Piece.State.response(state, "peer-b", len, block) + + assert state.contributors == %{"peer-a" => 1, "peer-b" => 1} + assert Piece.State.sole_contributor(state) == nil + end + + test "a block for a subpiece nobody asked for credits no one", %{hash: hash} do + block = :binary.copy(<<1>>, Piece.max_length()) + + state = + %Piece.State{index: 0, hash: hash, waiting: [], mode: nil} + |> Piece.State.response("peer-a", 0, block) + + assert state.contributors == %{} + assert Piece.State.sole_contributor(state) == nil + end + + test "a block outside the piece credits no one", %{hash: hash} do + block = :binary.copy(<<1>>, Piece.max_length()) + + state = + %Piece.State{index: 0, hash: hash, waiting: [{@piece_len, Piece.max_length()}], mode: nil} + |> Piece.State.response("peer-a", @piece_len, block) + + assert state.contributors == %{} + end + end + + describe "peer strikes" do + setup do + hash = :crypto.strong_rand_bytes(20) + + state = %ControllerState{ + hash: hash, + id: Peer.id(), + fast_extension: nil, + status: nil, + pieces_count: 16, + socket: nil + } + + {:ok, state: state} + end + + test "enough distinct bad pieces disconnects the peer", %{state: state} do + limit = ControllerState.max_hash_failures() + + final = + Enum.reduce(1..(limit - 1), state, fn index, acc -> + assert %ControllerState{hash_failures: failures} = + next = ControllerState.hash_check_failed(acc, index) + + assert MapSet.size(failures) == index + next + end) + + assert {:error, :corrupt_pieces, %ControllerState{}} = + ControllerState.hash_check_failed(final, limit) + end + + test "a local race on one piece does not cost a good peer its connection", %{state: state} do + assert %ControllerState{hash_failures: failures} = + ControllerState.hash_check_failed(state, 3) + + assert MapSet.member?(failures, 3) + end + + test "a poisoned piece is never requested from that peer again", %{state: state} do + state = ControllerState.hash_check_failed(state, 226) + + unchoked = + ControllerState.handle_unchoke(%ControllerState{ + state + | status: 226, + interested: true, + choke_me: true + }) + + # The pin is dropped instead of re-requested, so the peer can move to a + # piece it has not already ruined. + assert unchoked.status == nil + end + + test "the same bad piece twice is still one bad piece", %{state: state} do + limit = ControllerState.max_hash_failures() + + final = + Enum.reduce(1..(limit + 2), state, fn _, acc -> + assert %ControllerState{} = next = ControllerState.hash_check_failed(acc, 226) + next + end) + + assert MapSet.to_list(final.hash_failures) == [226] + end + end +end diff --git a/test/dial_backoff_test.exs b/test/dial_backoff_test.exs index 2533b8e..a7388a6 100644 --- a/test/dial_backoff_test.exs +++ b/test/dial_backoff_test.exs @@ -31,21 +31,35 @@ defmodule Peer.DialBackoffTest do assert DialBackoff.filter([p], @hash, 0) == [] end - test "a churn block is NOT re-added even under min_count pressure" do - p = peer(2, 6882) - DialBackoff.record(@hash, p.ip, p.port, :churn) + test "a sticky block yields to every other candidate before it is re-dialled" do + churned = peer(2, 6882) + soft = peer(21, 6902) + free = peer(22, 6903) + + DialBackoff.record(@hash, churned.ip, churned.port, :churn) + DialBackoff.record(@hash, soft.ip, soft.port, :timeout) _ = :sys.get_state(DialBackoff) - assert DialBackoff.filter([p], @hash, 5) == [] - assert DialBackoff.filter([p], @hash, 0) == [] + # Room for two: the unblocked peer, then the soft block. Sticky stays out + # while anything else can fill the batch. + assert DialBackoff.filter([churned, soft, free], @hash, 2) == [free, soft] + + # Without min_count pressure only genuinely unblocked peers pass. + assert DialBackoff.filter([churned, soft, free], @hash, 0) == [free] end - test "a hard-failure block is sticky too" do - p = peer(3, 6883) - DialBackoff.record(@hash, p.ip, p.port, :econnrefused) + test "a sticky block is re-dialled when the torrent has nothing else to dial" do + churn = peer(2, 6882) + hard = peer(3, 6883) + + DialBackoff.record(@hash, churn.ip, churn.port, :churn) + DialBackoff.record(@hash, hard.ip, hard.port, :econnrefused) _ = :sys.get_state(DialBackoff) - assert DialBackoff.filter([p], @hash, 5) == [] + # Refusing the only candidates leaves the torrent with no dial at all, which + # costs more than the wasted SYNs. + assert DialBackoff.filter([churn, hard], @hash, 5) == [churn, hard] + assert DialBackoff.filter([churn, hard], @hash, 0) == [] end test "unblocked peers always pass through" do @@ -69,13 +83,70 @@ defmodule Peer.DialBackoffTest do assert DialBackoff.filter([p], @hash, 5) == [p] # The 3rd :timeout crosses @hard_fail_threshold and promotes the row to - # sticky. It must now be excluded even under aggressive min_count pressure. + # sticky: it now loses to any other candidate, and only comes back when the + # torrent would otherwise have nothing to dial. DialBackoff.record(@hash, p.ip, p.port, :timeout) _ = :sys.get_state(DialBackoff) - assert DialBackoff.filter([p], @hash, 5) == [] + other = peer(23, 6904) + assert DialBackoff.filter([p, other], @hash, 1) == [other] + assert DialBackoff.filter([p], @hash, 5) == [p] assert DialBackoff.filter([p], @hash, 0) == [] end + test "a live connection clears the endpoint's failure history" do + p = peer(24, 6905) + + for _ <- 1..3, do: DialBackoff.record(@hash, p.ip, p.port, :timeout) + _ = :sys.get_state(DialBackoff) + assert DialBackoff.blocked?(@hash, p.ip, p.port) + + # Connect + handshake settles the only question this table asks. Without + # this, a peer that keeps us choked never delivers bytes, so + # `mark_productive/3` never fires and the row survives the whole session. + DialBackoff.record_success(@hash, p.ip, p.port) + _ = :sys.get_state(DialBackoff) + + refute DialBackoff.blocked?(@hash, p.ip, p.port) + assert DialBackoff.filter([p], @hash, 0) == [p] + end + + test "failures far enough apart start a new streak instead of escalating" do + p = peer(25, 6906) + + for _ <- 1..2, do: DialBackoff.record(@hash, p.ip, p.port, :timeout) + _ = :sys.get_state(DialBackoff) + + # Age the last failure past the streak window. Two failures an hour apart are + # two incidents, not evidence of a dead endpoint — without this the counter + # only ever climbed and live rows reached fail_count 109. + key = {@hash, p.ip, p.port} + [{^key, blocked_until, retention, sticky?, count}] = fresh_row(key) + + :ets.insert( + :peer_dial_backoff, + {key, blocked_until, retention, sticky?, count, + System.monotonic_time(:millisecond) - 60 * 60 * 1_000} + ) + + DialBackoff.record(@hash, p.ip, p.port, :timeout) + _ = :sys.get_state(DialBackoff) + + # Third failure overall, first of this streak → still soft, so min_count + # pressure can pick it ahead of written-off endpoints. + other = peer(26, 6907) + DialBackoff.record(@hash, other.ip, other.port, :churn) + _ = :sys.get_state(DialBackoff) + + assert DialBackoff.filter([other, p], @hash, 1) == [p] + end + + defp fresh_row(key) do + [{^key, blocked_until, retention, sticky?, count, _last_at}] = + :ets.lookup(:peer_dial_backoff, key) + + [{key, blocked_until, retention, sticky?, count}] + end + test "non-reachability outcomes (already_connected / not_connectable) do not accumulate" do p = peer(6, 6886) diff --git a/test/downloads_piece_endgame_test.exs b/test/downloads_piece_endgame_test.exs index 90b4868..4a52cc8 100644 --- a/test/downloads_piece_endgame_test.exs +++ b/test/downloads_piece_endgame_test.exs @@ -1,7 +1,7 @@ defmodule DownloadsPieceEndgameTest do use ExUnit.Case, async: false - alias Torrent.Downloads.Piece.State + alias Torrent.Downloads.Piece.{Request, State} setup do {:ok, _} = Application.ensure_all_started(:elixir_torrent) @@ -28,6 +28,44 @@ defmodule DownloadsPieceEndgameTest do end) end + test "entering endgame re-opens blocks that are exclusively in flight" do + hash = :crypto.strong_rand_bytes(20) + peer_a = Peer.id() + peer_b = Peer.id() + noop = fn _index, _begin, _length -> :ok end + + with_model(sample_torrent(hash, 0), fn _ -> + # Normal mode with every block claimed by one peer. `waiting` is empty, so + # a second peer holding the piece has nothing it may ask for — one block + # belongs to one peer. That is correct mid-download and fatal on the last + # piece: if the holder stalls, the block is only retried on its own + # timeout, and the peers that could supply it just wait. + claimed = State.make({hash, 0}).waiting + requests = Enum.map(claimed, &%Request{peer_id: peer_a, subpiece: &1, timer: nil}) + + normal = + State.make({hash, 0}) + |> Map.merge(%{waiting: [], requests: requests, monitoring: %{peer_b => make_ref()}}) + + assert State.request(normal, peer_b, noop) == normal + + endgame = State.enter_endgame(normal) + + assert endgame.mode == :endgame + assert Enum.sort(endgame.waiting) == Enum.sort(claimed) + # The in-flight requests survive: endgame adds sources for a block, it + # does not take it away from the peer already fetching it. + assert endgame.requests == requests + + served = State.request(endgame, peer_b, noop) + assert Enum.any?(served.requests, &(&1.peer_id == peer_b)) + + # Level-triggered from the controller's reconcile pump, so it runs every + # couple of seconds for the whole endgame: it must not re-queue anything. + assert State.enter_endgame(endgame) == endgame + end) + end + defp sample_torrent(hash, last_index) do piece_len = 16 * 384 diff --git a/test/endgame_pin_monopoly_test.exs b/test/endgame_pin_monopoly_test.exs index aba793e..f265b9f 100644 --- a/test/endgame_pin_monopoly_test.exs +++ b/test/endgame_pin_monopoly_test.exs @@ -108,6 +108,67 @@ defmodule EndgamePinMonopolyTest do end end) end + + test "controller reconcile upgrades a piece started before the endgame transition" do + hash = :crypto.strong_rand_bytes(20) + torrent = mid_download_torrent(hash) + subpiece = {0, @piece_len} + + with_model(torrent, fn _ -> + start_swarm(hash) + start_downloads(hash) + Downloads.piece(hash, 0, fn -> :ok end, fn -> :ok end) + assert wait_active_pieces(hash, [0]) + + piece = Downloads.Piece.whereis(hash, 0) + refute Model.get(hash, :mode) == :endgame + assert :sys.get_state(piece).mode == nil + + # The live shape of the stall: every block of the piece claimed by one + # peer, so `waiting` is empty and no other peer may ask for those blocks + # — a piece worker only ever hands a block to a single peer outside + # endgame. The in-flight request also carries the worker past the orphan + # sweep this same reconcile runs. + :sys.replace_state(piece, fn state -> + %{ + state + | waiting: [], + timer: nil, + requests: [ + %Torrent.Downloads.Piece.Request{ + peer_id: @peer_endgame_b, + subpiece: subpiece, + timer: nil + } + ] + } + end) + + # The torrent crosses the endgame threshold while that worker is already + # running, which is the case `State.download/3` cannot see: it read the + # mode once, at start. + :sys.replace_state(model_via(hash), &%{&1 | left: @piece_len}) + assert Model.get(hash, :mode) == :endgame + + {:ok, controller} = GenServer.start(Torrent.Controller, hash) + + try do + send(controller, :reconcile_pump) + TestSupport.Sync.sync(controller) + TestSupport.Sync.sync(piece) + + state = :sys.get_state(piece) + + assert state.mode == :endgame + # Re-opened for a second source, without taking it from the peer that + # is already fetching it. + assert state.waiting == [subpiece] + assert length(state.requests) == 1 + after + safe_stop(controller) + end + end) + end end describe "endgame reject re-queue" do @@ -140,6 +201,8 @@ defmodule EndgamePinMonopolyTest do defp downloads_via(hash), do: {:via, Registry, {Registry, {hash, Torrent.Downloads}}} + defp model_via(hash), do: {:via, Registry, {Registry, {hash, Model}}} + defp endgame_torrent(hash) do %Torrent{ hash: hash, @@ -152,6 +215,12 @@ defmodule EndgamePinMonopolyTest do } end + # Same torrent, but with enough left that Model.get(:mode) is still nil — the + # state a piece worker is born into before the endgame transition. + defp mid_download_torrent(hash) do + %{endgame_torrent(hash) | left: 11 * @piece_len, last_index: 15} + end + defp both_pieces do Torrent.Bitfield.make(4) |> Torrent.Bitfield.set(0, 1) diff --git a/test/peer_controller_callbacks_test.exs b/test/peer_controller_callbacks_test.exs index 50621e3..ee3b0ea 100644 --- a/test/peer_controller_callbacks_test.exs +++ b/test/peer_controller_callbacks_test.exs @@ -253,7 +253,7 @@ defmodule PeerControllerCallbacksTest do end) end - test "fast-extension wire without negotiation stops with protocol_error" do + test "a reject for a block we are not waiting on keeps the connection" do hash = :crypto.strong_rand_bytes(20) id = <<4::160>> key = Peer.make_key(hash, id) @@ -264,7 +264,8 @@ defmodule PeerControllerCallbacksTest do ref = Process.monitor(ctrl_pid) assert :ok = Peer.Controller.handle_reject(key, 0, 0, @piece_len) - assert_receive {:DOWN, ^ref, :process, ^ctrl_pid, {:shutdown, :protocol_error}}, 2_000 + assert %{unsolicited_blocks: 1} = controller_state(key) + refute_receive {:DOWN, ^ref, :process, ^ctrl_pid, _}, 200 end) end diff --git a/test/peer_controller_coverage_batch_test.exs b/test/peer_controller_coverage_batch_test.exs index bac8080..7c251d5 100644 --- a/test/peer_controller_coverage_batch_test.exs +++ b/test/peer_controller_coverage_batch_test.exs @@ -558,7 +558,7 @@ defmodule PeerControllerCoverageBatchTest do end describe "Fast-extension guards and bootstrap leniency" do - test "non-negotiated Fast messages stop with protocol_error outside magnet bootstrap" do + test "advisory Fast messages are ignored, not punished, when Fast was not negotiated" do hash = :crypto.strong_rand_bytes(20) id = <<37::160>> key = Peer.make_key(hash, id) @@ -566,10 +566,38 @@ defmodule PeerControllerCoverageBatchTest do with_model(sample_torrent(hash, 4), fn _ -> {:ok, ctrl} = start_controller(hash, id, reserved_no_fast) - ref = Process.monitor(ctrl) + # Dropping the peer here also blacklists its id for every torrent, which + # is far past what an advisory message deserves. assert :ok = Peer.Controller.handle_suggest_piece(key, 0) - assert_receive {:DOWN, ^ref, :process, ^ctrl, {:shutdown, :protocol_error}}, @timeout + assert :ok = Peer.Controller.handle_allowed_fast(key, 0) + assert :ok = Peer.Controller.handle_reject(key, 0, 0, 16_384) + TestSupport.Sync.sync(ctrl) + + assert Process.alive?(ctrl) + refute Acceptor.BlackList.member?(id) + stop_quietly(ctrl) + end) + end + + test "have_all still counts when Fast was not negotiated" do + hash = :crypto.strong_rand_bytes(20) + id = <<47::160>> + key = Peer.make_key(hash, id) + reserved_no_fast = reserved_without_fast() + + with_model(sample_torrent(hash, 4), fn _ -> + {:ok, ctrl} = start_controller(hash, id, reserved_no_fast) + + # It says exactly what a full bitfield says, so there is nothing to gain + # from discarding it — and a seeder we record as having nothing is a + # connection we can never request from. + assert :ok = Peer.Controller.handle_have_all(key) + TestSupport.Sync.sync(ctrl) + + assert Process.alive?(ctrl) + assert :sys.get_state(ctrl).bitfield == :all + stop_quietly(ctrl) end) end diff --git a/test/peer_controller_state_test.exs b/test/peer_controller_state_test.exs index d09dd3c..f999bb0 100644 --- a/test/peer_controller_state_test.exs +++ b/test/peer_controller_state_test.exs @@ -246,15 +246,18 @@ defmodule PeerControllerStateTest do assert {:error, :protocol_error, ^state} = State.handle_have_none(state) end - test "non-negotiated have_all and have_none are rejected without mutating state" do + test "non-negotiated have_none is still applied" do hash = :crypto.strong_rand_bytes(20) - state = base_state(hash, 4, status: 0) - - assert {:stop, {:shutdown, :protocol_error}, ^state} = - Peer.Controller.handle_cast({:handle_have_all, []}, state) - assert {:stop, {:shutdown, :protocol_error}, ^state} = - Peer.Controller.handle_cast({:handle_have_none, []}, state) + # The message needs no Fast state to be read: it says what an empty bitfield + # says. Refusing it cost us the peer and blacklisted its id, for information + # we could simply have used. `have_all` takes the same path but also touches + # PiecesStatistic, so it is covered where a model is running. + assert {:noreply, %State{bitfield: :none}} = + Peer.Controller.handle_cast( + {:handle_have_none, []}, + base_state(hash, 4, status: 0) + ) end end @@ -388,6 +391,31 @@ defmodule PeerControllerStateTest do assert State.stale_useless_pin?(state) refute State.stale_useless_pin?(%{state | pin_downloaded_bytes: 1}) end + + test "stale_useless_pin? also releases an unchoked peer that delivers nothing" do + hash = :crypto.strong_rand_bytes(20) + now = System.monotonic_time(:millisecond) + + # The damaging case: a choked peer holds no requests, but an unchoked one + # keeps a full pipeline, so the blocks it sits on are unavailable to anyone + # else and merely time out every 30 s. Live, two such peers held one whole + # piece each with 0 and 48 KiB delivered while two pieces with all 64 blocks + # free had no peer at all, and the torrent stopped at 99%. + state = + base_state(hash, 4, + status: 0, + choke_me: false, + pin_downloaded_bytes: 0, + pinned_at: now - 65_000 + ) + + assert State.stale_useless_pin?(state) + + # A merely slow peer gets longer than the choked threshold, and longer than + # one block timeout, to produce its first block. + refute State.stale_useless_pin?(%{state | pinned_at: now - 30_000}) + refute State.stale_useless_pin?(%{state | pin_downloaded_bytes: 1}) + end end describe "reject, cancel, and piece accounting" do @@ -413,25 +441,109 @@ defmodule PeerControllerStateTest do end) end - test "handle_reject for unknown request is protocol_error" do + test "a reject for a block we cancelled is expected, not a protocol error" do hash = :crypto.strong_rand_bytes(20) - state = base_state(hash, 4, requests: MapSet.new()) - assert {:error, :protocol_error, ^state} = + # BEP 6 obliges the peer to answer every request exactly once, so the + # cancel we just sent comes back as this reject. Banning for it cost us + # precisely the peers that implement the extension correctly. + state = base_state(hash, 4, withdrawn: %{{0, 0, @piece_len} => 1}) + + assert %State{withdrawn: withdrawn, unsolicited_blocks: 0} = State.handle_reject(state, 0, 0, @piece_len) + + assert withdrawn == %{} end - test "handle_piece without matching request is protocol_error" do + test "a block still in flight when we cancelled is counted, not punished" do hash = :crypto.strong_rand_bytes(20) state = base_state(hash, 4, status: 0, - requests: MapSet.new([{0, 0, @piece_len}]) + withdrawn: %{{0, 0, @piece_len} => 1}, + rank: 0, + downloaded_bytes: 0 ) - assert {:error, :protocol_error, ^state} = - State.handle_piece(state, 0, @piece_len, @piece_len) + assert %State{ + rank: @piece_len, + downloaded_bytes: @piece_len, + withdrawn: withdrawn, + unsolicited_blocks: 0 + } = State.handle_piece(state, 0, 0, @piece_len) + + assert withdrawn == %{} + end + + test "a choke moves in-flight requests into the withdrawn window" do + hash = :crypto.strong_rand_bytes(20) + + with_model(sample_torrent(hash, 4), fn _ -> + {:ok, piece_pid} = start_piece_worker(hash, 0) + _peer = ensure_peer_registered(hash) + on_exit(fn -> stop_worker(piece_pid) end) + + state = base_state(hash, 4, status: 0, requests: MapSet.new([{0, 0, @piece_len}])) + + assert %State{requests: reqs, withdrawn: withdrawn} = State.handle_choke(state) + assert MapSet.size(reqs) == 0 + assert withdrawn == %{{0, 0, @piece_len} => 1} + end) + end + + test "the withdrawn window survives more withdrawals than it can hold" do + hash = :crypto.strong_rand_bytes(20) + + # A peer that is repinned repeatedly withdraws far more blocks than the + # window holds. Evicting by clearing it wholesale turned every answer + # still in flight into an "unsolicited" block; a live run produced 4392 of + # them in four minutes. Eviction has to drop the oldest, not everything. + state = + Enum.reduce(1..600, base_state(hash, 4, status: 0), fn n, acc -> + State.cancel(%{acc | requests: MapSet.new([{n, 0, @piece_len}])}, n, 0, @piece_len) + end) + + assert %State{unsolicited_blocks: 0} = State.handle_reject(state, 600, 0, @piece_len) + assert %State{unsolicited_blocks: 0} = State.handle_reject(state, 400, 0, @piece_len) + assert %State{unsolicited_blocks: 1} = State.handle_reject(state, 1, 0, @piece_len) + end + + test "a block withdrawn twice is owed two answers" do + hash = :crypto.strong_rand_bytes(20) + block = {7, 16_384, @piece_len} + + # Live logs showed the same block requested, cancelled and requested again + # up to seven times against one peer. A set forgets the block on the first + # answer, so every later one read as unsolicited. + state = base_state(hash, 4, status: 0, requests: MapSet.new([block])) + state = State.cancel(state, 7, 16_384, @piece_len) + state = State.cancel(%{state | requests: MapSet.new([block])}, 7, 16_384, @piece_len) + + assert %State{withdrawn: %{^block => 2}} = state + + assert %State{unsolicited_blocks: 0} = + state = State.handle_reject(state, 7, 16_384, @piece_len) + + assert %State{unsolicited_blocks: 0} = + state = State.handle_reject(state, 7, 16_384, @piece_len) + + assert %State{unsolicited_blocks: 1} = State.handle_reject(state, 7, 16_384, @piece_len) + end + + test "blocks we never asked for are counted and eventually end the connection" do + hash = :crypto.strong_rand_bytes(20) + state = base_state(hash, 4, status: 0) + + assert %State{unsolicited_blocks: 1} = + state = State.handle_piece(state, 0, 0, @piece_len) + + # Wasteful rather than malicious: the connection ends without the + # torrent-wide ban that :protocol_error carries. + state = %{state | unsolicited_blocks: 512} + + assert {:error, :unsolicited_blocks, %State{}} = + State.handle_piece(state, 0, 0, @piece_len) end test "handle_piece with matching request updates counters" do diff --git a/test/peer_endpoints_test.exs b/test/peer_endpoints_test.exs index 0a2ce84..cd71ac3 100644 --- a/test/peer_endpoints_test.exs +++ b/test/peer_endpoints_test.exs @@ -77,6 +77,66 @@ defmodule PeerEndpointsTest do TestSupport.Sync.await_down(new_monitor, new_peer, @timeout) end + @tag race_group: :endpoints + test "a noted disconnect reason replaces the supervisor's bare :shutdown in the log" do + endpoint = {{9, 9, 9, 7}, 6883} + {peer, monitor, _release} = TestSupport.Sync.spawn_blocked() + + :ok = Peer.Endpoints.register(@hash, elem(endpoint, 0), elem(endpoint, 1), peer) + + :ok = + Peer.Endpoints.note_disconnect_reason( + @hash, + elem(endpoint, 0), + elem(endpoint, 1), + {:shutdown, :no_mutual_interest} + ) + + TestSupport.Sync.sync(Peer.Endpoints) + + log = + ExUnit.CaptureLog.capture_log(fn -> + Process.exit(peer, :shutdown) + TestSupport.Sync.await_down(monitor, peer, @timeout) + TestSupport.Sync.sync(Peer.Endpoints) + end) + + assert log =~ "no_mutual_interest" + end + + @tag race_group: :endpoints + test "the note is dropped when the endpoint is registered again" do + endpoint = {{9, 9, 9, 6}, 6884} + {first, first_monitor, _} = TestSupport.Sync.spawn_blocked() + {second, second_monitor, _} = TestSupport.Sync.spawn_blocked() + + :ok = Peer.Endpoints.register(@hash, elem(endpoint, 0), elem(endpoint, 1), first) + + :ok = + Peer.Endpoints.note_disconnect_reason( + @hash, + elem(endpoint, 0), + elem(endpoint, 1), + {:shutdown, :protocol_error} + ) + + # The note describes the connection that just ended, so a fresh connection to + # the same endpoint must not inherit it. + Process.exit(first, :kill) + TestSupport.Sync.await_down(first_monitor, first, @timeout) + :ok = Peer.Endpoints.register(@hash, elem(endpoint, 0), elem(endpoint, 1), second) + + log = + ExUnit.CaptureLog.capture_log(fn -> + Process.exit(second, :shutdown) + TestSupport.Sync.await_down(second_monitor, second, @timeout) + TestSupport.Sync.sync(Peer.Endpoints) + end) + + refute log =~ "protocol_error" + assert log =~ "reason=:shutdown" + end + test "list/1 returns registered endpoints" do peer_a = spawn(fn -> diff --git a/test/state_machines_test.exs b/test/state_machines_test.exs index b4a9146..110ca7b 100644 --- a/test/state_machines_test.exs +++ b/test/state_machines_test.exs @@ -190,8 +190,7 @@ defmodule Peer.DialBackoffStateM do @spec model_filter(model(), [Peer.t()], non_neg_integer()) :: [Peer.t()] defp model_filter(state, peers, min_count) do {allowed, blocked} = split_allowed_peers(state, peers) - soft_blocked = filter_soft_blocked(state, blocked) - resolve_filtered_peers(state, allowed, soft_blocked, min_count) + resolve_filtered_peers(state, allowed, blocked, min_count) end defp split_allowed_peers(state, peers) do @@ -200,48 +199,43 @@ defmodule Peer.DialBackoffStateM do end) end - defp filter_soft_blocked(state, blocked) do - Enum.reject(blocked, fn %Peer{ip: ip, port: port} -> - match?(%{sticky: true}, Map.get(state.blocks, {ip, port})) - end) - end - - defp resolve_filtered_peers(_state, allowed, _soft_blocked, min_count) + defp resolve_filtered_peers(_state, allowed, _blocked, min_count) when min_count <= 0 do allowed end - defp resolve_filtered_peers(_state, allowed, soft_blocked, _min_count) - when soft_blocked == [] do + defp resolve_filtered_peers(_state, allowed, blocked, _min_count) + when blocked == [] do allowed end - defp resolve_filtered_peers(_state, allowed, _soft_blocked, min_count) + defp resolve_filtered_peers(_state, allowed, _blocked, min_count) when length(allowed) >= min_count do allowed end - defp resolve_filtered_peers(state, allowed, soft_blocked, min_count) do - need = min(min_count - length(allowed), length(soft_blocked)) - allowed ++ take_soft(state, soft_blocked, need) + defp resolve_filtered_peers(state, allowed, blocked, min_count) do + need = min(min_count - length(allowed), length(blocked)) + allowed ++ take_blocked(state, blocked, need) end - defp take_soft(_state, _blocked, 0), do: [] - - defp take_soft(state, blocked, need) do - {productive, rest} = - Enum.split_with(blocked, fn %Peer{ip: ip, port: port} -> - MapSet.member?(state.productive, {ip, port}) - end) + defp take_blocked(_state, _blocked, 0), do: [] - {v6, v4} = - Enum.split_with(rest, fn %Peer{ip: ip} -> - tuple_size(ip) == 8 - end) + # Mirrors Peer.DialBackoff: productive, then soft before sticky, then v6 before + # v4, then fewest failures. Sticky is last-resort rather than excluded, so a + # torrent with nothing dialable still gets a batch. + defp take_blocked(state, blocked, need) do + blocked + |> Enum.sort_by(fn %Peer{ip: ip, port: port} -> + block = Map.get(state.blocks, {ip, port}, %{fail_count: 0, sticky: false}) - productive - |> Kernel.++(v6) - |> Kernel.++(v4) + { + if(MapSet.member?(state.productive, {ip, port}), do: 0, else: 1), + if(block.sticky, do: 1, else: 0), + if(tuple_size(ip) == 8, do: 0, else: 1), + block.fail_count + } + end) |> Enum.take(need) end diff --git a/test/torrent_storage_coverage_batch_test.exs b/test/torrent_storage_coverage_batch_test.exs index f13942a..0213ebe 100644 --- a/test/torrent_storage_coverage_batch_test.exs +++ b/test/torrent_storage_coverage_batch_test.exs @@ -236,6 +236,73 @@ defmodule TorrentStorageCoverageBatchTest do end) end + @tag race_group: :uploader + test "a peer too slow to accept the block cancels the upload instead of crashing" do + piece0 = random_piece() + {torrent, _} = build_tiny_torrent([piece0]) + hash = torrent.hash + parent = self() + + # Never answers the call, so it times out — a peer whose socket has + # stopped draining. One slow peer with a deep request queue must not + # produce one task crash per in-flight block. + controller = spawn(fn -> Process.sleep(:infinity) end) + on_exit(fn -> Process.exit(controller, :kill) end) + + with_storage_stack(torrent, fn _ -> + write_piece!(hash, 0, piece0) + start_uploader_supervisor(hash) + + assert {:ok, task_pid} = + Uploader.request(hash, @peer_a, 0, 0, 128, fn block -> + send(parent, {:upload_callback_ready, self()}) + GenServer.call(controller, {:complete_upload, 0, 0, 128, block}, 50) + end) + + assert_receive {:upload_callback_ready, ^task_pid}, 2_000 + task_ref = Process.monitor(task_pid) + + assert_receive {:DOWN, ^task_ref, :process, ^task_pid, :normal}, 2_000 + assert Model.get(hash, :uploaded) == 0 + end) + end + + @tag race_group: :uploader + test "peer shutting down mid-call quietly cancels the upload" do + piece0 = random_piece() + {torrent, _} = build_tiny_torrent([piece0]) + hash = torrent.hash + parent = self() + + # A peer that exits with a reason of its own while the upload callback is + # blocked in GenServer.call — a protocol error, say. That is the peer + # going away, not an upload fault, so the task must end :normal rather + # than crash and log an [error] per in-flight block. + controller = + spawn(fn -> + receive do + {:"$gen_call", _from, _msg} -> exit({:shutdown, :protocol_error}) + end + end) + + with_storage_stack(torrent, fn _ -> + write_piece!(hash, 0, piece0) + start_uploader_supervisor(hash) + + assert {:ok, task_pid} = + Uploader.request(hash, @peer_a, 0, 0, 128, fn block -> + send(parent, {:upload_callback_ready, self()}) + GenServer.call(controller, {:complete_upload, 0, 0, 128, block}) + end) + + assert_receive {:upload_callback_ready, ^task_pid}, 2_000 + task_ref = Process.monitor(task_pid) + + assert_receive {:DOWN, ^task_ref, :process, ^task_pid, :normal}, 2_000 + assert Model.get(hash, :uploaded) == 0 + end) + end + @tag race_group: :uploader test "peer teardown during completion quietly cancels the upload" do piece0 = random_piece() @@ -646,6 +713,170 @@ defmodule TorrentStorageCoverageBatchTest do end) end + test "assign_peer_to_piece?/3 will not move a pinned peer onto a drained piece" do + hash = :crypto.strong_rand_bytes(20) + + with_model(drained_pin_torrent(hash), fn _ -> + key = drained_piece_with_peer(hash, @peer_b, index: 0) + + refute Swarm.assign_peer_to_piece?(hash, key, 1) + end) + end + + test "assign_peer_to_piece?/3 still gives an unpinned peer a drained piece" do + hash = :crypto.strong_rand_bytes(20) + + with_model(drained_pin_torrent(hash), fn _ -> + key = drained_piece_with_peer(hash, @peer_b, index: nil) + + # A peer with no pin never reaches check_interested/1, so it would never + # send BEP 3 `interested` and never be unchoked. It gets the pin anyway. + assert Swarm.assign_peer_to_piece?(hash, key, 1) + end) + end + + test "assign_peer_to_piece?/3 frees a pin whose blocks are all in flight elsewhere" do + hash = :crypto.strong_rand_bytes(20) + + with_model(drained_pin_torrent(hash), fn _ -> + start_swarm(hash) + start_downloads(hash) + + # Piece 0: every block claimed by another peer, none delivered yet. + Downloads.piece(hash, 0, fn -> :ok end, fn -> :ok end) + + :sys.replace_state(Piece.whereis(hash, 0), fn state -> + %{ + state + | waiting: [], + requests: [%Request{peer_id: @peer_a, subpiece: {0, 16_384}, timer: nil}] + } + end) + + assert Downloads.piece_has_waiting?(hash, 0) + refute Downloads.piece_has_unclaimed?(hash, 0) + + Downloads.piece(hash, 1, fn -> :ok end, fn -> :ok end) + + {_pid, key} = + add_swarm_peer(hash, @peer_b, + index: 0, + bitfield: drained_pin_bitfield(), + choke_me: false, + stale: false + ) + + # There is nothing left on piece 0 to hand this peer, so it must be + # allowed onto piece 1 rather than idling until the in-flight blocks + # of a different peer resolve. + assert Swarm.assign_peer_to_piece?(hash, key, 1) + end) + end + + test "assign_peer_to_piece?/3 keeps a peer whose own blocks are the ones in flight" do + hash = :crypto.strong_rand_bytes(20) + + with_model(drained_pin_torrent(hash), fn _ -> + start_swarm(hash) + start_downloads(hash) + + # Same setup as the test above, except the in-flight block belongs to the + # pinned peer itself. Moving it would cancel the very requests that made + # the piece look drained, returning them to `waiting` — so the next + # reconcile tick pins the peer straight back. Live, that oscillated at the + # 2 s tick and the peer delivered nothing at all. + Downloads.piece(hash, 0, fn -> :ok end, fn -> :ok end) + + :sys.replace_state(Piece.whereis(hash, 0), fn state -> + %{ + state + | waiting: [], + requests: [%Request{peer_id: @peer_b, subpiece: {0, 16_384}, timer: nil}] + } + end) + + refute Downloads.piece_has_unclaimed?(hash, 0) + assert Downloads.piece_serves_peer?(hash, 0, @peer_b) + refute Downloads.piece_serves_peer?(hash, 0, @peer_a) + + Downloads.piece(hash, 1, fn -> :ok end, fn -> :ok end) + + {_pid, key} = + add_swarm_peer(hash, @peer_b, + index: 0, + bitfield: drained_pin_bitfield(), + choke_me: false, + stale: false + ) + + refute Swarm.assign_peer_to_piece?(hash, key, 1) + end) + end + + test "assign_peer_to_piece?/3 frees an endgame pin that cannot serve this peer" do + hash = :crypto.strong_rand_bytes(20) + active = [0, 1, 2] + + # Endgame spreads peers across the remaining indices by hashing the peer id, + # so the index this peer is entitled to is known up front — and pinning it + # elsewhere is what makes this exercise a move rather than the same-index + # shortcut. + preferred = Enum.at(active, :erlang.phash2(@peer_b, length(active))) + pin = Enum.find(active, &(&1 != preferred)) + bf = Enum.reduce(active, Bitfield.make(4), fn i, acc -> Bitfield.set(acc, i, 1) end) + + # A torrent enters endgame while the workers it already started keep running + # in normal mode, and a normal-mode worker never hands out a block that is + # already in flight. Judging the pin by the torrent's mode read those + # in-flight blocks — someone else's — as a reason to hold this peer, which + # parked a live torrent at 99.18% with 8 of 10 peers on one such piece while + # six pieces sat untouched. + with_model(endgame_torrent(hash), fn _ -> + assert Torrent.Model.get(hash, :mode) == :endgame + + start_swarm(hash) + start_downloads(hash) + Downloads.piece(hash, pin, fn -> :ok end, fn -> :ok end) + + piece_pid = Piece.whereis(hash, pin) + assert is_pid(piece_pid) + # `Downloads.piece/4` fills the worker asynchronously; without this barrier + # the overwrite below races that fill. + TestSupport.Sync.sync(piece_pid) + + :sys.replace_state(piece_pid, fn state -> + %{ + state + | mode: nil, + waiting: [], + requests: [%Request{peer_id: @peer_a, subpiece: {0, 16_384}, timer: nil}] + } + end) + + refute Downloads.piece_serves_peer?(hash, pin, @peer_b) + assert Downloads.piece_serves_peer?(hash, pin, @peer_a) + + {_pid, key_b} = + add_swarm_peer(hash, @peer_b, index: pin, bitfield: bf, choke_me: false, stale: false) + + {_pid, key_a} = + add_swarm_peer(hash, @peer_a, index: pin, bitfield: bf, choke_me: false, stale: false) + + targets = active -- [pin] + + # Freed, and onto its own index only. Endgame accepts a re-pin onto any + # active piece, so without the per-peer gate every freed peer would follow + # `reconcile_refresh_interest/4` onto whichever index it reaches first, + # trading one crowded piece for another. + assert Enum.filter(targets, &Swarm.assign_peer_to_piece?(hash, key_b, &1, active)) == + [preferred] + + # The peer whose own requests emptied the piece is the one working there; + # moving it would cancel exactly those requests. + assert Enum.filter(targets, &Swarm.assign_peer_to_piece?(hash, key_a, &1, active)) == [] + end) + end + test "sort_peers_seeders_first ranks seeders ahead of leechers" do hash = :crypto.strong_rand_bytes(20) torrent = endgame_torrent(hash) @@ -743,6 +974,41 @@ defmodule TorrentStorageCoverageBatchTest do end) end + test "each pump trigger cancels the previous pending {:next_piece} wake" do + hash = :crypto.strong_rand_bytes(20) + torrent = sample_torrent(hash, 4) + + with_model(torrent, fn _ -> + # No peers, so the handler takes the connecting_to_peers branch and arms + # a retry every time — the shape that used to accumulate one immortal + # chain per trigger. + start_swarm(hash) + start_downloads(hash) + + pending_wake = fn pid -> + {:dictionary, dict} = Process.info(pid, :dictionary) + Keyword.get(dict, :next_piece_timer) + end + + with_controller(hash, fn pid -> + send(pid, {:next_piece, :rare}) + TestSupport.Sync.sync(pid) + first = pending_wake.(pid) + + assert is_reference(first) + assert Process.read_timer(first) + + send(pid, {:next_piece, :rare}) + TestSupport.Sync.sync(pid) + second = pending_wake.(pid) + + assert is_reference(second) + refute second == first + refute Process.read_timer(first) + end) + end) + end + test ":reconcile_pump kicks next_piece when peers exist and capacity remains" do hash = :crypto.strong_rand_bytes(20) torrent = sample_torrent(hash, 4) @@ -881,6 +1147,36 @@ defmodule TorrentStorageCoverageBatchTest do end) end + test "orphan_no_sources?/1 is true when the swarm has peers but none has this piece" do + hash = :crypto.strong_rand_bytes(20) + torrent = sample_torrent(hash, 2) + + with_model(torrent, fn _ -> + start_swarm(hash) + # Holds piece 1 only, so the worker on piece 0 has no source and must + # release its @max_parallel_pieces slot rather than idle on it forever. + bf = Bitfield.make(2) |> Bitfield.set(1, 1) + add_swarm_peer(hash, @peer_a, index: nil, bitfield: bf) + + state = %State{hash: hash, index: 0, waiting: [{0, 16_384}], monitoring: %{}} + assert State.orphan_no_sources?(state) + end) + end + + test "orphan_no_sources?/1 is false while a connected peer has this piece" do + hash = :crypto.strong_rand_bytes(20) + torrent = sample_torrent(hash, 2) + + with_model(torrent, fn _ -> + start_swarm(hash) + bf = Bitfield.make(2) |> Bitfield.set(0, 1) + add_swarm_peer(hash, @peer_a, index: nil, bitfield: bf) + + state = %State{hash: hash, index: 0, waiting: [{0, 16_384}], monitoring: %{}} + refute State.orphan_no_sources?(state) + end) + end + test "release_in_flight_requests/1 cancels tracked requests" do hash = :crypto.strong_rand_bytes(20) subpiece = {0, 16_384} @@ -1286,6 +1582,52 @@ defmodule TorrentStorageCoverageBatchTest do } end + # 20 pieces keeps `left` above Model's `@until_endgame * piece_length`, so the + # torrent is NOT in endgame — endgame deliberately allows several peers onto + # the same drained piece, which is the opposite of what these tests check. + @drained_pin_pieces 20 + + defp drained_pin_torrent(hash) do + %Torrent{ + hash: hash, + metadata: %{"info" => %{"name" => "drained-target", "piece length" => 16_384}}, + left: @drained_pin_pieces * 16_384, + last_index: @drained_pin_pieces - 1, + last_piece_length: 16_384, + peer_status: nil + } + end + + # Starts piece 1, empties its waiting list, and adds a peer holding both + # pieces 0 and 1. Emptying via `:sys.replace_state` rather than a real + # `Downloads.request/4` keeps it deterministic — a handed-out subpiece is + # re-queued the moment its request timer fires. + defp drained_piece_with_peer(hash, peer_id, opts) do + start_swarm(hash) + start_downloads(hash) + Downloads.piece(hash, 1, fn -> :ok end, fn -> :ok end) + + piece_pid = Piece.whereis(hash, 1) + assert is_pid(piece_pid) + :sys.replace_state(piece_pid, &%{&1 | waiting: []}) + refute Downloads.piece_has_waiting?(hash, 1) + + {_pid, key} = + add_swarm_peer( + hash, + peer_id, + Keyword.merge([bitfield: drained_pin_bitfield(), choke_me: false, stale: false], opts) + ) + + key + end + + defp drained_pin_bitfield do + Bitfield.make(@drained_pin_pieces) + |> Bitfield.set(0, 1) + |> Bitfield.set(1, 1) + end + defp both_pieces do Bitfield.make(4) |> Bitfield.set(0, 1)