RUST-PAR-017: release reask leases by the TCP key, not the UDP endpoint

DL-11: ReaskEvent::SourceReleased carried the peer's UDP routing endpoint
(ip, udp_port), but the consumer release_direct_download_source_leases
(crates/emulebb-core/src/lib.rs) matches it against TCP-endpoint-keyed sets:
active_download_peer_endpoints and the download source registry leases are
keyed by source_endpoint_key = (ip, tcp_port), and the detach-exclusion path
retains the lease under that same TCP key. Since a peer's UDP port normally
differs from its TCP port (4672 vs 4662), every release raised on the reask
drop paths (FNF DropSource, low-rank re-engage, Remove command, RetryTcp
timeout) was a silent no-op: the lease leaked, acquire_* deferred the
endpoint forever, and the source could never be re-engaged until
disconnect_ed2k's full lease reset.

Fix: carry core's lease key through the detach registration and address all
release events by it.

- ReaskDetachArgs gains tcp_port (crates/emulebb-ed2k/src/ed2k_client_udp/
  runtime.rs); both producers set it (the TCP session detach passes the
  connected socket's remote port in ed2k_tcp/download/session/reask_detach.rs,
  the Kad buddy path passes source.tcp_port in
  crates/emulebb-core/src/ed2k_buddy_reask.rs).
- ReaskSource stores lease_endpoint = (ip, tcp_port) (ed2k_client_udp/
  state.rs); ReaskService::remove_source returns it and RoutedReply carries
  it (service.rs, source_set.rs), so every SourceReleased on the four drop
  paths now addresses the lease core actually holds.
- SourceDead is unchanged: it keeps the UDP endpoint and core's (ip, file)
  registry resolver, which keys on the IP only.
- NNP keying audited and left as is: registry NNP holds are (ip, tcp_port)
  in the registry keyspace, and the MarkNoNeededParts loop command is
  (ip, udp_port) in the loop keyspace - both consistent.

Tests pin the contract with distinct UDP/TCP ports on every path (runtime,
service, detach-args producers), plus a core test asserting a UDP endpoint
never matches the TCP-keyed lease sets while source_endpoint_key frees them.
Pure internal bookkeeping; no wire behavior changes.

Author: Threepwood-7

crates/emulebb-core/src/ed2k_buddy_reask.rs

