rest: surface our queue position at each download source (queueRank)

The download session decoded OP_QUEUERANK / OP_QUEUERANKING but only logged it,
so the REST source list always showed queueRank=0 -- the UI could not tell you
where you sit in each peer's upload queue. Record the rank on the live download
source (Ed2kSourceActivity.queue_rank via note_download_source_queue_rank at both
decode sites), carry it on Ed2kLiveSource, and overlay it in enrich_sources_with_live.
clientSoftware on download sources remains a follow-up (needs the peer HELLO
software captured into the source registry).

Author: Threepwood-7

crates/emulebb-core/src/views.rs

@@ -315,6 +315,9 @@ pub(crate) fn enrich_sources_with_live(
         source.download_speed_ki_bps = live_source.download_speed_bytes_per_sec as f64 / 1024.0;
         source.available_parts = live_source.available_parts;
         source.part_count = part_count;
+        if let Some(rank) = live_source.queue_rank {
+            source.queue_rank = rank;
+        }
         let state = if live_source.transferring {
             "downloading"
         } else {

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

@@ -851,6 +851,7 @@ pub(in crate::ed2k_tcp) async fn drive_download_session(
                 (OP_EDONKEYPROT, OP_QUEUERANK) => {
                     let rank = decode_edonkey_queue_rank_payload(&packet.payload)?;
                     session_state.queued_until = Some(tokio::time::Instant::now() + QUEUE_RANK_GRACE);
+                    transfer_runtime.note_download_source_queue_rank(file_hash_hex, peer_addr, rank);
                     dump_ed2k_tcp_download_meta(
                         peer_addr,
                         Some(transport.mode),
@@ -861,6 +862,11 @@ pub(in crate::ed2k_tcp) async fn drive_download_session(
                 (OP_EMULEPROT, OP_QUEUERANKING) => {
                     let rank = decode_emule_queue_ranking_payload(&packet.payload)?;
                     session_state.queued_until = Some(tokio::time::Instant::now() + QUEUE_RANK_GRACE);
+                    transfer_runtime.note_download_source_queue_rank(
+                        file_hash_hex,
+                        peer_addr,
+                        u32::from(rank),
+                    );
                     dump_ed2k_tcp_download_meta(
                         peer_addr,
                         Some(transport.mode),

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

@@ -80,6 +80,9 @@ pub(super) struct Ed2kSourceActivity {
     /// Per-part availability advertised by the peer (OP_FILESTATUS). `None`
     /// until a status frame is seen.
     part_bitmap: Option<Vec<bool>>,
+    /// Our queue position at this peer (OP_QUEUERANK / OP_QUEUERANKING). `None`
+    /// until the peer sends a ranking.
+    queue_rank: Option<u32>,
 }
 
 /// Snapshot of one live source for the REST transfer-source/detail views.
@@ -91,6 +94,7 @@ pub struct Ed2kLiveSource {
     pub download_speed_bytes_per_sec: u64,
     pub transferring: bool,
     pub available_parts: u32,
+    pub queue_rank: Option<u32>,
 }
 
 impl Ed2kTransferRuntime {
@@ -185,6 +189,7 @@ impl Ed2kTransferRuntime {
                 last_payload_at: None,
                 window: DatarateWindow::default(),
                 part_bitmap: None,
+                queue_rank: None,
             });
         entry.endpoint = peer;
         if user_hash.is_some() {
@@ -225,6 +230,7 @@ impl Ed2kTransferRuntime {
                 last_payload_at: None,
                 window: DatarateWindow::default(),
                 part_bitmap: None,
+                queue_rank: None,
             });
         entry.endpoint = peer;
         if user_hash.is_some() {
@@ -237,6 +243,37 @@ impl Ed2kTransferRuntime {
         entry.part_bitmap = Some(bitmap);
     }
 
+    /// Record our queue position at a peer (OP_QUEUERANK / OP_QUEUERANKING) so the
+    /// REST source list can show it. The session entry usually already exists from
+    /// a payload/status frame; create a minimal one if the ranking arrives first.
+    pub(crate) fn note_download_source_queue_rank(
+        &self,
+        file_hash: &str,
+        peer: SocketAddr,
+        rank: u32,
+    ) {
+        let now = Instant::now();
+        let Ok(mut sources) = self.download_sources.lock() else {
+            return;
+        };
+        let peers = sources.entry(file_hash.to_string()).or_default();
+        evict_stale_sources(peers, now);
+        let entry = peers
+            .entry(peer.to_string())
+            .or_insert_with(|| Ed2kSourceActivity {
+                endpoint: peer,
+                user_hash: None,
+                connect_options: None,
+                last_seen_at: now,
+                last_payload_at: None,
+                window: DatarateWindow::default(),
+                part_bitmap: None,
+                queue_rank: None,
+            });
+        entry.last_seen_at = now;
+        entry.queue_rank = Some(rank);
+    }
+
     /// Drop the live-source registry for a file (e.g. on transfer removal).
     pub(crate) fn clear_download_sources(&self, file_hash: &str) {
         if let Ok(mut sources) = self.download_sources.lock() {
@@ -274,6 +311,7 @@ impl Ed2kTransferRuntime {
                     .as_ref()
                     .map(|bitmap| bitmap.iter().filter(|present| **present).count() as u32)
                     .unwrap_or(0),
+                queue_rank: peer.queue_rank,
             })
             .collect()
     }