rustial_engine/io/

fetch_pool.rs

//! Concurrent fetch pool with configurable concurrency and viewport-center
//! priority ordering.
//!
//! # Purpose
//!
//! [`FetchPool`] sits between the engine (which decides *which* tiles to
//! fetch) and the host-provided [`HttpClient`] (which does the actual I/O).
//! It adds two things the raw client does not have:
//!
//! 1. **Concurrency limit** -- prevents flooding the network with hundreds
//!    of requests at once when the user zooms out.
//! 2. **Priority queue** -- tiles closest to the viewport center are sent
//!    first, so the area the user is looking at loads before periphery.
//!
//! # Request lifecycle
//!
//! ```text
//! enqueue(url, priority)       -- adds to priority queue
//!    |
//!    v
//! +----------+  flush()   +------------+
//! |  queue   | ---------> |  HttpClient |
//! |  (heap)  |            | (in-flight) |
//! +----------+            +------+------
//!                                | poll()
//!                                v
//!                          completed results
//! ```
//!
//! Call [`enqueue`](FetchPool::enqueue) to add requests, then
//! [`flush`](FetchPool::flush) to dispatch up to the concurrency limit.
//! [`poll`](FetchPool::poll) collects completed responses and
//! automatically flushes freed slots.
//!
//! # Thread safety
//!
//! The pool is `Send + Sync`.  Internal state is protected by mutexes so
//! it can be called from any thread.  Lock scopes are kept short to
//! avoid contention.
//!
//! # Cancellation
//!
//! Call [`cancel`](FetchPool::cancel) to remove a queued (not yet sent)
//! request.  Already in-flight requests cannot be cancelled at this level
//! -- the `HttpClient` may or may not support that.

use crate::io::{HttpClient, HttpRequest, HttpResponse};
use std::cmp::Ordering;
use std::collections::{BinaryHeap, HashSet};
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
use std::sync::Mutex;

// ---------------------------------------------------------------------------
// Priority wrapper
// ---------------------------------------------------------------------------

/// A fetch request annotated with a priority score.
///
/// Lower `priority` values are dispatched first (nearest to viewport center).
#[derive(Debug)]
struct PrioritizedRequest {
    request: HttpRequest,
    /// Distance-like metric from the viewport center.
    /// Lower = more important = dispatched sooner.
    priority: f64,
    sequence: u64,
}

impl PartialEq for PrioritizedRequest {
    fn eq(&self, other: &Self) -> bool {
        self.request.url == other.request.url
    }
}
impl Eq for PrioritizedRequest {}

impl PartialOrd for PrioritizedRequest {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for PrioritizedRequest {
    fn cmp(&self, other: &Self) -> Ordering {
        // BinaryHeap is a max-heap.  We want lower priority values at the
        // top, so we reverse the comparison.  NaN is treated as worst
        // priority (pushed to the bottom).
        other
            .priority
            .partial_cmp(&self.priority)
            .unwrap_or(Ordering::Equal)
            .then_with(|| other.sequence.cmp(&self.sequence))
    }
}

// ---------------------------------------------------------------------------
// FetchPool
// ---------------------------------------------------------------------------

/// A concurrency-limited, priority-ordered download scheduler.
///
/// Wraps an [`HttpClient`] and adds a bounded queue that dispatches
/// the most important requests first.
///
/// # Example
///
/// ```rust,ignore
/// use rustial_engine::{FetchPool, HttpClient};
///
/// let pool = FetchPool::new(my_http_client, 6);
///
/// // Batch enqueue -- no requests are sent yet.
/// pool.enqueue("https://tile.example.com/10/512/340.png".into(), 0.0);
/// pool.enqueue("https://tile.example.com/10/513/340.png".into(), 1.0);
///
/// // Dispatch the highest-priority requests up to the limit.
/// pool.flush();
///
/// // Later, in the frame loop:
/// let completed = pool.poll();  // also flushes freed slots
/// for (url, result) in completed {
///     // handle response
/// }
/// ```
pub struct FetchPool {
    /// The host-provided HTTP transport.
    client: Box<dyn HttpClient>,

    /// Maximum number of requests that may be in-flight simultaneously.
    max_concurrent: usize,

    /// Requests waiting to be dispatched, ordered by priority.
    queue: Mutex<BinaryHeap<PrioritizedRequest>>,

