sectorsync_core/
replication.rs

1//! Replication planning helpers.
2
3use std::collections::BTreeMap;
4
5use crate::ids::{ClientId, EntityHandle, Tick};
6use crate::interest::{ViewerQuery, VisibilityFilter};
7use crate::policy::{CompiledSyncPolicy, PolicyTable};
8use crate::spatial_index::{CellIndex, CellQueryScratch, CellQueryStats};
9use crate::station::Station;
10
11/// Per-client replication budget.
12#[derive(Clone, Copy, Debug, PartialEq, Eq)]
13pub struct ReplicationBudget {
14    /// Maximum entities to include in a frame.
15    pub max_entities: usize,
16    /// Estimated byte budget for a frame.
17    pub max_bytes: usize,
18    /// Estimated bytes charged per selected entity by simple planners.
19    pub estimated_entity_bytes: usize,
20}
21
22impl Default for ReplicationBudget {
23    fn default() -> Self {
24        Self {
25            max_entities: 300,
26            max_bytes: 16 * 1024,
27            estimated_entity_bytes: 32,
28        }
29    }
30}
31
32/// Replication planner result.
33#[derive(Clone, Debug, Default, PartialEq, Eq)]
34pub struct ReplicationPlan {
35    /// Selected entity handles.
36    pub entities: Vec<EntityHandle>,
37    /// Planner statistics.
38    pub stats: ReplicationStats,
39}
40
41/// Bounded per-client replication tracking configuration.
42#[derive(Clone, Copy, Debug, PartialEq, Eq)]
43pub struct ReplicationTrackerConfig {
44    /// Maximum tracked client/entity entries.
45    pub max_entries: usize,
46}
47
48impl Default for ReplicationTrackerConfig {
49    fn default() -> Self {
50        Self {
51            max_entries: 65_536,
52        }
53    }
54}
55
56/// Per-client/entity replication tracking key.
57#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
58pub struct ReplicationTrackKey {
59    /// Client that received the entity update.
60    pub client_id: ClientId,
61    /// Station-local entity handle selected by the planner.
62    pub entity: EntityHandle,
63}
64
65/// Per-client/entity replication tracking record.
66#[derive(Clone, Copy, Debug, PartialEq, Eq)]
67pub struct ReplicationTrackRecord {
68    /// Client that received the entity update.
69    pub client_id: ClientId,
70    /// Station-local entity handle selected by the planner.
71    pub entity: EntityHandle,
72    /// Last tick where this entity was sent to the client.
73    pub last_sent: Tick,
74    /// Last tick where the caller confirmed delivery.
75    pub last_acked: Option<Tick>,
76}
77
78/// Replication tracker statistics.
79#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
80pub struct ReplicationTrackerStats {
81    /// Currently tracked entries.
82    pub entries: usize,
83    /// Total record insertions or updates.
84    pub sent_records: usize,
85    /// Total ACK updates applied.
86    pub acked_records: usize,
87    /// Records pruned by explicit cleanup.
88    pub pruned_records: usize,
89}
90
91/// Replication tracking error.
92#[derive(Clone, Copy, Debug, PartialEq, Eq)]
93pub enum ReplicationTrackerError {
94    /// Recording would exceed the configured entry capacity.
95    CapacityExceeded {
96        /// Entries currently tracked.
97        current: usize,
98        /// New entries needed for this operation.
99        needed: usize,
100        /// Maximum tracked entries.
101        max: usize,
102    },
103}
104
105impl core::fmt::Display for ReplicationTrackerError {
106    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
107        match self {
108            Self::CapacityExceeded {
109                current,
110                needed,
111                max,
112            } => write!(
113                f,
114                "replication tracker capacity exceeded: current {current}, needed {needed}, max {max}"
115            ),
116        }
117    }
118}
119
120impl std::error::Error for ReplicationTrackerError {}
121
122/// Bounded per-client replication send/ACK tracker.
123#[derive(Clone, Debug)]
124pub struct ReplicationTracker {
125    config: ReplicationTrackerConfig,
126    records: BTreeMap<ReplicationTrackKey, ReplicationTrackRecord>,
127    stats: ReplicationTrackerStats,
128}
129
130impl Default for ReplicationTracker {
131    fn default() -> Self {
132        Self::new(ReplicationTrackerConfig::default())
133    }
134}
135
136impl ReplicationTracker {
137    /// Creates an empty tracker.
138    pub fn new(config: ReplicationTrackerConfig) -> Self {
139        Self {
140            config,
141            records: BTreeMap::new(),
142            stats: ReplicationTrackerStats::default(),
143        }
144    }
145
146    /// Returns tracker configuration.
147    pub const fn config(&self) -> ReplicationTrackerConfig {
148        self.config
149    }
150
151    /// Returns tracker statistics.
152    pub const fn stats(&self) -> ReplicationTrackerStats {
153        self.stats
154    }
155
156    /// Returns tracked entry count.
157    pub fn len(&self) -> usize {
158        self.records.len()
159    }
160
161    /// Returns whether no entries are tracked.
162    pub fn is_empty(&self) -> bool {
163        self.records.is_empty()
164    }
165
166    /// Returns the last sent tick for a client/entity pair.
167    pub fn last_sent(&self, client_id: ClientId, entity: EntityHandle) -> Option<Tick> {
168        self.records
169            .get(&ReplicationTrackKey { client_id, entity })
170            .map(|record| record.last_sent)
171    }
172
173    /// Returns a tracked record for a client/entity pair.
174    pub fn get(&self, client_id: ClientId, entity: EntityHandle) -> Option<ReplicationTrackRecord> {
175        self.records
176            .get(&ReplicationTrackKey { client_id, entity })
177            .copied()
178    }
179
180    /// Records that a planned set of entities was sent to a client.
181    pub fn record_plan_sent(
182        &mut self,
183        client_id: ClientId,
184        plan: &ReplicationPlan,
185        sent_at: Tick,
186    ) -> Result<usize, ReplicationTrackerError> {
187        self.ensure_capacity_for(client_id, &plan.entities)?;
188        let mut recorded = 0;
189        for entity in &plan.entities {
190            let key = ReplicationTrackKey {
191                client_id,
192                entity: *entity,
193            };
194            self.records.insert(
195                key,
196                ReplicationTrackRecord {
197                    client_id,
198                    entity: *entity,
199                    last_sent: sent_at,
200                    last_acked: None,
201                },
202            );
203            recorded += 1;
204        }
205        self.refresh_entry_count();
206        self.stats.sent_records = self.stats.sent_records.saturating_add(recorded);
207        Ok(recorded)
208    }
209
210    /// Records delivery acknowledgement for one client/entity pair.
211    pub fn acknowledge(
212        &mut self,
213        client_id: ClientId,
214        entity: EntityHandle,
215        acked_at: Tick,
216    ) -> bool {
217        let Some(record) = self
218            .records
219            .get_mut(&ReplicationTrackKey { client_id, entity })
220        else {
221            return false;
222        };
223        record.last_acked = Some(acked_at);
224        self.stats.acked_records = self.stats.acked_records.saturating_add(1);
225        true
226    }
227
228    /// Records delivery acknowledgement for every entity in a plan.
229    pub fn acknowledge_plan(
230        &mut self,
231        client_id: ClientId,
232        plan: &ReplicationPlan,
233        acked_at: Tick,
234    ) -> usize {
235        plan.entities
236            .iter()
237            .filter(|entity| self.acknowledge(client_id, **entity, acked_at))
238            .count()
239    }
240
241    /// Removes all entries for one client.
242    pub fn clear_client(&mut self, client_id: ClientId) -> usize {
243        let before = self.records.len();
244        self.records.retain(|key, _| key.client_id != client_id);
245        let pruned = before.saturating_sub(self.records.len());
246        self.stats.pruned_records = self.stats.pruned_records.saturating_add(pruned);
247        self.refresh_entry_count();
248        pruned
249    }
250
251    /// Removes entries last sent before `older_than`.
252    pub fn prune_sent_before(&mut self, older_than: Tick) -> usize {
253        let before = self.records.len();
254        self.records
255            .retain(|_, record| record.last_sent.get() >= older_than.get());
256        let pruned = before.saturating_sub(self.records.len());
257        self.stats.pruned_records = self.stats.pruned_records.saturating_add(pruned);
258        self.refresh_entry_count();
259        pruned
260    }
261
262    fn ensure_capacity_for(
263        &self,
264        client_id: ClientId,
265        entities: &[EntityHandle],
266    ) -> Result<(), ReplicationTrackerError> {
267        let mut needed = 0_usize;
268        for entity in entities {
269            if !self.records.contains_key(&ReplicationTrackKey {
270                client_id,
271                entity: *entity,
272            }) {
273                needed = needed.saturating_add(1);
274            }
275        }
276        if self.records.len().saturating_add(needed) > self.config.max_entries {
277            return Err(ReplicationTrackerError::CapacityExceeded {
278                current: self.records.len(),
279                needed,
280                max: self.config.max_entries,
281            });
282        }
283        Ok(())
284    }
285
286    fn refresh_entry_count(&mut self) {
287        self.stats.entries = self.records.len();
288    }
289}
290
291/// Replication planner statistics.
292#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
293pub struct ReplicationStats {
294    /// Candidate handles returned from the spatial index.
295    pub candidates: usize,
296    /// Candidate records considered after stale handle filtering.
297    pub considered: usize,
298    /// Selected entities.
299    pub selected: usize,
300    /// Entities skipped because the budget was exhausted.
301    pub skipped_by_budget: usize,
302    /// Entities skipped because their cadence interval has not elapsed.
303    pub skipped_by_cadence: usize,
304    /// Estimated frame bytes.
305    pub estimated_bytes: usize,
306}
307
308/// Stateless distance-based replication cadence helper.
309#[derive(Clone, Copy, Debug, Default)]
310pub struct ReplicationCadence;
311
312impl ReplicationCadence {
313    /// Returns the target update frequency for a policy at a squared distance.
314    pub fn target_hz(policy: &CompiledSyncPolicy, distance_squared: f32) -> u16 {
315        let min_hz = policy.min_hz.max(1);
316        let max_hz = policy.max_hz.max(min_hz);
317        let radius_squared = policy.interest_radius * policy.interest_radius;
318        let closeness =
319            if radius_squared.is_finite() && radius_squared > 0.0 && distance_squared.is_finite() {
320                1.0 - (distance_squared / radius_squared).clamp(0.0, 1.0)
321            } else {
322                1.0
323            };
324        let span = f32::from(max_hz - min_hz);
325        let target = f32::from(min_hz) + span * closeness;
326        rounded_frequency_to_u16(target, min_hz, max_hz)
327    }
328
329    /// Returns the tick interval for a policy at a squared distance.
330    pub fn interval_ticks(
331        policy: &CompiledSyncPolicy,
332        station_tick_rate_hz: u16,
333        distance_squared: f32,
334    ) -> u64 {
335        let tick_rate = u64::from(station_tick_rate_hz.max(1));
336        let target_hz = u64::from(Self::target_hz(policy, distance_squared).max(1));
337        tick_rate.div_ceil(target_hz).max(1)
338    }
339
340    /// Returns whether a replication update should be sent at `now`.
341    pub fn should_send(
342        policy: &CompiledSyncPolicy,
343        station_tick_rate_hz: u16,
344        distance_squared: f32,
345        now: Tick,
346        last_sent: Option<Tick>,
347    ) -> bool {
348        let Some(last_sent) = last_sent else {
349            return true;
350        };
351        let interval = Self::interval_ticks(policy, station_tick_rate_hz, distance_squared);
352        now.get().saturating_sub(last_sent.get()) >= interval
353    }
354}
355
356/// Stateless replication priority scoring helper.
357#[derive(Clone, Copy, Debug, Default)]
358pub struct ReplicationPriority;
359
360#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
361fn rounded_frequency_to_u16(target: f32, min_hz: u16, max_hz: u16) -> u16 {
362    let bounded = target.round().clamp(f32::from(min_hz), f32::from(max_hz));
363    bounded as u16
364}
365
366#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
367fn normalized_score_to_u64(closeness: f32) -> u64 {
368    debug_assert!(closeness.is_finite() && (0.0..=1.0).contains(&closeness));
369    (closeness * 1_000_000.0).round() as u64
370}
371
372impl ReplicationPriority {
373    /// Returns a deterministic priority score for budgeted selection.
374    pub fn score(policy: &CompiledSyncPolicy, distance_squared: f32) -> u64 {
375        let weight = u64::from(policy.priority_weight.max(1));
376        let radius_squared = policy.interest_radius * policy.interest_radius;
377        let distance_score =
378            if radius_squared.is_finite() && radius_squared > 0.0 && distance_squared.is_finite() {
379                let closeness = 1.0 - (distance_squared / radius_squared).clamp(0.0, 1.0);
380                normalized_score_to_u64(closeness)
381            } else {
382                1_000_000
383            };
384        weight
385            .saturating_mul(1_000_000)
386            .saturating_add(distance_score)
387    }
388}
389
390#[derive(Clone, Copy, Debug, PartialEq)]
391struct PrioritizedReplicationCandidate {
392    handle: EntityHandle,
393    score: u64,
394    distance_squared: f32,
395}
396
397/// Reusable scratch storage for allocation-aware replication planning.
398#[derive(Clone, Debug, Default)]
399pub struct ReplicationScratch {
400    cell_query: CellQueryScratch,
401    prioritized: Vec<PrioritizedReplicationCandidate>,
402}
403
404impl ReplicationScratch {
405    /// Clears retained planning results while keeping allocated capacity.
406    pub fn clear(&mut self) {
407        self.cell_query.clear();
408        self.prioritized.clear();
409    }
410
411    /// Number of spatial candidates retained from the last query.
412    pub fn candidate_count(&self) -> usize {
413        self.cell_query.len()
414    }
415
416    /// Capacity retained for priority candidate sorting.
417    pub fn prioritized_capacity(&self) -> usize {
418        self.prioritized.capacity()
419    }
420
421    /// Work counters from the last spatial candidate query.
422    pub const fn query_stats(&self) -> CellQueryStats {
423        self.cell_query.stats()
424    }
425
426    /// Capacity retained for spatial candidate handles.
427    pub fn candidate_capacity(&self) -> usize {
428        self.cell_query.handle_capacity()
429    }
430
431    /// Capacity retained by spatial candidate deduplication.
432    pub fn candidate_dedup_capacity(&self) -> usize {
433        self.cell_query.dedup_capacity()
434    }
435
436    /// Capacity retained for cells matched by sparse spatial queries.
437    pub fn matching_cell_capacity(&self) -> usize {
438        self.cell_query.matching_cell_capacity()
439    }
440}
441
442/// Simple range/visibility-based replication planner.
443#[derive(Clone, Copy, Debug, Default)]
444pub struct ReplicationPlanner;
445
446impl ReplicationPlanner {
447    /// Plans a frame for one viewer using the station-local spatial index.
448    pub fn plan_for_viewer<F: VisibilityFilter>(
449        station: &Station,
450        index: &CellIndex,
451        policies: &PolicyTable,
452        viewer: &ViewerQuery,
453        filter: &F,
454        budget: ReplicationBudget,
455    ) -> ReplicationPlan {
456        let candidates = index.query_sphere(viewer.position, viewer.radius);
457        Self::plan_for_candidates_inner(
458            station,
459            &candidates,
460            policies,
461            viewer,
462            filter,
463            budget,
464            |_, _, _| true,
465        )
466    }
467
468    /// Plans a frame using caller-provided scratch storage.
469    pub fn plan_for_viewer_with_scratch<F: VisibilityFilter>(
470        station: &Station,
471        index: &CellIndex,
472        policies: &PolicyTable,
473        viewer: &ViewerQuery,
474        filter: &F,
475        budget: ReplicationBudget,
476        scratch: &mut ReplicationScratch,
477    ) -> ReplicationPlan {
478        let candidates =
479            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
480        Self::plan_for_candidates_inner(
481            station,
482            candidates,
483            policies,
484            viewer,
485            filter,
486            budget,
487            |_, _, _| true,
488        )
489    }
490
491    /// Plans a frame and skips entities whose distance-based cadence has not elapsed.
492    pub fn plan_for_viewer_with_cadence<F, L>(
493        station: &Station,
494        index: &CellIndex,
495        policies: &PolicyTable,
496        viewer: &ViewerQuery,
497        filter: &F,
498        budget: ReplicationBudget,
499        last_sent: L,
500    ) -> ReplicationPlan
501    where
502        F: VisibilityFilter,
503        L: Fn(EntityHandle) -> Option<Tick>,
504    {
505        let tick_rate_hz = station.config().tick_rate_hz;
506        let now = station.tick();
507        let candidates = index.query_sphere(viewer.position, viewer.radius);
508        Self::plan_for_candidates_inner(
509            station,
510            &candidates,
511            policies,
512            viewer,
513            filter,
514            budget,
515            |handle, policy, distance_squared| {
516                ReplicationCadence::should_send(
517                    policy,
518                    tick_rate_hz,
519                    distance_squared,
520                    now,
521                    last_sent(handle),
522                )
523            },
524        )
525    }
526
527    /// Plans a cadence-aware frame using caller-provided scratch storage.
528    #[allow(clippy::too_many_arguments)]
529    pub fn plan_for_viewer_with_cadence_and_scratch<F, L>(
530        station: &Station,
531        index: &CellIndex,
532        policies: &PolicyTable,
533        viewer: &ViewerQuery,
534        filter: &F,
535        budget: ReplicationBudget,
536        last_sent: L,
537        scratch: &mut ReplicationScratch,
538    ) -> ReplicationPlan
539    where
540        F: VisibilityFilter,
541        L: Fn(EntityHandle) -> Option<Tick>,
542    {
543        let tick_rate_hz = station.config().tick_rate_hz;
544        let now = station.tick();
545        let candidates =
546            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
547        Self::plan_for_candidates_inner(
548            station,
549            candidates,
550            policies,
551            viewer,
552            filter,
553            budget,
554            |handle, policy, distance_squared| {
555                ReplicationCadence::should_send(
556                    policy,
557                    tick_rate_hz,
558                    distance_squared,
559                    now,
560                    last_sent(handle),
561                )
562            },
563        )
564    }
565
566    /// Plans a frame and selects the highest-priority entities when budgeted.
567    pub fn plan_for_viewer_prioritized<F: VisibilityFilter>(
568        station: &Station,
569        index: &CellIndex,
570        policies: &PolicyTable,
571        viewer: &ViewerQuery,
572        filter: &F,
573        budget: ReplicationBudget,
574    ) -> ReplicationPlan {
575        let candidates = index.query_sphere(viewer.position, viewer.radius);
576        let mut prioritized = Vec::new();
577        Self::plan_for_candidates_prioritized_inner(
578            station,
579            &candidates,
580            policies,
581            viewer,
582            filter,
583            budget,
584            &mut prioritized,
585            |_, _, _| true,
586        )
587    }
588
589    /// Plans a budgeted priority frame using caller-provided scratch storage.
590    pub fn plan_for_viewer_prioritized_with_scratch<F: VisibilityFilter>(
591        station: &Station,
592        index: &CellIndex,
593        policies: &PolicyTable,
594        viewer: &ViewerQuery,
595        filter: &F,
596        budget: ReplicationBudget,
597        scratch: &mut ReplicationScratch,
598    ) -> ReplicationPlan {
599        let candidates =
600            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
601        Self::plan_for_candidates_prioritized_inner(
602            station,
603            candidates,
604            policies,
605            viewer,
606            filter,
607            budget,
608            &mut scratch.prioritized,
609            |_, _, _| true,
610        )
611    }
612
613    /// Plans a budgeted priority frame with distance-based cadence checks.
614    pub fn plan_for_viewer_prioritized_with_cadence<F, L>(
615        station: &Station,
616        index: &CellIndex,
617        policies: &PolicyTable,
618        viewer: &ViewerQuery,
619        filter: &F,
620        budget: ReplicationBudget,
621        last_sent: L,
622    ) -> ReplicationPlan
623    where
624        F: VisibilityFilter,
625        L: Fn(EntityHandle) -> Option<Tick>,
626    {
627        let tick_rate_hz = station.config().tick_rate_hz;
628        let now = station.tick();
629        let candidates = index.query_sphere(viewer.position, viewer.radius);
630        let mut prioritized = Vec::new();
631        Self::plan_for_candidates_prioritized_inner(
632            station,
633            &candidates,
634            policies,
635            viewer,
636            filter,
637            budget,
638            &mut prioritized,
639            |handle, policy, distance_squared| {
640                ReplicationCadence::should_send(
641                    policy,
642                    tick_rate_hz,
643                    distance_squared,
644                    now,
645                    last_sent(handle),
646                )
647            },
648        )
649    }
650
651    /// Plans a priority/cadence frame using caller-provided scratch storage.
652    #[allow(clippy::too_many_arguments)]
653    pub fn plan_for_viewer_prioritized_with_cadence_and_scratch<F, L>(
654        station: &Station,
655        index: &CellIndex,
656        policies: &PolicyTable,
657        viewer: &ViewerQuery,
658        filter: &F,
659        budget: ReplicationBudget,
660        last_sent: L,
661        scratch: &mut ReplicationScratch,
662    ) -> ReplicationPlan
663    where
664        F: VisibilityFilter,
665        L: Fn(EntityHandle) -> Option<Tick>,
666    {
667        let tick_rate_hz = station.config().tick_rate_hz;
668        let now = station.tick();
669        let candidates =
670            index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
671        Self::plan_for_candidates_prioritized_inner(
672            station,
673            candidates,
674            policies,
675            viewer,
676            filter,
677            budget,
678            &mut scratch.prioritized,
679            |handle, policy, distance_squared| {
680                ReplicationCadence::should_send(
681                    policy,
682                    tick_rate_hz,
683                    distance_squared,
684                    now,
685                    last_sent(handle),
686                )
687            },
688        )
689    }
690
691    fn plan_for_candidates_inner<F, C>(
692        station: &Station,
693        candidates: &[EntityHandle],
694        policies: &PolicyTable,
695        viewer: &ViewerQuery,
696        filter: &F,
697        budget: ReplicationBudget,
698        cadence_allows: C,
699    ) -> ReplicationPlan
700    where
701        F: VisibilityFilter,
702        C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
703    {
704        let max_entities = viewer.max_entities.min(budget.max_entities);
705        let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
706        let hard_limit = max_entities.min(max_by_bytes);
707
708        let mut plan = ReplicationPlan {
709            entities: Vec::with_capacity(hard_limit),
710            stats: ReplicationStats {
711                candidates: candidates.len(),
712                ..ReplicationStats::default()
713            },
714        };
715
716        for handle in candidates {
717            let Some(entity) = station.get(*handle) else {
718                continue;
719            };
720            plan.stats.considered += 1;
721
722            let Some(policy) = policies.get(entity.policy_id) else {
723                continue;
724            };
725            let distance_squared = entity.position.distance_squared(viewer.position);
726            let policy_radius_sq = policy.interest_radius * policy.interest_radius;
727            if distance_squared > policy_radius_sq {
728                continue;
729            }
730            if !filter.is_visible(viewer, entity) {
731                continue;
732            }
733            if !cadence_allows(*handle, policy, distance_squared) {
734                plan.stats.skipped_by_cadence += 1;
735                continue;
736            }
737
738            if plan.entities.len() >= hard_limit {
739                plan.stats.skipped_by_budget += 1;
740                continue;
741            }
742
743            plan.entities.push(*handle);
744        }
745
746        plan.stats.selected = plan.entities.len();
747        plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
748        plan
749    }
750
751    #[allow(clippy::too_many_arguments)]
752    fn plan_for_candidates_prioritized_inner<F, C>(
753        station: &Station,
754        candidates: &[EntityHandle],
755        policies: &PolicyTable,
756        viewer: &ViewerQuery,
757        filter: &F,
758        budget: ReplicationBudget,
759        eligible: &mut Vec<PrioritizedReplicationCandidate>,
760        cadence_allows: C,
761    ) -> ReplicationPlan
762    where
763        F: VisibilityFilter,
764        C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
765    {
766        let max_entities = viewer.max_entities.min(budget.max_entities);
767        let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
768        let hard_limit = max_entities.min(max_by_bytes);
769        let mut plan = ReplicationPlan {
770            entities: Vec::with_capacity(hard_limit),
771            stats: ReplicationStats {
772                candidates: candidates.len(),
773                ..ReplicationStats::default()
774            },
775        };
776        eligible.clear();
777
778        for handle in candidates {
779            let Some(entity) = station.get(*handle) else {
780                continue;
781            };
782            plan.stats.considered += 1;
783
784            let Some(policy) = policies.get(entity.policy_id) else {
785                continue;
786            };
787            let distance_squared = entity.position.distance_squared(viewer.position);
788            let policy_radius_sq = policy.interest_radius * policy.interest_radius;
789            if distance_squared > policy_radius_sq {
790                continue;
791            }
792            if !filter.is_visible(viewer, entity) {
793                continue;
794            }
795            if !cadence_allows(*handle, policy, distance_squared) {
796                plan.stats.skipped_by_cadence += 1;
797                continue;
798            }
799
800            eligible.push(PrioritizedReplicationCandidate {
801                handle: *handle,
802                score: ReplicationPriority::score(policy, distance_squared),
803                distance_squared,
804            });
805        }
806
807        eligible.sort_by(|left, right| {
808            right
809                .score
810                .cmp(&left.score)
811                .then_with(|| left.distance_squared.total_cmp(&right.distance_squared))
812                .then_with(|| left.handle.cmp(&right.handle))
813        });
814
815        plan.stats.skipped_by_budget = eligible.len().saturating_sub(hard_limit);
816        plan.entities.extend(
817            eligible
818                .iter()
819                .take(hard_limit)
820                .map(|candidate| candidate.handle),
821        );
822        plan.stats.selected = plan.entities.len();
823        plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
824        plan
825    }
826}
827
828#[cfg(test)]
829mod tests {
830    use super::*;
831    use crate::entity::EntityTags;
832    use crate::ids::{ClientId, EntityId, InstanceId, NodeId, PolicyId, StationId};
833    use crate::interest::{AndVisibility, FrustumVisibility, RangeOnlyVisibility, TagVisibility};
834    use crate::policy::CompiledSyncPolicy;
835    use crate::spatial::{Aabb3, Bounds, Frustum3, GridSpec, Position3};
836    use crate::station::{Station, StationConfig};
837
838    #[test]
839    fn planner_applies_composed_frustum_visibility_filter() {
840        let mut station = Station::new(StationConfig {
841            station_id: StationId::new(1),
842            node_id: NodeId::new(1),
843            instance_id: InstanceId::new(1),
844            tick_rate_hz: 20,
845        });
846        let grid = GridSpec::new(16.0).expect("grid is valid");
847        let mut index = CellIndex::new(grid);
848        let mut policies = PolicyTable::default();
849        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
850
851        let visible = station
852            .spawn_owned(
853                EntityId::new(1),
854                Position3::new(10.0, 0.0, 0.0),
855                Bounds::Point,
856                PolicyId::new(1),
857            )
858            .expect("spawn visible");
859        let outside_frustum = station
860            .spawn_owned(
861                EntityId::new(2),
862                Position3::new(-10.0, 0.0, 0.0),
863                Bounds::Point,
864                PolicyId::new(1),
865            )
866            .expect("spawn outside frustum");
867        index.upsert(visible, Position3::new(10.0, 0.0, 0.0), Bounds::Point);
868        index.upsert(
869            outside_frustum,
870            Position3::new(-10.0, 0.0, 0.0),
871            Bounds::Point,
872        );
873
874        let viewer = ViewerQuery {
875            client_id: ClientId::new(7),
876            position: Position3::new(0.0, 0.0, 0.0),
877            radius: 128.0,
878            max_entities: 8,
879        };
880        let frustum = Frustum3::from_aabb(Aabb3::new(
881            Position3::new(0.0, -20.0, -20.0),
882            Position3::new(80.0, 20.0, 20.0),
883        ));
884        let filter = AndVisibility::new(RangeOnlyVisibility, FrustumVisibility::new(frustum));
885
886        let plan = ReplicationPlanner::plan_for_viewer(
887            &station,
888            &index,
889            &policies,
890            &viewer,
891            &filter,
892            ReplicationBudget::default(),
893        );
894
895        assert_eq!(plan.entities, vec![visible]);
896        assert_eq!(plan.stats.selected, 1);
897        assert_eq!(plan.stats.considered, 2);
898    }
899
900    #[test]
901    fn planner_applies_tag_visibility_filter() {
902        let mut station = Station::new(StationConfig {
903            station_id: StationId::new(1),
904            node_id: NodeId::new(1),
905            instance_id: InstanceId::new(1),
906            tick_rate_hz: 20,
907        });
908        let grid = GridSpec::new(16.0).expect("grid is valid");
909        let mut index = CellIndex::new(grid);
910        let mut policies = PolicyTable::default();
911        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
912
913        let static_visible = station
914            .spawn_owned(
915                EntityId::new(1),
916                Position3::new(10.0, 0.0, 0.0),
917                Bounds::Point,
918                PolicyId::new(1),
919            )
920            .expect("spawn static");
921        let fast_mover = station
922            .spawn_owned(
923                EntityId::new(2),
924                Position3::new(12.0, 0.0, 0.0),
925                Bounds::Point,
926                PolicyId::new(1),
927            )
928            .expect("spawn mover");
929        station
930            .set_tags(static_visible, EntityTags::from_bits(0b001))
931            .expect("tag static");
932        station
933            .set_tags(fast_mover, EntityTags::from_bits(0b010))
934            .expect("tag mover");
935        index.upsert(
936            static_visible,
937            Position3::new(10.0, 0.0, 0.0),
938            Bounds::Point,
939        );
940        index.upsert(fast_mover, Position3::new(12.0, 0.0, 0.0), Bounds::Point);
941
942        let viewer = ViewerQuery {
943            client_id: ClientId::new(7),
944            position: Position3::new(0.0, 0.0, 0.0),
945            radius: 128.0,
946            max_entities: 8,
947        };
948        let filter = AndVisibility::new(
949            RangeOnlyVisibility,
950            TagVisibility::new(EntityTags::from_bits(0b001), EntityTags::from_bits(0b010)),
951        );
952
953        let plan = ReplicationPlanner::plan_for_viewer(
954            &station,
955            &index,
956            &policies,
957            &viewer,
958            &filter,
959            ReplicationBudget::default(),
960        );
961
962        assert_eq!(plan.entities, vec![static_visible]);
963        assert_eq!(plan.stats.selected, 1);
964        assert_eq!(plan.stats.considered, 2);
965    }
966
967    #[test]
968    fn cadence_scales_interval_by_squared_distance() {
969        let policy = CompiledSyncPolicy::new(PolicyId::new(1), 2, 20, 100.0);
970
971        assert_eq!(ReplicationCadence::target_hz(&policy, 0.0), 20);
972        assert_eq!(ReplicationCadence::interval_ticks(&policy, 20, 0.0), 1);
973        assert_eq!(ReplicationCadence::target_hz(&policy, 100.0_f32 * 100.0), 2);
974        assert_eq!(
975            ReplicationCadence::interval_ticks(&policy, 20, 100.0_f32 * 100.0),
976            10
977        );
978    }
979
980    #[test]
981    fn priority_score_prefers_weight_then_distance() {
982        let mut low = CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 100.0);
983        low.priority_weight = 1;
984        let mut high = CompiledSyncPolicy::new(PolicyId::new(2), 1, 20, 100.0);
985        high.priority_weight = 10;
986
987        assert!(
988            ReplicationPriority::score(&high, 90.0 * 90.0) > ReplicationPriority::score(&low, 0.0)
989        );
990        assert!(
991            ReplicationPriority::score(&low, 0.0) > ReplicationPriority::score(&low, 90.0 * 90.0)
992        );
993    }
994
995    #[test]
996    fn planner_with_cadence_skips_recent_far_entities() {
997        let mut station = Station::new(StationConfig {
998            station_id: StationId::new(1),
999            node_id: NodeId::new(1),
1000            instance_id: InstanceId::new(1),
1001            tick_rate_hz: 20,
1002        });
1003        for _ in 0..10 {
1004            station.advance_tick();
1005        }
1006        let grid = GridSpec::new(16.0).expect("grid is valid");
1007        let mut index = CellIndex::new(grid);
1008        let mut policies = PolicyTable::default();
1009        policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 2, 20, 128.0));
1010
1011        let near = station
1012            .spawn_owned(
1013                EntityId::new(1),
1014                Position3::new(0.0, 0.0, 0.0),
1015                Bounds::Point,
1016                PolicyId::new(1),
1017            )
1018            .expect("spawn near");
1019        let far = station
1020            .spawn_owned(
1021                EntityId::new(2),
1022                Position3::new(120.0, 0.0, 0.0),
1023                Bounds::Point,
1024                PolicyId::new(1),
1025            )
1026            .expect("spawn far");
1027        index.upsert(near, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
1028        index.upsert(far, Position3::new(120.0, 0.0, 0.0), Bounds::Point);
1029
1030        let viewer = ViewerQuery {
1031            client_id: ClientId::new(7),
1032            position: Position3::new(0.0, 0.0, 0.0),
1033            radius: 128.0,
1034            max_entities: 8,
1035        };
1036        let plan = ReplicationPlanner::plan_for_viewer_with_cadence(
1037            &station,
1038            &index,
1039            &policies,
1040            &viewer,
1041            &RangeOnlyVisibility,
1042            ReplicationBudget::default(),
1043            |_| Some(Tick::new(9)),
1044        );
1045
1046        assert_eq!(plan.entities, vec![near]);
1047        assert_eq!(plan.stats.selected, 1);
1048        assert_eq!(plan.stats.skipped_by_cadence, 1);
1049    }
1050
1051    #[test]
1052    fn prioritized_planner_uses_policy_weight_under_budget() {
1053        let mut station = Station::new(StationConfig {
1054            station_id: StationId::new(1),
1055            node_id: NodeId::new(1),
1056            instance_id: InstanceId::new(1),
1057            tick_rate_hz: 20,
1058        });
1059        let grid = GridSpec::new(16.0).expect("grid is valid");
1060        let mut index = CellIndex::new(grid);
1061        let mut policies = PolicyTable::default();
1062        let mut low = CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0);
1063        low.priority_weight = 1;
1064        let mut high = CompiledSyncPolicy::new(PolicyId::new(2), 1, 20, 128.0);
1065        high.priority_weight = 10;
1066        policies.set(low);
1067        policies.set(high);
1068
1069        let near_low = station
1070            .spawn_owned(
1071                EntityId::new(1),
1072                Position3::new(0.0, 0.0, 0.0),
1073                Bounds::Point,
1074                PolicyId::new(1),
1075            )
1076            .expect("spawn near low priority");
1077        let far_high = station
1078            .spawn_owned(
1079                EntityId::new(2),
1080                Position3::new(96.0, 0.0, 0.0),
1081                Bounds::Point,
1082                PolicyId::new(2),
1083            )
1084            .expect("spawn far high priority");
1085        index.upsert(near_low, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
1086        index.upsert(far_high, Position3::new(96.0, 0.0, 0.0), Bounds::Point);
1087
1088        let viewer = ViewerQuery {
1089            client_id: ClientId::new(7),
1090            position: Position3::new(0.0, 0.0, 0.0),
1091            radius: 128.0,
1092            max_entities: 1,
1093        };
1094        let plan = ReplicationPlanner::plan_for_viewer_prioritized(
1095            &station,
1096            &index,
1097            &policies,
1098            &viewer,
1099            &RangeOnlyVisibility,
1100            ReplicationBudget {
1101                max_entities: 1,
1102                max_bytes: 32,
1103                estimated_entity_bytes: 32,
1104            },
1105        );
1106
1107        assert_eq!(plan.entities, vec![far_high]);
1108        assert_eq!(plan.stats.selected, 1);
1109        assert_eq!(plan.stats.skipped_by_budget, 1);
1110
1111        let mut scratch = ReplicationScratch::default();
1112        let scratch_plan = ReplicationPlanner::plan_for_viewer_prioritized_with_scratch(
1113            &station,
1114            &index,
1115            &policies,
1116            &viewer,
1117            &RangeOnlyVisibility,
1118            ReplicationBudget {
1119                max_entities: 1,
1120                max_bytes: 32,
1121                estimated_entity_bytes: 32,
1122            },
1123            &mut scratch,
1124        );
1125        assert_eq!(scratch_plan.entities, plan.entities);
1126        assert_eq!(scratch_plan.stats, plan.stats);
1127        assert_eq!(scratch.candidate_count(), 2);
1128        assert!(scratch.prioritized_capacity() >= 2);
1129        assert_eq!(scratch.query_stats().candidate_handles, 2);
1130        assert!(scratch.candidate_capacity() >= 2);
1131        assert!(scratch.candidate_dedup_capacity() >= 2);
1132    }
1133
1134    #[test]
1135    fn replication_tracker_records_sent_ack_and_prune() {
1136        let client_id = ClientId::new(7);
1137        let first = EntityHandle::new(1, 0);
1138        let second = EntityHandle::new(2, 0);
1139        let plan = ReplicationPlan {
1140            entities: vec![first, second],
1141            stats: ReplicationStats::default(),
1142        };
1143        let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 4 });
1144
1145        let recorded = tracker
1146            .record_plan_sent(client_id, &plan, Tick::new(10))
1147            .expect("recording should fit");
1148        assert_eq!(recorded, 2);
1149        assert_eq!(tracker.last_sent(client_id, first), Some(Tick::new(10)));
1150        assert_eq!(tracker.stats().entries, 2);
1151        assert_eq!(tracker.stats().sent_records, 2);
1152
1153        assert!(tracker.acknowledge(client_id, first, Tick::new(11)));
1154        assert_eq!(
1155            tracker
1156                .get(client_id, first)
1157                .expect("tracked record")
1158                .last_acked,
1159            Some(Tick::new(11))
1160        );
1161        assert_eq!(tracker.stats().acked_records, 1);
1162
1163        assert_eq!(tracker.prune_sent_before(Tick::new(11)), 2);
1164        assert!(tracker.is_empty());
1165        assert_eq!(tracker.stats().pruned_records, 2);
1166    }
1167
1168    #[test]
1169    fn replication_tracker_rejects_capacity_without_partial_insert() {
1170        let client_id = ClientId::new(7);
1171        let plan = ReplicationPlan {
1172            entities: vec![EntityHandle::new(1, 0), EntityHandle::new(2, 0)],
1173            stats: ReplicationStats::default(),
1174        };
1175        let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 1 });
1176
1177        let error = tracker
1178            .record_plan_sent(client_id, &plan, Tick::new(10))
1179            .expect_err("recording should exceed capacity");
1180
1181        assert_eq!(
1182            error,
1183            ReplicationTrackerError::CapacityExceeded {
1184                current: 0,
1185                needed: 2,
1186                max: 1,
1187            }
1188        );
1189        assert!(tracker.is_empty());
1190        assert_eq!(tracker.stats().sent_records, 0);
1191    }
1192}
sectorsync_core/replication.rs

sectorsync_core/
replication.rs
