tuwunel_service/rooms/event_handler/
backoff.rs

1use std::{ops::Range, time::Duration};
2
3use ruma::EventId;
4use tuwunel_core::{
5	implement,
6	utils::{
7		continue_exponential_backoff,
8		stream::{ReadyExt, TryIgnore},
9		time::now_secs,
10	},
11};
12use tuwunel_database::{Ignore, Interfix};
13
14/// Bucket width in seconds. Records within one bucket collide onto a single key
15/// (`<=` the smallest call-site backoff floor), coalescing concurrent failures.
16const QUANTUM: u64 = 60;
17
18/// Accumulated `Pending` records at which the rate brake engages.
19const SUPPRESS_AFTER: u32 = 3;
20
21/// Retry window for the `Upgrade` context.
22///
23/// Bounds how often a re-delivered event repeats the full upgrade. A
24/// soft-failed event is re-evaluated on this widening schedule rather than
25/// rejected forever, so a lapsed policy-server refusal heals.
26pub(super) const UPGRADE_RETRY: Range<Duration> =
27	Duration::from_mins(5)..Duration::from_hours(24);
28
29/// Federation step that recorded a decision; the key's leading discriminant.
30#[derive(Clone, Copy)]
31pub(super) enum Context {
32	Fetch = 0,
33	Auth = 1,
34	Upgrade = 2,
35}
36
37impl From<Context> for u8 {
38	#[inline]
39	fn from(context: Context) -> Self {
40		match context {
41			| Context::Fetch => 0,
42			| Context::Auth => 1,
43			| Context::Upgrade => 2,
44		}
45	}
46}
47
48/// Permanence of a recorded decision. Unknown discriminants decode to the
49/// weakest (`Pending`) so a future encoding can only soften, never wrongly
50/// escalate, a verdict against an old binary. `Permanent` is never written by
51/// this store.
52#[derive(Clone, Copy, Default)]
53pub(super) enum Disposition {
54	#[default]
55	Pending = 0,
56	Transient = 1,
57	Permanent = 2,
58}
59
60/// Verdict from consulting the store before a federation step.
61pub(super) enum Suppression {
62	Allow,
63	Deny,
64}
65
66#[derive(Default)]
67struct Summary {
68	total: u32,
69	pending: u32,
70	latest_secs: u64,
71	latest_class: Disposition,
72}
73
74impl From<u64> for Disposition {
75	#[inline]
76	fn from(disc: u64) -> Self {
77		match disc {
78			| 1 => Self::Transient,
79			| 2 => Self::Permanent,
80			| _ => Self::Pending,
81		}
82	}
83}
84
85impl From<Disposition> for u64 {
86	#[inline]
87	fn from(disposition: Disposition) -> Self {
88		match disposition {
89			| Disposition::Pending => 0,
90			| Disposition::Transient => 1,
91			| Disposition::Permanent => 2,
92		}
93	}
94}
95
96impl Suppression {
97	#[inline]
98	pub(super) fn is_deny(&self) -> bool { matches!(self, Self::Deny) }
99}
100
101impl Summary {
102	fn tally(mut self, (_, (class, secs)): (Ignore, (u64, u64))) -> Self {
103		let class = Disposition::from(class);
104
105		self.total = self.total.saturating_add(1);
106		if matches!(class, Disposition::Pending) {
107			self.pending = self.pending.saturating_add(1);
108		}
109
110		if secs >= self.latest_secs {
111			self.latest_secs = secs;
112			self.latest_class = class;
113		}
114
115		self
116	}
117}
118
119/// Record a federation attempt before a cancellable await, so a premature
120/// cancellation still leaves a `Pending` row behind to rate-gate against.
121#[implement(super::Service)]
122pub(super) fn record_attempt(&self, ctx: Context, event_id: &EventId) {
123	self.record_outcome(ctx, event_id, Disposition::Pending);
124}
125
126#[implement(super::Service)]
127pub(super) fn record_outcome(&self, ctx: Context, event_id: &EventId, disposition: Disposition) {
128	self.db.eventid_backoff.put(
129		(u8::from(ctx), event_id, current_bucket()),
130		(u64::from(disposition), now_secs()),
131	);
132}
133
134/// Clears the upgrade backoff recorded against an event.
135///
136/// The soft-fail marker and this backoff gate the same retry, so operator
137/// recovery has to drop both for the next delivery to evaluate the event
138/// without waiting out the window.
139#[implement(super::Service)]
140pub async fn clear_upgrade_backoff(&self, event_id: &EventId) {
141	self.record_success(Context::Upgrade, event_id)
142		.await;
143}
144
145#[implement(super::Service)]
146pub(super) async fn record_success(&self, ctx: Context, event_id: &EventId) {
147	self.db
148		.eventid_backoff
149		.del_prefix(&(u8::from(ctx), event_id, Interfix))
150		.await;
151}
152
153#[implement(super::Service)]
154pub(super) async fn is_suppressed(
155	&self,
156	ctx: Context,
157	event_id: &EventId,
158	range: Range<Duration>,
159) -> Suppression {
160	let summary = self
161		.db
162		.eventid_backoff
163		.stream_prefix::<Ignore, (u64, u64), _>(&(u8::from(ctx), event_id, Interfix))
164		.ignore_err()
165		.ready_fold(Summary::default(), Summary::tally)
166		.await;
167
168	if summary.total == 0 {
169		return Suppression::Allow;
170	}
171
172	if matches!(summary.latest_class, Disposition::Permanent) {
173		return Suppression::Deny;
174	}
175
176	let elapsed = Duration::from_secs(now_secs().saturating_sub(summary.latest_secs));
177	let (tries, rate_ok) = match summary.latest_class {
178		| Disposition::Pending => (summary.pending, summary.pending >= SUPPRESS_AFTER),
179		| _ => (summary.total, true),
180	};
181
182	(rate_ok && continue_exponential_backoff(range.start, range.end, elapsed, tries))
183		.then_some(Suppression::Deny)
184		.unwrap_or(Suppression::Allow)
185}
186
187fn current_bucket() -> u32 { u32::try_from(now_secs() / QUANTUM).unwrap_or(u32::MAX) }
tuwunel_service/rooms/event_handler/backoff.rs

tuwunel_service/rooms/event_handler/
backoff.rs
