MirrorHub/conduit
1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2024-11-09 06:11:04 +01:00
Code Issues Releases Activity

Merge branch 'rocksdb' into 'next'

Browse source 
feat: rocksdb backend

See merge request famedly/conduit!217
This commit is contained in:
Timo Kösters 2022-01-13 22:12:51 +00:00
commit 1119c2f510
18 changed files with 1390 additions and 548 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

727
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

8
Cargo.toml
View file

@ -78,17 +78,19 @@ crossbeam = { version = "0.8.1", optional = true }
num_cpus = "1.13.0"
threadpool = "1.8.1"
heed = { git = "https://github.com/timokoesters/heed.git", rev = "f6f825da7fb2c758867e05ad973ef800a6fe1d5d", optional = true }
rocksdb = { version = "0.17.0", features = ["multi-threaded-cf"], optional = true }
thread_local = "1.1.3"
# used for TURN server authentication
hmac = "0.11.0"
sha-1 = "0.9.8"
[features]
default = ["conduit_bin", "backend_sqlite"]
default = ["conduit_bin", "backend_sqlite", "backend_rocksdb"]
backend_sled = ["sled"]
backend_sqlite = ["sqlite"]
backend_heed = ["heed", "crossbeam", "parking_lot"]
sqlite = ["rusqlite", "parking_lot", "crossbeam", "tokio/signal"]
backend_heed = ["heed", "crossbeam"]
backend_rocksdb = ["rocksdb"]
sqlite = ["rusqlite", "parking_lot", "tokio/signal"]
conduit_bin = [] # TODO: add rocket to this when it is optional
[[bin]]

15
conduit-example.toml
View file

@ -1,11 +1,15 @@
[global]
# The server_name is the name of this server. It is used as a suffix for user
# The server_name is the pretty name of this server. It is used as a suffix for user
# and room ids. Examples: matrix.org, conduit.rs
# The Conduit server needs to be reachable at https://your.server.name/ on port
# 443 (client-server) and 8448 (federation) OR you can create /.well-known
# files to redirect requests. See
# The Conduit server needs all /_matrix/ requests to be reachable at
# https://your.server.name/ on port 443 (client-server) and 8448 (federation).
# If that's not possible for you, you can create /.well-known files to redirect
# requests. See
# https://matrix.org/docs/spec/client_server/latest#get-well-known-matrix-client
# and https://matrix.org/docs/spec/server_server/r0.1.4#get-well-known-matrix-server
# and
# https://matrix.org/docs/spec/server_server/r0.1.4#get-well-known-matrix-server
# for more information
# YOU NEED TO EDIT THIS
@ -13,6 +17,7 @@
# This is the only directory where Conduit will save its data
database_path = "/var/lib/conduit/"
database_backend = "rocksdb"
# The port Conduit will be running on. You need to set up a reverse proxy in
# your web server (e.g. apache or nginx), so all requests to /_matrix on port

62
src/client_server/context.rs
View file

@ -1,5 +1,9 @@
use crate::{database::DatabaseGuard, ConduitResult, Error, Ruma};
use ruma::api::client::{error::ErrorKind, r0::context::get_context};
use ruma::{
api::client::{error::ErrorKind, r0::context::get_context},
events::EventType,
};
use std::collections::HashSet;
use std::convert::TryFrom;
#[cfg(feature = "conduit_bin")]
@ -21,6 +25,7 @@ pub async fn get_context_route(
body: Ruma<get_context::Request<'_>>,
) -> ConduitResult<get_context::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let sender_device = body.sender_device.as_ref().expect("user is authenticated");
if !db.rooms.is_joined(sender_user, &body.room_id)? {
return Err(Error::BadRequest(
@ -29,6 +34,8 @@ pub async fn get_context_route(
));
}
let mut lazy_loaded = HashSet::new();
let base_pdu_id = db
.rooms
.get_pdu_id(&body.event_id)?
@ -45,8 +52,18 @@ pub async fn get_context_route(
.ok_or(Error::BadRequest(
ErrorKind::NotFound,
"Base event not found.",
))?
.to_room_event();
))?;
if !db.rooms.lazy_load_was_sent_before(
&sender_user,
&sender_device,
&body.room_id,
&base_event.sender,
)? {
lazy_loaded.insert(base_event.sender.clone());
}
let base_event = base_event.to_room_event();
let events_before: Vec<_> = db
.rooms
@ -60,6 +77,17 @@ pub async fn get_context_route(
.filter_map(|r| r.ok()) // Remove buggy events
.collect();
for (_, event) in &events_before {
if !db.rooms.lazy_load_was_sent_before(
&sender_user,
&sender_device,
&body.room_id,
&event.sender,
)? {
lazy_loaded.insert(event.sender.clone());
}
}
let start_token = events_before
.last()
.and_then(|(pdu_id, _)| db.rooms.pdu_count(pdu_id).ok())
@ -82,6 +110,17 @@ pub async fn get_context_route(
.filter_map(|r| r.ok()) // Remove buggy events
.collect();
for (_, event) in &events_after {
if !db.rooms.lazy_load_was_sent_before(
&sender_user,
&sender_device,
&body.room_id,
&event.sender,
)? {
lazy_loaded.insert(event.sender.clone());
}
}
let end_token = events_after
.last()
.and_then(|(pdu_id, _)| db.rooms.pdu_count(pdu_id).ok())
@ -92,18 +131,23 @@ pub async fn get_context_route(
.map(|(_, pdu)| pdu.to_room_event())
.collect();
let mut state = Vec::new();
for ll_id in &lazy_loaded {
if let Some(member_event) =
db.rooms
.room_state_get(&body.room_id, &EventType::RoomMember, ll_id.as_str())?
{
state.push(member_event.to_state_event());
}
}
let resp = get_context::Response {
start: start_token,
end: end_token,
events_before,
event: Some(base_event),
events_after,
state: db // TODO: State at event
.rooms
.room_state_full(&body.room_id)?
.values()
.map(|pdu| pdu.to_state_event())
.collect(),
state,
};
Ok(resp.into())

2
src/client_server/device.rs
View file

@ -85,7 +85,7 @@ pub async fn update_device_route(
Ok(update_device::Response {}.into())
}
/// # `PUT /_matrix/client/r0/devices/{deviceId}`
/// # `DELETE /_matrix/client/r0/devices/{deviceId}`
///
/// Deletes the given device.
///

57
src/client_server/filter.rs
View file

@ -1,32 +1,47 @@
use crate::{utils, ConduitResult};
use ruma::api::client::r0::filter::{self, create_filter, get_filter};
use crate::{database::DatabaseGuard, ConduitResult, Error, Ruma};
use ruma::api::client::{
error::ErrorKind,
r0::filter::{create_filter, get_filter},
};
#[cfg(feature = "conduit_bin")]
use rocket::{get, post};
/// # `GET /_matrix/client/r0/user/{userId}/filter/{filterId}`
///
/// TODO: Loads a filter that was previously created.
#[cfg_attr(feature = "conduit_bin", get("/_matrix/client/r0/user/<_>/filter/<_>"))]
#[tracing::instrument]
pub async fn get_filter_route() -> ConduitResult<get_filter::Response> {
// TODO
Ok(get_filter::Response::new(filter::IncomingFilterDefinition {
event_fields: None,
event_format: filter::EventFormat::default(),
account_data: filter::IncomingFilter::default(),
room: filter::IncomingRoomFilter::default(),
presence: filter::IncomingFilter::default(),
})
.into())
/// Loads a filter that was previously created.
///
/// - A user can only access their own filters
#[cfg_attr(
feature = "conduit_bin",
get("/_matrix/client/r0/user/<_>/filter/<_>", data = "<body>")
)]
#[tracing::instrument(skip(db, body))]
pub async fn get_filter_route(
db: DatabaseGuard,
body: Ruma<get_filter::Request<'_>>,
) -> ConduitResult<get_filter::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let filter = match db.users.get_filter(sender_user, &body.filter_id)? {
Some(filter) => filter,
None => return Err(Error::BadRequest(ErrorKind::NotFound, "Filter not found.")),
};
Ok(get_filter::Response::new(filter).into())
}
/// # `PUT /_matrix/client/r0/user/{userId}/filter`
///
/// TODO: Creates a new filter to be used by other endpoints.
#[cfg_attr(feature = "conduit_bin", post("/_matrix/client/r0/user/<_>/filter"))]
#[tracing::instrument]
pub async fn create_filter_route() -> ConduitResult<create_filter::Response> {
// TODO
Ok(create_filter::Response::new(utils::random_string(10)).into())
/// Creates a new filter to be used by other endpoints.
#[cfg_attr(
feature = "conduit_bin",
post("/_matrix/client/r0/user/<_>/filter", data = "<body>")
)]
#[tracing::instrument(skip(db, body))]
pub async fn create_filter_route(
db: DatabaseGuard,
body: Ruma<create_filter::Request<'_>>,
) -> ConduitResult<create_filter::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
Ok(create_filter::Response::new(db.users.create_filter(sender_user, &body.filter)?).into())
}

