tuwunel_service/fetcher/
transport.rs

1//! Federation transport: an [`Op`] and target server in, raw response bytes
2//! out.
3//!
4//! The [`Transport`] seam isolates the network behind a trait the tests mock;
5//! [`FederationTransport`] is the production impl.
6
7use std::{num::NonZeroUsize, sync::Arc};
8
9use async_trait::async_trait;
10use bytes::Bytes;
11use ruma::{
12	MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, ServerName, UInt,
13	api::federation::{
14		authorization::get_event_authorization::v1::Request as EventAuthRequest,
15		backfill::get_backfill::v1::Request as BackfillRequest,
16		event::{
17			get_event::v1::Request as EventRequest,
18			get_event_by_timestamp::v1::Request as TimestampRequest,
19			get_missing_events::v1::Request as MissingEventsRequest,
20			get_room_state_ids::v1::Request as StateIdsRequest,
21		},
22	},
23};
24use tuwunel_core::{
25	Result, err,
26	utils::{BoolExt, math::ruma_from_usize_saturating},
27};
28
29use super::{Op, Opts};
30use crate::services::OnceServices;
31
32/// Abstracts the network operation for one federation fetch attempt.
33///
34/// The production implementation routes through federation execution while
35/// tests substitute a scripted mock.
36#[async_trait]
37pub(super) trait Transport: Send + Sync {
38	/// Executes one endpoint operation and returns its raw response body.
39	///
40	/// Required option fields are validated before the federation request is sent.
41	async fn fetch_raw(&self, op: Op, server: &ServerName, opts: &Opts) -> Result<Bytes>;
42}
43
44/// Production transport backed by the federation request service.
45///
46/// Each operation is converted to its corresponding ruma federation request.
47pub(super) struct FederationTransport {
48	/// Services used to execute typed federation requests.
49	pub(super) services: Arc<OnceServices>,
50}
51
52#[async_trait]
53impl Transport for FederationTransport {
54	#[tracing::instrument(
55		level = "debug",
56		skip(self, opts),
57		fields(
58			%server,
59		),
60	)]
61	async fn fetch_raw(&self, op: Op, server: &ServerName, opts: &Opts) -> Result<Bytes> {
62		let federation = &self.services.federation;
63
64		match op {
65			| Op::Event | Op::AuthEvent => {
66				let event_id = require_event_id(opts)?;
67				let res = federation
68					.execute(server, EventRequest { event_id })
69					.await?;
70
71				Ok(Bytes::copy_from_slice(res.pdu.get().as_bytes()))
72			},
73			| Op::AuthChain => {
74				let event_id = require_event_id(opts)?;
75				let room_id = require_room_id(opts)?;
76				let res = federation
77					.execute(server, EventAuthRequest { room_id, event_id })
78					.await?;
79
80				to_bytes(&res.auth_chain)
81			},
82			| Op::Backfill => {
83				let event_id = require_event_id(opts)?;
84				let room_id = require_room_id(opts)?;
85				let res = federation
86					.execute(server, BackfillRequest {
87						room_id,
88						v: vec![event_id],
89						limit: batch_limit(opts),
90					})
91					.await?;
92
93				to_bytes(&res.pdus)
94			},
95			| Op::StateIds => {
96				let event_id = require_event_id(opts)?;
97				let room_id = require_room_id(opts)?;
98				let res = federation
99					.execute(server, StateIdsRequest { room_id, event_id })
100					.await?;
101
102				to_bytes(&serde_json::json!({
103					"auth_chain_ids": res.auth_chain_ids,
104					"pdu_ids": res.pdu_ids,
105				}))
106			},
107			| Op::MissingEvents => {
108				require_latest_events(opts)?;
109				let room_id = require_room_id(opts)?;
110				let req = MissingEventsRequest {
111					room_id,
112					earliest_events: opts.earliest_events.to_vec(),
113					latest_events: opts.latest_events.to_vec(),
114					limit: batch_limit(opts),
115					min_depth: UInt::default(),
116				};
117
118				let res = federation.execute(server, req).await?;
119
120				to_bytes(&res.events)
121			},
122			| Op::TimestampToEvent => {
123				let room_id = require_room_id(opts)?;
124				let ts = require_ts(opts)?;
125				let res = federation
126					.execute(
127						server,
128						TimestampRequest::new(room_id, ts, opts.dir.unwrap_or_default()),
129					)
130					.await?;
131
132				to_bytes(&serde_json::json!({
133					"event_id": res.event_id,
134					"origin_server_ts": res.origin_server_ts,
135				}))
136			},
137		}
138	}
139}
140
141fn require_event_id(opts: &Opts) -> Result<OwnedEventId> {
142	opts.event_id
143		.clone()
144		.ok_or_else(|| err!(Request(InvalidParam("event_id is required for op {:?}", opts.op))))
145}
146
147fn require_room_id(opts: &Opts) -> Result<OwnedRoomId> {
148	opts.room_id
149		.clone()
150		.ok_or_else(|| err!(Request(InvalidParam("room_id is required for op {:?}", opts.op))))
151}
152
153fn require_ts(opts: &Opts) -> Result<MilliSecondsSinceUnixEpoch> {
154	opts.ts
155		.ok_or_else(|| err!(Request(InvalidParam("ts is required for op {:?}", opts.op))))
156}
157
158fn require_latest_events(opts: &Opts) -> Result {
159	opts.latest_events
160		.is_empty()
161		.is_false()
162		.then_some(())
163		.ok_or_else(|| {
164			err!(Request(InvalidParam("latest_events is required for op {:?}", opts.op)))
165		})
166}
167
168/// Event count requested per batch op, defaulting to the federation default of
169/// 10 and saturating an oversized cap to the wire `UInt`.
170fn batch_limit(opts: &Opts) -> UInt {
171	opts.backfill_limit
172		.map(NonZeroUsize::get)
173		.map_or_else(|| UInt::from(10_u8), ruma_from_usize_saturating)
174}
175
176fn to_bytes<T: serde::Serialize>(value: &T) -> Result<Bytes> {
177	serde_json::to_vec(value)
178		.map(Bytes::from)
179		.map_err(|e| err!(BadServerResponse("failed to re-encode federation response: {e}")))
180}
tuwunel_service/fetcher/transport.rs

tuwunel_service/fetcher/
transport.rs
