crappyrules
/
conduit
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Use eventId when saving outliers

merge-requests/14/head
Devin Ragotzy 5 years ago
parent
e09be2f7ee
commit
64374b4679
4 changed files with 85 additions and 66 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      src/client_server/sync.rs
  2. 2
      src/database.rs
  3. 76
      src/database/rooms.rs
  4. 71
      src/server_server.rs

2
src/client_server/sync.rs
Unescape View File

@ -701,7 +701,7 @@ pub async fn sync_events_route(
let delay = tokio::time::sleep(duration); let delay = tokio::time::sleep(duration);
tokio::pin!(delay); tokio::pin!(delay);
tokio::select! { tokio::select! {
_ = &mut delay, if delay.is_elapsed() => {} _ = &mut delay => {}
_ = watcher => {} _ = watcher => {}
} }
} }

2
src/database.rs
Unescape View File

@ -159,7 +159,7 @@ impl Database {
stateid_pduid: db.open_tree("stateid_pduid")?, stateid_pduid: db.open_tree("stateid_pduid")?,
pduid_statehash: db.open_tree("pduid_statehash")?, pduid_statehash: db.open_tree("pduid_statehash")?,
roomid_statehash: db.open_tree("roomid_statehash")?, roomid_statehash: db.open_tree("roomid_statehash")?,
pduid_outlierpdu: db.open_tree("pduid_outlierpdu")?, eventid_outlierpdu: db.open_tree("eventid_outlierpdu")?,
}, },
account_data: account_data::AccountData { account_data: account_data::AccountData {
roomuserdataid_accountdata: db.open_tree("roomuserdataid_accountdata")?, roomuserdataid_accountdata: db.open_tree("roomuserdataid_accountdata")?,

76
src/database/rooms.rs
Unescape View File

@ -27,10 +27,9 @@ use std::{
convert::{TryFrom, TryInto}, convert::{TryFrom, TryInto},
mem, mem,
sync::Arc, sync::Arc,
time::Duration,
}; };
use super::{admin::AdminCommand, sending::Sending}; use super::admin::AdminCommand;
/// The unique identifier of each state group. /// The unique identifier of each state group.
/// ///
@ -67,13 +66,16 @@ pub struct Rooms {
pub(super) statekey_short: sled::Tree, // StateKey = EventType + StateKey, Short = Count pub(super) statekey_short: sled::Tree, // StateKey = EventType + StateKey, Short = Count
pub(super) stateid_pduid: sled::Tree, // StateId = StateHash + Short, PduId = Count (without roomid) pub(super) stateid_pduid: sled::Tree, // StateId = StateHash + Short, PduId = Count (without roomid)
/// Any pdu that has passed the steps up to auth with auth_events. /// RoomId + EventId -> outlier PDU.
pub(super) pduid_outlierpdu: sled::Tree, /// Any pdu that has passed the steps 1-8 in the incoming event /federation/send/txn.
pub(super) eventid_outlierpdu: sled::Tree,
} }
impl Rooms { impl Rooms {
/// Builds a StateMap by iterating over all keys that start /// Builds a StateMap by iterating over all keys that start
/// with state_hash, this gives the full state for the given state_hash. /// with state_hash, this gives the full state for the given state_hash.
///
/// TODO: Should this check for outliers, it does now.
pub fn state_full( pub fn state_full(
&self, &self,
room_id: &RoomId, room_id: &RoomId,
@ -90,7 +92,7 @@ impl Rooms {
Some(b) => serde_json::from_slice::<PduEvent>(&b) Some(b) => serde_json::from_slice::<PduEvent>(&b)
.map_err(|_| Error::bad_database("Invalid PDU in db.")), .map_err(|_| Error::bad_database("Invalid PDU in db.")),
None => self None => self
.pduid_outlierpdu .eventid_outlierpdu
.get(pduid)? .get(pduid)?
.map(|b| { .map(|b| {
serde_json::from_slice::<PduEvent>(&b) serde_json::from_slice::<PduEvent>(&b)
@ -118,6 +120,8 @@ impl Rooms {
} }
/// Returns a single PDU from `room_id` with key (`event_type`, `state_key`). /// Returns a single PDU from `room_id` with key (`event_type`, `state_key`).
///
/// TODO: Should this check for outliers, it does now.
pub fn state_get( pub fn state_get(
&self, &self,
room_id: &RoomId, room_id: &RoomId,
@ -149,7 +153,7 @@ impl Rooms {
Some(b) => serde_json::from_slice::<PduEvent>(&b) Some(b) => serde_json::from_slice::<PduEvent>(&b)
.map_err(|_| Error::bad_database("Invalid PDU in db."))?, .map_err(|_| Error::bad_database("Invalid PDU in db."))?,
None => self None => self
.pduid_outlierpdu .eventid_outlierpdu
.get(pdu_id)? .get(pdu_id)?
.map(|b| { .map(|b| {
serde_json::from_slice::<PduEvent>(&b) serde_json::from_slice::<PduEvent>(&b)
@ -260,6 +264,8 @@ impl Rooms {
} }
}; };
// Because of outliers this could also be an eventID but that
// is handled by `state_full`
let pdu_id_short = pdu_id let pdu_id_short = pdu_id
.splitn(2, |&b| b == 0xff) .splitn(2, |&b| b == 0xff)
.nth(1) .nth(1)
@ -325,9 +331,12 @@ impl Rooms {
Ok(Some( Ok(Some(
serde_json::from_slice(&match self.pduid_pdu.get(&pdu_id)? { serde_json::from_slice(&match self.pduid_pdu.get(&pdu_id)? {
Some(b) => b, Some(b) => b,
None => self.pduid_outlierpdu.get(pdu_id)?.ok_or_else(|| { None => self
Error::bad_database("Event is not in pdu tree or outliers.") .eventid_outlierpdu
})?, .get(event_id.as_bytes())?
.ok_or_else(|| {
Error::bad_database("Event is not in pdu tree or outliers.")
})?,
}) })
.map_err(|_| Error::bad_database("Invalid PDU in db."))?, .map_err(|_| Error::bad_database("Invalid PDU in db."))?,
)) ))
@ -342,6 +351,8 @@ impl Rooms {
} }
/// Returns the pdu. /// Returns the pdu.
///
/// Checks the `eventid_outlierpdu` Tree if not found in the timeline.
pub fn get_pdu(&self, event_id: &EventId) -> Result<Option<PduEvent>> { pub fn get_pdu(&self, event_id: &EventId) -> Result<Option<PduEvent>> {
self.eventid_pduid self.eventid_pduid
.get(event_id.as_bytes())? .get(event_id.as_bytes())?
@ -349,9 +360,12 @@ impl Rooms {
Ok(Some( Ok(Some(
serde_json::from_slice(&match self.pduid_pdu.get(&pdu_id)? { serde_json::from_slice(&match self.pduid_pdu.get(&pdu_id)? {
Some(b) => b, Some(b) => b,
None => self.pduid_outlierpdu.get(pdu_id)?.ok_or_else(|| { None => self
Error::bad_database("Event is not in pdu tree or outliers.") .eventid_outlierpdu
})?, .get(event_id.as_bytes())?
.ok_or_else(|| {
Error::bad_database("Event is not in pdu tree or outliers.")
})?,
}) })
.map_err(|_| Error::bad_database("Invalid PDU in db."))?, .map_err(|_| Error::bad_database("Invalid PDU in db."))?,
)) ))
@ -443,27 +457,34 @@ impl Rooms {
/// Returns the pdu from the outlier tree. /// Returns the pdu from the outlier tree.
pub fn get_pdu_outlier(&self, event_id: &EventId) -> Result<Option<PduEvent>> { pub fn get_pdu_outlier(&self, event_id: &EventId) -> Result<Option<PduEvent>> {
if let Some(id) = self.eventid_pduid.get(event_id.as_bytes())? { self.eventid_outlierpdu
self.pduid_outlierpdu.get(id)?.map_or(Ok(None), |pdu| { .get(event_id.as_bytes())?
.map_or(Ok(None), |pdu| {
serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db.")) serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db."))
}) })
} else {
Ok(None)
}
} }
/// Returns true if the event_id was previously inserted. /// Returns true if the event_id was previously inserted.
pub fn append_pdu_outlier(&self, pdu_id: &[u8], pdu: &PduEvent) -> Result<bool> { pub fn append_pdu_outlier(&self, pdu: &PduEvent) -> Result<bool> {
log::info!("Number of outlier pdu's {}", self.pduid_outlierpdu.len()); log::info!(
"Number of outlier pdu's {:#?}",
self.eventid_outlierpdu
.iter()
.map(|pair| {
let (_k, v) = pair.unwrap();
serde_json::from_slice::<PduBuilder>(&v).unwrap()
})
.collect::<Vec<_>>()
);
// we need to be able to find it by event_id let mut key = pdu.room_id().as_bytes().to_vec();
self.eventid_pduid key.push(0xff);
.insert(pdu.event_id.as_bytes(), &*pdu_id)?; key.extend_from_slice(pdu.event_id().as_bytes());
let res = self let res = self
.pduid_outlierpdu .eventid_outlierpdu
.insert( .insert(
pdu_id, &key,
&*serde_json::to_string(&pdu).expect("PduEvent is always a valid String"), &*serde_json::to_string(&pdu).expect("PduEvent is always a valid String"),
) )
.map(|op| op.is_some())?; .map(|op| op.is_some())?;
@ -511,9 +532,10 @@ impl Rooms {
} }
// We no longer keep this pdu as an outlier // We no longer keep this pdu as an outlier
if let Some(id) = self.eventid_pduid.remove(pdu.event_id().as_bytes())? { let mut key = pdu.room_id().as_bytes().to_vec();
self.pduid_outlierpdu.remove(id)?; key.push(0xff);
} key.extend_from_slice(pdu.event_id().as_bytes());
self.eventid_outlierpdu.remove(key)?;
self.replace_pdu_leaves(&pdu.room_id, leaves)?; self.replace_pdu_leaves(&pdu.room_id, leaves)?;

71
src/server_server.rs
Unescape View File

@ -18,7 +18,6 @@ use ruma::{
OutgoingRequest, OutgoingRequest,
}, },
directory::{IncomingFilter, IncomingRoomNetwork}, directory::{IncomingFilter, IncomingRoomNetwork},
events::EventType,
serde::to_canonical_value, serde::to_canonical_value,
signatures::{CanonicalJsonObject, CanonicalJsonValue, PublicKeyMap}, signatures::{CanonicalJsonObject, CanonicalJsonValue, PublicKeyMap},
EventId, RoomId, RoomVersionId, ServerName, ServerSigningKeyId, UserId, EventId, RoomId, RoomVersionId, ServerName, ServerSigningKeyId, UserId,
@ -638,12 +637,8 @@ pub async fn send_transaction_message_route<'a>(
None None
}; };
let count = db.globals.next_count()?;
let mut pdu_id = pdu.room_id.as_bytes().to_vec();
pdu_id.push(0xff);
pdu_id.extend_from_slice(&count.to_be_bytes());
// 6. persist the event as an outlier. // 6. persist the event as an outlier.
db.rooms.append_pdu_outlier(&pdu_id, &pdu)?; db.rooms.append_pdu_outlier(&pdu)?;
// Step 9. fetch missing state by calling /state_ids at backwards extremities doing all // Step 9. fetch missing state by calling /state_ids at backwards extremities doing all
// the checks in this list starting at 1. These are not timeline events. // the checks in this list starting at 1. These are not timeline events.
@ -1079,37 +1074,28 @@ pub(crate) async fn fetch_events(
) -> Result<Vec<Arc<PduEvent>>> { ) -> Result<Vec<Arc<PduEvent>>> {
let mut pdus = vec![]; let mut pdus = vec![];
for id in events { for id in events {
// `get_pdu` checks the outliers tree for us
let pdu = match db.rooms.get_pdu(&id)? { let pdu = match db.rooms.get_pdu(&id)? {
Some(pdu) => Arc::new(pdu), Some(pdu) => Arc::new(pdu),
None => match db.rooms.get_pdu_outlier(&id)? { None => match db
Some(pdu) => Arc::new(pdu), .sending
None => match db .send_federation_request(
.sending &db.globals,
.send_federation_request( origin,
&db.globals, get_event::v1::Request { event_id: &id },
origin, )
get_event::v1::Request { event_id: &id }, .await
) {
.await Ok(res) => {
{ let (event_id, value) = crate::pdu::gen_event_id_canonical_json(&res.pdu);
Ok(res) => { let (pdu, _) = validate_event(db, value, event_id, key_map, origin, auth_cache)
let (event_id, value) = crate::pdu::gen_event_id_canonical_json(&res.pdu); .await
let (pdu, _) = .map_err(|_| Error::Conflict("Authentication of event failed"))?;
validate_event(db, value, event_id, key_map, origin, auth_cache)
.await db.rooms.append_pdu_outlier(&pdu)?;
.map_err(|_| Error::Conflict("Authentication of event failed"))?; pdu
}
// create the pduid for this event but stick it in the outliers DB Err(_) => return Err(Error::BadServerResponse("Failed to fetch event")),
let count = db.globals.next_count()?;
let mut pdu_id = pdu.room_id.as_bytes().to_vec();
pdu_id.push(0xff);
pdu_id.extend_from_slice(&count.to_be_bytes());
db.rooms.append_pdu_outlier(&pdu_id, &pdu)?;
pdu
}
Err(_) => return Err(Error::BadServerResponse("Failed to fetch event")),
},
}, },
}; };
pdus.push(pdu); pdus.push(pdu);
@ -1193,7 +1179,7 @@ pub(crate) async fn calculate_forward_extremities(
/// This should always be called after the incoming event has been appended to the DB. /// This should always be called after the incoming event has been appended to the DB.
/// ///
/// This guarentees that the incoming event will be in the state sets (at least our servers /// This guarantees that the incoming event will be in the state sets (at least our servers
/// and the sending server). /// and the sending server).
pub(crate) async fn build_forward_extremity_snapshots( pub(crate) async fn build_forward_extremity_snapshots(
db: &Database, db: &Database,
@ -1303,7 +1289,18 @@ pub(crate) fn update_resolved_state(
); );
} }
None => { None => {
error!("We didn't append an event as an outlier\n{:?}", pdu); let mut pduid = pdu.room_id().as_bytes().to_vec();
pduid.push(0xff);
pduid.extend_from_slice(pdu.event_id().as_bytes());
new_state.insert(
(
ev_type,
state_k.ok_or_else(|| {
Error::Conflict("State contained non state event")
})?,
),
pduid,
);
} }
} }
} }

Write Preview
Loading…
Cancel
Save