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

Hot-Reloading Refactor

Browse source 
Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk 2024-05-09 15:59:08 -07:00 • committed by June 🍓🦴
parent ae1a4fd283
commit 6c1434c165
212 changed files with 5679 additions and 4206 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/service/sending/data.rs
View file

@ -6,7 +6,7 @@ use crate::Result;
type OutgoingSendingIter<'a> = Box<dyn Iterator<Item = Result<(Vec<u8>, Destination, SendingEvent)>> + 'a>;
type SendingEventIter<'a> = Box<dyn Iterator<Item = Result<(Vec<u8>, SendingEvent)>> + 'a>;
pub(crate) trait Data: Send + Sync {
pub trait Data: Send + Sync {
fn active_requests(&self) -> OutgoingSendingIter<'_>;
fn active_requests_for(&self, destination: &Destination) -> SendingEventIter<'_>;
fn delete_active_request(&self, key: Vec<u8>) -> Result<()>;

71
src/service/sending/mod.rs
View file

@ -1,27 +1,28 @@
use std::{fmt::Debug, sync::Arc};
pub(crate) use data::Data;
pub use data::Data;
use ruma::{
api::{appservice::Registration, OutgoingRequest},
OwnedServerName, OwnedUserId, RoomId, ServerName, UserId,
};
use tokio::sync::Mutex;
use tracing::warn;
use tokio::{sync::Mutex, task::JoinHandle};
use tracing::{error, warn};
use crate::{services, utils::server_name::server_is_ours, Config, Error, Result};
use crate::{server_is_ours, services, Config, Error, Result};
mod appservice;
mod data;
pub(crate) mod send;
pub(crate) mod sender;
pub(crate) use send::FedDest;
pub mod send;
pub mod sender;
pub use send::FedDest;
pub(crate) struct Service {
pub(crate) db: &'static dyn Data,
pub struct Service {
pub db: Arc<dyn Data>,
/// The state for a given state hash.
sender: loole::Sender<Msg>,
receiver: Mutex<loole::Receiver<Msg>>,
handler_join: Mutex<Option<JoinHandle<()>>>,
startup_netburst: bool,
startup_netburst_keep: i64,
}
@ -34,7 +35,7 @@ struct Msg {
}
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub(crate) enum Destination {
pub enum Destination {
Appservice(String),
Push(OwnedUserId, String), // user and pushkey
Normal(OwnedServerName),
@ -42,26 +43,42 @@ pub(crate) enum Destination {
#[allow(clippy::module_name_repetitions)]
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub(crate) enum SendingEvent {
pub enum SendingEvent {
Pdu(Vec<u8>), // pduid
Edu(Vec<u8>), // pdu json
Flush, // none
}
impl Service {
pub(crate) fn build(db: &'static dyn Data, config: &Config) -> Arc<Self> {
pub fn build(db: Arc<dyn Data>, config: &Config) -> Arc<Self> {
let (sender, receiver) = loole::unbounded();
Arc::new(Self {
db,
sender,
receiver: Mutex::new(receiver),
handler_join: Mutex::new(None),
startup_netburst: config.startup_netburst,
startup_netburst_keep: config.startup_netburst_keep,
})
}
pub async fn close(&self) {
self.interrupt();
if let Some(handler_join) = self.handler_join.lock().await.take() {
if let Err(e) = handler_join.await {
error!("Failed to shutdown: {e:?}");
}
}
}
pub fn interrupt(&self) {
if !self.sender.is_closed() {
self.sender.close();
}
}
#[tracing::instrument(skip(self, pdu_id, user, pushkey))]
pub(crate) fn send_pdu_push(&self, pdu_id: &[u8], user: &UserId, pushkey: String) -> Result<()> {
pub fn send_pdu_push(&self, pdu_id: &[u8], user: &UserId, pushkey: String) -> Result<()> {
let dest = Destination::Push(user.to_owned(), pushkey);
let event = SendingEvent::Pdu(pdu_id.to_owned());
let _cork = services().globals.db.cork()?;
@ -74,7 +91,7 @@ impl Service {
}
#[tracing::instrument(skip(self))]
pub(crate) fn send_pdu_appservice(&self, appservice_id: String, pdu_id: Vec<u8>) -> Result<()> {
pub fn send_pdu_appservice(&self, appservice_id: String, pdu_id: Vec<u8>) -> Result<()> {
let dest = Destination::Appservice(appservice_id);
let event = SendingEvent::Pdu(pdu_id);
let _cork = services().globals.db.cork()?;
@ -87,7 +104,7 @@ impl Service {
}
#[tracing::instrument(skip(self, room_id, pdu_id))]
pub(crate) fn send_pdu_room(&self, room_id: &RoomId, pdu_id: &[u8]) -> Result<()> {
pub fn send_pdu_room(&self, room_id: &RoomId, pdu_id: &[u8]) -> Result<()> {
let servers = services()
.rooms
.state_cache
@ -99,9 +116,7 @@ impl Service {
}
#[tracing::instrument(skip(self, servers, pdu_id))]
pub(crate) fn send_pdu_servers<I: Iterator<Item = OwnedServerName>>(
&self, servers: I, pdu_id: &[u8],
) -> Result<()> {
pub fn send_pdu_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I, pdu_id: &[u8]) -> Result<()> {
let requests = servers
.into_iter()
.map(|server| (Destination::Normal(server), SendingEvent::Pdu(pdu_id.to_owned())))
@ -125,7 +140,7 @@ impl Service {
}
#[tracing::instrument(skip(self, server, serialized))]
pub(crate) fn send_edu_server(&self, server: &ServerName, serialized: Vec<u8>) -> Result<()> {
pub fn send_edu_server(&self, server: &ServerName, serialized: Vec<u8>) -> Result<()> {
let dest = Destination::Normal(server.to_owned());
let event = SendingEvent::Edu(serialized);
let _cork = services().globals.db.cork()?;
@ -138,7 +153,7 @@ impl Service {
}
#[tracing::instrument(skip(self, room_id, serialized))]
pub(crate) fn send_edu_room(&self, room_id: &RoomId, serialized: Vec<u8>) -> Result<()> {
pub fn send_edu_room(&self, room_id: &RoomId, serialized: Vec<u8>) -> Result<()> {
let servers = services()
.rooms
.state_cache
@ -150,9 +165,7 @@ impl Service {
}
#[tracing::instrument(skip(self, servers, serialized))]
pub(crate) fn send_edu_servers<I: Iterator<Item = OwnedServerName>>(
&self, servers: I, serialized: Vec<u8>,
) -> Result<()> {
pub fn send_edu_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I, serialized: Vec<u8>) -> Result<()> {
let requests = servers
.into_iter()
.map(|server| (Destination::Normal(server), SendingEvent::Edu(serialized.clone())))
@ -177,7 +190,7 @@ impl Service {
}
#[tracing::instrument(skip(self, room_id))]
pub(crate) fn flush_room(&self, room_id: &RoomId) -> Result<()> {
pub fn flush_room(&self, room_id: &RoomId) -> Result<()> {
let servers = services()
.rooms
.state_cache
@ -189,7 +202,7 @@ impl Service {
}
#[tracing::instrument(skip(self, servers))]
pub(crate) fn flush_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I) -> Result<()> {
pub fn flush_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I) -> Result<()> {
let requests = servers.into_iter().map(Destination::Normal);
for dest in requests {
self.dispatch(Msg {
@ -203,7 +216,7 @@ impl Service {
}
#[tracing::instrument(skip(self, request), name = "request")]
pub(crate) async fn send_federation_request<T>(&self, dest: &ServerName, request: T) -> Result<T::IncomingResponse>
pub async fn send_federation_request<T>(&self, dest: &ServerName, request: T) -> Result<T::IncomingResponse>
where
T: OutgoingRequest + Debug,
{
@ -215,7 +228,7 @@ impl Service {
///
/// Only returns None if there is no url specified in the appservice
/// registration file
pub(crate) async fn send_appservice_request<T>(
pub async fn send_appservice_request<T>(
&self, registration: Registration, request: T,
) -> Result<Option<T::IncomingResponse>>
where
@ -227,7 +240,7 @@ impl Service {
/// Cleanup event data
/// Used for instance after we remove an appservice registration
#[tracing::instrument(skip(self))]
pub(crate) fn cleanup_events(&self, appservice_id: String) -> Result<()> {
pub fn cleanup_events(&self, appservice_id: String) -> Result<()> {
self.db
.delete_all_requests_for(&Destination::Appservice(appservice_id))?;
@ -243,7 +256,7 @@ impl Service {
impl Destination {
#[tracing::instrument(skip(self))]
pub(crate) fn get_prefix(&self) -> Vec<u8> {
pub fn get_prefix(&self) -> Vec<u8> {
let mut prefix = match self {
Destination::Appservice(server) => {
let mut p = b"+".to_vec();

222
src/service/sending/send.rs
View file

@ -13,7 +13,6 @@ use ruma::{
client::error::Error as RumaError, EndpointError, IncomingResponse, MatrixVersion, OutgoingRequest,
SendAccessToken,
},
events::room::message::RoomMessageEventContent,
OwnedServerName, ServerName,
};
use tracing::{debug, error, trace};
@ -28,7 +27,7 @@ use crate::{debug_error, debug_info, debug_warn, services, Error, Result};
///
/// # Examples:
/// ```rust
/// # use conduit::api::server_server::FedDest;
/// # use conduit_service::sending::FedDest;
/// # fn main() -> Result<(), std::net::AddrParseError> {
/// FedDest::Literal("198.51.100.3:8448".parse()?);
/// FedDest::Literal("[2001:db8::4:5]:443".parse()?);
@ -39,7 +38,7 @@ use crate::{debug_error, debug_info, debug_warn, services, Error, Result};
/// # }
/// ```
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum FedDest {
pub enum FedDest {
Literal(SocketAddr),
Named(String, String),
}
@ -52,7 +51,7 @@ struct ActualDest {
}
#[tracing::instrument(skip_all, name = "send")]
pub(crate) async fn send<T>(client: &Client, dest: &ServerName, req: T) -> Result<T::IncomingResponse>
pub async fn send<T>(client: &Client, dest: &ServerName, req: T) -> Result<T::IncomingResponse>
where
T: OutgoingRequest + Debug,
{
@ -195,7 +194,7 @@ async fn get_actual_dest(server_name: &ServerName) -> Result<ActualDest> {
} else {
cached = false;
validate_dest(server_name)?;
resolve_actual_dest(server_name, false, false).await?
resolve_actual_dest(server_name, false).await?
};
let string = dest.clone().into_https_string();
@ -211,162 +210,49 @@ async fn get_actual_dest(server_name: &ServerName) -> Result<ActualDest> {
/// Implemented according to the specification at <https://matrix.org/docs/spec/server_server/r0.1.4#resolving-server-names>
/// Numbers in comments below refer to bullet points in linked section of
/// specification
pub(crate) async fn resolve_actual_dest(
dest: &'_ ServerName, no_cache_dest: bool, admin_room_caller: bool,
) -> Result<(FedDest, String)> {
pub async fn resolve_actual_dest(dest: &ServerName, no_cache_dest: bool) -> Result<(FedDest, String)> {
trace!("Finding actual destination for {dest}");
let dest_str = dest.as_str().to_owned();
let mut hostname = dest_str.clone();
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"Checking for 1: IP literal with provided or default port",
))
.await;
}
#[allow(clippy::single_match_else)]
let actual_dest = match get_ip_with_port(&dest_str) {
Some(host_port) => {
debug!("1: IP literal with provided or default port");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"1: IP literal with provided or default port\n\nHost and Port: {host_port:?}"
)))
.await;
}
host_port
},
None => {
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"Checking for 2: Hostname with included port",
))
.await;
}
if let Some(pos) = dest_str.find(':') {
debug!("2: Hostname with included port");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain("2: Hostname with included port"))
.await;
}
let (host, port) = dest_str.split_at(pos);
if !no_cache_dest {
query_and_cache_override(host, host, port.parse::<u16>().unwrap_or(8448)).await?;
}
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!("Host: {host} | Port: {port}")))
.await;
}
FedDest::Named(host.to_owned(), port.to_owned())
} else {
trace!("Requesting well known for {dest}");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"Checking for 3: A .well-known file is available. Requesting well-known for {dest}"
)))
.await;
}
if let Some(delegated_hostname) = request_well_known(dest.as_str()).await? {
debug!("3: A .well-known file is available");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain("3: A .well-known file is available"))
.await;
}
hostname = add_port_to_hostname(&delegated_hostname).into_uri_string();
match get_ip_with_port(&delegated_hostname) {
Some(host_and_port) => {
debug!("3.1: IP literal in .well-known file");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"3.1: IP literal in .well-known file\n\nHost and Port: {host_and_port:?}"
)))
.await;
}
host_and_port
},
None => {
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"Checking for 3.2: Hostname with port in .well-known file",
))
.await;
}
if let Some(pos) = delegated_hostname.find(':') {
debug!("3.2: Hostname with port in .well-known file");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"3.2: Hostname with port in .well-known file",
))
.await;
}
let (host, port) = delegated_hostname.split_at(pos);
if !no_cache_dest {
query_and_cache_override(host, host, port.parse::<u16>().unwrap_or(8448)).await?;
}
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"Host: {host} | Port: {port}"
)))
.await;
}
FedDest::Named(host.to_owned(), port.to_owned())
} else {
trace!("Delegated hostname has no port in this branch");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"Delegated hostname has no port specified",
))
.await;
}
if let Some(hostname_override) = query_srv_record(&delegated_hostname).await? {
debug!("3.3: SRV lookup successful");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"3.3: SRV lookup successful",
))
.await;
}
let force_port = hostname_override.port();
if !no_cache_dest {
query_and_cache_override(
@ -378,53 +264,17 @@ pub(crate) async fn resolve_actual_dest(
}
if let Some(port) = force_port {
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"Host: {delegated_hostname} | Port: {port}"
)))
.await;
}
FedDest::Named(delegated_hostname, format!(":{port}"))
} else {
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"Host: {delegated_hostname} | Port: 8448"
)))
.await;
}
add_port_to_hostname(&delegated_hostname)
}
} else {
debug!("3.4: No SRV records, just use the hostname from .well-known");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"3.4: No SRV records, just use the hostname from .well-known",
))
.await;
}
if !no_cache_dest {
query_and_cache_override(&delegated_hostname, &delegated_hostname, 8448)
.await?;
}
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"Host: {delegated_hostname} | Port: 8448"
)))
.await;
}
add_port_to_hostname(&delegated_hostname)
}
}
@ -432,26 +282,8 @@ pub(crate) async fn resolve_actual_dest(
}
} else {
trace!("4: No .well-known or an error occured");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"4: No .well-known or an error occured",
))
.await;
}
if let Some(hostname_override) = query_srv_record(&dest_str).await? {
debug!("4: No .well-known; SRV record found");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"4: No .well-known; SRV record found",
))
.await;
}
let force_port = hostname_override.port();
if !no_cache_dest {
@ -464,52 +296,16 @@ pub(crate) async fn resolve_actual_dest(
}
if let Some(port) = force_port {
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"Host: {hostname} | Port: {port}"
)))
.await;
}
FedDest::Named(hostname.clone(), format!(":{port}"))
} else {
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"Host: {hostname} | Port: 8448"
)))
.await;
}
add_port_to_hostname(&hostname)
}
} else {
debug!("4: No .well-known; 5: No SRV record found");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(
"4: No .well-known; 5: No SRV record found",
))
.await;
}
if !no_cache_dest {
query_and_cache_override(&dest_str, &dest_str, 8448).await?;
}
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"Host: {dest_str} | Port: 8448"
)))
.await;
}
add_port_to_hostname(&dest_str)
}
}
@ -531,14 +327,6 @@ pub(crate) async fn resolve_actual_dest(
};
debug!("Actual destination: {actual_dest:?} hostname: {hostname:?}");
if admin_room_caller {
services()
.admin
.send_message(RoomMessageEventContent::notice_plain(format!(
"Actual destination: {actual_dest:?} | Hostname: {hostname:?}"
)))
.await;
}
Ok((actual_dest, hostname.into_uri_string()))
}

