use std::collections::BTreeMap;
use crate::ids::{ClientId, EntityHandle, Tick};
use crate::interest::{ViewerQuery, VisibilityFilter};
use crate::policy::{CompiledSyncPolicy, PolicyTable};
use crate::spatial_index::{CellIndex, CellQueryScratch, CellQueryStats};
use crate::station::Station;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplicationBudget {
pub max_entities: usize,
pub max_bytes: usize,
pub estimated_entity_bytes: usize,
}
impl Default for ReplicationBudget {
fn default() -> Self {
Self {
max_entities: 300,
max_bytes: 16 * 1024,
estimated_entity_bytes: 32,
}
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ReplicationPlan {
pub entities: Vec<EntityHandle>,
pub stats: ReplicationStats,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplicationTrackerConfig {
pub max_entries: usize,
}
impl Default for ReplicationTrackerConfig {
fn default() -> Self {
Self {
max_entries: 65_536,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ReplicationTrackKey {
pub client_id: ClientId,
pub entity: EntityHandle,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplicationTrackRecord {
pub client_id: ClientId,
pub entity: EntityHandle,
pub last_sent: Tick,
pub last_acked: Option<Tick>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ReplicationTrackerStats {
pub entries: usize,
pub sent_records: usize,
pub acked_records: usize,
pub pruned_records: usize,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReplicationTrackerError {
CapacityExceeded {
current: usize,
needed: usize,
max: usize,
},
}
impl core::fmt::Display for ReplicationTrackerError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::CapacityExceeded {
current,
needed,
max,
} => write!(
f,
"replication tracker capacity exceeded: current {current}, needed {needed}, max {max}"
),
}
}
}
impl std::error::Error for ReplicationTrackerError {}
#[derive(Clone, Debug)]
pub struct ReplicationTracker {
config: ReplicationTrackerConfig,
records: BTreeMap<ReplicationTrackKey, ReplicationTrackRecord>,
stats: ReplicationTrackerStats,
}
impl Default for ReplicationTracker {
fn default() -> Self {
Self::new(ReplicationTrackerConfig::default())
}
}
impl ReplicationTracker {
pub fn new(config: ReplicationTrackerConfig) -> Self {
Self {
config,
records: BTreeMap::new(),
stats: ReplicationTrackerStats::default(),
}
}
pub const fn config(&self) -> ReplicationTrackerConfig {
self.config
}
pub const fn stats(&self) -> ReplicationTrackerStats {
self.stats
}
pub fn len(&self) -> usize {
self.records.len()
}
pub fn is_empty(&self) -> bool {
self.records.is_empty()
}
pub fn last_sent(&self, client_id: ClientId, entity: EntityHandle) -> Option<Tick> {
self.records
.get(&ReplicationTrackKey { client_id, entity })
.map(|record| record.last_sent)
}
pub fn get(&self, client_id: ClientId, entity: EntityHandle) -> Option<ReplicationTrackRecord> {
self.records
.get(&ReplicationTrackKey { client_id, entity })
.copied()
}
pub fn record_plan_sent(
&mut self,
client_id: ClientId,
plan: &ReplicationPlan,
sent_at: Tick,
) -> Result<usize, ReplicationTrackerError> {
self.ensure_capacity_for(client_id, &plan.entities)?;
let mut recorded = 0;
for entity in &plan.entities {
let key = ReplicationTrackKey {
client_id,
entity: *entity,
};
self.records.insert(
key,
ReplicationTrackRecord {
client_id,
entity: *entity,
last_sent: sent_at,
last_acked: None,
},
);
recorded += 1;
}
self.refresh_entry_count();
self.stats.sent_records = self.stats.sent_records.saturating_add(recorded);
Ok(recorded)
}
pub fn acknowledge(
&mut self,
client_id: ClientId,
entity: EntityHandle,
acked_at: Tick,
) -> bool {
let Some(record) = self
.records
.get_mut(&ReplicationTrackKey { client_id, entity })
else {
return false;
};
record.last_acked = Some(acked_at);
self.stats.acked_records = self.stats.acked_records.saturating_add(1);
true
}
pub fn acknowledge_plan(
&mut self,
client_id: ClientId,
plan: &ReplicationPlan,
acked_at: Tick,
) -> usize {
plan.entities
.iter()
.filter(|entity| self.acknowledge(client_id, **entity, acked_at))
.count()
}
pub fn clear_client(&mut self, client_id: ClientId) -> usize {
let before = self.records.len();
self.records.retain(|key, _| key.client_id != client_id);
let pruned = before.saturating_sub(self.records.len());
self.stats.pruned_records = self.stats.pruned_records.saturating_add(pruned);
self.refresh_entry_count();
pruned
}
pub fn prune_sent_before(&mut self, older_than: Tick) -> usize {
let before = self.records.len();
self.records
.retain(|_, record| record.last_sent.get() >= older_than.get());
let pruned = before.saturating_sub(self.records.len());
self.stats.pruned_records = self.stats.pruned_records.saturating_add(pruned);
self.refresh_entry_count();
pruned
}
fn ensure_capacity_for(
&self,
client_id: ClientId,
entities: &[EntityHandle],
) -> Result<(), ReplicationTrackerError> {
let mut needed = 0_usize;
for entity in entities {
if !self.records.contains_key(&ReplicationTrackKey {
client_id,
entity: *entity,
}) {
needed = needed.saturating_add(1);
}
}
if self.records.len().saturating_add(needed) > self.config.max_entries {
return Err(ReplicationTrackerError::CapacityExceeded {
current: self.records.len(),
needed,
max: self.config.max_entries,
});
}
Ok(())
}
fn refresh_entry_count(&mut self) {
self.stats.entries = self.records.len();
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ReplicationStats {
pub candidates: usize,
pub considered: usize,
pub selected: usize,
pub skipped_by_budget: usize,
pub skipped_by_cadence: usize,
pub estimated_bytes: usize,
}
#[derive(Clone, Copy, Debug, Default)]
pub struct ReplicationCadence;
impl ReplicationCadence {
pub fn target_hz(policy: &CompiledSyncPolicy, distance_squared: f32) -> u16 {
let min_hz = policy.min_hz.max(1);
let max_hz = policy.max_hz.max(min_hz);
let radius_squared = policy.interest_radius * policy.interest_radius;
let closeness =
if radius_squared.is_finite() && radius_squared > 0.0 && distance_squared.is_finite() {
1.0 - (distance_squared / radius_squared).clamp(0.0, 1.0)
} else {
1.0
};
let span = f32::from(max_hz - min_hz);
let target = f32::from(min_hz) + span * closeness;
rounded_frequency_to_u16(target, min_hz, max_hz)
}
pub fn interval_ticks(
policy: &CompiledSyncPolicy,
station_tick_rate_hz: u16,
distance_squared: f32,
) -> u64 {
let tick_rate = u64::from(station_tick_rate_hz.max(1));
let target_hz = u64::from(Self::target_hz(policy, distance_squared).max(1));
tick_rate.div_ceil(target_hz).max(1)
}
pub fn should_send(
policy: &CompiledSyncPolicy,
station_tick_rate_hz: u16,
distance_squared: f32,
now: Tick,
last_sent: Option<Tick>,
) -> bool {
let Some(last_sent) = last_sent else {
return true;
};
let interval = Self::interval_ticks(policy, station_tick_rate_hz, distance_squared);
now.get().saturating_sub(last_sent.get()) >= interval
}
}
#[derive(Clone, Copy, Debug, Default)]
pub struct ReplicationPriority;
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
fn rounded_frequency_to_u16(target: f32, min_hz: u16, max_hz: u16) -> u16 {
let bounded = target.round().clamp(f32::from(min_hz), f32::from(max_hz));
bounded as u16
}
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
fn normalized_score_to_u64(closeness: f32) -> u64 {
debug_assert!(closeness.is_finite() && (0.0..=1.0).contains(&closeness));
(closeness * 1_000_000.0).round() as u64
}
impl ReplicationPriority {
pub fn score(policy: &CompiledSyncPolicy, distance_squared: f32) -> u64 {
let weight = u64::from(policy.priority_weight.max(1));
let radius_squared = policy.interest_radius * policy.interest_radius;
let distance_score =
if radius_squared.is_finite() && radius_squared > 0.0 && distance_squared.is_finite() {
let closeness = 1.0 - (distance_squared / radius_squared).clamp(0.0, 1.0);
normalized_score_to_u64(closeness)
} else {
1_000_000
};
weight
.saturating_mul(1_000_000)
.saturating_add(distance_score)
}
}
#[derive(Clone, Copy, Debug, PartialEq)]
struct PrioritizedReplicationCandidate {
handle: EntityHandle,
score: u64,
distance_squared: f32,
}
#[derive(Clone, Debug, Default)]
pub struct ReplicationScratch {
cell_query: CellQueryScratch,
prioritized: Vec<PrioritizedReplicationCandidate>,
}
impl ReplicationScratch {
pub fn clear(&mut self) {
self.cell_query.clear();
self.prioritized.clear();
}
pub fn candidate_count(&self) -> usize {
self.cell_query.len()
}
pub fn prioritized_capacity(&self) -> usize {
self.prioritized.capacity()
}
pub const fn query_stats(&self) -> CellQueryStats {
self.cell_query.stats()
}
pub fn candidate_capacity(&self) -> usize {
self.cell_query.handle_capacity()
}
pub fn candidate_dedup_capacity(&self) -> usize {
self.cell_query.dedup_capacity()
}
pub fn matching_cell_capacity(&self) -> usize {
self.cell_query.matching_cell_capacity()
}
}
#[derive(Clone, Copy, Debug, Default)]
pub struct ReplicationPlanner;
impl ReplicationPlanner {
pub fn plan_for_viewer<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
) -> ReplicationPlan {
let candidates = index.query_sphere(viewer.position, viewer.radius);
Self::plan_for_candidates_inner(
station,
&candidates,
policies,
viewer,
filter,
budget,
|_, _, _| true,
)
}
pub fn plan_for_viewer_with_scratch<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
) -> ReplicationPlan {
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_candidates_inner(
station,
candidates,
policies,
viewer,
filter,
budget,
|_, _, _| true,
)
}
pub fn plan_for_viewer_with_cadence<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
) -> ReplicationPlan
where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let tick_rate_hz = station.config().tick_rate_hz;
let now = station.tick();
let candidates = index.query_sphere(viewer.position, viewer.radius);
Self::plan_for_candidates_inner(
station,
&candidates,
policies,
viewer,
filter,
budget,
|handle, policy, distance_squared| {
ReplicationCadence::should_send(
policy,
tick_rate_hz,
distance_squared,
now,
last_sent(handle),
)
},
)
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_with_cadence_and_scratch<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
scratch: &mut ReplicationScratch,
) -> ReplicationPlan
where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let tick_rate_hz = station.config().tick_rate_hz;
let now = station.tick();
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_candidates_inner(
station,
candidates,
policies,
viewer,
filter,
budget,
|handle, policy, distance_squared| {
ReplicationCadence::should_send(
policy,
tick_rate_hz,
distance_squared,
now,
last_sent(handle),
)
},
)
}
pub fn plan_for_viewer_prioritized<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
) -> ReplicationPlan {
let candidates = index.query_sphere(viewer.position, viewer.radius);
let mut prioritized = Vec::new();
Self::plan_for_candidates_prioritized_inner(
station,
&candidates,
policies,
viewer,
filter,
budget,
&mut prioritized,
|_, _, _| true,
)
}
pub fn plan_for_viewer_prioritized_with_scratch<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
) -> ReplicationPlan {
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_candidates_prioritized_inner(
station,
candidates,
policies,
viewer,
filter,
budget,
&mut scratch.prioritized,
|_, _, _| true,
)
}
pub fn plan_for_viewer_prioritized_with_cadence<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
) -> ReplicationPlan
where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let tick_rate_hz = station.config().tick_rate_hz;
let now = station.tick();
let candidates = index.query_sphere(viewer.position, viewer.radius);
let mut prioritized = Vec::new();
Self::plan_for_candidates_prioritized_inner(
station,
&candidates,
policies,
viewer,
filter,
budget,
&mut prioritized,
|handle, policy, distance_squared| {
ReplicationCadence::should_send(
policy,
tick_rate_hz,
distance_squared,
now,
last_sent(handle),
)
},
)
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_prioritized_with_cadence_and_scratch<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
scratch: &mut ReplicationScratch,
) -> ReplicationPlan
where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let tick_rate_hz = station.config().tick_rate_hz;
let now = station.tick();
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_candidates_prioritized_inner(
station,
candidates,
policies,
viewer,
filter,
budget,
&mut scratch.prioritized,
|handle, policy, distance_squared| {
ReplicationCadence::should_send(
policy,
tick_rate_hz,
distance_squared,
now,
last_sent(handle),
)
},
)
}
fn plan_for_candidates_inner<F, C>(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
cadence_allows: C,
) -> ReplicationPlan
where
F: VisibilityFilter,
C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
{
let max_entities = viewer.max_entities.min(budget.max_entities);
let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
let hard_limit = max_entities.min(max_by_bytes);
let mut plan = ReplicationPlan {
entities: Vec::with_capacity(hard_limit),
stats: ReplicationStats {
candidates: candidates.len(),
..ReplicationStats::default()
},
};
for handle in candidates {
let Some(entity) = station.get(*handle) else {
continue;
};
plan.stats.considered += 1;
let Some(policy) = policies.get(entity.policy_id) else {
continue;
};
let distance_squared = entity.position.distance_squared(viewer.position);
let policy_radius_sq = policy.interest_radius * policy.interest_radius;
if distance_squared > policy_radius_sq {
continue;
}
if !filter.is_visible(viewer, entity) {
continue;
}
if !cadence_allows(*handle, policy, distance_squared) {
plan.stats.skipped_by_cadence += 1;
continue;
}
if plan.entities.len() >= hard_limit {
plan.stats.skipped_by_budget += 1;
continue;
}
plan.entities.push(*handle);
}
plan.stats.selected = plan.entities.len();
plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
plan
}
#[allow(clippy::too_many_arguments)]
fn plan_for_candidates_prioritized_inner<F, C>(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
eligible: &mut Vec<PrioritizedReplicationCandidate>,
cadence_allows: C,
) -> ReplicationPlan
where
F: VisibilityFilter,
C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
{
let max_entities = viewer.max_entities.min(budget.max_entities);
let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
let hard_limit = max_entities.min(max_by_bytes);
let mut plan = ReplicationPlan {
entities: Vec::with_capacity(hard_limit),
stats: ReplicationStats {
candidates: candidates.len(),
..ReplicationStats::default()
},
};
eligible.clear();
for handle in candidates {
let Some(entity) = station.get(*handle) else {
continue;
};
plan.stats.considered += 1;
let Some(policy) = policies.get(entity.policy_id) else {
continue;
};
let distance_squared = entity.position.distance_squared(viewer.position);
let policy_radius_sq = policy.interest_radius * policy.interest_radius;
if distance_squared > policy_radius_sq {
continue;
}
if !filter.is_visible(viewer, entity) {
continue;
}
if !cadence_allows(*handle, policy, distance_squared) {
plan.stats.skipped_by_cadence += 1;
continue;
}
eligible.push(PrioritizedReplicationCandidate {
handle: *handle,
score: ReplicationPriority::score(policy, distance_squared),
distance_squared,
});
}
eligible.sort_by(|left, right| {
right
.score
.cmp(&left.score)
.then_with(|| left.distance_squared.total_cmp(&right.distance_squared))
.then_with(|| left.handle.cmp(&right.handle))
});
plan.stats.skipped_by_budget = eligible.len().saturating_sub(hard_limit);
plan.entities.extend(
eligible
.iter()
.take(hard_limit)
.map(|candidate| candidate.handle),
);
plan.stats.selected = plan.entities.len();
plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
plan
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::entity::EntityTags;
use crate::ids::{ClientId, EntityId, InstanceId, NodeId, PolicyId, StationId};
use crate::interest::{AndVisibility, FrustumVisibility, RangeOnlyVisibility, TagVisibility};
use crate::policy::CompiledSyncPolicy;
use crate::spatial::{Aabb3, Bounds, Frustum3, GridSpec, Position3};
use crate::station::{Station, StationConfig};
#[test]
fn planner_applies_composed_frustum_visibility_filter() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let grid = GridSpec::new(16.0).expect("grid is valid");
let mut index = CellIndex::new(grid);
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
let visible = station
.spawn_owned(
EntityId::new(1),
Position3::new(10.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn visible");
let outside_frustum = station
.spawn_owned(
EntityId::new(2),
Position3::new(-10.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn outside frustum");
index.upsert(visible, Position3::new(10.0, 0.0, 0.0), Bounds::Point);
index.upsert(
outside_frustum,
Position3::new(-10.0, 0.0, 0.0),
Bounds::Point,
);
let viewer = ViewerQuery {
client_id: ClientId::new(7),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 8,
};
let frustum = Frustum3::from_aabb(Aabb3::new(
Position3::new(0.0, -20.0, -20.0),
Position3::new(80.0, 20.0, 20.0),
));
let filter = AndVisibility::new(RangeOnlyVisibility, FrustumVisibility::new(frustum));
let plan = ReplicationPlanner::plan_for_viewer(
&station,
&index,
&policies,
&viewer,
&filter,
ReplicationBudget::default(),
);
assert_eq!(plan.entities, vec![visible]);
assert_eq!(plan.stats.selected, 1);
assert_eq!(plan.stats.considered, 2);
}
#[test]
fn planner_applies_tag_visibility_filter() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let grid = GridSpec::new(16.0).expect("grid is valid");
let mut index = CellIndex::new(grid);
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
let static_visible = station
.spawn_owned(
EntityId::new(1),
Position3::new(10.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn static");
let fast_mover = station
.spawn_owned(
EntityId::new(2),
Position3::new(12.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn mover");
station
.set_tags(static_visible, EntityTags::from_bits(0b001))
.expect("tag static");
station
.set_tags(fast_mover, EntityTags::from_bits(0b010))
.expect("tag mover");
index.upsert(
static_visible,
Position3::new(10.0, 0.0, 0.0),
Bounds::Point,
);
index.upsert(fast_mover, Position3::new(12.0, 0.0, 0.0), Bounds::Point);
let viewer = ViewerQuery {
client_id: ClientId::new(7),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 8,
};
let filter = AndVisibility::new(
RangeOnlyVisibility,
TagVisibility::new(EntityTags::from_bits(0b001), EntityTags::from_bits(0b010)),
);
let plan = ReplicationPlanner::plan_for_viewer(
&station,
&index,
&policies,
&viewer,
&filter,
ReplicationBudget::default(),
);
assert_eq!(plan.entities, vec![static_visible]);
assert_eq!(plan.stats.selected, 1);
assert_eq!(plan.stats.considered, 2);
}
#[test]
fn cadence_scales_interval_by_squared_distance() {
let policy = CompiledSyncPolicy::new(PolicyId::new(1), 2, 20, 100.0);
assert_eq!(ReplicationCadence::target_hz(&policy, 0.0), 20);
assert_eq!(ReplicationCadence::interval_ticks(&policy, 20, 0.0), 1);
assert_eq!(ReplicationCadence::target_hz(&policy, 100.0_f32 * 100.0), 2);
assert_eq!(
ReplicationCadence::interval_ticks(&policy, 20, 100.0_f32 * 100.0),
10
);
}
#[test]
fn priority_score_prefers_weight_then_distance() {
let mut low = CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 100.0);
low.priority_weight = 1;
let mut high = CompiledSyncPolicy::new(PolicyId::new(2), 1, 20, 100.0);
high.priority_weight = 10;
assert!(
ReplicationPriority::score(&high, 90.0 * 90.0) > ReplicationPriority::score(&low, 0.0)
);
assert!(
ReplicationPriority::score(&low, 0.0) > ReplicationPriority::score(&low, 90.0 * 90.0)
);
}
#[test]
fn planner_with_cadence_skips_recent_far_entities() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
for _ in 0..10 {
station.advance_tick();
}
let grid = GridSpec::new(16.0).expect("grid is valid");
let mut index = CellIndex::new(grid);
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 2, 20, 128.0));
let near = station
.spawn_owned(
EntityId::new(1),
Position3::new(0.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn near");
let far = station
.spawn_owned(
EntityId::new(2),
Position3::new(120.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn far");
index.upsert(near, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
index.upsert(far, Position3::new(120.0, 0.0, 0.0), Bounds::Point);
let viewer = ViewerQuery {
client_id: ClientId::new(7),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 8,
};
let plan = ReplicationPlanner::plan_for_viewer_with_cadence(
&station,
&index,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget::default(),
|_| Some(Tick::new(9)),
);
assert_eq!(plan.entities, vec![near]);
assert_eq!(plan.stats.selected, 1);
assert_eq!(plan.stats.skipped_by_cadence, 1);
}
#[test]
fn prioritized_planner_uses_policy_weight_under_budget() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let grid = GridSpec::new(16.0).expect("grid is valid");
let mut index = CellIndex::new(grid);
let mut policies = PolicyTable::default();
let mut low = CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0);
low.priority_weight = 1;
let mut high = CompiledSyncPolicy::new(PolicyId::new(2), 1, 20, 128.0);
high.priority_weight = 10;
policies.set(low);
policies.set(high);
let near_low = station
.spawn_owned(
EntityId::new(1),
Position3::new(0.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn near low priority");
let far_high = station
.spawn_owned(
EntityId::new(2),
Position3::new(96.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(2),
)
.expect("spawn far high priority");
index.upsert(near_low, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
index.upsert(far_high, Position3::new(96.0, 0.0, 0.0), Bounds::Point);
let viewer = ViewerQuery {
client_id: ClientId::new(7),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 1,
};
let plan = ReplicationPlanner::plan_for_viewer_prioritized(
&station,
&index,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget {
max_entities: 1,
max_bytes: 32,
estimated_entity_bytes: 32,
},
);
assert_eq!(plan.entities, vec![far_high]);
assert_eq!(plan.stats.selected, 1);
assert_eq!(plan.stats.skipped_by_budget, 1);
let mut scratch = ReplicationScratch::default();
let scratch_plan = ReplicationPlanner::plan_for_viewer_prioritized_with_scratch(
&station,
&index,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget {
max_entities: 1,
max_bytes: 32,
estimated_entity_bytes: 32,
},
&mut scratch,
);
assert_eq!(scratch_plan.entities, plan.entities);
assert_eq!(scratch_plan.stats, plan.stats);
assert_eq!(scratch.candidate_count(), 2);
assert!(scratch.prioritized_capacity() >= 2);
assert_eq!(scratch.query_stats().candidate_handles, 2);
assert!(scratch.candidate_capacity() >= 2);
assert!(scratch.candidate_dedup_capacity() >= 2);
}
#[test]
fn replication_tracker_records_sent_ack_and_prune() {
let client_id = ClientId::new(7);
let first = EntityHandle::new(1, 0);
let second = EntityHandle::new(2, 0);
let plan = ReplicationPlan {
entities: vec![first, second],
stats: ReplicationStats::default(),
};
let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 4 });
let recorded = tracker
.record_plan_sent(client_id, &plan, Tick::new(10))
.expect("recording should fit");
assert_eq!(recorded, 2);
assert_eq!(tracker.last_sent(client_id, first), Some(Tick::new(10)));
assert_eq!(tracker.stats().entries, 2);
assert_eq!(tracker.stats().sent_records, 2);
assert!(tracker.acknowledge(client_id, first, Tick::new(11)));
assert_eq!(
tracker
.get(client_id, first)
.expect("tracked record")
.last_acked,
Some(Tick::new(11))
);
assert_eq!(tracker.stats().acked_records, 1);
assert_eq!(tracker.prune_sent_before(Tick::new(11)), 2);
assert!(tracker.is_empty());
assert_eq!(tracker.stats().pruned_records, 2);
}
#[test]
fn replication_tracker_rejects_capacity_without_partial_insert() {
let client_id = ClientId::new(7);
let plan = ReplicationPlan {
entities: vec![EntityHandle::new(1, 0), EntityHandle::new(2, 0)],
stats: ReplicationStats::default(),
};
let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 1 });
let error = tracker
.record_plan_sent(client_id, &plan, Tick::new(10))
.expect_err("recording should exceed capacity");
assert_eq!(
error,
ReplicationTrackerError::CapacityExceeded {
current: 0,
needed: 2,
max: 1,
}
);
assert!(tracker.is_empty());
assert_eq!(tracker.stats().sent_records, 0);
}
}