@@ -62,6 +62,10 @@ pub(crate) fn detach_kad_buddy_sources_for_reask(
         let registered = reask_handle.register_kad_buddy_source(ReaskDetachArgs {
             file_hash,
             endpoint,
+            // Core's lease key port (source_endpoint_key). A buddy source holds
+            // no direct-download lease, but its release events must still be
+            // addressed consistently by the TCP key.
+            tcp_port: source.tcp_port,
             udp_version: ED2K_DEFAULT_UDP_VERSION,
             // Kad buddy sources have no direct TCP file request to timestamp, so
             // the first buddy-relayed UDP reask is due immediately.
@@ -155,6 +159,7 @@ mod tests {
                 assert_eq!(args.file_hash, file_hash);
                 assert!(args.low_id);
                 assert_eq!(args.endpoint, (Ipv4Addr::new(192, 0, 2, 77), 4672));
+                assert_eq!(args.tcp_port, 4662);
                 assert_eq!(args.buddy_id, Some([0x5a; 16]));
                 assert_eq!(
                     args.buddy_endpoint,

crates/emulebb-core/src/ed2k_net_drivers.rs

@@ -296,7 +296,9 @@ async fn handle_reask_event(
             // download cycle — or the SourceReady that follows — can re-acquire
             // and reconnect this endpoint over TCP. Without this the endpoint
             // stays leased forever and acquire_direct_download_source_leases
-            // defers it, leaking the lease and killing re-engage.
+            // defers it, leaking the lease and killing re-engage. `endpoint` is
+            // the source's TCP endpoint (the loop carries core's lease key,
+            // RUST-PAR-017 DL-11), so it matches these TCP-keyed sets directly.
             core.release_direct_download_source_leases(&[endpoint])
                 .await;
         }

crates/emulebb-core/src/lib.rs

@@ -11249,6 +11249,68 @@ mod tests {
         assert!(re_retry_delay.is_some());
     }
 
+    #[tokio::test]
+    async fn lease_release_is_tcp_keyed_so_a_udp_endpoint_never_matches() {
+        // RUST-PAR-017 DL-11: core's lease sets (active_download_peer_endpoints +
+        // the registry leased peers) are keyed by (ip, tcp_port), while the UDP
+        // reask loop routes sources by (ip, udp_port). A SourceReleased carrying
+        // the UDP endpoint therefore releases NOTHING — the lease leaks and the
+        // source can never be re-engaged. This pins the constraint that forced
+        // the loop to carry the TCP lease key in its release events.
+        let core = EmulebbCore::new_in_memory("test", FileIndex::in_memory().unwrap()).unwrap();
+        let file_hash = Ed2kHash::from_bytes([0x5b; 16]).to_string();
+        let ip = Ipv4Addr::new(192, 0, 2, 60);
+        let tcp_port = 4662u16;
+        let udp_port = 4672u16;
+        let source = direct_test_source(Ed2kHash::from_bytes([0x5b; 16]), ip, tcp_port);
+        {
+            let mut state = core.state.lock().await;
+            state.download_source_registry.add_candidate(
+                Instant::now(),
+                DownloadSourceCandidate {
+                    file_hash: file_hash.clone(),
+                    file_priority: 5,
+                    needed_parts: 4,
+                    rare_parts: 0,
+                    source: source.clone(),
+                    last_seen: Instant::now(),
+                },
+            );
+        }
+        let (engaged, _, _) = core
+            .acquire_direct_download_source_leases(&file_hash, std::slice::from_ref(&source))
+            .await;
+        assert_eq!(engaged, vec![source.clone()]);
+
+        // Releasing by the peer's UDP endpoint (what the reask loop routes on)
+        // must not free the TCP-keyed lease — the endpoints live in different
+        // keyspaces, so this is a no-op by construction.
+        core.release_direct_download_source_leases(&[(ip, udp_port)])
+            .await;
+        {
+            let state = core.state.lock().await;
+            assert_eq!(
+                state.active_download_peer_endpoints.len(),
+                1,
+                "a UDP endpoint must not match the TCP-keyed active set"
+            );
+            assert_eq!(
+                state.download_source_registry.leased_peer_count(),
+                1,
+                "a UDP endpoint must not match the TCP-keyed registry lease"
+            );
+        }
+
+        // Releasing by the TCP lease key (what SourceReleased now carries) frees it.
+        core.release_direct_download_source_leases(&[source_endpoint_key(&source)])
+            .await;
+        {
+            let state = core.state.lock().await;
+            assert!(state.active_download_peer_endpoints.is_empty());
+            assert_eq!(state.download_source_registry.leased_peer_count(), 0);
+        }
+    }
+
     #[tokio::test]
     async fn run_attempt_stops_immediately_when_pre_cancelled() {
         // The requery loop checks the per-hash cancel token at the top of each

crates/emulebb-ed2k/src/ed2k_client_udp/runtime.rs

@@ -90,6 +90,12 @@ pub type ReaskCommandReceiver = mpsc::Receiver<ReaskCommand>;
 pub struct ReaskDetachArgs {
     pub file_hash: Ed2kHash,
     pub endpoint: (Ipv4Addr, u16),
+    /// The source's eD2k **TCP** port. Core's lease bookkeeping
+    /// (`active_download_peer_endpoints` + the download source registry) is keyed
+    /// by `(ip, tcp_port)` while `endpoint` is the UDP routing key, so the loop
+    /// must address `SourceReleased` events by `(endpoint.ip, tcp_port)` or the
+    /// release never matches and the lease leaks (RUST-PAR-017 DL-11).
+    pub tcp_port: u16,
     pub udp_version: u8,
     /// Delay before the first detached UDP reask is due.
     ///
@@ -179,6 +185,10 @@ pub enum ReaskEvent {
     /// registry. The loop raises this whenever it drops a detached source, so core
     /// frees the lease and a later cycle (or the following `SourceReady`) reconnects
     /// it over TCP. Must precede the `SourceReady` for the same source.
+    ///
+    /// `endpoint` is the source's **TCP** endpoint (`ReaskDetachArgs::tcp_port`) —
+    /// the key core leased it under — NOT the UDP endpoint the loop routes on;
+    /// core matches it directly against its TCP-keyed lease sets.
     SourceReleased { endpoint: (Ipv4Addr, u16) },
     /// A detached source UDP-answered `OP_FILENOTFOUND` for `file_hash` (oracle
     /// `CUpDownClient::UDPReaskFNF`, `DownloadClient.cpp:1774-1795`): besides the
@@ -334,23 +344,32 @@ fn routed_reply_events(
     action: ReaskAction,
     file_hash: Ed2kHash,
     endpoint: (Ipv4Addr, u16),
+    lease_endpoint: (Ipv4Addr, u16),
 ) -> Vec<ReaskEvent> {
     match action {
         // Slot imminent: release the lease, then ask core to reconnect over TCP.
+        // The release carries the TCP lease key, not the UDP routing endpoint
+        // (core's lease sets are TCP-keyed; RUST-PAR-017 DL-11).
         ReaskAction::UpdatedRank(rank) if rank <= REENGAGE_RANK_THRESHOLD => vec![
-            ReaskEvent::SourceReleased { endpoint },
+            ReaskEvent::SourceReleased {
+                endpoint: lease_endpoint,
+            },
             ReaskEvent::SourceReady { file_hash },
         ],
         // Uploader no longer has the file (OP_FILENOTFOUND): dead-list it first
         // (oracle UDPReaskFNF AddDeadSource, DownloadClient.cpp:1781), then free
         // its lease. Order matters: the dead-list gate must be in place before
-        // the released source becomes re-acquirable.
+        // the released source becomes re-acquirable. SourceDead keeps the UDP
+        // endpoint (core's FNF resolver keys on the IP only); the release again
+        // carries the TCP lease key.
         ReaskAction::DropSource => vec![
             ReaskEvent::SourceDead {
                 file_hash,
                 endpoint,
             },
-            ReaskEvent::SourceReleased { endpoint },
+            ReaskEvent::SourceReleased {
+                endpoint: lease_endpoint,
+            },
         ],
         // Still queued / transient: keep the source on UDP reask, lease unchanged.
         _ => Vec::new(),
@@ -411,6 +430,7 @@ async fn handle_inbound_datagram(
         ReaskInboundOutcome::RoutedReply {
             file_hash,
             endpoint,
+            lease_endpoint,
             action,
         } => {
             trace!("ed2k udp reask: routed reply from {from}: {action:?} (file {file_hash})");
@@ -419,7 +439,7 @@ async fn handle_inbound_datagram(
                 service.remove_source(endpoint.0, endpoint.1);
             }
             // Emit loop->core events (lease release ordered before any re-engage).
-            for event in routed_reply_events(action, file_hash, endpoint) {
+            for event in routed_reply_events(action, file_hash, endpoint, lease_endpoint) {
                 if let ReaskEvent::SourceReady { file_hash } = &event {
                     trace!("ed2k udp reask: re-engage SourceReady for {file_hash}");
                 }
@@ -490,6 +510,7 @@ fn apply_reask_command(
             let ReaskDetachArgs {
                 file_hash,
                 endpoint,
+                tcp_port,
                 udp_version,
                 initial_reask_delay,
                 user_hash,
@@ -499,7 +520,8 @@ fn apply_reask_command(
                 buddy_id,
             } = args;
             let now = Instant::now();
-            let mut source = ReaskSource::new(endpoint, file_hash, udp_version, now);
+            let mut source = ReaskSource::new(endpoint, file_hash, udp_version, now)
+                .with_lease_endpoint((endpoint.0, tcp_port));
             source.next_reask = now + initial_reask_delay;
             if let Some(hash) = user_hash {
                 source = source.with_obfuscation(hash, should_crypt);
@@ -513,9 +535,13 @@ fn apply_reask_command(
         }
         ReaskCommand::Remove { endpoint } => {
             // Only release the lease if this endpoint was a detached source we held;
-            // a Remove for an unknown endpoint must not free a lease core never gave us.
-            if service.remove_source(endpoint.0, endpoint.1) {
-                let _ = events.send(ReaskEvent::SourceReleased { endpoint });
+            // a Remove for an unknown endpoint must not free a lease core never gave
+            // us. The release carries the source's TCP lease key, not the UDP
+            // routing endpoint (core's lease sets are TCP-keyed).
+            if let Some(lease_endpoint) = service.remove_source(endpoint.0, endpoint.1) {
+                let _ = events.send(ReaskEvent::SourceReleased {
+                    endpoint: lease_endpoint,
+                });
             }
         }
         ReaskCommand::MarkNoNeededParts { endpoint } => {
@@ -640,8 +666,10 @@ async fn drive_reask_tick(
             && let SocketAddr::V4(v4) = addr
         {
             let endpoint = (*v4.ip(), v4.port());
-            if service.remove_source(endpoint.0, endpoint.1) {
-                let _ = events.send(ReaskEvent::SourceReleased { endpoint });
+            if let Some(lease_endpoint) = service.remove_source(endpoint.0, endpoint.1) {
+                let _ = events.send(ReaskEvent::SourceReleased {
+                    endpoint: lease_endpoint,
+                });
             }
         }
     }

crates/emulebb-ed2k/src/ed2k_client_udp/runtime/tests.rs

@@ -12,13 +12,18 @@ fn register_command_adds_a_source_and_remove_drops_it() {
     let mut svc = service();
     let (events, mut rx) = reask_event_channel();
     let file_hash = Ed2kHash::from_bytes([0xAB; 16]);
+    // Distinct UDP routing port (4672) vs TCP lease port (4662): core's lease
+    // sets are keyed by (ip, tcp_port), so a SourceReleased carrying the UDP
+    // endpoint would never match and the lease would leak (RUST-PAR-017 DL-11).
     let endpoint = (Ipv4Addr::new(198, 51, 100, 7), 4672);
+    let lease_endpoint = (Ipv4Addr::new(198, 51, 100, 7), 4662);
     apply_reask_command(
         &mut svc,
         &events,
         ReaskCommand::Register(ReaskDetachArgs {
             file_hash,
             endpoint,
+            tcp_port: lease_endpoint.1,
             udp_version: 4,
             initial_reask_delay: Duration::ZERO,
             user_hash: Some([0x55; 16]),
@@ -36,7 +41,10 @@ fn register_command_adds_a_source_and_remove_drops_it() {
     assert_eq!(svc.source_count(), 0);
     assert!(svc.registered_file_hashes().is_empty());
     match rx.try_recv().expect("a SourceReleased event") {
-        ReaskEvent::SourceReleased { endpoint: released } => assert_eq!(released, endpoint),
+        ReaskEvent::SourceReleased { endpoint: released } => assert_eq!(
+            released, lease_endpoint,
+            "the release must carry the TCP lease key, not the UDP routing endpoint"
+        ),
         other => panic!("expected SourceReleased, got {other:?}"),
     }
 }
@@ -53,6 +61,7 @@ fn register_command_respects_initial_reask_delay() {
         ReaskCommand::Register(ReaskDetachArgs {
             file_hash,
             endpoint,
+            tcp_port: 4662,
             udp_version: 4,
             initial_reask_delay: crate::ed2k_client_udp::FILE_REASK_TIME,
             user_hash: None,
@@ -106,14 +115,19 @@ fn remove_command_for_unknown_endpoint_releases_no_lease() {
 fn reengage_releases_lease_before_signalling_source_ready() {
     let file_hash = Ed2kHash::from_bytes([0x33; 16]);
     let endpoint = (Ipv4Addr::new(198, 51, 100, 9), 4672);
+    let lease_endpoint = (Ipv4Addr::new(198, 51, 100, 9), 4662);
     let events = routed_reply_events(
         ReaskAction::UpdatedRank(REENGAGE_RANK_THRESHOLD),
         file_hash,
         endpoint,
+        lease_endpoint,
     );
     assert_eq!(events.len(), 2);
     match &events[0] {
-        ReaskEvent::SourceReleased { endpoint: released } => assert_eq!(*released, endpoint),
+        ReaskEvent::SourceReleased { endpoint: released } => assert_eq!(
+            *released, lease_endpoint,
+            "the release must carry the TCP lease key, not the UDP routing endpoint"
+        ),
         other => panic!("expected SourceReleased first, got {other:?}"),
     }
     match &events[1] {
@@ -128,6 +142,7 @@ fn deep_rank_keeps_source_and_releases_no_lease() {
         ReaskAction::UpdatedRank(REENGAGE_RANK_THRESHOLD + 1),
         Ed2kHash::from_bytes([0x44; 16]),
         (Ipv4Addr::new(198, 51, 100, 10), 4672),
+        (Ipv4Addr::new(198, 51, 100, 10), 4662),
     );
     assert!(
         events.is_empty(),
@@ -139,10 +154,13 @@ fn deep_rank_keeps_source_and_releases_no_lease() {
 fn dropped_source_is_dead_listed_then_releases_its_lease() {
     // UDP FNF (oracle UDPReaskFNF, DownloadClient.cpp:1774-1795): the source is
     // dead-listed BEFORE its lease is released, so the 45-minute block gates
-    // re-acquisition the moment the endpoint becomes free again.
+    // re-acquisition the moment the endpoint becomes free again. SourceDead
+    // carries the UDP endpoint (core resolves by IP); SourceReleased carries
+    // the TCP lease key core's sets match against (RUST-PAR-017 DL-11).
     let endpoint = (Ipv4Addr::new(198, 51, 100, 11), 4672);
+    let lease_endpoint = (Ipv4Addr::new(198, 51, 100, 11), 4662);
     let hash = Ed2kHash::from_bytes([0x55; 16]);
-    let events = routed_reply_events(ReaskAction::DropSource, hash, endpoint);
+    let events = routed_reply_events(ReaskAction::DropSource, hash, endpoint, lease_endpoint);
     assert_eq!(events.len(), 2);
     match &events[0] {
         ReaskEvent::SourceDead {
@@ -155,7 +173,10 @@ fn dropped_source_is_dead_listed_then_releases_its_lease() {
         other => panic!("expected SourceDead, got {other:?}"),
     }
     match &events[1] {
-        ReaskEvent::SourceReleased { endpoint: released } => assert_eq!(*released, endpoint),
+        ReaskEvent::SourceReleased { endpoint: released } => assert_eq!(
+            *released, lease_endpoint,
+            "the release must carry the TCP lease key, not the UDP routing endpoint"
+        ),
         other => panic!("expected SourceReleased, got {other:?}"),
     }
 }
@@ -166,12 +187,14 @@ fn retry_tcp_timeout_releases_held_lease() {
     let (events, mut rx) = reask_event_channel();
     let file_hash = Ed2kHash::from_bytes([0x66; 16]);
     let endpoint = (Ipv4Addr::new(198, 51, 100, 12), 4672);
+    let lease_endpoint = (Ipv4Addr::new(198, 51, 100, 12), 4662);
     apply_reask_command(
         &mut svc,
         &events,
         ReaskCommand::Register(ReaskDetachArgs {
             file_hash,
             endpoint,
+            tcp_port: lease_endpoint.1,
             udp_version: 4,
             initial_reask_delay: Duration::ZERO,
             user_hash: None,
@@ -181,13 +204,19 @@ fn retry_tcp_timeout_releases_held_lease() {
             buddy_id: None,
         }),
     );
-    assert!(svc.remove_source(endpoint.0, endpoint.1));
-    let _ = events.send(ReaskEvent::SourceReleased { endpoint });
+    // remove_source hands back the TCP lease key the release event must carry.
+    assert_eq!(
+        svc.remove_source(endpoint.0, endpoint.1),
+        Some(lease_endpoint)
+    );
+    let _ = events.send(ReaskEvent::SourceReleased {
+        endpoint: lease_endpoint,
+    });
     match rx.try_recv().expect("a SourceReleased event") {
-        ReaskEvent::SourceReleased { endpoint: released } => assert_eq!(released, endpoint),
+        ReaskEvent::SourceReleased { endpoint: released } => assert_eq!(released, lease_endpoint),
         other => panic!("expected SourceReleased, got {other:?}"),
     }
-    assert!(!svc.remove_source(endpoint.0, endpoint.1));
+    assert_eq!(svc.remove_source(endpoint.0, endpoint.1), None);
 }
 
 #[tokio::test]
@@ -258,6 +287,7 @@ fn detach_handle_register_is_received_as_a_command() {
     assert!(handle.register_kad_buddy_source(ReaskDetachArgs {
         file_hash,
         endpoint: (Ipv4Addr::new(10, 0, 0, 1), 5000),
+        tcp_port: 4662,
         udp_version: 4,
         initial_reask_delay: Duration::ZERO,
         user_hash: None,
@@ -269,6 +299,7 @@ fn detach_handle_register_is_received_as_a_command() {
     match rx.try_recv().expect("a queued command") {
         ReaskCommand::Register(args) => {
             assert_eq!(args.endpoint, (Ipv4Addr::new(10, 0, 0, 1), 5000));
+            assert_eq!(args.tcp_port, 4662);
             assert!(args.low_id);
             assert_eq!(
                 args.buddy_endpoint,