sectorsync_core/
station.rs

1//! Station-local entity storage and runtime primitives.
2
3use std::collections::HashMap;
4
5use crate::entity::{DirtyMask, EntityRecord, EntityRole, EntityTags};
6use crate::handoff::HandoffTransfer;
7use crate::ids::{
8    EntityHandle, EntityId, InstanceId, NodeId, OwnerEpoch, PolicyId, StationId, Tick,
9};
10use crate::snapshot::{SnapshotMeta, SnapshotVersion, StationSnapshot};
11use crate::spatial::{Bounds, Position3};
12
13/// Station runtime configuration.
14#[derive(Clone, Copy, Debug, PartialEq, Eq)]
15pub struct StationConfig {
16    /// Station id.
17    pub station_id: StationId,
18    /// Node id selected by the embedding application.
19    pub node_id: NodeId,
20    /// World instance id.
21    pub instance_id: InstanceId,
22    /// Fixed authoritative tick rate in hertz.
23    pub tick_rate_hz: u16,
24}
25
26/// Station-local storage and metadata.
27#[derive(Clone, Debug)]
28pub struct Station {
29    config: StationConfig,
30    tick: Tick,
31    owner_epoch: OwnerEpoch,
32    records: Vec<Option<EntityRecord>>,
33    generations: Vec<u32>,
34    free: Vec<u32>,
35    by_id: HashMap<EntityId, EntityHandle>,
36}
37
38impl Station {
39    /// Creates an empty station.
40    pub fn new(config: StationConfig) -> Self {
41        Self {
42            config,
43            tick: Tick::new(0),
44            owner_epoch: OwnerEpoch::new(0),
45            records: Vec::new(),
46            generations: Vec::new(),
47            free: Vec::new(),
48            by_id: HashMap::new(),
49        }
50    }
51
52    /// Returns station configuration.
53    pub const fn config(&self) -> StationConfig {
54        self.config
55    }
56
57    /// Current station tick.
58    pub const fn tick(&self) -> Tick {
59        self.tick
60    }
61
62    /// Current owner epoch.
63    pub const fn owner_epoch(&self) -> OwnerEpoch {
64        self.owner_epoch
65    }
66
67    /// Reserves and returns the next owner epoch for this station.
68    pub fn next_owner_epoch(&mut self) -> OwnerEpoch {
69        self.owner_epoch = OwnerEpoch::new(self.owner_epoch.get().saturating_add(1));
70        self.owner_epoch
71    }
72
73    /// Number of live entity records.
74    pub fn len(&self) -> usize {
75        self.by_id.len()
76    }
77
78    /// Returns whether the station has no entity records.
79    pub fn is_empty(&self) -> bool {
80        self.by_id.is_empty()
81    }
82
83    /// Advances the authoritative station tick.
84    pub fn advance_tick(&mut self) {
85        self.tick = Tick::new(self.tick.get().saturating_add(1));
86    }
87
88    /// Spawns an authoritative entity in this station.
89    pub fn spawn_owned(
90        &mut self,
91        id: EntityId,
92        position: Position3,
93        bounds: Bounds,
94        policy_id: PolicyId,
95    ) -> Result<EntityHandle, StationError> {
96        if self.by_id.contains_key(&id) {
97            return Err(StationError::DuplicateEntity(id));
98        }
99
100        let handle = self.allocate_handle();
101        let record = EntityRecord::owned(id, handle, position, bounds, policy_id, self.owner_epoch);
102        self.insert_allocated(handle, record);
103        Ok(handle)
104    }
105
106    /// Inserts or refreshes a read-only ghost entity.
107    #[allow(clippy::too_many_arguments)]
108    pub fn upsert_ghost(
109        &mut self,
110        id: EntityId,
111        position: Position3,
112        bounds: Bounds,
113        policy_id: PolicyId,
114        owner_station: StationId,
115        owner_epoch: OwnerEpoch,
116        expires_at: Tick,
117    ) -> EntityHandle {
118        if let Some(handle) = self.by_id.get(&id).copied() {
119            if let Some(record) = self.get_mut(handle) {
120                record.position = position;
121                record.bounds = bounds;
122                record.policy_id = policy_id;
123                record.role = EntityRole::Ghost {
124                    owner_station,
125                    owner_epoch,
126                    expires_at,
127                };
128                record.dirty.insert(DirtyMask::TRANSFORM);
129            }
130            return handle;
131        }
132
133        let handle = self.allocate_handle();
134        let record = EntityRecord::ghost(
135            id,
136            handle,
137            position,
138            bounds,
139            policy_id,
140            owner_station,
141            owner_epoch,
142            expires_at,
143        );
144        self.insert_allocated(handle, record);
145        handle
146    }
147
148    /// Gets an entity by handle if the generation is still valid.
149    pub fn get(&self, handle: EntityHandle) -> Option<&EntityRecord> {
150        let index = usize::try_from(handle.index()).ok()?;
151        let generation = self.generations.get(index).copied()?;
152        if generation != handle.generation() {
153            return None;
154        }
155        self.records.get(index)?.as_ref()
156    }
157
158    /// Gets a mutable entity record by handle.
159    pub fn get_mut(&mut self, handle: EntityHandle) -> Option<&mut EntityRecord> {
160        let index = usize::try_from(handle.index()).ok()?;
161        let generation = self.generations.get(index).copied()?;
162        if generation != handle.generation() {
163            return None;
164        }
165        self.records.get_mut(index)?.as_mut()
166    }
167
168    /// Gets an entity by stable id.
169    pub fn get_by_id(&self, id: EntityId) -> Option<&EntityRecord> {
170        let handle = self.by_id.get(&id).copied()?;
171        self.get(handle)
172    }
173
174    /// Gets a mutable entity record by stable id.
175    pub fn get_by_id_mut(&mut self, id: EntityId) -> Option<&mut EntityRecord> {
176        let handle = self.by_id.get(&id).copied()?;
177        self.get_mut(handle)
178    }
179
180    /// Gets a station-local handle by stable id.
181    pub fn handle_by_id(&self, id: EntityId) -> Option<EntityHandle> {
182        self.by_id.get(&id).copied()
183    }
184
185    /// Moves an authoritative entity and marks transform dirty.
186    pub fn move_owned(
187        &mut self,
188        handle: EntityHandle,
189        position: Position3,
190    ) -> Result<(), StationError> {
191        let record = self
192            .get_mut(handle)
193            .ok_or(StationError::MissingEntityHandle(handle))?;
194        if !record.is_owned() {
195            return Err(StationError::NotOwner(record.id));
196        }
197        record.position = position;
198        record.dirty.insert(DirtyMask::TRANSFORM);
199        Ok(())
200    }
201
202    /// Replaces authoritative entity tags and marks tags dirty.
203    pub fn set_tags(&mut self, handle: EntityHandle, tags: EntityTags) -> Result<(), StationError> {
204        let record = self
205            .get_mut(handle)
206            .ok_or(StationError::MissingEntityHandle(handle))?;
207        if !record.is_owned() {
208            return Err(StationError::NotOwner(record.id));
209        }
210        record.tags = tags;
211        record.dirty.insert(DirtyMask::TAGS);
212        Ok(())
213    }
214
215    /// Clears selected dirty bits for a local entity record.
216    pub fn clear_dirty(
217        &mut self,
218        handle: EntityHandle,
219        mask: DirtyMask,
220    ) -> Result<(), StationError> {
221        let record = self
222            .get_mut(handle)
223            .ok_or(StationError::MissingEntityHandle(handle))?;
224        record.dirty.remove(mask);
225        Ok(())
226    }
227
228    /// Iterates over live records.
229    pub fn iter(&self) -> impl Iterator<Item = &EntityRecord> {
230        self.records.iter().filter_map(Option::as_ref)
231    }
232
233    /// Removes an entity record by handle.
234    pub fn remove(&mut self, handle: EntityHandle) -> Result<EntityRecord, StationError> {
235        let index = usize::try_from(handle.index())
236            .map_err(|_| StationError::MissingEntityHandle(handle))?;
237        let generation = self
238            .generations
239            .get(index)
240            .copied()
241            .ok_or(StationError::MissingEntityHandle(handle))?;
242        if generation != handle.generation() {
243            return Err(StationError::MissingEntityHandle(handle));
244        }
245
246        let record = self.records[index]
247            .take()
248            .ok_or(StationError::MissingEntityHandle(handle))?;
249        self.by_id.remove(&record.id);
250        self.generations[index] = self.generations[index].saturating_add(1);
251        self.free.push(handle.index());
252        Ok(record)
253    }
254
255    /// Removes an entity record by stable id.
256    pub fn remove_by_id(&mut self, id: EntityId) -> Result<EntityRecord, StationError> {
257        let handle = self
258            .by_id
259            .get(&id)
260            .copied()
261            .ok_or(StationError::MissingEntity(id))?;
262        self.remove(handle)
263    }
264
265    /// Prepares an outgoing handoff transfer without mutating ownership yet.
266    pub fn prepare_outgoing_handoff(
267        &self,
268        entity_id: EntityId,
269        target_station: StationId,
270        target_owner_epoch: OwnerEpoch,
271        source_ghost_expires_at: Tick,
272    ) -> Result<HandoffTransfer, StationError> {
273        if target_station == self.config.station_id {
274            return Err(StationError::HandoffTargetIsSource(target_station));
275        }
276
277        let entity = self
278            .get_by_id(entity_id)
279            .ok_or(StationError::MissingEntity(entity_id))?;
280        if !entity.is_owned() {
281            return Err(StationError::NotOwner(entity_id));
282        }
283
284        Ok(HandoffTransfer {
285            entity_id,
286            source_station: self.config.station_id,
287            target_station,
288            source_owner_epoch: entity.role.owner_epoch(),
289            target_owner_epoch,
290            prepared_at: self.tick,
291            source_ghost_expires_at,
292            entity: entity.clone(),
293        })
294    }
295
296    /// Prewarms or refreshes a target-side ghost before owner commit.
297    pub fn prewarm_handoff_ghost(
298        &mut self,
299        transfer: &HandoffTransfer,
300    ) -> Result<EntityHandle, StationError> {
301        if transfer.target_station != self.config.station_id {
302            return Err(StationError::WrongHandoffTarget {
303                expected: self.config.station_id,
304                actual: transfer.target_station,
305            });
306        }
307
308        Ok(self.upsert_ghost(
309            transfer.entity_id,
310            transfer.entity.position,
311            transfer.entity.bounds,
312            transfer.entity.policy_id,
313            transfer.source_station,
314            transfer.source_owner_epoch,
315            transfer.source_ghost_expires_at,
316        ))
317    }
318
319    /// Commits the target side of an incoming handoff and becomes authoritative.
320    pub fn commit_incoming_handoff(
321        &mut self,
322        transfer: HandoffTransfer,
323    ) -> Result<EntityHandle, StationError> {
324        if transfer.target_station != self.config.station_id {
325            return Err(StationError::WrongHandoffTarget {
326                expected: self.config.station_id,
327                actual: transfer.target_station,
328            });
329        }
330
331        if let Some(handle) = self.handle_by_id(transfer.entity_id) {
332            let record = self
333                .get_mut(handle)
334                .ok_or(StationError::MissingEntityHandle(handle))?;
335            if record.is_owned() {
336                return Err(StationError::AlreadyOwner(transfer.entity_id));
337            }
338
339            *record = transfer.entity;
340            record.handle = handle;
341            record.role = EntityRole::Owned {
342                owner_epoch: transfer.target_owner_epoch,
343            };
344            record.dirty.insert(DirtyMask::TRANSFORM);
345            self.owner_epoch = transfer.target_owner_epoch;
346            return Ok(handle);
347        }
348
349        let handle = self.allocate_handle();
350        let mut record = transfer.entity;
351        record.handle = handle;
352        record.role = EntityRole::Owned {
353            owner_epoch: transfer.target_owner_epoch,
354        };
355        record.dirty.insert(DirtyMask::TRANSFORM);
356        self.owner_epoch = transfer.target_owner_epoch;
357        self.insert_allocated(handle, record);
358        Ok(handle)
359    }
360
361    /// Commits the source side of an outgoing handoff and keeps a short-lived ghost.
362    pub fn commit_outgoing_handoff(
363        &mut self,
364        transfer: &HandoffTransfer,
365    ) -> Result<EntityHandle, StationError> {
366        if transfer.source_station != self.config.station_id {
367            return Err(StationError::WrongHandoffSource {
368                expected: self.config.station_id,
369                actual: transfer.source_station,
370            });
371        }
372
373        let handle = self
374            .handle_by_id(transfer.entity_id)
375            .ok_or(StationError::MissingEntity(transfer.entity_id))?;
376        let record = self
377            .get_mut(handle)
378            .ok_or(StationError::MissingEntityHandle(handle))?;
379        if !record.is_owned() {
380            return Err(StationError::NotOwner(transfer.entity_id));
381        }
382
383        record.role = EntityRole::Ghost {
384            owner_station: transfer.target_station,
385            owner_epoch: transfer.target_owner_epoch,
386            expires_at: transfer.source_ghost_expires_at,
387        };
388        record.dirty.insert(DirtyMask::TRANSFORM);
389        Ok(handle)
390    }
391
392    /// Exports an in-memory station snapshot.
393    pub fn snapshot(&self, version: SnapshotVersion) -> StationSnapshot {
394        let entities = self.iter().cloned().collect::<Vec<_>>();
395        StationSnapshot {
396            meta: SnapshotMeta {
397                instance_id: self.config.instance_id,
398                station_id: self.config.station_id,
399                tick: self.tick,
400                entity_count: entities.len(),
401                owner_epoch: self.owner_epoch,
402                version,
403            },
404            entities,
405        }
406    }
407
408    /// Restores station state from a snapshot.
409    pub fn restore(config: StationConfig, snapshot: StationSnapshot) -> Result<Self, StationError> {
410        if snapshot.meta.station_id != config.station_id {
411            return Err(StationError::SnapshotStationMismatch {
412                expected: config.station_id,
413                actual: snapshot.meta.station_id,
414            });
415        }
416
417        let mut station = Self::new(config);
418        station.tick = snapshot.meta.tick;
419        station.owner_epoch = snapshot.meta.owner_epoch;
420
421        for mut record in snapshot.entities {
422            if station.by_id.contains_key(&record.id) {
423                return Err(StationError::DuplicateEntity(record.id));
424            }
425            let handle = station.allocate_handle();
426            record.handle = handle;
427            station.insert_allocated(handle, record);
428        }
429
430        Ok(station)
431    }
432
433    fn allocate_handle(&mut self) -> EntityHandle {
434        if let Some(index) = self.free.pop() {
435            let generation = self.generations[index as usize];
436            EntityHandle::new(index, generation)
437        } else {
438            let index =
439                u32::try_from(self.records.len()).expect("station entity capacity exceeded");
440            self.records.push(None);
441            self.generations.push(0);
442            EntityHandle::new(index, 0)
443        }
444    }
445
446    fn insert_allocated(&mut self, handle: EntityHandle, record: EntityRecord) {
447        let index = handle.index() as usize;
448        self.by_id.insert(record.id, handle);
449        self.records[index] = Some(record);
450    }
451}
452
453/// Station operation error.
454#[derive(Clone, Copy, Debug, PartialEq, Eq)]
455pub enum StationError {
456    /// Entity id already exists in this station.
457    DuplicateEntity(EntityId),
458    /// Entity id does not exist in this station.
459    MissingEntity(EntityId),
460    /// Entity handle is missing or stale.
461    MissingEntityHandle(EntityHandle),
462    /// Operation requires an authoritative entity.
463    NotOwner(EntityId),
464    /// This station is already authoritative for the entity.
465    AlreadyOwner(EntityId),
466    /// Handoff target cannot be the source station.
467    HandoffTargetIsSource(StationId),
468    /// Incoming transfer was addressed to a different target station.
469    WrongHandoffTarget {
470        /// Expected target station id.
471        expected: StationId,
472        /// Actual target station id in transfer.
473        actual: StationId,
474    },
475    /// Outgoing transfer was addressed from a different source station.
476    WrongHandoffSource {
477        /// Expected source station id.
478        expected: StationId,
479        /// Actual source station id in transfer.
480        actual: StationId,
481    },
482    /// Snapshot was captured from a different station.
483    SnapshotStationMismatch {
484        /// Expected station id.
485        expected: StationId,
486        /// Actual snapshot station id.
487        actual: StationId,
488    },
489}
490
491impl core::fmt::Display for StationError {
492    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
493        match self {
494            Self::DuplicateEntity(id) => write!(f, "duplicate entity id {}", id.get()),
495            Self::MissingEntity(id) => write!(f, "missing entity id {}", id.get()),
496            Self::MissingEntityHandle(handle) => {
497                write!(
498                    f,
499                    "missing entity handle index={} generation={}",
500                    handle.index(),
501                    handle.generation()
502                )
503            }
504            Self::NotOwner(id) => write!(f, "entity {} is not authoritative here", id.get()),
505            Self::AlreadyOwner(id) => {
506                write!(f, "entity {} is already authoritative here", id.get())
507            }
508            Self::HandoffTargetIsSource(station_id) => {
509                write!(
510                    f,
511                    "handoff target {} is the source station",
512                    station_id.get()
513                )
514            }
515            Self::WrongHandoffTarget { expected, actual } => write!(
516                f,
517                "wrong handoff target: expected {}, got {}",
518                expected.get(),
519                actual.get()
520            ),
521            Self::WrongHandoffSource { expected, actual } => write!(
522                f,
523                "wrong handoff source: expected {}, got {}",
524                expected.get(),
525                actual.get()
526            ),
527            Self::SnapshotStationMismatch { expected, actual } => write!(
528                f,
529                "snapshot station mismatch: expected {}, got {}",
530                expected.get(),
531                actual.get()
532            ),
533        }
534    }
535}
536
537impl std::error::Error for StationError {}
538
539#[cfg(test)]
540mod tests {
541    use super::*;
542
543    fn config() -> StationConfig {
544        StationConfig {
545            station_id: StationId::new(1),
546            node_id: NodeId::new(1),
547            instance_id: InstanceId::new(7),
548            tick_rate_hz: 20,
549        }
550    }
551
552    fn config_with_station(station_id: u32) -> StationConfig {
553        StationConfig {
554            station_id: StationId::new(station_id),
555            node_id: NodeId::new(1),
556            instance_id: InstanceId::new(7),
557            tick_rate_hz: 20,
558        }
559    }
560
561    #[test]
562    fn owned_entities_can_move_and_snapshot_restore() {
563        let mut station = Station::new(config());
564        let handle = station
565            .spawn_owned(
566                EntityId::new(42),
567                Position3::new(1.0, 2.0, 3.0),
568                Bounds::Point,
569                PolicyId::new(0),
570            )
571            .expect("spawn should work");
572
573        station
574            .move_owned(handle, Position3::new(2.0, 3.0, 4.0))
575            .expect("owned move should work");
576        station.advance_tick();
577
578        let snapshot = station.snapshot(SnapshotVersion::default());
579        let restored = Station::restore(config(), snapshot).expect("restore should work");
580
581        assert_eq!(restored.len(), 1);
582        assert_eq!(restored.tick(), Tick::new(1));
583        assert_eq!(
584            restored
585                .get_by_id(EntityId::new(42))
586                .expect("entity should exist")
587                .position,
588            Position3::new(2.0, 3.0, 4.0)
589        );
590    }
591
592    #[test]
593    fn ghosts_cannot_be_moved_as_owned() {
594        let mut station = Station::new(config());
595        let handle = station.upsert_ghost(
596            EntityId::new(5),
597            Position3::new(0.0, 0.0, 0.0),
598            Bounds::Point,
599            PolicyId::new(0),
600            StationId::new(2),
601            OwnerEpoch::new(1),
602            Tick::new(10),
603        );
604
605        let error = station
606            .move_owned(handle, Position3::new(1.0, 0.0, 0.0))
607            .expect_err("ghost move should fail");
608        assert_eq!(error, StationError::NotOwner(EntityId::new(5)));
609    }
610
611    #[test]
612    fn owned_tags_can_be_replaced_and_mark_dirty() {
613        let mut station = Station::new(config());
614        let handle = station
615            .spawn_owned(
616                EntityId::new(6),
617                Position3::new(0.0, 0.0, 0.0),
618                Bounds::Point,
619                PolicyId::new(0),
620            )
621            .expect("spawn should work");
622        let tags = EntityTags::from_bits(0b101);
623
624        station
625            .set_tags(handle, tags)
626            .expect("set tags should work");
627        let record = station.get(handle).expect("entity should exist");
628
629        assert_eq!(record.tags, tags);
630        assert!(record.dirty.contains(DirtyMask::TAGS));
631    }
632
633    #[test]
634    fn selected_dirty_bits_can_be_cleared() {
635        let mut station = Station::new(config());
636        let handle = station
637            .spawn_owned(
638                EntityId::new(8),
639                Position3::new(0.0, 0.0, 0.0),
640                Bounds::Point,
641                PolicyId::new(0),
642            )
643            .expect("spawn should work");
644        station
645            .set_tags(handle, EntityTags::from_bits(1))
646            .expect("set tags should work");
647
648        station
649            .clear_dirty(handle, DirtyMask::TAGS)
650            .expect("clear dirty should work");
651        let record = station.get(handle).expect("entity should exist");
652
653        assert!(!record.dirty.contains(DirtyMask::TAGS));
654        assert!(record.dirty.contains(DirtyMask::TRANSFORM));
655    }
656
657    #[test]
658    fn ghost_tags_cannot_be_replaced_as_owned() {
659        let mut station = Station::new(config());
660        let handle = station.upsert_ghost(
661            EntityId::new(7),
662            Position3::new(0.0, 0.0, 0.0),
663            Bounds::Point,
664            PolicyId::new(0),
665            StationId::new(2),
666            OwnerEpoch::new(1),
667            Tick::new(10),
668        );
669
670        let error = station
671            .set_tags(handle, EntityTags::from_bits(1))
672            .expect_err("ghost tags should fail");
673        assert_eq!(error, StationError::NotOwner(EntityId::new(7)));
674    }
675
676    #[test]
677    fn two_phase_handoff_prewarms_target_and_commits_owner_switch() {
678        let mut source = Station::new(config_with_station(1));
679        let mut target = Station::new(config_with_station(2));
680
681        source
682            .spawn_owned(
683                EntityId::new(9),
684                Position3::new(10.0, 20.0, 30.0),
685                Bounds::Point,
686                PolicyId::new(0),
687            )
688            .expect("spawn should work");
689        source.advance_tick();
690
691        let target_epoch = target.next_owner_epoch();
692        let transfer = source
693            .prepare_outgoing_handoff(
694                EntityId::new(9),
695                StationId::new(2),
696                target_epoch,
697                Tick::new(8),
698            )
699            .expect("prepare should work");
700
701        let ghost_handle = target
702            .prewarm_handoff_ghost(&transfer)
703            .expect("prewarm should work");
704        assert!(!target.get(ghost_handle).expect("ghost exists").is_owned());
705
706        let owner_handle = target
707            .commit_incoming_handoff(transfer.clone())
708            .expect("incoming commit should work");
709        assert!(target.get(owner_handle).expect("owner exists").is_owned());
710
711        let source_handle = source
712            .commit_outgoing_handoff(&transfer)
713            .expect("outgoing commit should work");
714        let source_record = source.get(source_handle).expect("source ghost exists");
715        assert!(!source_record.is_owned());
716        assert_eq!(source_record.role.owner_epoch(), target_epoch);
717    }
718}
sectorsync_core/station.rs

sectorsync_core/
station.rs
