mirror of
https://forgejo.ellis.link/continuwuation/continuwuity.git
synced 2025-09-11 23:13:02 +02:00
Compare commits
17 commits
1b4890fd26
...
dfc33bf1b9
Author | SHA1 | Date | |
---|---|---|---|
|
dfc33bf1b9 | ||
|
5e1c396ab2 | ||
|
cae1f0244e | ||
|
1374126853 | ||
|
a29ab94e03 | ||
|
4df3794894 | ||
|
cc8880f65f | ||
|
1f18a1f537 | ||
|
7de49b2961 | ||
|
ed28159800 | ||
|
5c32436e8a | ||
|
0040331c96 | ||
|
819e9696ab | ||
|
62928623db | ||
|
cf64def999 | ||
|
7806a694a0 | ||
|
2ec33db6c3 |
2 changed files with 7 additions and 29 deletions
|
@ -3,7 +3,7 @@ use std::{fmt::Debug, mem};
|
|||
use bytes::Bytes;
|
||||
use conduwuit::{
|
||||
Err, Error, Result, debug, debug::INFO_SPAN_LEVEL, debug_error, debug_warn, err,
|
||||
error::inspect_debug_log, implement, trace, utils::string::EMPTY, warn,
|
||||
error::inspect_debug_log, implement, trace, utils::string::EMPTY,
|
||||
};
|
||||
use http::{HeaderValue, header::AUTHORIZATION};
|
||||
use ipaddress::IPAddress;
|
||||
|
@ -193,7 +193,7 @@ fn handle_error(
|
|||
) -> Result {
|
||||
if e.is_timeout() || e.is_connect() {
|
||||
e = e.without_url();
|
||||
debug_warn!(?url, "network error while sending request: {e:?}");
|
||||
debug_warn!("{e:?}");
|
||||
} else if e.is_redirect() {
|
||||
debug_error!(
|
||||
method = ?method,
|
||||
|
@ -204,7 +204,7 @@ fn handle_error(
|
|||
e,
|
||||
);
|
||||
} else {
|
||||
warn!(?url, "failed to send federation request: {e:?}");
|
||||
debug_error!("{e:?}");
|
||||
}
|
||||
|
||||
Err(e.into())
|
||||
|
|
|
@ -10,7 +10,7 @@ use std::{
|
|||
|
||||
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
|
||||
use conduwuit_core::{
|
||||
Error, Event, Result, debug, err, error, info,
|
||||
Error, Event, Result, debug, err, error,
|
||||
result::LogErr,
|
||||
trace,
|
||||
utils::{
|
||||
|
@ -142,7 +142,7 @@ impl Service {
|
|||
}
|
||||
|
||||
fn handle_response_err(dest: Destination, statuses: &mut CurTransactionStatus, e: &Error) {
|
||||
debug!(dest = ?dest, "error response: {e:?}");
|
||||
debug!(dest = ?dest, "{e:?}");
|
||||
statuses.entry(dest).and_modify(|e| {
|
||||
*e = match e {
|
||||
| TransactionStatus::Running => TransactionStatus::Failed(1, Instant::now()),
|
||||
|
@ -177,21 +177,7 @@ impl Service {
|
|||
if !new_events.is_empty() {
|
||||
self.db.mark_as_active(new_events.iter());
|
||||
|
||||
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
|
||||
info!(
|
||||
?dest,
|
||||
"Catching up previously failed destination with {}+ new events",
|
||||
new_events_vec.len()
|
||||
);
|
||||
statuses.insert(dest.clone(), TransactionStatus::Running);
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
|
@ -873,20 +859,12 @@ impl Service {
|
|||
pdus,
|
||||
edus,
|
||||
};
|
||||
let pdu_count = request.pdus.len();
|
||||
let edu_count = request.edus.len();
|
||||
|
||||
let result = self
|
||||
.services
|
||||
.federation
|
||||
.execute_on(&self.services.client.sender, &server, request)
|
||||
.await
|
||||
.inspect(|_| {
|
||||
info!(%txn_id, %server, "Sent {} PDUs, {} EDUs", pdu_count, edu_count);
|
||||
})
|
||||
.inspect_err(|e| {
|
||||
info!(%txn_id, %server, "Failed to send transaction ({} PDUs, {} EDUs): {e:?}", pdu_count, edu_count);
|
||||
});
|
||||
.await;
|
||||
|
||||
for (event_id, result) in result.iter().flat_map(|resp| resp.pdus.iter()) {
|
||||
if let Err(e) = result {
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue