mirror of
https://forgejo.ellis.link/continuwuation/continuwuity/
synced 2026-05-12 21:05:43 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Renovate Bot
|
a71de68296 | ||
|
timedout
|
10dd8bebfe |
@@ -71,7 +71,7 @@ runs:
|
||||
|
||||
- name: Install timelord-cli and git-warp-time
|
||||
if: steps.check-binaries.outputs.need-install == 'true'
|
||||
uses: https://github.com/taiki-e/install-action@b5fddbb5361bce8a06fb168c9d403a6cc552b084 # v2
|
||||
uses: https://github.com/taiki-e/install-action@c070f87102a1c75b3183910f391c1cb887fe13c8 # v2
|
||||
with:
|
||||
tool: git-warp-time,timelord-cli@3.0.1
|
||||
|
||||
|
||||
@@ -1,9 +0,0 @@
|
||||
Implemented event rejection, which should resolve and prevent future netsplits of the kinds observed
|
||||
within some Continuwuity rooms.
|
||||
Also resolved several bugs related to both soft-failing events, and event backfilling, which should
|
||||
improve state resolution stability.
|
||||
The `!admin debug get-pdu` command was updated to disambiguate event acceptance status, and
|
||||
`!admin debug show-auth-chain` was added to visually display event auth chains, which may assist
|
||||
developers in debugging strangely complex events.
|
||||
|
||||
Contributed by @nex.
|
||||
+9
-213
@@ -1,5 +1,5 @@
|
||||
use std::{
|
||||
collections::{HashMap, HashSet},
|
||||
collections::HashMap,
|
||||
fmt::Write,
|
||||
iter::once,
|
||||
time::{Instant, SystemTime},
|
||||
@@ -22,7 +22,7 @@
|
||||
use lettre::message::Mailbox;
|
||||
use ruma::{
|
||||
CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId,
|
||||
OwnedRoomOrAliasId, OwnedServerName, RoomId, RoomVersionId, UInt,
|
||||
OwnedRoomOrAliasId, OwnedServerName, RoomId, RoomVersionId,
|
||||
api::federation::event::get_room_state, events::AnyStateEvent, serde::Raw,
|
||||
};
|
||||
use service::rooms::{
|
||||
@@ -69,189 +69,6 @@ pub(super) async fn get_auth_chain(&self, event_id: OwnedEventId) -> Result {
|
||||
self.write_str(&out).await
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Eq, PartialEq)]
|
||||
enum NodeStatus {
|
||||
Normal(bool),
|
||||
SoftFailed(bool),
|
||||
Rejected(bool),
|
||||
}
|
||||
|
||||
struct AuthChild {
|
||||
node_id: String,
|
||||
event_id: OwnedEventId,
|
||||
depth: UInt,
|
||||
ts: UInt,
|
||||
first_seen: bool,
|
||||
pdu: Option<PduEvent>,
|
||||
}
|
||||
|
||||
fn render_node(
|
||||
graph: &mut String,
|
||||
node_id: &str,
|
||||
event_id: &EventId,
|
||||
status: NodeStatus,
|
||||
) -> Result {
|
||||
let evt_str = event_id.to_string();
|
||||
|
||||
let status_label = match status {
|
||||
| NodeStatus::Normal(false) => evt_str,
|
||||
| NodeStatus::Normal(true) => format!("{evt_str} (missing locally)"),
|
||||
| NodeStatus::SoftFailed(false) => format!("{evt_str} (soft-failed)"),
|
||||
| NodeStatus::SoftFailed(true) => format!("{evt_str} (soft-failed & missing locally)"),
|
||||
| NodeStatus::Rejected(false) => format!("{evt_str} (rejected)"),
|
||||
| NodeStatus::Rejected(true) => format!("{evt_str} (rejected & missing locally)"),
|
||||
};
|
||||
|
||||
writeln!(graph, "{node_id}[\"{}\"]", status_label.as_str())?;
|
||||
|
||||
match status {
|
||||
| NodeStatus::Rejected(_) => writeln!(graph, "class {node_id} rejected;")?,
|
||||
| NodeStatus::SoftFailed(_) => writeln!(graph, "class {node_id} soft_failed;")?,
|
||||
| NodeStatus::Normal(_) => {},
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[admin_command]
|
||||
pub(super) async fn show_auth_chain(&self, event_id: OwnedEventId) -> Result {
|
||||
let node_status = async |event_id: &EventId, missing: bool| -> NodeStatus {
|
||||
if self
|
||||
.services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
.is_event_rejected(event_id)
|
||||
.await
|
||||
{
|
||||
NodeStatus::Rejected(missing)
|
||||
} else if self
|
||||
.services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
.is_event_soft_failed(event_id)
|
||||
.await
|
||||
{
|
||||
NodeStatus::SoftFailed(missing)
|
||||
} else {
|
||||
NodeStatus::Normal(missing)
|
||||
}
|
||||
};
|
||||
|
||||
let Ok(root) = self.services.rooms.timeline.get_pdu(&event_id).await else {
|
||||
return Err!("Event not found.");
|
||||
};
|
||||
|
||||
let mut graph = String::from(
|
||||
"```mermaid\n%% This is a mermaid graph. You can plug this output into\n\
|
||||
%% https://mermaid.live/edit to visualise it on-the-fly.\nflowchart TD\n\
|
||||
classDef rejected fill:#ffe5e5,stroke:#cc0000,stroke-width:2px,color:#000;\n\
|
||||
classDef soft_failed fill:#fff6cc,stroke:#c9a400,stroke-width:2px,color:#000;\n"
|
||||
);
|
||||
|
||||
let mut node_ids: HashMap<OwnedEventId, String> = HashMap::new();
|
||||
let mut cached_events: HashMap<OwnedEventId, PduEvent> =
|
||||
HashMap::from([(event_id.clone(), root.clone())]);
|
||||
let mut scheduled: HashSet<OwnedEventId> = HashSet::from([event_id.clone()]);
|
||||
let mut visited: HashSet<OwnedEventId> = HashSet::new();
|
||||
let mut stack = vec![root];
|
||||
let mut next_node_id = 0_usize;
|
||||
|
||||
let node_id_for = |event_id: &OwnedEventId,
|
||||
node_ids: &mut HashMap<OwnedEventId, String>,
|
||||
next_node_id: &mut usize| {
|
||||
node_ids
|
||||
.entry(event_id.clone())
|
||||
.or_insert_with(|| {
|
||||
let id = format!("n{}", *next_node_id);
|
||||
*next_node_id = next_node_id.saturating_add(1);
|
||||
id
|
||||
})
|
||||
.clone()
|
||||
};
|
||||
|
||||
while let Some(event) = stack.pop() {
|
||||
let current_event_id = event.event_id().to_owned();
|
||||
if !visited.insert(current_event_id.clone()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
let current_node_id = node_id_for(¤t_event_id, &mut node_ids, &mut next_node_id);
|
||||
let current_status = node_status(¤t_event_id, false).await;
|
||||
|
||||
render_node(&mut graph, ¤t_node_id, ¤t_event_id, current_status)?;
|
||||
|
||||
let mut children = Vec::with_capacity(event.auth_events.len());
|
||||
for auth_event_id in event.auth_events().rev() {
|
||||
let auth_event_id = auth_event_id.to_owned();
|
||||
let auth_node_id = node_id_for(&auth_event_id, &mut node_ids, &mut next_node_id);
|
||||
writeln!(graph, "{current_node_id} --> {auth_node_id}")?;
|
||||
|
||||
let first_seen = scheduled.insert(auth_event_id.clone());
|
||||
let auth_pdu = if let Some(auth_pdu) = cached_events.get(&auth_event_id) {
|
||||
// NOTE: events might be referenced multiple times (like the create event)
|
||||
// so this saves some cheeky db lookup time
|
||||
Some(auth_pdu.clone())
|
||||
} else if first_seen {
|
||||
match self.services.rooms.timeline.get_pdu(&auth_event_id).await {
|
||||
| Ok(auth_event) => {
|
||||
cached_events.insert(auth_event_id.clone(), auth_event.clone());
|
||||
Some(auth_event)
|
||||
},
|
||||
| Err(_) => None,
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// NOTE: Depth is used as the primary sorting key here, even though it has no
|
||||
// bearing on state resolution or anything. Timestamp is used as a
|
||||
// tiebreaker, failing back to lexicographical comparison.
|
||||
let (depth, ts) = auth_pdu
|
||||
.as_ref()
|
||||
.map_or((UInt::MAX, UInt::MAX), |pdu| (pdu.depth, pdu.origin_server_ts));
|
||||
|
||||
children.push(AuthChild {
|
||||
node_id: auth_node_id,
|
||||
event_id: auth_event_id,
|
||||
depth,
|
||||
ts,
|
||||
first_seen,
|
||||
pdu: auth_pdu,
|
||||
});
|
||||
}
|
||||
|
||||
children.sort_by(|a, b| {
|
||||
a.depth
|
||||
.cmp(&b.depth)
|
||||
.then(a.ts.cmp(&b.ts))
|
||||
.then(a.event_id.as_str().cmp(b.event_id.as_str()))
|
||||
});
|
||||
|
||||
for child in children.into_iter().rev() {
|
||||
if !child.first_seen {
|
||||
continue;
|
||||
}
|
||||
|
||||
if let Some(child_pdu) = child.pdu {
|
||||
// We have this PDU so will want to traverse it.
|
||||
stack.push(child_pdu);
|
||||
} else {
|
||||
// We don't have this PDU locally so we can't traverse its auth events,
|
||||
// but we can still render it as a node.
|
||||
render_node(
|
||||
&mut graph,
|
||||
&child.node_id,
|
||||
&child.event_id,
|
||||
node_status(&child.event_id, true).await,
|
||||
)?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
graph.push_str("```\n");
|
||||
self.write_str(&graph).await
|
||||
}
|
||||
|
||||
#[admin_command]
|
||||
pub(super) async fn parse_pdu(&self) -> Result {
|
||||
if self.body.len() < 2
|
||||
@@ -294,31 +111,15 @@ pub(super) async fn get_pdu(&self, event_id: OwnedEventId) -> Result {
|
||||
outlier = true;
|
||||
pdu_json = self.services.rooms.timeline.get_pdu_json(&event_id).await;
|
||||
}
|
||||
let rejected = self
|
||||
.services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
.is_event_rejected(&event_id)
|
||||
.await;
|
||||
let soft_failed = self
|
||||
.services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
.is_event_soft_failed(&event_id)
|
||||
.await;
|
||||
|
||||
match pdu_json {
|
||||
| Err(_) => return Err!("PDU not found locally."),
|
||||
| Ok(json) => {
|
||||
let text = serde_json::to_string_pretty(&json)?;
|
||||
let msg = if rejected {
|
||||
"Rejected PDU:"
|
||||
} else if soft_failed {
|
||||
"Soft-failed PDU:"
|
||||
} else if outlier {
|
||||
"Outlier PDU:"
|
||||
let msg = if outlier {
|
||||
"Outlier (Rejected / Soft Failed) PDU found in our database"
|
||||
} else {
|
||||
"PDU:"
|
||||
"PDU found in our database"
|
||||
};
|
||||
write!(self, "{msg}\n```json\n{text}\n```")
|
||||
},
|
||||
@@ -813,10 +614,6 @@ pub(super) async fn force_set_room_state_from_server(
|
||||
.await;
|
||||
|
||||
state.insert(shortstatekey, pdu.event_id.clone());
|
||||
self.services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
.clear_pdu_markers(pdu.event_id());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -834,10 +631,6 @@ pub(super) async fn force_set_room_state_from_server(
|
||||
.rooms
|
||||
.outlier
|
||||
.add_pdu_outlier(&event_id, &value);
|
||||
self.services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
.clear_pdu_markers(&event_id);
|
||||
}
|
||||
|
||||
info!("Resolving new room state");
|
||||
@@ -869,7 +662,10 @@ pub(super) async fn force_set_room_state_from_server(
|
||||
.force_state(room_id.clone().as_ref(), short_state_hash, added, removed, &state_lock)
|
||||
.await?;
|
||||
|
||||
info!("Updating joined counts for room");
|
||||
info!(
|
||||
"Updating joined counts for room just in case (e.g. we may have found a difference in \
|
||||
the room's m.room.member state"
|
||||
);
|
||||
self.services
|
||||
.rooms
|
||||
.state_cache
|
||||
|
||||
+1
-10
@@ -17,21 +17,12 @@ pub enum DebugCommand {
|
||||
message: Vec<String>,
|
||||
},
|
||||
|
||||
/// Loads the auth_chain of a PDU, reporting how long it took.
|
||||
/// Get the auth_chain of a PDU
|
||||
GetAuthChain {
|
||||
/// An event ID (the $ character followed by the base64 reference hash)
|
||||
event_id: OwnedEventId,
|
||||
},
|
||||
|
||||
/// Walks & displays the auth_chain of a PDU in a mermaid graph format.
|
||||
///
|
||||
/// This is useless to basically anyone but developers, and is also probably
|
||||
/// slow and memory hungry.
|
||||
ShowAuthChain {
|
||||
/// The root event ID to start walking back from.
|
||||
event_id: OwnedEventId,
|
||||
},
|
||||
|
||||
/// Parse and print a PDU from a JSON
|
||||
///
|
||||
/// The PDU event is only checked for validity and is not added to the
|
||||
|
||||
@@ -763,41 +763,6 @@ pub(super) async fn force_join_room(
|
||||
.await
|
||||
}
|
||||
|
||||
#[admin_command]
|
||||
pub(super) async fn force_join_room_remotely(
|
||||
&self,
|
||||
user_id: String,
|
||||
room_id: OwnedRoomOrAliasId,
|
||||
via: String,
|
||||
) -> Result {
|
||||
let via = ServerName::parse(&via)?;
|
||||
let user_id = parse_local_user_id(self.services, &user_id)?;
|
||||
let (room_id, mut servers) = self
|
||||
.services
|
||||
.rooms
|
||||
.alias
|
||||
.resolve_with_servers(&room_id, None)
|
||||
.await?;
|
||||
if servers.contains(&via) {
|
||||
servers.retain(|n| *n != via);
|
||||
}
|
||||
servers.insert(0, via);
|
||||
|
||||
assert!(
|
||||
self.services.globals.user_is_local(&user_id),
|
||||
"Parsed user_id must be a local user"
|
||||
);
|
||||
let state_lock = self.services.rooms.state.mutex.lock(&*room_id).await;
|
||||
self.services
|
||||
.rooms
|
||||
.membership
|
||||
.join_remote_room(&user_id, &room_id, None, &servers, state_lock)
|
||||
.await?;
|
||||
|
||||
self.write_str(&format!("{user_id} has been joined to {room_id}."))
|
||||
.await
|
||||
}
|
||||
|
||||
#[admin_command]
|
||||
pub(super) async fn force_leave_room(
|
||||
&self,
|
||||
|
||||
@@ -183,20 +183,6 @@ pub enum UserCommand {
|
||||
room_id: OwnedRoomOrAliasId,
|
||||
},
|
||||
|
||||
/// Manually join a local user to a room via a remote server, regardless of
|
||||
/// our current residency.
|
||||
ForceJoinRoomRemotely {
|
||||
/// The user to join
|
||||
user_id: String,
|
||||
/// The room to join
|
||||
room_id: OwnedRoomOrAliasId,
|
||||
/// The server name to join via.
|
||||
///
|
||||
/// This server will always be tried first, however if more are
|
||||
/// available, they may be tried after.
|
||||
via: String,
|
||||
},
|
||||
|
||||
/// Manually leave a local user from a room.
|
||||
ForceLeaveRoom {
|
||||
user_id: String,
|
||||
|
||||
@@ -187,7 +187,7 @@ pub(crate) async fn change_password_route(
|
||||
if services.server.config.admin_room_notices {
|
||||
services
|
||||
.admin
|
||||
.notice(&format!("User {sender_user} changed their password."))
|
||||
.notice(&format!("User {} changed their password.", &sender_user))
|
||||
.await;
|
||||
}
|
||||
|
||||
|
||||
@@ -21,10 +21,6 @@ pub(crate) async fn get_mutual_rooms_route(
|
||||
return Err!(Request(Unknown("You cannot request rooms in common with yourself.")));
|
||||
}
|
||||
|
||||
if !services.users.exists(&body.user_id).await {
|
||||
return Ok(mutual_rooms::unstable::Response::new(vec![]));
|
||||
}
|
||||
|
||||
let mutual_rooms = services
|
||||
.rooms
|
||||
.state_cache
|
||||
|
||||
@@ -537,7 +537,10 @@ pub(crate) async fn create_room_route(
|
||||
if services.server.config.admin_room_notices {
|
||||
services
|
||||
.admin
|
||||
.send_text(&format!("{sender_user} made {room_id} public to the room directory"))
|
||||
.send_text(&format!(
|
||||
"{sender_user} made {} public to the room directory",
|
||||
&room_id
|
||||
))
|
||||
.await;
|
||||
}
|
||||
info!("{sender_user} made {0} public to the room directory", &room_id);
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use axum::{Json, extract::State, response::IntoResponse};
|
||||
use conduwuit::Result;
|
||||
use conduwuit::{
|
||||
Result,
|
||||
matrix::versions::{unstable_features, versions},
|
||||
};
|
||||
use futures::StreamExt;
|
||||
use ruma::{api::client::discovery::get_supported_versions, assign};
|
||||
|
||||
@@ -22,47 +23,10 @@
|
||||
pub(crate) async fn get_supported_versions_route(
|
||||
_body: Ruma<get_supported_versions::Request>,
|
||||
) -> Result<get_supported_versions::Response> {
|
||||
let versions = vec![
|
||||
"r0.0.1".to_owned(),
|
||||
"r0.1.0".to_owned(),
|
||||
"r0.2.0".to_owned(),
|
||||
"r0.3.0".to_owned(),
|
||||
"r0.4.0".to_owned(),
|
||||
"r0.5.0".to_owned(),
|
||||
"r0.6.0".to_owned(),
|
||||
"r0.6.1".to_owned(),
|
||||
"v1.1".to_owned(),
|
||||
"v1.2".to_owned(),
|
||||
"v1.3".to_owned(),
|
||||
"v1.4".to_owned(),
|
||||
"v1.5".to_owned(),
|
||||
"v1.8".to_owned(),
|
||||
"v1.11".to_owned(),
|
||||
"v1.12".to_owned(),
|
||||
"v1.13".to_owned(),
|
||||
"v1.14".to_owned(),
|
||||
];
|
||||
|
||||
let unstable_features = BTreeMap::from_iter([
|
||||
("org.matrix.e2e_cross_signing".to_owned(), true),
|
||||
("org.matrix.msc2285.stable".to_owned(), true), /* private read receipts (https://github.com/matrix-org/matrix-spec-proposals/pull/2285) */
|
||||
("uk.half-shot.msc2666.query_mutual_rooms.stable".to_owned(), true), /* query mutual rooms (https://github.com/matrix-org/matrix-spec-proposals/pull/2666) */
|
||||
("org.matrix.msc2836".to_owned(), true), /* threading/threads (https://github.com/matrix-org/matrix-spec-proposals/pull/2836) */
|
||||
("org.matrix.msc2946".to_owned(), true), /* spaces/hierarchy summaries (https://github.com/matrix-org/matrix-spec-proposals/pull/2946) */
|
||||
("org.matrix.msc3026.busy_presence".to_owned(), true), /* busy presence status (https://github.com/matrix-org/matrix-spec-proposals/pull/3026) */
|
||||
("org.matrix.msc3814".to_owned(), true), /* dehydrated devices */
|
||||
("org.matrix.msc3827".to_owned(), true), /* filtering of /publicRooms by room type (https://github.com/matrix-org/matrix-spec-proposals/pull/3827) */
|
||||
("org.matrix.msc3952_intentional_mentions".to_owned(), true), /* intentional mentions (https://github.com/matrix-org/matrix-spec-proposals/pull/3952) */
|
||||
("org.matrix.msc3916.stable".to_owned(), true), /* authenticated media (https://github.com/matrix-org/matrix-spec-proposals/pull/3916) */
|
||||
("org.matrix.msc4180".to_owned(), true), /* stable flag for 3916 (https://github.com/matrix-org/matrix-spec-proposals/pull/4180) */
|
||||
("uk.tcpip.msc4133".to_owned(), true), /* Extending User Profile API with Key:Value Pairs (https://github.com/matrix-org/matrix-spec-proposals/pull/4133) */
|
||||
("us.cloke.msc4175".to_owned(), true), /* Profile field for user time zone (https://github.com/matrix-org/matrix-spec-proposals/pull/4175) */
|
||||
("org.matrix.simplified_msc3575".to_owned(), true), /* Simplified Sliding sync (https://github.com/matrix-org/matrix-spec-proposals/pull/4186) */
|
||||
("uk.timedout.msc4323".to_owned(), true), /* agnostic suspend (https://github.com/matrix-org/matrix-spec-proposals/pull/4323) */
|
||||
("org.matrix.msc4155".to_owned(), true), /* invite filtering (https://github.com/matrix-org/matrix-spec-proposals/pull/4155) */
|
||||
]);
|
||||
|
||||
Ok(assign!(get_supported_versions::Response::new(versions), { unstable_features }))
|
||||
Ok(assign!(
|
||||
get_supported_versions::Response::new(versions()),
|
||||
{ unstable_features: unstable_features() }
|
||||
))
|
||||
}
|
||||
|
||||
/// # `GET /_conduwuit/server_version`
|
||||
|
||||
@@ -260,7 +260,7 @@ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "{real_error}")
|
||||
}
|
||||
} else {
|
||||
write!(f, "Request error: {}", self.0)
|
||||
write!(f, "Request error: {}", &self.0)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -29,7 +29,7 @@ pub fn unstable_features() -> BTreeMap<String, bool> {
|
||||
BTreeMap::from_iter([
|
||||
("org.matrix.e2e_cross_signing".to_owned(), true),
|
||||
("org.matrix.msc2285.stable".to_owned(), true), /* private read receipts (https://github.com/matrix-org/matrix-spec-proposals/pull/2285) */
|
||||
("uk.half-shot.msc2666.query_mutual_rooms.stable".to_owned(), true), /* query mutual rooms (https://github.com/matrix-org/matrix-spec-proposals/pull/2666) */
|
||||
("uk.half-shot.msc2666.query_mutual_rooms".to_owned(), true), /* query mutual rooms (https://github.com/matrix-org/matrix-spec-proposals/pull/2666) */
|
||||
("org.matrix.msc2836".to_owned(), true), /* threading/threads (https://github.com/matrix-org/matrix-spec-proposals/pull/2836) */
|
||||
("org.matrix.msc2946".to_owned(), true), /* spaces/hierarchy summaries (https://github.com/matrix-org/matrix-spec-proposals/pull/2946) */
|
||||
("org.matrix.msc3026.busy_presence".to_owned(), true), /* busy presence status (https://github.com/matrix-org/matrix-spec-proposals/pull/3026) */
|
||||
|
||||
@@ -311,11 +311,6 @@ pub(super) fn open_list(db: &Arc<Engine>, maps: &[Descriptor]) -> Result<Maps> {
|
||||
key_size_hint: Some(48),
|
||||
..descriptor::RANDOM_SMALL
|
||||
},
|
||||
Descriptor {
|
||||
name: "rejectedeventids",
|
||||
key_size_hint: Some(48),
|
||||
..descriptor::RANDOM_SMALL
|
||||
},
|
||||
Descriptor {
|
||||
name: "statehash_shortstatehash",
|
||||
val_size_hint: Some(8),
|
||||
|
||||
@@ -44,7 +44,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
|
||||
db,
|
||||
server: args.server.clone(),
|
||||
bad_event_ratelimiter: Arc::new(SyncRwLock::new(HashMap::new())),
|
||||
admin_alias: OwnedRoomAliasId::try_from(format!("#admins:{}", args.server.name))
|
||||
admin_alias: OwnedRoomAliasId::try_from(format!("#admins:{}", &args.server.name))
|
||||
.expect("#admins:server_name is valid alias name"),
|
||||
server_user: UserId::parse_with_server_name(
|
||||
String::from("conduit"),
|
||||
|
||||
@@ -37,7 +37,7 @@ pub struct PasswordReset<'a> {
|
||||
}
|
||||
|
||||
impl MessageTemplate for PasswordReset<'_> {
|
||||
fn subject(&self) -> String { format!("Password reset request for {}", self.user_id) }
|
||||
fn subject(&self) -> String { format!("Password reset request for {}", &self.user_id) }
|
||||
}
|
||||
|
||||
#[derive(Template)]
|
||||
|
||||
@@ -48,7 +48,7 @@ pub fn is_valid(&self) -> bool {
|
||||
|
||||
impl std::fmt::Display for DatabaseTokenInfo {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "Token created by {} and used {} times. ", self.creator, self.uses)?;
|
||||
write!(f, "Token created by {} and used {} times. ", &self.creator, self.uses)?;
|
||||
if let Some(expires) = &self.expires {
|
||||
write!(f, "{expires}.")?;
|
||||
} else {
|
||||
|
||||
@@ -35,7 +35,7 @@ pub struct ValidToken {
|
||||
|
||||
impl std::fmt::Display for ValidToken {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "`{}` --- {}", self.token, self.source)
|
||||
write!(f, "`{}` --- {}", self.token, &self.source)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -215,17 +215,6 @@ pub async fn handle_incoming_pdu<'a>(
|
||||
.get_room_create_event(room_id)
|
||||
.await;
|
||||
|
||||
let start_time = Instant::now();
|
||||
self.federation_handletime
|
||||
.write()
|
||||
.insert(room_id.into(), (event_id.to_owned(), start_time));
|
||||
|
||||
defer! {{
|
||||
self.federation_handletime
|
||||
.write()
|
||||
.remove(room_id);
|
||||
}};
|
||||
|
||||
let (incoming_pdu, val) = self
|
||||
.handle_outlier_pdu(origin, create_event, event_id, room_id, value, false)
|
||||
.await?;
|
||||
@@ -292,6 +281,17 @@ pub async fn handle_incoming_pdu<'a>(
|
||||
.await?;
|
||||
|
||||
// Done with prev events, now handling the incoming event
|
||||
let start_time = Instant::now();
|
||||
self.federation_handletime
|
||||
.write()
|
||||
.insert(room_id.into(), (event_id.to_owned(), start_time));
|
||||
|
||||
defer! {{
|
||||
self.federation_handletime
|
||||
.write()
|
||||
.remove(room_id);
|
||||
}};
|
||||
|
||||
self.upgrade_outlier_to_timeline_pdu(incoming_pdu, val, create_event, origin, room_id)
|
||||
.boxed()
|
||||
.await
|
||||
|
||||
@@ -133,8 +133,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
|
||||
.filter(|id| !auth_events.contains_key(*id))
|
||||
.collect::<Vec<_>>();
|
||||
if !still_missing.is_empty() {
|
||||
// Don't reject: this could be a temporary condition
|
||||
// TODO: use get_missing_events?
|
||||
return Err!(Request(InvalidParam(
|
||||
"Could not fetch all auth events for outlier event {event_id}, still missing: \
|
||||
{still_missing:?}"
|
||||
@@ -165,10 +163,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
|
||||
v.insert(auth_event);
|
||||
},
|
||||
| hash_map::Entry::Occupied(_) => {
|
||||
self.services
|
||||
.outlier
|
||||
.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
|
||||
self.services.pdu_metadata.mark_event_rejected(event_id);
|
||||
return Err!(Request(InvalidParam(
|
||||
"Auth event's type and state_key combination exists multiple times: {}, {}",
|
||||
auth_event.kind,
|
||||
@@ -183,10 +177,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
|
||||
auth_events_by_key.get(&(StateEventType::RoomCreate, String::new().into())),
|
||||
Some(_) | None
|
||||
) {
|
||||
self.services.pdu_metadata.mark_event_rejected(event_id);
|
||||
self.services
|
||||
.outlier
|
||||
.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
|
||||
return Err!(Request(InvalidParam("Incoming event refers to wrong create event.")));
|
||||
}
|
||||
|
||||
@@ -195,7 +185,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
|
||||
ready(auth_events_by_key.get(&key).map(ToOwned::to_owned))
|
||||
};
|
||||
|
||||
// PDU check: 3
|
||||
let auth_check = state_res::event_auth::auth_check(
|
||||
&room_version_rules,
|
||||
&pdu_event,
|
||||
@@ -207,13 +196,7 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
|
||||
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
|
||||
|
||||
if !auth_check {
|
||||
self.services.pdu_metadata.mark_event_rejected(event_id);
|
||||
self.services
|
||||
.outlier
|
||||
.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
|
||||
return Err!(Request(Forbidden(
|
||||
"Event authorisation fails based on event's claimed auth events"
|
||||
)));
|
||||
return Err!(Request(Forbidden("Auth check failed")));
|
||||
}
|
||||
|
||||
trace!("Validation successful.");
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
};
|
||||
|
||||
use conduwuit::{
|
||||
Result, debug, err, error, implement,
|
||||
Result, debug, err, implement,
|
||||
matrix::{Event, StateMap},
|
||||
trace,
|
||||
utils::stream::{BroadbandExt, IterStream, ReadyExt, TryBroadbandExt, TryWidebandExt},
|
||||
@@ -121,7 +121,6 @@ pub(super) async fn state_at_incoming_resolved<Pdu>(
|
||||
.state_resolution(room_version_rules, fork_states.iter(), &auth_chain_sets)
|
||||
.boxed()
|
||||
.await
|
||||
.inspect_err(|e| error!("State resolution failed: {e:?}"))
|
||||
else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
@@ -1,18 +1,14 @@
|
||||
use std::{borrow::Borrow, collections::BTreeMap, sync::Arc, time::Instant};
|
||||
use std::{borrow::Borrow, collections::BTreeMap, iter::once, sync::Arc, time::Instant};
|
||||
|
||||
use conduwuit::{
|
||||
Err, Result, debug, debug_info, debug_warn, err, implement, is_equal_to,
|
||||
Err, Result, debug, debug_info, err, implement, info, is_equal_to,
|
||||
matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
|
||||
trace,
|
||||
utils::{
|
||||
IterStream,
|
||||
stream::{BroadbandExt, ReadyExt},
|
||||
},
|
||||
utils::stream::{BroadbandExt, ReadyExt},
|
||||
warn,
|
||||
};
|
||||
use futures::{FutureExt, StreamExt, future::ready};
|
||||
use ruma::{CanonicalJsonValue, RoomId, ServerName, events::StateEventType};
|
||||
use tokio::join;
|
||||
|
||||
use super::get_room_version_rules;
|
||||
use crate::rooms::{
|
||||
@@ -42,33 +38,13 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
|
||||
return Ok(Some(pduid));
|
||||
}
|
||||
|
||||
let (rejected, soft_failed) = join!(
|
||||
self.services
|
||||
.pdu_metadata
|
||||
.is_event_rejected(incoming_pdu.event_id()),
|
||||
self.services
|
||||
.pdu_metadata
|
||||
.is_event_soft_failed(incoming_pdu.event_id())
|
||||
);
|
||||
if rejected {
|
||||
return Err!(Request(InvalidParam("Event has been rejected")));
|
||||
} else if soft_failed {
|
||||
return Err!(Request(InvalidParam("Event has been soft-failed")));
|
||||
}
|
||||
|
||||
// If any of the auth events are rejected, this event is also rejected.
|
||||
for aid in incoming_pdu.auth_events() {
|
||||
if self.services.pdu_metadata.is_event_rejected(aid).await {
|
||||
// TODO: debug_warn instead of warn
|
||||
warn!(
|
||||
"Rejecting incoming event {} which depends on rejected auth event {aid}",
|
||||
incoming_pdu.event_id()
|
||||
);
|
||||
self.services
|
||||
.pdu_metadata
|
||||
.mark_event_rejected(incoming_pdu.event_id());
|
||||
return Err!(Request(InvalidParam("Event has rejected auth event: {aid}")));
|
||||
}
|
||||
if self
|
||||
.services
|
||||
.pdu_metadata
|
||||
.is_event_soft_failed(incoming_pdu.event_id())
|
||||
.await
|
||||
{
|
||||
return Err!(Request(InvalidParam("Event has been soft failed")));
|
||||
}
|
||||
|
||||
debug!(
|
||||
@@ -119,7 +95,6 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
|
||||
event_id = %incoming_pdu.event_id,
|
||||
"Running initial auth check"
|
||||
);
|
||||
// PDU check: 5
|
||||
let auth_check = state_res::event_auth::auth_check(
|
||||
&room_version_rules,
|
||||
&incoming_pdu,
|
||||
@@ -131,12 +106,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
|
||||
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
|
||||
|
||||
if !auth_check {
|
||||
self.services
|
||||
.pdu_metadata
|
||||
.mark_event_rejected(incoming_pdu.event_id());
|
||||
return Err!(Request(Forbidden(
|
||||
"Event authorisation fails based on the state before the event"
|
||||
)));
|
||||
return Err!(Request(Forbidden("Event has failed auth check with state at the event.")));
|
||||
}
|
||||
|
||||
debug!(
|
||||
@@ -165,7 +135,6 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
|
||||
event_id = %incoming_pdu.event_id,
|
||||
"Running auth check with claimed state auth"
|
||||
);
|
||||
// PDU check: 6
|
||||
let auth_check = state_res::event_auth::auth_check(
|
||||
&room_version_rules,
|
||||
&incoming_pdu,
|
||||
@@ -175,12 +144,6 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
|
||||
)
|
||||
.await
|
||||
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
|
||||
if !auth_check {
|
||||
warn!(
|
||||
event_id = %incoming_pdu.event_id,
|
||||
"Event authentication fails based on the current state of the room"
|
||||
);
|
||||
}
|
||||
|
||||
// Soft fail check before doing state res
|
||||
debug!(
|
||||
@@ -190,22 +153,16 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
|
||||
let mut soft_fail = match (auth_check, incoming_pdu.redacts_id(&room_version_rules)) {
|
||||
| (false, _) => true,
|
||||
| (true, None) => false,
|
||||
| (true, Some(redact_id)) => {
|
||||
if !self
|
||||
| (true, Some(redact_id)) =>
|
||||
!self
|
||||
.services
|
||||
.state_accessor
|
||||
.user_can_redact(&redact_id, incoming_pdu.sender(), room_id, true)
|
||||
.await?
|
||||
{
|
||||
warn!(redacts = %redact_id, "User is not allowed to redact event");
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
},
|
||||
.await?,
|
||||
};
|
||||
|
||||
// 13. Use state resolution to find new room state
|
||||
|
||||
// We start looking at current room state now, so lets lock the room
|
||||
trace!(
|
||||
room_id = %room_id,
|
||||
@@ -213,6 +170,36 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
|
||||
);
|
||||
let state_lock = self.services.state.mutex.lock(room_id).await;
|
||||
|
||||
// Now we calculate the set of extremities this room has after the incoming
|
||||
// event has been applied. We start with the previous extremities (aka leaves)
|
||||
trace!("Calculating extremities");
|
||||
let mut extremities: Vec<_> = self
|
||||
.services
|
||||
.state
|
||||
.get_forward_extremities(room_id)
|
||||
.ready_filter(|event_id| {
|
||||
// Remove any that are referenced by this incoming event's prev_events
|
||||
!incoming_pdu.prev_events().any(is_equal_to!(event_id))
|
||||
})
|
||||
.broad_filter_map(|event_id| async move {
|
||||
// Only keep those extremities were not referenced yet
|
||||
self.services
|
||||
.pdu_metadata
|
||||
.is_event_referenced(room_id, &event_id)
|
||||
.await
|
||||
.eq(&false)
|
||||
.then_some(event_id)
|
||||
})
|
||||
.collect()
|
||||
.await;
|
||||
extremities.push(incoming_pdu.event_id().to_owned());
|
||||
|
||||
debug!(
|
||||
"Retained {} extremities checked against {} prev_events",
|
||||
extremities.len(),
|
||||
incoming_pdu.prev_events().count()
|
||||
);
|
||||
|
||||
let state_ids_compressed: Arc<CompressedState> = self
|
||||
.services
|
||||
.state_compressor
|
||||
@@ -307,88 +294,81 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
|
||||
.is_event_soft_failed(&redact_id)
|
||||
.await
|
||||
{
|
||||
// TODO: This should avoid pushing the event to the timeline instead of using
|
||||
// soft-fails as a hack
|
||||
warn!(
|
||||
redact_id = %redact_id,
|
||||
"Redaction is for a soft-failed event"
|
||||
"Redaction is for a soft-failed event, soft failing the redaction"
|
||||
);
|
||||
soft_fail = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
trace!("Appending pdu to timeline");
|
||||
let mut extremities: Vec<_> = self
|
||||
.services
|
||||
.state
|
||||
.get_forward_extremities(room_id)
|
||||
.collect()
|
||||
.await;
|
||||
if !soft_fail {
|
||||
// Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events
|
||||
// are not added as forward extremities.
|
||||
|
||||
// Now we calculate the set of extremities this room has after the incoming
|
||||
// event has been applied. We start with the previous extremities (aka leaves)
|
||||
trace!("Calculating extremities");
|
||||
extremities = extremities
|
||||
.into_iter()
|
||||
.stream()
|
||||
.ready_filter(|event_id| {
|
||||
// Remove any that are referenced by this incoming event's prev_events
|
||||
!incoming_pdu.prev_events().any(is_equal_to!(event_id))
|
||||
})
|
||||
.broad_filter_map(|event_id| async move {
|
||||
// Only keep those extremities were not referenced yet
|
||||
self.services
|
||||
.pdu_metadata
|
||||
.is_event_referenced(room_id, &event_id)
|
||||
.await
|
||||
.eq(&false)
|
||||
.then_some(event_id)
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
.await;
|
||||
extremities.push(incoming_pdu.event_id().to_owned());
|
||||
debug!(
|
||||
"Retained {} extremities checked against {} prev_events",
|
||||
extremities.len(),
|
||||
incoming_pdu.prev_events().count()
|
||||
// 14. Check if the event passes auth based on the "current state" of the room,
|
||||
// if not soft fail it
|
||||
if soft_fail {
|
||||
info!(
|
||||
event_id = %incoming_pdu.event_id,
|
||||
"Soft failing event"
|
||||
);
|
||||
assert!(!extremities.is_empty(), "extremities must not empty");
|
||||
// assert!(extremities.is_empty(), "soft_fail extremities empty");
|
||||
let extremities = extremities.iter().map(Borrow::borrow);
|
||||
debug_assert!(extremities.clone().count() > 0, "extremities not empty");
|
||||
|
||||
self.services
|
||||
.timeline
|
||||
.append_incoming_pdu(
|
||||
&incoming_pdu,
|
||||
val,
|
||||
extremities,
|
||||
state_ids_compressed,
|
||||
soft_fail,
|
||||
&state_lock,
|
||||
room_id,
|
||||
)
|
||||
.await?;
|
||||
|
||||
// Soft fail, we keep the event as an outlier but don't add it to the timeline
|
||||
self.services
|
||||
.pdu_metadata
|
||||
.mark_event_soft_failed(incoming_pdu.event_id());
|
||||
|
||||
warn!(
|
||||
event_id = %incoming_pdu.event_id,
|
||||
"Event was soft failed"
|
||||
);
|
||||
return Err!(Request(InvalidParam("Event has been soft failed")));
|
||||
}
|
||||
|
||||
// Now that the event has passed all auth it is added into the timeline.
|
||||
// We use the `state_at_event` instead of `state_after` so we accurately
|
||||
// represent the state for this event.
|
||||
trace!("Appending pdu to timeline");
|
||||
let extremities = extremities
|
||||
.iter()
|
||||
.map(Borrow::borrow)
|
||||
.chain(once(incoming_pdu.event_id()));
|
||||
debug_assert!(extremities.clone().count() > 0, "extremities not empty");
|
||||
|
||||
let pdu_id = self
|
||||
.services
|
||||
.timeline
|
||||
.append_incoming_pdu(
|
||||
&incoming_pdu,
|
||||
val,
|
||||
extremities.iter().map(Borrow::borrow),
|
||||
extremities,
|
||||
state_ids_compressed,
|
||||
soft_fail,
|
||||
&state_lock,
|
||||
room_id,
|
||||
)
|
||||
.await?;
|
||||
if soft_fail {
|
||||
self.services
|
||||
.pdu_metadata
|
||||
.mark_event_soft_failed(incoming_pdu.event_id());
|
||||
debug_warn!(
|
||||
elapsed = ?timer.elapsed(),
|
||||
"Event has been soft-failed",
|
||||
);
|
||||
} else {
|
||||
debug_info!(
|
||||
elapsed = ?timer.elapsed(),
|
||||
"Accepted",
|
||||
);
|
||||
}
|
||||
|
||||
// Event has passed all auth/stateres checks
|
||||
drop(state_lock);
|
||||
debug_info!(
|
||||
elapsed = ?timer.elapsed(),
|
||||
"Accepted",
|
||||
);
|
||||
|
||||
Ok(pdu_id)
|
||||
}
|
||||
|
||||
@@ -34,7 +34,7 @@
|
||||
use crate::{
|
||||
Dep, antispam, globals,
|
||||
rooms::{
|
||||
metadata, outlier, pdu_metadata, short,
|
||||
metadata, outlier, short,
|
||||
state::{self, RoomMutexGuard},
|
||||
state_accessor, state_cache,
|
||||
state_compressor::{self, CompressedState, HashSetCompressStateEvent},
|
||||
@@ -54,7 +54,6 @@ struct Services {
|
||||
globals: Dep<globals::Service>,
|
||||
metadata: Dep<metadata::Service>,
|
||||
outlier: Dep<outlier::Service>,
|
||||
pdu_metadata: Dep<pdu_metadata::Service>,
|
||||
sending: Dep<sending::Service>,
|
||||
server_keys: Dep<server_keys::Service>,
|
||||
short: Dep<short::Service>,
|
||||
@@ -76,7 +75,6 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
|
||||
globals: args.depend::<globals::Service>("globals"),
|
||||
metadata: args.depend::<metadata::Service>("metadata"),
|
||||
outlier: args.depend::<outlier::Service>("rooms::outlier"),
|
||||
pdu_metadata: args.depend::<pdu_metadata::Service>("rooms::pdu_metadata"),
|
||||
sending: args.depend::<sending::Service>("sending"),
|
||||
server_keys: args.depend::<server_keys::Service>("server_keys"),
|
||||
short: args.depend::<short::Service>("rooms::short"),
|
||||
@@ -288,7 +286,7 @@ async fn join_local_room(
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all, fields(%sender_user, %room_id), name = "join_remote_room", level = "info")]
|
||||
pub async fn join_remote_room(
|
||||
async fn join_remote_room(
|
||||
&self,
|
||||
sender_user: &UserId,
|
||||
room_id: &RoomId,
|
||||
@@ -296,7 +294,6 @@ pub async fn join_remote_room(
|
||||
servers: &[OwnedServerName],
|
||||
state_lock: RoomMutexGuard,
|
||||
) -> Result {
|
||||
// public so the admin command force-join-room-remotely works
|
||||
info!("Joining {room_id} over federation.");
|
||||
|
||||
let (make_join_response, remote_server) = self
|
||||
@@ -517,7 +514,6 @@ pub async fn join_remote_room(
|
||||
return state;
|
||||
}
|
||||
self.services.outlier.add_pdu_outlier(&event_id, &value);
|
||||
self.services.pdu_metadata.clear_pdu_markers(&event_id);
|
||||
if let Some(state_key) = &pdu.state_key {
|
||||
let shortstatekey = self
|
||||
.services
|
||||
@@ -549,7 +545,6 @@ pub async fn join_remote_room(
|
||||
.ready_for_each(|(event_id, value)| {
|
||||
trace!(%event_id, "Adding PDU as an outlier from send_join auth_chain");
|
||||
self.services.outlier.add_pdu_outlier(&event_id, &value);
|
||||
self.services.pdu_metadata.clear_pdu_markers(&event_id);
|
||||
})
|
||||
.await;
|
||||
|
||||
|
||||
@@ -26,7 +26,6 @@ pub(super) struct Data {
|
||||
tofrom_relation: Arc<Map>,
|
||||
referencedevents: Arc<Map>,
|
||||
softfailedeventids: Arc<Map>,
|
||||
rejectedeventids: Arc<Map>,
|
||||
services: Services,
|
||||
}
|
||||
|
||||
@@ -41,7 +40,6 @@ pub(super) fn new(args: &crate::Args<'_>) -> Self {
|
||||
tofrom_relation: db["tofrom_relation"].clone(),
|
||||
referencedevents: db["referencedevents"].clone(),
|
||||
softfailedeventids: db["softfailedeventids"].clone(),
|
||||
rejectedeventids: db["rejectedeventids"].clone(),
|
||||
services: Services {
|
||||
timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
|
||||
},
|
||||
@@ -120,29 +118,7 @@ pub(super) fn mark_event_soft_failed(&self, event_id: &EventId) {
|
||||
self.softfailedeventids.insert(event_id, []);
|
||||
}
|
||||
|
||||
pub(super) fn unmark_event_soft_failed(&self, event_id: &EventId) {
|
||||
self.softfailedeventids.remove(event_id);
|
||||
}
|
||||
|
||||
pub(super) async fn is_event_soft_failed(&self, event_id: &EventId) -> bool {
|
||||
self.softfailedeventids.get(event_id).await.is_ok()
|
||||
}
|
||||
|
||||
pub(super) fn mark_event_rejected(&self, event_id: &EventId) {
|
||||
self.rejectedeventids.insert(event_id, []);
|
||||
}
|
||||
|
||||
pub(super) fn unmark_event_rejected(&self, event_id: &EventId) {
|
||||
self.rejectedeventids.remove(event_id);
|
||||
}
|
||||
|
||||
pub(super) async fn is_event_rejected(&self, event_id: &EventId) -> bool {
|
||||
self.rejectedeventids.get(event_id).await.is_ok()
|
||||
}
|
||||
|
||||
/// Removes any soft-fail or rejection markers applied to the target PDU
|
||||
pub(super) fn clear_pdu_markers(&self, event_id: &EventId) {
|
||||
self.unmark_event_rejected(event_id);
|
||||
self.unmark_event_soft_failed(event_id);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -140,28 +140,4 @@ pub fn mark_event_soft_failed(&self, event_id: &EventId) {
|
||||
pub async fn is_event_soft_failed(&self, event_id: &EventId) -> bool {
|
||||
self.db.is_event_soft_failed(event_id).await
|
||||
}
|
||||
|
||||
pub async fn is_event_rejected(&self, event_id: &EventId) -> bool {
|
||||
self.db.is_event_rejected(event_id).await
|
||||
}
|
||||
|
||||
pub fn mark_event_rejected(&self, event_id: &EventId) {
|
||||
self.db.mark_event_rejected(event_id);
|
||||
}
|
||||
|
||||
pub fn unmark_event_soft_failed(&self, event_id: &EventId) {
|
||||
self.db.unmark_event_soft_failed(event_id);
|
||||
}
|
||||
|
||||
pub fn unmark_event_rejected(&self, event_id: &EventId) {
|
||||
self.db.unmark_event_rejected(event_id);
|
||||
}
|
||||
|
||||
/// Returns true if the event is neither soft-failed nor rejected.
|
||||
pub async fn is_event_accepted(&self, event_id: &EventId) -> bool {
|
||||
!self.db.is_event_rejected(event_id).await
|
||||
&& !self.db.is_event_soft_failed(event_id).await
|
||||
}
|
||||
|
||||
pub fn clear_pdu_markers(&self, event_id: &EventId) { self.db.clear_pdu_markers(event_id); }
|
||||
}
|
||||
|
||||
@@ -53,7 +53,15 @@ pub async fn append_incoming_pdu<'a, Leaves>(
|
||||
.await?;
|
||||
|
||||
if soft_fail {
|
||||
// Nothing else to do with a soft-failed event.
|
||||
self.services
|
||||
.pdu_metadata
|
||||
.mark_as_referenced(room_id, pdu.prev_events.iter().map(AsRef::as_ref));
|
||||
|
||||
// self.services
|
||||
// .state
|
||||
// .set_forward_extremities(room_id, new_room_leaves, state_lock)
|
||||
// .await;
|
||||
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user