38
src/service/sending/sender.rs
View file

@ -23,12 +23,7 @@ use ruma::{
use tracing::{debug, error, warn};
use super::{appservice, send, Destination, Msg, SendingEvent, Service};
use crate::{
service::presence::Presence,
services,
utils::{calculate_hash, user_id::user_is_local},
Error, PduEvent, Result,
};
use crate::{service::presence::Presence, services, user_is_local, utils::calculate_hash, Error, PduEvent, Result};
#[derive(Debug)]
enum TransactionStatus {
@ -47,25 +42,31 @@ const DEQUEUE_LIMIT: usize = 48;
const SELECT_EDU_LIMIT: usize = 16;
impl Service {
pub(crate) fn start_handler(self: &Arc<Self>) {
let self2 = Arc::clone(self);
tokio::spawn(async move {
self2.handler().await;
pub async fn start_handler(self: &Arc<Self>) {
let self_ = Arc::clone(self);
let handle = services().server.runtime().spawn(async move {
self_
.handler()
.await
.expect("Failed to start sending handler");
});
_ = self.handler_join.lock().await.insert(handle);
}
#[tracing::instrument(skip_all, name = "sender")]
async fn handler(&self) {
async fn handler(&self) -> Result<()> {
let receiver = self.receiver.lock().await;
debug_assert!(!receiver.is_closed(), "channel error");
let mut futures: SendingFutures<'_> = FuturesUnordered::new();
let mut statuses: CurTransactionStatus = CurTransactionStatus::new();
self.initial_transactions(&mut futures, &mut statuses);
loop {
debug_assert!(!receiver.is_closed(), "channel error");
tokio::select! {
Ok(request) = receiver.recv_async() => {
self.handle_request(request, &mut futures, &mut statuses);
request = receiver.recv_async() => match request {
Ok(request) => self.handle_request(request, &mut futures, &mut statuses),
Err(_) => return Ok(()),
},
Some(response) = futures.next() => {
self.handle_response(response, &mut futures, &mut statuses);
@ -396,7 +397,7 @@ fn select_edus_receipts(
}
async fn send_events(dest: Destination, events: Vec<SendingEvent>) -> SendingResult {
debug_assert!(!events.is_empty(), "sending empty transaction");
//debug_assert!(!events.is_empty(), "sending empty transaction");
match dest {
Destination::Normal(ref server) => send_events_dest_normal(&dest, server, events).await,
Destination::Appservice(ref id) => send_events_dest_appservice(&dest, id, events).await,
@ -433,7 +434,7 @@ async fn send_events_dest_appservice(dest: &Destination, id: &String, events: Ve
}
}
debug_assert!(!pdu_jsons.is_empty(), "sending empty transaction");
//debug_assert!(!pdu_jsons.is_empty(), "sending empty transaction");
match appservice::send_request(
services()
.appservice
@ -584,7 +585,8 @@ async fn send_events_dest_normal(
}
let client = &services().globals.client.sender;
debug_assert!(pdu_jsons.len() + edu_jsons.len() > 0, "sending empty transaction");
//debug_assert!(pdu_jsons.len() + edu_jsons.len() > 0, "sending empty
// transaction");
send::send(
client,
server_name,