git-mirrors/continuwuity
1
0
Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2025-09-10 05:02:49 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix(fed): Improve transaction flushing

Browse source
This commit is contained in:
nexy7574 2025-08-15 04:10:40 +01:00 • committed by Jacob Taylor
commit 0abfc192d4
2 changed files with 29 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/service/federation/execute.rs
View file

@ -3,7 +3,7 @@ use std::{fmt::Debug, mem};
use bytes::Bytes; use bytes::Bytes;
use conduwuit::{ use conduwuit::{
Err, Error, Result, debug, debug::INFO_SPAN_LEVEL, debug_error, debug_warn, err, Err, Error, Result, debug, debug::INFO_SPAN_LEVEL, debug_error, debug_warn, err,
error::inspect_debug_log, implement, trace, utils::string::EMPTY, error::inspect_debug_log, implement, trace, utils::string::EMPTY, warn,
}; };
use http::{HeaderValue, header::AUTHORIZATION}; use http::{HeaderValue, header::AUTHORIZATION};
use ipaddress::IPAddress; use ipaddress::IPAddress;
@ -193,7 +193,7 @@ fn handle_error(
) -> Result { ) -> Result {
if e.is_timeout() || e.is_connect() { if e.is_timeout() || e.is_connect() {
e = e.without_url(); e = e.without_url();
debug_warn!("{e:?}"); debug_warn!(?url, "network error while sending request: {e:?}");
} else if e.is_redirect() { } else if e.is_redirect() {
debug_error!( debug_error!(
method = ?method, method = ?method,
@ -204,7 +204,7 @@ fn handle_error(
e, e,
); );
} else { } else {
debug_error!("{e:?}"); warn!(?url, "failed to send federation request: {e:?}");
} }
Err(e.into()) Err(e.into())

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

@ -10,7 +10,7 @@ use std::{
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use conduwuit_core::{ use conduwuit_core::{
Error, Event, Result, debug, err, error, Error, Event, Result, debug, err, error, info,
result::LogErr, result::LogErr,
trace, trace,
utils::{ utils::{
@ -142,7 +142,7 @@ impl Service {
} }
fn handle_response_err(dest: Destination, statuses: &mut CurTransactionStatus, e: &Error) { fn handle_response_err(dest: Destination, statuses: &mut CurTransactionStatus, e: &Error) {
debug!(dest = ?dest, "{e:?}"); debug!(dest = ?dest, "error response: {e:?}");
statuses.entry(dest).and_modify(|e| { statuses.entry(dest).and_modify(|e| {
*e = match e { *e = match e {
| TransactionStatus::Running => TransactionStatus::Failed(1, Instant::now()), | TransactionStatus::Running => TransactionStatus::Failed(1, Instant::now()),
@ -177,7 +177,21 @@ impl Service {
if !new_events.is_empty() { if !new_events.is_empty() {
self.db.mark_as_active(new_events.iter()); self.db.mark_as_active(new_events.iter());
let new_events_vec = new_events.into_iter().map(|(_, event)| event).collect(); let new_events_vec: Vec<SendingEvent> =
new_events.into_iter().map(|(_, event)| event).collect();
if let Some(status) = statuses.get(&dest.clone()) {
if matches!(status, TransactionStatus::Running) {
// If the server is in backoff, clear it
warn!(
?dest,
"Catching up destination with {} new events",
new_events_vec.len()
);
statuses.insert(dest.clone(), TransactionStatus::Running);
}
}
futures.push(self.send_events(dest.clone(), new_events_vec)); futures.push(self.send_events(dest.clone(), new_events_vec));
} else { } else {
statuses.remove(dest); statuses.remove(dest);
@ -859,12 +873,20 @@ impl Service {
pdus, pdus,
edus, edus,
}; };
let pdu_count = request.pdus.len();
let edu_count = request.edus.len();
let result = self let result = self
.services .services
.federation .federation
.execute_on(&self.services.client.sender, &server, request) .execute_on(&self.services.client.sender, &server, request)
.await; .await
.inspect(|_| {
info!(%txn_id, %server, "Sent {} PDUs, {} EDUs", pdu_count, edu_count);
})
.inspect_err(|e| {
error!(%txn_id, %server, "Failed to send transaction ({} PDUs, {} EDUs): {e:?}", pdu_count, edu_count);
});
for (event_id, result) in result.iter().flat_map(|resp| resp.pdus.iter()) { for (event_id, result) in result.iter().flat_map(|resp| resp.pdus.iter()) {
if let Err(e) = result { if let Err(e) = result {