iris-chat-protocol 0.3.17

Reusable Iris Chat double-ratchet protocol engine
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

use super::*;

// Keep real durable work in the measured windows even when missing-roster
// retries only update volatile backoff. The original 44 intents stay untouched.
pub(super) struct BatchedExpiringControls {
    controls: Vec<(String, u64, bool)>,
    pub(super) retired: usize,
}

impl BatchedExpiringControls {
    pub(super) fn seed(engine: &mut ProtocolEngine, recovery_epoch: u64, now: u64) -> Self {
        assert_eq!(engine.pending_remote_sends.len(), 44);
        let prototype = engine.pending_remote_sends[0].clone();
        assert!(engine
            .session_manager
            .roster(prototype.recipient_owner)
            .is_none());
        let mut controls = Vec::new();
        for (index, offset) in [210, 270].into_iter().enumerate() {
            let expires = recovery_epoch + offset;
            assert!(
                expires > now,
                "restore exceeded the control scheduling margin"
            );
            let id = format!("cpu-expiring-control-{index}");
            assert!(!engine
                .pending_remote_sends
                .iter()
                .any(|pending| pending.message_id == id));
            let mut rumor = pairwise_codec::typing_event(
                engine.owner_pubkey,
                pairwise_codec::EncodeOptions::new(now, now * 1_000).with_expiration(expires),
            )
            .unwrap();
            rumor.ensure_id();
            let mut pending = prototype.clone();
            pending.message_id = id.clone();
            pending.inner_event_id = rumor.id.map(|id| id.to_hex());
            pending.payload = serde_json::to_vec(&rumor).unwrap();
            pending.created_at_secs = now;
            pending.next_retry_at_secs = expires;
            pending.expires_at_secs = Some(expires);
            engine.pending_remote_sends.push(pending);
            controls.push((id, expires, false));
        }
        engine.persist().unwrap();
        Self {
            controls,
            retired: 0,
        }
    }

    pub(super) fn observe_turn(
        &mut self,
        engine: &ProtocolEngine,
        now: u64,
        puts_before: usize,
        puts_after: usize,
    ) {
        for (id, expires, retired) in &mut self.controls {
            let pending = engine
                .pending_remote_sends
                .iter()
                .find(|pending| pending.message_id == *id);
            if *retired {
                assert!(pending.is_none(), "retired control returned");
            } else if let Some(pending) = pending {
                assert_eq!(pending.expires_at_secs, Some(*expires));
                assert_eq!(pending.next_retry_at_secs, *expires);
            } else {
                assert!(now >= *expires, "control retired before expiry");
                assert!(
                    puts_after > puts_before,
                    "control retirement must flush in its own retry turn"
                );
                *retired = true;
                self.retired += 1;
            }
        }
    }
}