rustial_engine/

tile_request_coordinator.rs

// ---------------------------------------------------------------------------
//! # Cross-source tile request coordinator
//!
//! [`TileRequestCoordinator`] centralises tile-request scheduling across
//! all active sources (raster tile layers, streamed vector sources, terrain
//! DEM sources) so that requests are batched, prioritised, and rate-limited
//! at the **map** level rather than the individual source level.
//!
//! ## Motivation
//!
//! In MapLibre GL JS the `SourceCache` object batches tile operations
//! across sources and the `Scheduler` enforces a global worker budget.
//! Without a coordinator, each Rustial source independently queues
//! requests against its own `FetchPool` / `HttpClient`, leading to:
//!
//! - **Bandwidth contention** -- raster and vector sources compete for
//!   the same connection slots, starving whichever starts later.
//! - **Priority inversion** -- a burst of low-priority vector tile
//!   requests can delay the more user-visible raster imagery.
//! - **Redundant selection** -- each source independently computes the
//!   same tile-coverage set from the camera, duplicating work.
//!
//! ## Architecture
//!
//! ```text
//! MapState::update_tile_layers()
//!     |
//!     +-- raster TileLayer::update_with_view()
//!     |       reports request_count -> coordinator
//!     |
//!     +-- vector SourceLayer::update_with_view()  (x N)
//!     |       reports request_count -> coordinator
//!     |
//!     +-- TerrainManager::update()
//!             reports request_count -> coordinator
//!
//! coordinator.finish_frame()
//!     -> computes per-source budget for next frame
//!     -> records cross-source diagnostics
//! ```
//!
//! ## Budget allocation
//!
//! Each frame, the coordinator distributes a global request budget
//! across source classes using configurable priority weights.
//! The default weights (`raster=3, vector=2, terrain=1`) ensure that
//! raster imagery -- the most visible layer -- receives the lion's
//! share of available request slots during contention.
//!
//! When a source class has no pending requests, its unused budget is
//! redistributed to the remaining classes proportionally.
//!
//! ## Integration
//!
//! The coordinator is **non-invasive**: it does not modify `TileManager`
//! internals.  Instead, `MapState` queries the coordinator for per-source
//! request budgets before calling each source's update method, and
//! `TileSelectionConfig::max_requests_per_frame` limits how many new
//! tile requests `TileManager` issues in a single update pass.
// ---------------------------------------------------------------------------

use rustial_math::TileId;
use std::collections::HashSet;

// ---------------------------------------------------------------------------
// Configuration
// ---------------------------------------------------------------------------

/// Priority class for a tile source.
///
/// Higher-priority classes receive a larger share of the global
/// per-frame request budget.  The numeric weight is used directly
/// in the proportional budget allocation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum SourcePriority {
    /// Raster imagery -- the most user-visible layer.
    Raster,
    /// Streamed vector tile sources.
    Vector,
    /// Terrain DEM elevation data.
    Terrain,
}

impl SourcePriority {
    /// Default priority weight.
    ///
    /// Raster gets the highest share because blank tiles are the
    /// most visible artefact.  Vector and terrain are lower because
    /// they are either overlays or only affect depth/elevation.
    fn default_weight(self) -> u32 {
        match self {
            Self::Raster => 3,
            Self::Vector => 2,
            Self::Terrain => 1,
        }
    }
}

/// Configuration for the cross-source request coordinator.
#[derive(Debug, Clone)]
pub struct CoordinatorConfig {
    /// Maximum number of new tile requests that may be issued across
    /// **all** sources in a single frame.
    ///
    /// This global cap prevents a sudden camera move from flooding the
    /// network with hundreds of queued requests.  The budget is split
    /// among source classes proportionally to their weights.
    ///
    /// Set to `0` to disable coordination (each source uses its own
    /// internal limits).  Defaults to 32.
    pub global_request_budget: usize,

    /// Per-source-class priority weight overrides.
    ///
    /// If `None`, the default weights are used (raster=3, vector=2,
    /// terrain=1).  The weights are relative -- only their ratio
    /// matters.
    pub weights: Option<[u32; 3]>,
}

impl Default for CoordinatorConfig {
    fn default() -> Self {
        Self {
            global_request_budget: 32,
            weights: None,
        }
    }
}

// ---------------------------------------------------------------------------
// Per-source tracking
// ---------------------------------------------------------------------------

