mirror of
https://forgejo.ellis.link/continuwuation/continuwuity.git
synced 2026-11-13 04:06:54 +00:00
fix: Throttle client-triggered remote event fetches
This commit is contained in:
1 parent
03f60c7002
commit
20d146bd5e
3 files changed
+46
-10
No files matched your search
@@ -0,0 +1 @@
|
||||
Fixed events being repeatedly re-requested over federation when they were fetched successfully but could not be resolved, and throttled remote event fetches made on behalf of clients. Contributed by @eleboucher.
|
||||
@@ -1,6 +1,5 @@
|
||||
use axum::extract::State;
|
||||
use conduwuit::{Err, Event, Result, debug_warn, err};
|
||||
use futures::{FutureExt, TryFutureExt, future::try_join};
|
||||
use ruma::api::client::room::get_room_event;
|
||||
|
||||
use crate::{Ruma, client::is_ignored_pdu};
|
||||
@@ -16,19 +15,39 @@ pub(crate) async fn get_room_event_route(
|
||||
let event_id = &body.event_id;
|
||||
let room_id = &body.room_id;
|
||||
|
||||
let event = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_remote_pdu(room_id, event_id)
|
||||
.map_err(|_| err!(Request(NotFound("Event {} not found.", event_id))));
|
||||
let mut event = match services.rooms.timeline.get_pdu(event_id).await {
|
||||
| Ok(event) => event,
|
||||
| Err(_) => {
|
||||
// Only fetch over federation for users who could see the event.
|
||||
if !services
|
||||
.rooms
|
||||
.state_cache
|
||||
.is_joined(sender_user, room_id)
|
||||
.await && !services
|
||||
.rooms
|
||||
.state_accessor
|
||||
.is_world_readable(room_id)
|
||||
.await
|
||||
{
|
||||
return Err!(Request(NotFound("Event {} not found.", event_id)));
|
||||
}
|
||||
|
||||
services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_remote_pdu(room_id, event_id)
|
||||
.await
|
||||
.map_err(|_| err!(Request(NotFound("Event {} not found.", event_id))))?
|
||||
},
|
||||
};
|
||||
|
||||
// NOTE: checked after fetching, as visibility of an unknown event cannot
|
||||
// be determined (user_can_see_event returns true for those).
|
||||
let visible = services
|
||||
.rooms
|
||||
.state_accessor
|
||||
.user_can_see_event(sender_user, room_id, event_id)
|
||||
.map(Ok);
|
||||
|
||||
let (mut event, visible) = try_join(event, visible).await?;
|
||||
.await;
|
||||
|
||||
if !visible || is_ignored_pdu(services, &event, sender_user).await? {
|
||||
return Err!(Request(Forbidden("You don't have permission to view this event.")));
|
||||
|
||||
@@ -120,6 +120,13 @@ impl super::Service {
|
||||
return Err!(Request(NotFound("No one can backfill this PDU, room is empty.")));
|
||||
}
|
||||
|
||||
self.services
|
||||
.event_handler
|
||||
.ensure_can_pull_event(event_id)
|
||||
.inspect_err(|e| {
|
||||
debug_warn!(%event_id, error=?e, "Not attempting to fetch remote PDU");
|
||||
})?;
|
||||
|
||||
let servers = self.candidate_backfill_servers(room_id).await;
|
||||
|
||||
for backfill_server in servers {
|
||||
@@ -145,7 +152,12 @@ impl super::Service {
|
||||
.event_handler
|
||||
.handle_incoming_pdu(&backfill_server, room_id, event_id, value, true)
|
||||
.boxed()
|
||||
.await?;
|
||||
.await
|
||||
.inspect_err(|_| {
|
||||
self.services
|
||||
.event_handler
|
||||
.hit_failed_pdu_pull(event_id.to_owned());
|
||||
})?;
|
||||
debug!("Successfully backfilled {event_id} from {backfill_server}");
|
||||
Some(self.get_pdu(event_id).await)
|
||||
},
|
||||
@@ -156,10 +168,14 @@ impl super::Service {
|
||||
};
|
||||
if let Some(pdu) = pdu {
|
||||
debug!("Fetched {event_id} from {backfill_server}");
|
||||
self.services.event_handler.clear_failed_pdu(event_id);
|
||||
return pdu;
|
||||
}
|
||||
}
|
||||
|
||||
self.services
|
||||
.event_handler
|
||||
.hit_failed_pdu_pull(event_id.to_owned());
|
||||
Err!("No servers could be used to fetch {} in {}.", room_id, event_id)
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user