RUST-PAR-017: dial disconnected waiters for slot grants; rank only on re-ask

There was no way to deliver a slot grant to a waiter without a live
connection, so rust compensated by holding every waiting connection open
forever (500 ms poll) and pushing unsolicited OP_QUEUERANKING refreshes
every 10 s -- packets the oracle never sends. In the oracle, when the
queue grants a slot to a client with no connected socket, AddUpNextClient
sets US_CONNECTING and opens an OUTBOUND client connection (TryToConnect,
UploadQueue.cpp:327-361); once established, the client sends OP_HELLO and
pushes OP_ACCEPTUPLOADREQ on that connection (ConnectionEstablished,
BaseClient.cpp:1634-1641), and a failed connect deletes the client,
dropping the grant so the slot moves to the next waiter.

The new ed2k_tcp::upload_promote driver (spawned by the listener) drains
the pending promotions recorded by the runtime queue, dials each HighID
peer's advertised endpoint -- a LowID grant without a live callback path
is dropped like the oracle failed-connect path -- and serves the session
through the standard listener dispatch (Ed2kSessionSource::
PromotedUpload seeds the peer identity and pushes HELLO + accept before
the packet loop).

Queue rank now goes out only in response to a re-ask, matching the
oracle SendRankingInfo call sites (UploadQueue.cpp:1866,1963,1986): the
10 s refresh timer and the rank-changed push are removed. An idle
WAITING connection is closed after 30 s like the oracle client-socket
timeout (CClientReqSocket::CheckTimeOut, thePrefs.GetConnectionTimeout()
default 30 s on this branch) while the queue entry survives the close
and is dialed back on a grant.

Residual divergence: an ACTIVE (granted/uploading) connection is still
not idle-closed by rust; the underfill slot recycle handles stalled
active slots, so only waiting connections adopt the socket timeout here.

Author: Threepwood-7

crates/emulebb-ed2k/src/ed2k_tcp.rs

@@ -47,6 +47,7 @@ mod identity;
 mod listener;
 mod obfuscation;
 mod transport;
