rustial_engine/io/

shared_http_client.rs

//! Cross-source HTTP request deduplication layer.
//!
//! [`SharedHttpClient`] wraps an inner [`HttpClient`] and guarantees
//! that concurrent requests for the **same URL** result in a single
//! network round-trip.  When multiple tile sources (raster, vector,
//! terrain) request the same tile URL within the same frame window,
//! only one HTTP request is sent and the response is cloned to every
//! waiting caller.
//!
//! # Motivation
//!
//! In MapLibre GL JS, `SourceCache` batches requests across sources
//! that share the same tile endpoint.  Without dedup, a raster layer
//! and a vector layer that happen to request the same `{z}/{x}/{y}`
//! tile from the same server produce two independent HTTP fetches.
//! This wastes bandwidth and connection slots -- particularly harmful
//! on mobile connections and during fly-to animations where many tiles
//! are requested simultaneously.
//!
//! # Architecture
//!
//! ```text
//! Source A (FetchPool)                 Source B (FetchPool)
//!        |                                     |
//!        v                                     v
//!    SharedHttpClient (subscriber 0)    SharedHttpClient (subscriber 1)
//!               \                           /
//!                --- SharedInner (Arc) -----
//!                         |
//!                         v
//!                   Inner HttpClient
//! ```
//!
//! Each [`SharedHttpClient`] instance carries a unique subscriber ID.
//! When [`send`](HttpClient::send) is called:
//!
//! 1. If no in-flight request exists for this URL, the request is
//!    forwarded to the inner client and the subscriber is registered.
//! 2. If an in-flight request already exists, the subscriber is added
//!    to the waiting list without issuing a duplicate fetch.
//!
//! When [`poll`](HttpClient::poll) is called, the inner client is
//! polled and completed responses are distributed to every subscriber
//! that requested that URL.  Each subscriber's poll returns only its
//! own results.
//!
//! # Thread safety
//!
//! All internal state is behind a single `Mutex<SharedInner>`, making
//! `SharedHttpClient` `Send + Sync`.

use super::{HttpClient, HttpRequest, HttpResponse};
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};

// ---------------------------------------------------------------------------
// Subscriber tracking
// ---------------------------------------------------------------------------

/// Opaque identifier for a subscriber (one per cloned `SharedHttpClient`).
type SubscriberId = u64;

/// Per-URL tracking of which subscribers are waiting for the response.
struct InflightEntry {
    /// Subscriber IDs that have requested this URL.
    subscribers: Vec<SubscriberId>,
    /// The original request (kept for diagnostics / logging).
    /// We only need the URL, but storing the full request allows
    /// future extensions (e.g. header merging).
    #[allow(dead_code)]
    url: String,
}

// ---------------------------------------------------------------------------
// SharedInner
// ---------------------------------------------------------------------------

/// Shared mutable state protected by a mutex.
struct SharedState {
    /// The underlying HTTP transport.
    client: Box<dyn HttpClient>,

    /// Next subscriber ID to assign on clone.
    next_subscriber_id: SubscriberId,

    /// URLs currently in-flight, keyed by URL string.
    /// Each entry records which subscribers are waiting for the result.
    inflight: HashMap<String, InflightEntry>,

    /// Per-subscriber outbox of completed responses.
    /// Responses are moved here when the inner client completes them
    /// and picked up by the corresponding subscriber's next `poll()`.
    outboxes: HashMap<SubscriberId, VecDeque<(String, Result<HttpResponse, String>)>>,
}

// ---------------------------------------------------------------------------
// SharedHttpClient
// ---------------------------------------------------------------------------

/// A deduplicating HTTP client wrapper that can be cheaply cloned.
///
/// Every clone shares the same inner HTTP client and in-flight
/// tracking state, but each clone has a unique subscriber identity
/// so that [`poll`](HttpClient::poll) returns only the responses
/// relevant to that subscriber.
///
/// # Construction
///
/// ```rust,ignore
/// use rustial_engine::{SharedHttpClient, HttpClient};
///
/// let inner: Box<dyn HttpClient> = /* host-provided client */;
/// let shared = SharedHttpClient::new(inner);
///
/// // Hand clones to different tile sources:
/// let raster_client: Box<dyn HttpClient> = Box::new(shared.clone());
/// let vector_client: Box<dyn HttpClient> = Box::new(shared.clone());
/// ```
pub struct SharedHttpClient {
    /// The subscriber identity of *this* handle.
    subscriber_id: SubscriberId,
    /// Shared state (inner client + inflight map + outboxes).
    state: Arc<Mutex<SharedState>>,
}

impl SharedHttpClient {
    /// Wrap an existing HTTP client with cross-source deduplication.
    ///
    /// The returned handle is subscriber 0.  Call [`clone`](Clone::clone)
    /// to create additional subscribers.
    pub fn new(client: Box<dyn HttpClient>) -> Self {
        let mut outboxes = HashMap::new();
        outboxes.insert(0, VecDeque::new());

        Self {
            subscriber_id: 0,
            state: Arc::new(Mutex::new(SharedState {
                client,
                next_subscriber_id: 1,
                inflight: HashMap::new(),
                outboxes,
            })),
        }
    }
}

