umsh_text/engine/
sequence.rs

1//! Per-stream sequence state: serial-number arithmetic, duplicate windows,
2//! and wire-ID-to-handle mappings.
3//!
4//! Sequence IDs are scoped to `(conversation, sender)`. Each stream keeps a
5//! windowed bitmap of recently seen IDs (half-range window of 128) plus a
6//! bounded ring mapping recent wire IDs to stable application handles.
7
8use umsh_core::PublicKey;
9
10use crate::model::{ConversationKey, SenderScope};
11
12/// Serial-number delta from `last` to `id`, modulo 256.
13///
14/// Deltas 1–127 are newer; 128–255 are old or ambiguous.
15pub fn serial_delta(last: u8, id: u8) -> u8 {
16    id.wrapping_sub(last)
17}
18
19/// Classification of a received ID relative to a stream baseline.
20#[derive(Clone, Copy, Debug, PartialEq, Eq)]
21pub enum SerialClass {
22    /// Equal to the baseline.
23    Baseline,
24    /// Newer by the contained forward delta (1–127).
25    Newer(u8),
26    /// Older by the contained backward distance (1–127).
27    Older(u8),
28    /// Exactly half the range away: old or ambiguous.
29    Ambiguous,
30}
31
32pub fn classify(last: u8, id: u8) -> SerialClass {
33    match serial_delta(last, id) {
34        0 => SerialClass::Baseline,
35        delta @ 1..=127 => SerialClass::Newer(delta),
36        128 => SerialClass::Ambiguous,
37        delta => SerialClass::Older(delta.wrapping_neg()),
38    }
39}
40
41/// Stable application identity of a transcript message.
42#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
43pub struct MessageHandle(pub u32);
44
45/// Bounded ring mapping recent wire IDs to message handles within one
46/// stream epoch. When an ID is reused or evicted, the old mapping retires.
47#[derive(Clone, Debug, Default)]
48pub struct RefRing {
49    entries: heapless::Vec<(u8, MessageHandle), 16>,
50}
51
52impl RefRing {
53    pub fn clear(&mut self) {
54        self.entries.clear();
55    }
56
57    /// Record `id -> handle`, retiring any previous mapping of `id` and the
58    /// oldest entry when full.
59    pub fn record(&mut self, id: u8, handle: MessageHandle) {
60        self.entries.retain(|(entry_id, _)| *entry_id != id);
61        if self.entries.is_full() {
62            self.entries.remove(0);
63        }
64        let _ = self.entries.push((id, handle));
65    }
66
67    pub fn lookup(&self, id: u8) -> Option<MessageHandle> {
68        self.entries
69            .iter()
70            .find(|(entry_id, _)| *entry_id == id)
71            .map(|(_, handle)| *handle)
72    }
73
74    pub fn lookup_handle(&self, handle: MessageHandle) -> Option<u8> {
75        self.entries
76            .iter()
77            .find(|(_, entry)| *entry == handle)
78            .map(|(id, _)| *id)
79    }
80
81    /// Retire the mapping for `id`, if present.
82    pub fn retire(&mut self, id: u8) {
83        self.entries.retain(|(entry_id, _)| *entry_id != id);
84    }
85}
86
87/// Bounded ring mapping an edit's own wire ID back to the ID it replaced.
88///
89/// Senders name the *original* when referencing an edited message, but a
90/// sender that names the edit instead is easy to accommodate and impossible
91/// to distinguish from a stale reference otherwise, so references are chased
92/// through this ring before lookup. Recording chases one hop, so an edit of
93/// an edit collapses to the first original rather than forming a chain.
94#[derive(Clone, Debug, Default)]
95pub struct EditRing {
96    entries: heapless::Vec<(u8, u8), 8>,
97}
98
99impl EditRing {
100    pub fn clear(&mut self) {
101        self.entries.clear();
102    }
103
104    /// Record `edit_id -> original_id`, collapsing through any existing
105    /// mapping of `original_id` and evicting the oldest entry when full.
106    pub fn record(&mut self, edit_id: u8, original_id: u8) {
107        let original_id = self.resolve(original_id);
108        if edit_id == original_id {
109            return;
110        }
111        self.entries.retain(|(entry_id, _)| *entry_id != edit_id);
112        if self.entries.is_full() {
113            self.entries.remove(0);
114        }
115        let _ = self.entries.push((edit_id, original_id));
116    }
117
118    /// The original an ID stands for, or the ID itself when unmapped.
119    pub fn resolve(&self, id: u8) -> u8 {
120        self.entries
121            .iter()
122            .find(|(entry_id, _)| *entry_id == id)
123            .map(|(_, original)| *original)
124            .unwrap_or(id)
125    }
126
127    /// Retire any mapping involving `id`, in either position.
128    pub fn retire(&mut self, id: u8) {
129        self.entries
130            .retain(|(edit_id, original_id)| *edit_id != id && *original_id != id);
131    }
132}
133
134/// 256-bit seen-ID bitmap with half-range window semantics.
135#[derive(Clone, Debug, Default)]
136pub struct SeenWindow {
137    bits: [u32; 8],
138}
139
140impl SeenWindow {
141    pub fn clear(&mut self) {
142        self.bits = [0; 8];
143    }
144
145    pub fn contains(&self, id: u8) -> bool {
146        self.bits[(id >> 5) as usize] & (1 << (id & 31)) != 0
147    }
148
149    pub fn insert(&mut self, id: u8) {
150        self.bits[(id >> 5) as usize] |= 1 << (id & 31);
151    }
152
153    fn remove(&mut self, id: u8) {
154        self.bits[(id >> 5) as usize] &= !(1 << (id & 31));
155    }
156
157    /// Advance the baseline from `last` by `delta`, clearing stale bits for
158    /// the IDs entering the window so quarter-old duplicates never alias
159    /// wrapped IDs.
160    pub fn advance(&mut self, last: u8, delta: u8) {
161        for step in 1..=delta {
162            self.remove(last.wrapping_add(step));
163        }
164    }
165}
166
167/// A pending automatic repair request for one missing frame.
168#[derive(Clone, Copy, Debug, PartialEq, Eq)]
169pub struct PendingRepair {
170    pub message_id: u8,
171    /// Missing fragment index, or `None` for the whole-message 1-byte form.
172    pub fragment: Option<u8>,
173    /// Earliest time the request may be transmitted (grace plus jitter).
174    pub deadline_ms: u64,
175    pub attempts: u8,
176    /// Ordered-slot placeholder reserved for this missing whole message, so a
177    /// terminal repair outcome can flip the same slot to `Unavailable`.
178    /// `None` for fragment repairs (their slot already exists).
179    pub handle: Option<MessageHandle>,
180}
181
182/// Inbound stream state for one `(conversation, sender)` pair.
183#[derive(Clone, Debug)]
184pub struct InboundStream {
185    /// Local epoch counter; bumped on reset so stale cached state can never
186    /// merge across a reset.
187    pub epoch: u16,
188    /// Most recent (serial-order) accepted ID.
189    pub baseline: Option<u8>,
190    pub seen: SeenWindow,
191    pub refs: RefRing,
192    /// Edit IDs on this stream, mapped back to the originals they replaced.
193    pub edit_refs: EditRing,
194    pub pending: heapless::Vec<PendingRepair, 8>,
195    /// Resolved full key of a claimed multicast member, when known. Needed to
196    /// address group repair requests.
197    pub sender_key: Option<PublicKey>,
198    /// Two known peer keys collide on this stream's hint; automatic repair is
199    /// suppressed and references resolve only when unambiguous.
200    pub collided: bool,
201    pub last_request_ms: u64,
202    pub last_active_ms: u64,
203}
204
205impl InboundStream {
206    pub fn new(now_ms: u64) -> Self {
207        Self {
208            epoch: 0,
209            baseline: None,
210            seen: SeenWindow::default(),
211            refs: RefRing::default(),
212            edit_refs: EditRing::default(),
213            pending: heapless::Vec::new(),
214            sender_key: None,
215            collided: false,
216            last_request_ms: 0,
217            last_active_ms: now_ms,
218        }
219    }
220
221    /// Start a new epoch, discarding cached wire mappings and repair state
222    /// but not transcript history.
223    pub fn reset_epoch(&mut self, new_baseline: Option<u8>) {
224        self.epoch = self.epoch.wrapping_add(1);
225        self.baseline = new_baseline;
226        self.seen.clear();
227        self.refs.clear();
228        self.edit_refs.clear();
229        self.pending.clear();
230        if let Some(id) = new_baseline {
231            self.seen.insert(id);
232        }
233    }
234
235    /// Cancel pending repairs satisfied by an arrival: a whole-message
236    /// arrival (`fragment == None`) satisfies everything for that ID, while a
237    /// fragment arrival satisfies its own request and any whole-message
238    /// request (which fragment zero or reassembly tracking supersedes).
239    pub fn cancel_pending(&mut self, message_id: u8, fragment: Option<u8>) {
240        self.pending.retain(|pending| {
241            if pending.message_id != message_id {
242                return true;
243            }
244            match (fragment, pending.fragment) {
245                (None, _) | (Some(_), None) => false,
246                (Some(arrived), Some(wanted)) => wanted != arrived,
247            }
248        });
249    }
250}
251
252/// Outbound stream state for the local sender in one conversation.
253#[derive(Clone, Debug)]
254pub struct OutboundStream {
255    pub next_id: u8,
256    pub epoch: u16,
257    /// Include Sequence Reset on the next message sent in this conversation.
258    pub announce_reset: bool,
259    pub refs: RefRing,
260    /// Our own edit IDs, mapped back to the originals they replaced, so a
261    /// peer that references an edit of ours still lands on the original.
262    pub edit_refs: EditRing,
263    pub last_active_ms: u64,
264}
265
266impl OutboundStream {
267    pub fn fresh(now_ms: u64) -> Self {
268        Self {
269            next_id: 0,
270            epoch: 0,
271            announce_reset: true,
272            refs: RefRing::default(),
273            edit_refs: EditRing::default(),
274            last_active_ms: now_ms,
275        }
276    }
277
278    /// Allocate the next wire ID, retiring any wrapped mapping.
279    pub fn allocate(&mut self) -> u8 {
280        let id = self.next_id;
281        self.next_id = self.next_id.wrapping_add(1);
282        self.refs.retire(id);
283        self.edit_refs.retire(id);
284        id
285    }
286}
287
288/// Key of an inbound stream.
289#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
290pub struct StreamKey {
291    pub conversation: ConversationKey,
292    pub sender: SenderScope,
293}
umsh_text/engine/sequence.rs

umsh_text/engine/
sequence.rs