/// Per-source-class state tracked across frames.
#[derive(Debug, Default)]
struct SourceSlot {
    /// Number of tile requests the source reported as pending (wanted
    /// but not yet issued) at the end of the previous frame.
    demand: usize,
    /// Budget allocated for this frame (set by `begin_frame`).
    budget: usize,
    /// Number of requests the source actually issued this frame.
    issued: usize,
    /// Number of desired tiles that still need loading (fallback +
    /// missing), reported by the source alongside `issued`.
    pending_demand: usize,
    /// Tile IDs that the source reported as its desired set this frame.
    desired: HashSet<TileId>,
}

// ---------------------------------------------------------------------------
// Diagnostics
// ---------------------------------------------------------------------------

/// Per-frame cross-source coordination diagnostics.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct CoordinatorStats {
    /// Total global request budget for this frame.
    pub budget_total: usize,
    /// Budget allocated to raster sources.
    pub budget_raster: usize,
    /// Budget allocated to vector sources.
    pub budget_vector: usize,
    /// Budget allocated to terrain sources.
    pub budget_terrain: usize,
    /// Number of tile IDs that appeared in more than one source's
    /// desired set (shared tiles).
    pub shared_tile_count: usize,
    /// Number of unique tile IDs across all source desired sets.
    pub unique_desired_tiles: usize,
}

// ---------------------------------------------------------------------------
// Coordinator
// ---------------------------------------------------------------------------

/// Centralises tile-request scheduling across all active sources.
///
/// See the [module-level documentation](self) for the design rationale
/// and integration points.
#[derive(Debug)]
pub struct TileRequestCoordinator {
    config: CoordinatorConfig,
    slots: [SourceSlot; 3],
    stats: CoordinatorStats,
}

impl TileRequestCoordinator {
    /// Create a new coordinator with the given configuration.
    pub fn new(config: CoordinatorConfig) -> Self {
        Self {
            config,
            slots: Default::default(),
            stats: CoordinatorStats::default(),
        }
    }

    // -- Budget queries (called before each source updates) ----------------

    /// Return the request budget allocated for `priority` this frame.
    ///
    /// `MapState` should call this before invoking
    /// `TileLayer::update_with_view` and pass the result as
    /// `TileSelectionConfig::max_requests_per_frame` so that the
    /// `TileManager` respects the coordinator's global budget.
    ///
    /// Returns `usize::MAX` when coordination is disabled
    /// (`global_request_budget == 0`).
    pub fn budget_for(&self, priority: SourcePriority) -> usize {
        if self.config.global_request_budget == 0 {
            return usize::MAX;
        }
        self.slots[Self::idx(priority)].budget
    }

    // -- Source registration (called after each source updates) ------------

    /// Record the desired tile set and request count for a source class.
    ///
    /// `desired` is the set of tile IDs the source wants to render.
    /// `issued` is the number of new HTTP requests actually dispatched
    /// during this frame's update.
    /// `pending_demand` is the number of desired tiles that still need
    /// loading (fallback + missing).  This drives the budget allocation
    /// for the next frame even when the current frame's budget was zero.
    pub fn report(
        &mut self,
        priority: SourcePriority,
        desired: HashSet<TileId>,
        issued: usize,
        pending_demand: usize,
    ) {
        let slot = &mut self.slots[Self::idx(priority)];
        slot.desired = desired;
        slot.issued = issued;
        slot.pending_demand = pending_demand;
    }

    /// Record only the demand (pending request count) for a source class,
    /// without replacing its desired tile set.
    ///
    /// Useful for sources that don't expose their full desired set
    /// (e.g. terrain DEM, which computes tiles internally).
    pub fn report_demand(&mut self, priority: SourcePriority, demand: usize) {
        self.slots[Self::idx(priority)].demand = demand;
    }

    // -- Frame lifecycle ---------------------------------------------------