86
src/client_server/message.rs
View file

@ -6,7 +6,11 @@ use ruma::{
},
events::EventType,
};
use std::{collections::BTreeMap, convert::TryInto, sync::Arc};
use std::{
collections::{BTreeMap, HashSet},
convert::TryInto,
sync::Arc,
};
#[cfg(feature = "conduit_bin")]
use rocket::{get, put};
@ -117,6 +121,7 @@ pub async fn get_message_events_route(
body: Ruma<get_message_events::Request<'_>>,
) -> ConduitResult<get_message_events::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let sender_device = body.sender_device.as_ref().expect("user is authenticated");
if !db.rooms.is_joined(sender_user, &body.room_id)? {
return Err(Error::BadRequest(
@ -133,9 +138,18 @@ pub async fn get_message_events_route(
let to = body.to.as_ref().map(|t| t.parse());
db.rooms
.lazy_load_confirm_delivery(&sender_user, &sender_device, &body.room_id, from)?;
// Use limit or else 10
let limit = body.limit.try_into().map_or(10_usize, |l: u32| l as usize);
let next_token;
let mut resp = get_message_events::Response::new();
let mut lazy_loaded = HashSet::new();
match body.dir {
get_message_events::Direction::Forward => {
let events_after: Vec<_> = db
@ -152,21 +166,27 @@ pub async fn get_message_events_route(
.take_while(|&(k, _)| Some(Ok(k)) != to) // Stop at `to`
.collect();
let end_token = events_after.last().map(|(count, _)| count.to_string());
for (_, event) in &events_after {
if !db.rooms.lazy_load_was_sent_before(
&sender_user,
&sender_device,
&body.room_id,
&event.sender,
)? {
lazy_loaded.insert(event.sender.clone());
}
}
next_token = events_after.last().map(|(count, _)| count).copied();
let events_after: Vec<_> = events_after
.into_iter()
.map(|(_, pdu)| pdu.to_room_event())
.collect();
let resp = get_message_events::Response {
start: body.from.to_owned(),
end: end_token,
chunk: events_after,
state: Vec::new(),
};
Ok(resp.into())
resp.start = body.from.to_owned();
resp.end = next_token.map(|count| count.to_string());
resp.chunk = events_after;
}
get_message_events::Direction::Backward => {
let events_before: Vec<_> = db
@ -183,21 +203,49 @@ pub async fn get_message_events_route(
.take_while(|&(k, _)| Some(Ok(k)) != to) // Stop at `to`
.collect();
let start_token = events_before.last().map(|(count, _)| count.to_string());
for (_, event) in &events_before {
if !db.rooms.lazy_load_was_sent_before(
&sender_user,
&sender_device,
&body.room_id,
&event.sender,
)? {
lazy_loaded.insert(event.sender.clone());
}
}
next_token = events_before.last().map(|(count, _)| count).copied();
let events_before: Vec<_> = events_before
.into_iter()
.map(|(_, pdu)| pdu.to_room_event())
.collect();
let resp = get_message_events::Response {
start: body.from.to_owned(),
end: start_token,
chunk: events_before,
state: Vec::new(),
};
resp.start = body.from.to_owned();
resp.end = next_token.map(|count| count.to_string());
resp.chunk = events_before;
}
}
resp.state = Vec::new();
for ll_id in &lazy_loaded {
if let Some(member_event) =
db.rooms
.room_state_get(&body.room_id, &EventType::RoomMember, ll_id.as_str())?
{
resp.state.push(member_event.to_state_event());
}
}
if let Some(next_token) = next_token {
db.rooms.lazy_load_mark_sent(
&sender_user,
&sender_device,
&body.room_id,
lazy_loaded,
next_token,
);
}
Ok(resp.into())
}
}
}

199
src/client_server/sync.rs
View file

@ -1,6 +1,10 @@
use crate::{database::DatabaseGuard, ConduitResult, Database, Error, Result, Ruma, RumaResponse};
use ruma::{
api::client::r0::{sync::sync_events, uiaa::UiaaResponse},
api::client::r0::{
filter::{IncomingFilterDefinition, LazyLoadOptions},
sync::sync_events,
uiaa::UiaaResponse,
},
events::{
room::member::{MembershipState, RoomMemberEventContent},
AnySyncEphemeralRoomEvent, EventType,
@ -36,13 +40,15 @@ use rocket::{get, tokio};
/// Calling this endpoint with a `since` parameter from a previous `next_batch` returns:
/// For joined rooms:
/// - Some of the most recent events of each timeline that happened after since
/// - If user joined the room after since: All state events and device list updates in that room
/// - If user joined the room after since: All state events (unless lazy loading is activated) and
/// all device list updates in that room
/// - If the user was already in the room: A list of all events that are in the state now, but were
/// not in the state at `since`
/// - If the state we send contains a member event: Joined and invited member counts, heroes
/// - Device list updates that happened after `since`
/// - If there are events in the timeline we send or the user send updated his read mark: Notification counts
/// - EDUs that are active now (read receipts, typing updates, presence)
/// - TODO: Allow multiple sync streams to support Pantalaimon
///
/// For invited rooms:
/// - If the user was invited after `since`: A subset of the state of the room at the point of the invite
@ -77,34 +83,32 @@ pub async fn sync_events_route(
Entry::Vacant(v) => {
let (tx, rx) = tokio::sync::watch::channel(None);
v.insert((body.since.clone(), rx.clone()));
tokio::spawn(sync_helper_wrapper(
Arc::clone(&arc_db),
sender_user.clone(),
sender_device.clone(),
body.since.clone(),
body.full_state,
body.timeout,
body,
tx,
));
v.insert((body.since.clone(), rx)).1.clone()
rx
}
Entry::Occupied(mut o) => {
if o.get().0 != body.since {
let (tx, rx) = tokio::sync::watch::channel(None);
o.insert((body.since.clone(), rx.clone()));
tokio::spawn(sync_helper_wrapper(
Arc::clone(&arc_db),
sender_user.clone(),
sender_device.clone(),
body.since.clone(),
body.full_state,
body.timeout,
body,
tx,
));
o.insert((body.since.clone(), rx.clone()));
rx
} else {
o.get().1.clone()
@ -135,18 +139,16 @@ async fn sync_helper_wrapper(
db: Arc<DatabaseGuard>,
sender_user: Box<UserId>,
sender_device: Box<DeviceId>,
since: Option<String>,
full_state: bool,
timeout: Option<Duration>,
body: sync_events::IncomingRequest,
tx: Sender<Option<ConduitResult<sync_events::Response>>>,
) {
let since = body.since.clone();
let r = sync_helper(
Arc::clone(&db),
sender_user.clone(),
sender_device.clone(),
since.clone(),
full_state,
timeout,
body,
)
.await;
@ -179,9 +181,7 @@ async fn sync_helper(
db: Arc<DatabaseGuard>,
sender_user: Box<UserId>,
sender_device: Box<DeviceId>,
since: Option<String>,
full_state: bool,
timeout: Option<Duration>,
body: sync_events::IncomingRequest,
// bool = caching allowed
) -> Result<(sync_events::Response, bool), Error> {
// TODO: match body.set_presence {
@ -193,8 +193,26 @@ async fn sync_helper(
let next_batch = db.globals.current_count()?;
let next_batch_string = next_batch.to_string();
// Load filter
let filter = match body.filter {
None => IncomingFilterDefinition::default(),
Some(sync_events::IncomingFilter::FilterDefinition(filter)) => filter,
Some(sync_events::IncomingFilter::FilterId(filter_id)) => db
.users
.get_filter(&sender_user, &filter_id)?
.unwrap_or_default(),
};
let (lazy_load_enabled, lazy_load_send_redundant) = match filter.room.state.lazy_load_options {
LazyLoadOptions::Enabled {
include_redundant_members: redundant,
} => (true, redundant),
_ => (false, false),
};
let mut joined_rooms = BTreeMap::new();
let since = since
let since = body
.since
.clone()
.and_then(|string| string.parse().ok())
.unwrap_or(0);
@ -264,6 +282,14 @@ async fn sync_helper(
// limited unless there are events in non_timeline_pdus
let limited = non_timeline_pdus.next().is_some();
let mut timeline_users = HashSet::new();
for (_, event) in &timeline_pdus {
timeline_users.insert(event.sender.as_str().to_owned());
}
db.rooms
.lazy_load_confirm_delivery(&sender_user, &sender_device, &room_id, since)?;
// Database queries:
let current_shortstatehash = db
@ -344,14 +370,58 @@ async fn sync_helper(
state_events,
) = if since_shortstatehash.is_none() {
// Probably since = 0, we will do an initial sync
let (joined_member_count, invited_member_count, heroes) = calculate_counts()?;
let current_state_ids = db.rooms.state_full_ids(current_shortstatehash)?;
let state_events: Vec<_> = current_state_ids
.iter()
.map(|(_, id)| db.rooms.get_pdu(id))
.filter_map(|r| r.ok().flatten())
.collect();
let mut state_events = Vec::new();
let mut lazy_loaded = HashSet::new();
for (shortstatekey, id) in current_state_ids {
let (event_type, state_key) = db.rooms.get_statekey_from_short(shortstatekey)?;
if event_type != EventType::RoomMember {
let pdu = match db.rooms.get_pdu(&id)? {
Some(pdu) => pdu,
None => {
error!("Pdu in state not found: {}", id);
continue;
}
};
state_events.push(pdu);
} else if !lazy_load_enabled
|| body.full_state
|| timeline_users.contains(&state_key)
{
let pdu = match db.rooms.get_pdu(&id)? {
Some(pdu) => pdu,
None => {
error!("Pdu in state not found: {}", id);
continue;
}
};
lazy_loaded.insert(
UserId::parse(state_key.as_ref())
.expect("they are in timeline_users, so they should be correct"),
);
state_events.push(pdu);
}
}
// Reset lazy loading because this is an initial sync
db.rooms
.lazy_load_reset(&sender_user, &sender_device, &room_id)?;
// The state_events above should contain all timeline_users, let's mark them as lazy
// loaded.
db.rooms.lazy_load_mark_sent(
&sender_user,
&sender_device,
&room_id,
lazy_loaded,
next_batch,
);
(
heroes,
@ -387,21 +457,68 @@ async fn sync_helper(
let since_state_ids = db.rooms.state_full_ids(since_shortstatehash)?;
let state_events = if joined_since_last_sync {
current_state_ids
.iter()
.map(|(_, id)| db.rooms.get_pdu(id))
.filter_map(|r| r.ok().flatten())
.collect::<Vec<_>>()
} else {
current_state_ids
.iter()
.filter(|(key, id)| since_state_ids.get(key) != Some(id))
.map(|(_, id)| db.rooms.get_pdu(id))
.filter_map(|r| r.ok().flatten())
.collect()
let mut state_events = Vec::new();
let mut lazy_loaded = HashSet::new();
for (key, id) in current_state_ids {
if body.full_state || since_state_ids.get(&key) != Some(&id) {
let pdu = match db.rooms.get_pdu(&id)? {
Some(pdu) => pdu,
None => {
error!("Pdu in state not found: {}", id);
continue;
}
};
if pdu.kind == EventType::RoomMember {
match UserId::parse(
pdu.state_key
.as_ref()
.expect("State event has state key")
.clone(),
) {
Ok(state_key_userid) => {
lazy_loaded.insert(state_key_userid);
}
Err(e) => error!("Invalid state key for member event: {}", e),
}
}
state_events.push(pdu);
}
}
for (_, event) in &timeline_pdus {
if lazy_loaded.contains(&event.sender) {
continue;
}
if !db.rooms.lazy_load_was_sent_before(
&sender_user,
&sender_device,
&room_id,
&event.sender,
)? || lazy_load_send_redundant
{
if let Some(member_event) = db.rooms.room_state_get(
&room_id,
&EventType::RoomMember,
event.sender.as_str(),
)? {
lazy_loaded.insert(event.sender.clone());
state_events.push(member_event);
}
}
}
db.rooms.lazy_load_mark_sent(
&sender_user,
&sender_device,
&room_id,
lazy_loaded,
next_batch,
);
let encrypted_room = db
.rooms
.state_get(current_shortstatehash, &EventType::RoomEncryption, "")?
@ -767,7 +884,7 @@ async fn sync_helper(
};
// TODO: Retry the endpoint instead of returning (waiting for #118)
if !full_state
if !body.full_state
&& response.rooms.is_empty()
&& response.presence.is_empty()
&& response.account_data.is_empty()
@ -776,7 +893,7 @@ async fn sync_helper(
{
// Hang a few seconds so requests are not spammed
// Stop hanging if new info arrives
let mut duration = timeout.unwrap_or_default();
let mut duration = body.timeout.unwrap_or_default();
if duration.as_secs() > 30 {
duration = Duration::from_secs(30);
}

127
src/database.rs
View file

@ -44,13 +44,15 @@ use self::proxy::ProxyConfig;
#[derive(Clone, Debug, Deserialize)]
pub struct Config {
server_name: Box<ServerName>,
#[serde(default = "default_database_backend")]
database_backend: String,
database_path: String,
#[serde(default = "default_db_cache_capacity_mb")]
db_cache_capacity_mb: f64,
#[serde(default = "default_pdu_cache_capacity")]
pdu_cache_capacity: u32,
#[serde(default = "default_sqlite_wal_clean_second_interval")]
sqlite_wal_clean_second_interval: u32,
#[serde(default = "default_cleanup_second_interval")]
cleanup_second_interval: u32,
#[serde(default = "default_max_request_size")]
max_request_size: u32,
#[serde(default = "default_max_concurrent_requests")]
@ -117,15 +119,19 @@ fn true_fn() -> bool {
true
}
fn default_database_backend() -> String {
"sqlite".to_owned()
}
fn default_db_cache_capacity_mb() -> f64 {
200.0
}
fn default_pdu_cache_capacity() -> u32 {
100_000
1_000_000
}
fn default_sqlite_wal_clean_second_interval() -> u32 {
fn default_cleanup_second_interval() -> u32 {
1 * 60 // every minute
}
@ -145,17 +151,8 @@ fn default_turn_ttl() -> u64 {
60 * 60 * 24
}
#[cfg(feature = "sled")]
pub type Engine = abstraction::sled::Engine;
#[cfg(feature = "sqlite")]
pub type Engine = abstraction::sqlite::Engine;
#[cfg(feature = "heed")]
pub type Engine = abstraction::heed::Engine;
pub struct Database {
_db: Arc<Engine>,
_db: Arc<dyn DatabaseEngine>,
pub globals: globals::Globals,
pub users: users::Users,
pub uiaa: uiaa::Uiaa,
@ -183,28 +180,54 @@ impl Database {
Ok(())
}
fn check_sled_or_sqlite_db(config: &Config) -> Result<()> {
#[cfg(feature = "backend_sqlite")]
{
fn check_db_setup(config: &Config) -> Result<()> {
let path = Path::new(&config.database_path);
let sled_exists = path.join("db").exists();
let sqlite_exists = path.join("conduit.db").exists();
let rocksdb_exists = path.join("IDENTITY").exists();
let mut count = 0;
if sled_exists {
count += 1;
}
if sqlite_exists {
// most likely an in-place directory, only warn
warn!("Both sled and sqlite databases are detected in database directory");
warn!("Currently running from the sqlite database, but consider removing sled database files to free up space")
} else {
error!(
"Sled database detected, conduit now uses sqlite for database operations"
);
error!("This database must be converted to sqlite, go to https://github.com/ShadowJonathan/conduit_toolbox#conduit_sled_to_sqlite");
count += 1;
}
if rocksdb_exists {
count += 1;
}
if count > 1 {
warn!("Multiple databases at database_path detected");
return Ok(());
}
if sled_exists {
if config.database_backend != "sled" {
return Err(Error::bad_config(
"sled database detected, migrate to sqlite",
"Found sled at database_path, but is not specified in config.",
));
}
}
if sqlite_exists {
if config.database_backend != "sqlite" {
return Err(Error::bad_config(
"Found sqlite at database_path, but is not specified in config.",
));
}
}
if rocksdb_exists {
if config.database_backend != "rocksdb" {
return Err(Error::bad_config(
"Found rocksdb at database_path, but is not specified in config.",
));
}
}
Ok(())
@ -212,14 +235,30 @@ impl Database {
/// Load an existing database or create a new one.
pub async fn load_or_create(config: &Config) -> Result<Arc<TokioRwLock<Self>>> {
Self::check_sled_or_sqlite_db(config)?;
Self::check_db_setup(config)?;
if !Path::new(&config.database_path).exists() {
std::fs::create_dir_all(&config.database_path)
.map_err(|_| Error::BadConfig("Database folder doesn't exists and couldn't be created (e.g. due to missing permissions). Please create the database folder yourself."))?;
}
let builder = Engine::open(config)?;
let builder: Arc<dyn DatabaseEngine> = match &*config.database_backend {
"sqlite" => {
#[cfg(not(feature = "sqlite"))]
return Err(Error::BadConfig("Database backend not found."));
#[cfg(feature = "sqlite")]
Arc::new(Arc::<abstraction::sqlite::Engine>::open(config)?)
}
"rocksdb" => {
#[cfg(not(feature = "rocksdb"))]
return Err(Error::BadConfig("Database backend not found."));
#[cfg(feature = "rocksdb")]
Arc::new(Arc::<abstraction::rocksdb::Engine>::open(config)?)
}
_ => {
return Err(Error::BadConfig("Database backend not found."));
}
};
if config.max_request_size < 1024 {
eprintln!("ERROR: Max request size is less than 1KB. Please increase it.");
@ -246,6 +285,7 @@ impl Database {
userid_masterkeyid: builder.open_tree("userid_masterkeyid")?,
userid_selfsigningkeyid: builder.open_tree("userid_selfsigningkeyid")?,
userid_usersigningkeyid: builder.open_tree("userid_usersigningkeyid")?,
userfilterid_filter: builder.open_tree("userfilterid_filter")?,
todeviceid_events: builder.open_tree("todeviceid_events")?,
},
uiaa: uiaa::Uiaa {
@ -285,6 +325,8 @@ impl Database {
userroomid_leftstate: builder.open_tree("userroomid_leftstate")?,
roomuserid_leftcount: builder.open_tree("roomuserid_leftcount")?,
lazyloadedids: builder.open_tree("lazyloadedids")?,
userroomid_notificationcount: builder.open_tree("userroomid_notificationcount")?,
userroomid_highlightcount: builder.open_tree("userroomid_highlightcount")?,
@ -320,6 +362,7 @@ impl Database {
statekeyshort_cache: Mutex::new(LruCache::new(1_000_000)),
our_real_users_cache: RwLock::new(HashMap::new()),
appservice_in_room_cache: RwLock::new(HashMap::new()),
lazy_load_waiting: Mutex::new(HashMap::new()),
stateinfo_cache: Mutex::new(LruCache::new(1000)),
},
account_data: account_data::AccountData {
@ -777,10 +820,7 @@ impl Database {
drop(guard);
#[cfg(feature = "sqlite")]
{
Self::start_wal_clean_task(Arc::clone(&db), config).await;
}
Self::start_cleanup_task(Arc::clone(&db), config).await;
Ok(db)
}
@ -918,15 +958,8 @@ impl Database {
res
}
#[cfg(feature = "sqlite")]
#[tracing::instrument(skip(self))]
pub fn flush_wal(&self) -> Result<()> {
self._db.flush_wal()
}
#[cfg(feature = "sqlite")]
#[tracing::instrument(skip(db, config))]
pub async fn start_wal_clean_task(db: Arc<TokioRwLock<Self>>, config: &Config) {
pub async fn start_cleanup_task(db: Arc<TokioRwLock<Self>>, config: &Config) {
use tokio::time::interval;
#[cfg(unix)]
@ -935,7 +968,7 @@ impl Database {
use std::time::{Duration, Instant};
let timer_interval = Duration::from_secs(config.sqlite_wal_clean_second_interval as u64);
let timer_interval = Duration::from_secs(config.cleanup_second_interval as u64);
tokio::spawn(async move {
let mut i = interval(timer_interval);
@ -946,23 +979,23 @@ impl Database {
#[cfg(unix)]
tokio::select! {
_ = i.tick() => {
info!("wal-trunc: Timer ticked");
info!("cleanup: Timer ticked");
}
_ = s.recv() => {
info!("wal-trunc: Received SIGHUP");
info!("cleanup: Received SIGHUP");
}
};
#[cfg(not(unix))]
{
i.tick().await;
info!("wal-trunc: Timer ticked")
info!("cleanup: Timer ticked")
}
let start = Instant::now();
if let Err(e) = db.read().await.flush_wal() {
error!("wal-trunc: Errored: {}", e);
if let Err(e) = db.read().await._db.cleanup() {
error!("cleanup: Errored: {}", e);
} else {
info!("wal-trunc: Flushed in {:?}", start.elapsed());
info!("cleanup: Finished in {:?}", start.elapsed());
}
}
});

21
src/database/abstraction.rs
View file

@ -12,13 +12,24 @@ pub mod sqlite;
#[cfg(feature = "heed")]
pub mod heed;
#[cfg(any(feature = "sqlite", feature = "heed"))]
#[cfg(feature = "rocksdb")]
pub mod rocksdb;
#[cfg(any(feature = "sqlite", feature = "rocksdb", feature = "heed"))]
pub mod watchers;
pub trait DatabaseEngine: Sized {
fn open(config: &Config) -> Result<Arc<Self>>;
fn open_tree(self: &Arc<Self>, name: &'static str) -> Result<Arc<dyn Tree>>;
fn flush(self: &Arc<Self>) -> Result<()>;
pub trait DatabaseEngine: Send + Sync {
fn open(config: &Config) -> Result<Self>
where
Self: Sized;
fn open_tree(&self, name: &'static str) -> Result<Arc<dyn Tree>>;
fn flush(self: &Self) -> Result<()>;
fn cleanup(self: &Self) -> Result<()> {
Ok(())
}
fn memory_usage(self: &Self) -> Result<String> {
Ok("Current database engine does not support memory usage reporting.".to_string())
}
}
pub trait Tree: Send + Sync {

231
src/database/abstraction/rocksdb.rs Normal file
View file

@ -0,0 +1,231 @@
use super::{super::Config, watchers::Watchers, DatabaseEngine, Tree};
use crate::{utils, Result};
use std::{future::Future, pin::Pin, sync::Arc, sync::RwLock};
pub struct Engine {
rocks: rocksdb::DBWithThreadMode<rocksdb::MultiThreaded>,
cache_capacity_bytes: usize,
cache: rocksdb::Cache,
old_cfs: Vec<String>,
}
pub struct RocksDbEngineTree<'a> {
db: Arc<Engine>,
name: &'a str,
watchers: Watchers,
write_lock: RwLock<()>,
}
fn db_options(cache_capacity_bytes: usize, rocksdb_cache: &rocksdb::Cache) -> rocksdb::Options {
let mut block_based_options = rocksdb::BlockBasedOptions::default();
block_based_options.set_block_cache(rocksdb_cache);
// "Difference of spinning disk"
// https://zhangyuchi.gitbooks.io/rocksdbbook/content/RocksDB-Tuning-Guide.html
block_based_options.set_block_size(4 * 1024);
block_based_options.set_cache_index_and_filter_blocks(true);
let mut db_opts = rocksdb::Options::default();
db_opts.set_block_based_table_factory(&block_based_options);
db_opts.set_optimize_filters_for_hits(true);
db_opts.set_skip_stats_update_on_db_open(true);
db_opts.set_level_compaction_dynamic_level_bytes(true);
db_opts.set_target_file_size_base(256 * 1024 * 1024);
//db_opts.set_compaction_readahead_size(2 * 1024 * 1024);
//db_opts.set_use_direct_reads(true);
//db_opts.set_use_direct_io_for_flush_and_compaction(true);
db_opts.create_if_missing(true);
db_opts.increase_parallelism(num_cpus::get() as i32);
db_opts.set_max_open_files(512);
db_opts.set_compression_type(rocksdb::DBCompressionType::Zstd);
db_opts.set_compaction_style(rocksdb::DBCompactionStyle::Level);
db_opts.optimize_level_style_compaction(cache_capacity_bytes);
let prefix_extractor = rocksdb::SliceTransform::create_fixed_prefix(1);
db_opts.set_prefix_extractor(prefix_extractor);
db_opts
}
impl DatabaseEngine for Arc<Engine> {
fn open(config: &Config) -> Result<Self> {
let cache_capacity_bytes = (config.db_cache_capacity_mb * 1024.0 * 1024.0) as usize;
let rocksdb_cache = rocksdb::Cache::new_lru_cache(cache_capacity_bytes).unwrap();
let db_opts = db_options(cache_capacity_bytes, &rocksdb_cache);
let cfs = rocksdb::DBWithThreadMode::<rocksdb::MultiThreaded>::list_cf(
&db_opts,
&config.database_path,
)
.unwrap_or_default();
let db = rocksdb::DBWithThreadMode::<rocksdb::MultiThreaded>::open_cf_descriptors(
&db_opts,
&config.database_path,
cfs.iter().map(|name| {
rocksdb::ColumnFamilyDescriptor::new(
name,
db_options(cache_capacity_bytes, &rocksdb_cache),
)
}),
)?;
Ok(Arc::new(Engine {
rocks: db,
cache_capacity_bytes,
cache: rocksdb_cache,
old_cfs: cfs,
}))
}
fn open_tree(&self, name: &'static str) -> Result<Arc<dyn Tree>> {
if !self.old_cfs.contains(&name.to_owned()) {
// Create if it didn't exist
let _ = self
.rocks
.create_cf(name, &db_options(self.cache_capacity_bytes, &self.cache));
}
Ok(Arc::new(RocksDbEngineTree {
name,
db: Arc::clone(self),
watchers: Watchers::default(),
write_lock: RwLock::new(()),
}))
}
fn flush(&self) -> Result<()> {
// TODO?
Ok(())
}
fn memory_usage(&self) -> Result<String> {
let stats =
rocksdb::perf::get_memory_usage_stats(Some(&[&self.rocks]), Some(&[&self.cache]))?;
Ok(format!(
"Approximate memory usage of all the mem-tables: {:.3} MB\n\
Approximate memory usage of un-flushed mem-tables: {:.3} MB\n\
Approximate memory usage of all the table readers: {:.3} MB\n\
Approximate memory usage by cache: {:.3} MB\n\
Approximate memory usage by cache pinned: {:.3} MB\n\
",
stats.mem_table_total as f64 / 1024.0 / 1024.0,
stats.mem_table_unflushed as f64 / 1024.0 / 1024.0,
stats.mem_table_readers_total as f64 / 1024.0 / 1024.0,
stats.cache_total as f64 / 1024.0 / 1024.0,
self.cache.get_pinned_usage() as f64 / 1024.0 / 1024.0,
))
}
}
impl RocksDbEngineTree<'_> {
fn cf(&self) -> Arc<rocksdb::BoundColumnFamily<'_>> {
self.db.rocks.cf_handle(self.name).unwrap()
}
}
impl Tree for RocksDbEngineTree<'_> {
fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
Ok(self.db.rocks.get_cf(&self.cf(), key)?)
}
fn insert(&self, key: &[u8], value: &[u8]) -> Result<()> {
let lock = self.write_lock.read().unwrap();
self.db.rocks.put_cf(&self.cf(), key, value)?;
drop(lock);
self.watchers.wake(key);
Ok(())
}
fn insert_batch<'a>(&self, iter: &mut dyn Iterator<Item = (Vec<u8>, Vec<u8>)>) -> Result<()> {
for (key, value) in iter {
self.db.rocks.put_cf(&self.cf(), key, value)?;
}
Ok(())
}
fn remove(&self, key: &[u8]) -> Result<()> {
Ok(self.db.rocks.delete_cf(&self.cf(), key)?)
}
fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> {
Box::new(
self.db
.rocks
.iterator_cf(&self.cf(), rocksdb::IteratorMode::Start)
.map(|(k, v)| (Vec::from(k), Vec::from(v))),
)
}
fn iter_from<'a>(
&'a self,
from: &[u8],
backwards: bool,
) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> {
Box::new(
self.db
.rocks
.iterator_cf(
&self.cf(),
rocksdb::IteratorMode::From(
from,
if backwards {
rocksdb::Direction::Reverse
} else {
rocksdb::Direction::Forward
},
),
)
.map(|(k, v)| (Vec::from(k), Vec::from(v))),
)
}
fn increment(&self, key: &[u8]) -> Result<Vec<u8>> {
let lock = self.write_lock.write().unwrap();
let old = self.db.rocks.get_cf(&self.cf(), &key)?;
let new = utils::increment(old.as_deref()).unwrap();
self.db.rocks.put_cf(&self.cf(), key, &new)?;
drop(lock);
Ok(new)
}
fn increment_batch<'a>(&self, iter: &mut dyn Iterator<Item = Vec<u8>>) -> Result<()> {
let lock = self.write_lock.write().unwrap();
for key in iter {
let old = self.db.rocks.get_cf(&self.cf(), &key)?;
let new = utils::increment(old.as_deref()).unwrap();
self.db.rocks.put_cf(&self.cf(), key, new)?;
}
drop(lock);
Ok(())
}
fn scan_prefix<'a>(
&'a self,
prefix: Vec<u8>,
) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> {
Box::new(
self.db
.rocks
.iterator_cf(
&self.cf(),
rocksdb::IteratorMode::From(&prefix, rocksdb::Direction::Forward),
)
.map(|(k, v)| (Vec::from(k), Vec::from(v)))
.take_while(move |(k, _)| k.starts_with(&prefix)),
)
}
fn watch_prefix<'a>(&'a self, prefix: &[u8]) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
self.watchers.watch(prefix)
}
}

14
src/database/abstraction/sqlite.rs
View file

@ -80,8 +80,8 @@ impl Engine {
}
}
impl DatabaseEngine for Engine {
fn open(config: &Config) -> Result<Arc<Self>> {
impl DatabaseEngine for Arc<Engine> {
fn open(config: &Config) -> Result<Self> {
let path = Path::new(&config.database_path).join("conduit.db");
// calculates cache-size per permanent connection
@ -92,7 +92,7 @@ impl DatabaseEngine for Engine {
/ ((num_cpus::get().max(1) * 2) + 1) as f64)
as u32;
let writer = Mutex::new(Self::prepare_conn(&path, cache_size_per_thread)?);
let writer = Mutex::new(Engine::prepare_conn(&path, cache_size_per_thread)?);
let arc = Arc::new(Engine {
writer,
@ -105,7 +105,7 @@ impl DatabaseEngine for Engine {
Ok(arc)
}
fn open_tree(self: &Arc<Self>, name: &str) -> Result<Arc<dyn Tree>> {
fn open_tree(&self, name: &str) -> Result<Arc<dyn Tree>> {
self.write_lock().execute(&format!("CREATE TABLE IF NOT EXISTS {} ( \"key\" BLOB PRIMARY KEY, \"value\" BLOB NOT NULL )", name), [])?;
Ok(Arc::new(SqliteTable {
@ -115,10 +115,14 @@ impl DatabaseEngine for Engine {
}))
}
fn flush(self: &Arc<Self>) -> Result<()> {
fn flush(&self) -> Result<()> {
// we enabled PRAGMA synchronous=normal, so this should not be necessary
Ok(())
}
fn cleanup(&self) -> Result<()> {
self.flush_wal()
}
}
pub struct SqliteTable {

8
src/database/abstraction/watchers.rs
View file

@ -1,8 +1,8 @@
use parking_lot::RwLock;
use std::{
collections::{hash_map, HashMap},
future::Future,
pin::Pin,
sync::RwLock,
};
use tokio::sync::watch;
@ -16,7 +16,7 @@ impl Watchers {
&'a self,
prefix: &[u8],
) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
let mut rx = match self.watchers.write().entry(prefix.to_vec()) {
let mut rx = match self.watchers.write().unwrap().entry(prefix.to_vec()) {
hash_map::Entry::Occupied(o) => o.get().1.clone(),
hash_map::Entry::Vacant(v) => {
let (tx, rx) = tokio::sync::watch::channel(());
@ -31,7 +31,7 @@ impl Watchers {
})
}
pub(super) fn wake(&self, key: &[u8]) {
let watchers = self.watchers.read();
let watchers = self.watchers.read().unwrap();
let mut triggered = Vec::new();
for length in 0..=key.len() {
@ -43,7 +43,7 @@ impl Watchers {
drop(watchers);
if !triggered.is_empty() {
let mut watchers = self.watchers.write();
let mut watchers = self.watchers.write().unwrap();
for prefix in triggered {
if let Some(tx) = watchers.remove(prefix) {
let _ = tx.0.send(());

8
src/database/admin.rs
View file

@ -14,6 +14,7 @@ pub enum AdminCommand {
RegisterAppservice(serde_yaml::Value),
UnregisterAppservice(String),
ListAppservices,
ShowMemoryUsage,
SendMessage(RoomMessageEventContent),
}
@ -113,6 +114,13 @@ impl Admin {
send_message(RoomMessageEventContent::text_plain("Failed to get appservices."), guard, &state_lock);
}
}
AdminCommand::ShowMemoryUsage => {
if let Ok(response) = guard._db.memory_usage() {
send_message(RoomMessageEventContent::text_plain(response), guard, &state_lock);
} else {
send_message(RoomMessageEventContent::text_plain("Failed to get database memory usage.".to_string()), guard, &state_lock);
}
}
AdminCommand::SendMessage(message) => {
send_message(message, guard, &state_lock);
}

99
src/database/rooms.rs
View file

@ -28,7 +28,7 @@ use ruma::{
push::{Action, Ruleset, Tweak},
serde::{CanonicalJsonObject, CanonicalJsonValue, Raw},
state_res::{self, RoomVersion, StateMap},
uint, EventId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
uint, DeviceId, EventId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
};
use serde::Deserialize;
use serde_json::value::to_raw_value;
@ -79,6 +79,8 @@ pub struct Rooms {
pub(super) userroomid_leftstate: Arc<dyn Tree>,
pub(super) roomuserid_leftcount: Arc<dyn Tree>,
pub(super) lazyloadedids: Arc<dyn Tree>, // LazyLoadedIds = UserId + DeviceId + RoomId + LazyLoadedUserId
pub(super) userroomid_notificationcount: Arc<dyn Tree>, // NotifyCount = u64
pub(super) userroomid_highlightcount: Arc<dyn Tree>, // HightlightCount = u64
@ -117,6 +119,8 @@ pub struct Rooms {
pub(super) shortstatekey_cache: Mutex<LruCache<u64, (EventType, String)>>,
pub(super) our_real_users_cache: RwLock<HashMap<Box<RoomId>, Arc<HashSet<Box<UserId>>>>>,
pub(super) appservice_in_room_cache: RwLock<HashMap<Box<RoomId>, HashMap<String, bool>>>,
pub(super) lazy_load_waiting:
Mutex<HashMap<(Box<UserId>, Box<DeviceId>, Box<RoomId>, u64), HashSet<Box<UserId>>>>,
pub(super) stateinfo_cache: Mutex<
LruCache<
u64,
@ -1689,6 +1693,9 @@ impl Rooms {
));
}
}
"database_memory_usage" => {
db.admin.send(AdminCommand::ShowMemoryUsage);
}
_ => {
db.admin.send(AdminCommand::SendMessage(
RoomMessageEventContent::text_plain(format!(
@ -3466,4 +3473,94 @@ impl Rooms {
Ok(())
}
#[tracing::instrument(skip(self))]
pub fn lazy_load_was_sent_before(
&self,
user_id: &UserId,
device_id: &DeviceId,
room_id: &RoomId,
ll_user: &UserId,
) -> Result<bool> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&device_id.as_bytes());
key.push(0xff);
key.extend_from_slice(&room_id.as_bytes());
key.push(0xff);
key.extend_from_slice(&ll_user.as_bytes());
Ok(self.lazyloadedids.get(&key)?.is_some())
}
#[tracing::instrument(skip(self))]
pub fn lazy_load_mark_sent(
&self,
user_id: &UserId,
device_id: &DeviceId,
room_id: &RoomId,
lazy_load: HashSet<Box<UserId>>,
count: u64,
) {
self.lazy_load_waiting.lock().unwrap().insert(
(
user_id.to_owned(),
device_id.to_owned(),
room_id.to_owned(),
count,
),
lazy_load,
);
}
#[tracing::instrument(skip(self))]
pub fn lazy_load_confirm_delivery(
&self,
user_id: &UserId,
device_id: &DeviceId,
room_id: &RoomId,
since: u64,
) -> Result<()> {
if let Some(user_ids) = self.lazy_load_waiting.lock().unwrap().remove(&(
user_id.to_owned(),
device_id.to_owned(),
room_id.to_owned(),
since,
)) {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
prefix.extend_from_slice(&device_id.as_bytes());
prefix.push(0xff);
prefix.extend_from_slice(&room_id.as_bytes());
prefix.push(0xff);
for ll_id in user_ids {
let mut key = prefix.clone();
key.extend_from_slice(&ll_id.as_bytes());
self.lazyloadedids.insert(&key, &[])?;
}
}
Ok(())
}
#[tracing::instrument(skip(self))]
pub fn lazy_load_reset(
&self,
user_id: &Box<UserId>,
device_id: &Box<DeviceId>,
room_id: &Box<RoomId>,
) -> Result<()> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
prefix.extend_from_slice(&device_id.as_bytes());
prefix.push(0xff);
prefix.extend_from_slice(&room_id.as_bytes());
prefix.push(0xff);
for (key, _) in self.lazyloadedids.scan_prefix(prefix) {
self.lazyloadedids.remove(&key)?;
}
Ok(())
}
}

48
src/database/users.rs
View file

@ -1,6 +1,9 @@
use crate::{utils, Error, Result};
use ruma::{
api::client::{error::ErrorKind, r0::device::Device},
api::client::{
error::ErrorKind,
r0::{device::Device, filter::IncomingFilterDefinition},
},
encryption::{CrossSigningKey, DeviceKeys, OneTimeKey},
events::{AnyToDeviceEvent, EventType},
identifiers::MxcUri,
@ -36,6 +39,8 @@ pub struct Users {
pub(super) userid_selfsigningkeyid: Arc<dyn Tree>,
pub(super) userid_usersigningkeyid: Arc<dyn Tree>,
pub(super) userfilterid_filter: Arc<dyn Tree>, // UserFilterId = UserId + FilterId
pub(super) todeviceid_events: Arc<dyn Tree>, // ToDeviceId = UserId + DeviceId + Count
}
@ -996,6 +1001,47 @@ impl Users {
// TODO: Unhook 3PID
Ok(())
}
/// Creates a new sync filter. Returns the filter id.
#[tracing::instrument(skip(self))]
pub fn create_filter(
&self,
user_id: &UserId,
filter: &IncomingFilterDefinition,
) -> Result<String> {
let filter_id = utils::random_string(4);
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(filter_id.as_bytes());
self.userfilterid_filter.insert(
&key,
&serde_json::to_vec(&filter).expect("filter is valid json"),
)?;
Ok(filter_id)
}
#[tracing::instrument(skip(self))]
pub fn get_filter(
&self,
user_id: &UserId,
filter_id: &str,
) -> Result<Option<IncomingFilterDefinition>> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(filter_id.as_bytes());
let raw = self.userfilterid_filter.get(&key)?;
if let Some(raw) = raw {
serde_json::from_slice(&raw)
.map_err(|_| Error::bad_database("Invalid filter event in db."))
} else {
Ok(None)
}
}
}
/// Ensure that a user only sees signatures from themselves and the target user

6
src/error.rs
View file

@ -39,6 +39,12 @@ pub enum Error {
#[cfg(feature = "heed")]
#[error("There was a problem with the connection to the heed database: {error}")]
HeedError { error: String },
#[cfg(feature = "rocksdb")]
#[error("There was a problem with the connection to the rocksdb database: {source}")]
RocksDbError {
#[from]
source: rocksdb::Error,
},
#[error("Could not generate an image.")]
ImageError {
#[from]

132
src/server_server.rs
View file

@ -1392,12 +1392,11 @@ async fn upgrade_outlier_to_timeline_pdu(
let mut starting_events = Vec::with_capacity(leaf_state.len());
for (k, id) in leaf_state {
let k = db
.rooms
.get_statekey_from_short(k)
.map_err(|_| "Failed to get_statekey_from_short.".to_owned())?;
if let Ok(k) = db.rooms.get_statekey_from_short(k) {
state.insert(k, id.clone());
} else {
warn!("Failed to get_statekey_from_short.");
}
starting_events.push(id);
}
@ -1687,25 +1686,6 @@ async fn upgrade_outlier_to_timeline_pdu(
// We do this by adding the current state to the list of fork states
extremity_sstatehashes.remove(&current_sstatehash);
fork_states.push(current_state_ids);
dbg!(&extremity_sstatehashes);
for (sstatehash, leaf_pdu) in extremity_sstatehashes {
let mut leaf_state = db
.rooms
.state_full_ids(sstatehash)
.map_err(|_| "Failed to ask db for room state.".to_owned())?;
if let Some(state_key) = &leaf_pdu.state_key {
let shortstatekey = db
.rooms
.get_or_create_shortstatekey(&leaf_pdu.kind, state_key, &db.globals)
.map_err(|_| "Failed to create shortstatekey.".to_owned())?;
leaf_state.insert(shortstatekey, Arc::from(&*leaf_pdu.event_id));
// Now it's the state after the pdu
}
fork_states.push(leaf_state);
}
// We also add state after incoming event to the fork states
let mut state_after = state_at_incoming_event.clone();
@ -1755,11 +1735,16 @@ async fn upgrade_outlier_to_timeline_pdu(
.into_iter()
.map(|map| {
map.into_iter()
.map(|(k, id)| db.rooms.get_statekey_from_short(k).map(|k| (k, id)))
.collect::<Result<StateMap<_>>>()
.filter_map(|(k, id)| {
db.rooms
.get_statekey_from_short(k)
.map(|k| (k, id))
.map_err(|e| warn!("Failed to get_statekey_from_short: {}", e))
.ok()
})
.collect::<Result<_>>()
.map_err(|_| "Failed to get_statekey_from_short.".to_owned())?;
.collect::<StateMap<_>>()
})
.collect();
let state = match state_res::resolve(
room_version_id,
@ -1871,45 +1856,86 @@ pub(crate) fn fetch_and_handle_outliers<'a>(
// a. Look in the main timeline (pduid_pdu tree)
// b. Look at outlier pdu tree
// (get_pdu_json checks both)
let local_pdu = db.rooms.get_pdu(id);
let pdu = match local_pdu {
Ok(Some(pdu)) => {
if let Ok(Some(local_pdu)) = db.rooms.get_pdu(id) {
trace!("Found {} in db", id);
(pdu, None)
pdus.push((local_pdu, None));
continue;
}
Ok(None) => {
// c. Ask origin server over federation
warn!("Fetching {} over federation.", id);
// We also handle its auth chain here so we don't get a stack overflow in
// handle_outlier_pdu.
let mut todo_auth_events = vec![Arc::clone(id)];
let mut events_in_reverse_order = Vec::new();
let mut events_all = HashSet::new();
while let Some(next_id) = todo_auth_events.pop() {
if events_all.contains(&next_id) {
continue;
}
if let Ok(Some(_)) = db.rooms.get_pdu(&next_id) {
trace!("Found {} in db", id);
continue;
}
warn!("Fetching {} over federation.", next_id);
match db
.sending
.send_federation_request(
&db.globals,
origin,
get_event::v1::Request { event_id: id },
get_event::v1::Request { event_id: &next_id },
)
.await
{
Ok(res) => {
warn!("Got {} over federation", id);
warn!("Got {} over federation", next_id);
let (calculated_event_id, value) =
match crate::pdu::gen_event_id_canonical_json(&res.pdu) {
Ok(t) => t,
Err(_) => {
back_off((**id).to_owned());
back_off((*next_id).to_owned());
continue;
}
};
if calculated_event_id != **id {
if calculated_event_id != *next_id {
warn!("Server didn't return event id we requested: requested: {}, we got {}. Event: {:?}",
id, calculated_event_id, &res.pdu);
next_id, calculated_event_id, &res.pdu);
}
// This will also fetch the auth chain
if let Some(auth_events) =
value.get("auth_events").and_then(|c| c.as_array())
{
for auth_event in auth_events {
if let Ok(auth_event) =
serde_json::from_value(auth_event.clone().into())
{
let a: Arc<EventId> = auth_event;
todo_auth_events.push(a);
} else {
warn!("Auth event id is not valid");
}
}
} else {
warn!("Auth event list invalid");
}
events_in_reverse_order.push((next_id.clone(), value));
events_all.insert(next_id);
}
Err(_) => {
warn!("Failed to fetch event: {}", next_id);
back_off((*next_id).to_owned());
}
}
}
for (next_id, value) in events_in_reverse_order.iter().rev() {
match handle_outlier_pdu(
origin,
create_event,
id,
&next_id,
room_id,
value.clone(),
db,
@ -1917,27 +1943,17 @@ pub(crate) fn fetch_and_handle_outliers<'a>(
)
.await
{
Ok((pdu, json)) => (pdu, Some(json)),
Err(e) => {
warn!("Authentication of event {} failed: {:?}", id, e);
back_off((**id).to_owned());
continue;
}
}
}
Err(_) => {
warn!("Failed to fetch event: {}", id);
back_off((**id).to_owned());
continue;
}
Ok((pdu, json)) => {
if next_id == id {
pdus.push((pdu, Some(json)));
}
}
Err(e) => {
warn!("Error loading {}: {}", id, e);
continue;
warn!("Authentication of event {} failed: {:?}", next_id, e);
back_off((**next_id).to_owned());
}
}
}
};
pdus.push(pdu);
}
pdus
})