1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
//! Command-wide health of each provider (ADR 039 ยง Failure handling:
//! reassign-only tail).
//!
//! One `PeerHealth` spans a whole command, so a bundle pull cools a flaky
//! provider once for every entry. A source is never removed: a delivery fault
//! cools it for a while that doubles with each consecutive fault, and a
//! refusal on price parks it until the pool deposit rises.
use std::collections::HashMap;
use std::sync::{Mutex, PoisonError};
use alloy::primitives::{Address, U256};
use tokio::time::{Duration, Instant};
use crate::fault::Fault;
/// The first cooldown after a delivery fault.
pub(crate) const COOL_BASE: Duration = Duration::from_secs(2);
/// The longest cooldown, reached after consecutive faults.
pub(crate) const COOL_CAP: Duration = Duration::from_mins(1);
/// One provider's health.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[doc(hidden)]
pub enum Health {
/// Usable. `streak` counts delivery faults since its last verified byte.
Healthy {
/// Consecutive delivery faults without progress.
streak: u32,
},
/// Not usable before `until`.
Cooling {
/// When the source is usable again.
until: Instant,
/// Consecutive delivery faults without progress.
streak: u32,
},
/// Not usable until the pool deposit exceeds `at_deposit`.
Unaffordable {
/// The deposit the source refused at.
at_deposit: U256,
},
}
impl Health {
const fn streak(self) -> u32 {
match self {
Self::Healthy { streak } | Self::Cooling { streak, .. } => streak,
Self::Unaffordable { .. } => 0,
}
}
}
/// Every provider's [`Health`] for one command.
#[derive(Debug, Default)]
pub struct PeerHealth {
map: Mutex<HashMap<Address, Health>>,
}
impl PeerHealth {
/// `provider`'s health. An unseen provider is healthy.
#[must_use]
#[doc(hidden)]
pub fn health(&self, provider: Address) -> Health {
self.lock()
.get(&provider)
.copied()
.unwrap_or(Health::Healthy { streak: 0 })
}
/// Whether `provider` may take work at `now` against `deposit`.
#[must_use]
#[doc(hidden)]
pub fn usable(&self, provider: Address, now: Instant, deposit: U256) -> bool {
match self.health(provider) {
Health::Healthy { .. } => true,
Health::Cooling { until, .. } => now >= until,
Health::Unaffordable { at_deposit } => deposit > at_deposit,
}
}
/// Record a fault `provider` raised at `now` while the deposit was `deposit`.
#[doc(hidden)]
pub fn record(&self, provider: Address, fault: Fault, now: Instant, deposit: U256) {
let mut map = self.lock();
let current = map
.get(&provider)
.copied()
.unwrap_or(Health::Healthy { streak: 0 });
let next = match fault {
Fault::Source => {
let streak = current.streak().saturating_add(1);
Health::Cooling {
until: now + cool_for(streak),
streak,
}
}
Fault::Unaffordable => Health::Unaffordable {
at_deposit: deposit,
},
Fault::Fatal(_) | Fault::Transient => current,
};
map.insert(provider, next);
}
/// Hold `provider` off until `until` without counting a fault: its streak
/// stays. The acquire loop holds a source that refuses the pool's funding
/// while a funding recovery step's top-up settles.
#[doc(hidden)]
pub fn hold(&self, provider: Address, until: Instant) {
let mut map = self.lock();
let streak = map.get(&provider).copied().map_or(0, Health::streak);
map.insert(provider, Health::Cooling { until, streak });
}
/// Make every source priced out at the current funding usable again: a
/// swapped credential changes what the sources refused.
#[doc(hidden)]
pub fn clear_unaffordable(&self) {
for health in self.lock().values_mut() {
if matches!(health, Health::Unaffordable { .. }) {
*health = Health::Healthy { streak: 0 };
}
}
}
/// Record a verified byte from `provider`: it is healthy with no streak.
#[doc(hidden)]
pub fn record_progress(&self, provider: Address) {
self.lock().insert(provider, Health::Healthy { streak: 0 });
}
/// When `provider` stops cooling, if it is cooling at `now`.
#[must_use]
#[doc(hidden)]
pub fn cooling_until(&self, provider: Address, now: Instant) -> Option<Instant> {
match self.health(provider) {
Health::Cooling { until, .. } if until > now => Some(until),
_ => None,
}
}
fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<Address, Health>> {
self.map.lock().unwrap_or_else(PoisonError::into_inner)
}
}
/// The cooldown after the `streak`-th consecutive fault: 2 s, doubling, capped at 60 s.
fn cool_for(streak: u32) -> Duration {
COOL_BASE
.saturating_mul(2u32.saturating_pow(streak.saturating_sub(1)))
.min(COOL_CAP)
}
#[cfg(test)]
mod tests;