mirror of
https://forgejo.ellis.link/continuwuation/continuwuity.git
synced 2026-11-14 20:56:49 +00:00
fix: Address review comments
This commit is contained in:
1 parent
9af60bcf62
commit
f1a77e0404
9 files changed
+45
-57
No files matched your search
@@ -256,8 +256,15 @@ impl super::Service {
|
||||
parts,
|
||||
body.as_ref(),
|
||||
))
|
||||
.inspect(|_| self.mark_healthy(dest))
|
||||
.map_err(|e| err!(BadServerResponse("Server returned bad 200 response: {e:?}")))
|
||||
.inspect(|_| {
|
||||
self.mark_healthy(dest);
|
||||
})
|
||||
.map_err(|e| {
|
||||
// Bad 200 response is usually a sign of bad routing, so we'll mark as stale.
|
||||
// We won't mark as offline though since we technically got a response.
|
||||
self.mark_destination_stale(dest);
|
||||
err!(BadServerResponse("Server returned bad 200 response: {e:?}"))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -8,10 +8,7 @@ use std::{
|
||||
|
||||
use assign::assign;
|
||||
use async_trait::async_trait;
|
||||
use conduwuit::{
|
||||
Error, Result, Server, SyncRwLock, debug,
|
||||
utils::{millis_since_unix_epoch, time::exponential_backoff::next_interval},
|
||||
};
|
||||
use conduwuit::{Error, Result, Server, SyncRwLock, debug, utils::millis_since_unix_epoch};
|
||||
pub(crate) use execute::FederationPathBuilderInput;
|
||||
use http::StatusCode;
|
||||
use ruma::{
|
||||
@@ -83,7 +80,9 @@ impl Service {
|
||||
|
||||
/// Marks or updates a remote's health status as unhealthy. If the remote is
|
||||
/// not already marked as unhealthy, a new entry is created. Otherwise, the
|
||||
/// retry count is incremented and
|
||||
/// retry count is incremented and next retry window is updated.
|
||||
///
|
||||
/// Does not update the marker if the backoff period is already in effect.
|
||||
pub fn hit_unhealthy(&self, server_name: OwnedServerName) {
|
||||
let unix_now = millis_since_unix_epoch();
|
||||
let mut map = self.remote_health.write();
|
||||
@@ -101,8 +100,9 @@ impl Service {
|
||||
let max = Duration::from_secs(self.services.server.config.sender_retry_backoff_limit);
|
||||
|
||||
*retries = retries.saturating_add(1);
|
||||
let next_interval = min.saturating_mul(*retries).min(max);
|
||||
*next_retry = unix_now.saturating_add(
|
||||
u64::try_from(next_interval(min, max, *retries).as_millis())
|
||||
u64::try_from(next_interval.as_millis())
|
||||
.expect("backoff milliseconds should not exceed u64::MAX"),
|
||||
);
|
||||
debug!(
|
||||
@@ -113,12 +113,12 @@ impl Service {
|
||||
|
||||
/// Marks a server as "healthy" by removing it from the health map.
|
||||
///
|
||||
/// TODO: flush senders too
|
||||
pub fn mark_healthy(&self, server_name: &ServerName) {
|
||||
// TODO: We need to make sure the sender flush DOESN'T trigger if this is called
|
||||
// by the senders themselves.
|
||||
/// Returns true if the server was previously marked as unhealthy, false
|
||||
/// otherwise.
|
||||
pub fn mark_healthy(&self, server_name: &ServerName) -> bool {
|
||||
let mut health_map = self.remote_health.write();
|
||||
if health_map.remove(server_name).is_some() {
|
||||
let was_unhealthy = health_map.remove(server_name).is_some();
|
||||
if was_unhealthy {
|
||||
debug!("{} is now healthy", server_name);
|
||||
}
|
||||
// The lock for remote_health is deliberately retained until the end of the
|
||||
@@ -136,6 +136,8 @@ impl Service {
|
||||
.matrix_resolver
|
||||
.remove_cache_entry(server_name.as_str());
|
||||
}
|
||||
|
||||
was_unhealthy
|
||||
}
|
||||
|
||||
/// Returns a rate-limited error if the remote is unhealthy.
|
||||
|
||||
@@ -9,10 +9,7 @@ use std::{
|
||||
};
|
||||
|
||||
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
|
||||
use conduwuit::{
|
||||
debug_info, debug_warn, info, trace,
|
||||
utils::{should_continue_backoff, time::exponential_backoff::next_interval},
|
||||
};
|
||||
use conduwuit::{debug_info, debug_warn, info, trace};
|
||||
use conduwuit_core::{
|
||||
Error, Event, Result, at, debug, err, error,
|
||||
matrix::pdu::sticky,
|
||||
@@ -173,11 +170,7 @@ impl Service {
|
||||
TransactionStatus::Failed(n.saturating_add(1), Instant::now()),
|
||||
| TransactionStatus::Failed(..) => {
|
||||
panic!(
|
||||
"{}",
|
||||
format!(
|
||||
"Request to {dest2:?} that was not even running ({e:?}) failed: \
|
||||
{er:?}"
|
||||
)
|
||||
"Request to {dest2:?} that was not even running ({e:?}) failed: {er:?}"
|
||||
)
|
||||
},
|
||||
}
|
||||
@@ -446,16 +439,16 @@ impl Service {
|
||||
});
|
||||
return Ok((allow, retry));
|
||||
}
|
||||
|
||||
// Legacy backoff tracking for non-server remotes
|
||||
statuses
|
||||
.entry(dest.clone())
|
||||
.and_modify(|e| match e {
|
||||
| TransactionStatus::Failed(tries, time) => {
|
||||
let min = Duration::from_secs(self.server.config.sender_retry_backoff_base);
|
||||
let max = Duration::from_secs(self.server.config.sender_retry_backoff_limit);
|
||||
if should_continue_backoff(min, max, time.elapsed(), *tries)
|
||||
&& !matches!(dest, Destination::Appservice(_))
|
||||
{
|
||||
let retry_after = next_interval(min, max, *tries);
|
||||
let retry_after = min.saturating_mul(*tries).min(max);
|
||||
if time.elapsed() < retry_after {
|
||||
debug_warn!("Not retrying destination for another {retry_after:?}");
|
||||
allow = false;
|
||||
} else {
|
||||
|
||||
Reference in new issue
Block a user