meerkat_runtime/
input_ledger.rs

1//! InputLedger — in-memory ledger of InputState entries.
2//!
3//! IndexMap<InputId, InputState>.
4
5use indexmap::IndexMap;
6use meerkat_core::lifecycle::InputId;
7use std::collections::BTreeSet;
8
9use crate::input_state::InputState;
10
11/// In-memory ledger tracking InputState for all inputs.
12#[derive(Debug, Default, Clone)]
13pub struct InputLedger {
14    /// InputId → InputState (insertion order preserved).
15    states: IndexMap<InputId, InputState>,
16    /// Canonical owners of unfinished terminal work, ordered for stable paging.
17    pending_terminal_owners: BTreeSet<uuid::Uuid>,
18}
19
20impl InputLedger {
21    /// Create a new empty ledger.
22    pub fn new() -> Self {
23        Self::default()
24    }
25
26    /// Accept a new InputState into the ledger.
27    pub fn accept(&mut self, state: InputState) {
28        let input_id = state.input_id.clone();
29        self.states.insert(input_id.clone(), state);
30        self.refresh_pending_terminal_owner(&input_id);
31    }
32
33    /// Recover an InputState after generated recovery authority retained it.
34    ///
35    /// Recovery retention and idempotency ownership live in the generated
36    /// MeerkatMachine recovery/admission path, not in the ledger.
37    /// Returns `true` if the state was inserted.
38    pub fn recover(&mut self, state: InputState) -> bool {
39        let input_id = state.input_id.clone();
40        self.states.insert(input_id.clone(), state);
41        self.refresh_pending_terminal_owner(&input_id);
42        true
43    }
44
45    /// Get the state of a specific input.
46    pub fn get(&self, input_id: &InputId) -> Option<&InputState> {
47        self.states.get(input_id)
48    }
49
50    /// Remove an input from the ledger.
51    pub fn remove(&mut self, input_id: &InputId) -> Option<InputState> {
52        self.pending_terminal_owners.remove(&input_id.0);
53        self.states.shift_remove(input_id)
54    }
55
56    /// Get mutable reference to the state of a specific input.
57    pub fn get_mut(&mut self, input_id: &InputId) -> Option<&mut InputState> {
58        self.states.get_mut(input_id)
59    }
60
61    /// Iterate over all input states. "Active" (non-terminal) filtering must
62    /// happen at the driver level, which has DSL access; the ledger by itself
63    /// carries only shell metadata.
64    pub fn iter(&self) -> impl Iterator<Item = (&InputId, &InputState)> {
65        self.states.iter()
66    }
67
68    /// Refresh the secondary pending-terminal owner index after an in-place
69    /// shell mutation of one row.
70    pub(crate) fn refresh_pending_terminal_owner(&mut self, input_id: &InputId) {
71        let pending = self
72            .states
73            .get(input_id)
74            .is_some_and(crate::store::input_state_is_pending_terminal_owner);
75        if pending {
76            self.pending_terminal_owners.insert(input_id.0);
77        } else {
78            self.pending_terminal_owners.remove(&input_id.0);
79        }
80    }
81
82    /// Stable canonical owner ids for unfinished terminal work.
83    pub(crate) fn pending_terminal_owner_ids(&self) -> Vec<InputId> {
84        self.pending_terminal_owners
85            .iter()
86            .copied()
87            .map(InputId::from_uuid)
88            .collect()
89    }
90
91    /// Number of entries in the ledger.
92    pub fn len(&self) -> usize {
93        self.states.len()
94    }
95
96    /// Check if the ledger is empty.
97    pub fn is_empty(&self) -> bool {
98        self.states.is_empty()
99    }
100}
101
102#[cfg(test)]
103#[allow(clippy::unwrap_used)]
104mod tests {
105    use super::*;
106
107    #[test]
108    fn accept_and_retrieve() {
109        let mut ledger = InputLedger::new();
110        let id = InputId::new();
111        let state = InputState::new_accepted(id.clone());
112        ledger.accept(state);
113
114        assert_eq!(ledger.len(), 1);
115        assert!(!ledger.is_empty());
116        let retrieved = ledger.get(&id).unwrap();
117        assert_eq!(retrieved.input_id, id);
118    }
119
120    #[test]
121    fn accept_preserves_idempotency_key_as_metadata_only() {
122        let mut ledger = InputLedger::new();
123
124        let input_id = InputId::new();
125        let mut state = InputState::new_accepted(input_id.clone());
126        state.idempotency_key = Some(crate::identifiers::IdempotencyKey::new("req-123"));
127        ledger.accept(state);
128
129        assert_eq!(ledger.len(), 1);
130        assert_eq!(
131            ledger
132                .get(&input_id)
133                .and_then(|state| state.idempotency_key.as_ref()),
134            Some(&crate::identifiers::IdempotencyKey::new("req-123"))
135        );
136    }
137
138    #[test]
139    fn recover_does_not_interpret_durability() {
140        let mut ledger = InputLedger::new();
141
142        let mut state = InputState::new_accepted(InputId::new());
143        state.durability = Some(crate::input::InputDurability::Ephemeral);
144        assert!(
145            ledger.recover(state),
146            "the ledger inserts rows after machine authority has made the retention decision"
147        );
148        assert_eq!(ledger.len(), 1);
149    }
150}
meerkat_runtime/input_ledger.rs

meerkat_runtime/
input_ledger.rs
