1use 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#[derive(Clone, Copy, Debug, PartialEq, Eq)]
105pub struct ReplicationBudget {
106 pub max_entities: usize,
108 pub max_bytes: usize,
110 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#[derive(Clone, Debug, Default, PartialEq, Eq)]
126pub struct ReplicationPlan {
127 pub entities: Vec<EntityHandle>,
129 pub stats: ReplicationStats,
131}
132
133#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
135pub struct ReplicationBatchStats {
136 pub viewers: usize,
138 pub candidates: usize,
140 pub considered: usize,
142 pub selected: usize,
144 pub unexamined_after_budget: usize,
146 pub estimated_bytes: usize,
148 pub grid_queries: usize,
150 pub occupied_queries: usize,
152 pub grid_cells_probed: usize,
154 pub occupied_cells_scanned: usize,
156 pub matched_cells: usize,
158 pub candidate_capacity_max: usize,
160 pub dedup_capacity_max: usize,
162 pub matching_cell_capacity_max: usize,
164 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 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#[derive(Clone, Debug, Default, PartialEq, Eq)]
237pub struct ReplicationBatchResult {
238 pub plans: Vec<ReplicationPlan>,
240 pub stats: ReplicationBatchStats,
242}
243
244#[derive(Clone, Copy, Debug, PartialEq, Eq)]
246pub struct ReplicationBatchView<'a> {
247 pub plans: &'a [ReplicationPlan],
249 pub stats: ReplicationBatchStats,
251}
252
253#[derive(Clone, Debug, Default)]
259pub struct ReplicationBatchScratch {
260 plans: Vec<ReplicationPlan>,
261 active_plans: usize,
262 stats: ReplicationBatchStats,
263}
264
265impl ReplicationBatchScratch {
266 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 pub fn retained_plan_slots(&self) -> usize {
293 self.plans.len()
294 }
295
296 pub fn retained_entity_capacity(&self) -> usize {
298 self.plans.iter().map(|plan| plan.entities.capacity()).sum()
299 }
300
301 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#[derive(Clone, Copy, Debug, PartialEq, Eq)]
320pub struct ReplicationTrackerConfig {
321 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#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
335pub struct ReplicationTrackKey {
336 pub client_id: ClientId,
338 pub entity: EntityHandle,
340}
341
342#[derive(Clone, Copy, Debug, PartialEq, Eq)]
344pub struct ReplicationTrackRecord {
345 pub client_id: ClientId,
347 pub entity: EntityHandle,
349 pub last_sent: Tick,
351 pub last_acked: Option<Tick>,
353}
354
355#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
357pub struct ReplicationTrackerStats {
358 pub entries: usize,
360 pub sent_records: usize,
362 pub acked_records: usize,
364 pub pruned_records: usize,
366}
367
368#[derive(Clone, Copy, Debug, PartialEq, Eq)]
370pub enum ReplicationTrackerError {
371 CapacityExceeded {
373 current: usize,
375 needed: usize,
377 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#[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 pub fn new(config: ReplicationTrackerConfig) -> Self {
416 Self {
417 config,
418 records: AdaptiveTrackMap::new(),
419 stats: ReplicationTrackerStats::default(),
420 }
421 }
422
423 pub const fn config(&self) -> ReplicationTrackerConfig {
425 self.config
426 }
427
428 pub const fn stats(&self) -> ReplicationTrackerStats {
430 self.stats
431 }
432
433 pub fn len(&self) -> usize {
435 self.records.len()
436 }
437
438 pub fn is_empty(&self) -> bool {
440 self.records.is_empty()
441 }
442
443 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 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 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 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 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 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 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#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
573pub struct ReplicationStats {
574 pub candidates: usize,
576 pub considered: usize,
578 pub selected: usize,
580 pub skipped_by_budget: usize,
582 pub unexamined_after_budget: usize,
584 pub skipped_by_cadence: usize,
586 pub estimated_bytes: usize,
588}
589
590#[derive(Clone, Copy, Debug, Default)]
592pub struct ReplicationCadence;
593
594impl ReplicationCadence {
595 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 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 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#[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 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#[derive(Clone, Debug, Default)]
681pub struct ReplicationScratch {
682 cell_query: CellQueryScratch,
683 prioritized: Vec<PrioritizedReplicationCandidate>,
684}
685
686impl ReplicationScratch {
687 pub fn clear(&mut self) {
689 self.cell_query.clear();
690 self.prioritized.clear();
691 }
692
693 pub fn candidate_count(&self) -> usize {
695 self.cell_query.len()
696 }
697
698 pub fn prioritized_capacity(&self) -> usize {
700 self.prioritized.capacity()
701 }
702
703 pub const fn query_stats(&self) -> CellQueryStats {
705 self.cell_query.stats()
706 }
707
708 pub fn candidate_capacity(&self) -> usize {
710 self.cell_query.handle_capacity()
711 }
712
713 pub fn candidate_dedup_capacity(&self) -> usize {
715 self.cell_query.dedup_capacity()
716 }
717
718 pub fn matching_cell_capacity(&self) -> usize {
720 self.cell_query.matching_cell_capacity()
721 }
722}
723
724#[derive(Clone, Copy, Debug, Default)]
726pub struct ReplicationPlanner;
727
728impl ReplicationPlanner {
729 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 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 #[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 #[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 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 #[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 #[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 #[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 #[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 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 #[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 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 #[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 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 #[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 #[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 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 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 #[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 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 #[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 #[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}