RUST-BUG-085: apply SX2 answer cooldown

Author: Threepwood-7

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

@@ -1104,6 +1104,10 @@ async fn remember_source_exchange_sources(
         return Ok(());
     }
 
+    transfer_runtime
+        .note_source_exchange_answer(file_hash_hex, std::time::Instant::now())
+        .await;
+
     for source in sources {
         if source.tcp_port == 0 || source.ip == [0, 0, 0, 0] {
             continue;

crates/emulebb-ed2k/src/ed2k_transfer.rs

@@ -50,6 +50,7 @@ mod reask_reciprocity;
 mod requested_block;
 mod salvage;
 mod shared_catalog;
+mod source_exchange;
 mod store;
 mod transfer_sql;
 mod upload;
@@ -76,6 +77,7 @@ pub use model::{
     Ed2kCallbackIntent, Ed2kLocalIngestSummary, Ed2kPieceState, Ed2kResumeManifest, Ed2kSourceHint,
     Ed2kTransferJob, Ed2kTransferState,
 };
+use source_exchange::SourceExchangeState;
 use upload_queue::DEFAULT_SOFT_QUEUE_SIZE;
 use upload_queue::Ed2kUploadQueueState;
 pub(crate) use upload_queue::{
@@ -140,16 +142,6 @@ pub fn ed2k_part_count(file_size: u64) -> u16 {
 /// Canonical eMule upload block size used inside one ED2K part request.
 pub(crate) const ED2K_EMBLOCK_SIZE: u64 = 184_320;
 const PAYLOAD_FILE_NAME: &str = "pieces.bin";
-const SOURCE_EXCHANGE_REASK_INTERVAL: Duration = Duration::from_secs(40 * 60);
-const SOURCE_EXCHANGE_COMMON_REASK_INTERVAL: Duration = Duration::from_secs(160 * 60);
-const SOURCE_EXCHANGE_RARE_FILE: usize = 50;
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-struct SourceExchangeRequestKey {
-    file_hash: String,
-    peer_addr: SocketAddr,
-    user_hash: Option<[u8; 16]>,
-}
 
 /// Runtime owner for ED2K transfer manifests, piece-store payloads, and the
 /// transfer-backed shared catalog.
@@ -162,7 +154,7 @@ pub struct Ed2kTransferRuntime {
     manifest_io: Arc<Mutex<()>>,
     manifest_cache: Arc<Mutex<HashMap<String, Ed2kResumeManifest>>>,
     manifest_checkpoint_state: Arc<Mutex<HashMap<String, Ed2kManifestCheckpointState>>>,
-    source_exchange_requests: Arc<Mutex<HashMap<SourceExchangeRequestKey, Instant>>>,
+    source_exchange: SourceExchangeState,
     /// Per-file accumulator of network-proposed AICH roots and their distinct
     /// proposing IPs. A network-learned root is only promoted to the
     /// salvage-authorizing `manifest.aich_root` once it clears the master's
@@ -295,7 +287,7 @@ impl Ed2kTransferRuntime {
             manifest_io: Arc::new(Mutex::new(())),
             manifest_cache: Arc::new(Mutex::new(HashMap::new())),
             manifest_checkpoint_state: Arc::new(Mutex::new(HashMap::new())),
-            source_exchange_requests: Arc::new(Mutex::new(HashMap::new())),
+            source_exchange: SourceExchangeState::default(),
             aich_root_corroboration: Arc::new(StdMutex::new(HashMap::new())),
             download_activity: Arc::new(StdMutex::new(HashMap::new())),
             download_sources: Arc::new(StdMutex::new(HashMap::new())),
@@ -512,28 +504,13 @@ impl Ed2kTransferRuntime {
         if !self.can_engage_file_source(current_source_count) {
             return false;
         }
-        let reask_interval = source_exchange_reask_interval(current_source_count);
-        let key = SourceExchangeRequestKey {
-            file_hash: file_hash.to_string(),
-            peer_addr,
-            user_hash,
-        };
-        let mut requests = self.source_exchange_requests.lock().await;
-        let allowed = requests
-            .get(&key)
-            .is_none_or(|last_requested| now.duration_since(*last_requested) > reask_interval);
-        if allowed {
-            requests.insert(key, now);
-        }
-        allowed
+        self.source_exchange
+            .should_request(file_hash, peer_addr, user_hash, current_source_count, now)
+            .await
     }
-}
 
-fn source_exchange_reask_interval(current_source_count: usize) -> Duration {
-    if current_source_count > SOURCE_EXCHANGE_RARE_FILE {
-        SOURCE_EXCHANGE_COMMON_REASK_INTERVAL
-    } else {
-        SOURCE_EXCHANGE_REASK_INTERVAL
+    pub(crate) async fn note_source_exchange_answer(&self, file_hash: &str, now: Instant) {
+        self.source_exchange.note_answer(file_hash, now).await;
     }
 }
 

crates/emulebb-ed2k/src/ed2k_transfer/source_exchange.rs

@@ -0,0 +1,89 @@
+use std::{
+    collections::HashMap,
+    net::SocketAddr,
+    sync::Arc,
+    time::{Duration, Instant},
+};
+
+use tokio::sync::Mutex;
+
+const SOURCE_EXCHANGE_REASK_INTERVAL: Duration = Duration::from_secs(40 * 60);
+const SOURCE_EXCHANGE_COMMON_REASK_INTERVAL: Duration = Duration::from_secs(160 * 60);
+const SOURCE_EXCHANGE_FILE_ANSWER_INTERVAL: Duration = Duration::from_secs(5 * 60);
+const SOURCE_EXCHANGE_COMMON_FILE_ANSWER_INTERVAL: Duration = Duration::from_secs(20 * 60);
+const SOURCE_EXCHANGE_RARE_FILE: usize = 50;
+const SOURCE_EXCHANGE_VERY_RARE_FILE: usize = SOURCE_EXCHANGE_RARE_FILE / 5;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct SourceExchangeRequestKey {
+    file_hash: String,
+    peer_addr: SocketAddr,
+    user_hash: Option<[u8; 16]>,
+}
+
+#[derive(Debug, Default)]
+pub(super) struct SourceExchangeState {
+    requests: Arc<Mutex<HashMap<SourceExchangeRequestKey, Instant>>>,
+    file_answers: Arc<Mutex<HashMap<String, Instant>>>,
+}
+
+impl SourceExchangeState {
+    pub(super) async fn should_request(
+        &self,
+        file_hash: &str,
+        peer_addr: SocketAddr,
+        user_hash: Option<[u8; 16]>,
+        current_source_count: usize,
+        now: Instant,
+    ) -> bool {
+        let file_answer_interval = file_answer_interval(current_source_count);
+        if !file_answer_interval.is_zero() {
+            let answers = self.file_answers.lock().await;
+            if answers.get(file_hash).is_some_and(|last_answered| {
+                now.duration_since(*last_answered) <= file_answer_interval
+            }) {
+                return false;
+            }
+        }
+
+        let reask_interval = reask_interval(current_source_count);
+        let key = SourceExchangeRequestKey {
+            file_hash: file_hash.to_string(),
+            peer_addr,
+            user_hash,
+        };
+        let mut requests = self.requests.lock().await;
+        let allowed = requests
+            .get(&key)
+            .is_none_or(|last_requested| now.duration_since(*last_requested) > reask_interval);
+        if allowed {
+            requests.insert(key, now);
+        }
+        allowed
+    }
+
+    pub(super) async fn note_answer(&self, file_hash: &str, now: Instant) {
+        self.file_answers
+            .lock()
+            .await
+            .insert(file_hash.to_string(), now);
+    }
+}
+
+fn reask_interval(current_source_count: usize) -> Duration {
+    if current_source_count > SOURCE_EXCHANGE_RARE_FILE {
+        SOURCE_EXCHANGE_COMMON_REASK_INTERVAL
+    } else {
+        SOURCE_EXCHANGE_REASK_INTERVAL
+    }
+}
+
+fn file_answer_interval(current_source_count: usize) -> Duration {
+    if current_source_count <= SOURCE_EXCHANGE_VERY_RARE_FILE {
+        Duration::ZERO
+    } else if current_source_count <= SOURCE_EXCHANGE_RARE_FILE {
+        SOURCE_EXCHANGE_FILE_ANSWER_INTERVAL
+    } else {
+        SOURCE_EXCHANGE_COMMON_FILE_ANSWER_INTERVAL
+    }
+}

crates/emulebb-ed2k/src/ed2k_transfer/tests/source_exchange.rs

@@ -127,3 +127,67 @@ async fn source_exchange_respects_soft_source_cap() {
         "a cap-denied request must not consume the peer throttle"
     );
 }
+
+#[tokio::test]
+async fn source_exchange_answer_cools_down_rare_and_common_files() {
+    let root = unique_test_dir("ed2k-transfer-source-exchange-answer-cooldown");
+    let runtime = Ed2kTransferRuntime::load_or_create(&root).unwrap();
+    let now = Instant::now();
+    let peer_addr = SocketAddr::from((Ipv4Addr::new(10, 1, 2, 6), 4662));
+
+    runtime.note_source_exchange_answer("very_rare", now).await;
+    assert!(
+        runtime
+            .should_request_source_exchange("very_rare", peer_addr, None, 10, now)
+            .await,
+        "very rare files do not use the file-level answer cooldown"
+    );
+
+    runtime.note_source_exchange_answer("rare", now).await;
+    assert!(
+        !runtime
+            .should_request_source_exchange(
+                "rare",
+                peer_addr,
+                None,
+                11,
+                now + Duration::from_secs(60)
+            )
+            .await
+    );
+    assert!(
+        runtime
+            .should_request_source_exchange(
+                "rare",
+                peer_addr,
+                None,
+                11,
+                now + Duration::from_secs(5 * 60 + 1),
+            )
+            .await
+    );
+
+    runtime.note_source_exchange_answer("common", now).await;
+    assert!(
+        !runtime
+            .should_request_source_exchange(
+                "common",
+                peer_addr,
+                None,
+                51,
+                now + Duration::from_secs(5 * 60 + 1),
+            )
+            .await
+    );
+    assert!(
+        runtime
+            .should_request_source_exchange(
+                "common",
+                peer_addr,
+                None,
+                51,
+                now + Duration::from_secs(20 * 60 + 1),
+            )
+            .await
+    );
+}

docs/active/INDEX.md

@@ -96,6 +96,7 @@ items.
 | [RUST-BUG-082](items/RUST-BUG-082.md) | Minor | DONE | Report SX2 answer source counts in live-wire evidence |
 | [RUST-BUG-083](items/RUST-BUG-083.md) | Major | DONE | Gate SX2 requests on the MFC soft source cap |
 | [RUST-BUG-084](items/RUST-BUG-084.md) | Major | DONE | Use MFC common-file SX2 reask timing |
+| [RUST-BUG-085](items/RUST-BUG-085.md) | Major | DONE | Apply MFC SX2 answer cooldown per file |
 
 ## Refactors (`REF`)
 

docs/active/items/RUST-BUG-085.md

@@ -0,0 +1,47 @@
+---
+id: RUST-BUG-085
+title: Apply MFC SX2 answer cooldown per file
+status: done
+priority: Major
+category: bug
+workflow: local
+---
+
+# RUST-BUG-085: Apply MFC SX2 answer cooldown per file
+
+## Problem
+
+Rust decoded `OP_ANSWERSOURCES2` and remembered any sources it carried, but it
+did not update a per-file answer timestamp. eMuleBB MFC calls
+`CPartFile::SetLastAnsweredTime()` as soon as a matching SX2 answer arrives,
+before adding sources, so empty answers still suppress subsequent SX2 requests
+for that file.
+
+That leaves a selected-surface drift after `RUST-BUG-084`: Rust had the peer
+reask timing, but not the file-level answer cooldown that MFC applies to rare
+and common files.
+
+## Acceptance
+
+- [x] A matching `OP_ANSWERSOURCES2` records the file's SX2 answer time even when
+      the answer contains zero sources.
+- [x] Very rare files stay exempt from the file-level answer cooldown.
+- [x] Rare files apply the MFC 5-minute file answer cooldown.
+- [x] Common files apply the MFC 20-minute file answer cooldown.
+- [x] Mismatched SX2 answer hashes are still ignored.
+
+## Implementation Notes
+
+- Added in-memory per-file SX2 answer timestamps to `Ed2kTransferRuntime`.
+- The SX2 request gate now checks the MFC rarity class before the peer/file
+  throttle map is updated.
+- The download session records the answer timestamp after the answer hash
+  matches the requested file and before source ingestion.
+
+## Evidence
+
+- MFC comparison:
+  `srchybrid/DownloadClient.cpp` `CUpDownClient::IsSourceRequestAllowed`,
+  `ListenSocket.cpp` `OP_ANSWERSOURCES2` handling, and `PartFile.h`
+  `SetLastAnsweredTime`.
+- `cargo test -p emulebb-ed2k source_exchange -- --nocapture`