umsh_text/engine/
repair.rs

1//! Resend-service bookkeeping: pending archive lookups, response coalescing,
2//! and deterministic jitter.
3
4use umsh_core::PublicKey;
5
6use crate::model::{ConversationKey, MessageSequence};
7
8/// An archive lookup the platform has been asked to perform.
9#[derive(Clone, Copy, Debug)]
10pub struct PendingLookup {
11    pub request_id: u32,
12    /// The stream the request selects (the original conversation).
13    pub conversation: ConversationKey,
14    /// The authenticated requester (used only for diagnostics; responses are
15    /// addressed by the conversation's delivery mode).
16    pub requester: PublicKey,
17    pub sequence: MessageSequence,
18}
19
20/// Ring of recently transmitted or answered frames, for coalescing resend
21/// requests: duplicate requests from multiple group members, and requests
22/// for a frame whose original transmission just left this node (a slow
23/// serialized link can deliver frames later than the requester's patience).
24#[derive(Clone, Debug, Default)]
25pub struct CoalesceRing {
26    entries: heapless::Vec<(ConversationKey, u8, Option<u8>, u64), 16>,
27}
28
29impl CoalesceRing {
30    /// True when an equivalent request was answered within `window_ms`.
31    pub fn recently_answered(
32        &self,
33        conversation: &ConversationKey,
34        sequence: &MessageSequence,
35        now_ms: u64,
36        window_ms: u64,
37    ) -> bool {
38        let fragment = sequence.fragment.map(|fragment| fragment.index);
39        self.entries.iter().any(|(conv, id, frag, at)| {
40            conv == conversation
41                && *id == sequence.message_id
42                && *frag == fragment
43                && now_ms.saturating_sub(*at) < window_ms
44        })
45    }
46
47    pub fn record(
48        &mut self,
49        conversation: ConversationKey,
50        sequence: &MessageSequence,
51        now_ms: u64,
52    ) {
53        let fragment = sequence.fragment.map(|fragment| fragment.index);
54        self.record_frame(conversation, sequence.message_id, fragment, now_ms);
55    }
56
57    /// Record one frame by its archive coordinates, refreshing any existing
58    /// entry for the same frame.
59    pub fn record_frame(
60        &mut self,
61        conversation: ConversationKey,
62        message_id: u8,
63        fragment: Option<u8>,
64        now_ms: u64,
65    ) {
66        self.entries.retain(|(conv, id, frag, _)| {
67            !(*conv == conversation && *id == message_id && *frag == fragment)
68        });
69        if self.entries.is_full() {
70            self.entries.remove(0);
71        }
72        let _ = self
73            .entries
74            .push((conversation, message_id, fragment, now_ms));
75    }
76}
77
78/// SplitMix64: deterministic scheduling jitter.
79///
80/// This is scheduling randomness only — it desynchronizes group repair
81/// requests — and is never used as security material. Supplying the seed at
82/// construction keeps the reducer deterministic under test.
83#[derive(Clone, Debug)]
84pub struct JitterSource {
85    state: u64,
86}
87
88impl JitterSource {
89    pub fn new(seed: u64) -> Self {
90        Self { state: seed }
91    }
92
93    pub fn next_u64(&mut self) -> u64 {
94        self.state = self.state.wrapping_add(0x9E37_79B9_7F4A_7C15);
95        let mut value = self.state;
96        value = (value ^ (value >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
97        value = (value ^ (value >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
98        value ^ (value >> 31)
99    }
100
101    /// Uniform-ish value in `0..bound_ms` (0 when the bound is 0).
102    pub fn jitter_ms(&mut self, bound_ms: u64) -> u64 {
103        if bound_ms == 0 {
104            0
105        } else {
106            self.next_u64() % bound_ms
107        }
108    }
109}
umsh_text/engine/repair.rs

umsh_text/engine/
repair.rs
