iris_chat_protocol/protocol_engine/
engine_persistence.rs

1impl ProtocolEngine {
2    fn has_authoritative_local_roster(&self) -> bool {
3        if self.local_app_keys_observed {
4            return true;
5        }
6        self.session_manager
7            .snapshot()
8            .users
9            .into_iter()
10            .find(|user| user.owner_pubkey == self.local_owner)
11            .and_then(|user| user.roster)
12            .is_some_and(|roster| {
13                let devices = roster.devices();
14                !devices.is_empty()
15                    && (devices.len() > 1 || devices[0].device_pubkey != self.local_device)
16            })
17    }
18
19    #[cfg_attr(all(test, target_os = "linux"), track_caller)]
20    fn persist(&mut self) -> anyhow::Result<()> {
21        #[cfg(all(test, target_os = "linux"))]
22        tests::audit_retry_persist_requested(std::panic::Location::caller());
23        if self.batch_depth.get() > 0 {
24            self.batch_persist_dirty.set(true);
25            return Ok(());
26        }
27        self.persist_now()
28    }
29
30    #[cfg_attr(all(test, target_os = "linux"), track_caller)]
31    fn persist_now(&mut self) -> anyhow::Result<()> {
32        let state = ProtocolEnginePersistedState {
33            version: PROTOCOL_ENGINE_STATE_VERSION,
34            group_invite_uses: self.group_invite_uses.clone(),
35            session_manager: self.session_manager.snapshot(),
36            group_manager: self.group_manager.snapshot(),
37            verified_app_keys_owners: self.verified_app_keys_owners.clone(),
38            app_keys_provenance_version: PROTOCOL_APP_KEYS_PROVENANCE_VERSION,
39            invite_owner_app_keys_evidence: self.invite_owner_app_keys_evidence.clone(),
40            processed_private_invite_response_ids: self
41                .processed_private_invite_response_ids
42                .clone(),
43            pending_inbound: Vec::new(),
44            pending_group_fanouts: Vec::new(),
45            ready_group_publications: self.ready_group_publications.clone(),
46            pending_local_sibling_sends: self.pending_local_sibling_sends.clone(),
47            pending_remote_sends: self.pending_remote_sends.clone(),
48            pending_group_pairwise_payloads: self.pending_group_pairwise_payloads.clone(),
49            pending_group_sender_key_messages: PendingGroupMessages::default(),
50            pending_group_sender_key_repairs: self.pending_group_sender_key_repairs.clone(),
51            processed_group_sender_key_messages: self.processed_group_sender_key_messages.clone(),
52            answered_group_sender_key_repairs: self.answered_group_sender_key_repairs.clone(),
53            pending_decrypted_deliveries: self
54                .pending_decrypted_deliveries
55                .iter()
56                .filter(|delivery| !delivery.discarded)
57                .cloned()
58                .collect(),
59            group_roster_fact_histories: self.group_roster_fact_histories.clone(),
60            subscription_generation: self.subscription_generation,
61        };
62        let (checkpoint, layout) = record_protocol_checkpoint(
63            state,
64            &self.checkpoint_layout.borrow(),
65            &self.pending_group_sender_key_messages,
66            &self.pending_group_fanouts,
67            &self.pending_inbound,
68        )?;
69        self.storage
70            .put_checkpoint(PROTOCOL_ENGINE_STATE_KEY, &checkpoint)?;
71        #[cfg(all(test, target_os = "linux"))]
72        tests::audit_retry_persisted(std::panic::Location::caller(), self.batch_depth.get());
73        self.batch_persist_dirty.set(false);
74        self.pending_decrypted_deliveries
75            .retain(|delivery| !delivery.discarded);
76        *self.checkpoint_layout.borrow_mut() = layout;
77        Ok(())
78    }
79
80    pub fn enter_batch(&self) {
81        if self.batch_depth.get() == 0 {
82            self.group_sender_key_retry.borrow_mut().reset_budget();
83        }
84        self.batch_depth
85            .set(self.batch_depth.get().saturating_add(1));
86    }
87
88    pub fn exit_batch(&mut self) -> anyhow::Result<()> {
89        let depth = self.batch_depth.get();
90        if depth == 0 {
91            return Ok(());
92        }
93        self.batch_depth.set(depth - 1);
94        if self.batch_depth.get() == 0 && self.batch_persist_dirty.get() {
95            self.persist_now()?;
96        }
97        Ok(())
98    }
99}
100
101fn record_protocol_checkpoint(
102    core: ProtocolEnginePersistedState,
103    previous: &ProtocolCheckpointLayout,
104    pending: &PendingGroupMessages,
105    fanouts: &[ProtocolPendingGroupFanout],
106    inbound: &[ProtocolPendingInbound],
107) -> anyhow::Result<(crate::StorageCheckpoint, ProtocolCheckpointLayout)> {
108    let core_json = serde_json::to_string(&core)?;
109    let empty_prefix = format!(
110        "{{\"version\":{},\"pending_group_sender_key_messages\":[],\"pending_group_fanouts\":[],\"pending_inbound\":[]",
111        PROTOCOL_ENGINE_STATE_VERSION
112    );
113    // These are the first four fields of our own serializer, not untrusted
114    // substring matches. A schema/order change must update this contract.
115    anyhow::ensure!(
116        core_json.starts_with(&empty_prefix),
117        "unexpected checkpoint core prefix"
118    );
119    let core_previous = previous.core.as_deref();
120    let empty_layout = ProtocolCheckpointLayout::default();
121    let (core_json, core_layout) = layout_protocol_checkpoint_with_cached_fields(
122        core_json,
123        core_previous.unwrap_or(&empty_layout),
124        None,
125        &[],
126    );
127    let tail = core_json
128        .strip_prefix(&empty_prefix)
129        .ok_or_else(|| anyhow::anyhow!("unexpected checkpoint layout prefix"))?;
130    let mut value = crate::StorageCheckpoint::new();
131    value.push_inline(
132        "prefix",
133        format!(
134            "{{\"version\":{},\"pending_group_sender_key_messages\":[",
135            PROTOCOL_ENGINE_STATE_VERSION
136        ),
137        0,
138    );
139    for (index, body) in pending.record_bodies()?.into_iter().enumerate() {
140        value.push_record(
141            0,
142            body,
143            if index == 0 { "" } else { "," }.into(),
144            String::new(),
145            0,
146        );
147    }
148    value.push_inline("between", "],\"pending_group_fanouts\":[".into(), 0);
149    #[derive(Serialize)]
150    struct FanoutTail<'a> {
151        remaining_devices: &'a Option<BTreeSet<NdrDevicePubkey>>,
152        inner_event_id: &'a Option<String>,
153        created_at_secs: u64,
154        next_retry_at_secs: u64,
155    }
156    for (index, pending) in fanouts.iter().enumerate() {
157        let suffix = serde_json::to_string(&FanoutTail {
158            remaining_devices: &pending.remaining_devices,
159            inner_event_id: &pending.inner_event_id,
160            created_at_secs: pending.created_at_secs,
161            next_retry_at_secs: pending.next_retry_at_secs,
162        })?;
163        let prefix = format!(
164            "{}{{\"group_id\":{},\"fanout\":",
165            if index == 0 { "" } else { "," },
166            serde_json::to_string(&pending.group_id)?
167        );
168        value.push_record(
169            1,
170            pending.fanout.checkpoint_body()?,
171            prefix,
172            format!(",{}", &suffix[1..]),
173            0,
174        );
175    }
176    value.push_inline("before_inbound", "],\"pending_inbound\":[".into(), 0);
177    append_checkpoint_inbound(&mut value, inbound)?;
178    value.push_inline("core", format!("]{tail}"), 0);
179    let compact_bytes = value.len();
180    if compact_bytes < CHECKPOINT_RESERVE {
181        return Ok((value, ProtocolCheckpointLayout::default()));
182    }
183    // Padding belongs to the record, not its old ordinal. Append/removal
184    // cannot dirty the descriptors of every surviving backlog record.
185    value.pad_records(CHECKPOINT_PAGE, CHECKPOINT_PAGE / 2);
186    let reuse = previous.bytes <= compact_bytes.saturating_mul(3) + CHECKPOINT_RESERVE;
187    let target = if reuse && value.len() <= previous.bytes {
188        previous.bytes
189    } else {
190        value.len().div_ceil(CHECKPOINT_RESERVE) * CHECKPOINT_RESERVE
191    };
192    value.push_inline("reserve", String::new(), target - value.len());
193    let layout = ProtocolCheckpointLayout {
194        bytes: value.len(),
195        record_ends: Vec::new(),
196        core: Some(Box::new(core_layout)),
197    };
198    Ok((value, layout))
199}
iris_chat_protocol/protocol_engine/engine_persistence.rs

iris_chat_protocol/protocol_engine/
engine_persistence.rs
