monoloop_loop/transaction/
finalization.rs

1//! Exactly-once finalization guard and event sequence allocation.
2
3use monoloop_contracts::{
4    ChannelId, CompletionCallback, EventDeliveryOutcome, SessionId, TransactionEnd,
5    TransactionEndKind, TransactionId, TransactionUsage,
6};
7use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
8use std::sync::{Arc, Mutex};
9
10/// Sole allocator of transaction event sequence numbers (starts at 1).
11#[derive(Debug)]
12pub struct EventSequencer {
13    next: AtomicU64,
14}
15
16impl EventSequencer {
17    /// Create a sequencer; first allocated sequence is 1.
18    pub fn new() -> Self {
19        Self {
20            next: AtomicU64::new(1),
21        }
22    }
23
24    /// Next sequence that [`Self::allocate`] will return (D-036 peek-before-enqueue).
25    pub fn peek_next(&self) -> u64 {
26        self.next.load(Ordering::SeqCst)
27    }
28
29    /// Allocate the next contiguous sequence number.
30    pub fn allocate(&self) -> u64 {
31        self.next.fetch_add(1, Ordering::SeqCst)
32    }
33
34    /// Last allocated sequence (0 if none).
35    pub fn last_allocated(&self) -> u64 {
36        self.next.load(Ordering::SeqCst).saturating_sub(1)
37    }
38}
39
40impl Default for EventSequencer {
41    fn default() -> Self {
42        Self::new()
43    }
44}
45
46/// Material taken exactly once by the winning finalization path.
47pub struct FinalizationPayload {
48    /// Completion callback.
49    pub callback: Box<dyn CompletionCallback>,
50    /// Channel id for terminal.
51    pub channel_id: ChannelId,
52    /// Session when known.
53    pub session_id: Option<SessionId>,
54    /// Transaction id.
55    pub transaction_id: TransactionId,
56}
57
58/// Atomic exactly-once finalization claim shared by actor and shutdown supervisor.
59pub struct FinalizationGuard {
60    claimed: AtomicBool,
61    payload: Mutex<Option<FinalizationPayload>>,
62    sequencer: Arc<EventSequencer>,
63    /// Whether the callback was scheduled (tests / shutdown accounting).
64    callback_scheduled: AtomicBool,
65}
66
67impl FinalizationGuard {
68    /// Create a guard holding the one-shot callback.
69    pub fn new(
70        transaction_id: TransactionId,
71        channel_id: ChannelId,
72        session_id: Option<SessionId>,
73        callback: Box<dyn CompletionCallback>,
74        sequencer: Arc<EventSequencer>,
75    ) -> Arc<Self> {
76        Arc::new(Self {
77            claimed: AtomicBool::new(false),
78            payload: Mutex::new(Some(FinalizationPayload {
79                callback,
80                channel_id,
81                session_id,
82                transaction_id,
83            })),
84            sequencer,
85            callback_scheduled: AtomicBool::new(false),
86        })
87    }
88
89    /// Event sequencer for this transaction.
90    pub fn sequencer(&self) -> &Arc<EventSequencer> {
91        &self.sequencer
92    }
93
94    /// Update session id on the payload before claim (session establishment).
95    pub fn set_session_id(&self, session_id: SessionId) {
96        if let Ok(mut g) = self.payload.lock() {
97            if let Some(p) = g.as_mut() {
98                p.session_id = Some(session_id);
99            }
100        }
101    }
102
103    /// Claim exactly once. Winner takes payload; losers get `None`.
104    pub fn try_claim(&self) -> Option<FinalizationPayload> {
105        if self
106            .claimed
107            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
108            .is_err()
109        {
110            return None;
111        }
112        self.payload.lock().ok().and_then(|mut g| g.take())
113    }
114
115    /// Whether already claimed.
116    pub fn is_claimed(&self) -> bool {
117        self.claimed.load(Ordering::SeqCst)
118    }
119
120    /// Mark that a callback was scheduled (accounting).
121    pub fn mark_callback_scheduled(&self) {
122        self.callback_scheduled.store(true, Ordering::SeqCst);
123    }
124
125    /// Whether callback was scheduled.
126    pub fn callback_was_scheduled(&self) -> bool {
127        self.callback_scheduled.load(Ordering::SeqCst)
128    }
129}
130
131/// Build a terminal end event payload fields helper.
132pub fn build_transaction_end(
133    payload: &FinalizationPayload,
134    kind: TransactionEndKind,
135    prior: Option<TransactionEndKind>,
136    event_delivery: EventDeliveryOutcome,
137    emitted_events: u64,
138) -> TransactionEnd {
139    TransactionEnd {
140        transaction_id: payload.transaction_id,
141        session_id: payload.session_id.clone(),
142        channel_id: payload.channel_id.clone(),
143        kind,
144        prior_terminal_cause: prior,
145        event_delivery,
146        emitted_events,
147        usage: TransactionUsage::default(),
148        diagnostics: vec![],
149    }
150}
151
152/// Bound a diagnostic list by count and per-message bytes (D-015).
153pub fn bound_diagnostics(
154    mut diagnostics: Vec<monoloop_contracts::TransactionDiagnostic>,
155    max_count: usize,
156    max_message_bytes: usize,
157) -> Vec<monoloop_contracts::TransactionDiagnostic> {
158    if diagnostics.len() > max_count {
159        diagnostics.truncate(max_count.max(1));
160    }
161    for d in &mut diagnostics {
162        if let Some(ref mut msg) = d.diagnostic.message {
163            if msg.len() > max_message_bytes {
164                msg.truncate(max_message_bytes);
165                while !msg.is_char_boundary(msg.len()) {
166                    msg.pop();
167                }
168            }
169        }
170    }
171    diagnostics
172}
monoloop_loop/transaction/finalization.rs

monoloop_loop/transaction/
finalization.rs