    /// URLs that are currently queued *or* in-flight (dedup set).
    /// An entry is inserted on `enqueue` and removed when `poll`
    /// returns the completed response, preventing duplicate requests
    /// for the same URL.
    known_urls: Mutex<HashSet<String>>,

    /// URLs currently in-flight (sent to the HTTP client, awaiting response).
    ///
    /// Using a set of URLs instead of a plain counter eliminates
    /// counting mismatches that arise when a deduplicating HTTP client
    /// wrapper (e.g. [`SharedHttpClient`](crate::SharedHttpClient))
    /// silently coalesces a re-sent URL with an existing in-flight
    /// request.  With a set, `flush` inserts the URL (idempotent if
    /// already present), `poll` removes it, and `force_cancel` removes
    /// it immediately -- the set length is always accurate.
    in_flight_urls: Mutex<HashSet<String>>,

    /// URLs that were force-cancelled while already in-flight.
    ///
    /// When the HTTP response for such a URL arrives in [`poll`], it
    /// is silently discarded (the URL was already removed from
    /// `in_flight_urls` by `force_cancel`).  This set exists so that
    /// `poll` can distinguish ghost responses from real completions.
    cancelled_in_flight: Mutex<HashSet<String>>,

    /// Monotonic insertion counter used to preserve FIFO ordering among
    /// equal-priority requests.
    sequence: AtomicU64,
}

impl FetchPool {
    /// Create a new pool wrapping `client` with at most `max_concurrent`
    /// simultaneous downloads.
    ///
    /// A `max_concurrent` of 0 is silently clamped to 1.
    pub fn new(client: Box<dyn HttpClient>, max_concurrent: usize) -> Self {
        Self {
            client,
            max_concurrent: max_concurrent.max(1),
            queue: Mutex::new(BinaryHeap::new()),
            known_urls: Mutex::new(HashSet::new()),
            in_flight_urls: Mutex::new(HashSet::new()),
            cancelled_in_flight: Mutex::new(HashSet::new()),
            sequence: AtomicU64::new(0),
        }
    }

    /// Add a URL to the priority queue.
    ///
    /// Lower `priority` = dispatched sooner.  Duplicate URLs (already
    /// queued or in-flight) are silently ignored.
    ///
    /// The request is **not** sent until [`flush`](Self::flush) is called
    /// (or implicitly via [`poll`](Self::poll)).  This allows the caller
    /// to batch-enqueue a frame's worth of tiles before dispatching, so
    /// the priority heap can sort them correctly.
    pub fn enqueue(&self, request: HttpRequest, priority: f64) {
        let url = request.url.clone();
        let mut known = match self.known_urls.lock() {
            Ok(u) => u,
            Err(_) => return,
        };
        if !known.insert(url.clone()) {
            return;
        }
        drop(known);

        // If this URL was previously force-cancelled while in-flight but is
        // now being re-enqueued, the eventual response should satisfy the new
        // logical request rather than being discarded as a ghost.  Clear the
        // ghost marker here so `poll` treats the next completion as real.
        if let Ok(mut cancelled) = self.cancelled_in_flight.lock() {
            cancelled.remove(&url);
        }

        if let Ok(mut queue) = self.queue.lock() {
            queue.push(PrioritizedRequest {
                request,
                priority,
                sequence: self.sequence.fetch_add(1, AtomicOrdering::Relaxed),
            });
        }
    }

    /// Dispatch queued requests up to the concurrency limit.
    ///
    /// The highest-priority (lowest score) requests are sent first.
    /// Call this once per frame after all [`enqueue`](Self::enqueue)
    /// calls for that frame.
    pub fn flush(&self) {
        let mut in_flight = match self.in_flight_urls.lock() {
            Ok(f) => f,
            Err(_) => return,
        };
        let mut queue = match self.queue.lock() {
            Ok(q) => q,
            Err(_) => return,
        };

        while in_flight.len() < self.max_concurrent {
            match queue.pop() {
                Some(req) => {
                    let url = req.request.url.clone();
                    self.client.send(req.request);
                    in_flight.insert(url);
                }
                None => break,
            }
        }
    }

