loam_runtime/
store.rs

1use std::any::type_name;
2use std::fmt;
3use std::ops::Range;
4use std::sync::atomic::{AtomicU64, Ordering};
5
6use crate::entity::{Entities, Entity, EntityKey, SceneId};
7use crate::relation::LinkId;
8use crate::session::Stamp;
9
10#[derive(Clone, Copy, Debug, PartialEq, Eq)]
11pub enum StoreError {
12    Foreign(Entity),
13    Stale(Entity),
14    Occupied(Entity),
15    Missing(Entity),
16    Unlinked(LinkId),
17    Capacity,
18}
19
20impl fmt::Display for StoreError {
21    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
22        match self {
23            Self::Foreign(entity) => write!(f, "{entity} belongs to another session"),
24            Self::Stale(entity) => write!(f, "{entity} is stale"),
25            Self::Occupied(entity) => write!(f, "{entity} already has a row"),
26            Self::Missing(entity) => write!(f, "{entity} has no row"),
27            Self::Unlinked(link) => write!(f, "{link} is unlinked"),
28            Self::Capacity => f.write_str("the store is full"),
29        }
30    }
31}
32
33#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, PartialOrd, Ord)]
34pub struct Version(u64);
35
36impl Version {
37    pub fn bump(self) -> Self {
38        Self(self.0.wrapping_add(1))
39    }
40}
41
42/// Retained events; independent of live row count.
43#[derive(Clone, Copy, Debug, PartialEq, Eq)]
44pub struct LogCapacity {
45    pub dirty: usize,
46    pub removals: usize,
47}
48
49pub const DEFAULT_LOG_CAPACITY: LogCapacity = LogCapacity {
50    dirty: 4096,
51    removals: 1024,
52};
53
54impl Default for LogCapacity {
55    fn default() -> Self {
56        DEFAULT_LOG_CAPACITY
57    }
58}
59
60const CURSOR_EXPIRY_BOUNDARIES: u64 = 8;
61
62/// A consumer's position in a store's logs; the default cursor is stale and forces a resync.
63#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
64pub struct Cursor {
65    position: u64,
66    removed: u64,
67    boundary: u64,
68    synced: bool,
69}
70
71impl Cursor {
72    pub fn position(self) -> u64 {
73        self.position
74    }
75}
76
77#[derive(Clone, Copy, Debug, PartialEq, Eq)]
78pub struct Removal {
79    pub entity: Entity,
80    pub version: Version,
81}
82
83pub enum Change<'a, T> {
84    Row(Entity, &'a T),
85    Removed(Removal),
86}
87
88/// Deltas since the cursor, or every live row and retained removal after a resync.
89pub struct Changes<'a, T> {
90    store: &'a Store<T>,
91    resync: bool,
92    removals: Range<u64>,
93    dirty: Range<u64>,
94    rows: Range<usize>,
95}
96
97impl<T> Changes<'_, T> {
98    pub fn is_resync(&self) -> bool {
99        self.resync
100    }
101}
102
103impl<'a, T> Iterator for Changes<'a, T> {
104    type Item = Change<'a, T>;
105
106    fn next(&mut self) -> Option<Self::Item> {
107        let store = self.store;
108        if let Some(tracking) = &store.tracking {
109            for seq in &mut self.removals {
110                if let Some(removal) = tracking.removals.get(seq) {
111                    return Some(Change::Removed(removal));
112                }
113            }
114            for seq in &mut self.dirty {
115                let Some(key) = tracking.dirty.get(seq) else {
116                    continue;
117                };
118                let Some(dense) = store.dense_of(key) else {
119                    continue;
120                };
121                if tracking.logged[dense] == seq {
122                    return Some(Change::Row(
123                        Entity::new(store.scene, key),
124                        &store.dense[dense],
125                    ));
126                }
127            }
128        }
129        let dense = self.rows.next()?;
130        Some(Change::Row(
131            Entity::new(store.scene, store.keys[dense]),
132            &store.dense[dense],
133        ))
134    }
135}
136
137/// Disjoint mutable row ranges of one store, checked once when built.
138pub struct Partition<'a, T> {
139    rows: &'a mut [T],
140    cuts: &'a [usize],
141}
142
143impl<'a, T> Partition<'a, T> {
144    pub fn len(&self) -> usize {
145        self.rows.len()
146    }
147
148    pub fn is_empty(&self) -> bool {
149        self.rows.is_empty()
150    }
151
152    pub fn part_count(&self) -> usize {
153        self.cuts.len() + 1
154    }
155
156    pub fn parts(self) -> Parts<'a, T> {
157        Parts {
158            rows: self.rows,
159            cuts: self.cuts,
160            first: 0,
161            done: false,
162        }
163    }
164}
165
166pub struct Part<'a, T> {
167    pub first: usize,
168    pub rows: &'a mut [T],
169}
170
171pub struct Parts<'a, T> {
172    rows: &'a mut [T],
173    cuts: &'a [usize],
174    first: usize,
175    done: bool,
176}
177
178impl<'a, T> Iterator for Parts<'a, T> {
179    type Item = Part<'a, T>;
180
181    fn next(&mut self) -> Option<Self::Item> {
182        if self.done {
183            return None;
184        }
185        let rows = std::mem::take(&mut self.rows);
186        let first = self.first;
187        match self.cuts.split_first() {
188            Some((&cut, rest)) => {
189                let (head, tail) = rows.split_at_mut(cut - first);
190                self.rows = tail;
191                self.cuts = rest;
192                self.first = cut;
193                Some(Part { first, rows: head })
194            }
195            None => {
196                self.done = true;
197                Some(Part { first, rows })
198            }
199        }
200    }
201}
202
203#[derive(Clone, Copy, Debug, PartialEq, Eq)]
204pub enum PartitionError {
205    Unsorted { at: usize },
206    OutOfRange { cut: usize, len: usize },
207}
208
209const NO_RECORD: u32 = u32::MAX;
210
211pub struct RecordBuffer<R> {
212    rows: Vec<R>,
213    entities: Vec<Entity>,
214    positions: Vec<u32>,
215    stamp: Stamp,
216}
217
218impl<R> RecordBuffer<R> {
219    pub fn rows(&self) -> &[R] {
220        &self.rows
221    }
222
223    pub fn stamp(&self) -> Stamp {
224        self.stamp
225    }
226
227    fn position(&self, entity: Entity) -> Option<usize> {
228        let position = *self.positions.get(entity.key().slot() as usize)?;
229        (position != NO_RECORD && self.entities[position as usize] == entity)
230            .then_some(position as usize)
231    }
232
233    fn upsert(&mut self, entity: Entity, record: R) {
234        if let Some(position) = self.position(entity) {
235            self.rows[position] = record;
236            return;
237        }
238        let slot = entity.key().slot() as usize;
239        if slot >= self.positions.len() {
240            self.positions.resize(slot + 1, NO_RECORD);
241        }
242        self.positions[slot] = self.rows.len() as u32;
243        self.rows.push(record);
244        self.entities.push(entity);
245    }
246
247    fn clear(&mut self) {
248        self.rows.clear();
249        self.entities.clear();
250        self.positions.fill(NO_RECORD);
251    }
252
253    pub(crate) fn restamp(&mut self, stamp: Stamp) {
254        self.stamp = stamp;
255    }
256
257    pub(crate) fn update(&mut self, entity: Entity, record: R) -> bool {
258        let Some(position) = self.position(entity) else {
259            return false;
260        };
261        self.rows[position] = record;
262        true
263    }
264
265    pub(crate) fn replace(&mut self, records: impl Iterator<Item = (Entity, R)>, stamp: Stamp) {
266        self.clear();
267        for (entity, record) in records {
268            self.upsert(entity, record);
269        }
270        self.stamp = stamp;
271    }
272}
273
274impl<R> Default for RecordBuffer<R> {
275    fn default() -> Self {
276        Self {
277            rows: Vec::new(),
278            entities: Vec::new(),
279            positions: Vec::new(),
280            stamp: Stamp::default(),
281        }
282    }
283}
284
285#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
286pub struct SchemaId(&'static str);
287
288impl SchemaId {
289    pub fn of<T: 'static>() -> Self {
290        Self(type_name::<T>())
291    }
292
293    pub fn name(self) -> &'static str {
294        self.0
295    }
296}
297
298#[derive(Clone, Copy)]
299pub struct Owner {
300    _private: (),
301}
302
303impl Owner {
304    pub(crate) fn new() -> Self {
305        Self { _private: () }
306    }
307}
308
309/// A storage-field contract; only `Stores` implementations call its lifetime hooks.
310pub trait StoreField: Send + 'static {
311    type Snapshot: Send + 'static;
312
313    fn bind(&mut self, scene: SceneId, owner: Owner);
314
315    fn snapshot(&self) -> Self::Snapshot;
316
317    fn restore(&mut self, from: &Self::Snapshot, scene: SceneId, owner: Owner);
318
319    fn boundary(&mut self, _owner: Owner) {}
320
321    fn release(&mut self, _entity: Entity, _owner: Owner) {}
322}
323
324pub struct StoreSnapshot<T> {
325    rows: Vec<T>,
326    keys: Vec<EntityKey>,
327}
328
329impl<T> StoreSnapshot<T> {
330    pub fn rows(&self) -> &[T] {
331        &self.rows
332    }
333
334    pub fn keys(&self) -> &[EntityKey] {
335        &self.keys
336    }
337}
338
339#[derive(Clone, Copy)]
340struct SparseEntry {
341    generation: u32,
342    dense: u32,
343}
344
345struct Ring<E> {
346    entries: Vec<E>,
347    capacity: usize,
348    pushed: u64,
349}
350
351impl<E: Copy> Ring<E> {
352    fn new(capacity: usize) -> Self {
353        Self {
354            entries: Vec::with_capacity(capacity),
355            capacity,
356            pushed: 0,
357        }
358    }
359
360    fn push(&mut self, entry: E) -> u64 {
361        if self.entries.len() < self.capacity {
362            self.entries.push(entry);
363        } else if self.capacity > 0 {
364            self.entries[(self.pushed % self.capacity as u64) as usize] = entry;
365        }
366        self.pushed += 1;
367        self.pushed
368    }
369
370    fn get(&self, seq: u64) -> Option<E> {
371        let index = seq.checked_sub(1)? % self.capacity.max(1) as u64;
372        self.entries.get(index as usize).copied()
373    }
374
375    fn retained(&self) -> Range<u64> {
376        self.pushed - self.entries.len() as u64 + 1..self.pushed + 1
377    }
378
379    fn lost(&self, consumed: u64) -> bool {
380        consumed + 1 < self.retained().start
381    }
382
383    fn reset(&mut self) {
384        self.entries.clear();
385        let stride = self.capacity.max(1) as u64;
386        self.pushed = self.pushed / stride * stride + stride;
387    }
388}
389
390struct Tracking {
391    capacity: LogCapacity,
392    versions: Vec<Version>,
393    logged: Vec<u64>,
394    dirty: Ring<EntityKey>,
395    removals: Ring<Removal>,
396    last_read: AtomicU64,
397    boundary: u64,
398}
399
400impl Tracking {
401    fn new(capacity: LogCapacity) -> Self {
402        Self {
403            capacity,
404            versions: Vec::new(),
405            logged: Vec::new(),
406            dirty: Ring::new(capacity.dirty),
407            removals: Ring::new(capacity.removals),
408            last_read: AtomicU64::new(0),
409            boundary: 0,
410        }
411    }
412
413    fn push_row(&mut self) {
414        self.versions.push(Version::default());
415        self.logged.push(0);
416    }
417
418    fn touch(&mut self, dense: usize, key: EntityKey) {
419        self.versions[dense] = self.versions[dense].bump();
420        if self.logged[dense] > self.last_read.load(Ordering::Relaxed) {
421            return;
422        }
423        self.logged[dense] = self.dirty.push(key);
424    }
425
426    fn touch_all(&mut self, keys: &[EntityKey]) {
427        let last_read = self.last_read.load(Ordering::Relaxed);
428        for (dense, &key) in keys.iter().enumerate() {
429            self.versions[dense] = self.versions[dense].bump();
430            if self.logged[dense] <= last_read {
431                self.logged[dense] = self.dirty.push(key);
432            }
433        }
434    }
435
436    fn remove(&mut self, dense: usize, entity: Entity) {
437        let version = self.versions.swap_remove(dense);
438        self.logged.swap_remove(dense);
439        self.removals.push(Removal { entity, version });
440    }
441
442    fn stale(&self, cursor: &Cursor) -> bool {
443        !cursor.synced
444            || self.boundary.saturating_sub(cursor.boundary) > CURSOR_EXPIRY_BOUNDARIES
445            || self.dirty.lost(cursor.position)
446            || self.removals.lost(cursor.removed)
447    }
448
449    fn reset(&mut self, rows: usize) {
450        self.versions.clear();
451        self.versions.resize(rows, Version::default());
452        self.logged.clear();
453        self.logged.resize(rows, 0);
454        self.dirty.reset();
455        self.removals.reset();
456    }
457}
458
459pub struct Store<T> {
460    scene: SceneId,
461    dense: Vec<T>,
462    keys: Vec<EntityKey>,
463    sparse: Vec<Option<SparseEntry>>,
464    tracking: Option<Tracking>,
465}
466
467impl<T> Default for Store<T> {
468    fn default() -> Self {
469        Self::untracked()
470    }
471}
472
473impl<T> Store<T> {
474    pub fn untracked() -> Self {
475        Self::with_tracking(None)
476    }
477
478    pub fn tracked(capacity: LogCapacity) -> Self {
479        Self::with_tracking(Some(Tracking::new(capacity)))
480    }
481
482    fn with_tracking(tracking: Option<Tracking>) -> Self {
483        Self {
484            scene: SceneId::UNBOUND,
485            dense: Vec::new(),
486            keys: Vec::new(),
487            sparse: Vec::new(),
488            tracking,
489        }
490    }
491
492    pub(crate) fn bind(&mut self, scene: SceneId) {
493        self.scene = scene;
494    }
495
496    pub fn is_tracked(&self) -> bool {
497        self.tracking.is_some()
498    }
499
500    pub fn log_capacity(&self) -> Option<LogCapacity> {
501        self.tracking.as_ref().map(|tracking| tracking.capacity)
502    }
503
504    pub fn scene(&self) -> SceneId {
505        self.scene
506    }
507
508    pub fn len(&self) -> usize {
509        self.dense.len()
510    }
511
512    pub fn is_empty(&self) -> bool {
513        self.dense.is_empty()
514    }
515
516    pub fn contains(&self, entity: Entity) -> bool {
517        self.dense_index(entity).is_some()
518    }
519
520    pub(crate) fn dense_versions(&self) -> Option<(&[EntityKey], &[Version])> {
521        let tracking = self.tracking.as_ref()?;
522        Some((&self.keys, &tracking.versions))
523    }
524
525    /// Stable between structural boundaries and meaningless across them.
526    pub fn dense_index(&self, entity: Entity) -> Option<usize> {
527        if entity.scene() != self.scene {
528            return None;
529        }
530        self.dense_of(entity.key())
531    }
532
533    fn dense_of(&self, key: EntityKey) -> Option<usize> {
534        let entry = (*self.sparse.get(key.slot() as usize)?)?;
535        (entry.generation == key.generation()).then_some(entry.dense as usize)
536    }
537
538    pub fn entity_at(&self, dense: usize) -> Option<Entity> {
539        self.keys
540            .get(dense)
541            .map(|&key| Entity::new(self.scene, key))
542    }
543
544    pub fn get(&self, entity: Entity) -> Option<&T> {
545        self.dense_index(entity).map(|dense| &self.dense[dense])
546    }
547
548    pub fn get_mut(&mut self, entity: Entity) -> Option<&mut T> {
549        let dense = self.dense_index(entity)?;
550        if let Some(tracking) = &mut self.tracking {
551            tracking.touch(dense, self.keys[dense]);
552        }
553        Some(&mut self.dense[dense])
554    }
555
556    /// Two disjoint mutable rows of one store; equal handles yield `None`.
557    pub fn pair_mut(&mut self, a: Entity, b: Entity) -> Option<(&mut T, &mut T)> {
558        let i = self.dense_index(a)?;
559        let j = self.dense_index(b)?;
560        if i == j {
561            return None;
562        }
563        if let Some(tracking) = &mut self.tracking {
564            tracking.touch(i, self.keys[i]);
565            tracking.touch(j, self.keys[j]);
566        }
567        let (low, high) = self.dense.split_at_mut(i.max(j));
568        if i < j {
569            Some((&mut low[i], &mut high[0]))
570        } else {
571            Some((&mut high[0], &mut low[j]))
572        }
573    }
574
575    pub fn rows(&self) -> &[T] {
576        &self.dense
577    }
578
579    pub(crate) fn rows_mut_untracked(&mut self) -> &mut [T] {
580        &mut self.dense
581    }
582
583    pub fn iter(&self) -> impl Iterator<Item = (Entity, &T)> {
584        let scene = self.scene;
585        self.keys
586            .iter()
587            .zip(&self.dense)
588            .map(move |(&key, row)| (Entity::new(scene, key), row))
589    }
590
591    pub fn iter_mut(&mut self) -> impl Iterator<Item = (Entity, &mut T)> {
592        if let Some(tracking) = &mut self.tracking {
593            tracking.touch_all(&self.keys);
594        }
595        let scene = self.scene;
596        self.keys
597            .iter()
598            .zip(&mut self.dense)
599            .map(move |(&key, row)| (Entity::new(scene, key), row))
600    }
601
602    pub fn insert(
603        &mut self,
604        entities: &Entities,
605        entity: Entity,
606        row: T,
607    ) -> Result<(), StoreError> {
608        if entity.scene() != entities.scene() {
609            return Err(StoreError::Foreign(entity));
610        }
611        if entities.resolve(entity).is_none() {
612            return Err(StoreError::Stale(entity));
613        }
614        if self.scene == SceneId::UNBOUND {
615            self.scene = entities.scene();
616        }
617        self.insert_raw(entity, row)
618    }
619
620    pub(crate) fn insert_raw(&mut self, entity: Entity, row: T) -> Result<(), StoreError> {
621        if entity.scene() != self.scene {
622            return Err(StoreError::Foreign(entity));
623        }
624        let key = entity.key();
625        let slot = key.slot() as usize;
626        if slot >= self.sparse.len() {
627            self.sparse.resize(slot + 1, None);
628        }
629        if self.sparse[slot].is_some() {
630            return Err(StoreError::Occupied(entity));
631        }
632        let dense = self.dense.len();
633        let dense32 = u32::try_from(dense).map_err(|_| StoreError::Capacity)?;
634        self.sparse[slot] = Some(SparseEntry {
635            generation: key.generation(),
636            dense: dense32,
637        });
638        self.dense.push(row);
639        self.keys.push(key);
640        if let Some(tracking) = &mut self.tracking {
641            tracking.push_row();
642            tracking.touch(dense, key);
643        }
644        Ok(())
645    }
646
647    pub fn remove(&mut self, entity: Entity) -> Result<T, StoreError> {
648        let dense = self
649            .dense_index(entity)
650            .ok_or(StoreError::Missing(entity))?;
651        let key = entity.key();
652        self.sparse[key.slot() as usize] = None;
653        let row = self.dense.swap_remove(dense);
654        self.keys.swap_remove(dense);
655        if let Some(&moved) = self.keys.get(dense) {
656            if let Some(entry) = &mut self.sparse[moved.slot() as usize] {
657                entry.dense = dense as u32;
658            }
659        }
660        if let Some(tracking) = &mut self.tracking {
661            tracking.remove(dense, entity);
662        }
663        Ok(row)
664    }
665
666    /// `cuts` must be strictly ascending and within `len`.
667    pub fn partition<'a>(
668        &'a mut self,
669        cuts: &'a [usize],
670    ) -> Result<Partition<'a, T>, PartitionError> {
671        let len = self.dense.len();
672        let mut previous = None;
673        for (at, &cut) in cuts.iter().enumerate() {
674            if cut > len {
675                return Err(PartitionError::OutOfRange { cut, len });
676            }
677            if previous.is_some_and(|previous| cut <= previous) {
678                return Err(PartitionError::Unsorted { at });
679            }
680            previous = Some(cut);
681        }
682        if let Some(tracking) = &mut self.tracking {
683            tracking.touch_all(&self.keys);
684        }
685        Ok(Partition {
686            rows: &mut self.dense,
687            cuts,
688        })
689    }
690
691    /// Counts toward cursor expiry; the session calls it once per boundary.
692    pub fn boundary(&mut self) {
693        if let Some(tracking) = &mut self.tracking {
694            tracking.boundary += 1;
695        }
696    }
697
698    /// `None` for an untracked store or an entity it does not hold.
699    pub fn version(&self, entity: Entity) -> Option<Version> {
700        let dense = self.dense_index(entity)?;
701        self.tracking
702            .as_ref()
703            .map(|tracking| tracking.versions[dense])
704    }
705
706    pub fn changes(&self, cursor: &mut Cursor) -> Changes<'_, T> {
707        let Some(tracking) = &self.tracking else {
708            *cursor = Cursor {
709                synced: true,
710                ..Cursor::default()
711            };
712            return Changes {
713                store: self,
714                resync: true,
715                removals: 0..0,
716                dirty: 0..0,
717                rows: 0..self.dense.len(),
718            };
719        };
720        let resync = tracking.stale(cursor);
721        let removals = if resync {
722            tracking.removals.retained()
723        } else {
724            cursor.removed + 1..tracking.removals.pushed + 1
725        };
726        let dirty = if resync {
727            0..0
728        } else {
729            cursor.position + 1..tracking.dirty.pushed + 1
730        };
731        let rows = if resync { 0..self.dense.len() } else { 0..0 };
732        *cursor = Cursor {
733            position: tracking.dirty.pushed,
734            removed: tracking.removals.pushed,
735            boundary: tracking.boundary,
736            synced: true,
737        };
738        tracking
739            .last_read
740            .fetch_max(tracking.dirty.pushed, Ordering::Relaxed);
741        Changes {
742            store: self,
743            resync,
744            removals,
745            dirty,
746            rows,
747        }
748    }
749
750    /// An untracked store or an expired cursor counts as changed; a clean check renews the cursor's expiry.
751    pub fn changed_since(&self, cursor: &mut Cursor) -> bool {
752        let Some(tracking) = &self.tracking else {
753            return true;
754        };
755        if tracking.stale(cursor)
756            || cursor.position != tracking.dirty.pushed
757            || cursor.removed != tracking.removals.pushed
758        {
759            return true;
760        }
761        cursor.boundary = tracking.boundary;
762        false
763    }
764
765    pub fn catch_up(&self, cursor: &mut Cursor) {
766        let _ = self.changes(cursor);
767    }
768}
769
770impl<T: Clone + Send + 'static> StoreField for Store<T> {
771    type Snapshot = StoreSnapshot<T>;
772
773    fn bind(&mut self, scene: SceneId, _owner: Owner) {
774        Store::bind(self, scene);
775    }
776
777    fn snapshot(&self) -> StoreSnapshot<T> {
778        StoreSnapshot {
779            rows: self.dense.clone(),
780            keys: self.keys.clone(),
781        }
782    }
783
784    fn restore(&mut self, from: &StoreSnapshot<T>, scene: SceneId, _owner: Owner) {
785        self.scene = scene;
786        self.dense.clone_from(&from.rows);
787        self.keys.clone_from(&from.keys);
788        let slots = self
789            .keys
790            .iter()
791            .map(|key| key.slot() as usize + 1)
792            .max()
793            .unwrap_or(0);
794        self.sparse.clear();
795        self.sparse.resize(slots, None);
796        for (dense, key) in self.keys.iter().enumerate() {
797            self.sparse[key.slot() as usize] = Some(SparseEntry {
798                generation: key.generation(),
799                dense: dense as u32,
800            });
801        }
802        if let Some(tracking) = &mut self.tracking {
803            tracking.reset(self.dense.len());
804        }
805    }
806
807    fn boundary(&mut self, _owner: Owner) {
808        Store::boundary(self);
809    }
810
811    fn release(&mut self, entity: Entity, _owner: Owner) {
812        let _ = Store::remove(self, entity);
813    }
814}
815
816#[cfg(test)]
817pub(crate) mod tests {
818    use std::alloc::System;
819    use std::collections::BTreeMap;
820
821    use loam_time::alloc::{bytes_allocated_by, CountingAllocator};
822
823    use super::*;
824    use crate::entity::{Entities, Epoch, RuntimeId};
825    use crate::relation::Relation;
826
827    #[global_allocator]
828    static COUNTING_ALLOCATOR: CountingAllocator<System> = CountingAllocator::new(System);
829
830    const SMALL: LogCapacity = LogCapacity {
831        dirty: 4,
832        removals: 2,
833    };
834
835    fn entities() -> Entities {
836        Entities::new(SceneId {
837            runtime: RuntimeId::allocate(),
838            epoch: Epoch::default(),
839        })
840    }
841
842    fn filled(count: u32, capacity: Option<LogCapacity>) -> (Entities, Store<u32>, Vec<Entity>) {
843        let entities = entities();
844        let mut store = capacity.map_or_else(Store::untracked, Store::tracked);
845        StoreField::bind(&mut store, entities.scene(), Owner::new());
846        let mut entities = entities;
847        let spawned: Vec<Entity> = (0..count)
848            .map(|value| {
849                let entity = entities.spawn();
850                store.insert(&entities, entity, value).unwrap();
851                entity
852            })
853            .collect();
854        (entities, store, spawned)
855    }
856
857    #[derive(Default)]
858    struct Mirror {
859        rows: BTreeMap<Entity, u32>,
860        cursor: Cursor,
861    }
862
863    impl Mirror {
864        fn sync(&mut self, store: &Store<u32>) -> (bool, Vec<Entity>) {
865            let changes = store.changes(&mut self.cursor);
866            let resync = changes.is_resync();
867            if resync {
868                self.rows.clear();
869            }
870            let mut seen = Vec::new();
871            for change in changes {
872                match change {
873                    Change::Row(entity, &value) => {
874                        seen.push(entity);
875                        self.rows.insert(entity, value);
876                    }
877                    Change::Removed(removal) => {
878                        self.rows.remove(&removal.entity);
879                    }
880                }
881            }
882            (resync, seen)
883        }
884    }
885
886    fn live(store: &Store<u32>) -> BTreeMap<Entity, u32> {
887        store
888            .iter()
889            .map(|(entity, &value)| (entity, value))
890            .collect()
891    }
892
893    #[test]
894    fn pair_mut_and_partition_never_alias_and_never_refuse_distinct_rows() {
895        let (_, mut store, e) = filled(5, None);
896        assert!(store.pair_mut(e[1], e[1]).is_none());
897        let (a, b) = store.pair_mut(e[3], e[1]).unwrap();
898        *a += 10;
899        *b += 20;
900        assert_eq!(store.rows(), [0, 21, 2, 13, 4]);
901
902        assert_eq!(
903            store.partition(&[1, 6]).err(),
904            Some(PartitionError::OutOfRange { cut: 6, len: 5 })
905        );
906        assert_eq!(
907            store.partition(&[2, 2]).err(),
908            Some(PartitionError::Unsorted { at: 1 })
909        );
910        let partition = store.partition(&[1, 3]).unwrap();
911        assert_eq!(partition.part_count(), 3);
912        let mut covered = Vec::new();
913        for (index, part) in partition.parts().enumerate() {
914            for (offset, row) in part.rows.iter_mut().enumerate() {
915                *row = index as u32;
916                covered.push(part.first + offset);
917            }
918        }
919        assert_eq!(covered, [0, 1, 2, 3, 4]);
920        assert_eq!(store.rows(), [0, 1, 1, 2, 2]);
921    }
922
923    #[test]
924    fn swap_remove_keeps_every_survivor_once_in_iteration_and_the_sparse_index() {
925        let (_, mut store, e) = filled(5, None);
926        assert_eq!(store.remove(e[1]), Ok(1));
927        assert_eq!(store.remove(e[1]), Err(StoreError::Missing(e[1])));
928        let mut seen: Vec<u32> = store.iter().map(|(_, &value)| value).collect();
929        seen.sort_unstable();
930        assert_eq!(seen, [0, 2, 3, 4]);
931        for dense in 0..store.len() {
932            let entity = store.entity_at(dense).unwrap();
933            assert_eq!(store.dense_index(entity), Some(dense));
934            assert_eq!(store.get(entity), Some(&store.rows()[dense]));
935        }
936        assert_eq!(store.get(e[4]), Some(&4));
937        assert!(!store.contains(e[1]));
938    }
939
940    #[test]
941    fn deleting_the_last_rows_keeps_their_removals() {
942        let retaining = LogCapacity {
943            dirty: SMALL.dirty,
944            removals: 3,
945        };
946        let (_, mut store, e) = filled(3, Some(retaining));
947        let mut mirror = Mirror::default();
948        mirror.sync(&store);
949        assert_eq!(mirror.rows, live(&store));
950        for &entity in [e[2], e[0], e[1]].iter() {
951            store.remove(entity).unwrap();
952        }
953        let mut cursor = mirror.cursor;
954        let changes = store.changes(&mut cursor);
955        assert!(!changes.is_resync());
956        let mut removed: Vec<Entity> = changes
957            .map(|change| match change {
958                Change::Removed(removal) => removal.entity,
959                Change::Row(entity, _) => panic!("{entity:?} was published after its removal"),
960            })
961            .collect();
962        removed.sort();
963        assert_eq!(removed, e);
964        assert!(store.is_empty());
965    }
966
967    #[test]
968    fn churn_or_cursor_overflow_forces_a_resync_that_rebuilds_the_consumer() {
969        let (mut entities, mut store, e) = filled(3, Some(SMALL));
970        let mut mirror = Mirror::default();
971        assert!(mirror.sync(&store).0);
972        for _ in 0..3 {
973            store.boundary();
974        }
975        *store.get_mut(e[1]).unwrap() = 11;
976        *store.get_mut(e[1]).unwrap() = 12;
977        let (resync, seen) = mirror.sync(&store);
978        assert!(!resync);
979        assert_eq!(seen, [e[1]]);
980        assert_eq!(mirror.rows, live(&store));
981
982        for value in 0..SMALL.dirty as u32 + 1 {
983            let entity = entities.spawn();
984            store.insert(&entities, entity, 100 + value).unwrap();
985        }
986        store.remove(e[0]).unwrap();
987        assert!(mirror.sync(&store).0);
988        assert_eq!(mirror.rows, live(&store));
989
990        for _ in 0..=CURSOR_EXPIRY_BOUNDARIES {
991            store.boundary();
992        }
993        assert!(mirror.sync(&store).0);
994        store.boundary();
995        assert!(!mirror.sync(&store).0);
996
997        for &entity in &e[1..] {
998            store.remove(entity).unwrap();
999        }
1000        store.remove(store.entity_at(0).unwrap()).unwrap();
1001        assert!(mirror.sync(&store).0);
1002        assert_eq!(mirror.rows, live(&store));
1003    }
1004
1005    #[test]
1006    fn old_generation_removal_never_deletes_a_reused_slots_new_object() {
1007        let (mut entities, mut store, e) = filled(1, Some(SMALL));
1008        let mut mirror = Mirror::default();
1009        mirror.sync(&store);
1010        store.remove(e[0]).unwrap();
1011        entities.despawn(e[0]).unwrap();
1012        let reused = entities.spawn();
1013        assert_eq!(reused.key().slot(), e[0].key().slot());
1014        store.insert(&entities, reused, 7).unwrap();
1015
1016        let mut cursor = mirror.cursor;
1017        let removals: Vec<Removal> = store
1018            .changes(&mut cursor)
1019            .filter_map(|change| match change {
1020                Change::Removed(removal) => Some(removal),
1021                Change::Row(..) => None,
1022            })
1023            .collect();
1024        assert_eq!(removals.len(), 1);
1025        assert_eq!(removals[0].entity, e[0]);
1026        assert_ne!(removals[0].entity, reused);
1027        mirror.sync(&store);
1028        assert_eq!(mirror.rows.get(&reused), Some(&7));
1029        assert_eq!(mirror.rows.len(), 1);
1030    }
1031
1032    #[test]
1033    fn warm_read_iterate_pair_and_partition_paths_allocate_nothing() {
1034        let (_, mut store, e) = filled(64, Some(LogCapacity::default()));
1035        let mut cursor = Cursor::default();
1036        let mut warm = |store: &mut Store<u32>| {
1037            for (_, row) in store.iter_mut() {
1038                *row += 1;
1039            }
1040            store.changes(&mut cursor).count()
1041        };
1042        warm(&mut store);
1043        assert_eq!(warm(&mut store), 64);
1044
1045        let bytes = bytes_allocated_by(|| {
1046            for _ in 0..16 {
1047                let mut sum = 0u32;
1048                for &entity in &e {
1049                    sum = sum.wrapping_add(*store.get(entity).unwrap());
1050                }
1051                for (_, row) in store.iter() {
1052                    sum = sum.wrapping_add(*row);
1053                }
1054                let (a, b) = store.pair_mut(e[0], e[63]).unwrap();
1055                *a = sum;
1056                *b += 1;
1057                for part in store.partition(&[16, 48]).unwrap().parts() {
1058                    for row in part.rows {
1059                        *row += 1;
1060                    }
1061                }
1062                warm(&mut store);
1063            }
1064        })
1065        .expect("the counting allocator is installed");
1066        assert_eq!(
1067            bytes, 0,
1068            "16 warmed passes asked the allocator for {bytes} bytes"
1069        );
1070    }
1071
1072    #[test]
1073    fn edits_and_removals_after_a_restore_reach_a_resynchronized_cursor() {
1074        const RETAINING: LogCapacity = LogCapacity {
1075            dirty: 3,
1076            removals: 2,
1077        };
1078        let (mut entities, mut store, e) = filled(3, Some(RETAINING));
1079        let mut mirror = Mirror::default();
1080        mirror.sync(&store);
1081        let (snapshot, rows) = (entities.snapshot(), store.snapshot());
1082
1083        entities.restore(&snapshot);
1084        let scene = entities.scene();
1085        StoreField::restore(&mut store, &rows, scene, Owner::new());
1086        let rebased: Vec<Entity> = e
1087            .iter()
1088            .map(|entity| Entity::new(scene, entity.key()))
1089            .collect();
1090        assert!(mirror.sync(&store).0, "the restore never forced a resync");
1091
1092        *store.get_mut(rebased[0]).unwrap() = 40;
1093        *store.get_mut(rebased[1]).unwrap() = 41;
1094        store.remove(rebased[2]).unwrap();
1095        let (resync, seen) = mirror.sync(&store);
1096        assert!(!resync, "a second resync hid the lost log entries");
1097        assert_eq!(seen, [rebased[0], rebased[1]]);
1098        assert_eq!(mirror.rows, live(&store));
1099    }
1100
1101    #[test]
1102    fn restore_advances_the_epoch_so_old_handles_fail_and_relations_survive() {
1103        let (mut entities, mut store, e) = filled(2, Some(SMALL));
1104        let mut relation = Relation::<u8>::new();
1105        StoreField::bind(&mut relation, entities.scene(), Owner::new());
1106        let link = relation.link(&entities, e[0], e[1], 5).unwrap();
1107        let mut mirror = Mirror::default();
1108        mirror.sync(&store);
1109        let (snapshot, rows, links) = (entities.snapshot(), store.snapshot(), relation.snapshot());
1110
1111        let extra = entities.spawn();
1112        store.insert(&entities, extra, 9).unwrap();
1113        relation.unlink(link).unwrap();
1114        *store.get_mut(e[0]).unwrap() = 8;
1115
1116        entities.restore(&snapshot);
1117        let scene = entities.scene();
1118        assert_eq!(scene.epoch(), Epoch::default().advance());
1119        StoreField::restore(&mut store, &rows, scene, Owner::new());
1120        StoreField::restore(&mut relation, &links, scene, Owner::new());
1121
1122        assert_eq!(store.scene(), scene);
1123        assert_eq!(store.get(e[0]), None);
1124        assert_eq!(store.get(extra), None);
1125        assert_eq!(relation.outgoing(e[0]).count(), 0);
1126        assert!(relation.get(link).is_none());
1127
1128        let rebased: Vec<Entity> = e
1129            .iter()
1130            .map(|entity| Entity::new(scene, entity.key()))
1131            .collect();
1132        assert_eq!(store.get(rebased[0]), Some(&0));
1133        assert_eq!(store.len(), 2);
1134        let survivor = relation.outgoing(rebased[0]).next().unwrap();
1135        let survivor = relation.get(survivor).unwrap();
1136        assert_eq!(
1137            (survivor.from(), survivor.to(), survivor.data),
1138            (rebased[0], rebased[1], 5)
1139        );
1140        assert_eq!(store.get(survivor.to()), Some(&1));
1141        assert!(mirror.sync(&store).0);
1142        assert_eq!(mirror.rows, live(&store));
1143    }
1144}
loam_runtime/store.rs

loam_runtime/
store.rs
