tuwunel_api_client/sync/v5/rooms/
collected.rs

1//! What one delivery read of a room's required state.
2//!
3//! The result decides the snapshot the delivery records and the keys it leaves
4//! for the next delivery to send.
5
6use tuwunel_core::{implement, utils::BoolExt};
7use tuwunel_service::{rooms::short::ShortStateHash, sync::UnreadState};
8
9use super::{CollectedState, StateEvent};
10
11/// One selected key's outcome: its event, if sent, or why it was left out.
12///
13/// A sent event carries its key's fingerprint when the last delivery had left
14/// that key unread.
15pub(super) type Outcome = Result<Option<(StateEvent, Option<u64>)>, Unread>;
16
17/// Why a selected key's event was left out of a delivery, when not by choice.
18///
19/// A due event that could not be read leaves its key for the next delivery to
20/// send; a key that could not be resolved makes the delivery record zero.
21pub(super) enum Unread {
22	/// The key could not be resolved, so whether its event was due is unknown.
23	Unresolved,
24
25	/// The event was due but could not be read; carries its key's fingerprint.
26	Unreadable(u64),
27}
28
29/// The required state a delivery read, folded from its candidates.
30///
31/// A selected key adds its event when sent, its fingerprint when its due event
32/// could not be read, or the mark of an unresolved key. A sent event whose key
33/// the last delivery left unread also adds that key's fingerprint.
34#[derive(Default)]
35pub(super) struct Collected {
36	/// The events read.
37	pub(super) events: Vec<StateEvent>,
38
39	/// Fingerprints of the keys whose due events could not be read.
40	pub(super) unread: UnreadState,
41
42	/// Fingerprints of keys the last delivery left unread whose events were sent.
43	pub(super) settled: UnreadState,
44
45	/// Whether a key could not be resolved.
46	pub(super) unresolved: bool,
47}
48
49#[implement(Collected)]
50pub(super) fn push(mut self, outcome: Outcome) -> Self {
51	match outcome {
52		| Ok(None) => {},
53		| Err(Unread::Unreadable(fingerprint)) => self.unread.push(fingerprint),
54		| Err(Unread::Unresolved) => self.unresolved = true,
55		| Ok(Some((event, settled))) => {
56			self.events.push(event);
57			self.settled.extend(settled);
58		},
59	}
60
61	self
62}
63
64/// The events read, the snapshot the delivery records, and the keys it leaves
65/// unread.
66///
67/// `previous` holds the keys the last delivery left unread. A complete delivery,
68/// one that compared every selected key against `snapshot` and resolved them
69/// all, keeps none of them; any other keeps those whose events it did not send.
70#[implement(Collected)]
71pub(super) fn finish(self, snapshot: Option<ShortStateHash>, previous: &[u64]) -> CollectedState {
72	let resolved = self.unresolved.is_false();
73	let complete = snapshot.is_some() && resolved;
74	let snapshot = recorded_snapshot(snapshot, resolved);
75	let previous = complete
76		.is_false()
77		.then_some(previous)
78		.unwrap_or_default();
79
80	let kept = previous
81		.iter()
82		.copied()
83		.filter(|fingerprint| !self.settled.contains(fingerprint));
84
85	let unread = unread_state(kept, self.unread);
86
87	(self.events, (snapshot, unread))
88}
89
90/// The snapshot a delivery records, given whether it resolved every selected key.
91///
92/// A key whose listing or current event could not be read may hide an event that
93/// was due, so no snapshot vouches for what the client holds: the delivery
94/// records zero, and the next delta compares against the cursor and resends
95/// outliers.
96pub(super) fn recorded_snapshot(
97	snapshot: Option<ShortStateHash>,
98	resolved: bool,
99) -> Option<ShortStateHash> {
100	resolved.copy_or(Some(0), snapshot)
101}
102
103/// The keys the next delivery sends whatever snapshot it compares against.
104///
105/// They are the keys whose due events this delivery could not read, and the
106/// `kept` keys an earlier delivery left unread, each once and in order.
107pub(super) fn unread_state(
108	kept: impl Iterator<Item = u64>,
109	mut unread: UnreadState,
110) -> UnreadState {
111	unread.extend(kept);
112	unread.sort_unstable();
113	unread.dedup();
114	unread.shrink_to_fit();
115	unread
116}
tuwunel_api_client/sync/v5/rooms/collected.rs

tuwunel_api_client/sync/v5/rooms/
collected.rs
