mirror of
https://github.com/element-hq/synapse.git
synced 2026-11-04 09:07:22 +00:00
Forget abandoned virtual-clock sleeps, drop TransportError::IpBlocked
A dropped Clock::sleep left its waker in VIRTUAL_SLEEPERS until the virtual time passed its deadline, which a later test case's reactor may never do. Prune closed wakers when sleeping and when the time moves. Nothing constructs IpBlocked: a blocked address is a failed DNS lookup, as in Python. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb
This commit is contained in:
1 parent
9d4d050a0b
commit
db4febd65a
2 files changed
+15
-6
No files matched your search
@@ -83,8 +83,12 @@ pub fn set_virtual_time_msec(millis: u64) {
|
||||
|
||||
let due = {
|
||||
let mut sleepers = VIRTUAL_SLEEPERS.lock().expect("poisoned");
|
||||
// Drop the sleeps nobody waits on any more. A later test case's
|
||||
// reactor may start before their deadline, so they would otherwise
|
||||
// linger.
|
||||
let (due, pending) = sleepers
|
||||
.drain(..)
|
||||
.filter(|(_, waker)| !waker.is_closed())
|
||||
.partition::<Vec<_>, _>(|(wake_at, _)| *wake_at <= millis);
|
||||
*sleepers = pending;
|
||||
due
|
||||
@@ -165,6 +169,9 @@ impl Clock {
|
||||
if wake_at <= now {
|
||||
let _ = tx.send(());
|
||||
} else {
|
||||
// Also drop abandoned sleeps here, so they cannot
|
||||
// pile up while the virtual time stands still.
|
||||
sleepers.retain(|(_, waker)| !waker.is_closed());
|
||||
sleepers.push((wake_at, tx));
|
||||
}
|
||||
Some(rx)
|
||||
@@ -243,6 +250,11 @@ mod tests {
|
||||
set_virtual_time_msec(500);
|
||||
assert!(sleep.as_mut().poll(&mut cx).is_ready());
|
||||
|
||||
// A dropped sleep is forgotten.
|
||||
drop(clock.sleep(Duration::from_millis(1_000)));
|
||||
set_virtual_time_msec(500);
|
||||
assert!(VIRTUAL_SLEEPERS.lock().unwrap().is_empty());
|
||||
|
||||
// A zero-length sleep is ready at once.
|
||||
let mut sleep = std::pin::pin!(clock.sleep(Duration::ZERO));
|
||||
assert!(sleep.as_mut().poll(&mut cx).is_ready());
|
||||
|
||||
@@ -53,18 +53,16 @@ pub trait FederationTransport: Send + Sync + 'static {
|
||||
/// Why a request got no response.
|
||||
///
|
||||
/// The request layer needs to tell these apart: a DNS failure is not
|
||||
/// retried when the caller asked for `retry_on_dns_fail=False`, and a
|
||||
/// blocked IP address is reported as such.
|
||||
/// retried when the caller asked for `retry_on_dns_fail=False`.
|
||||
#[derive(Debug)]
|
||||
pub enum TransportError {
|
||||
/// Looking up the server's addresses (SRV or A/AAAA) failed, or found
|
||||
/// nothing we may connect to.
|
||||
/// nothing we may connect to. An address the IP range blocklist covers
|
||||
/// counts as not found, as in Python.
|
||||
Dns(anyhow::Error),
|
||||
/// The server publishes a single SRV record with target `.`, which means
|
||||
/// it offers no federation service.
|
||||
ServiceUnavailable(String),
|
||||
/// The destination is an IP address that the IP range blocklist covers.
|
||||
IpBlocked(std::net::IpAddr),
|
||||
/// Every TCP connection attempt failed, or the proxy refused the tunnel.
|
||||
Connect(anyhow::Error),
|
||||
/// The TLS handshake or certificate verification failed.
|
||||
@@ -92,7 +90,6 @@ impl fmt::Display for TransportError {
|
||||
match self {
|
||||
TransportError::Dns(e) => write!(f, "DNS lookup failed: {e:#}"),
|
||||
TransportError::ServiceUnavailable(name) => write!(f, "service {name} unavailable"),
|
||||
TransportError::IpBlocked(ip) => write!(f, "IP address {ip} blocked"),
|
||||
TransportError::Connect(e) => write!(f, "connection failed: {e:#}"),
|
||||
TransportError::Tls(e) => write!(f, "TLS failed: {e:#}"),
|
||||
TransportError::InvalidRequest(e) => write!(f, "invalid request: {e:#}"),
|
||||
|
||||
Reference in new issue
Block a user