chris/ruwuma
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

state-res: Use unique over dedup, remove Vec -> BTreeSet conversions

Browse Source
This commit is contained in:
Devin Ragotzy 2021-05-14 18:53:35 -04:00 committed by Jonas Platte
parent 282ffd9c6d
commit ce6fd1e65e
No known key found for this signature in database
GPG Key ID: CC154DE0E30B7C67
8 changed files with 77 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
crates/ruma-state-res/benches/state_res_bench.rs
View File

@ -16,7 +16,7 @@ use std::{
use criterion::{criterion_group, criterion_main, Criterion};
use event::StateEvent;
use js_int::uint;
use maplit::btreemap;
use maplit::{btreemap, btreeset};
use ruma_common::MilliSecondsSinceUnixEpoch;
use ruma_events::{
pdu::{EventHash, Pdu, RoomV3Pdu},
@ -35,11 +35,11 @@ static SERVER_TIMESTAMP: AtomicU64 = AtomicU64::new(0);
fn lexico_topo_sort(c: &mut Criterion) {
c.bench_function("lexicographical topological sort", |b| {
let graph = btreemap! {
event_id("l") => vec![event_id("o")],
event_id("m") => vec![event_id("n"), event_id("o")],
event_id("n") => vec![event_id("o")],
event_id("o") => vec![], // "o" has zero outgoing edges but 4 incoming edges
event_id("p") => vec![event_id("o")],
event_id("l") => btreeset![event_id("o")],
event_id("m") => btreeset![event_id("n"), event_id("o")],
event_id("n") => btreeset![event_id("o")],
event_id("o") => btreeset![], // "o" has zero outgoing edges but 4 incoming edges
event_id("p") => btreeset![event_id("o")],
};
b.iter(|| {
let _ = StateResolution::lexicographical_topological_sort(&graph, |id| {
@ -61,7 +61,7 @@ fn resolution_shallow_auth_chain(c: &mut Criterion) {
let state_sets = vec![state_at_bob.clone(), state_at_charlie.clone()];
let _ = match StateResolution::resolve::<StateEvent>(
&room_id(),
&RoomVersionId::Version2,
&RoomVersionId::Version6,
&state_sets,
state_sets
.iter()
@ -82,10 +82,9 @@ fn resolution_shallow_auth_chain(c: &mut Criterion) {
fn resolve_deeper_event_set(c: &mut Criterion) {
c.bench_function("resolve state of 10 events 3 conflicting", |b| {
let init = INITIAL_EVENTS();
let mut inner = INITIAL_EVENTS();
let ban = BAN_STATE_SET();
let mut inner = init;
inner.extend(ban);
let store = TestStore(inner.clone());
@ -119,7 +118,7 @@ fn resolve_deeper_event_set(c: &mut Criterion) {
let state_sets = vec![state_set_a.clone(), state_set_b.clone()];
let _ = match StateResolution::resolve::<StateEvent>(
&room_id(),
&RoomVersionId::Version2,
&RoomVersionId::Version6,
&state_sets,
state_sets
.iter()
@ -330,25 +329,11 @@ fn room_id() -> RoomId {
}
fn member_content_ban() -> JsonValue {
serde_json::to_value(MemberEventContent {
membership: MembershipState::Ban,
displayname: None,
avatar_url: None,
is_direct: None,
third_party_invite: None,
})
.unwrap()
serde_json::to_value(MemberEventContent::new(MembershipState::Ban)).unwrap()
}
fn member_content_join() -> JsonValue {
serde_json::to_value(MemberEventContent {
membership: MembershipState::Join,
displayname: None,
avatar_url: None,
is_direct: None,
third_party_invite: None,
})
.unwrap()
serde_json::to_value(MemberEventContent::new(MembershipState::Join)).unwrap()
}
pub fn to_pdu_event<S>(
@ -363,6 +348,8 @@ pub fn to_pdu_event<S>(
where
S: AsRef<str>,
{
// We don't care if the addition happens in order just that it is atomic
// (each event has its own value)
let ts = SERVER_TIMESTAMP.fetch_add(1, SeqCst);
let id = if id.contains('$') { id.to_string() } else { format!("${}:foo", id) };
let auth_events = auth_events.iter().map(AsRef::as_ref).map(event_id).collect::<Vec<_>>();

5
crates/ruma-state-res/src/event_auth.rs
View File

@ -304,7 +304,7 @@ pub fn valid_membership_change<E: Event>(
auth_events: &StateMap<Arc<E>>,
) -> Result<bool> {
let target_membership = serde_json::from_value::<MembershipState>(
content.get("membership").expect("we should test before that this field exists").clone(),
content.get("membership").expect("we test before that this field exists").clone(),
)?;
let third_party_invite = content
@ -493,7 +493,7 @@ pub fn can_send_event<E: Event>(event: &Arc<E>, auth_events: &StateMap<Arc<E>>)
let event_type_power_level = get_send_level(&event.kind(), event.state_key(), ple);
let user_level = get_user_power_level(event.sender(), auth_events);
debug!("{} ev_type {} usr {}", event.event_id().as_str(), event_type_power_level, user_level);
debug!("{} ev_type {} usr {}", event.event_id(), event_type_power_level, user_level);
if user_level < event_type_power_level {
return false;
@ -756,7 +756,6 @@ pub fn get_send_level<E: Event>(
state_key: Option<String>,
power_lvl: Option<&Arc<E>>,
) -> i64 {
debug!("{:?} {:?}", e_type, state_key);
power_lvl
.and_then(|ple| {
serde_json::from_value::<PowerLevelsEventContent>(ple.content())

89
crates/ruma-state-res/src/lib.rs
View File

@ -77,19 +77,20 @@ impl StateResolution {
// Add the auth_diff to conflicting now we have a full set of conflicting events
auth_diff.extend(conflicting.values().cloned().flatten());
let mut all_conflicted =
auth_diff.into_iter().collect::<BTreeSet<_>>().into_iter().collect::<Vec<_>>();
// `all_conflicted` contains unique items
// synapse says `full_set = {eid for eid in full_conflicted_set if eid in event_map}`
//
// Don't honor events we cannot "verify"
// TODO: BTreeSet::retain() when stable 1.53
let all_conflicted =
auth_diff.into_iter().filter(|id| event_map.contains_key(id)).collect::<BTreeSet<_>>();
info!("full conflicted set is {} events", all_conflicted.len());
// We used to check that all events are events from the correct room
// this is now a check the caller of `resolve` must make.
// Synapse says `full_set = {eid for eid in full_conflicted_set if eid in event_map}`
//
// Don't honor events we cannot "verify"
all_conflicted.retain(|id| event_map.contains_key(id));
// Get only the control events with a state_key: "" or ban/kick event (sender != state_key)
let control_events = all_conflicted
.iter()
@ -98,7 +99,7 @@ impl StateResolution {
.collect::<Vec<_>>();
// Sort the control events based on power_level/clock/event_id and outgoing/incoming edges
let mut sorted_control_levels = StateResolution::reverse_topological_power_sort(
let sorted_control_levels = StateResolution::reverse_topological_power_sort(
room_id,
&control_events,
event_map,
@ -117,15 +118,11 @@ impl StateResolution {
event_map,
)?;
debug!(
"AUTHED {:?}",
resolved_control.iter().map(|(key, id)| (key, id.to_string())).collect::<Vec<_>>()
);
debug!("AUTHED {:?}", resolved_control.iter().collect::<Vec<_>>());
// At this point the control_events have been resolved we now have to
// sort the remaining events using the mainline of the resolved power level.
sorted_control_levels.dedup();
let deduped_power_ev = sorted_control_levels;
let deduped_power_ev = sorted_control_levels.into_iter().collect::<BTreeSet<_>>();
// This removes the control events that passed auth and more importantly those that failed
// auth
@ -135,7 +132,7 @@ impl StateResolution {
.cloned()
.collect::<Vec<_>>();
debug!("LEFT {:?}", events_to_resolve.iter().map(ToString::to_string).collect::<Vec<_>>());
debug!("LEFT {:?}", events_to_resolve.iter().collect::<Vec<_>>());
// This "epochs" power level event
let power_event = resolved_control.get(&(EventType::RoomPowerLevels, "".into()));
@ -145,10 +142,7 @@ impl StateResolution {
let sorted_left_events =
StateResolution::mainline_sort(room_id, &events_to_resolve, power_event, event_map);
debug!(
"SORTED LEFT {:?}",
sorted_left_events.iter().map(ToString::to_string).collect::<Vec<_>>()
);
debug!("SORTED LEFT {:?}", sorted_left_events.iter().collect::<Vec<_>>());
let mut resolved_state = StateResolution::iterative_auth_check(
room_id,
@ -180,16 +174,13 @@ impl StateResolution {
let mut unconflicted_state = StateMap::new();
let mut conflicted_state = StateMap::new();
for key in state_sets.iter().flat_map(|map| map.keys()).dedup() {
for key in state_sets.iter().flat_map(|map| map.keys()).unique() {
let mut event_ids =
state_sets.iter().map(|state_set| state_set.get(key)).dedup().collect::<Vec<_>>();
state_sets.iter().map(|state_set| state_set.get(key)).unique().collect::<Vec<_>>();
if event_ids.len() == 1 {
if let Some(Some(id)) = event_ids.pop() {
unconflicted_state.insert(key.clone(), id.clone());
} else {
panic!()
}
let id = event_ids.remove(0).expect("unconflicting `EventId` is not None");
unconflicted_state.insert(key.clone(), id.clone());
} else {
conflicted_state.insert(
key.clone(),
@ -205,9 +196,7 @@ impl StateResolution {
pub fn get_auth_chain_diff(
_room_id: &RoomId,
auth_event_ids: &[Vec<EventId>],
) -> Result<Vec<EventId>> {
use itertools::Itertools;
) -> Result<BTreeSet<EventId>> {
let mut chains = vec![];
for ids in auth_event_ids {
@ -221,9 +210,9 @@ impl StateResolution {
let rest = chains.iter().skip(1).flatten().cloned().collect();
let common = chain.intersection(&rest).collect::<Vec<_>>();
Ok(chains.into_iter().flatten().filter(|id| !common.contains(&id)).dedup().collect())
Ok(chains.into_iter().flatten().filter(|id| !common.contains(&id)).collect())
} else {
Ok(vec![])
Ok(btreeset![])
}
}
@ -238,7 +227,7 @@ impl StateResolution {
room_id: &RoomId,
events_to_sort: &[EventId],
event_map: &mut EventMap<Arc<E>>,
auth_diff: &[EventId],
auth_diff: &BTreeSet<EventId>,
) -> Vec<EventId> {
debug!("reverse topological sort of power events");
@ -257,7 +246,7 @@ impl StateResolution {
let mut event_to_pl = BTreeMap::new();
for event_id in graph.keys() {
let pl = StateResolution::get_power_level_for_sender(room_id, event_id, event_map);
info!("{} power level {}", event_id.to_string(), pl);
info!("{} power level {}", event_id, pl);
event_to_pl.insert(event_id.clone(), pl);
@ -283,7 +272,7 @@ impl StateResolution {
/// `key_fn` is used as a tie breaker. The tie breaker happens based on
/// power level, age, and event_id.
pub fn lexicographical_topological_sort<F>(
graph: &BTreeMap<EventId, Vec<EventId>>,
graph: &BTreeMap<EventId, BTreeSet<EventId>>,
key_fn: F,
) -> Vec<EventId>
where
@ -300,8 +289,7 @@ impl StateResolution {
// TODO make the BTreeSet conversion cleaner ??
// outdegree_map is an event referring to the events before it, the
// more outdegree's the more recent the event.
let mut outdegree_map: BTreeMap<EventId, BTreeSet<EventId>> =
graph.iter().map(|(k, v)| (k.clone(), v.iter().cloned().collect())).collect();
let mut outdegree_map = graph.clone();
// The number of events that depend on the given event (the eventId key)
let mut reverse_graph = BTreeMap::new();
@ -353,7 +341,7 @@ impl StateResolution {
event_id: &EventId,
event_map: &mut EventMap<Arc<E>>,
) -> i64 {
info!("fetch event ({}) senders power level", event_id.to_string());
info!("fetch event ({}) senders power level", event_id);
let event = StateResolution::get_or_load_event(room_id, event_id, event_map);
let mut pl = None;
@ -378,7 +366,7 @@ impl StateResolution {
{
if let Ok(ev) = event {
if let Some(user) = content.users.get(ev.sender()) {
debug!("found {} at power_level {}", ev.sender().as_str(), user);
debug!("found {} at power_level {}", ev.sender(), user);
return (*user).into();
}
}
@ -407,10 +395,7 @@ impl StateResolution {
) -> Result<StateMap<EventId>> {
info!("starting iterative auth check");
debug!(
"performing auth checks on {:?}",
events_to_check.iter().map(ToString::to_string).collect::<Vec<_>>()
);
debug!("performing auth checks on {:?}", events_to_check.iter().collect::<Vec<_>>());
let mut resolved_state = unconflicted_state.clone();
@ -454,7 +439,7 @@ impl StateResolution {
}
}
debug!("event to check {:?}", event.event_id().as_str());
debug!("event to check {:?}", event.event_id());
let most_recent_prev_event = event
.prev_events()
@ -483,7 +468,7 @@ impl StateResolution {
resolved_state.insert((event.kind(), state_key), event_id.clone());
} else {
// synapse passes here on AuthError. We do not add this event to resolved_state.
warn!("event {} failed the authentication check", event_id.to_string());
warn!("event {} failed the authentication check", event_id);
}
// TODO: if these functions are ever made async here
@ -582,7 +567,7 @@ impl StateResolution {
event_map: &mut EventMap<Arc<E>>,
) -> Result<usize> {
while let Some(sort_ev) = event {
debug!("mainline event_id {}", sort_ev.event_id().to_string());
debug!("mainline event_id {}", sort_ev.event_id());
let id = &sort_ev.event_id();
if let Some(depth) = mainline_map.get(id) {
return Ok(*depth);
@ -604,20 +589,20 @@ impl StateResolution {
fn add_event_and_auth_chain_to_graph<E: Event>(
room_id: &RoomId,
graph: &mut BTreeMap<EventId, Vec<EventId>>,
graph: &mut BTreeMap<EventId, BTreeSet<EventId>>,
event_id: &EventId,
event_map: &mut EventMap<Arc<E>>,
auth_diff: &[EventId],
auth_diff: &BTreeSet<EventId>,
) {
let mut state = vec![event_id.clone()];
while !state.is_empty() {
// We just checked if it was empty so unwrap is fine
let eid = state.pop().unwrap();
graph.entry(eid.clone()).or_insert_with(Vec::new);
graph.entry(eid.clone()).or_insert(btreeset![]);
// Prefer the store to event as the store filters dedups the events
// Otherwise it seems we can loop forever
for aid in
&StateResolution::get_or_load_event(room_id, &eid, event_map).unwrap().auth_events()
for aid in &StateResolution::get_or_load_event(room_id, &eid, event_map)
.map(|ev| ev.auth_events())
.unwrap_or_default()
{
if auth_diff.contains(aid) {
if !graph.contains_key(aid) {
@ -625,7 +610,7 @@ impl StateResolution {
}
// We just inserted this at the start of the while loop
graph.get_mut(&eid).unwrap().push(aid.clone());
graph.get_mut(&eid).unwrap().insert(aid.clone());
}
}
}

3
crates/ruma-state-res/tests/event_auth.rs
View File

@ -2,9 +2,6 @@ use std::sync::Arc;
use ruma_events::EventType;
use ruma_state_res::{event_auth::valid_membership_change, StateMap};
// use ruma_state_res::event_auth:::{
// auth_check, auth_types_for_event, can_federate, check_power_levels, check_redaction,
// };
mod utils;
use utils::{alice, charlie, event_id, member_content_ban, to_pdu_event, INITIAL_EVENTS};

6
crates/ruma-state-res/tests/event_sorting.rs
View File

@ -1,4 +1,4 @@
use std::collections::BTreeMap;
use std::collections::{BTreeMap, BTreeSet};
use rand::seq::SliceRandom;
use ruma_events::EventType;
@ -15,7 +15,7 @@ fn test_event_sort() {
.map(|ev| ((ev.kind(), ev.state_key()), ev.clone()))
.collect::<StateMap<_>>();
let auth_chain = &[] as &[_];
let auth_chain = BTreeSet::new();
let power_events = event_map
.values()
@ -30,7 +30,7 @@ fn test_event_sort() {
&room_id(),
&power_events,
&mut events,
auth_chain,
&auth_chain,
);
// This is a TODO in conduit

2
crates/ruma-state-res/tests/res_with_auth_ids.rs
View File

@ -67,7 +67,7 @@ fn ban_with_auth_chains2() {
let state_sets = vec![state_set_a, state_set_b];
let resolved = match StateResolution::resolve::<StateEvent>(
&room_id(),
&RoomVersionId::Version2,
&RoomVersionId::Version6,
&state_sets,
state_sets
.iter()

23
crates/ruma-state-res/tests/state_res.rs
View File

@ -1,7 +1,7 @@
use std::sync::Arc;
use js_int::uint;
use maplit::btreemap;
use maplit::{btreemap, btreeset};
use ruma_common::MilliSecondsSinceUnixEpoch;
use ruma_events::{room::join_rules::JoinRule, EventType};
use ruma_identifiers::{EventId, RoomVersionId};
@ -240,6 +240,13 @@ fn topic_setting() {
#[test]
fn test_event_map_none() {
// This is the only test that does not use `do_check` so we
// have to start the logger if this is what we're running.
//
// to activate logging use `RUST_LOG=debug cargo t one_test_only`
let _ = LOGGER
.call_once(|| tracer::fmt().with_env_filter(tracer::EnvFilter::from_default_env()).init());
let mut store = TestStore::<StateEvent>(btreemap! {});
// build up the DAG
@ -271,11 +278,11 @@ fn test_event_map_none() {
#[test]
fn test_lexicographical_sort() {
let graph = btreemap! {
event_id("l") => vec![event_id("o")],
event_id("m") => vec![event_id("n"), event_id("o")],
event_id("n") => vec![event_id("o")],
event_id("o") => vec![], // "o" has zero outgoing edges but 4 incoming edges
event_id("p") => vec![event_id("o")],
event_id("l") => btreeset![event_id("o")],
event_id("m") => btreeset![event_id("n"), event_id("o")],
event_id("n") => btreeset![event_id("o")],
event_id("o") => btreeset![], // "o" has zero outgoing edges but 4 incoming edges
event_id("p") => btreeset![event_id("o")],
};
let res = StateResolution::lexicographical_topological_sort(&graph, |id| {
@ -296,10 +303,6 @@ fn test_lexicographical_sort() {
//
impl TestStore<StateEvent> {
pub fn set_up(&mut self) -> (StateMap<EventId>, StateMap<EventId>, StateMap<EventId>) {
// to activate logging use `RUST_LOG=debug cargo t one_test_only`
let _ = LOGGER.call_once(|| {
tracer::fmt().with_env_filter(tracer::EnvFilter::from_default_env()).init()
});
let create_event = to_pdu_event::<EventId>(
"CREATE",
alice(),

17
crates/ruma-state-res/tests/utils.rs
View File

@ -10,7 +10,7 @@ use std::{
};
use js_int::uint;
use maplit::btreemap;
use maplit::{btreemap, btreeset};
use ruma_common::MilliSecondsSinceUnixEpoch;
use ruma_events::{
pdu::{EventHash, Pdu, RoomV3Pdu},
@ -37,9 +37,8 @@ pub fn do_check(
edges: Vec<Vec<EventId>>,
expected_state_ids: Vec<EventId>,
) {
// to activate logging use `RUST_LOG=debug cargo t`
let _ = LOGGER
.call_once(|| tracer::fmt().with_env_filter(tracer::EnvFilter::from_default_env()).init());
// To activate logging use `RUST_LOG=debug cargo t`
// The logger is initialized in the `INITIAL_EVENTS` function.
let init_events = INITIAL_EVENTS();
@ -55,20 +54,20 @@ pub fn do_check(
// Create the DB of events that led up to this point
// TODO maybe clean up some of these clones it is just tests but...
for ev in init_events.values().chain(events) {
graph.insert(ev.event_id().clone(), vec![]);
graph.insert(ev.event_id().clone(), btreeset![]);
fake_event_map.insert(ev.event_id().clone(), ev.clone());
}
for pair in INITIAL_EDGES().windows(2) {
if let [a, b] = &pair {
graph.entry(a.clone()).or_insert(vec![]).push(b.clone());
graph.entry(a.clone()).or_insert(btreeset![]).insert(b.clone());
}
}
for edge_list in edges {
for pair in edge_list.windows(2) {
if let [a, b] = &pair {
graph.entry(a.clone()).or_insert(vec![]).push(b.clone());
graph.entry(a.clone()).or_insert(btreeset![]).insert(b.clone());
}
}
}
@ -91,7 +90,7 @@ pub fn do_check(
let state_before: StateMap<EventId> = if prev_events.is_empty() {
BTreeMap::new()
} else if prev_events.len() == 1 {
state_at_event.get(&prev_events[0]).unwrap().clone()
state_at_event.get(prev_events.iter().next().unwrap()).unwrap().clone()
} else {
let state_sets = prev_events
.iter()
@ -161,7 +160,7 @@ pub fn do_check(
Some(&e.state_key()),
e.content(),
&auth_events,
prev_events,
&prev_events.iter().cloned().collect::<Vec<_>>(),
);
// We have to update our store, an actual user of this lib would