    /// Begin a new frame: allocate per-source budgets from the global cap.
    ///
    /// Must be called once per frame **before** any source updates.
    pub fn begin_frame(&mut self) {
        if self.config.global_request_budget == 0 {
            // Coordination disabled -- give each source unlimited budget.
            for slot in &mut self.slots {
                slot.budget = usize::MAX;
                slot.issued = 0;
                slot.desired.clear();
            }
            return;
        }

        let weights = self.config.weights.unwrap_or([
            SourcePriority::Raster.default_weight(),
            SourcePriority::Vector.default_weight(),
            SourcePriority::Terrain.default_weight(),
        ]);

        // Phase 1: compute raw proportional budgets.
        let total_weight: u32 = weights.iter().sum();
        let budget = self.config.global_request_budget;
        let mut budgets = [0usize; 3];
        let mut remainder = budget;

        if total_weight > 0 {
            for (i, &w) in weights.iter().enumerate() {
                budgets[i] = budget * w as usize / total_weight as usize;
                remainder -= budgets[i];
            }
            // Distribute rounding remainder to the highest-priority
            // source(s) that still have demand.
            for i in 0..3 {
                if remainder == 0 {
                    break;
                }
                if self.slots[i].demand > 0 {
                    budgets[i] += 1;
                    remainder -= 1;
                }
            }
        }

        // Phase 2: redistribute unused budget from sources that have
        // no demand to those that do.
        let mut excess = 0usize;
        for i in 0..3 {
            if self.slots[i].demand == 0 {
                excess += budgets[i];
                budgets[i] = 0;
            }
        }
        if excess > 0 {
            let needy: Vec<usize> = (0..3)
                .filter(|&i| self.slots[i].demand > 0)
                .collect();
            let needy_weight: u32 = needy.iter().map(|&i| weights[i]).sum();
            if needy_weight > 0 {
                let mut leftover = excess;
                for &i in &needy {
                    let share = excess * weights[i] as usize / needy_weight as usize;
                    budgets[i] += share;
                    leftover -= share;
                }
                // Give any leftover to the first needy source.
                if leftover > 0 {
                    if let Some(&first) = needy.first() {
                        budgets[first] += leftover;
                    }
                }
            }
        }

        for (i, slot) in self.slots.iter_mut().enumerate() {
            slot.budget = budgets[i];
            slot.issued = 0;
            slot.desired.clear();
        }
    }

    /// Finish the frame: compute cross-source diagnostics.
    ///
    /// Must be called once per frame **after** all source updates.
    pub fn finish_frame(&mut self) {
        // Compute shared / unique tile counts.
        let mut all_tiles: HashSet<TileId> = HashSet::new();
        let mut total_per_source = 0usize;

        for slot in &self.slots {
            total_per_source += slot.desired.len();
            all_tiles.extend(slot.desired.iter());
        }

        let unique = all_tiles.len();
        let shared = total_per_source.saturating_sub(unique);

        // Update demand from this frame's desired sets for next frame's
        // budget allocation.
        for slot in &mut self.slots {
            // Demand = the greater of requests actually issued and tiles
            // that still need loading.  Using only `issued` creates a
            // deadlock: when budget is 0, issued is 0, so demand stays 0
            // and the budget never recovers.  `pending_demand` reflects
            // the true need regardless of the current budget.
            slot.demand = slot.issued.max(slot.pending_demand);
        }

        self.stats = CoordinatorStats {
            budget_total: self.config.global_request_budget,
            budget_raster: self.slots[0].budget,
            budget_vector: self.slots[1].budget,
            budget_terrain: self.slots[2].budget,
            shared_tile_count: shared,
            unique_desired_tiles: unique,
        };
    }

    /// Read-only access to the most recent cross-source diagnostics.
    pub fn stats(&self) -> &CoordinatorStats {
        &self.stats
    }

    /// Read-only access to the coordinator configuration.
    pub fn config(&self) -> &CoordinatorConfig {
        &self.config
    }

    /// Replace the coordinator configuration.
    ///
    /// Takes effect on the next `begin_frame()` call.
    pub fn set_config(&mut self, config: CoordinatorConfig) {
        self.config = config;
    }

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

    /// Map a source priority to a slot index.
    #[inline]
    fn idx(priority: SourcePriority) -> usize {
        match priority {
            SourcePriority::Raster => 0,
            SourcePriority::Vector => 1,
            SourcePriority::Terrain => 2,
        }
    }
}