+mod upload_promote;
 pub(in crate::ed2k_tcp) use codec::{
     PeerSourceExchangeRequest, SourceExchangePeer, decode_aich_file_hash_answer,
     decode_aich_recovery_answer_payload, decode_aich_recovery_request_payload,
@@ -127,7 +128,7 @@ pub(in crate::ed2k_tcp) use identity::{
 };
 pub(crate) use listener::reply_with_firewall_udp;
 #[cfg(test)]
-use listener::{Ed2kConnectionContext, handle_connection};
+use listener::{Ed2kConnectionContext, Ed2kSessionSource, handle_connection};
 pub use listener::{Ed2kListenerOptions, run_ed2k_listener};
 use obfuscation::{
     Rc4KeyStream, accept_incoming_obfuscation_handshake, is_plain_ed2k_protocol_marker,
@@ -220,10 +221,12 @@ pub(crate) const MAX_ED2K_PACKET_LEN: usize = 2_000_000;
 const MAX_PEER_DECOMPRESSED_PACKET_LEN: usize = 50_000;
 const ED2K_CONNECTION_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
 const ED2K_UPLOAD_QUEUE_POLL_INTERVAL: Duration = Duration::from_millis(500);
-#[cfg(not(test))]
-const ED2K_UPLOAD_QUEUE_REFRESH_INTERVAL: Duration = Duration::from_secs(10);
-#[cfg(test)]
-const ED2K_UPLOAD_QUEUE_REFRESH_INTERVAL: Duration = Duration::from_millis(200);
+/// Idle cutoff for a connection whose peer only holds a WAITING queue entry:
+/// the oracle closes an idle client socket after the connection timeout
+/// (`CClientReqSocket::CheckTimeOut`, `thePrefs.GetConnectionTimeout()` default
+/// 30 s on this fork) while the US_ONUPLOADQUEUE entry itself survives up to
+/// MAX_PURGEQUEUETIME between re-asks.
+const ED2K_WAITING_CONNECTION_IDLE_TIMEOUT: Duration = ED2K_CONNECTION_IDLE_TIMEOUT;
 const FIREWALL_HELPER_POST_REQUEST_KEEPALIVE_SECS: u64 = 10;
 const ED2K_UPLOAD_PACKET_SPLIT_THRESHOLD: usize = 13_000;
 const ED2K_UPLOAD_PACKET_FRAGMENT_LEN: usize = 10_240;

crates/emulebb-ed2k/src/ed2k_tcp/listener/mod.rs

@@ -24,8 +24,9 @@ use super::{Ed2kHelloIdentity, Ed2kSecureIdent};
 mod session;
 
 pub(crate) use session::reply_with_firewall_udp;
-#[cfg(test)]
-pub(in crate::ed2k_tcp) use session::{Ed2kConnectionContext, handle_connection};
+pub(in crate::ed2k_tcp) use session::{
+    Ed2kConnectionContext, Ed2kSessionSource, handle_connection,
+};
 
 /// Inputs for the long-lived ED2K TCP listener task.
 pub struct Ed2kListenerOptions {
@@ -65,11 +66,11 @@ pub async fn run_ed2k_listener(options: Ed2kListenerOptions) {
     // Resolve the VPN bind interface index once (from the listener's local addr)
     // so each accepted socket can egress-pin to the tunnel (IP_UNICAST_IF) without
     // a per-connection interface lookup.
-    let bind_if_index = match listener.local_addr() {
+    let (bind_ip, bind_if_index) = match listener.local_addr() {
         Ok(addr) => match addr.ip() {
             std::net::IpAddr::V4(v4) => {
                 match crate::networking::require_bind_if_index(v4, "eD2K listener") {
-                    Ok(index) => index,
+                    Ok(index) => (v4, index),
                     Err(error) => {
                         warn!("eD2K listener disabled: {error:#}");
                         return;
@@ -86,6 +87,25 @@ pub async fn run_ed2k_listener(options: Ed2kListenerOptions) {
             return;
         }
     };
+    // Outbound promote-connect driver: hands upload slots to waiters whose
+    // connection is gone by dialing their advertised endpoint and pushing
+    // OP_ACCEPTUPLOADREQ (master AddUpNextClient connect-out,
+    // UploadQueue.cpp:327-361).
+    tokio::spawn(
+        super::upload_promote::UploadPromoteDriver {
+            dht: dht.clone(),
+            server_state: Arc::clone(&server_state),
+            kad_firewall: Arc::clone(&kad_firewall),
+            secure_ident: Arc::clone(&secure_ident),
+            transfer_runtime: Arc::clone(&transfer_runtime),
+            hello_identity,
+            reachability: reachability.clone(),
+            buddy_registry: buddy_registry.clone(),
+            bind_ip,
+            shutdown: Arc::clone(&shutdown),
+        }
+        .run(),
+    );
     while !shutdown.load(Ordering::Relaxed) {
         match listener.accept().await {
             Ok((stream, peer_addr)) => {
@@ -144,7 +164,7 @@ pub async fn run_ed2k_listener(options: Ed2kListenerOptions) {
                     // inbound slot is released on every exit path (Drop).
                     let _inbound_guard = inbound_guard;
                     if let Err(error) = session::handle_connection(
-                        stream,
+                        session::Ed2kSessionSource::Inbound(stream),
                         peer_addr,
                         session::Ed2kConnectionContext {
                             dht: &dht,

crates/emulebb-ed2k/src/ed2k_tcp/listener/session.rs

@@ -18,7 +18,7 @@ use tokio::sync::mpsc;
 use crate::{
     buddy_socket::BuddySocketRegistry,
     ed2k_server::Ed2kServerState,
-    ed2k_transfer::{Ed2kTransferRuntime, Ed2kUploadPeerIdentity},
+    ed2k_transfer::{Ed2kTransferRuntime, Ed2kUploadPeerIdentity, Ed2kUploadPendingPromotion},
     kad_firewall::KadFirewallState,
 };
 
@@ -36,7 +36,7 @@ use super::super::dump::{
 };
 use super::super::hello::{
     DecodedHelloProfile, build_hello_responses, decode_emule_info_profile, decode_hello_profile,
-    encode_emule_info_answer,
+    encode_emule_info_answer, encode_hello_request,
 };
 use super::super::identity::{Ed2kPeerSecureIdentState, begin_secure_ident_probe};
 use super::super::{
@@ -71,6 +71,20 @@ use shared_file::{
 use upload_payload::{UploadPayloadOutcome, UploadPayloadRequest, serve_upload_payload};
 use upload_queue::{ListenerQueuePoll, ListenerUploadQueue};
 
+/// Transport source for one eD2k peer session: an accepted inbound socket, or
+/// an already-established OUTBOUND connection dialed to hand an upload slot to
+/// a disconnected waiter (oracle `AddUpNextClient` US_CONNECTING connect-out,
+/// UploadQueue.cpp:327-361).
+pub(in crate::ed2k_tcp) enum Ed2kSessionSource {
+    Inbound(TcpStream),
+    PromotedUpload {
+        // Boxed: the established transport dwarfs the plain inbound socket
+        // (clippy::large_enum_variant).
+        transport: Box<Ed2kTransport>,
+        grant: Box<Ed2kUploadPendingPromotion>,
+    },
+}
+
 pub(in crate::ed2k_tcp) struct Ed2kConnectionContext<'a> {
     pub(in crate::ed2k_tcp) dht: &'a DhtNode,
     pub(in crate::ed2k_tcp) server_state: &'a Arc<RwLock<Ed2kServerState>>,
@@ -87,7 +101,7 @@ pub(in crate::ed2k_tcp) struct Ed2kConnectionContext<'a> {
 
 #[allow(clippy::cognitive_complexity)]
 pub(in crate::ed2k_tcp) async fn handle_connection(
-    stream: TcpStream,
+    source: Ed2kSessionSource,
     peer_addr: SocketAddr,
     context: Ed2kConnectionContext<'_>,
 ) -> Result<()> {
@@ -101,15 +115,6 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
         reachability,
         buddy_registry,
     } = context;
-    let local_addr = stream.local_addr().with_context(|| {
-        format!("failed to resolve local eD2k listener address for {peer_addr}")
-    })?;
-    dump_ed2k_tcp_listener_meta(
-        peer_addr,
-        None,
-        "tcp_accept",
-        format!("local_addr={local_addr}"),
-    );
     let kad_udp_port = dht
         .bind_addr()
         .context("failed to resolve Kad bind address for eD2k hello response")?
@@ -125,51 +130,83 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
     };
     let response_identity =
         enrich_hello_identity(response_identity, server_state, kad_firewall).await;
-    let mut transport = match tokio::time::timeout(
-        ED2K_CONNECTION_IDLE_TIMEOUT,
-        Ed2kTransport::accept(stream, hello_identity.user_hash),
-    )
-    .await
-    {
-        Ok(Ok(transport)) => transport,
-        Ok(Err(error)) => {
+    let (mut transport, promoted_grant) = match source {
+        Ed2kSessionSource::Inbound(stream) => {
+            let local_addr = stream.local_addr().with_context(|| {
+                format!("failed to resolve local eD2k listener address for {peer_addr}")
+            })?;
             dump_ed2k_tcp_listener_meta(
                 peer_addr,
                 None,
-                "accept_failed",
-                format!("local_addr={local_addr} error={error:#}"),
+                "tcp_accept",
+                format!("local_addr={local_addr}"),
             );
-            return Err(error).with_context(|| {
-                format!("failed to accept inbound eD2k peer transport from {peer_addr}")
-            });
+            let transport = match tokio::time::timeout(
+                ED2K_CONNECTION_IDLE_TIMEOUT,
+                Ed2kTransport::accept(stream, hello_identity.user_hash),
+            )
+            .await
+            {
+                Ok(Ok(transport)) => transport,
+                Ok(Err(error)) => {
+                    dump_ed2k_tcp_listener_meta(
+                        peer_addr,
+                        None,
+                        "accept_failed",
+                        format!("local_addr={local_addr} error={error:#}"),
+                    );
+                    return Err(error).with_context(|| {
+                        format!("failed to accept inbound eD2k peer transport from {peer_addr}")
+                    });
+                }
+                Err(_) => {
+                    dump_ed2k_tcp_listener_meta(
+                        peer_addr,
+                        None,
+                        "accept_timeout",
+                        format!(
+                            "local_addr={local_addr} idle_timeout_secs={}",
+                            ED2K_CONNECTION_IDLE_TIMEOUT.as_secs()
+                        ),
+                    );
+                    anyhow::bail!("timed out waiting for initial eD2k peer bytes");
+                }
+            };
+            (transport, None)
         }
-        Err(_) => {
+        // A slot grant for a disconnected waiter arrives on a connection WE
+        // dialed (master AddUpNextClient US_CONNECTING connect-out); the
+        // transport handshake already happened in the promote driver.
+        Ed2kSessionSource::PromotedUpload { transport, grant } => {
             dump_ed2k_tcp_listener_meta(
                 peer_addr,
-                None,
-                "accept_timeout",
-                format!(
-                    "local_addr={local_addr} idle_timeout_secs={}",
-                    ED2K_CONNECTION_IDLE_TIMEOUT.as_secs()
-                ),
+                Some(transport.mode),
+                "promote_connect",
+                format!("file_hash={}", grant.file_hash),
             );
-            anyhow::bail!("timed out waiting for initial eD2k peer bytes");
+            (*transport, Some(*grant))
         }
     };
+    let local_addr = transport
+        .stream
+        .local_addr()
+        .with_context(|| format!("failed to resolve local eD2k session address for {peer_addr}"))?;
     transport
         .stream
         .set_nodelay(true)
-        .with_context(|| format!("failed to enable TCP_NODELAY for inbound peer {peer_addr}"))?;
+        .with_context(|| format!("failed to enable TCP_NODELAY for peer {peer_addr}"))?;
     debug!(
-        "accepted eD2k TCP peer from {peer_addr} transport={}",
+        "eD2k TCP peer session with {peer_addr} transport={}",
         transport.mode.as_str()
     );
-    dump_ed2k_tcp_listener_meta(
-        peer_addr,
-        Some(transport.mode),
-        "accept",
-        format!("udp_port={kad_udp_port}"),
-    );
+    if promoted_grant.is_none() {
+        dump_ed2k_tcp_listener_meta(
+            peer_addr,
+            Some(transport.mode),
+            "accept",
+            format!("udp_port={kad_udp_port}"),
+        );
+    }
     let mut peer_secure_ident = Ed2kPeerSecureIdentState::default();
     let mut requested_file_hash: Option<Ed2kHash> = None;
     let mut peer_supports_aich = false;
@@ -192,13 +229,52 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
     // a flood of these is a share-probe and the peer is banned (MFC file_request_flood).
     let mut failed_file_req_count: u32 = 0;
 
+    // A promoted-upload outbound session announces itself and pushes the slot
+    // grant before entering the dispatch loop: the oracle sends OP_HELLO and
+    // then OP_ACCEPTUPLOADREQ on the fresh connection it opened for the
+    // promoted waiter (ConnectionEstablished, BaseClient.cpp:1634-1641).
+    if let Some(grant) = promoted_grant {
+        peer_upload_identity = grant.peer.clone();
+        peer_upload_identity.should_crypt = transport.mode.is_obfuscated();
+        bound_user_hash = grant.peer.user_hash;
+        let file_hash = Ed2kHash::from_str(&grant.file_hash)
+            .with_context(|| format!("invalid promoted upload file hash {}", grant.file_hash))?;
+        requested_file_hash = Some(file_hash);
+        let hello_packet = encode_hello_request(response_identity);
+        dump_ed2k_tcp_listener_send(peer_addr, transport.mode, "hello_request", &hello_packet);
+        transport.write_all(&hello_packet).await.with_context(|| {
+            format!("failed to send OP_HELLO to promoted upload peer {peer_addr}")
+        })?;
+        let attached = upload_queue
+            .attach_promoted_grant(
+                transfer_runtime,
+                &grant.peer,
+                grant.handle,
+                file_hash,
+                &mut transport,
+                peer_addr,
+            )
+            .await?;
+        if !attached {
+            // The grant went stale while connecting (aged out or re-owned by an
+            // inbound reconnect): nothing to serve on this connection.
+            return Ok(());
+        }
+    }
+
     // Run the session loop inside a fallible async scope so that EVERY exit
     // path -- a clean `break`, a propagated `?` I/O error, or any other early
     // return from the loop body -- lands in `result` and falls through to the
-    // unconditional `upload_queue.release(...)` below. The master always frees
-    // the slot on teardown (`CUpDownClient::Disconnected`, BaseClient.cpp:1172);
-    // without this wrapper an in-loop `?` would skip the release and leave the
-    // slot pinned until the idle reaper reclaimed it.
+    // unconditional `upload_queue.release(...)` below. The master frees an
+    // ACTIVE slot on teardown (`CUpDownClient::Disconnected` removes
+    // US_UPLOADING/US_CONNECTING, BaseClient.cpp:1172-1175) while a waiting
+    // queue entry survives the disconnect (BaseClient.cpp:1229); the runtime
+    // release applies exactly that split. Without this wrapper an in-loop `?`
+    // would skip the release and leave the slot pinned until the idle reaper
+    // reclaimed it.
+    // Last time the peer actually sent us a packet: feeds the waiting-connection
+    // idle close (oracle socket timeout) inside poll_on_timeout.
+    let mut last_packet_at = tokio::time::Instant::now();
     let result: Result<()> = async {
         loop {
         let read_timeout = upload_queue.read_timeout();
@@ -233,7 +309,12 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
                         Ok(packet) => packet
                             .with_context(|| format!("failed to read eD2k packet from {peer_addr}"))?,
                         Err(_) => match upload_queue
-                            .poll_on_timeout(transfer_runtime, &mut transport, peer_addr)
+                            .poll_on_timeout(
+                                transfer_runtime,
+                                &mut transport,
+                                peer_addr,
+                                last_packet_at.elapsed(),
+                            )
                             .await?
                         {
                             ListenerQueuePoll::Continue => continue,
@@ -248,7 +329,12 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
                     packet.with_context(|| format!("failed to read eD2k packet from {peer_addr}"))?
                 }
                 Err(_) => match upload_queue
-                    .poll_on_timeout(transfer_runtime, &mut transport, peer_addr)
+                    .poll_on_timeout(
+                        transfer_runtime,
+                        &mut transport,
+                        peer_addr,
+                        last_packet_at.elapsed(),
+                    )
                     .await?
                 {
                     ListenerQueuePoll::Continue => continue,
@@ -259,6 +345,7 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
         let Some(packet) = packet else {
             break Ok(());
         };
+        last_packet_at = tokio::time::Instant::now();
         dump_ed2k_tcp_listener_recv(peer_addr, transport.mode, "session", &packet);
 
         match (packet.protocol, packet.opcode) {