    /// Remove a URL from the pending queue if it has not been sent yet.
    ///
    /// Returns `true` if the URL was found and removed.  Already in-flight
    /// requests are not affected.
    pub fn cancel(&self, url: &str) -> bool {
        let in_queue = self
            .queue
            .lock()
            .is_ok_and(|q| q.iter().any(|r| r.request.url == url));
        if !in_queue {
            return false;
        }

        // Remove from the dedup set.
        if let Ok(mut known) = self.known_urls.lock() {
            known.remove(url);
        }

        // Rebuild the heap without the cancelled URL.  O(n) but
        // cancellation is infrequent.
        if let Ok(mut queue) = self.queue.lock() {
            let old: Vec<_> = queue.drain().collect();
            for req in old {
                if req.request.url != url {
                    queue.push(req);
                }
            }
        }
        true
    }

    /// Remove a URL from the dedup set regardless of whether it is
    /// queued or in-flight.
    ///
    /// If the URL is still in the queue it is also removed from the
    /// heap.  If it is already in-flight, the concurrency slot is
    /// reclaimed immediately so that new requests can be dispatched
    /// without waiting for the ghost HTTP response to arrive.  The
    /// URL is recorded in an internal set so that [`poll`](Self::poll)
    /// does not double-decrement `in_flight` when the ghost response
    /// eventually completes.
    ///
    /// This allows the same URL to be re-enqueued immediately, which
    /// is essential when the tile manager cancels a pending tile and
    /// then re-requests it on a subsequent frame.
    pub fn force_cancel(&self, url: &str) {
        // Check whether the URL is still in the queue (not yet sent).
        let was_queued = self
            .queue
            .lock()
            .is_ok_and(|q| q.iter().any(|r| r.request.url == url));

        // Always remove from the dedup set so re-enqueue is possible.
        if let Ok(mut known) = self.known_urls.lock() {
            known.remove(url);
        }

        if was_queued {
            // Remove from the heap -- the request was never sent.
            if let Ok(mut queue) = self.queue.lock() {
                let old: Vec<_> = queue.drain().collect();
                for req in old {
                    if req.request.url != url {
                        queue.push(req);
                    }
                }
            }
        } else {
            // The URL is in-flight.  Immediately reclaim the concurrency
            // slot so new requests can be dispatched without waiting for
            // the ghost HTTP response to arrive.
            if let Ok(mut in_flight) = self.in_flight_urls.lock() {
                in_flight.remove(url);
            }
            // Record the ghost so `poll` knows the eventual response is
            // orphaned and should not be forwarded to the caller.
            if let Ok(mut cancelled) = self.cancelled_in_flight.lock() {
                cancelled.insert(url.to_owned());
            }
        }
    }

    /// Discard all queued (not yet sent) requests.
    ///
    /// In-flight requests are unaffected -- they will still appear in
    /// future [`poll`](Self::poll) results.
    pub fn clear_queue(&self) {
        if let Ok(mut queue) = self.queue.lock() {
            // Remove queued URLs from the known set so they can be
            // re-enqueued later if needed.
            if let Ok(mut known) = self.known_urls.lock() {
                for req in queue.iter() {
                    known.remove(&req.request.url);
                }
            }
            queue.clear();
        }
    }

    /// Number of requests waiting in the queue (not yet sent).
    pub fn queued_count(&self) -> usize {
        self.queue.lock().map(|q| q.len()).unwrap_or(0)
    }

    /// Maximum number of concurrent in-flight requests.
    #[inline]
    pub fn max_concurrent(&self) -> usize {
        self.max_concurrent
    }

    /// Number of requests currently in-flight (sent, awaiting response).
    pub fn in_flight_count(&self) -> usize {
        self.in_flight_urls.lock().map(|g| g.len()).unwrap_or(0)
    }

    /// Number of URLs currently tracked by the dedup set.
    pub fn known_count(&self) -> usize {
        self.known_urls.lock().map(|k| k.len()).unwrap_or(0)
    }

    /// Number of URLs recorded as force-cancelled while in-flight.
    pub fn cancelled_in_flight_count(&self) -> usize {
        self.cancelled_in_flight
            .lock()
            .map(|set| set.len())
            .unwrap_or(0)
    }

    /// Returns `true` if `url` is known to the pool (queued or in-flight).
    pub fn is_known(&self, url: &str) -> bool {
        self.known_urls.lock().is_ok_and(|k| k.contains(url))
    }

    /// Poll the underlying [`HttpClient`] for completed responses and
    /// release their concurrency slots.
    ///
    /// Any freed slots are immediately filled from the priority queue
    /// (implicit flush).
    ///
    /// Returns `(url, Result<HttpResponse, String>)` for each completed
    /// request.
    pub fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
        let results = self.client.poll();

