tuwunel_service/fetcher/
opts.rs

1//! Caller contract and result types for a fetch: [`Opts`] in, [`Outcome`] out.
2//!
3//! [`Op`] selects the federation endpoint and folds into the single-flight
4//! dedup key; [`FanoutGrowth`] schedules the staged fan-out width.
5
6use std::num::NonZeroUsize;
7
8use bytes::Bytes;
9use ruma::{
10	MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedServerName, RoomVersionId,
11	api::Direction,
12};
13use tuwunel_core::smallvec::SmallVec;
14
15use crate::federation::Candidates;
16
17/// Event-id window for the batch ops, inline-sized for the common single-prev
18/// case and spilling to the heap past that.
19pub type EventWindow = SmallVec<[OwnedEventId; 1]>;
20
21/// Federation endpoint a fetch targets. The dedup key folds this in, so two
22/// callers asking for the same event over different endpoints do not coalesce.
23#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
24pub enum Op {
25	/// `GET /_matrix/federation/v1/event/{eventId}`
26	Event,
27
28	/// `GET /_matrix/federation/v1/event/{eventId}` for an event fetched while
29	/// reconstructing an auth chain; routed like [`Op::Event`] but pins the
30	/// room's authority server ahead of the popularity ranking.
31	AuthEvent,
32
33	/// `GET /_matrix/federation/v1/event_auth/{roomId}/{eventId}`
34	AuthChain,
35
36	/// `GET /_matrix/federation/v1/backfill/{roomId}`
37	Backfill,
38
39	/// `GET /_matrix/federation/v1/state_ids/{roomId}?event_id=`
40	StateIds,
41
42	/// `POST /_matrix/federation/v1/get_missing_events/{roomId}`
43	MissingEvents,
44
45	/// `GET /_matrix/federation/v1/timestamp_to_event/{roomId}?ts=&dir=`
46	TimestampToEvent,
47}
48
49/// Schedules the concurrent candidate width for each staged fan-out round.
50///
51/// The worker clamps each computed width to its per-round ceiling and remaining
52/// attempt budget. `Fixed(1)`, the `Opts::new` default, makes attempts strictly
53/// sequential.
54#[derive(Clone, Copy, Debug, Eq, PartialEq)]
55pub enum FanoutGrowth {
56	/// Every round races the same width.
57	Fixed(NonZeroUsize),
58
59	/// `base`, `base + step`, `base + 2*step`, ...
60	Linear {
61		base: NonZeroUsize,
62		step: NonZeroUsize,
63	},
64
65	/// `base`, `base * factor`, `base * factor^2`, ...  Base 1, factor 2 is the
66	/// 1 -> 2 -> 4 -> 8 hedging ramp.
67	Geometric {
68		base: NonZeroUsize,
69		factor: NonZeroUsize,
70	},
71}
72
73impl FanoutGrowth {
74	/// Width for round `round` (0-based). Always >= 1; saturating, so a runaway
75	/// exponent cannot overflow (the candidate pool and `attempt_limit` clamp
76	/// the value to something small regardless).
77	#[must_use]
78	pub fn round_width(self, round: usize) -> usize {
79		match self {
80			| Self::Fixed(width) => width.get(),
81			| Self::Linear { base, step } => base
82				.get()
83				.saturating_add(step.get().saturating_mul(round)),
84			| Self::Geometric { base, factor } => {
85				let exp = u32::try_from(round).unwrap_or(u32::MAX);
86
87				base.get()
88					.saturating_mul(factor.get().saturating_pow(exp))
89			},
90		}
91	}
92}
93
94/// Caller contract. `event_id` is the sought datum for [`Op::Event`] /
95/// [`Op::AuthEvent`] / [`Op::AuthChain`] / [`Op::StateIds`] and a reference
96/// point for the others.
97#[derive(Clone, Debug)]
98pub struct Opts {
99	/// Federation endpoint this fetch targets.
100	pub op: Op,
101
102	/// Room the fetch is scoped to, or `None` for an unscoped id-addressed
103	/// fetch.
104	pub room_id: Option<OwnedRoomId>,
105
106	/// Event to fetch (id-addressed ops) or anchor from (room-scoped ops).
107	pub event_id: Option<OwnedEventId>,
108
109	/// Timestamp the [`Op::TimestampToEvent`] search starts from; `None` for
110	/// every other op.
111	pub ts: Option<MilliSecondsSinceUnixEpoch>,
112
113	/// Direction the [`Op::TimestampToEvent`] search runs; `None` for every
114	/// other op.
115	pub dir: Option<Direction>,
116
117	/// Boundary events the requester already holds; an [`Op::MissingEvents`]
118	/// window stops its backward walk here. Empty for every other op.
119	pub earliest_events: EventWindow,
120
121	/// Frontier events an [`Op::MissingEvents`] window fills the predecessors
122	/// of. Empty for every other op.
123	pub latest_events: EventWindow,
124
125	/// Server to try ahead of the ranked candidates.
126	pub hint: Option<OwnedServerName>,
127
128	/// Caller-supplied candidate pool tried in place of the room-derived
129	/// ranking; empty defers to the room-derived candidates.
130	pub candidates: Candidates,
131
132	/// Room version governing id and signature checks; `None` assumes V11.
133	pub room_version: Option<RoomVersionId>,
134
135	/// Cap on candidate servers tried; `None` tries every candidate.
136	pub attempt_limit: Option<NonZeroUsize>,
137
138	/// Event count requested per [`Op::Backfill`] / [`Op::MissingEvents`] batch
139	/// response; defaults to 10.
140	pub backfill_limit: Option<NonZeroUsize>,
141
142	/// Per-round width curve for staged fan-out. `Fixed(1)` is sequential.
143	pub fanout_growth: FanoutGrowth,
144
145	/// Per-round concurrency ceiling. `None` lets the curve run free, clamped
146	/// only by the candidate pool and `attempt_limit`; `Some(n)` caps each
147	/// round at `n`.
148	pub fanout_max_width: Option<NonZeroUsize>,
149
150	/// Cap on escalation rounds before giving up. `None` runs until exhaustion.
151	pub fanout_rounds: Option<NonZeroUsize>,
152
153	/// Reject a response whose event does not hash to the requested id.
154	pub check_event_id: bool,
155
156	/// Reject a response that is not well-formed JSON.
157	pub check_conforms: bool,
158
159	/// Reject a response that fails content-hash verification.
160	pub check_hashes: bool,
161
162	/// Accepted but not yet consulted; redaction-aware hash verification is
163	/// unimplemented.
164	pub authoritative_redaction: bool,
165
166	/// Reject a response that fails signature verification.
167	pub check_signature: bool,
168}
169
170impl Opts {
171	/// Scope a fetch to a room.
172	#[must_use]
173	pub fn new(op: Op, room_id: OwnedRoomId) -> Self { Self::with_room_id(op, Some(room_id)) }
174
175	/// A fetch with no room scope, for id-addressed callers such as
176	/// `get-remote-pdu`; room-derived candidate ranking is skipped, leaving the
177	/// hint, the caller-supplied pool, and the event id's origin.
178	#[must_use]
179	pub fn unscoped(op: Op) -> Self { Self::with_room_id(op, None) }
180
181	/// All validation toggles default on; the caller relaxes them per request.
182	fn with_room_id(op: Op, room_id: Option<OwnedRoomId>) -> Self {
183		Self {
184			op,
185			room_id,
186			event_id: None,
187			ts: None,
188			dir: None,
189			earliest_events: EventWindow::new(),
190			latest_events: EventWindow::new(),
191			hint: None,
192			candidates: Candidates::new(),
193			room_version: None,
194			attempt_limit: None,
195			backfill_limit: None,
196			fanout_growth: FanoutGrowth::Fixed(NonZeroUsize::MIN),
197			fanout_max_width: None,
198			fanout_rounds: None,
199			check_event_id: true,
200			check_conforms: true,
201			check_hashes: true,
202			authoritative_redaction: true,
203			check_signature: true,
204		}
205	}
206
207	/// Set the target event; required for the id-addressed ops.
208	#[must_use]
209	pub fn event_id(self, event_id: OwnedEventId) -> Self {
210		Self { event_id: Some(event_id), ..self }
211	}
212
213	/// Set the timestamp the [`Op::TimestampToEvent`] search starts from.
214	#[must_use]
215	pub fn ts(self, ts: MilliSecondsSinceUnixEpoch) -> Self { Self { ts: Some(ts), ..self } }
216
217	/// Set the direction the [`Op::TimestampToEvent`] search runs.
218	#[must_use]
219	pub fn dir(self, dir: Direction) -> Self { Self { dir: Some(dir), ..self } }
220
221	/// Set the boundary the [`Op::MissingEvents`] backward walk stops at.
222	#[must_use]
223	pub fn earliest_events<I>(self, earliest_events: I) -> Self
224	where
225		I: IntoIterator<Item = OwnedEventId>,
226	{
227		Self {
228			earliest_events: earliest_events.into_iter().collect(),
229			..self
230		}
231	}
232
233	/// Set the frontier an [`Op::MissingEvents`] window fills behind.
234	#[must_use]
235	pub fn latest_events<I>(self, latest_events: I) -> Self
236	where
237		I: IntoIterator<Item = OwnedEventId>,
238	{
239		Self {
240			latest_events: latest_events.into_iter().collect(),
241			..self
242		}
243	}
244
245	/// Try the named server ahead of the ranked candidates.
246	#[must_use]
247	pub fn hint(self, hint: OwnedServerName) -> Self { Self { hint: Some(hint), ..self } }
248
249	/// Supply the candidate pool verbatim, bypassing the room-derived ranking.
250	#[must_use]
251	pub fn candidates<I>(self, candidates: I) -> Self
252	where
253		I: IntoIterator<Item = OwnedServerName>,
254	{
255		Self {
256			candidates: candidates.into_iter().collect(),
257			..self
258		}
259	}
260
261	/// Room version for [`Op::Event`] id and signature checks. `None` keeps the
262	/// V11 default, so callers on a non-V11 room must name it to avoid a
263	/// spurious rejection.
264	#[must_use]
265	pub fn room_version(self, room_version: RoomVersionId) -> Self {
266		Self { room_version: Some(room_version), ..self }
267	}
268
269	/// Cap the number of candidate servers tried.
270	#[must_use]
271	pub fn attempt_limit(self, attempt_limit: NonZeroUsize) -> Self {
272		Self {
273			attempt_limit: Some(attempt_limit),
274			..self
275		}
276	}
277
278	/// Set the events requested per [`Op::Backfill`] / [`Op::MissingEvents`]
279	/// batch.
280	#[must_use]
281	pub fn backfill_limit(self, backfill_limit: NonZeroUsize) -> Self {
282		Self {
283			backfill_limit: Some(backfill_limit),
284			..self
285		}
286	}
287
288	/// Set the per-round fan-out width schedule.
289	#[must_use]
290	pub fn fanout(self, growth: FanoutGrowth) -> Self { Self { fanout_growth: growth, ..self } }
291
292	/// Cap the per-round fan-out concurrency.
293	#[must_use]
294	pub fn fanout_max_width(self, max_width: NonZeroUsize) -> Self {
295		Self {
296			fanout_max_width: Some(max_width),
297			..self
298		}
299	}
300
301	/// Cap the number of escalation rounds.
302	#[must_use]
303	pub fn fanout_rounds(self, rounds: NonZeroUsize) -> Self {
304		Self { fanout_rounds: Some(rounds), ..self }
305	}
306
307	/// Apply the op's advised staged-fan-out ramp. `Opts::new` is otherwise
308	/// dark on every op, so a callsite opts in by chaining this; the generic
309	/// and single-shot-batch ops keep the sequential default.
310	#[must_use]
311	pub fn fanout_for_op(self) -> Self {
312		use FanoutGrowth::{Geometric, Linear};
313
314		const ONE: NonZeroUsize = NonZeroUsize::new(1).unwrap();
315		const TWO: NonZeroUsize = NonZeroUsize::new(2).unwrap();
316		const THREE: NonZeroUsize = NonZeroUsize::new(3).unwrap();
317		const FOUR: NonZeroUsize = NonZeroUsize::new(4).unwrap();
318		const FIVE: NonZeroUsize = NonZeroUsize::new(5).unwrap();
319
320		match self.op {
321			| Op::AuthEvent => self
322				.fanout(Geometric { base: ONE, factor: TWO })
323				.fanout_max_width(FOUR)
324				.fanout_rounds(FIVE),
325			| Op::AuthChain => self
326				.fanout(Linear { base: ONE, step: ONE })
327				.fanout_max_width(TWO)
328				.fanout_rounds(TWO),
329			| Op::StateIds => self
330				.fanout(Linear { base: ONE, step: ONE })
331				.fanout_max_width(THREE)
332				.fanout_rounds(THREE),
333			| Op::MissingEvents => self
334				.fanout(Geometric { base: ONE, factor: TWO })
335				.fanout_rounds(THREE),
336			| Op::Event | Op::Backfill | Op::TimestampToEvent => self,
337		}
338	}
339
340	/// Toggle every validation gate at once. Callers that re-validate
341	/// downstream pass `false` to fetch raw bytes without rejecting non-V11
342	/// events.
343	#[must_use]
344	pub fn checks(self, enabled: bool) -> Self {
345		Self {
346			check_event_id: enabled,
347			check_conforms: enabled,
348			check_hashes: enabled,
349			check_signature: enabled,
350			..self
351		}
352	}
353}
354
355/// Raw response body plus the server that answered. `bytes` is ref-counted so
356/// concurrent callers coalesced onto one fetch share a single buffer.
357#[derive(Debug)]
358pub struct Outcome {
359	pub bytes: Bytes,
360	pub origin: OwnedServerName,
361}
tuwunel_service/fetcher/opts.rs

tuwunel_service/fetcher/
opts.rs
