sectorsync_core/
hotspot.rs

1//! Hotspot metrics and split planning primitives.
2
3use crate::ids::StationId;
4use crate::spatial::CellCoord3;
5
6/// Load sample for one spatial cell.
7#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
8pub struct CellLoadSample {
9    /// Cell coordinate.
10    pub cell: CellCoord3,
11    /// Owned entity count observed in this cell.
12    pub owned_entities: usize,
13    /// Ghost entity count observed in this cell.
14    pub ghost_entities: usize,
15    /// Estimated subscribers interested in this cell.
16    pub subscribers: usize,
17    /// Estimated updates generated by this cell for the current window.
18    pub estimated_updates: usize,
19    /// Estimated payload bytes generated by this cell for the current window.
20    pub estimated_bytes: usize,
21    /// Runtime-defined tick cost units for this cell.
22    pub tick_cost_units: u64,
23    /// Runtime-defined event pressure units for this cell.
24    pub event_pressure: usize,
25}
26
27impl CellLoadSample {
28    /// Returns a deterministic weighted pressure score for ordering cells.
29    pub fn pressure_score(self) -> u64 {
30        let entities = (self.owned_entities + self.ghost_entities) as u64;
31        let subscribers = self.subscribers as u64;
32        let updates = self.estimated_updates as u64;
33        let bytes = (self.estimated_bytes / 256) as u64;
34        let events = self.event_pressure as u64;
35        entities
36            .saturating_mul(8)
37            .saturating_add(subscribers.saturating_mul(4))
38            .saturating_add(updates.saturating_mul(2))
39            .saturating_add(bytes)
40            .saturating_add(events.saturating_mul(16))
41            .saturating_add(self.tick_cost_units)
42    }
43}
44
45/// Load sample for one station over a measurement window.
46#[derive(Clone, Debug, Default, PartialEq, Eq)]
47pub struct StationLoadSample {
48    /// Station being measured.
49    pub station_id: StationId,
50    /// Number of authoritative entities.
51    pub owned_entities: usize,
52    /// Number of read-only ghost entities.
53    pub ghost_entities: usize,
54    /// Estimated subscribers routed to this station.
55    pub subscribers: usize,
56    /// Total queued cross-station events.
57    pub queued_events: usize,
58    /// Estimated frame bytes generated by this station.
59    pub estimated_bytes: usize,
60    /// Runtime-defined station tick cost units.
61    pub tick_cost_units: u64,
62    /// Per-cell load samples.
63    pub cells: Vec<CellLoadSample>,
64}
65
66impl StationLoadSample {
67    /// Returns total entity count.
68    pub const fn total_entities(&self) -> usize {
69        self.owned_entities + self.ghost_entities
70    }
71
72    /// Returns the highest per-cell pressure score.
73    pub fn max_cell_pressure(&self) -> u64 {
74        self.cells
75            .iter()
76            .map(|cell| cell.pressure_score())
77            .max()
78            .unwrap_or(0)
79    }
80}
81
82/// Thresholds used to classify hotspot pressure.
83#[derive(Clone, Copy, Debug, PartialEq, Eq)]
84pub struct HotspotThresholds {
85    /// Maximum total entities before a station is considered hot.
86    pub max_station_entities: usize,
87    /// Maximum subscribers before a station is considered hot.
88    pub max_station_subscribers: usize,
89    /// Maximum queued events before a station is considered hot.
90    pub max_queued_events: usize,
91    /// Maximum estimated bytes before a station is considered hot.
92    pub max_estimated_bytes: usize,
93    /// Maximum tick cost units before a station is considered hot.
94    pub max_tick_cost_units: u64,
95    /// Maximum pressure score for a single cell before split is suggested.
96    pub max_cell_pressure: u64,
97}
98
99impl Default for HotspotThresholds {
100    fn default() -> Self {
101        Self {
102            max_station_entities: 50_000,
103            max_station_subscribers: 2_000,
104            max_queued_events: 10_000,
105            max_estimated_bytes: 64 * 1024 * 1024,
106            max_tick_cost_units: 16_000,
107            max_cell_pressure: 10_000,
108        }
109    }
110}
111
112/// Hotspot classification.
113#[derive(Clone, Copy, Debug, PartialEq, Eq)]
114pub enum HotspotSeverity {
115    /// Load is within configured limits.
116    Normal,
117    /// Load is near or slightly over limits.
118    Warm,
119    /// Load is clearly over limits and should trigger mitigation.
120    Hot,
121}
122
123/// Hotspot evaluation result.
124#[derive(Clone, Debug, PartialEq, Eq)]
125pub struct HotspotDecision {
126    /// Station evaluated by this decision.
127    pub station_id: StationId,
128    /// Classification severity.
129    pub severity: HotspotSeverity,
130    /// Weighted pressure score.
131    pub score: u64,
132    /// Human-readable reason codes.
133    pub reasons: Vec<&'static str>,
134}
135
136/// Cell split proposal for external schedulers.
137#[derive(Clone, Debug, Default, PartialEq, Eq)]
138pub struct SplitProposal {
139    /// Source station that should give up some cell ownership.
140    pub source_station: StationId,
141    /// Cells recommended for movement to a new or less loaded station.
142    pub cells_to_move: Vec<CellCoord3>,
143    /// Total pressure score covered by `cells_to_move`.
144    pub moved_pressure_score: u64,
145}
146
147/// Hotspot evaluator and split planner.
148#[derive(Clone, Copy, Debug, Default)]
149pub struct HotspotPlanner;
150
151impl HotspotPlanner {
152    /// Evaluates a station load sample against thresholds.
153    pub fn evaluate(sample: &StationLoadSample, thresholds: HotspotThresholds) -> HotspotDecision {
154        let mut score = 0_u64;
155        let mut reasons = Vec::new();
156
157        if sample.total_entities() > thresholds.max_station_entities {
158            score = score.saturating_add(1);
159            reasons.push("station_entities");
160        }
161        if sample.subscribers > thresholds.max_station_subscribers {
162            score = score.saturating_add(1);
163            reasons.push("station_subscribers");
164        }
165        if sample.queued_events > thresholds.max_queued_events {
166            score = score.saturating_add(1);
167            reasons.push("queued_events");
168        }
169        if sample.estimated_bytes > thresholds.max_estimated_bytes {
170            score = score.saturating_add(1);
171            reasons.push("estimated_bytes");
172        }
173        if sample.tick_cost_units > thresholds.max_tick_cost_units {
174            score = score.saturating_add(1);
175            reasons.push("tick_cost");
176        }
177        if sample.max_cell_pressure() > thresholds.max_cell_pressure {
178            score = score.saturating_add(1);
179            reasons.push("cell_pressure");
180        }
181
182        let severity = match score {
183            0 => HotspotSeverity::Normal,
184            1 => HotspotSeverity::Warm,
185            _ => HotspotSeverity::Hot,
186        };
187
188        HotspotDecision {
189            station_id: sample.station_id,
190            severity,
191            score,
192            reasons,
193        }
194    }
195
196    /// Proposes cells to move by selecting the highest-pressure cells first.
197    pub fn propose_cell_split(
198        sample: &StationLoadSample,
199        max_cells_to_move: usize,
200    ) -> SplitProposal {
201        let mut cells = sample.cells.clone();
202        cells.sort_by(|left, right| {
203            right
204                .pressure_score()
205                .cmp(&left.pressure_score())
206                .then_with(|| left.cell.cmp(&right.cell))
207        });
208
209        let mut proposal = SplitProposal {
210            source_station: sample.station_id,
211            cells_to_move: Vec::with_capacity(max_cells_to_move.min(cells.len())),
212            moved_pressure_score: 0,
213        };
214
215        for cell in cells.into_iter().take(max_cells_to_move) {
216            proposal.moved_pressure_score = proposal
217                .moved_pressure_score
218                .saturating_add(cell.pressure_score());
219            proposal.cells_to_move.push(cell.cell);
220        }
221
222        proposal
223    }
224}
225
226#[cfg(test)]
227mod tests {
228    use super::*;
229
230    #[test]
231    fn planner_detects_hot_station_and_suggests_hottest_cells() {
232        let sample = StationLoadSample {
233            station_id: StationId::new(3),
234            owned_entities: 100,
235            subscribers: 20,
236            cells: vec![
237                CellLoadSample {
238                    cell: CellCoord3::new(0, 0, 0),
239                    owned_entities: 1,
240                    ..CellLoadSample::default()
241                },
242                CellLoadSample {
243                    cell: CellCoord3::new(1, 0, 0),
244                    owned_entities: 100,
245                    event_pressure: 50,
246                    ..CellLoadSample::default()
247                },
248            ],
249            ..StationLoadSample::default()
250        };
251        let thresholds = HotspotThresholds {
252            max_station_entities: 50,
253            max_cell_pressure: 10,
254            ..HotspotThresholds::default()
255        };
256
257        let decision = HotspotPlanner::evaluate(&sample, thresholds);
258        assert_eq!(decision.severity, HotspotSeverity::Hot);
259
260        let proposal = HotspotPlanner::propose_cell_split(&sample, 1);
261        assert_eq!(proposal.cells_to_move, vec![CellCoord3::new(1, 0, 0)]);
262    }
263}
sectorsync_core/hotspot.rs

sectorsync_core/
hotspot.rs
