iris-chat-protocol 0.1.12

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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

impl ProtocolEngine {
    fn prepare_pending_local_sibling_send(
        &mut self,
        pending: &mut ProtocolPendingLocalSiblingSend,
        now: NdrUnixSeconds,
    ) -> anyhow::Result<(Vec<ProtocolEffect>, bool)> {
        pending.next_retry_at_secs = next_pending_retry_at_secs(pending.created_at_secs, now);
        // The provisional one-device roster created at startup cannot prove
        // there are no other devices. Keep the intent until discovery finishes.
        if !self.local_app_keys_observed && !self.has_authoritative_local_roster() {
            return Ok((Vec::new(), false));
        }
        let snapshot = self.session_manager.snapshot();
        let Some(devices) =
            user_record_snapshot(&snapshot, self.local_owner).and_then(roster_device_pubkeys)
        else {
            return Ok((Vec::new(), false));
        };
        let targets = devices
            .into_iter()
            .filter(|device| {
                *device != self.local_device
                    && !pending.completed_devices.contains(device)
                    && pending
                        .eligible_devices
                        .as_ref()
                        .is_none_or(|eligible| eligible.contains(device))
            })
            .collect::<Vec<_>>();
        if targets.is_empty() {
            return Ok((Vec::new(), true));
        }
        let mut rng = OsRng;
        let mut ctx = ProtocolContext::new(now, &mut rng);
        let prepared = self.session_manager.prepare_local_sibling_send_to_devices(
            &mut ctx,
            targets,
            pending.payload.clone(),
        )?;
        let mut event_ids = Vec::new();
        let effects = protocol_effects_from_prepared(
            &prepared,
            self.local_handshake_owner_proof(),
            Some(pending.message_id.clone()),
            pending.chat_id.clone(),
            &mut event_ids,
        )?;
        pending.completed_devices.extend(
            prepared
                .deliveries
                .iter()
                .map(|delivery| delivery.device_pubkey),
        );
        Ok((effects, prepared.relay_gaps.is_empty()))
    }

    fn retry_pending_local_sibling_sends(
        &mut self,
        now: NdrUnixSeconds,
    ) -> anyhow::Result<Vec<ProtocolEffect>> {
        if !self
            .pending_local_sibling_sends
            .iter()
            .any(|pending| pending.next_retry_at_secs <= now.get())
        {
            return Ok(Vec::new());
        }
        self.with_state_checkpoint(|engine| {
            let mut effects = Vec::new();
            let mut processed = 0usize;
            for mut pending in std::mem::take(&mut engine.pending_local_sibling_sends) {
                if pending.next_retry_at_secs > now.get()
                    || processed >= PENDING_GROUP_FANOUT_RETRY_BATCH_SIZE
                {
                    engine.pending_local_sibling_sends.push(pending);
                    continue;
                }
                processed += 1;
                // A malformed session must not lose this or later queued work.
                let sessions = engine.session_manager.clone();
                match engine.prepare_pending_local_sibling_send(&mut pending, now) {
                    Ok((ready, complete)) => {
                        effects.extend(ready);
                        if !complete {
                            engine.pending_local_sibling_sends.push(pending);
                        }
                    }
                    Err(_) => {
                        engine.session_manager = sessions;
                        engine.pending_local_sibling_sends.push(pending);
                    }
                }
            }
            engine.persist()?;
            Ok(effects)
        })
    }
}