arkhe_kernel/runtime/
stage.rs

1//! `StepStage` — transactional COW staging for one `step()` (10 buckets).
2//!
3//! Every commit-conditional write the kernel performs during a step is
4//! buffered here; on commit, `Instance::apply_stage` drains in canonical
5//! canonical order; on rollback, the stage is dropped without effect.
6//!
7//! Buckets (10):
8//! 1. `state_ops` — entity/component mutations
9//! 2. `events` — KernelEvent emissions (kernel-level drain)
10//! 3. `schedule_deltas` — scheduler add/cancel
11//! 4. `pending_signals` — outbound IPC (kernel routes post-commit)
12//! 5. `id_counters` — monotonic ID counter advances
13//! 6. `ledger_delta` — ResourceLedger updates
14//! 7. `inflight_refs_delta` — drain-refcount per RouteId (signed delta)
15//! 8. `wall_remainder_delta` — sub-tick time accumulator advance
16//! 9. `local_tick_delta` — logical tick advance
17//! 10. `observer_eviction_pending` — observers slated for eviction
18
19use std::collections::{BTreeMap, VecDeque};
20
21use bytes::Bytes;
22use serde::{Deserialize, Serialize};
23
24use crate::abi::{EntityId, InstanceId, Principal, RouteId, TypeCode};
25use crate::state::{EntityMeta, ScheduledActionId, ScheduledEntry};
26
27use super::event::{KernelEvent, ObserverHandle};
28
29#[derive(Debug, Default, Clone, Serialize, Deserialize)]
30pub(crate) struct StepStage {
31    pub state_ops: Vec<StagedStateDelta>,
32    pub events: VecDeque<KernelEvent>,
33    pub schedule_deltas: Vec<ScheduledEntryDelta>,
34    pub pending_signals: Vec<PendingSignal>,
35    pub id_counters: IdCountersDelta,
36    pub ledger_delta: ResourceLedgerDelta,
37    pub inflight_refs_delta: BTreeMap<RouteId, i32>,
38    pub wall_remainder_delta: u128,
39    pub local_tick_delta: u64,
40    pub observer_eviction_pending: Vec<ObserverHandle>,
41}
42
43#[derive(Debug, Clone, Serialize, Deserialize)]
44pub(crate) enum StagedStateDelta {
45    SpawnEntity {
46        id: EntityId,
47        meta: EntityMeta,
48    },
49    DespawnEntity {
50        id: EntityId,
51    },
52    SetComponent {
53        entity: EntityId,
54        type_code: TypeCode,
55        bytes: Bytes,
56        size: u64,
57    },
58    RemoveComponent {
59        entity: EntityId,
60        type_code: TypeCode,
61        size: u64,
62    },
63}
64
65#[derive(Debug, Clone, Serialize, Deserialize)]
66pub(crate) enum ScheduledEntryDelta {
67    Add(ScheduledEntry),
68    Remove(ScheduledActionId),
69}
70
71#[derive(Debug, Clone, Serialize, Deserialize)]
72pub(crate) struct PendingSignal {
73    pub target: InstanceId,
74    pub route: RouteId,
75    pub payload: Bytes,
76    pub principal: Principal,
77}
78
79#[derive(Debug, Default, Clone, Serialize, Deserialize)]
80pub(crate) struct IdCountersDelta {
81    pub next_entity_advance: u64,
82    pub next_scheduled_advance: u64,
83    pub next_source_seq_advance: u64,
84}
85
86#[derive(Debug, Default, Clone, Serialize, Deserialize)]
87pub(crate) struct ResourceLedgerDelta {
88    pub ops: Vec<LedgerOp>,
89}
90
91#[derive(Debug, Clone, Serialize, Deserialize)]
92pub(crate) enum LedgerOp {
93    AddEntity(EntityId),
94    RemoveEntity(EntityId),
95    AddComponent {
96        entity: EntityId,
97        type_code: TypeCode,
98        size: u64,
99    },
100    RemoveComponent {
101        entity: EntityId,
102        type_code: TypeCode,
103        size: u64,
104    },
105}
106
107/// Net component-byte delta accumulated in `stage.ledger_delta` so far.
108/// Used by `step()` budget enforcement to project the post-commit total
109/// before each Op is dispatched. `RemoveEntity` does not carry per-entity
110/// byte counts at this layer — its bytes are recovered from the ledger
111/// at apply time, so this helper is conservative (under-counts freed
112/// bytes within a single step), favoring false-deny over false-allow.
113///
114/// `size` is a caller-supplied `u64`; a value above `i64::MAX` must NOT
115/// wrap to a negative `i64` (which would shrink the projection and let an
116/// oversized component slip past the budget gate). `i64::try_from(..)
117/// .unwrap_or(i64::MAX)` clamps an oversized add to the maximum positive
118/// magnitude, so it always trips the deny path rather than bypassing it.
119pub(crate) fn bytes_delta(stage: &StepStage) -> i64 {
120    let mut d: i64 = 0;
121    for op in &stage.ledger_delta.ops {
122        match op {
123            LedgerOp::AddComponent { size, .. } => {
124                d = d.saturating_add(i64::try_from(*size).unwrap_or(i64::MAX));
125            }
126            LedgerOp::RemoveComponent { size, .. } => {
127                d = d.saturating_sub(i64::try_from(*size).unwrap_or(i64::MAX));
128            }
129            LedgerOp::AddEntity(_) | LedgerOp::RemoveEntity(_) => {}
130        }
131    }
132    d
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138    use crate::abi::Tick;
139
140    #[test]
141    fn step_stage_default_all_buckets_empty() {
142        let s = StepStage::default();
143        assert!(s.state_ops.is_empty());
144        assert!(s.events.is_empty());
145        assert!(s.schedule_deltas.is_empty());
146        assert!(s.pending_signals.is_empty());
147        assert_eq!(s.id_counters.next_entity_advance, 0);
148        assert_eq!(s.id_counters.next_scheduled_advance, 0);
149        assert_eq!(s.id_counters.next_source_seq_advance, 0);
150        assert!(s.ledger_delta.ops.is_empty());
151        assert!(s.inflight_refs_delta.is_empty());
152        assert_eq!(s.wall_remainder_delta, 0);
153        assert_eq!(s.local_tick_delta, 0);
154        assert!(s.observer_eviction_pending.is_empty());
155    }
156
157    #[test]
158    fn staged_state_delta_variants_clone() {
159        let id = EntityId::new(1).unwrap();
160        let meta = EntityMeta {
161            owner: Principal::System,
162            created: Tick(0),
163        };
164        let _ = StagedStateDelta::SpawnEntity { id, meta }.clone();
165        let _ = StagedStateDelta::DespawnEntity { id }.clone();
166        let _ = StagedStateDelta::SetComponent {
167            entity: id,
168            type_code: TypeCode(1),
169            bytes: Bytes::from_static(b"x"),
170            size: 1,
171        }
172        .clone();
173        let _ = StagedStateDelta::RemoveComponent {
174            entity: id,
175            type_code: TypeCode(1),
176            size: 1,
177        }
178        .clone();
179    }
180
181    #[test]
182    fn scheduled_entry_delta_variants() {
183        let entry = ScheduledEntry {
184            id: ScheduledActionId::new(1).unwrap(),
185            at: Tick(0),
186            actor: None,
187            principal: Principal::System,
188            action_type_code: TypeCode(0),
189            action_bytes: vec![],
190        };
191        let _ = ScheduledEntryDelta::Add(entry).clone();
192        let _ = ScheduledEntryDelta::Remove(ScheduledActionId::new(1).unwrap()).clone();
193    }
194
195    #[test]
196    fn bytes_delta_clamps_oversized_size_no_negative_wrap() {
197        // Regression: a size above i64::MAX must clamp to i64::MAX (positive),
198        // never wrap negative — otherwise the budget projection would shrink
199        // and let an oversized component bypass the memory budget gate.
200        let id = EntityId::new(1).unwrap();
201        let mut stage = StepStage::default();
202        stage.ledger_delta.ops.push(LedgerOp::AddComponent {
203            entity: id,
204            type_code: TypeCode(1),
205            size: u64::MAX,
206        });
207        assert_eq!(bytes_delta(&stage), i64::MAX);
208    }
209
210    #[test]
211    fn ledger_op_variants() {
212        let id = EntityId::new(1).unwrap();
213        let _ = LedgerOp::AddEntity(id).clone();
214        let _ = LedgerOp::RemoveEntity(id).clone();
215        let _ = LedgerOp::AddComponent {
216            entity: id,
217            type_code: TypeCode(1),
218            size: 100,
219        }
220        .clone();
221        let _ = LedgerOp::RemoveComponent {
222            entity: id,
223            type_code: TypeCode(1),
224            size: 100,
225        }
226        .clone();
227    }
228
229    #[test]
230    fn pending_signal_construction() {
231        let s = PendingSignal {
232            target: InstanceId::new(1).unwrap(),
233            route: RouteId(1),
234            payload: Bytes::from_static(b"hello"),
235            principal: Principal::System,
236        };
237        assert_eq!(s.payload.len(), 5);
238    }
239}
arkhe_kernel/runtime/stage.rs

arkhe_kernel/runtime/
stage.rs
