feat: add peer and tracker management - #258
Open
artrixdotdev wants to merge 2 commits into
Open
Conversation
Contributor
WalkthroughTorrent live APIs now support manual peer management and tracker lifecycle controls. Actor message flows return typed results, tracker registration is supervised and protocol-checked, live hub lookups resolve tracker relationships, and integration tests cover peer and tracker operations. ChangesPeer and tracker lifecycle
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related issues
Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
coderabbitai
Bot
added
enhancement
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
crates/libtortillas/src/torrent/messages.rs (1)
249-271: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
tell(...).awaitper tracker can stall the torrent actor.
force_reannounceruns inside the torrent actor's handler and awaits each tracker's bounded mailbox sequentially, so one saturated tracker mailbox blocks the whole torrent loop (and every other command behind it). The sibling broadcast paths already usefor_each_concurrentortry_sendfor this reason. Considertry_send(counting refusals as failures) or a concurrent send bounded bytracker_broadcast_concurrency.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/libtortillas/src/torrent/messages.rs` around lines 249 - 271, The force_reannounce handler must not await each tracker mailbox sequentially. Update force_reannounce to use the existing non-blocking or bounded-concurrency tracker broadcast pattern, such as try_send or tracker_broadcast_concurrency, while preserving queued-counting and first-error reporting for failed sends.crates/libtortillas/src/torrent/handle.rs (1)
271-279: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUnbounded
listener.recv()in tests.Both new tests await
listener.recv()without atimeout(...)in a couple of places (here and in the tracker test's firstrecvs they do use timeouts). An unmet event hangs the test until the harness-wide limit instead of failing fast. Wrapping this one intimeoutkeeps the suite diagnosable.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/libtortillas/src/torrent/handle.rs` around lines 271 - 279, Wrap the await on listener.recv() in the disconnect-peer test with the existing timeout mechanism, preserving the current Disconnected event assertion while making a missing event fail promptly; apply the same bounded receive to the other new unbounded listener.recv() calls in the tracker test.crates/libtortillas/src/lib.rs (1)
570-575: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDocument why the fixture never returns.
std::future::pending()deliberately keeps the connection open so the peer under test doesn't observe an immediate EOF after the scripted messages; cleanup relies onLocalPeer::dropaborting the accept task and, with it, theJoinSet. That contract is non-obvious — worth a one-line comment so a future reader doesn't "fix" it back toOk(()).📝 Suggested comment
+ // Keep the connection open so the client does not see an EOF right after + // the scripted messages. `LocalPeer::drop` aborts this task. std::future::pending().await🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/libtortillas/src/lib.rs` around lines 570 - 575, Add a concise comment immediately before std::future::pending().await in the fixture’s message-stream handling to document that it intentionally keeps the connection open, preventing EOF until LocalPeer::drop aborts the accept task and its JoinSet. Do not change the existing control flow.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/libtortillas/src/torrent/swarm.rs`:
- Around line 46-91: Bound the spawned connection future in the add-peer flow
around PeerStream::connect, send_handshake, and recv_handshake_message using the
configured peer-handshake timeout. Convert expiration into the established typed
TorrentError timeout variant so the awaited add_peer request and discovery task
terminate predictably, while preserving existing handshake and registration
errors.
---
Nitpick comments:
In `@crates/libtortillas/src/lib.rs`:
- Around line 570-575: Add a concise comment immediately before
std::future::pending().await in the fixture’s message-stream handling to
document that it intentionally keeps the connection open, preventing EOF until
LocalPeer::drop aborts the accept task and its JoinSet. Do not change the
existing control flow.
In `@crates/libtortillas/src/torrent/handle.rs`:
- Around line 271-279: Wrap the await on listener.recv() in the disconnect-peer
test with the existing timeout mechanism, preserving the current Disconnected
event assertion while making a missing event fail promptly; apply the same
bounded receive to the other new unbounded listener.recv() calls in the tracker
test.
In `@crates/libtortillas/src/torrent/messages.rs`:
- Around line 249-271: The force_reannounce handler must not await each tracker
mailbox sequentially. Update force_reannounce to use the existing non-blocking
or bounded-concurrency tracker broadcast pattern, such as try_send or
tracker_broadcast_concurrency, while preserving queued-counting and first-error
reporting for failed sends.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 688a3e94-4158-421e-972b-63eb0954e6fe
📒 Files selected for processing (10)
crates/libtortillas/src/errors.rscrates/libtortillas/src/facade.rscrates/libtortillas/src/lib.rscrates/libtortillas/src/live/hub.rscrates/libtortillas/src/torrent/actor.rscrates/libtortillas/src/torrent/handle.rscrates/libtortillas/src/torrent/messages.rscrates/libtortillas/src/torrent/mod.rscrates/libtortillas/src/torrent/piece_flow.rscrates/libtortillas/src/torrent/swarm.rs
Comment on lines
46
to
+91
| tokio::spawn(async move { | ||
| let mut id = peer.id; | ||
| let (stream, reserved) = match stream { | ||
| Some((mut stream, reserved)) => { | ||
| let handshake = Handshake::new(info_hash, our_id); | ||
| if let Err(err) = stream.send(PeerMessages::Handshake(handshake)).await { | ||
| debug!(error = %err, peer_addr = %peer.socket_addr(), "Failed to send handshake to peer"); | ||
| return; | ||
| let connection = async { | ||
| let (stream, reserved) = match stream { | ||
| Some((mut stream, reserved)) => { | ||
| let handshake = Handshake::new(info_hash, our_id); | ||
| stream.send(PeerMessages::Handshake(handshake)).await?; | ||
| (stream, reserved) | ||
| } | ||
| (stream, reserved) | ||
| } | ||
| None => { | ||
| let stream = PeerStream::connect(peer.socket_addr(), Some(utp_server)).await; | ||
| match stream { | ||
| Ok(mut stream) => match stream.send_handshake(our_id, info_hash).await { | ||
| Ok(_) => match stream.recv_handshake_message().await { | ||
| Ok(handshake) => { | ||
| if let Err(err) = | ||
| validate_handshake(&handshake, peer.socket_addr(), info_hash) | ||
| { | ||
| trace!(error = %err, peer_addr = %peer.socket_addr(), "Failed to validate peer handshake; exiting"); | ||
| return; | ||
| } | ||
| id = Some(handshake.peer_id); | ||
| (stream, handshake.reserved) | ||
| } | ||
| Err(err) => { | ||
| trace!(error = %err, peer_addr = %peer.socket_addr(), "Failed to receive handshake from peer; exiting"); | ||
| return; | ||
| } | ||
| }, | ||
| Err(err) => { | ||
| trace!(error = %err, peer_addr = %peer.socket_addr(), "Failed to send handshake to peer; exiting"); | ||
| return; | ||
| } | ||
| }, | ||
| Err(err) => { | ||
| trace!(error = %err, peer_addr = %peer.socket_addr(), "Failed to connect to peer; exiting"); | ||
| return; | ||
| } | ||
| None => { | ||
| let mut stream = | ||
| PeerStream::connect(peer.socket_addr(), Some(utp_server)).await?; | ||
| stream.send_handshake(our_id, info_hash).await?; | ||
| let handshake = stream.recv_handshake_message().await?; | ||
| validate_handshake(&handshake, peer.socket_addr(), info_hash)?; | ||
| id = Some(handshake.peer_id); | ||
| (stream, handshake.reserved) | ||
| } | ||
| }; | ||
|
|
||
| let id = id.ok_or_else(|| TorrentError::InvalidOperation { | ||
| operation: "add peer", | ||
| reason: "peer connection completed without a peer id".to_string(), | ||
| })?; | ||
|
|
||
| if id == our_id { | ||
| return Err(TorrentError::InvalidOperation { | ||
| operation: "add peer", | ||
| reason: "a torrent cannot connect to its own peer id".to_string(), | ||
| }); | ||
| } | ||
| }; | ||
|
|
||
| let Some(id) = id else { | ||
| trace!(peer_addr = %peer.socket_addr(), "Peer connection completed without a peer id; exiting"); | ||
| return; | ||
| }; | ||
| peer.id = Some(id); | ||
|
|
||
| if id == our_id { | ||
| return; | ||
| // Registration is the boundary where the connection becomes part | ||
| // of the swarm, so a manual add does not succeed before this ask. | ||
| actor_ref | ||
| .ask(PeerConnected { | ||
| peer, | ||
| reserved, | ||
| stream, | ||
| }) | ||
| .await | ||
| .map_err(|error| map_torrent_send_error("register connected peer", error)) | ||
| } | ||
| .await; |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Unbounded connect/handshake makes Torrent::add_peer hang indefinitely.
The spawned connection future has no timeout around PeerStream::connect, send_handshake, or recv_handshake_message. For discovery this only leaks a task, but the new public add_peer awaits this oneshot, so a black-holed address leaves the caller (and the TUI) blocked until the OS TCP timeout — or forever for a peer that accepts and never sends a handshake (which is exactly what the LocalPeer fixture now does after std::future::pending()). Consider bounding the connection future with a configured peer-handshake timeout and surfacing a typed timeout error.
🛡️ Sketch
- let connection = async {
+ let connection = match tokio::time::timeout(handshake_timeout, async {
...
- }
- .await;
+ })
+ .await
+ {
+ Ok(result) => result,
+ Err(_) => Err(TorrentError::InvalidOperation {
+ operation: "add peer",
+ reason: "peer handshake timed out".to_string(),
+ }),
+ };🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/libtortillas/src/torrent/swarm.rs` around lines 46 - 91, Bound the
spawned connection future in the add-peer flow around PeerStream::connect,
send_handshake, and recv_handshake_message using the configured peer-handshake
timeout. Convert expiration into the established typed TorrentError timeout
variant so the awaited add_peer request and discovery task terminate
predictably, while preserving existing handshake and registration errors.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
TorrentAPIs to add and disconnect peers, add and remove trackers, and queue per-tracker or torrent-wide reannouncesTorrentas the sole command surface while peer and tracker handles provide stable identities and live status/event accessAPI boundary
Peer connection succeeds after the handshake and swarm registration boundary. Tracker registration and announce commands follow the existing actor model: command delivery is guaranteed, while tracker initialization and network responses remain asynchronous and observable through
TrackerHandleviews and listeners. Tracker removal immediately updates the authoritative registry and live terminal state; its stopped announce is best-effort.The implementation reuses the existing peer removal path, tracker actor construction, live scope registry, and tracker status/event projection. Inline comments document only the non-obvious handshake, shared tracker-registration, and shutdown flows.
Closes #226
Validation
cargo fmt --all -- --checkcargo clippy --workspace --all-targets --all-features -- -D warningscargo clippy -p libtortillas --no-default-features --all-targets -- -D warningscargo nextest run --workspace --all-features— 194 passed, 9 skippedcargo nextest run --workspace --all-features --run-ignored only— 9 passedcargo test -p libtortillas --all-features --doc— 28 passed, 6 ignoredRUSTDOCFLAGS="-D warnings" cargo doc --workspace --all-features --no-depsbuild_and_testandlint