tuwunel_admin/room/
clear_soft_failed_events.rs

1use ruma::{EventId, OwnedEventId, OwnedRoomOrAliasId};
2use tuwunel_core::{
3	Result, debug_warn,
4	matrix::Event,
5	utils::{ReadyExt, stream::BroadbandExt},
6};
7use tuwunel_service::Services;
8
9use crate::admin_command;
10
11#[derive(Default)]
12struct ClearSummary {
13	cleared: usize,
14	unreadable: usize,
15}
16
17#[admin_command]
18pub(super) async fn room_clear_soft_failed_events(&self, room_id: OwnedRoomOrAliasId) -> Result {
19	let room_id = self
20		.services
21		.alias
22		.maybe_resolve(&room_id)
23		.await?;
24
25	let _federation_lock = self
26		.services
27		.event_handler
28		.mutex_federation
29		.lock(&room_id)
30		.await;
31
32	let clear =
33		async |event_id: OwnedEventId| match self.services.timeline.get_pdu(&event_id).await {
34			| Ok(pdu) if pdu.room_id() == room_id => clear_event(self.services, &event_id).await,
35			| Ok(_) => ClearSummary::default(),
36			| Err(error) => {
37				debug_warn!(%event_id, %error, "Unable to read soft-failed event");
38				ClearSummary { unreadable: 1, ..Default::default() }
39			},
40		};
41
42	let ClearSummary { cleared, unreadable } = self
43		.services
44		.pdu_metadata
45		.soft_failed_event_ids()
46		.broad_then(clear)
47		.ready_fold(ClearSummary::default(), ClearSummary::merge)
48		.await;
49
50	write!(
51		self,
52		"Cleared {cleared} soft-failed event markers for {room_id} and removed any \
53		 corresponding cached policy decisions. Events will be checked again when federation \
54		 supplies them. Skipped {unreadable} markers whose events could not be read."
55	)
56	.await
57}
58
59/// Drops every stored verdict gating one event's re-evaluation.
60///
61/// The cached policy decision, the retry backoff, and the soft-fail marker each
62/// hold the next delivery back, so recovery has to clear all three. Dropping
63/// the backoff is what opens the gate, and it follows the policy decision, so
64/// an interrupted run never re-evaluates against a stale verdict.
65async fn clear_event(services: &Services, event_id: &EventId) -> ClearSummary {
66	services
67		.event_handler
68		.clear_policy_signature_state(event_id);
69
70	services
71		.event_handler
72		.clear_upgrade_backoff(event_id)
73		.await;
74
75	services
76		.pdu_metadata
77		.clear_event_soft_failed(event_id);
78
79	ClearSummary { cleared: 1, ..Default::default() }
80}
81
82impl ClearSummary {
83	fn merge(mut self, Self { cleared, unreadable }: Self) -> Self {
84		self.cleared = self.cleared.saturating_add(cleared);
85		self.unreadable = self.unreadable.saturating_add(unreadable);
86		self
87	}
88}
tuwunel_admin/room/clear_soft_failed_events.rs

tuwunel_admin/room/
clear_soft_failed_events.rs
