studio/tuwunel
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

chain_width to 50

Browse Source 
Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk
2025-04-22 04:42:26 +00:00
parent 9b658d86b2
commit 76509830e6
190 changed files with 3469 additions and 930 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/service/sending/appservice.rs
View File

@@ -54,14 +54,22 @@ where
.parse()
.unwrap(),
);
*http_request.uri_mut() = parts.try_into().expect("our manipulation is always valid");
*http_request.uri_mut() = parts
.try_into()
.expect("our manipulation is always valid");
let reqwest_request = reqwest::Request::try_from(http_request)?;
let mut response = client.execute(reqwest_request).await.map_err(|e| {
warn!("Could not send request to appservice \"{}\" at {dest}: {e:?}", registration.id);
e
})?;
let mut response = client
.execute(reqwest_request)
.await
.map_err(|e| {
warn!(
"Could not send request to appservice \"{}\" at {dest}: {e:?}",
registration.id
);
e
})?;
// reqwest::Response -> http::Response conversion
let status = response.status();

15
src/service/sending/data.rs
View File

@@ -170,7 +170,8 @@ impl Data {
}
pub(super) fn set_latest_educount(&self, server_name: &ServerName, last_count: u64) {
self.servername_educount.raw_put(server_name, last_count);
self.servername_educount
.raw_put(server_name, last_count);
}
pub async fn get_latest_educount(&self, server_name: &ServerName) -> u64 {
@@ -187,7 +188,9 @@ fn parse_servercurrentevent(key: &[u8], value: &[u8]) -> Result<(Destination, Se
Ok::<_, Error>(if key.starts_with(b"+") {
let mut parts = key[1..].splitn(2, |&b| b == 0xFF);
let server = parts.next().expect("splitn always returns one element");
let server = parts
.next()
.expect("splitn always returns one element");
let event = parts
.next()
.ok_or_else(|| Error::bad_database("Invalid bytes in servercurrentpdus."))?;
@@ -207,7 +210,9 @@ fn parse_servercurrentevent(key: &[u8], value: &[u8]) -> Result<(Destination, Se
} else if key.starts_with(b"$") {
let mut parts = key[1..].splitn(3, |&b| b == 0xFF);
let user = parts.next().expect("splitn always returns one element");
let user = parts
.next()
.expect("splitn always returns one element");
let user_string = utils::str_from_bytes(user)
.map_err(|_| Error::bad_database("Invalid user string in servercurrentevent"))?;
let user_id = UserId::parse(user_string)
@@ -235,7 +240,9 @@ fn parse_servercurrentevent(key: &[u8], value: &[u8]) -> Result<(Destination, Se
} else {
let mut parts = key.splitn(2, |&b| b == 0xFF);
let server = parts.next().expect("splitn always returns one element");
let server = parts
.next()
.expect("splitn always returns one element");
let event = parts
.next()
.ok_or_else(|| Error::bad_database("Invalid bytes in servercurrentpdus."))?;

5
src/service/sending/dest.rs
View File

@@ -24,7 +24,10 @@ pub(super) fn get_prefix(&self) -> Vec<u8> {
},
| Self::Appservice(server) => {
let sigil = b"+";
let len = sigil.len().saturating_add(server.len()).saturating_add(1);
let len = sigil
.len()
.saturating_add(server.len())
.saturating_add(1);
let mut p = Vec::with_capacity(len);
p.extend_from_slice(sigil);

32
src/service/sending/mod.rs
View File

@@ -100,7 +100,9 @@ impl crate::Service for Service {
pusher: args.depend::<pusher::Service>("pusher"),
federation: args.depend::<federation::Service>("federation"),
},
channels: (0..num_senders).map(|_| loole::unbounded()).collect(),
channels: (0..num_senders)
.map(|_| loole::unbounded())
.collect(),
}))
}
@@ -160,7 +162,10 @@ impl Service {
self.dispatch(Msg {
dest,
event,
queue_id: keys.into_iter().next().expect("request queue key"),
queue_id: keys
.into_iter()
.next()
.expect("request queue key"),
})
}
@@ -173,7 +178,10 @@ impl Service {
self.dispatch(Msg {
dest,
event,
queue_id: keys.into_iter().next().expect("request queue key"),
queue_id: keys
.into_iter()
.next()
.expect("request queue key"),
})
}
@@ -201,7 +209,9 @@ impl Service {
.await;
let _cork = self.db.db.cork();
let keys = self.db.queue_requests(requests.iter().map(|(o, e)| (e, o)));
let keys = self
.db
.queue_requests(requests.iter().map(|(o, e)| (e, o)));
for ((dest, event), queue_id) in requests.into_iter().zip(keys) {
self.dispatch(Msg { dest, event, queue_id })?;
@@ -219,7 +229,10 @@ impl Service {
self.dispatch(Msg {
dest,
event,
queue_id: keys.into_iter().next().expect("request queue key"),
queue_id: keys
.into_iter()
.next()
.expect("request queue key"),
})
}
@@ -250,7 +263,9 @@ impl Service {
.await;
let _cork = self.db.db.cork();
let keys = self.db.queue_requests(requests.iter().map(|(o, e)| (e, o)));
let keys = self
.db
.queue_requests(requests.iter().map(|(o, e)| (e, o)));
for ((dest, event), queue_id) in requests.into_iter().zip(keys) {
self.dispatch(Msg { dest, event, queue_id })?;
@@ -299,7 +314,10 @@ impl Service {
where
T: OutgoingRequest + Debug + Send,
{
self.services.federation.execute(dest, request).await
self.services
.federation
.execute(dest, request)
.await
}
/// Like send_federation_request() but with a very large timeout

86
src/service/sending/sender.rs
View File

@@ -85,7 +85,8 @@ impl Service {
.boxed()
.await;
self.work_loop(id, &mut futures, &mut statuses).await;
self.work_loop(id, &mut futures, &mut statuses)
.await;
if !futures.is_empty() {
self.finish_responses(&mut futures).boxed().await;
@@ -136,7 +137,9 @@ impl Service {
statuses: &mut CurTransactionStatus,
) {
match response {
| Ok(dest) => self.handle_response_ok(&dest, futures, statuses).await,
| Ok(dest) =>
self.handle_response_ok(&dest, futures, statuses)
.await,
| Err((dest, e)) => Self::handle_response_err(dest, statuses, &e),
}
}
@@ -177,7 +180,10 @@ impl Service {
if !new_events.is_empty() {
self.db.mark_as_active(new_events.iter());
let new_events_vec = new_events.into_iter().map(|(_, event)| event).collect();
let new_events_vec = new_events
.into_iter()
.map(|(_, event)| event)
.collect();
futures.push(self.send_events(dest.clone(), new_events_vec));
} else {
statuses.remove(dest);
@@ -322,7 +328,8 @@ impl Service {
let select_edus = select_edus.into_iter().map(SendingEvent::Edu);
events.extend(select_edus);
self.db.set_latest_educount(server_name, last_count);
self.db
.set_latest_educount(server_name, last_count);
}
}
@@ -415,7 +422,10 @@ impl Service {
events_len: &AtomicUsize,
) -> EduVec {
let mut events = EduVec::new();
let server_rooms = self.services.state_cache.server_rooms(server_name);
let server_rooms = self
.services
.state_cache
.server_rooms(server_name);
pin_mut!(server_rooms);
let mut device_list_changes = HashSet::<OwnedUserId>::new();
@@ -562,7 +572,10 @@ impl Service {
event_ids: vec![event_id.clone()],
};
if read.insert(user_id.to_owned(), receipt_data).is_none() {
if read
.insert(user_id.to_owned(), receipt_data)
.is_none()
{
*num = num.saturating_add(1);
if *num >= SELECT_RECEIPT_LIMIT {
break;
@@ -621,7 +634,10 @@ impl Service {
let update = PresenceUpdate {
user_id: user_id.into(),
presence: presence_event.content.presence,
currently_active: presence_event.content.currently_active.unwrap_or(false),
currently_active: presence_event
.content
.currently_active
.unwrap_or(false),
status_msg: presence_event.content.status_msg,
last_active_ago: presence_event
.content
@@ -653,11 +669,15 @@ impl Service {
fn send_events(&self, dest: Destination, events: Vec<SendingEvent>) -> SendingFuture<'_> {
debug_assert!(!events.is_empty(), "sending empty transaction");
match dest {
| Destination::Federation(server) =>
self.send_events_dest_federation(server, events).boxed(),
| Destination::Appservice(id) => self.send_events_dest_appservice(id, events).boxed(),
| Destination::Push(user_id, pushkey) =>
self.send_events_dest_push(user_id, pushkey, events).boxed(),
| Destination::Federation(server) => self
.send_events_dest_federation(server, events)
.boxed(),
| Destination::Appservice(id) => self
.send_events_dest_appservice(id, events)
.boxed(),
| Destination::Push(user_id, pushkey) => self
.send_events_dest_push(user_id, pushkey, events)
.boxed(),
}
}
@@ -674,7 +694,12 @@ impl Service {
id: String,
events: Vec<SendingEvent>,
) -> SendingResult {
let Some(appservice) = self.services.appservice.get_registration(&id).await else {
let Some(appservice) = self
.services
.appservice
.get_registration(&id)
.await
else {
return Err((
Destination::Appservice(id.clone()),
err!(Database(warn!(?id, "Missing appservice registration"))),
@@ -696,7 +721,12 @@ impl Service {
for event in &events {
match event {
| SendingEvent::Pdu(pdu_id) => {
if let Ok(pdu) = self.services.timeline.get_pdu_from_id(pdu_id).await {
if let Ok(pdu) = self
.services
.timeline
.get_pdu_from_id(pdu_id)
.await
{
pdu_jsons.push(pdu.into_room_event());
}
},
@@ -752,7 +782,12 @@ impl Service {
pushkey: String,
events: Vec<SendingEvent>,
) -> SendingResult {
let Ok(pusher) = self.services.pusher.get_pusher(&user_id, &pushkey).await else {
let Ok(pusher) = self
.services
.pusher
.get_pusher(&user_id, &pushkey)
.await
else {
return Err((
Destination::Push(user_id.clone(), pushkey.clone()),
err!(Database(error!(?user_id, ?pushkey, "Missing pusher"))),
@@ -768,7 +803,12 @@ impl Service {
for event in &events {
match event {
| SendingEvent::Pdu(pdu_id) => {
if let Ok(pdu) = self.services.timeline.get_pdu_from_id(pdu_id).await {
if let Ok(pdu) = self
.services
.timeline
.get_pdu_from_id(pdu_id)
.await
{
pdus.push(pdu);
}
},
@@ -826,7 +866,12 @@ impl Service {
| _ => None,
})
.stream()
.wide_filter_map(|pdu_id| self.services.timeline.get_pdu_json_from_id(pdu_id).ok())
.wide_filter_map(|pdu_id| {
self.services
.timeline
.get_pdu_json_from_id(pdu_id)
.ok()
})
.wide_then(|pdu| self.convert_to_outgoing_federation_event(pdu))
.collect()
.await;
@@ -898,7 +943,12 @@ impl Service {
.get("room_id")
.and_then(|val| RoomId::parse(val.as_str()?).ok())
{
match self.services.state.get_room_version(room_id).await {
match self
.services
.state
.get_room_version(room_id)
.await
{
| Ok(room_version_id) => match room_version_id {
| RoomVersionId::V1 | RoomVersionId::V2 => {},
| _ => _ = pdu_json.remove("event_id"),