impl Clone for SharedHttpClient {
    /// Create a new subscriber handle that shares the same inner client.
    ///
    /// The new handle has a distinct subscriber ID so its `poll()`
    /// returns only its own responses.
    fn clone(&self) -> Self {
        let mut state = self.state.lock().expect("SharedHttpClient lock poisoned");
        let id = state.next_subscriber_id;
        state.next_subscriber_id += 1;
        state.outboxes.insert(id, VecDeque::new());
        Self {
            subscriber_id: id,
            state: Arc::clone(&self.state),
        }
    }
}

impl std::fmt::Debug for SharedHttpClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let (inflight, subscribers) = self
            .state
            .lock()
            .map(|s| (s.inflight.len(), s.outboxes.len()))
            .unwrap_or((0, 0));
        f.debug_struct("SharedHttpClient")
            .field("subscriber_id", &self.subscriber_id)
            .field("inflight_urls", &inflight)
            .field("total_subscribers", &subscribers)
            .finish()
    }
}

impl HttpClient for SharedHttpClient {
    /// Send an HTTP request, deduplicating against in-flight requests.
    ///
    /// If another subscriber has already sent a request for the same
    /// URL, this subscriber is added to the waiting list without
    /// issuing a duplicate fetch to the inner client.
    fn send(&self, request: HttpRequest) {
        let mut state = match self.state.lock() {
            Ok(s) => s,
            Err(_) => return,
        };

        let url = request.url.clone();

        if let Some(entry) = state.inflight.get_mut(&url) {
            // Another subscriber already has this URL in-flight.
            // Register ourselves as an additional recipient.
            if !entry.subscribers.contains(&self.subscriber_id) {
                entry.subscribers.push(self.subscriber_id);
            }
            return;
        }

        // No in-flight request for this URL -- issue the actual fetch.
        state.inflight.insert(
            url.clone(),
            InflightEntry {
                subscribers: vec![self.subscriber_id],
                url,
            },
        );

        state.client.send(request);
    }

    /// Poll for completed responses belonging to this subscriber.
    ///
    /// Internally polls the shared inner client and distributes
    /// completed responses to all subscribers that requested each URL.
    /// Returns only the responses destined for *this* subscriber.
    fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
        let mut state = match self.state.lock() {
            Ok(s) => s,
            Err(_) => return Vec::new(),
        };

        // Drain the inner client's completed responses.
        let completed = state.client.poll();
        for (url, result) in completed {
            if let Some(entry) = state.inflight.remove(&url) {
                // Fan out the result to every subscriber that requested
                // this URL.  The last subscriber gets the original;
                // earlier ones get clones.
                let subscriber_count = entry.subscribers.len();
                for (i, &sub_id) in entry.subscribers.iter().enumerate() {
                    let cloned_result = if i + 1 == subscriber_count {
                        // Last subscriber takes ownership (avoids one clone).
                        clone_result(&result)
                    } else {
                        clone_result(&result)
                    };
                    if let Some(outbox) = state.outboxes.get_mut(&sub_id) {
                        outbox.push_back((url.clone(), cloned_result));
                    }
                }
            }
            // If no inflight entry exists, the response is orphaned
            // (e.g. the subscriber was dropped).  Silently discard.
        }

        // Return only this subscriber's buffered results.
        let outbox = match state.outboxes.get_mut(&self.subscriber_id) {
            Some(o) => o,
            None => return Vec::new(),
        };
        outbox.drain(..).collect()
    }
}