        if !results.is_empty() {
            // Clean up ghost responses (force-cancelled while in-flight)
            // and release concurrency slots for real completions.
            let mut cancelled = self.cancelled_in_flight.lock().ok();
            let mut in_flight = self.in_flight_urls.lock().ok();

            for (url, _) in &results {
                // If this URL was force-cancelled, it is a ghost -- the
                // slot was already freed in force_cancel.  Just clean up
                // the tracking set.
                if let Some(ref mut set) = cancelled {
                    if set.remove(url.as_str()) {
                        continue;
                    }
                }
                // Real completion: free the concurrency slot.
                if let Some(ref mut urls) = in_flight {
                    urls.remove(url.as_str());
                }
            }
            drop(cancelled);
            drop(in_flight);

            // Remove completed URLs from the dedup set.
            if let Ok(mut known) = self.known_urls.lock() {
                for (url, _) in &results {
                    known.remove(url);
                }
            }
        }

        // Fill any freed slots from the queue.
        self.flush();

        results
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::io::shared_http_client::SharedHttpClient;
    use std::sync::Arc;

    // -- Helpers ----------------------------------------------------------

    /// Mock that instantly completes every request on the next `poll`.
    struct InstantMockClient {
        sent: Mutex<Vec<String>>,
    }

    impl InstantMockClient {
        fn new() -> Self {
            Self {
                sent: Mutex::new(Vec::new()),
            }
        }
    }

    impl HttpClient for InstantMockClient {
        fn send(&self, request: HttpRequest) {
            self.sent.lock().unwrap().push(request.url);
        }

        fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
            let sent = std::mem::take(&mut *self.sent.lock().unwrap());
            sent.into_iter()
                .map(|url| {
                    (
                        url,
                        Ok(HttpResponse {
                            status: 200,
                            body: Vec::new(),
                            headers: Vec::new(),
                        }),
                    )
                })
                .collect()
        }
    }

    /// Mock that records send order but never completes.
    struct DeferredMockClient {
        sent: Mutex<Vec<String>>,
    }

    impl DeferredMockClient {
        fn new() -> Self {
            Self {
                sent: Mutex::new(Vec::new()),
            }
        }
    }

    impl HttpClient for DeferredMockClient {
        fn send(&self, request: HttpRequest) {
            self.sent.lock().unwrap().push(request.url);
        }

        fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
            Vec::new()
        }
    }

    /// Shared mock for inspecting send order after handing ownership
    /// to FetchPool.
    struct SharedMock(Arc<Mutex<Vec<String>>>);

    impl HttpClient for SharedMock {
        fn send(&self, request: HttpRequest) {
            self.0.lock().unwrap().push(request.url);
        }
        fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
            Vec::new()
        }
    }

    #[derive(Clone, Default)]
    #[allow(clippy::type_complexity)]
    struct RecordingClient {
        sent: Arc<Mutex<Vec<String>>>,
        responses: Arc<Mutex<Vec<(String, Result<HttpResponse, String>)>>>,
    }

    impl RecordingClient {
        fn sent_urls(&self) -> Vec<String> {
            self.sent.lock().unwrap().clone()
        }

        fn complete(&self, url: &str) {
            self.responses.lock().unwrap().push((
                url.to_owned(),
                Ok(HttpResponse {
                    status: 200,
                    body: Vec::new(),
                    headers: Vec::new(),
                }),
            ));
        }
    }

    impl HttpClient for RecordingClient {
        fn send(&self, request: HttpRequest) {
            self.sent.lock().unwrap().push(request.url);
        }

        fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
            std::mem::take(&mut *self.responses.lock().unwrap())
        }
    }

    // -- Concurrency limit ------------------------------------------------

    #[test]
    fn respects_concurrency_limit() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 2);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("b"), 2.0);
        pool.enqueue(HttpRequest::get("c"), 3.0);
        pool.flush();

        // Only 2 sent, 1 remains queued.
        assert_eq!(pool.in_flight_count(), 2);
        assert_eq!(pool.queued_count(), 1);
    }

    #[test]
    fn freed_slots_dispatch_queued() {
        let pool = FetchPool::new(Box::new(InstantMockClient::new()), 1);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("b"), 2.0);
        pool.flush();

        // 'a' sent (lower priority value), 'b' queued.
        assert_eq!(pool.in_flight_count(), 1);
        assert_eq!(pool.queued_count(), 1);

        // Poll completes 'a', implicit flush sends 'b'.
        let results = pool.poll();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].0, "a");
        assert_eq!(pool.queued_count(), 0);
        assert_eq!(pool.in_flight_count(), 1); // 'b' now in-flight
    }

    // -- Priority ordering ------------------------------------------------

    #[test]
    fn priority_order_nearest_first() {
        let sent = Arc::new(Mutex::new(Vec::new()));
        let pool = FetchPool::new(Box::new(SharedMock(Arc::clone(&sent))), 10);

        // Enqueue in non-priority order.
        pool.enqueue(HttpRequest::get("far"), 100.0);
        pool.enqueue(HttpRequest::get("near"), 1.0);
        pool.enqueue(HttpRequest::get("mid"), 50.0);

        // Flush dispatches all three; heap sorts by priority.
        pool.flush();

        let order = sent.lock().unwrap().clone();
        assert_eq!(order.len(), 3);
        assert_eq!(order[0], "near");
        assert_eq!(order[1], "mid");
        assert_eq!(order[2], "far");
    }

    #[test]
    fn equal_priority_preserves_enqueue_order() {
        let sent = Arc::new(Mutex::new(Vec::new()));
        let pool = FetchPool::new(Box::new(SharedMock(Arc::clone(&sent))), 10);

        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("b"), 1.0);
        pool.enqueue(HttpRequest::get("c"), 1.0);
        pool.flush();

        let order = sent.lock().unwrap().clone();
        assert_eq!(order, vec!["a", "b", "c"]);
    }

    // -- Duplicate suppression --------------------------------------------

    #[test]
    fn duplicate_enqueue_ignored() {
        let sent = Arc::new(Mutex::new(Vec::new()));
        let pool = FetchPool::new(Box::new(SharedMock(Arc::clone(&sent))), 10);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("a"), 2.0); // duplicate -- silently dropped
        pool.flush();

        assert_eq!(pool.in_flight_count(), 1);
        assert_eq!(sent.lock().unwrap().len(), 1);
    }

    #[test]
    fn duplicate_suppressed_while_in_flight() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 10);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.flush(); // 'a' is now in-flight

        // Try to enqueue 'a' again while it is in-flight.
        pool.enqueue(HttpRequest::get("a"), 2.0);
        assert_eq!(pool.queued_count(), 0, "should not re-queue in-flight URL");
    }

    #[test]
    fn can_re_enqueue_after_completion() {
        let pool = FetchPool::new(Box::new(InstantMockClient::new()), 10);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.flush();

        // Complete it.
        let results = pool.poll();
        assert_eq!(results.len(), 1);

        // Now re-enqueue should work.
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.flush();
        assert_eq!(pool.in_flight_count(), 1);
    }

    // -- Cancel -----------------------------------------------------------

    #[test]
    fn cancel_removes_queued_request() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 1);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("b"), 2.0);
        pool.enqueue(HttpRequest::get("c"), 3.0);
        pool.flush();

        // 'a' sent (priority 1.0), 'b' and 'c' queued.
        assert_eq!(pool.queued_count(), 2);
        assert!(pool.cancel("b"));
        assert_eq!(pool.queued_count(), 1);
    }

    #[test]
    fn cancel_nonexistent_returns_false() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 10);
        assert!(!pool.cancel("nope"));
    }

    #[test]
    fn cancel_in_flight_returns_false() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 10);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.flush();
        // 'a' is in-flight, not queued.
        assert!(!pool.cancel("a"));
    }

    // -- Clear queue ------------------------------------------------------

    #[test]
    fn clear_queue_discards_pending() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 1);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("b"), 2.0);
        pool.enqueue(HttpRequest::get("c"), 3.0);
        pool.flush();

        assert_eq!(pool.queued_count(), 2);
        pool.clear_queue();
        assert_eq!(pool.queued_count(), 0);
        // In-flight request is unaffected.
        assert_eq!(pool.in_flight_count(), 1);
    }

    #[test]
    fn cleared_urls_can_be_re_enqueued() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 1);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("b"), 2.0);
        pool.flush(); // sends 'a', 'b' queued
        pool.clear_queue(); // drops 'b' from queue and known set

        pool.enqueue(HttpRequest::get("b"), 2.0); // should succeed
        assert_eq!(pool.queued_count(), 1);
    }

    // -- Zero-concurrency edge case ---------------------------------------

    #[test]
    fn zero_concurrency_clamped_to_one() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 0);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.flush();
        assert_eq!(pool.in_flight_count(), 1);
    }

    // -- Empty pool -------------------------------------------------------

    #[test]
    fn poll_empty_returns_empty() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 10);
        assert!(pool.poll().is_empty());
        assert_eq!(pool.in_flight_count(), 0);
        assert_eq!(pool.queued_count(), 0);
    }

    #[test]
    fn flush_empty_is_noop() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 10);
        pool.flush(); // should not panic
        assert_eq!(pool.in_flight_count(), 0);
    }

    // -- Force-cancel ghost slot reclamation ------------------------------

    #[test]
    fn force_cancel_immediately_reclaims_in_flight_slot() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 2);

        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("b"), 2.0);
        pool.enqueue(HttpRequest::get("c"), 3.0);
        pool.flush();

        // 'a' and 'b' in-flight, 'c' queued.
        assert_eq!(pool.in_flight_count(), 2);
        assert_eq!(pool.queued_count(), 1);

        // Force-cancel 'a' while it is in-flight.
        pool.force_cancel("a");

        // The slot should be reclaimed immediately.
        assert_eq!(
            pool.in_flight_count(),
            1,
            "ghost slot must be reclaimed immediately"
        );
        assert_eq!(pool.cancelled_in_flight_count(), 1);

        // Flush should now dispatch 'c' into the freed slot.
        pool.flush();
        assert_eq!(
            pool.in_flight_count(),
            2,
            "freed slot should accept queued request"
        );
        assert_eq!(pool.queued_count(), 0);
    }

    #[test]
    fn ghost_response_does_not_double_decrement_in_flight() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 2);

        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("b"), 2.0);
        pool.flush();
        assert_eq!(pool.in_flight_count(), 2);

        // Force-cancel 'a': slot reclaimed immediately.
        pool.force_cancel("a");
        assert_eq!(pool.in_flight_count(), 1);
        assert_eq!(pool.cancelled_in_flight_count(), 1);

        // poll returns nothing (DeferredMockClient), so ghost persists.
        let _ = pool.poll();
        assert_eq!(pool.in_flight_count(), 1);
        assert_eq!(
            pool.cancelled_in_flight_count(),
            1,
            "ghost persists until response arrives"
        );
    }

    #[test]
    fn force_cancel_allows_re_enqueue_of_in_flight_url() {
        let pool = FetchPool::new(Box::new(DeferredMockClient::new()), 10);
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.flush();
        assert_eq!(pool.in_flight_count(), 1);

        pool.force_cancel("a");
        assert_eq!(pool.in_flight_count(), 0);

        // Should be able to re-enqueue immediately.
        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.flush();
        assert_eq!(pool.in_flight_count(), 1);
    }

    #[test]
    fn reenqueue_after_force_cancel_with_shared_dedup_completes_cleanly() {
        let inner = RecordingClient::default();
        let shared = SharedHttpClient::new(Box::new(inner.clone()));
        let pool = FetchPool::new(Box::new(shared), 2);

        pool.enqueue(HttpRequest::get("a"), 1.0);
        pool.enqueue(HttpRequest::get("b"), 2.0);
        pool.flush();
        assert_eq!(pool.in_flight_count(), 2);
        assert_eq!(inner.sent_urls(), vec!["a".to_string(), "b".to_string()]);

        // Cancel the in-flight request and immediately re-enqueue the same URL.
        pool.force_cancel("a");
        assert_eq!(pool.in_flight_count(), 1);
        assert_eq!(pool.cancelled_in_flight_count(), 1);

        pool.enqueue(HttpRequest::get("a"), 0.5);
        pool.flush();

        // SharedHttpClient should dedup the resend against the original
        // in-flight request, so no second network send for "a" occurs.
        assert_eq!(pool.in_flight_count(), 2);
        assert_eq!(inner.sent_urls(), vec!["a".to_string(), "b".to_string()]);
        assert_eq!(
            pool.cancelled_in_flight_count(),
            0,
            "re-enqueue should clear ghost marker"
        );

        // When the original response for "a" arrives, it must satisfy the new
        // logical request rather than being discarded as a ghost.
        inner.complete("a");
        let results = pool.poll();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].0, "a");
        assert_eq!(
            pool.in_flight_count(),
            1,
            "completion should retire the re-enqueued URL"
        );
    }
}