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
136impl Default for HotspotDecision {
137    fn default() -> Self {
138        Self {
139            station_id: StationId::default(),
140            severity: HotspotSeverity::Normal,
141            score: 0,
142            reasons: Vec::new(),
143        }
144    }
145}
146
147/// Cell split proposal for external schedulers.
148#[derive(Clone, Debug, Default, PartialEq, Eq)]
149pub struct SplitProposal {
150    /// Source station that should give up some cell ownership.
151    pub source_station: StationId,
152    /// Cells recommended for movement to a new or less loaded station.
153    pub cells_to_move: Vec<CellCoord3>,
154    /// Total pressure score covered by `cells_to_move`.
155    pub moved_pressure_score: u64,
156}
157
158/// Caller-owned reusable candidate storage for hotspot cell split planning.
159#[derive(Clone, Debug, Default)]
160pub struct HotspotSplitScratch {
161    cells: Vec<CellLoadSample>,
162}
163
164impl HotspotSplitScratch {
165    /// Creates empty split-planning scratch.
166    pub const fn new() -> Self {
167        Self { cells: Vec::new() }
168    }
169
170    /// Cell candidate capacity retained across calls.
171    pub fn candidate_capacity(&self) -> usize {
172        self.cells.capacity()
173    }
174}
175
176/// Hotspot evaluator and split planner.
177#[derive(Clone, Copy, Debug, Default)]
178pub struct HotspotPlanner;
179
180impl HotspotPlanner {
181    /// Evaluates a station load sample against thresholds.
182    pub fn evaluate(sample: &StationLoadSample, thresholds: HotspotThresholds) -> HotspotDecision {
183        let mut decision = HotspotDecision::default();
184        Self::evaluate_into(sample, thresholds, &mut decision);
185        decision
186    }
187
188    /// Evaluates a station into caller-owned output while retaining reason capacity.
189    pub fn evaluate_into(
190        sample: &StationLoadSample,
191        thresholds: HotspotThresholds,
192        decision: &mut HotspotDecision,
193    ) {
194        let mut score = 0_u64;
195        decision.station_id = sample.station_id;
196        decision.reasons.clear();
197
198        if sample.total_entities() > thresholds.max_station_entities {
199            score = score.saturating_add(1);
200            decision.reasons.push("station_entities");
201        }
202        if sample.subscribers > thresholds.max_station_subscribers {
203            score = score.saturating_add(1);
204            decision.reasons.push("station_subscribers");
205        }
206        if sample.queued_events > thresholds.max_queued_events {
207            score = score.saturating_add(1);
208            decision.reasons.push("queued_events");
209        }
210        if sample.estimated_bytes > thresholds.max_estimated_bytes {
211            score = score.saturating_add(1);
212            decision.reasons.push("estimated_bytes");
213        }
214        if sample.tick_cost_units > thresholds.max_tick_cost_units {
215            score = score.saturating_add(1);
216            decision.reasons.push("tick_cost");
217        }
218        if sample.max_cell_pressure() > thresholds.max_cell_pressure {
219            score = score.saturating_add(1);
220            decision.reasons.push("cell_pressure");
221        }
222
223        decision.severity = match score {
224            0 => HotspotSeverity::Normal,
225            1 => HotspotSeverity::Warm,
226            _ => HotspotSeverity::Hot,
227        };
228        decision.score = score;
229    }
230
231    /// Proposes cells to move by selecting the highest-pressure cells first.
232    pub fn propose_cell_split(
233        sample: &StationLoadSample,
234        max_cells_to_move: usize,
235    ) -> SplitProposal {
236        let mut scratch = HotspotSplitScratch::new();
237        let mut proposal = SplitProposal::default();
238        Self::propose_cell_split_into(sample, max_cells_to_move, &mut scratch, &mut proposal);
239        proposal
240    }
241
242    /// Proposes cells using caller-owned candidate scratch.
243    pub fn propose_cell_split_with_scratch(
244        sample: &StationLoadSample,
245        max_cells_to_move: usize,
246        scratch: &mut HotspotSplitScratch,
247    ) -> SplitProposal {
248        let mut proposal = SplitProposal::default();
249        Self::propose_cell_split_into(sample, max_cells_to_move, scratch, &mut proposal);
250        proposal
251    }
252
253    /// Writes a deterministic split proposal into fully reusable caller-owned storage.
254    pub fn propose_cell_split_into(
255        sample: &StationLoadSample,
256        max_cells_to_move: usize,
257        scratch: &mut HotspotSplitScratch,
258        proposal: &mut SplitProposal,
259    ) {
260        let selected = max_cells_to_move.min(sample.cells.len());
261        scratch.cells.clear();
262        scratch.cells.extend_from_slice(&sample.cells);
263        prioritize_cell_samples(&mut scratch.cells, selected);
264
265        proposal.source_station = sample.station_id;
266        proposal.cells_to_move.clear();
267        proposal.cells_to_move.reserve(selected);
268        proposal.moved_pressure_score = 0;
269        for cell in &scratch.cells[..selected] {
270            proposal.moved_pressure_score = proposal
271                .moved_pressure_score
272                .saturating_add(cell.pressure_score());
273            proposal.cells_to_move.push(cell.cell);
274        }
275    }
276}
277
278fn compare_cell_samples(left: &CellLoadSample, right: &CellLoadSample) -> core::cmp::Ordering {
279    right
280        .pressure_score()
281        .cmp(&left.pressure_score())
282        .then_with(|| left.cell.cmp(&right.cell))
283}
284
285fn prioritize_cell_samples(cells: &mut [CellLoadSample], selected: usize) {
286    if selected == 0 {
287        return;
288    }
289    if selected.saturating_mul(2) < cells.len() {
290        cells.select_nth_unstable_by(selected, compare_cell_samples);
291        cells[..selected].sort_by(compare_cell_samples);
292    } else {
293        cells.sort_by(compare_cell_samples);
294    }
295}
296
297#[cfg(test)]
298mod tests {
299    use super::*;
300
301    #[test]
302    fn planner_detects_hot_station_and_suggests_hottest_cells() {
303        let sample = StationLoadSample {
304            station_id: StationId::new(3),
305            owned_entities: 100,
306            subscribers: 20,
307            cells: vec![
308                CellLoadSample {
309                    cell: CellCoord3::new(0, 0, 0),
310                    owned_entities: 1,
311                    ..CellLoadSample::default()
312                },
313                CellLoadSample {
314                    cell: CellCoord3::new(1, 0, 0),
315                    owned_entities: 100,
316                    event_pressure: 50,
317                    ..CellLoadSample::default()
318                },
319            ],
320            ..StationLoadSample::default()
321        };
322        let thresholds = HotspotThresholds {
323            max_station_entities: 50,
324            max_cell_pressure: 10,
325            ..HotspotThresholds::default()
326        };
327
328        let decision = HotspotPlanner::evaluate(&sample, thresholds);
329        assert_eq!(decision.severity, HotspotSeverity::Hot);
330
331        let mut reused_decision = HotspotDecision::default();
332        HotspotPlanner::evaluate_into(&sample, thresholds, &mut reused_decision);
333        assert_eq!(reused_decision, decision);
334        let reason_capacity = reused_decision.reasons.capacity();
335        HotspotPlanner::evaluate_into(
336            &StationLoadSample {
337                station_id: StationId::new(4),
338                ..StationLoadSample::default()
339            },
340            thresholds,
341            &mut reused_decision,
342        );
343        assert_eq!(reused_decision.station_id, StationId::new(4));
344        assert_eq!(reused_decision.severity, HotspotSeverity::Normal);
345        assert!(reused_decision.reasons.is_empty());
346        assert_eq!(reused_decision.reasons.capacity(), reason_capacity);
347
348        let proposal = HotspotPlanner::propose_cell_split(&sample, 1);
349        assert_eq!(proposal.cells_to_move, vec![CellCoord3::new(1, 0, 0)]);
350    }
351
352    #[test]
353    fn split_top_k_matches_full_sort_and_reuses_capacity_at_budget_edges() {
354        let cells = (0_i32..257)
355            .map(|index| CellLoadSample {
356                cell: CellCoord3::new(index, index % 5, index % 7),
357                owned_entities: usize::try_from(index * 37 % 23).expect("non-negative"),
358                event_pressure: usize::try_from(index * 19 % 11).expect("non-negative"),
359                ..CellLoadSample::default()
360            })
361            .collect::<Vec<_>>();
362        let sample = StationLoadSample {
363            station_id: StationId::new(9),
364            cells: cells.clone(),
365            ..StationLoadSample::default()
366        };
367        let mut scratch = HotspotSplitScratch::new();
368        let mut proposal = SplitProposal::default();
369
370        for requested in [0, 1, 7, 64, 128, 129, 256, 257, 300] {
371            let selected = requested.min(cells.len());
372            let mut expected = cells.clone();
373            expected.sort_by(compare_cell_samples);
374            expected.truncate(selected);
375            HotspotPlanner::propose_cell_split_into(
376                &sample,
377                requested,
378                &mut scratch,
379                &mut proposal,
380            );
381
382            assert_eq!(
383                proposal.cells_to_move,
384                expected.iter().map(|cell| cell.cell).collect::<Vec<_>>()
385            );
386            assert_eq!(
387                proposal.moved_pressure_score,
388                expected.iter().fold(0_u64, |score, cell| {
389                    score.saturating_add(cell.pressure_score())
390                })
391            );
392        }
393        let candidate_capacity = scratch.candidate_capacity();
394        let output_capacity = proposal.cells_to_move.capacity();
395        HotspotPlanner::propose_cell_split_into(
396            &StationLoadSample {
397                station_id: StationId::new(10),
398                cells: cells[..8].to_vec(),
399                ..StationLoadSample::default()
400            },
401            2,
402            &mut scratch,
403            &mut proposal,
404        );
405        assert_eq!(proposal.source_station, StationId::new(10));
406        assert_eq!(scratch.candidate_capacity(), candidate_capacity);
407        assert_eq!(proposal.cells_to_move.capacity(), output_capacity);
408    }
409}
sectorsync_core/hotspot.rs

sectorsync_core/
hotspot.rs
