monoloop_loop/transaction/lifecycle/
ledger.rs

1//! Lifecycle ledger — continuous representation of admitted transactions (v2 §8).
2
3use super::capacity::TransactionReservations;
4use super::terminal::TerminalDecision;
5use crate::transaction::sticky_cancel::StickyCancel;
6use monoloop_contracts::{
7    CanonicalInput, ChannelId, EffectiveConfig, InvocationConfig, SessionConfig, SessionKey,
8    ToolId, TransactionCompletionSender, TransactionDelivery, TransactionId, TransactionUsage,
9};
10use std::collections::HashMap;
11use std::sync::Arc;
12
13/// Resource controls for cooperative cancel / shutdown wakeups.
14#[derive(Debug, Clone)]
15pub struct ResourceControls {
16    /// Sticky cancel / shutdown signal for the coordinator (flag before notify).
17    pub cancel: Arc<StickyCancel>,
18}
19
20impl Default for ResourceControls {
21    fn default() -> Self {
22        Self {
23            cancel: Arc::new(StickyCancel::new()),
24        }
25    }
26}
27
28/// Per-transaction ledger row.
29#[derive(Debug)]
30pub struct LedgerEntry {
31    /// Transaction id.
32    pub transaction_id: TransactionId,
33    /// Channel.
34    pub channel_id: ChannelId,
35    /// Session when known at admission or after claim.
36    pub session_key: Option<SessionKey>,
37    /// Current phase.
38    pub phase: TransactionPhase,
39    /// Immutable terminal decision once selected.
40    pub terminal: Option<TerminalDecision>,
41    /// Coordinator proposal parked before `WorkerExited` notify (join_next recovery).
42    pub pending_worker_proposal: Option<super::terminal::TerminalProposal>,
43    /// Last allocated event sequence (0 = none yet).
44    pub event_sequence: u64,
45    /// Full delivery ports at admit; taken at Start (split into publisher + completion).
46    pub delivery: Option<TransactionDelivery>,
47    /// Completion sender retained until Seal + publish.
48    pub completion_tx: Option<TransactionCompletionSender>,
49    /// Ordinary Publish/Establish admit gate into this transaction's event publisher.
50    pub publisher_cmd_tx: Option<super::event_publisher::OrdinaryCmdAdmit>,
51    /// Dedicated Seal sender (D-047 priority path; capacity 1).
52    pub publisher_seal_tx: Option<tokio::sync::mpsc::Sender<super::event_publisher::SealCommand>>,
53    /// Canonical input captured at admission.
54    pub input: CanonicalInput,
55    /// Invocation configuration (raw admit request; validated into `effective_config`).
56    pub invocation_config: InvocationConfig,
57    /// Optional session configuration (raw admit request).
58    pub session_config: Option<SessionConfig>,
59    /// Validated effective configuration (computed synchronously at admission, §9.2).
60    pub effective_config: EffectiveConfig,
61    /// Selected tool ids.
62    pub tools: Vec<ToolId>,
63    /// RAII reservations.
64    pub reservations: Option<TransactionReservations>,
65    /// Cancel / control knobs.
66    pub resources: ResourceControls,
67    /// Usage facts.
68    pub usage: TransactionUsage,
69    /// Bounded diagnostics count.
70    pub diagnostic_count: u32,
71}
72
73/// Ledger phase machine (v2 §8.2).
74#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
75pub enum TransactionPhase {
76    /// Admitted; supervisor has not started work.
77    Queued,
78    /// External session create/load in progress.
79    EstablishingSession,
80    /// Provider/tool work running.
81    Running,
82    /// Cancellation in progress.
83    Cancelling,
84    /// Terminal selected; publishing.
85    Finalizing,
86    /// Completion published; cleanup already done.
87    CompletionPublished,
88    /// Completion published; owned cleanup remains.
89    CleanupPending,
90}
91
92/// Source of truth from admission through completion publication.
93#[derive(Debug, Default)]
94pub struct LifecycleLedger {
95    by_transaction: HashMap<TransactionId, LedgerEntry>,
96    by_session: HashMap<SessionKey, TransactionId>,
97}
98
99impl LifecycleLedger {
100    /// Empty ledger.
101    pub fn new() -> Self {
102        Self::default()
103    }
104
105    /// Number of entries.
106    pub fn len(&self) -> usize {
107        self.by_transaction.len()
108    }
109
110    /// Whether empty.
111    pub fn is_empty(&self) -> bool {
112        self.by_transaction.is_empty()
113    }
114
115    /// Snapshot of all transaction ids (for shutdown).
116    pub fn transaction_ids(&self) -> Vec<TransactionId> {
117        self.by_transaction.keys().copied().collect()
118    }
119
120    /// Lookup by id.
121    pub fn get(&self, id: &TransactionId) -> Option<&LedgerEntry> {
122        self.by_transaction.get(id)
123    }
124
125    /// Mutable lookup by id.
126    pub fn get_mut(&mut self, id: &TransactionId) -> Option<&mut LedgerEntry> {
127        self.by_transaction.get_mut(id)
128    }
129
130    /// Whether a session key is already active.
131    pub fn session_active(&self, key: &SessionKey) -> bool {
132        self.by_session.contains_key(key)
133    }
134
135    /// Resolve the active transaction for a session key.
136    pub fn transaction_for_session(&self, key: &SessionKey) -> Option<TransactionId> {
137        self.by_session.get(key).copied()
138    }
139
140    /// Count distinct active sessions on a channel (D-015 / ChannelLimits).
141    pub fn distinct_sessions_on_channel(&self, channel: &ChannelId) -> usize {
142        self.by_session
143            .keys()
144            .filter(|k| k.channel_id == *channel)
145            .count()
146    }
147
148    /// Insert a complete Queued entry. Returns `Err` if id or session collides
149    /// or if `max_distinct_sessions` would be exceeded for a new SessionKey.
150    pub fn insert_queued(
151        &mut self,
152        entry: LedgerEntry,
153        max_distinct_sessions: Option<usize>,
154    ) -> Result<(), LedgerInsertError> {
155        if self.by_transaction.contains_key(&entry.transaction_id) {
156            return Err(LedgerInsertError::DuplicateTransaction);
157        }
158        if let Some(ref key) = entry.session_key {
159            if self.by_session.contains_key(key) {
160                return Err(LedgerInsertError::SessionAlreadyActive);
161            }
162            if let Some(max) = max_distinct_sessions {
163                if self.distinct_sessions_on_channel(&key.channel_id) >= max {
164                    return Err(LedgerInsertError::DistinctSessionsExceeded);
165                }
166            }
167        }
168        if let Some(ref key) = entry.session_key {
169            self.by_session.insert(key.clone(), entry.transaction_id);
170        }
171        self.by_transaction.insert(entry.transaction_id, entry);
172        Ok(())
173    }
174
175    /// Remove an entry and drop its reservations (via Drop).
176    pub fn remove(&mut self, id: &TransactionId) -> Option<LedgerEntry> {
177        let entry = self.by_transaction.remove(id)?;
178        if let Some(ref key) = entry.session_key {
179            if self.by_session.get(key) == Some(id) {
180                self.by_session.remove(key);
181            }
182        }
183        Some(entry)
184    }
185
186    /// Bind session key after external session claim (supervisor only).
187    ///
188    /// When `max_distinct_sessions` is `Some`, a net-new session on the channel
189    /// that would exceed the bound fails with `DistinctSessionsExceeded`.
190    /// Replacing an existing key on the same channel does not consume an extra
191    /// distinct slot.
192    ///
193    /// D-063: admission (`insert_queued`) already reserves `SessionKey` in
194    /// `by_session` for a resumed transaction (`session_id: Some(..)` on
195    /// `TransactionSubmitRequest`), bound to that transaction's own id, before
196    /// the claim below ever runs. If the existing holder *is* `id`, this call
197    /// is that same transaction re-confirming its own admission-time
198    /// reservation once the external session is established — not a new
199    /// distinct-session slot — so it must succeed as a no-op rather than
200    /// reject its own resume with `SessionAlreadyActive`.
201    pub fn bind_session(
202        &mut self,
203        id: &TransactionId,
204        key: SessionKey,
205        max_distinct_sessions: Option<usize>,
206    ) -> Result<(), LedgerInsertError> {
207        if let Some(holder) = self.by_session.get(&key) {
208            if holder != id {
209                return Err(LedgerInsertError::SessionAlreadyActive);
210            }
211            return Ok(());
212        }
213        let replacing_same_channel = self
214            .by_transaction
215            .get(id)
216            .ok_or(LedgerInsertError::UnknownTransaction)?
217            .session_key
218            .as_ref()
219            .is_some_and(|old| old.channel_id == key.channel_id);
220        if let Some(max) = max_distinct_sessions {
221            if !replacing_same_channel && self.distinct_sessions_on_channel(&key.channel_id) >= max
222            {
223                return Err(LedgerInsertError::DistinctSessionsExceeded);
224            }
225        }
226        let entry = self
227            .by_transaction
228            .get_mut(id)
229            .ok_or(LedgerInsertError::UnknownTransaction)?;
230        if let Some(ref old) = entry.session_key {
231            self.by_session.remove(old);
232        }
233        entry.session_key = Some(key.clone());
234        self.by_session.insert(key, *id);
235        Ok(())
236    }
237}
238
239/// Ledger install / bind failure.
240#[derive(Clone, Copy, Debug, PartialEq, Eq)]
241pub enum LedgerInsertError {
242    /// Transaction id already present.
243    DuplicateTransaction,
244    /// Session key already has an active transaction.
245    SessionAlreadyActive,
246    /// Channel `max_distinct_sessions` would be exceeded.
247    DistinctSessionsExceeded,
248    /// Unknown transaction id.
249    UnknownTransaction,
250}
monoloop_loop/transaction/lifecycle/ledger.rs

monoloop_loop/transaction/lifecycle/
ledger.rs
