magmaus3/continuwuity
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge remote-tracking branch 'upstream/master' into correct-sendtxn

Browse source
This commit is contained in:
Devin Ragotzy 2021-03-04 08:39:16 -05:00
commit d0df8b495c
49 changed files with 747 additions and 281 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/database/account_data.rs
View file

@ -74,6 +74,7 @@ impl AccountData {
}
/// Returns all changes to the account data that happened after `since`.
#[tracing::instrument(skip(self))]
pub fn changes_since(
&self,
room_id: Option<&RoomId>,

3
src/database/admin.rs
View file

@ -7,7 +7,6 @@ use ruma::{
events::{room::message, EventType},
UserId,
};
use tokio::select;
pub enum AdminCommand {
RegisterAppservice(serde_yaml::Value),
@ -67,7 +66,7 @@ impl Admin {
};
loop {
select! {
tokio::select! {
Some(event) = receiver.next() => {
match event {
AdminCommand::RegisterAppservice(yaml) => {

4
src/database/globals.rs
View file

@ -12,11 +12,11 @@ use trust_dns_resolver::TokioAsyncResolver;
pub const COUNTER: &str = "c";
pub type DestinationCache = Arc<RwLock<HashMap<Box<ServerName>, (String, Option<String>)>>>;
type WellKnownMap = HashMap<Box<ServerName>, (String, Option<String>)>;
#[derive(Clone)]
pub struct Globals {
pub actual_destination_cache: DestinationCache, // actual_destination, host
pub actual_destination_cache: Arc<RwLock<WellKnownMap>>, // actual_destination, host
pub(super) globals: sled::Tree,
config: Config,
keypair: Arc<ruma::signatures::Ed25519KeyPair>,

20
src/database/rooms.rs
View file

@ -83,8 +83,7 @@ pub struct Rooms {
impl Rooms {
/// Builds a StateMap by iterating over all keys that start
/// with state_hash, this gives the full state for the given state_hash.
///
/// TODO: Should this check for outliers, it does now.
#[tracing::instrument(skip(self))]
pub fn state_full(
&self,
room_id: &RoomId,
@ -129,6 +128,7 @@ impl Rooms {
}
/// Returns a single PDU from `room_id` with key (`event_type`, `state_key`).
#[tracing::instrument(skip(self))]
pub fn state_get(
&self,
room_id: &RoomId,
@ -178,11 +178,13 @@ impl Rooms {
}
/// Returns the state hash for this pdu.
#[tracing::instrument(skip(self))]
pub fn pdu_state_hash(&self, pdu_id: &[u8]) -> Result<Option<StateHashId>> {
Ok(self.pduid_statehash.get(pdu_id)?)
}
/// Returns the last state hash key added to the db for the given room.
#[tracing::instrument(skip(self))]
pub fn current_state_hash(&self, room_id: &RoomId) -> Result<Option<StateHashId>> {
Ok(self.roomid_statehash.get(room_id.as_bytes())?)
}
@ -290,6 +292,7 @@ impl Rooms {
}
/// Returns the full room state.
#[tracing::instrument(skip(self))]
pub fn room_state_full(
&self,
room_id: &RoomId,
@ -302,6 +305,7 @@ impl Rooms {
}
/// Returns a single PDU from `room_id` with key (`event_type`, `state_key`).
#[tracing::instrument(skip(self))]
pub fn room_state_get(
&self,
room_id: &RoomId,
@ -316,6 +320,7 @@ impl Rooms {
}
/// Returns the `count` of this pdu's id.
#[tracing::instrument(skip(self))]
pub fn pdu_count(&self, pdu_id: &[u8]) -> Result<u64> {
Ok(
utils::u64_from_bytes(&pdu_id[pdu_id.len() - mem::size_of::<u64>()..pdu_id.len()])
@ -515,6 +520,7 @@ impl Rooms {
///
/// By this point the incoming event should be fully authenticated, no auth happens
/// in `append_pdu`.
#[allow(clippy::too_many_arguments)]
pub fn append_pdu(
&self,
pdu: &PduEvent,
@ -1113,6 +1119,7 @@ impl Rooms {
let user_is_joined =
|bridge_user_id| self.is_joined(&bridge_user_id, room_id).unwrap_or(false);
let matching_users = |users: &Regex| {
users.is_match(pdu.sender.as_str())
|| pdu.kind == EventType::RoomMember
@ -1141,6 +1148,7 @@ impl Rooms {
}
/// Returns an iterator over all PDUs in a room.
#[tracing::instrument(skip(self))]
pub fn all_pdus(
&self,
user_id: &UserId,
@ -1151,6 +1159,7 @@ impl Rooms {
/// Returns a double-ended iterator over all events in a room that happened after the event with id `since`
/// in chronological order.
#[tracing::instrument(skip(self))]
pub fn pdus_since(
&self,
user_id: &UserId,
@ -1217,6 +1226,7 @@ impl Rooms {
/// Returns an iterator over all events and their token in a room that happened after the event
/// with id `from` in chronological order.
#[tracing::instrument(skip(self))]
pub fn pdus_after(
&self,
user_id: &UserId,
@ -1566,6 +1576,7 @@ impl Rooms {
))
}
#[tracing::instrument(skip(self))]
pub fn get_shared_rooms<'a>(
&'a self,
users: Vec<UserId>,
@ -1627,6 +1638,7 @@ impl Rooms {
}
/// Returns an iterator over all joined members of a room.
#[tracing::instrument(skip(self))]
pub fn room_members(&self, room_id: &RoomId) -> impl Iterator<Item = Result<UserId>> {
let mut prefix = room_id.as_bytes().to_vec();
prefix.push(0xff);
@ -1675,6 +1687,7 @@ impl Rooms {
}
/// Returns an iterator over all invited members of a room.
#[tracing::instrument(skip(self))]
pub fn room_members_invited(&self, room_id: &RoomId) -> impl Iterator<Item = Result<UserId>> {
let mut prefix = room_id.as_bytes().to_vec();
prefix.push(0xff);
@ -1699,6 +1712,7 @@ impl Rooms {
}
/// Returns an iterator over all rooms this user joined.
#[tracing::instrument(skip(self))]
pub fn rooms_joined(&self, user_id: &UserId) -> impl Iterator<Item = Result<RoomId>> {
self.userroomid_joined
.scan_prefix(user_id.as_bytes())
@ -1720,6 +1734,7 @@ impl Rooms {
}
/// Returns an iterator over all rooms a user was invited to.
#[tracing::instrument(skip(self))]
pub fn rooms_invited(&self, user_id: &UserId) -> impl Iterator<Item = Result<RoomId>> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
@ -1744,6 +1759,7 @@ impl Rooms {
}
/// Returns an iterator over all rooms a user left.
#[tracing::instrument(skip(self))]
pub fn rooms_left(&self, user_id: &UserId) -> impl Iterator<Item = Result<RoomId>> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);

5
src/database/rooms/edus.rs
View file

@ -70,6 +70,7 @@ impl RoomEdus {
}
/// Returns an iterator over the most recent read_receipts in a room that happened after the event with id `since`.
#[tracing::instrument(skip(self))]
pub fn readreceipts_since(
&self,
room_id: &RoomId,
@ -115,6 +116,7 @@ impl RoomEdus {
}
/// Returns the private read marker.
#[tracing::instrument(skip(self))]
pub fn private_read_get(&self, room_id: &RoomId, user_id: &UserId) -> Result<Option<u64>> {
let mut key = room_id.to_string().as_bytes().to_vec();
key.push(0xff);
@ -256,6 +258,7 @@ impl RoomEdus {
}
/// Returns the count of the last typing update in this room.
#[tracing::instrument(skip(self, globals))]
pub fn last_typing_update(
&self,
room_id: &RoomId,
@ -339,6 +342,7 @@ impl RoomEdus {
}
/// Resets the presence timeout, so the user will stay in their current presence state.
#[tracing::instrument(skip(self))]
pub fn ping_presence(&self, user_id: &UserId) -> Result<()> {
self.userid_lastpresenceupdate.insert(
&user_id.to_string().as_bytes(),
@ -429,6 +433,7 @@ impl RoomEdus {
}
/// Returns an iterator over the most recent presence updates that happened after the event with id `since`.
#[tracing::instrument(skip(self, globals, rooms))]
pub fn presence_since(
&self,
room_id: &RoomId,

96
src/database/sending.rs
View file

@ -8,7 +8,8 @@ use std::{
use crate::{appservice_server, server_server, utils, Error, PduEvent, Result};
use federation::transactions::send_transaction_message;
use log::info;
use log::{info, warn};
use ring::digest;
use rocket::futures::stream::{FuturesUnordered, StreamExt};
use ruma::{
api::{appservice, federation, OutgoingRequest},
@ -35,6 +36,7 @@ impl Sending {
) {
let servernamepduids = self.servernamepduids.clone();
let servercurrentpdus = self.servercurrentpdus.clone();
let maximum_requests = self.maximum_requests.clone();
let rooms = rooms.clone();
let globals = globals.clone();
let appservice = appservice.clone();
@ -43,23 +45,43 @@ impl Sending {
let mut futures = FuturesUnordered::new();
// Retry requests we could not finish yet
let mut current_transactions = HashMap::new();
let mut current_transactions = HashMap::<(Box<ServerName>, bool), Vec<IVec>>::new();
for (server, pdu, is_appservice) in servercurrentpdus
for (key, server, pdu, is_appservice) in servercurrentpdus
.iter()
.filter_map(|r| r.ok())
.filter_map(|(key, _)| Self::parse_servercurrentpdus(key).ok())
.filter(|(_, pdu, _)| !pdu.is_empty()) // Skip reservation key
.take(50)
// This should not contain more than 50 anyway
{
current_transactions
if pdu.is_empty() {
// Remove old reservation key
servercurrentpdus.remove(key).unwrap();
continue;
}
let entry = current_transactions
.entry((server, is_appservice))
.or_insert_with(Vec::new)
.push(pdu);
.or_insert_with(Vec::new);
if entry.len() > 30 {
warn!("Dropping some current pdus because too many were queued. This should not happen.");
servercurrentpdus.remove(key).unwrap();
continue;
}
entry.push(pdu);
}
for ((server, is_appservice), pdus) in current_transactions {
// Create new reservation
let mut prefix = if is_appservice {
b"+".to_vec()
} else {
Vec::new()
};
prefix.extend_from_slice(server.as_bytes());
prefix.push(0xff);
servercurrentpdus.insert(prefix, &[]).unwrap();
futures.push(Self::handle_event(
server,
is_appservice,
@ -67,6 +89,7 @@ impl Sending {
&globals,
&rooms,
&appservice,
&maximum_requests,
));
}
@ -105,7 +128,7 @@ impl Sending {
.map(|k| {
k.subslice(prefix.len(), k.len() - prefix.len())
})
.take(50)
.take(30)
.collect::<Vec<_>>();
if !new_pdus.is_empty() {
@ -116,7 +139,7 @@ impl Sending {
servernamepduids.remove(&current_key).unwrap();
}
futures.push(Self::handle_event(server, is_appservice, new_pdus, &globals, &rooms, &appservice));
futures.push(Self::handle_event(server, is_appservice, new_pdus, &globals, &rooms, &appservice, &maximum_requests));
} else {
servercurrentpdus.remove(&prefix).unwrap();
// servercurrentpdus with the prefix should be empty now
@ -202,7 +225,7 @@ impl Sending {
servercurrentpdus.insert(&key, &[]).unwrap();
servernamepduids.remove(&key).unwrap();
futures.push(Self::handle_event(server, is_appservice, vec![pdu_id.into()], &globals, &rooms, &appservice));
futures.push(Self::handle_event(server, is_appservice, vec![pdu_id.into()], &globals, &rooms, &appservice, &maximum_requests));
}
}
}
@ -211,6 +234,7 @@ impl Sending {
});
}
#[tracing::instrument(skip(self))]
pub fn send_pdu(&self, server: &ServerName, pdu_id: &[u8]) -> Result<()> {
let mut key = server.as_bytes().to_vec();
key.push(0xff);
@ -220,6 +244,7 @@ impl Sending {
Ok(())
}
#[tracing::instrument(skip(self))]
pub fn send_pdu_appservice(&self, appservice_id: &str, pdu_id: &[u8]) -> Result<()> {
let mut key = b"+".to_vec();
key.extend_from_slice(appservice_id.as_bytes());
@ -230,6 +255,15 @@ impl Sending {
Ok(())
}
#[tracing::instrument]
fn calculate_hash(keys: &[IVec]) -> Vec<u8> {
// We only hash the pdu's event ids, not the whole pdu
let bytes = keys.join(&0xff);
let hash = digest::digest(&digest::SHA256, &bytes);
hash.as_ref().to_owned()
}
#[tracing::instrument(skip(globals, rooms, appservice))]
async fn handle_event(
server: Box<ServerName>,
is_appservice: bool,
@ -237,6 +271,7 @@ impl Sending {
globals: &super::globals::Globals,
rooms: &super::rooms::Rooms,
appservice: &super::appservice::Appservice,
maximum_requests: &Semaphore,
) -> std::result::Result<(Box<ServerName>, bool), (Box<ServerName>, bool, Error)> {
if is_appservice {
let pdu_jsons = pdu_ids
@ -259,7 +294,9 @@ impl Sending {
})
.filter_map(|r| r.ok())
.collect::<Vec<_>>();
appservice_server::send_request(
let permit = maximum_requests.acquire().await;
let response = appservice_server::send_request(
&globals,
appservice
.get_registration(server.as_str())
@ -267,12 +304,19 @@ impl Sending {
.unwrap(), // TODO: handle error
appservice::event::push_events::v1::Request {
events: &pdu_jsons,
txn_id: &utils::random_string(16),
txn_id: &base64::encode_config(
Self::calculate_hash(&pdu_ids),
base64::URL_SAFE_NO_PAD,
),
},
)
.await
.map(|_response| (server.clone(), is_appservice))
.map_err(|e| (server, is_appservice, e))
.map_err(|e| (server, is_appservice, e));
drop(permit);
response
} else {
let pdu_jsons = pdu_ids
.iter()
@ -302,7 +346,8 @@ impl Sending {
.filter_map(|r| r.ok())
.collect::<Vec<_>>();
server_server::send_request(
let permit = maximum_requests.acquire().await;
let response = server_server::send_request(
&globals,
&*server,
send_transaction_message::v1::Request {
@ -310,17 +355,25 @@ impl Sending {
pdus: &pdu_jsons,
edus: &[],
origin_server_ts: SystemTime::now(),
transaction_id: &utils::random_string(16),
transaction_id: &base64::encode_config(
Self::calculate_hash(&pdu_ids),
base64::URL_SAFE_NO_PAD,
),
},
)
.await
.map(|_response| (server.clone(), is_appservice))
.map_err(|e| (server, is_appservice, e))
.map_err(|e| (server, is_appservice, e));
drop(permit);
response
}
}
fn parse_servercurrentpdus(key: IVec) -> Result<(Box<ServerName>, IVec, bool)> {
let mut parts = key.splitn(2, |&b| b == 0xff);
fn parse_servercurrentpdus(key: IVec) -> Result<(IVec, Box<ServerName>, IVec, bool)> {
let key2 = key.clone();
let mut parts = key2.splitn(2, |&b| b == 0xff);
let server = parts.next().expect("splitn always returns one element");
let pdu = parts
.next()
@ -338,6 +391,7 @@ impl Sending {
};
Ok::<_, Error>((
key,
Box::<ServerName>::try_from(server).map_err(|_| {
Error::bad_database("Invalid server string in server_currenttransaction")
})?,
@ -346,6 +400,7 @@ impl Sending {
))
}
#[tracing::instrument(skip(self, globals))]
pub async fn send_federation_request<T: OutgoingRequest>(
&self,
globals: &crate::database::globals::Globals,
@ -362,6 +417,7 @@ impl Sending {
response
}
#[tracing::instrument(skip(self, globals))]
pub async fn send_appservice_request<T: OutgoingRequest>(
&self,
globals: &crate::database::globals::Globals,

5
src/database/users.rs
View file

@ -311,6 +311,7 @@ impl Users {
Ok(())
}
#[tracing::instrument(skip(self))]
pub fn last_one_time_keys_update(&self, user_id: &UserId) -> Result<u64> {
self.userid_lastonetimekeyupdate
.get(&user_id.to_string().as_bytes())?
@ -364,6 +365,7 @@ impl Users {
.transpose()
}
#[tracing::instrument(skip(self))]
pub fn count_one_time_keys(
&self,
user_id: &UserId,
@ -563,6 +565,7 @@ impl Users {
Ok(())
}
#[tracing::instrument(skip(self))]
pub fn keys_changed(
&self,
user_or_room_id: &str,
@ -738,6 +741,7 @@ impl Users {
Ok(())
}
#[tracing::instrument(skip(self))]
pub fn get_to_device_events(
&self,
user_id: &UserId,
@ -760,6 +764,7 @@ impl Users {
Ok(events)
}
#[tracing::instrument(skip(self))]
pub fn remove_to_device_events(
&self,
user_id: &UserId,