impl Default for TileRequestCoordinator {
    fn default() -> Self {
        Self::new(CoordinatorConfig::default())
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn default_budget_allocation_respects_weights() {
        let mut coord = TileRequestCoordinator::default();

        // Simulate demand from all three source classes.
        coord.slots[0].demand = 10; // raster
        coord.slots[1].demand = 10; // vector
        coord.slots[2].demand = 10; // terrain

        coord.begin_frame();

        // Default weights: raster=3, vector=2, terrain=1.  Budget=32.
        // raster = 32 * 3/6 = 16, vector = 32 * 2/6 = 10,
        // terrain = 32 * 1/6 = 5, remainder 1 -> raster.
        assert_eq!(coord.budget_for(SourcePriority::Raster), 17);
        assert_eq!(coord.budget_for(SourcePriority::Vector), 10);
        assert_eq!(coord.budget_for(SourcePriority::Terrain), 5);

        // Total should equal global budget.
        let total = coord.budget_for(SourcePriority::Raster)
            + coord.budget_for(SourcePriority::Vector)
            + coord.budget_for(SourcePriority::Terrain);
        assert_eq!(total, 32);
    }

    #[test]
    fn unused_budget_redistributed_to_needy_sources() {
        let mut coord = TileRequestCoordinator::default();

        // Only raster has demand -- vector and terrain are idle.
        coord.slots[0].demand = 20; // raster
        coord.slots[1].demand = 0;  // vector (idle)
        coord.slots[2].demand = 0;  // terrain (idle)

        coord.begin_frame();

        // Raster should get the entire budget.
        assert_eq!(coord.budget_for(SourcePriority::Raster), 32);
        assert_eq!(coord.budget_for(SourcePriority::Vector), 0);
        assert_eq!(coord.budget_for(SourcePriority::Terrain), 0);
    }

    #[test]
    fn zero_budget_disables_coordination() {
        let config = CoordinatorConfig {
            global_request_budget: 0,
            ..Default::default()
        };
        let mut coord = TileRequestCoordinator::new(config);

        coord.begin_frame();

        assert_eq!(coord.budget_for(SourcePriority::Raster), usize::MAX);
        assert_eq!(coord.budget_for(SourcePriority::Vector), usize::MAX);
        assert_eq!(coord.budget_for(SourcePriority::Terrain), usize::MAX);
    }

    #[test]
    fn shared_tile_count_tracks_cross_source_overlap() {
        let mut coord = TileRequestCoordinator::default();
        coord.begin_frame();

        let tile_a = TileId::new(5, 10, 12);
        let tile_b = TileId::new(5, 11, 12);
        let tile_c = TileId::new(5, 12, 12);

        // Raster wants {a, b}, vector wants {b, c} -- tile_b is shared.
        coord.report(
            SourcePriority::Raster,
            [tile_a, tile_b].into_iter().collect(),
            2,
            0,
        );
        coord.report(
            SourcePriority::Vector,
            [tile_b, tile_c].into_iter().collect(),
            2,
            0,
        );

        coord.finish_frame();

        let stats = coord.stats();
        assert_eq!(stats.unique_desired_tiles, 3);
        assert_eq!(stats.shared_tile_count, 1);
    }

    #[test]
    fn finish_frame_updates_demand_for_next_frame() {
        let mut coord = TileRequestCoordinator::default();

        // Frame 1: all sources have demand.
        coord.slots[0].demand = 10;
        coord.slots[1].demand = 5;
        coord.slots[2].demand = 3;
        coord.begin_frame();

        // Raster issued 8 requests, vector 3, terrain 0.
        coord.report(SourcePriority::Raster, HashSet::new(), 8, 0);
        coord.report(SourcePriority::Vector, HashSet::new(), 3, 0);
        coord.report(SourcePriority::Terrain, HashSet::new(), 0, 0);
        coord.finish_frame();

        // Frame 2: demand should reflect issued counts from frame 1.
        coord.begin_frame();

        // Terrain had 0 issued -> 0 demand -> budget redistributed.
        assert_eq!(coord.budget_for(SourcePriority::Terrain), 0);
        // Raster and vector should share the full 32.
        let raster_budget = coord.budget_for(SourcePriority::Raster);
        let vector_budget = coord.budget_for(SourcePriority::Vector);
        assert_eq!(raster_budget + vector_budget, 32);
        // Raster (weight 3) should get more than vector (weight 2).
        assert!(raster_budget > vector_budget);
    }

    #[test]
    fn custom_weights_override_defaults() {
        let config = CoordinatorConfig {
            global_request_budget: 20,
            weights: Some([1, 1, 1]), // equal weights
        };
        let mut coord = TileRequestCoordinator::new(config);

        coord.slots[0].demand = 10;
        coord.slots[1].demand = 10;
        coord.slots[2].demand = 10;
        coord.begin_frame();

        // Equal weights: each gets ~6-7 (20/3 with remainder).
        let r = coord.budget_for(SourcePriority::Raster);
        let v = coord.budget_for(SourcePriority::Vector);
        let t = coord.budget_for(SourcePriority::Terrain);
        assert_eq!(r + v + t, 20);
        // With equal weights, difference should be at most 1.
        assert!(r.abs_diff(v) <= 1);
        assert!(v.abs_diff(t) <= 1);
    }
}