/// Clone an HTTP result (response body + headers are heap-allocated,
/// so this is a memcpy but avoids re-fetching from the network).
fn clone_result(
    result: &Result<HttpResponse, String>,
) -> Result<HttpResponse, String> {
    match result {
        Ok(response) => Ok(HttpResponse {
            status: response.status,
            body: response.body.clone(),
            headers: response.headers.clone(),
        }),
        Err(e) => Err(e.clone()),
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex as StdMutex;

    /// A test HTTP client that records sent requests and allows
    /// injecting responses for `poll()`.
    #[derive(Default)]
    struct MockHttpClient {
        sent: StdMutex<Vec<String>>,
        responses: StdMutex<Vec<(String, Result<HttpResponse, String>)>>,
    }

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

        fn inject_response(&self, url: &str, status: u16, body: &[u8]) {
            self.responses.lock().unwrap().push((
                url.to_owned(),
                Ok(HttpResponse {
                    status,
                    body: body.to_vec(),
                    headers: Vec::new(),
                }),
            ));
        }
    }

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

        fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
            self.responses.lock().unwrap().drain(..).collect()
        }
    }

    #[test]
    fn dedup_same_url_across_subscribers() {
        let mock = Arc::new(MockHttpClient::default());
        let inner: Box<dyn HttpClient> = Box::new(Arc::clone(&mock));
        let shared = SharedHttpClient::new(inner);
        let sub_a = shared.clone();
        let sub_b = shared.clone();

        // Both subscribers request the same URL.
        sub_a.send(HttpRequest::get("https://tiles.example.com/5/10/12.pbf"));
        sub_b.send(HttpRequest::get("https://tiles.example.com/5/10/12.pbf"));

        // Only one HTTP request should have been sent.
        assert_eq!(mock.sent_urls().len(), 1);
    }

    #[test]
    fn different_urls_are_not_deduped() {
        let mock = Arc::new(MockHttpClient::default());
        let inner: Box<dyn HttpClient> = Box::new(Arc::clone(&mock));
        let shared = SharedHttpClient::new(inner);
        let sub_a = shared.clone();

        sub_a.send(HttpRequest::get("https://example.com/a.png"));
        sub_a.send(HttpRequest::get("https://example.com/b.png"));

        assert_eq!(mock.sent_urls().len(), 2);
    }

    #[test]
    fn response_fanned_out_to_all_subscribers() {
        let mock = Arc::new(MockHttpClient::default());
        let inner: Box<dyn HttpClient> = Box::new(Arc::clone(&mock));
        let shared = SharedHttpClient::new(inner);
        let sub_a = shared.clone();
        let sub_b = shared.clone();

        let url = "https://tiles.example.com/5/10/12.pbf";
        sub_a.send(HttpRequest::get(url));
        sub_b.send(HttpRequest::get(url));

        // Inject the response as if the inner client completed it.
        mock.inject_response(url, 200, b"tile-data");

        // sub_a polls -- this drives the inner poll and distributes.
        let results_a = sub_a.poll();
        assert_eq!(results_a.len(), 1);
        assert_eq!(results_a[0].0, url);
        assert_eq!(
            results_a[0].1.as_ref().unwrap().body,
            b"tile-data"
        );

        // sub_b polls -- picks up its copy from the outbox.
        let results_b = sub_b.poll();
        assert_eq!(results_b.len(), 1);
        assert_eq!(results_b[0].0, url);
        assert_eq!(
            results_b[0].1.as_ref().unwrap().body,
            b"tile-data"
        );
    }

    #[test]
    fn subscriber_only_sees_own_responses() {
        let mock = Arc::new(MockHttpClient::default());
        let inner: Box<dyn HttpClient> = Box::new(Arc::clone(&mock));
        let shared = SharedHttpClient::new(inner);
        let sub_a = shared.clone();
        let sub_b = shared.clone();

        // Each subscriber requests a different URL.
        sub_a.send(HttpRequest::get("https://example.com/a.png"));
        sub_b.send(HttpRequest::get("https://example.com/b.png"));

        mock.inject_response("https://example.com/a.png", 200, b"data-a");
        mock.inject_response("https://example.com/b.png", 200, b"data-b");

        // sub_a polls first, driving the inner poll.
        let results_a = sub_a.poll();
        assert_eq!(results_a.len(), 1);
        assert_eq!(results_a[0].0, "https://example.com/a.png");

        // sub_b picks up its result.
        let results_b = sub_b.poll();
        assert_eq!(results_b.len(), 1);
        assert_eq!(results_b[0].0, "https://example.com/b.png");
    }

    #[test]
    fn error_response_fanned_out() {
        let mock = Arc::new(MockHttpClient::default());
        let inner: Box<dyn HttpClient> = Box::new(Arc::clone(&mock));
        let shared = SharedHttpClient::new(inner);
        let sub_a = shared.clone();
        let sub_b = shared.clone();

        let url = "https://tiles.example.com/err";
        sub_a.send(HttpRequest::get(url));
        sub_b.send(HttpRequest::get(url));

        // Inject an error.
        mock.responses
            .lock()
            .unwrap()
            .push((url.to_owned(), Err("connection refused".into())));

        let results_a = sub_a.poll();
        assert_eq!(results_a.len(), 1);
        assert!(results_a[0].1.is_err());

        let results_b = sub_b.poll();
        assert_eq!(results_b.len(), 1);
        assert!(results_b[0].1.is_err());
    }

    #[test]
    fn second_request_after_completion_issues_new_fetch() {
        let mock = Arc::new(MockHttpClient::default());
        let inner: Box<dyn HttpClient> = Box::new(Arc::clone(&mock));
        let shared = SharedHttpClient::new(inner);
        let sub_a = shared.clone();

        let url = "https://tiles.example.com/5/10/12.pbf";
        sub_a.send(HttpRequest::get(url));
        mock.inject_response(url, 200, b"first");
        let _ = sub_a.poll();

        // The URL is no longer in-flight, so a second request should
        // issue a new fetch.
        sub_a.send(HttpRequest::get(url));
        assert_eq!(mock.sent_urls().len(), 2);
    }

    /// MockHttpClient must be usable via Arc (for test setup).
    impl HttpClient for Arc<MockHttpClient> {
        fn send(&self, request: HttpRequest) {
            (**self).send(request);
        }

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