RUST-BUG-109: close ED2K callback and server parity gaps

crates/emulebb-core/src/core_state.rs

@@ -88,23 +88,23 @@ pub(crate) struct CoreState {
     pub(crate) ed2k_server_source_last_frame_at: Option<Instant>,
     pub(crate) ed2k_udp_source_batch_last_queried: HashMap<String, Instant>,
     pub(crate) ed2k_kad_source_last_queried: HashMap<String, (Instant, u8)>,
-    /// Last time we sent an outbound Kad `KADEMLIA_CALLBACK_REQ` for a firewalled
-    /// buddy source, keyed by (source ip, source tcp port, file hash). Enforces the
-    /// callback cooldown so a buddy-only source is not re-callbacked every requery
-    /// round (oracle `DS_WAITCALLBACKKAD` reap window).
+    /// Last settled outbound Kad callback per firewalled peer/route. One inbound
+    /// callback session can negotiate all A4AF files, while direct-buddy and
+    /// FINDSOURCE routes remain distinct. Enforces the oracle
+    /// `DS_WAITCALLBACKKAD` reap window without multiplying requests by file.
     pub(crate) ed2k_kad_callback_last_sent:
         HashMap<crate::kad_callback_initiator::KadCallbackKey, Instant>,
-    /// Last time we sent `OP_CALLBACKREQUEST` through the connected eD2K server,
-    /// keyed by (LowID client id, file hash). Mirrors eMule's per-source
-    /// `MIN2MS(20)` callback retry gate.
+    /// Last time we sent a classic server callback for one server-local LowID
+    /// peer, keyed by (server endpoint, client id). The peer-scoped gate covers
+    /// every A4AF file joined to that callback session and mirrors eMule's
+    /// per-client `MIN2MS(20)` retry gate.
     pub(crate) ed2k_server_callback_last_sent:
         HashMap<crate::ed2k_sources::Ed2kServerCallbackKey, Instant>,
-    /// Last time we originated an `OP_DIRECTCALLBACKREQ` to a firewalled type-6
-    /// source, keyed by (source ip, source tcp port, file hash). Enforces the
-    /// same callback cooldown as the Kad-buddy path so a direct-callback source
-    /// is not re-requested every requery round.
+    /// Last settled `OP_DIRECTCALLBACKREQ` dispatch per direct route/peer,
+    /// keyed by (source UDP endpoint, user hash). One connect-back session can
+    /// negotiate every A4AF file, so file identity is deliberately absent.
     pub(crate) ed2k_direct_callback_last_sent:
-        HashMap<crate::kad_callback_initiator::KadCallbackKey, Instant>,
+        HashMap<crate::ed2k_sources::Ed2kDirectCallbackKey, Instant>,
     pub(crate) shared_directories: Vec<SharedDirectoryRoot>,
     pub(crate) unshared_hashes: HashSet<String>,
     pub(crate) monitor_shared_hashes: HashMap<PathBuf, MonitoredSharedFile>,

crates/emulebb-core/src/diag_kad_event.rs

@@ -304,10 +304,17 @@ pub(crate) fn buddy(established: bool, peer: SocketAddr) {
 /// `callback` milestone (uniform-diagnostics-v2 §3.3, rust superset): we
 /// originated an outbound Kad `KADEMLIA_CALLBACK_REQ` to a firewalled buddy
 /// source's buddy so the source connects back to us (oracle `BaseClient.cpp`
-/// `CCS_KADCALLBACK`). `outcome` is `sent`/`send_failed`; `keys.peer` is the
-/// buddy relay endpoint the request went to. This has no direct MFC diag_event
-/// analog (MFC only `DebugSend`s), so it is an allowed rust-only superset event.
-pub(crate) fn callback(outcome: &str, buddy_peer: SocketAddr, source: SocketAddr, file_hash: &str) {
+/// `CCS_KADCALLBACK`). `outcome` is `sent`, `send_failed`, or
+/// `delivery_unknown`; `keys.peer` is the buddy relay endpoint the request went
+/// to when known, while FINDSOURCE traversal omits it. This has no direct MFC
+/// diag_event analog (MFC only `DebugSend`s), so it is an allowed rust-only
+/// superset event.
+pub(crate) fn callback(
+    outcome: &str,
+    buddy_peer: Option<SocketAddr>,
+    source: SocketAddr,
+    file_hash: &str,
+) {
     let milestone = match outcome {
         "sent" => "callback_requested",
         _ => "callback_request_failed",
@@ -320,13 +327,11 @@ pub(crate) fn callback(outcome: &str, buddy_peer: SocketAddr, source: SocketAddr
         "source": source.to_string(),
         "fileHash": file_hash,
     });
-    emit(
-        FAMILY,
-        "callback",
-        severity,
-        json!({ "peer": buddy_peer.to_string(), "fileHash": file_hash }),
-        body,
-    );
+    let mut keys = json!({ "fileHash": file_hash });
+    if let Some(peer) = buddy_peer {
+        keys["peer"] = json!(peer.to_string());
+    }
+    emit(FAMILY, "callback", severity, keys, body);
 }
 
 /// `lookup` milestone `lookup_complete` (schema §3.3): a Kad search/lookup
@@ -427,13 +432,19 @@ mod tests {
         buddy(false, "1.2.3.4:4672".parse().unwrap());
         callback(
             "sent",
-            "198.51.100.9:5000".parse().unwrap(),
+            Some("198.51.100.9:5000".parse().unwrap()),
+            "192.0.2.77:4662".parse().unwrap(),
+            "abc123",
+        );
+        callback(
+            "send_failed",
+            Some("198.51.100.9:5000".parse().unwrap()),
             "192.0.2.77:4662".parse().unwrap(),
             "abc123",
         );
         callback(
             "send_failed",
-            "198.51.100.9:5000".parse().unwrap(),
+            None,
             "192.0.2.77:4662".parse().unwrap(),
             "abc123",
         );

crates/emulebb-core/src/ed2k_buddy_reask.rs

@@ -12,10 +12,12 @@
 //! glue: it builds the [`ReaskDetachArgs`] and hands them to the reask handle.
 
 use std::collections::HashSet;
-use std::net::Ipv4Addr;
 use std::time::Duration;
 
-use emulebb_ed2k::{ReaskDetachArgs, ReaskSourceHandle, ed2k_server::Ed2kFoundSource};
+use emulebb_ed2k::{
+    ReaskDetachArgs, ReaskSourceHandle,
+    ed2k_server::{Ed2kFoundSource, Ed2kFoundSourceKey},
+};
 use emulebb_kad_proto::Ed2kHash;
 
 use crate::ed2k_sources::source_key;
@@ -26,7 +28,7 @@ use crate::ed2k_sources::source_key;
 /// for a firewalled-buddy Kad source).
 const ED2K_DEFAULT_UDP_VERSION: u8 = 4;
 
-type RequestedCallbackSources = HashSet<(Ipv4Addr, u16, Option<[u8; 16]>, Option<u8>)>;
+type RequestedCallbackSources = HashSet<Ed2kFoundSourceKey>;
 
 /// Detach every firewalled LowID Kad source with a known buddy onto the UDP reask
 /// loop so it originates an `OP_REASKCALLBACKUDP` to its buddy on the normal
@@ -93,6 +95,7 @@ mod tests {
     use crate::ed2k_sources::kad_source_result_to_ed2k_found_source;
     use emulebb_ed2k::{ReaskCommand, reask_command_channel};
     use emulebb_kad_dht::SourceResult;
+    use std::net::Ipv4Addr;
 
     fn buddy_source(file_hash: Ed2kHash) -> Ed2kFoundSource {
         Ed2kFoundSource {

crates/emulebb-core/src/ed2k_net_drivers.rs

@@ -23,7 +23,9 @@ use anyhow::Result;
 use emulebb_ed2k::ed2k_server::{
     Ed2kServerListEvent, Ed2kServerListEventReceiver, Ed2kServerState,
 };
-use emulebb_ed2k::ed2k_tcp::{Ed2kHelloIdentity, connect_callback_peer};
+use emulebb_ed2k::ed2k_tcp::{
+    Ed2kCallbackSessionHandle, Ed2kCallbackSessionOrigin, Ed2kCallbackSessionRequest,
+};
 use emulebb_ed2k::kad_firewall::KadFirewallState;
 use emulebb_ed2k::stun::{
     DEFAULT_STUN_TIMEOUT, NatMappingBehavior, stun_probe, stun_probe_mapping_behavior,
@@ -33,6 +35,7 @@ use emulebb_ed2k::{
     reachability::ExternalReachability,
 };
 use emulebb_ed2k::{ReaskEvent, ReaskEventReceiver};
+use emulebb_kad_dht::DhtNode;
 use tokio::net::TcpListener;
 use tokio::sync::{Mutex, RwLock};
 
@@ -44,10 +47,10 @@ use crate::{Ed2kNetworkConfig, EmulebbCore, current_tcp_firewalled};
 /// to mirror `TryToConnectOrDelete`. Cloned cheaply (all `Arc`/`Copy`).
 #[derive(Clone)]
 pub(crate) struct ReaskReengageContext {
-    /// Our eD2k bind IP for the outbound callback connection (egress-pinned).
-    pub bind_ip: Ipv4Addr,
-    /// Our advertised hello identity for the outbound `OP_HELLO`.
-    pub hello_identity: Ed2kHelloIdentity,
+    /// Live Kad runtime; stock accepts direct callbacks only while Kad is running.
+    pub dht: DhtNode,
+    /// Listener-owned dispatcher for the callback's long-lived peer session.
+    pub callback_sessions: Ed2kCallbackSessionHandle,
     /// The eD2k TCP listener, for the last-resort firewalled fallback verdict.
     pub ed2k_listener: Arc<TcpListener>,
     /// The eD2k server state (authoritative LowID flag).
@@ -329,17 +332,22 @@ fn direct_callback_target_is_valid(requester_ip: Ipv4Addr, tcp_port: u16) -> boo
 /// callback (MISCOPTIONS2 bit 12) — to connect out to it. Mirrors the oracle
 /// `ClientUDPSocket.cpp` `OP_DIRECTCALLBACKREQ` handler: gate on
 /// `Kademlia::IsRunning() && Kademlia::IsFirewalled()`, then connect out to the
-/// requester (`TryToConnectOrDelete`). The reask loop only runs with Kad up, so
-/// the running half holds; we re-check the firewalled half live here (it can
-/// change as our reachability is learned), only acting when we are still the
-/// firewalled side we advertised.
+/// requester (`TryToConnectOrDelete`). Re-check both halves live because the
+/// shared client-UDP ingress outlives Kad protocol activity and reachability can
+/// change after the source advertisement.
 async fn handle_direct_callback_req(
     ctx: &ReaskReengageContext,
     requester_ip: Ipv4Addr,
     tcp_port: u16,
     user_hash: [u8; 16],
     connect_options: u8,
 ) {
+    if !ctx.dht.is_running() {
+        tracing::debug!(
+            "ignoring OP_DIRECTCALLBACKREQ from {requester_ip}:{tcp_port}: Kad is not running"
+        );
+        return;
+    }
     if !direct_callback_target_is_valid(requester_ip, tcp_port) {
         tracing::debug!(
             "ignoring OP_DIRECTCALLBACKREQ: invalid connect-back endpoint {requester_ip}:{tcp_port}"
@@ -373,32 +381,18 @@ async fn handle_direct_callback_req(
     // (oracle AddTrackCallbackRequests, only on an accepted request).
     ctx.direct_callback_limiter.track(requester_ip);
     let peer_addr = SocketAddr::new(IpAddr::V4(requester_ip), tcp_port);
-    let bind_ip = ctx.bind_ip;
-    let hello_identity = ctx.hello_identity;
-    // Connect out like the oracle TryToConnectOrDelete: open an outgoing eD2k
-    // client connection and send OP_HELLO, carrying the requester's user hash +
-    // connect options for the obfuscated handshake. Spawned so a slow connect
-    // never stalls the event consumer.
-    tokio::spawn(async move {
-        match connect_callback_peer(
-            bind_ip,
-            peer_addr,
-            hello_identity,
-            Some(user_hash),
-            Some(connect_options),
-            Duration::from_secs(20),
-        )
-        .await
-        {
-            Ok(mode) => tracing::debug!(
-                "direct-UDP-callback connect-out to {peer_addr} completed transport={}",
-                mode.as_str()
-            ),
-            Err(error) => {
-                tracing::debug!("direct-UDP-callback connect-out to {peer_addr} failed: {error:#}")
-            }
-        }
-    });
+    // Queue the complete peer session without waiting on the dial from this UDP
+    // event consumer. The listener-owned supervisor sends HELLO and retains the
+    // socket for the requester's subsequent upload protocol.
+    if let Err(error) = ctx.callback_sessions.try_start(Ed2kCallbackSessionRequest {
+        peer_addr,
+        expected_user_hash: Some(user_hash),
+        peer_connect_options: Some(connect_options),
+        timeout: Duration::from_secs(20),
+        origin: Ed2kCallbackSessionOrigin::DirectUdp,
+    }) {
+        tracing::debug!("direct-UDP-callback dispatch to {peer_addr} failed: {error}");
+    }
 }
 
 /// Consumer for server-list feedback raised by the ED2K server session loop
@@ -572,4 +566,39 @@ mod tests {
             4662
         ));
     }
+
+    #[tokio::test]
+    async fn direct_callback_requires_live_kad_at_dispatch_time() {
+        let dht = DhtNode::new(emulebb_kad_dht::DhtConfig {
+            bind_addr: Some(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))),
+            ..emulebb_kad_dht::DhtConfig::default()
+        })
+        .await
+        .unwrap();
+        dht.set_running(false);
+        let listener = Arc::new(TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap());
+        let server_state = Arc::new(RwLock::new(Ed2kServerState {
+            client_id: Some(1),
+            ..Ed2kServerState::default()
+        }));
+        let (callback_sessions, _callback_inbox) =
+            emulebb_ed2k::ed2k_tcp::ed2k_callback_session_channel();
+        let limiter = Arc::new(crate::callback_tracker::DirectCallbackRateLimiter::new());
+        let requester_ip = Ipv4Addr::new(198, 51, 100, 7);
+        let context = ReaskReengageContext {
+            dht,
+            callback_sessions,
+            ed2k_listener: listener,
+            server_state,
+            kad_firewall: Arc::new(Mutex::new(KadFirewallState::default())),
+            direct_callback_limiter: Arc::clone(&limiter),
+        };
+
+        handle_direct_callback_req(&context, requester_ip, 4662, [0x5A; 16], 0x01).await;
+
+        assert!(
+            limiter.allow(requester_ip),
+            "a callback rejected because Kad stopped must not consume its rate-limit slot"
+        );
+    }
 }