Skip to main content

sectorsync_core/
replication.rs

1//! Replication planning helpers.
2
3use std::cmp::Ordering;
4use std::collections::{BTreeMap, HashMap};
5use std::hash::Hash;
6
7use crate::entity::EntityRecord;
8use crate::ids::{ClientId, EntityHandle, Tick};
9#[cfg(not(feature = "simd"))]
10use crate::interest::RangeOnlyVisibility;
11use crate::interest::{ViewerQuery, VisibilityFilter};
12use crate::policy::{CompiledSyncPolicy, PolicyTable};
13use crate::spatial_index::{CellIndex, CellQueryScratch, CellQueryStats, CellQueryStrategy};
14use crate::station::Station;
15
16const HASHED_REPLICATION_TRACKER_MIN_ENTRIES: usize = 2_048;
17
18#[derive(Clone, Debug)]
19enum AdaptiveTrackMap<K, V> {
20    Ordered(BTreeMap<K, V>),
21    Hashed(HashMap<K, V>),
22}
23
24impl<K: Copy + Eq + Hash + Ord, V> AdaptiveTrackMap<K, V> {
25    fn new() -> Self {
26        Self::Ordered(BTreeMap::new())
27    }
28
29    fn len(&self) -> usize {
30        match self {
31            Self::Ordered(entries) => entries.len(),
32            Self::Hashed(entries) => entries.len(),
33        }
34    }
35
36    fn is_empty(&self) -> bool {
37        match self {
38            Self::Ordered(entries) => entries.is_empty(),
39            Self::Hashed(entries) => entries.is_empty(),
40        }
41    }
42
43    fn contains_key(&self, key: &K) -> bool {
44        match self {
45            Self::Ordered(entries) => entries.contains_key(key),
46            Self::Hashed(entries) => entries.contains_key(key),
47        }
48    }
49
50    fn get(&self, key: &K) -> Option<&V> {
51        match self {
52            Self::Ordered(entries) => entries.get(key),
53            Self::Hashed(entries) => entries.get(key),
54        }
55    }
56
57    fn get_mut(&mut self, key: &K) -> Option<&mut V> {
58        match self {
59            Self::Ordered(entries) => entries.get_mut(key),
60            Self::Hashed(entries) => entries.get_mut(key),
61        }
62    }
63
64    fn insert(&mut self, key: K, value: V) -> Option<V> {
65        let promote = match self {
66            Self::Ordered(entries) => {
67                entries.len() >= HASHED_REPLICATION_TRACKER_MIN_ENTRIES.saturating_sub(1)
68                    && !entries.contains_key(&key)
69            }
70            Self::Hashed(_) => false,
71        };
72        if promote {
73            let Self::Ordered(ordered) = std::mem::replace(self, Self::Hashed(HashMap::new()))
74            else {
75                unreachable!("promotion starts from ordered tracker storage");
76            };
77            let mut hashed = HashMap::with_capacity(ordered.len().saturating_add(1));
78            hashed.extend(ordered);
79            *self = Self::Hashed(hashed);
80        }
81        match self {
82            Self::Ordered(entries) => entries.insert(key, value),
83            Self::Hashed(entries) => entries.insert(key, value),
84        }
85    }
86
87    fn retain<F>(&mut self, mut keep: F)
88    where
89        F: FnMut(&K, &mut V) -> bool,
90    {
91        match self {
92            Self::Ordered(entries) => entries.retain(|key, value| keep(key, value)),
93            Self::Hashed(entries) => entries.retain(|key, value| keep(key, value)),
94        }
95    }
96
97    #[cfg(test)]
98    fn is_hashed(&self) -> bool {
99        matches!(self, Self::Hashed(_))
100    }
101}
102
103/// Per-client replication budget.
104#[derive(Clone, Copy, Debug, PartialEq, Eq)]
105pub struct ReplicationBudget {
106    /// Maximum entities to include in a frame.
107    pub max_entities: usize,
108    /// Estimated byte budget for a frame.
109    pub max_bytes: usize,
110    /// Estimated bytes charged per selected entity by simple planners.
111    pub estimated_entity_bytes: usize,
112}
113
114impl Default for ReplicationBudget {
115    fn default() -> Self {
116        Self {
117            max_entities: 300,
118            max_bytes: 16 * 1024,
119            estimated_entity_bytes: 32,
120        }
121    }
122}
123
124/// Replication planner result.
125#[derive(Clone, Debug, Default, PartialEq, Eq)]
126pub struct ReplicationPlan {
127    /// Selected entity handles.
128    pub entities: Vec<EntityHandle>,
129    /// Planner statistics.
130    pub stats: ReplicationStats,
131}
132
133/// Aggregated work and retained-capacity signals from one viewer batch.
134#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
135pub struct ReplicationBatchStats {
136    /// Viewer queries planned in input order.
137    pub viewers: usize,
138    /// Spatial candidates returned across all viewers.
139    pub candidates: usize,
140    /// Candidate records considered after stale-handle filtering.
141    pub considered: usize,
142    /// Entities selected across all plans.
143    pub selected: usize,
144    /// Spatial candidates left unexamined after a first-fit budget filled.
145    pub unexamined_after_budget: usize,
146    /// Estimated payload bytes across all plans.
147    pub estimated_bytes: usize,
148    /// Queries that probed the regular cell grid.
149    pub grid_queries: usize,
150    /// Queries that scanned occupied cells.
151    pub occupied_queries: usize,
152    /// Grid cells probed across the batch.
153    pub grid_cells_probed: usize,
154    /// Occupied cells scanned across the batch.
155    pub occupied_cells_scanned: usize,
156    /// Cells intersecting viewer query bounds across the batch.
157    pub matched_cells: usize,
158    /// Largest retained candidate-handle capacity.
159    pub candidate_capacity_max: usize,
160    /// Largest retained candidate-deduplication capacity.
161    pub dedup_capacity_max: usize,
162    /// Largest retained matched-cell capacity.
163    pub matching_cell_capacity_max: usize,
164    /// Largest retained priority candidate capacity.
165    pub priority_capacity_max: usize,
166}
167
168impl ReplicationBatchStats {
169    fn record(&mut self, plan: &ReplicationPlan, scratch: &ReplicationScratch) {
170        self.viewers = self.viewers.saturating_add(1);
171        self.candidates = self.candidates.saturating_add(plan.stats.candidates);
172        self.considered = self.considered.saturating_add(plan.stats.considered);
173        self.selected = self.selected.saturating_add(plan.stats.selected);
174        self.unexamined_after_budget = self
175            .unexamined_after_budget
176            .saturating_add(plan.stats.unexamined_after_budget);
177        self.estimated_bytes = self
178            .estimated_bytes
179            .saturating_add(plan.stats.estimated_bytes);
180        let query = scratch.query_stats();
181        match query.strategy {
182            CellQueryStrategy::Grid => self.grid_queries = self.grid_queries.saturating_add(1),
183            CellQueryStrategy::OccupiedCells => {
184                self.occupied_queries = self.occupied_queries.saturating_add(1);
185            }
186        }
187        self.grid_cells_probed = self
188            .grid_cells_probed
189            .saturating_add(query.grid_cells_probed);
190        self.occupied_cells_scanned = self
191            .occupied_cells_scanned
192            .saturating_add(query.occupied_cells_scanned);
193        self.matched_cells = self.matched_cells.saturating_add(query.matched_cells);
194        self.candidate_capacity_max = self
195            .candidate_capacity_max
196            .max(scratch.candidate_capacity());
197        self.dedup_capacity_max = self
198            .dedup_capacity_max
199            .max(scratch.candidate_dedup_capacity());
200        self.matching_cell_capacity_max = self
201            .matching_cell_capacity_max
202            .max(scratch.matching_cell_capacity());
203        self.priority_capacity_max = self
204            .priority_capacity_max
205            .max(scratch.prioritized_capacity());
206    }
207
208    /// Merges another deterministic batch partition into this report.
209    pub fn merge(&mut self, other: Self) {
210        self.viewers = self.viewers.saturating_add(other.viewers);
211        self.candidates = self.candidates.saturating_add(other.candidates);
212        self.considered = self.considered.saturating_add(other.considered);
213        self.selected = self.selected.saturating_add(other.selected);
214        self.estimated_bytes = self.estimated_bytes.saturating_add(other.estimated_bytes);
215        self.grid_queries = self.grid_queries.saturating_add(other.grid_queries);
216        self.occupied_queries = self.occupied_queries.saturating_add(other.occupied_queries);
217        self.grid_cells_probed = self
218            .grid_cells_probed
219            .saturating_add(other.grid_cells_probed);
220        self.occupied_cells_scanned = self
221            .occupied_cells_scanned
222            .saturating_add(other.occupied_cells_scanned);
223        self.matched_cells = self.matched_cells.saturating_add(other.matched_cells);
224        self.candidate_capacity_max = self
225            .candidate_capacity_max
226            .max(other.candidate_capacity_max);
227        self.dedup_capacity_max = self.dedup_capacity_max.max(other.dedup_capacity_max);
228        self.matching_cell_capacity_max = self
229            .matching_cell_capacity_max
230            .max(other.matching_cell_capacity_max);
231        self.priority_capacity_max = self.priority_capacity_max.max(other.priority_capacity_max);
232    }
233}
234
235/// Ordered plans and aggregate statistics produced for a viewer batch.
236#[derive(Clone, Debug, Default, PartialEq, Eq)]
237pub struct ReplicationBatchResult {
238    /// One plan per input viewer, retaining input order.
239    pub plans: Vec<ReplicationPlan>,
240    /// Aggregate work signals for the batch.
241    pub stats: ReplicationBatchStats,
242}
243
244/// Borrowed ordered plans produced from reusable batch storage.
245#[derive(Clone, Copy, Debug, PartialEq, Eq)]
246pub struct ReplicationBatchView<'a> {
247    /// One plan per input viewer, retaining input order.
248    pub plans: &'a [ReplicationPlan],
249    /// Aggregate work signals for the active plans.
250    pub stats: ReplicationBatchStats,
251}
252
253/// Caller-owned reusable output storage for viewer batch planning.
254///
255/// Plan slots and their entity buffers grow to the largest observed batch and
256/// are retained for later calls. This storage contains no cross-client send or
257/// acknowledgement state.
258#[derive(Clone, Debug, Default)]
259pub struct ReplicationBatchScratch {
260    plans: Vec<ReplicationPlan>,
261    active_plans: usize,
262    stats: ReplicationBatchStats,
263}
264
265impl ReplicationBatchScratch {
266    /// Creates empty batch output storage.
267    pub const fn new() -> Self {
268        Self {
269            plans: Vec::new(),
270            active_plans: 0,
271            stats: ReplicationBatchStats {
272                viewers: 0,
273                candidates: 0,
274                considered: 0,
275                selected: 0,
276                unexamined_after_budget: 0,
277                estimated_bytes: 0,
278                grid_queries: 0,
279                occupied_queries: 0,
280                grid_cells_probed: 0,
281                occupied_cells_scanned: 0,
282                matched_cells: 0,
283                candidate_capacity_max: 0,
284                dedup_capacity_max: 0,
285                matching_cell_capacity_max: 0,
286                priority_capacity_max: 0,
287            },
288        }
289    }
290
291    /// Number of plan slots retained for reuse.
292    pub fn retained_plan_slots(&self) -> usize {
293        self.plans.len()
294    }
295
296    /// Total selected-entity capacity retained across all plan slots.
297    pub fn retained_entity_capacity(&self) -> usize {
298        self.plans.iter().map(|plan| plan.entities.capacity()).sum()
299    }
300
301    /// Returns the active result from the most recent planning call.
302    pub fn view(&self) -> ReplicationBatchView<'_> {
303        ReplicationBatchView {
304            plans: &self.plans[..self.active_plans],
305            stats: self.stats,
306        }
307    }
308
309    fn prepare(&mut self, plans: usize) {
310        if self.plans.len() < plans {
311            self.plans.resize_with(plans, ReplicationPlan::default);
312        }
313        self.active_plans = plans;
314        self.stats = ReplicationBatchStats::default();
315    }
316}
317
318/// Bounded per-client replication tracking configuration.
319#[derive(Clone, Copy, Debug, PartialEq, Eq)]
320pub struct ReplicationTrackerConfig {
321    /// Maximum tracked client/entity entries.
322    pub max_entries: usize,
323}
324
325impl Default for ReplicationTrackerConfig {
326    fn default() -> Self {
327        Self {
328            max_entries: 65_536,
329        }
330    }
331}
332
333/// Per-client/entity replication tracking key.
334#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
335pub struct ReplicationTrackKey {
336    /// Client that received the entity update.
337    pub client_id: ClientId,
338    /// Station-local entity handle selected by the planner.
339    pub entity: EntityHandle,
340}
341
342/// Per-client/entity replication tracking record.
343#[derive(Clone, Copy, Debug, PartialEq, Eq)]
344pub struct ReplicationTrackRecord {
345    /// Client that received the entity update.
346    pub client_id: ClientId,
347    /// Station-local entity handle selected by the planner.
348    pub entity: EntityHandle,
349    /// Last tick where this entity was sent to the client.
350    pub last_sent: Tick,
351    /// Last tick where the caller confirmed delivery.
352    pub last_acked: Option<Tick>,
353}
354
355/// Replication tracker statistics.
356#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
357pub struct ReplicationTrackerStats {
358    /// Currently tracked entries.
359    pub entries: usize,
360    /// Total record insertions or updates.
361    pub sent_records: usize,
362    /// Total ACK updates applied.
363    pub acked_records: usize,
364    /// Records pruned by explicit cleanup.
365    pub pruned_records: usize,
366}
367
368/// Replication tracking error.
369#[derive(Clone, Copy, Debug, PartialEq, Eq)]
370pub enum ReplicationTrackerError {
371    /// Recording would exceed the configured entry capacity.
372    CapacityExceeded {
373        /// Entries currently tracked.
374        current: usize,
375        /// New entries needed for this operation.
376        needed: usize,
377        /// Maximum tracked entries.
378        max: usize,
379    },
380}
381
382impl core::fmt::Display for ReplicationTrackerError {
383    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
384        match self {
385            Self::CapacityExceeded {
386                current,
387                needed,
388                max,
389            } => write!(
390                f,
391                "replication tracker capacity exceeded: current {current}, needed {needed}, max {max}"
392            ),
393        }
394    }
395}
396
397impl std::error::Error for ReplicationTrackerError {}
398
399/// Bounded per-client replication send/ACK tracker.
400#[derive(Clone, Debug)]
401pub struct ReplicationTracker {
402    config: ReplicationTrackerConfig,
403    records: AdaptiveTrackMap<ReplicationTrackKey, ReplicationTrackRecord>,
404    stats: ReplicationTrackerStats,
405}
406
407impl Default for ReplicationTracker {
408    fn default() -> Self {
409        Self::new(ReplicationTrackerConfig::default())
410    }
411}
412
413impl ReplicationTracker {
414    /// Creates an empty tracker.
415    pub fn new(config: ReplicationTrackerConfig) -> Self {
416        Self {
417            config,
418            records: AdaptiveTrackMap::new(),
419            stats: ReplicationTrackerStats::default(),
420        }
421    }
422
423    /// Returns tracker configuration.
424    pub const fn config(&self) -> ReplicationTrackerConfig {
425        self.config
426    }
427
428    /// Returns tracker statistics.
429    pub const fn stats(&self) -> ReplicationTrackerStats {
430        self.stats
431    }
432
433    /// Returns tracked entry count.
434    pub fn len(&self) -> usize {
435        self.records.len()
436    }
437
438    /// Returns whether no entries are tracked.
439    pub fn is_empty(&self) -> bool {
440        self.records.is_empty()
441    }
442
443    /// Returns the last sent tick for a client/entity pair.
444    pub fn last_sent(&self, client_id: ClientId, entity: EntityHandle) -> Option<Tick> {
445        self.records
446            .get(&ReplicationTrackKey { client_id, entity })
447            .map(|record| record.last_sent)
448    }
449
450    /// Returns a tracked record for a client/entity pair.
451    pub fn get(&self, client_id: ClientId, entity: EntityHandle) -> Option<ReplicationTrackRecord> {
452        self.records
453            .get(&ReplicationTrackKey { client_id, entity })
454            .copied()
455    }
456
457    /// Records that a planned set of entities was sent to a client.
458    pub fn record_plan_sent(
459        &mut self,
460        client_id: ClientId,
461        plan: &ReplicationPlan,
462        sent_at: Tick,
463    ) -> Result<usize, ReplicationTrackerError> {
464        self.ensure_capacity_for(client_id, &plan.entities)?;
465        let mut recorded = 0;
466        for entity in &plan.entities {
467            let key = ReplicationTrackKey {
468                client_id,
469                entity: *entity,
470            };
471            self.records.insert(
472                key,
473                ReplicationTrackRecord {
474                    client_id,
475                    entity: *entity,
476                    last_sent: sent_at,
477                    last_acked: None,
478                },
479            );
480            recorded += 1;
481        }
482        self.refresh_entry_count();
483        self.stats.sent_records = self.stats.sent_records.saturating_add(recorded);
484        Ok(recorded)
485    }
486
487    /// Records delivery acknowledgement for one client/entity pair.
488    pub fn acknowledge(
489        &mut self,
490        client_id: ClientId,
491        entity: EntityHandle,
492        acked_at: Tick,
493    ) -> bool {
494        let Some(record) = self
495            .records
496            .get_mut(&ReplicationTrackKey { client_id, entity })
497        else {
498            return false;
499        };
500        record.last_acked = Some(acked_at);
501        self.stats.acked_records = self.stats.acked_records.saturating_add(1);
502        true
503    }
504
505    /// Records delivery acknowledgement for every entity in a plan.
506    pub fn acknowledge_plan(
507        &mut self,
508        client_id: ClientId,
509        plan: &ReplicationPlan,
510        acked_at: Tick,
511    ) -> usize {
512        plan.entities
513            .iter()
514            .filter(|entity| self.acknowledge(client_id, **entity, acked_at))
515            .count()
516    }
517
518    /// Removes all entries for one client.
519    pub fn clear_client(&mut self, client_id: ClientId) -> usize {
520        let before = self.records.len();
521        self.records.retain(|key, _| key.client_id != client_id);
522        let pruned = before.saturating_sub(self.records.len());
523        self.stats.pruned_records = self.stats.pruned_records.saturating_add(pruned);
524        self.refresh_entry_count();
525        pruned
526    }
527
528    /// Removes entries last sent before `older_than`.
529    pub fn prune_sent_before(&mut self, older_than: Tick) -> usize {
530        let before = self.records.len();
531        self.records
532            .retain(|_, record| record.last_sent.get() >= older_than.get());
533        let pruned = before.saturating_sub(self.records.len());
534        self.stats.pruned_records = self.stats.pruned_records.saturating_add(pruned);
535        self.refresh_entry_count();
536        pruned
537    }
538
539    fn ensure_capacity_for(
540        &self,
541        client_id: ClientId,
542        entities: &[EntityHandle],
543    ) -> Result<(), ReplicationTrackerError> {
544        if self.records.len().saturating_add(entities.len()) <= self.config.max_entries {
545            return Ok(());
546        }
547        let mut needed = 0_usize;
548        for entity in entities {
549            if !self.records.contains_key(&ReplicationTrackKey {
550                client_id,
551                entity: *entity,
552            }) {
553                needed = needed.saturating_add(1);
554            }
555        }
556        if self.records.len().saturating_add(needed) > self.config.max_entries {
557            return Err(ReplicationTrackerError::CapacityExceeded {
558                current: self.records.len(),
559                needed,
560                max: self.config.max_entries,
561            });
562        }
563        Ok(())
564    }
565
566    fn refresh_entry_count(&mut self) {
567        self.stats.entries = self.records.len();
568    }
569}
570
571/// Replication planner statistics.
572#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
573pub struct ReplicationStats {
574    /// Candidate handles returned from the spatial index.
575    pub candidates: usize,
576    /// Candidate records considered after stale handle filtering.
577    pub considered: usize,
578    /// Selected entities.
579    pub selected: usize,
580    /// Entities skipped because the budget was exhausted.
581    pub skipped_by_budget: usize,
582    /// Spatial candidates not examined after a work-bounded budget filled.
583    pub unexamined_after_budget: usize,
584    /// Entities skipped because their cadence interval has not elapsed.
585    pub skipped_by_cadence: usize,
586    /// Estimated frame bytes.
587    pub estimated_bytes: usize,
588}
589
590/// Stateless distance-based replication cadence helper.
591#[derive(Clone, Copy, Debug, Default)]
592pub struct ReplicationCadence;
593
594impl ReplicationCadence {
595    /// Returns the target update frequency for a policy at a squared distance.
596    pub fn target_hz(policy: &CompiledSyncPolicy, distance_squared: f32) -> u16 {
597        let min_hz = policy.min_hz.max(1);
598        let max_hz = policy.max_hz.max(min_hz);
599        let radius_squared = policy.interest_radius * policy.interest_radius;
600        let closeness =
601            if radius_squared.is_finite() && radius_squared > 0.0 && distance_squared.is_finite() {
602                1.0 - (distance_squared / radius_squared).clamp(0.0, 1.0)
603            } else {
604                1.0
605            };
606        let span = f32::from(max_hz - min_hz);
607        let target = f32::from(min_hz) + span * closeness;
608        rounded_frequency_to_u16(target, min_hz, max_hz)
609    }
610
611    /// Returns the tick interval for a policy at a squared distance.
612    pub fn interval_ticks(
613        policy: &CompiledSyncPolicy,
614        station_tick_rate_hz: u16,
615        distance_squared: f32,
616    ) -> u64 {
617        let tick_rate = u64::from(station_tick_rate_hz.max(1));
618        let target_hz = u64::from(Self::target_hz(policy, distance_squared).max(1));
619        tick_rate.div_ceil(target_hz).max(1)
620    }
621
622    /// Returns whether a replication update should be sent at `now`.
623    pub fn should_send(
624        policy: &CompiledSyncPolicy,
625        station_tick_rate_hz: u16,
626        distance_squared: f32,
627        now: Tick,
628        last_sent: Option<Tick>,
629    ) -> bool {
630        let Some(last_sent) = last_sent else {
631            return true;
632        };
633        let interval = Self::interval_ticks(policy, station_tick_rate_hz, distance_squared);
634        now.get().saturating_sub(last_sent.get()) >= interval
635    }
636}
637
638/// Stateless replication priority scoring helper.
639#[derive(Clone, Copy, Debug, Default)]
640pub struct ReplicationPriority;
641
642#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
643fn rounded_frequency_to_u16(target: f32, min_hz: u16, max_hz: u16) -> u16 {
644    let bounded = target.round().clamp(f32::from(min_hz), f32::from(max_hz));
645    bounded as u16
646}
647
648#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
649fn normalized_score_to_u64(closeness: f32) -> u64 {
650    debug_assert!(closeness.is_finite() && (0.0..=1.0).contains(&closeness));
651    (closeness * 1_000_000.0).round() as u64
652}
653
654impl ReplicationPriority {
655    /// Returns a deterministic priority score for budgeted selection.
656    pub fn score(policy: &CompiledSyncPolicy, distance_squared: f32) -> u64 {
657        let weight = u64::from(policy.priority_weight.max(1));
658        let radius_squared = policy.interest_radius * policy.interest_radius;
659        let distance_score =
660            if radius_squared.is_finite() && radius_squared > 0.0 && distance_squared.is_finite() {
661                let closeness = 1.0 - (distance_squared / radius_squared).clamp(0.0, 1.0);
662                normalized_score_to_u64(closeness)
663            } else {
664                1_000_000
665            };
666        weight
667            .saturating_mul(1_000_000)
668            .saturating_add(distance_score)
669    }
670}
671
672#[derive(Clone, Copy, Debug, PartialEq)]
673struct PrioritizedReplicationCandidate {
674    handle: EntityHandle,
675    score: u64,
676    distance_squared: f32,
677}
678
679/// Reusable scratch storage for allocation-aware replication planning.
680#[derive(Clone, Debug, Default)]
681pub struct ReplicationScratch {
682    cell_query: CellQueryScratch,
683    prioritized: Vec<PrioritizedReplicationCandidate>,
684}
685
686impl ReplicationScratch {
687    /// Clears retained planning results while keeping allocated capacity.
688    pub fn clear(&mut self) {
689        self.cell_query.clear();
690        self.prioritized.clear();
691    }
692
693    /// Number of spatial candidates retained from the last query.
694    pub fn candidate_count(&self) -> usize {
695        self.cell_query.len()
696    }
697
698    /// Capacity retained for priority candidate sorting.
699    pub fn prioritized_capacity(&self) -> usize {
700        self.prioritized.capacity()
701    }
702
703    /// Work counters from the last spatial candidate query.
704    pub const fn query_stats(&self) -> CellQueryStats {
705        self.cell_query.stats()
706    }
707
708    /// Capacity retained for spatial candidate handles.
709    pub fn candidate_capacity(&self) -> usize {
710        self.cell_query.handle_capacity()
711    }
712
713    /// Capacity retained by spatial candidate deduplication.
714    pub fn candidate_dedup_capacity(&self) -> usize {
715        self.cell_query.dedup_capacity()
716    }
717
718    /// Capacity retained for cells matched by sparse spatial queries.
719    pub fn matching_cell_capacity(&self) -> usize {
720        self.cell_query.matching_cell_capacity()
721    }
722}
723
724/// Simple range/visibility-based replication planner.
725#[derive(Clone, Copy, Debug, Default)]
726pub struct ReplicationPlanner;
727
728impl ReplicationPlanner {
729    /// Plans a frame for one viewer using the station-local spatial index.
730    pub fn plan_for_viewer<F: VisibilityFilter>(
731        station: &Station,
732        index: &CellIndex,
733        policies: &PolicyTable,
734        viewer: &ViewerQuery,
735        filter: &F,
736        budget: ReplicationBudget,
737    ) -> ReplicationPlan {
738        let candidates = index.query_sphere(viewer.position, viewer.radius);
739        Self::plan_for_candidates_inner(
740            station,
741            &candidates,
742            policies,
743            viewer,
744            filter,
745            budget,
746            |_, _, _| true,
747        )
748    }
749
750    /// Plans a frame using caller-provided scratch storage.
751    pub fn plan_for_viewer_with_scratch<F: VisibilityFilter>(
752        station: &Station,
753        index: &CellIndex,
754        policies: &PolicyTable,
755        viewer: &ViewerQuery,
756        filter: &F,
757        budget: ReplicationBudget,
758        scratch: &mut ReplicationScratch,
759    ) -> ReplicationPlan {
760        let mut plan = ReplicationPlan::default();
761        Self::plan_for_viewer_with_scratch_into(
762            station, index, policies, viewer, filter, budget, scratch, &mut plan,
763        );
764        plan
765    }
766
767    /// Plans one viewer into caller-owned output while retaining its entity capacity.
768    #[allow(clippy::too_many_arguments)]
769    pub fn plan_for_viewer_with_scratch_into<F: VisibilityFilter>(
770        station: &Station,
771        index: &CellIndex,
772        policies: &PolicyTable,
773        viewer: &ViewerQuery,
774        filter: &F,
775        budget: ReplicationBudget,
776        scratch: &mut ReplicationScratch,
777        plan: &mut ReplicationPlan,
778    ) {
779        Self::plan_for_viewer_eligible_with_scratch_into(
780            station,
781            index,
782            policies,
783            viewer,
784            filter,
785            budget,
786            |_, _, _| true,
787            scratch,
788            plan,
789        );
790    }
791
792    /// Plans one viewer while applying caller-owned eligibility before budget use.
793    ///
794    /// The predicate can implement dirty, per-client delivery, or gameplay-owned
795    /// selection rules without transferring that state into `SectorSync`.
796    #[allow(clippy::too_many_arguments)]
797    pub fn plan_for_viewer_eligible_with_scratch_into<F, E>(
798        station: &Station,
799        index: &CellIndex,
800        policies: &PolicyTable,
801        viewer: &ViewerQuery,
802        filter: &F,
803        budget: ReplicationBudget,
804        eligible: E,
805        scratch: &mut ReplicationScratch,
806        plan: &mut ReplicationPlan,
807    ) where
808        F: VisibilityFilter,
809        E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
810    {
811        let candidates =
812            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
813        Self::plan_for_candidates_inner_into(
814            station,
815            candidates,
816            policies,
817            viewer,
818            filter,
819            budget,
820            |_, _, _| true,
821            eligible,
822            false,
823            plan,
824        );
825    }
826
827    /// Plans viewers in input order while reusing caller-provided scratch.
828    pub fn plan_for_viewers_with_scratch<F: VisibilityFilter>(
829        station: &Station,
830        index: &CellIndex,
831        policies: &PolicyTable,
832        viewers: &[ViewerQuery],
833        filter: &F,
834        budget: ReplicationBudget,
835        scratch: &mut ReplicationScratch,
836    ) -> ReplicationBatchResult {
837        let mut batch = ReplicationBatchResult {
838            plans: Vec::with_capacity(viewers.len()),
839            stats: ReplicationBatchStats::default(),
840        };
841        for viewer in viewers {
842            let plan = Self::plan_for_viewer_with_scratch(
843                station, index, policies, viewer, filter, budget, scratch,
844            );
845            batch.stats.record(&plan, scratch);
846            batch.plans.push(plan);
847        }
848        batch
849    }
850
851    /// Plans viewers into caller-owned output slots while retaining all capacities.
852    #[allow(clippy::too_many_arguments)]
853    pub fn plan_for_viewers_into<'a, F: VisibilityFilter>(
854        station: &Station,
855        index: &CellIndex,
856        policies: &PolicyTable,
857        viewers: &[ViewerQuery],
858        filter: &F,
859        budget: ReplicationBudget,
860        scratch: &mut ReplicationScratch,
861        batch: &'a mut ReplicationBatchScratch,
862    ) -> ReplicationBatchView<'a> {
863        Self::plan_for_viewers_eligible_into(
864            station,
865            index,
866            policies,
867            viewers,
868            filter,
869            budget,
870            |_, _, _| true,
871            scratch,
872            batch,
873        )
874    }
875
876    /// Plans viewers with caller-owned eligibility and reusable output storage.
877    #[allow(clippy::too_many_arguments)]
878    pub fn plan_for_viewers_eligible_into<'a, F, E>(
879        station: &Station,
880        index: &CellIndex,
881        policies: &PolicyTable,
882        viewers: &[ViewerQuery],
883        filter: &F,
884        budget: ReplicationBudget,
885        eligible: E,
886        scratch: &mut ReplicationScratch,
887        batch: &'a mut ReplicationBatchScratch,
888    ) -> ReplicationBatchView<'a>
889    where
890        F: VisibilityFilter,
891        E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
892    {
893        batch.prepare(viewers.len());
894        for (plan, viewer) in batch.plans[..viewers.len()].iter_mut().zip(viewers) {
895            Self::plan_for_viewer_eligible_with_scratch_into(
896                station, index, policies, viewer, filter, budget, &eligible, scratch, plan,
897            );
898            batch.stats.record(plan, scratch);
899        }
900        batch.view()
901    }
902
903    /// Plans one viewer using deterministic first-fit selection with a bounded scan.
904    ///
905    /// Planning stops as soon as the budget is full. This avoids exact
906    /// post-budget accounting and does not provide global priority ordering.
907    #[allow(clippy::too_many_arguments)]
908    pub fn plan_for_viewer_work_bounded_with_scratch_into<F, E>(
909        station: &Station,
910        index: &CellIndex,
911        policies: &PolicyTable,
912        viewer: &ViewerQuery,
913        filter: &F,
914        budget: ReplicationBudget,
915        eligible: E,
916        scratch: &mut ReplicationScratch,
917        plan: &mut ReplicationPlan,
918    ) where
919        F: VisibilityFilter,
920        E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
921    {
922        let candidates =
923            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
924        Self::plan_for_candidates_inner_into(
925            station,
926            candidates,
927            policies,
928            viewer,
929            filter,
930            budget,
931            |_, _, _| true,
932            eligible,
933            true,
934            plan,
935        );
936    }
937
938    /// Plans viewers using deterministic first-fit selection with bounded scans.
939    ///
940    /// Each plan stops examining candidates as soon as its budget is full. This
941    /// avoids exact post-budget accounting and does not provide global priority
942    /// ordering; see [`ReplicationStats::unexamined_after_budget`].
943    #[allow(clippy::too_many_arguments)]
944    pub fn plan_for_viewers_work_bounded_into<'a, F, E>(
945        station: &Station,
946        index: &CellIndex,
947        policies: &PolicyTable,
948        viewers: &[ViewerQuery],
949        filter: &F,
950        budget: ReplicationBudget,
951        eligible: E,
952        scratch: &mut ReplicationScratch,
953        batch: &'a mut ReplicationBatchScratch,
954    ) -> ReplicationBatchView<'a>
955    where
956        F: VisibilityFilter,
957        E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
958    {
959        batch.prepare(viewers.len());
960        for (plan, viewer) in batch.plans[..viewers.len()].iter_mut().zip(viewers) {
961            Self::plan_for_viewer_work_bounded_with_scratch_into(
962                station, index, policies, viewer, filter, budget, &eligible, scratch, plan,
963            );
964            batch.stats.record(plan, scratch);
965        }
966        batch.view()
967    }
968
969    /// Plans one range-only viewer using the optional SIMD candidate filter.
970    ///
971    /// With the `simd` feature this evaluates candidate distances in eight-lane
972    /// groups. Without it, the same API uses the scalar range-only planner.
973    pub fn plan_for_viewer_range_with_scratch(
974        station: &Station,
975        index: &CellIndex,
976        policies: &PolicyTable,
977        viewer: &ViewerQuery,
978        budget: ReplicationBudget,
979        scratch: &mut ReplicationScratch,
980    ) -> ReplicationPlan {
981        let mut plan = ReplicationPlan::default();
982        Self::plan_for_viewer_range_with_scratch_into(
983            station, index, policies, viewer, budget, scratch, &mut plan,
984        );
985        plan
986    }
987
988    /// Plans one range-only viewer into caller-owned reusable output.
989    #[allow(clippy::too_many_arguments)]
990    pub fn plan_for_viewer_range_with_scratch_into(
991        station: &Station,
992        index: &CellIndex,
993        policies: &PolicyTable,
994        viewer: &ViewerQuery,
995        budget: ReplicationBudget,
996        scratch: &mut ReplicationScratch,
997        plan: &mut ReplicationPlan,
998    ) {
999        let candidates =
1000            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
1001        Self::plan_for_range_candidates_into(station, candidates, policies, viewer, budget, plan);
1002    }
1003
1004    /// Plans a range-only viewer batch in input order with optional SIMD filtering.
1005    pub fn plan_for_viewers_range_with_scratch(
1006        station: &Station,
1007        index: &CellIndex,
1008        policies: &PolicyTable,
1009        viewers: &[ViewerQuery],
1010        budget: ReplicationBudget,
1011        scratch: &mut ReplicationScratch,
1012    ) -> ReplicationBatchResult {
1013        let mut batch = ReplicationBatchResult {
1014            plans: Vec::with_capacity(viewers.len()),
1015            stats: ReplicationBatchStats::default(),
1016        };
1017        for viewer in viewers {
1018            let plan = Self::plan_for_viewer_range_with_scratch(
1019                station, index, policies, viewer, budget, scratch,
1020            );
1021            batch.stats.record(&plan, scratch);
1022            batch.plans.push(plan);
1023        }
1024        batch
1025    }
1026
1027    /// Plans a range-only viewer batch into caller-owned reusable output slots.
1028    #[allow(clippy::too_many_arguments)]
1029    pub fn plan_for_viewers_range_into<'a>(
1030        station: &Station,
1031        index: &CellIndex,
1032        policies: &PolicyTable,
1033        viewers: &[ViewerQuery],
1034        budget: ReplicationBudget,
1035        scratch: &mut ReplicationScratch,
1036        batch: &'a mut ReplicationBatchScratch,
1037    ) -> ReplicationBatchView<'a> {
1038        batch.prepare(viewers.len());
1039        for (plan, viewer) in batch.plans[..viewers.len()].iter_mut().zip(viewers) {
1040            Self::plan_for_viewer_range_with_scratch_into(
1041                station, index, policies, viewer, budget, scratch, plan,
1042            );
1043            batch.stats.record(plan, scratch);
1044        }
1045        batch.view()
1046    }
1047
1048    /// Plans a frame and skips entities whose distance-based cadence has not elapsed.
1049    pub fn plan_for_viewer_with_cadence<F, L>(
1050        station: &Station,
1051        index: &CellIndex,
1052        policies: &PolicyTable,
1053        viewer: &ViewerQuery,
1054        filter: &F,
1055        budget: ReplicationBudget,
1056        last_sent: L,
1057    ) -> ReplicationPlan
1058    where
1059        F: VisibilityFilter,
1060        L: Fn(EntityHandle) -> Option<Tick>,
1061    {
1062        let tick_rate_hz = station.config().tick_rate_hz;
1063        let now = station.tick();
1064        let candidates = index.query_sphere(viewer.position, viewer.radius);
1065        Self::plan_for_candidates_inner(
1066            station,
1067            &candidates,
1068            policies,
1069            viewer,
1070            filter,
1071            budget,
1072            |handle, policy, distance_squared| {
1073                ReplicationCadence::should_send(
1074                    policy,
1075                    tick_rate_hz,
1076                    distance_squared,
1077                    now,
1078                    last_sent(handle),
1079                )
1080            },
1081        )
1082    }
1083
1084    /// Plans a cadence-aware frame using caller-provided scratch storage.
1085    #[allow(clippy::too_many_arguments)]
1086    pub fn plan_for_viewer_with_cadence_and_scratch<F, L>(
1087        station: &Station,
1088        index: &CellIndex,
1089        policies: &PolicyTable,
1090        viewer: &ViewerQuery,
1091        filter: &F,
1092        budget: ReplicationBudget,
1093        last_sent: L,
1094        scratch: &mut ReplicationScratch,
1095    ) -> ReplicationPlan
1096    where
1097        F: VisibilityFilter,
1098        L: Fn(EntityHandle) -> Option<Tick>,
1099    {
1100        let mut plan = ReplicationPlan::default();
1101        Self::plan_for_viewer_with_cadence_and_scratch_into(
1102            station, index, policies, viewer, filter, budget, last_sent, scratch, &mut plan,
1103        );
1104        plan
1105    }
1106
1107    /// Plans a cadence-aware frame into caller-owned reusable output.
1108    #[allow(clippy::too_many_arguments)]
1109    pub fn plan_for_viewer_with_cadence_and_scratch_into<F, L>(
1110        station: &Station,
1111        index: &CellIndex,
1112        policies: &PolicyTable,
1113        viewer: &ViewerQuery,
1114        filter: &F,
1115        budget: ReplicationBudget,
1116        last_sent: L,
1117        scratch: &mut ReplicationScratch,
1118        plan: &mut ReplicationPlan,
1119    ) where
1120        F: VisibilityFilter,
1121        L: Fn(EntityHandle) -> Option<Tick>,
1122    {
1123        let tick_rate_hz = station.config().tick_rate_hz;
1124        let now = station.tick();
1125        let candidates =
1126            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
1127        Self::plan_for_candidates_inner_into(
1128            station,
1129            candidates,
1130            policies,
1131            viewer,
1132            filter,
1133            budget,
1134            |handle, policy, distance_squared| {
1135                ReplicationCadence::should_send(
1136                    policy,
1137                    tick_rate_hz,
1138                    distance_squared,
1139                    now,
1140                    last_sent(handle),
1141                )
1142            },
1143            |_, _, _| true,
1144            false,
1145            plan,
1146        );
1147    }
1148
1149    /// Plans a frame and selects the highest-priority entities when budgeted.
1150    pub fn plan_for_viewer_prioritized<F: VisibilityFilter>(
1151        station: &Station,
1152        index: &CellIndex,
1153        policies: &PolicyTable,
1154        viewer: &ViewerQuery,
1155        filter: &F,
1156        budget: ReplicationBudget,
1157    ) -> ReplicationPlan {
1158        let candidates = index.query_sphere(viewer.position, viewer.radius);
1159        let mut prioritized = Vec::new();
1160        Self::plan_for_candidates_prioritized_inner(
1161            station,
1162            &candidates,
1163            policies,
1164            viewer,
1165            filter,
1166            budget,
1167            &mut prioritized,
1168            |_, _, _| true,
1169        )
1170    }
1171
1172    /// Plans a budgeted priority frame using caller-provided scratch storage.
1173    pub fn plan_for_viewer_prioritized_with_scratch<F: VisibilityFilter>(
1174        station: &Station,
1175        index: &CellIndex,
1176        policies: &PolicyTable,
1177        viewer: &ViewerQuery,
1178        filter: &F,
1179        budget: ReplicationBudget,
1180        scratch: &mut ReplicationScratch,
1181    ) -> ReplicationPlan {
1182        let mut plan = ReplicationPlan::default();
1183        Self::plan_for_viewer_prioritized_with_scratch_into(
1184            station, index, policies, viewer, filter, budget, scratch, &mut plan,
1185        );
1186        plan
1187    }
1188
1189    /// Plans a priority frame into caller-owned reusable output.
1190    #[allow(clippy::too_many_arguments)]
1191    pub fn plan_for_viewer_prioritized_with_scratch_into<F: VisibilityFilter>(
1192        station: &Station,
1193        index: &CellIndex,
1194        policies: &PolicyTable,
1195        viewer: &ViewerQuery,
1196        filter: &F,
1197        budget: ReplicationBudget,
1198        scratch: &mut ReplicationScratch,
1199        plan: &mut ReplicationPlan,
1200    ) {
1201        let candidates =
1202            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
1203        Self::plan_for_candidates_prioritized_inner_into(
1204            station,
1205            candidates,
1206            policies,
1207            viewer,
1208            filter,
1209            budget,
1210            &mut scratch.prioritized,
1211            |_, _, _| true,
1212            plan,
1213        );
1214    }
1215
1216    /// Plans a budgeted priority frame with distance-based cadence checks.
1217    pub fn plan_for_viewer_prioritized_with_cadence<F, L>(
1218        station: &Station,
1219        index: &CellIndex,
1220        policies: &PolicyTable,
1221        viewer: &ViewerQuery,
1222        filter: &F,
1223        budget: ReplicationBudget,
1224        last_sent: L,
1225    ) -> ReplicationPlan
1226    where
1227        F: VisibilityFilter,
1228        L: Fn(EntityHandle) -> Option<Tick>,
1229    {
1230        let tick_rate_hz = station.config().tick_rate_hz;
1231        let now = station.tick();
1232        let candidates = index.query_sphere(viewer.position, viewer.radius);
1233        let mut prioritized = Vec::new();
1234        Self::plan_for_candidates_prioritized_inner(
1235            station,
1236            &candidates,
1237            policies,
1238            viewer,
1239            filter,
1240            budget,
1241            &mut prioritized,
1242            |handle, policy, distance_squared| {
1243                ReplicationCadence::should_send(
1244                    policy,
1245                    tick_rate_hz,
1246                    distance_squared,
1247                    now,
1248                    last_sent(handle),
1249                )
1250            },
1251        )
1252    }
1253
1254    /// Plans a priority/cadence frame using caller-provided scratch storage.
1255    #[allow(clippy::too_many_arguments)]
1256    pub fn plan_for_viewer_prioritized_with_cadence_and_scratch<F, L>(
1257        station: &Station,
1258        index: &CellIndex,
1259        policies: &PolicyTable,
1260        viewer: &ViewerQuery,
1261        filter: &F,
1262        budget: ReplicationBudget,
1263        last_sent: L,
1264        scratch: &mut ReplicationScratch,
1265    ) -> ReplicationPlan
1266    where
1267        F: VisibilityFilter,
1268        L: Fn(EntityHandle) -> Option<Tick>,
1269    {
1270        let mut plan = ReplicationPlan::default();
1271        Self::plan_for_viewer_prioritized_with_cadence_and_scratch_into(
1272            station, index, policies, viewer, filter, budget, last_sent, scratch, &mut plan,
1273        );
1274        plan
1275    }
1276
1277    /// Plans a priority/cadence frame into caller-owned reusable output.
1278    #[allow(clippy::too_many_arguments)]
1279    pub fn plan_for_viewer_prioritized_with_cadence_and_scratch_into<F, L>(
1280        station: &Station,
1281        index: &CellIndex,
1282        policies: &PolicyTable,
1283        viewer: &ViewerQuery,
1284        filter: &F,
1285        budget: ReplicationBudget,
1286        last_sent: L,
1287        scratch: &mut ReplicationScratch,
1288        plan: &mut ReplicationPlan,
1289    ) where
1290        F: VisibilityFilter,
1291        L: Fn(EntityHandle) -> Option<Tick>,
1292    {
1293        let tick_rate_hz = station.config().tick_rate_hz;
1294        let now = station.tick();
1295        let candidates =
1296            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
1297        Self::plan_for_candidates_prioritized_inner_into(
1298            station,
1299            candidates,
1300            policies,
1301            viewer,
1302            filter,
1303            budget,
1304            &mut scratch.prioritized,
1305            |handle, policy, distance_squared| {
1306                ReplicationCadence::should_send(
1307                    policy,
1308                    tick_rate_hz,
1309                    distance_squared,
1310                    now,
1311                    last_sent(handle),
1312                )
1313            },
1314            plan,
1315        );
1316    }
1317
1318    fn plan_for_candidates_inner<F, C>(
1319        station: &Station,
1320        candidates: &[EntityHandle],
1321        policies: &PolicyTable,
1322        viewer: &ViewerQuery,
1323        filter: &F,
1324        budget: ReplicationBudget,
1325        cadence_allows: C,
1326    ) -> ReplicationPlan
1327    where
1328        F: VisibilityFilter,
1329        C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
1330    {
1331        let mut plan = ReplicationPlan::default();
1332        Self::plan_for_candidates_inner_into(
1333            station,
1334            candidates,
1335            policies,
1336            viewer,
1337            filter,
1338            budget,
1339            cadence_allows,
1340            |_, _, _| true,
1341            false,
1342            &mut plan,
1343        );
1344        plan
1345    }
1346
1347    #[allow(clippy::too_many_arguments)]
1348    fn plan_for_candidates_inner_into<F, C, E>(
1349        station: &Station,
1350        candidates: &[EntityHandle],
1351        policies: &PolicyTable,
1352        viewer: &ViewerQuery,
1353        filter: &F,
1354        budget: ReplicationBudget,
1355        cadence_allows: C,
1356        eligible: E,
1357        stop_when_budget_full: bool,
1358        plan: &mut ReplicationPlan,
1359    ) where
1360        F: VisibilityFilter,
1361        C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
1362        E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
1363    {
1364        let max_entities = viewer.max_entities.min(budget.max_entities);
1365        let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
1366        let hard_limit = max_entities.min(max_by_bytes);
1367        plan.entities.clear();
1368        plan.entities.reserve(hard_limit.min(candidates.len()));
1369        plan.stats = ReplicationStats {
1370            candidates: candidates.len(),
1371            ..ReplicationStats::default()
1372        };
1373
1374        if stop_when_budget_full && hard_limit == 0 {
1375            plan.stats.unexamined_after_budget = candidates.len();
1376            return;
1377        }
1378
1379        for (candidate_index, handle) in candidates.iter().enumerate() {
1380            let Some(entity) = station.get(*handle) else {
1381                continue;
1382            };
1383            plan.stats.considered += 1;
1384
1385            let Some(policy) = policies.get(entity.policy_id) else {
1386                continue;
1387            };
1388            let distance_squared = entity.position.distance_squared(viewer.position);
1389            let policy_radius_sq = policy.interest_radius * policy.interest_radius;
1390            if distance_squared > policy_radius_sq {
1391                continue;
1392            }
1393            if !filter.is_visible_with_distance(viewer, entity, distance_squared) {
1394                continue;
1395            }
1396            if !cadence_allows(*handle, policy, distance_squared) {
1397                plan.stats.skipped_by_cadence += 1;
1398                continue;
1399            }
1400            if !eligible(viewer, *handle, entity) {
1401                continue;
1402            }
1403
1404            if plan.entities.len() >= hard_limit {
1405                plan.stats.skipped_by_budget += 1;
1406                continue;
1407            }
1408
1409            plan.entities.push(*handle);
1410            if stop_when_budget_full && plan.entities.len() == hard_limit {
1411                plan.stats.unexamined_after_budget =
1412                    candidates.len().saturating_sub(candidate_index + 1);
1413                break;
1414            }
1415        }
1416
1417        plan.stats.selected = plan.entities.len();
1418        plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
1419    }
1420
1421    #[cfg(all(not(feature = "simd"), test))]
1422    fn plan_for_range_candidates(
1423        station: &Station,
1424        candidates: &[EntityHandle],
1425        policies: &PolicyTable,
1426        viewer: &ViewerQuery,
1427        budget: ReplicationBudget,
1428    ) -> ReplicationPlan {
1429        let mut plan = ReplicationPlan::default();
1430        Self::plan_for_range_candidates_into(
1431            station, candidates, policies, viewer, budget, &mut plan,
1432        );
1433        plan
1434    }
1435
1436    #[cfg(not(feature = "simd"))]
1437    fn plan_for_range_candidates_into(
1438        station: &Station,
1439        candidates: &[EntityHandle],
1440        policies: &PolicyTable,
1441        viewer: &ViewerQuery,
1442        budget: ReplicationBudget,
1443        plan: &mut ReplicationPlan,
1444    ) {
1445        Self::plan_for_candidates_inner_into(
1446            station,
1447            candidates,
1448            policies,
1449            viewer,
1450            &RangeOnlyVisibility,
1451            budget,
1452            |_, _, _| true,
1453            |_, _, _| true,
1454            false,
1455            plan,
1456        );
1457    }
1458
1459    #[cfg(all(feature = "simd", test))]
1460    fn plan_for_range_candidates(
1461        station: &Station,
1462        candidates: &[EntityHandle],
1463        policies: &PolicyTable,
1464        viewer: &ViewerQuery,
1465        budget: ReplicationBudget,
1466    ) -> ReplicationPlan {
1467        let mut plan = ReplicationPlan::default();
1468        Self::plan_for_range_candidates_into(
1469            station, candidates, policies, viewer, budget, &mut plan,
1470        );
1471        plan
1472    }
1473
1474    #[cfg(feature = "simd")]
1475    fn plan_for_range_candidates_into(
1476        station: &Station,
1477        candidates: &[EntityHandle],
1478        policies: &PolicyTable,
1479        viewer: &ViewerQuery,
1480        budget: ReplicationBudget,
1481        plan: &mut ReplicationPlan,
1482    ) {
1483        use wide::{CmpLe, f32x8};
1484
1485        const LANES: usize = 8;
1486        let max_entities = viewer.max_entities.min(budget.max_entities);
1487        let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
1488        let hard_limit = max_entities.min(max_by_bytes);
1489        plan.entities.clear();
1490        plan.entities.reserve(hard_limit.min(candidates.len()));
1491        plan.stats = ReplicationStats {
1492            candidates: candidates.len(),
1493            ..ReplicationStats::default()
1494        };
1495        let viewer_radius_squared = viewer.radius_squared();
1496
1497        for handles in candidates.chunks(LANES) {
1498            let mut distance_squared = [f32::NAN; LANES];
1499            let mut policy_radius_squared = [f32::NAN; LANES];
1500            let mut valid_lanes = 0_u8;
1501
1502            for (lane, handle) in handles.iter().copied().enumerate() {
1503                let Some(entity) = station.get(handle) else {
1504                    continue;
1505                };
1506                plan.stats.considered = plan.stats.considered.saturating_add(1);
1507                let Some(policy) = policies.get(entity.policy_id) else {
1508                    continue;
1509                };
1510                distance_squared[lane] = entity.position.distance_squared(viewer.position);
1511                policy_radius_squared[lane] = policy.interest_radius * policy.interest_radius;
1512                valid_lanes |= 1 << lane;
1513            }
1514
1515            let visible_lanes = u8::try_from(
1516                (f32x8::new(distance_squared).cmp_le(f32x8::new(policy_radius_squared))
1517                    & f32x8::new(distance_squared).cmp_le(f32x8::splat(viewer_radius_squared)))
1518                .move_mask(),
1519            )
1520            .expect("eight-lane SIMD mask fits u8")
1521                & valid_lanes;
1522
1523            for (lane, handle) in handles.iter().copied().enumerate() {
1524                if visible_lanes & (1 << lane) == 0 {
1525                    continue;
1526                }
1527                if plan.entities.len() >= hard_limit {
1528                    plan.stats.skipped_by_budget = plan.stats.skipped_by_budget.saturating_add(1);
1529                } else {
1530                    plan.entities.push(handle);
1531                }
1532            }
1533        }
1534
1535        plan.stats.selected = plan.entities.len();
1536        plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
1537    }
1538
1539    #[allow(clippy::too_many_arguments)]
1540    fn plan_for_candidates_prioritized_inner<F, C>(
1541        station: &Station,
1542        candidates: &[EntityHandle],
1543        policies: &PolicyTable,
1544        viewer: &ViewerQuery,
1545        filter: &F,
1546        budget: ReplicationBudget,
1547        eligible: &mut Vec<PrioritizedReplicationCandidate>,
1548        cadence_allows: C,
1549    ) -> ReplicationPlan
1550    where
1551        F: VisibilityFilter,
1552        C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
1553    {
1554        let mut plan = ReplicationPlan::default();
1555        Self::plan_for_candidates_prioritized_inner_into(
1556            station,
1557            candidates,
1558            policies,
1559            viewer,
1560            filter,
1561            budget,
1562            eligible,
1563            cadence_allows,
1564            &mut plan,
1565        );
1566        plan
1567    }
1568
1569    #[allow(clippy::too_many_arguments)]
1570    fn plan_for_candidates_prioritized_inner_into<F, C>(
1571        station: &Station,
1572        candidates: &[EntityHandle],
1573        policies: &PolicyTable,
1574        viewer: &ViewerQuery,
1575        filter: &F,
1576        budget: ReplicationBudget,
1577        eligible: &mut Vec<PrioritizedReplicationCandidate>,
1578        cadence_allows: C,
1579        plan: &mut ReplicationPlan,
1580    ) where
1581        F: VisibilityFilter,
1582        C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
1583    {
1584        let max_entities = viewer.max_entities.min(budget.max_entities);
1585        let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
1586        let hard_limit = max_entities.min(max_by_bytes);
1587        plan.entities.clear();
1588        plan.entities.reserve(hard_limit.min(candidates.len()));
1589        plan.stats = ReplicationStats {
1590            candidates: candidates.len(),
1591            ..ReplicationStats::default()
1592        };
1593        eligible.clear();
1594
1595        for handle in candidates {
1596            let Some(entity) = station.get(*handle) else {
1597                continue;
1598            };
1599            plan.stats.considered += 1;
1600
1601            let Some(policy) = policies.get(entity.policy_id) else {
1602                continue;
1603            };
1604            let distance_squared = entity.position.distance_squared(viewer.position);
1605            let policy_radius_sq = policy.interest_radius * policy.interest_radius;
1606            if distance_squared > policy_radius_sq {
1607                continue;
1608            }
1609            if !filter.is_visible_with_distance(viewer, entity, distance_squared) {
1610                continue;
1611            }
1612            if !cadence_allows(*handle, policy, distance_squared) {
1613                plan.stats.skipped_by_cadence += 1;
1614                continue;
1615            }
1616
1617            eligible.push(PrioritizedReplicationCandidate {
1618                handle: *handle,
1619                score: ReplicationPriority::score(policy, distance_squared),
1620                distance_squared,
1621            });
1622        }
1623
1624        let selected = prioritize_candidates(eligible, hard_limit);
1625
1626        plan.stats.skipped_by_budget = eligible.len().saturating_sub(selected);
1627        plan.entities.extend(
1628            eligible
1629                .iter()
1630                .take(selected)
1631                .map(|candidate| candidate.handle),
1632        );
1633        plan.stats.selected = plan.entities.len();
1634        plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
1635    }
1636}
1637
1638fn compare_prioritized_candidates(
1639    left: &PrioritizedReplicationCandidate,
1640    right: &PrioritizedReplicationCandidate,
1641) -> Ordering {
1642    right
1643        .score
1644        .cmp(&left.score)
1645        .then_with(|| left.distance_squared.total_cmp(&right.distance_squared))
1646        .then_with(|| left.handle.cmp(&right.handle))
1647}
1648
1649fn prioritize_candidates(eligible: &mut [PrioritizedReplicationCandidate], limit: usize) -> usize {
1650    let selected = eligible.len().min(limit);
1651    if selected == 0 {
1652        return 0;
1653    }
1654    if selected.saturating_mul(2) < eligible.len() {
1655        eligible.select_nth_unstable_by(selected, compare_prioritized_candidates);
1656        eligible[..selected].sort_by(compare_prioritized_candidates);
1657    } else {
1658        eligible.sort_by(compare_prioritized_candidates);
1659    }
1660    selected
1661}
1662
1663#[cfg(test)]
1664mod tests {
1665    use super::*;
1666    use crate::entity::EntityTags;
1667    use crate::ids::{ClientId, EntityId, InstanceId, NodeId, PolicyId, StationId};
1668    use crate::interest::{AndVisibility, FrustumVisibility, RangeOnlyVisibility, TagVisibility};
1669    use crate::policy::CompiledSyncPolicy;
1670    use crate::spatial::{Aabb3, Bounds, Frustum3, GridSpec, Position3};
1671    use crate::station::{Station, StationConfig};
1672
1673    #[test]
1674    fn planner_applies_composed_frustum_visibility_filter() {
1675        let mut station = Station::new(StationConfig {
1676            station_id: StationId::new(1),
1677            node_id: NodeId::new(1),
1678            instance_id: InstanceId::new(1),
1679            tick_rate_hz: 20,
1680        });
1681        let grid = GridSpec::new(16.0).expect("grid is valid");
1682        let mut index = CellIndex::new(grid);
1683        let mut policies = PolicyTable::default();
1684        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
1685
1686        let visible = station
1687            .spawn_owned(
1688                EntityId::new(1),
1689                Position3::new(10.0, 0.0, 0.0),
1690                Bounds::Point,
1691                PolicyId::new(1),
1692            )
1693            .expect("spawn visible");
1694        let outside_frustum = station
1695            .spawn_owned(
1696                EntityId::new(2),
1697                Position3::new(-10.0, 0.0, 0.0),
1698                Bounds::Point,
1699                PolicyId::new(1),
1700            )
1701            .expect("spawn outside frustum");
1702        index.upsert(visible, Position3::new(10.0, 0.0, 0.0), Bounds::Point);
1703        index.upsert(
1704            outside_frustum,
1705            Position3::new(-10.0, 0.0, 0.0),
1706            Bounds::Point,
1707        );
1708
1709        let viewer = ViewerQuery {
1710            client_id: ClientId::new(7),
1711            position: Position3::new(0.0, 0.0, 0.0),
1712            radius: 128.0,
1713            max_entities: 8,
1714        };
1715        let frustum = Frustum3::from_aabb(Aabb3::new(
1716            Position3::new(0.0, -20.0, -20.0),
1717            Position3::new(80.0, 20.0, 20.0),
1718        ));
1719        let filter = AndVisibility::new(RangeOnlyVisibility, FrustumVisibility::new(frustum));
1720
1721        let plan = ReplicationPlanner::plan_for_viewer(
1722            &station,
1723            &index,
1724            &policies,
1725            &viewer,
1726            &filter,
1727            ReplicationBudget::default(),
1728        );
1729
1730        assert_eq!(plan.entities, vec![visible]);
1731        assert_eq!(plan.stats.selected, 1);
1732        assert_eq!(plan.stats.considered, 2);
1733    }
1734
1735    #[test]
1736    fn planner_applies_tag_visibility_filter() {
1737        let mut station = Station::new(StationConfig {
1738            station_id: StationId::new(1),
1739            node_id: NodeId::new(1),
1740            instance_id: InstanceId::new(1),
1741            tick_rate_hz: 20,
1742        });
1743        let grid = GridSpec::new(16.0).expect("grid is valid");
1744        let mut index = CellIndex::new(grid);
1745        let mut policies = PolicyTable::default();
1746        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
1747
1748        let static_visible = station
1749            .spawn_owned(
1750                EntityId::new(1),
1751                Position3::new(10.0, 0.0, 0.0),
1752                Bounds::Point,
1753                PolicyId::new(1),
1754            )
1755            .expect("spawn static");
1756        let fast_mover = station
1757            .spawn_owned(
1758                EntityId::new(2),
1759                Position3::new(12.0, 0.0, 0.0),
1760                Bounds::Point,
1761                PolicyId::new(1),
1762            )
1763            .expect("spawn mover");
1764        station
1765            .set_tags(static_visible, EntityTags::from_bits(0b001))
1766            .expect("tag static");
1767        station
1768            .set_tags(fast_mover, EntityTags::from_bits(0b010))
1769            .expect("tag mover");
1770        index.upsert(
1771            static_visible,
1772            Position3::new(10.0, 0.0, 0.0),
1773            Bounds::Point,
1774        );
1775        index.upsert(fast_mover, Position3::new(12.0, 0.0, 0.0), Bounds::Point);
1776
1777        let viewer = ViewerQuery {
1778            client_id: ClientId::new(7),
1779            position: Position3::new(0.0, 0.0, 0.0),
1780            radius: 128.0,
1781            max_entities: 8,
1782        };
1783        let filter = AndVisibility::new(
1784            RangeOnlyVisibility,
1785            TagVisibility::new(EntityTags::from_bits(0b001), EntityTags::from_bits(0b010)),
1786        );
1787
1788        let plan = ReplicationPlanner::plan_for_viewer(
1789            &station,
1790            &index,
1791            &policies,
1792            &viewer,
1793            &filter,
1794            ReplicationBudget::default(),
1795        );
1796
1797        assert_eq!(plan.entities, vec![static_visible]);
1798        assert_eq!(plan.stats.selected, 1);
1799        assert_eq!(plan.stats.considered, 2);
1800    }
1801
1802    #[test]
1803    fn caller_eligibility_filters_before_replication_budget() {
1804        let mut station = Station::new(StationConfig {
1805            station_id: StationId::new(1),
1806            node_id: NodeId::new(1),
1807            instance_id: InstanceId::new(1),
1808            tick_rate_hz: 20,
1809        });
1810        let mut index = CellIndex::new(GridSpec::new(16.0).expect("grid is valid"));
1811        let mut policies = PolicyTable::default();
1812        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
1813        let handles = (1_u16..=3)
1814            .map(|id| {
1815                let position = Position3::new(f32::from(id), 0.0, 0.0);
1816                let handle = station
1817                    .spawn_owned(
1818                        EntityId::new(u64::from(id)),
1819                        position,
1820                        Bounds::Point,
1821                        PolicyId::new(1),
1822                    )
1823                    .expect("entity id is unique");
1824                index.upsert(handle, position, Bounds::Point);
1825                handle
1826            })
1827            .collect::<Vec<_>>();
1828        let viewer = ViewerQuery {
1829            client_id: ClientId::new(1),
1830            position: Position3::new(0.0, 0.0, 0.0),
1831            radius: 128.0,
1832            max_entities: 1,
1833        };
1834        let mut scratch = ReplicationScratch::default();
1835        let mut plan = ReplicationPlan::default();
1836
1837        ReplicationPlanner::plan_for_viewer_eligible_with_scratch_into(
1838            &station,
1839            &index,
1840            &policies,
1841            &viewer,
1842            &RangeOnlyVisibility,
1843            ReplicationBudget {
1844                max_entities: 1,
1845                ..ReplicationBudget::default()
1846            },
1847            |_, handle, _| handle == handles[2],
1848            &mut scratch,
1849            &mut plan,
1850        );
1851
1852        assert_eq!(plan.entities, vec![handles[2]]);
1853        assert_eq!(plan.stats.selected, 1);
1854        assert_eq!(plan.stats.skipped_by_budget, 0);
1855    }
1856
1857    #[test]
1858    fn work_bounded_planner_stops_after_first_fit_budget() {
1859        let mut station = Station::new(StationConfig {
1860            station_id: StationId::new(1),
1861            node_id: NodeId::new(1),
1862            instance_id: InstanceId::new(1),
1863            tick_rate_hz: 20,
1864        });
1865        let mut index = CellIndex::new(GridSpec::new(16.0).expect("grid is valid"));
1866        let mut policies = PolicyTable::default();
1867        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
1868        for id in 1_u16..=8 {
1869            let position = Position3::new(f32::from(id), 0.0, 0.0);
1870            let handle = station
1871                .spawn_owned(
1872                    EntityId::new(u64::from(id)),
1873                    position,
1874                    Bounds::Point,
1875                    PolicyId::new(1),
1876                )
1877                .expect("entity id is unique");
1878            index.upsert(handle, position, Bounds::Point);
1879        }
1880        let viewer = ViewerQuery {
1881            client_id: ClientId::new(1),
1882            position: Position3::new(0.0, 0.0, 0.0),
1883            radius: 128.0,
1884            max_entities: 2,
1885        };
1886        let mut scratch = ReplicationScratch::default();
1887        let mut plan = ReplicationPlan::default();
1888
1889        ReplicationPlanner::plan_for_viewer_work_bounded_with_scratch_into(
1890            &station,
1891            &index,
1892            &policies,
1893            &viewer,
1894            &RangeOnlyVisibility,
1895            ReplicationBudget {
1896                max_entities: 2,
1897                ..ReplicationBudget::default()
1898            },
1899            |_, _, _| true,
1900            &mut scratch,
1901            &mut plan,
1902        );
1903
1904        assert_eq!(plan.stats.selected, 2);
1905        assert_eq!(plan.stats.considered, 2);
1906        assert_eq!(plan.stats.skipped_by_budget, 0);
1907        assert_eq!(plan.stats.unexamined_after_budget, 6);
1908    }
1909
1910    #[test]
1911    #[allow(clippy::too_many_lines)]
1912    fn range_batch_matches_ordered_scalar_plans() {
1913        let mut station = Station::new(StationConfig {
1914            station_id: StationId::new(1),
1915            node_id: NodeId::new(1),
1916            instance_id: InstanceId::new(1),
1917            tick_rate_hz: 128,
1918        });
1919        let grid = GridSpec::new(16.0).expect("grid is valid");
1920        let mut index = CellIndex::new(grid);
1921        let mut policies = PolicyTable::default();
1922        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 128, 96.0));
1923        for entity_index in 0_u16..24 {
1924            let position = Position3::new(f32::from(entity_index) * 8.0 - 64.0, 0.0, 0.0);
1925            let handle = station
1926                .spawn_owned(
1927                    EntityId::new(u64::from(entity_index)),
1928                    position,
1929                    Bounds::Point,
1930                    PolicyId::new(1),
1931                )
1932                .expect("entity id is unique");
1933            index.upsert(handle, position, Bounds::Point);
1934        }
1935        let viewers = [
1936            ViewerQuery {
1937                client_id: ClientId::new(1),
1938                position: Position3::new(0.0, 0.0, 0.0),
1939                radius: 80.0,
1940                max_entities: 32,
1941            },
1942            ViewerQuery {
1943                client_id: ClientId::new(2),
1944                position: Position3::new(48.0, 0.0, 0.0),
1945                radius: 48.0,
1946                max_entities: 8,
1947            },
1948        ];
1949        let mut scalar_scratch = ReplicationScratch::default();
1950        let expected = viewers
1951            .iter()
1952            .map(|viewer| {
1953                ReplicationPlanner::plan_for_viewer_with_scratch(
1954                    &station,
1955                    &index,
1956                    &policies,
1957                    viewer,
1958                    &RangeOnlyVisibility,
1959                    ReplicationBudget::default(),
1960                    &mut scalar_scratch,
1961                )
1962            })
1963            .collect::<Vec<_>>();
1964
1965        let mut batch_scratch = ReplicationScratch::default();
1966        let batch = ReplicationPlanner::plan_for_viewers_range_with_scratch(
1967            &station,
1968            &index,
1969            &policies,
1970            &viewers,
1971            ReplicationBudget::default(),
1972            &mut batch_scratch,
1973        );
1974
1975        assert_eq!(batch.plans, expected);
1976        assert_eq!(batch.stats.viewers, viewers.len());
1977        assert_eq!(
1978            batch.stats.selected,
1979            expected.iter().map(|plan| plan.stats.selected).sum()
1980        );
1981        assert_eq!(
1982            batch.stats.grid_queries + batch.stats.occupied_queries,
1983            viewers.len()
1984        );
1985
1986        let mut reusable_planning = ReplicationScratch::default();
1987        let mut reusable_output = ReplicationBatchScratch::new();
1988        {
1989            let reused = ReplicationPlanner::plan_for_viewers_range_into(
1990                &station,
1991                &index,
1992                &policies,
1993                &viewers,
1994                ReplicationBudget::default(),
1995                &mut reusable_planning,
1996                &mut reusable_output,
1997            );
1998            assert_eq!(reused.plans, expected);
1999            assert_eq!(reused.stats, batch.stats);
2000        }
2001        let retained_capacity = reusable_output.retained_entity_capacity();
2002        assert_eq!(reusable_output.retained_plan_slots(), viewers.len());
2003
2004        {
2005            let reused = ReplicationPlanner::plan_for_viewers_into(
2006                &station,
2007                &index,
2008                &policies,
2009                &viewers,
2010                &RangeOnlyVisibility,
2011                ReplicationBudget::default(),
2012                &mut reusable_planning,
2013                &mut reusable_output,
2014            );
2015            assert_eq!(reused.plans, expected);
2016            assert_eq!(reused.stats, batch.stats);
2017        }
2018
2019        let reused = ReplicationPlanner::plan_for_viewers_range_into(
2020            &station,
2021            &index,
2022            &policies,
2023            &viewers[..1],
2024            ReplicationBudget::default(),
2025            &mut reusable_planning,
2026            &mut reusable_output,
2027        );
2028        assert_eq!(reused.plans, &expected[..1]);
2029        assert_eq!(reusable_output.retained_plan_slots(), viewers.len());
2030        assert_eq!(
2031            reusable_output.retained_entity_capacity(),
2032            retained_capacity
2033        );
2034    }
2035
2036    #[test]
2037    fn range_batch_preserves_scalar_nan_radius_semantics() {
2038        let mut station = Station::new(StationConfig {
2039            station_id: StationId::new(1),
2040            node_id: NodeId::new(1),
2041            instance_id: InstanceId::new(1),
2042            tick_rate_hz: 128,
2043        });
2044        let mut policies = PolicyTable::default();
2045        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 128, 96.0));
2046        let handle = station
2047            .spawn_owned(
2048                EntityId::new(1),
2049                Position3::new(1.0, 2.0, 3.0),
2050                Bounds::Point,
2051                PolicyId::new(1),
2052            )
2053            .expect("spawn entity");
2054        let viewer = ViewerQuery {
2055            client_id: ClientId::new(1),
2056            position: Position3::new(0.0, 0.0, 0.0),
2057            radius: f32::NAN,
2058            max_entities: 8,
2059        };
2060        let candidates = [handle];
2061        let scalar = ReplicationPlanner::plan_for_candidates_inner(
2062            &station,
2063            &candidates,
2064            &policies,
2065            &viewer,
2066            &RangeOnlyVisibility,
2067            ReplicationBudget::default(),
2068            |_, _, _| true,
2069        );
2070        let range = ReplicationPlanner::plan_for_range_candidates(
2071            &station,
2072            &candidates,
2073            &policies,
2074            &viewer,
2075            ReplicationBudget::default(),
2076        );
2077
2078        assert!(scalar.entities.is_empty());
2079        assert_eq!(range, scalar);
2080    }
2081
2082    #[test]
2083    fn cadence_scales_interval_by_squared_distance() {
2084        let policy = CompiledSyncPolicy::new(PolicyId::new(1), 2, 20, 100.0);
2085
2086        assert_eq!(ReplicationCadence::target_hz(&policy, 0.0), 20);
2087        assert_eq!(ReplicationCadence::interval_ticks(&policy, 20, 0.0), 1);
2088        assert_eq!(ReplicationCadence::target_hz(&policy, 100.0_f32 * 100.0), 2);
2089        assert_eq!(
2090            ReplicationCadence::interval_ticks(&policy, 20, 100.0_f32 * 100.0),
2091            10
2092        );
2093    }
2094
2095    #[test]
2096    fn priority_score_prefers_weight_then_distance() {
2097        let mut low = CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 100.0);
2098        low.priority_weight = 1;
2099        let mut high = CompiledSyncPolicy::new(PolicyId::new(2), 1, 20, 100.0);
2100        high.priority_weight = 10;
2101
2102        assert!(
2103            ReplicationPriority::score(&high, 90.0 * 90.0) > ReplicationPriority::score(&low, 0.0)
2104        );
2105        assert!(
2106            ReplicationPriority::score(&low, 0.0) > ReplicationPriority::score(&low, 90.0 * 90.0)
2107        );
2108    }
2109
2110    #[test]
2111    fn top_k_priority_selection_matches_full_sort_for_all_budget_edges() {
2112        let candidates = (0_u32..257)
2113            .map(|index| PrioritizedReplicationCandidate {
2114                handle: EntityHandle::new(index, index % 3),
2115                score: u64::from(index.wrapping_mul(37) % 23),
2116                distance_squared: f32::from(
2117                    u16::try_from(index.wrapping_mul(19) % 41).expect("distance fits u16"),
2118                ),
2119            })
2120            .collect::<Vec<_>>();
2121
2122        for limit in [0, 1, 7, 64, 256, 257, 300] {
2123            let mut expected = candidates.clone();
2124            expected.sort_by(compare_prioritized_candidates);
2125            expected.truncate(limit.min(expected.len()));
2126            let mut actual = candidates.clone();
2127            let selected = prioritize_candidates(&mut actual, limit);
2128
2129            assert_eq!(selected, expected.len());
2130            assert_eq!(&actual[..selected], expected.as_slice());
2131        }
2132    }
2133
2134    #[test]
2135    fn planner_with_cadence_skips_recent_far_entities() {
2136        let mut station = Station::new(StationConfig {
2137            station_id: StationId::new(1),
2138            node_id: NodeId::new(1),
2139            instance_id: InstanceId::new(1),
2140            tick_rate_hz: 20,
2141        });
2142        for _ in 0..10 {
2143            station.advance_tick();
2144        }
2145        let grid = GridSpec::new(16.0).expect("grid is valid");
2146        let mut index = CellIndex::new(grid);
2147        let mut policies = PolicyTable::default();
2148        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 2, 20, 128.0));
2149
2150        let near = station
2151            .spawn_owned(
2152                EntityId::new(1),
2153                Position3::new(0.0, 0.0, 0.0),
2154                Bounds::Point,
2155                PolicyId::new(1),
2156            )
2157            .expect("spawn near");
2158        let far = station
2159            .spawn_owned(
2160                EntityId::new(2),
2161                Position3::new(120.0, 0.0, 0.0),
2162                Bounds::Point,
2163                PolicyId::new(1),
2164            )
2165            .expect("spawn far");
2166        index.upsert(near, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
2167        index.upsert(far, Position3::new(120.0, 0.0, 0.0), Bounds::Point);
2168
2169        let viewer = ViewerQuery {
2170            client_id: ClientId::new(7),
2171            position: Position3::new(0.0, 0.0, 0.0),
2172            radius: 128.0,
2173            max_entities: 8,
2174        };
2175        let plan = ReplicationPlanner::plan_for_viewer_with_cadence(
2176            &station,
2177            &index,
2178            &policies,
2179            &viewer,
2180            &RangeOnlyVisibility,
2181            ReplicationBudget::default(),
2182            |_| Some(Tick::new(9)),
2183        );
2184
2185        assert_eq!(plan.entities, vec![near]);
2186        assert_eq!(plan.stats.selected, 1);
2187        assert_eq!(plan.stats.skipped_by_cadence, 1);
2188
2189        let mut scratch = ReplicationScratch::default();
2190        let mut reusable = ReplicationPlan::default();
2191        ReplicationPlanner::plan_for_viewer_with_cadence_and_scratch_into(
2192            &station,
2193            &index,
2194            &policies,
2195            &viewer,
2196            &RangeOnlyVisibility,
2197            ReplicationBudget::default(),
2198            |_| Some(Tick::new(9)),
2199            &mut scratch,
2200            &mut reusable,
2201        );
2202        assert_eq!(reusable, plan);
2203    }
2204
2205    #[test]
2206    fn prioritized_planner_uses_policy_weight_under_budget() {
2207        let mut station = Station::new(StationConfig {
2208            station_id: StationId::new(1),
2209            node_id: NodeId::new(1),
2210            instance_id: InstanceId::new(1),
2211            tick_rate_hz: 20,
2212        });
2213        let grid = GridSpec::new(16.0).expect("grid is valid");
2214        let mut index = CellIndex::new(grid);
2215        let mut policies = PolicyTable::default();
2216        let mut low = CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0);
2217        low.priority_weight = 1;
2218        let mut high = CompiledSyncPolicy::new(PolicyId::new(2), 1, 20, 128.0);
2219        high.priority_weight = 10;
2220        policies.set(low);
2221        policies.set(high);
2222
2223        let near_low = station
2224            .spawn_owned(
2225                EntityId::new(1),
2226                Position3::new(0.0, 0.0, 0.0),
2227                Bounds::Point,
2228                PolicyId::new(1),
2229            )
2230            .expect("spawn near low priority");
2231        let far_high = station
2232            .spawn_owned(
2233                EntityId::new(2),
2234                Position3::new(96.0, 0.0, 0.0),
2235                Bounds::Point,
2236                PolicyId::new(2),
2237            )
2238            .expect("spawn far high priority");
2239        index.upsert(near_low, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
2240        index.upsert(far_high, Position3::new(96.0, 0.0, 0.0), Bounds::Point);
2241
2242        let viewer = ViewerQuery {
2243            client_id: ClientId::new(7),
2244            position: Position3::new(0.0, 0.0, 0.0),
2245            radius: 128.0,
2246            max_entities: 1,
2247        };
2248        let plan = ReplicationPlanner::plan_for_viewer_prioritized(
2249            &station,
2250            &index,
2251            &policies,
2252            &viewer,
2253            &RangeOnlyVisibility,
2254            ReplicationBudget {
2255                max_entities: 1,
2256                max_bytes: 32,
2257                estimated_entity_bytes: 32,
2258            },
2259        );
2260
2261        assert_eq!(plan.entities, vec![far_high]);
2262        assert_eq!(plan.stats.selected, 1);
2263        assert_eq!(plan.stats.skipped_by_budget, 1);
2264
2265        let mut scratch = ReplicationScratch::default();
2266        let scratch_plan = ReplicationPlanner::plan_for_viewer_prioritized_with_scratch(
2267            &station,
2268            &index,
2269            &policies,
2270            &viewer,
2271            &RangeOnlyVisibility,
2272            ReplicationBudget {
2273                max_entities: 1,
2274                max_bytes: 32,
2275                estimated_entity_bytes: 32,
2276            },
2277            &mut scratch,
2278        );
2279        assert_eq!(scratch_plan.entities, plan.entities);
2280        assert_eq!(scratch_plan.stats, plan.stats);
2281        assert_eq!(scratch.candidate_count(), 2);
2282        assert!(scratch.prioritized_capacity() >= 2);
2283        assert_eq!(scratch.query_stats().candidate_handles, 2);
2284        assert!(scratch.candidate_capacity() >= 2);
2285        assert!(scratch.candidate_dedup_capacity() >= 2);
2286
2287        let budget = ReplicationBudget {
2288            max_entities: 1,
2289            max_bytes: 32,
2290            estimated_entity_bytes: 32,
2291        };
2292        assert_prioritized_output_reuse(
2293            &station,
2294            &index,
2295            &policies,
2296            &viewer,
2297            budget,
2298            &plan,
2299            &mut scratch,
2300        );
2301    }
2302
2303    fn assert_prioritized_output_reuse(
2304        station: &Station,
2305        index: &CellIndex,
2306        policies: &PolicyTable,
2307        viewer: &ViewerQuery,
2308        budget: ReplicationBudget,
2309        expected: &ReplicationPlan,
2310        scratch: &mut ReplicationScratch,
2311    ) {
2312        let mut reusable = ReplicationPlan::default();
2313        ReplicationPlanner::plan_for_viewer_prioritized_with_scratch_into(
2314            station,
2315            index,
2316            policies,
2317            viewer,
2318            &RangeOnlyVisibility,
2319            budget,
2320            scratch,
2321            &mut reusable,
2322        );
2323        let retained_entities = reusable.entities.as_ptr();
2324        assert_eq!(&reusable, expected);
2325        ReplicationPlanner::plan_for_viewer_prioritized_with_cadence_and_scratch_into(
2326            station,
2327            index,
2328            policies,
2329            viewer,
2330            &RangeOnlyVisibility,
2331            budget,
2332            |_| None,
2333            scratch,
2334            &mut reusable,
2335        );
2336        assert_eq!(reusable.entities.as_ptr(), retained_entities);
2337    }
2338
2339    #[test]
2340    fn replication_tracker_records_sent_ack_and_prune() {
2341        let client_id = ClientId::new(7);
2342        let first = EntityHandle::new(1, 0);
2343        let second = EntityHandle::new(2, 0);
2344        let plan = ReplicationPlan {
2345            entities: vec![first, second],
2346            stats: ReplicationStats::default(),
2347        };
2348        let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 4 });
2349
2350        let recorded = tracker
2351            .record_plan_sent(client_id, &plan, Tick::new(10))
2352            .expect("recording should fit");
2353        assert_eq!(recorded, 2);
2354        assert_eq!(tracker.last_sent(client_id, first), Some(Tick::new(10)));
2355        assert_eq!(tracker.stats().entries, 2);
2356        assert_eq!(tracker.stats().sent_records, 2);
2357
2358        assert!(tracker.acknowledge(client_id, first, Tick::new(11)));
2359        assert_eq!(
2360            tracker
2361                .get(client_id, first)
2362                .expect("tracked record")
2363                .last_acked,
2364            Some(Tick::new(11))
2365        );
2366        assert_eq!(tracker.stats().acked_records, 1);
2367
2368        assert_eq!(tracker.prune_sent_before(Tick::new(11)), 2);
2369        assert!(tracker.is_empty());
2370        assert_eq!(tracker.stats().pruned_records, 2);
2371    }
2372
2373    #[test]
2374    fn replication_tracker_rejects_capacity_without_partial_insert() {
2375        let client_id = ClientId::new(7);
2376        let plan = ReplicationPlan {
2377            entities: vec![EntityHandle::new(1, 0), EntityHandle::new(2, 0)],
2378            stats: ReplicationStats::default(),
2379        };
2380        let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 1 });
2381
2382        let error = tracker
2383            .record_plan_sent(client_id, &plan, Tick::new(10))
2384            .expect_err("recording should exceed capacity");
2385
2386        assert_eq!(
2387            error,
2388            ReplicationTrackerError::CapacityExceeded {
2389                current: 0,
2390                needed: 2,
2391                max: 1,
2392            }
2393        );
2394        assert!(tracker.is_empty());
2395        assert_eq!(tracker.stats().sent_records, 0);
2396    }
2397
2398    #[test]
2399    fn replication_tracker_uses_exact_capacity_check_near_limit() {
2400        let client_id = ClientId::new(7);
2401        let first = EntityHandle::new(1, 0);
2402        let second = EntityHandle::new(2, 0);
2403        let third = EntityHandle::new(3, 0);
2404        let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 2 });
2405        tracker
2406            .record_plan_sent(
2407                client_id,
2408                &ReplicationPlan {
2409                    entities: vec![first],
2410                    stats: ReplicationStats::default(),
2411                },
2412                Tick::new(1),
2413            )
2414            .expect("initial record should fit");
2415        tracker
2416            .record_plan_sent(
2417                client_id,
2418                &ReplicationPlan {
2419                    entities: vec![first, second],
2420                    stats: ReplicationStats::default(),
2421                },
2422                Tick::new(2),
2423            )
2424            .expect("one existing and one new record should fit exactly");
2425
2426        let error = tracker
2427            .record_plan_sent(
2428                client_id,
2429                &ReplicationPlan {
2430                    entities: vec![first, second, third],
2431                    stats: ReplicationStats::default(),
2432                },
2433                Tick::new(3),
2434            )
2435            .expect_err("new record should exceed exact capacity");
2436        assert_eq!(
2437            error,
2438            ReplicationTrackerError::CapacityExceeded {
2439                current: 2,
2440                needed: 1,
2441                max: 2,
2442            }
2443        );
2444        assert_eq!(tracker.last_sent(client_id, first), Some(Tick::new(2)));
2445        assert_eq!(tracker.last_sent(client_id, second), Some(Tick::new(2)));
2446        assert_eq!(tracker.get(client_id, third), None);
2447    }
2448
2449    #[test]
2450    fn replication_tracker_promotes_without_losing_ack_or_prune_state() {
2451        let first_client = ClientId::new(1);
2452        let second_client = ClientId::new(2);
2453        let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig {
2454            max_entries: HASHED_REPLICATION_TRACKER_MIN_ENTRIES + 1,
2455        });
2456        let initial_entities: Vec<_> = (0..HASHED_REPLICATION_TRACKER_MIN_ENTRIES - 2)
2457            .map(|index| {
2458                EntityHandle::new(u32::try_from(index).expect("test entity index fits u32"), 1)
2459            })
2460            .collect();
2461        tracker
2462            .record_plan_sent(
2463                first_client,
2464                &ReplicationPlan {
2465                    entities: initial_entities.clone(),
2466                    stats: ReplicationStats::default(),
2467                },
2468                Tick::new(1),
2469            )
2470            .expect("initial records should fit");
2471        let second_entity = EntityHandle::new(u32::MAX, 1);
2472        tracker
2473            .record_plan_sent(
2474                second_client,
2475                &ReplicationPlan {
2476                    entities: vec![second_entity],
2477                    stats: ReplicationStats::default(),
2478                },
2479                Tick::new(1),
2480            )
2481            .expect("second client record should fit");
2482        assert!(!tracker.records.is_hashed());
2483
2484        let first_entity = initial_entities[0];
2485        tracker
2486            .record_plan_sent(
2487                first_client,
2488                &ReplicationPlan {
2489                    entities: vec![first_entity],
2490                    stats: ReplicationStats::default(),
2491                },
2492                Tick::new(2),
2493            )
2494            .expect("existing record should update without promotion");
2495        assert!(!tracker.records.is_hashed());
2496
2497        let final_entity = EntityHandle::new(
2498            u32::try_from(HASHED_REPLICATION_TRACKER_MIN_ENTRIES)
2499                .expect("test entity index fits u32"),
2500            1,
2501        );
2502        tracker
2503            .record_plan_sent(
2504                first_client,
2505                &ReplicationPlan {
2506                    entities: vec![final_entity],
2507                    stats: ReplicationStats::default(),
2508                },
2509                Tick::new(2),
2510            )
2511            .expect("threshold record should promote");
2512        assert!(tracker.records.is_hashed());
2513        assert!(tracker.acknowledge(first_client, final_entity, Tick::new(3)));
2514        assert_eq!(tracker.clear_client(second_client), 1);
2515        assert!(tracker.records.is_hashed());
2516        assert_eq!(
2517            tracker.prune_sent_before(Tick::new(2)),
2518            initial_entities.len() - 1
2519        );
2520        assert!(tracker.records.is_hashed());
2521        assert_eq!(tracker.len(), 2);
2522        assert_eq!(tracker.stats().entries, 2);
2523        assert_eq!(tracker.stats().acked_records, 1);
2524        assert_eq!(
2525            tracker
2526                .get(first_client, final_entity)
2527                .expect("acked record should survive")
2528                .last_acked,
2529            Some(Tick::new(3))
2530        );
2531    }
2532}