"""Two failure detectors over the same heartbeat stream. A failure detector answers one question — "is that node dead?" — from one kind of evidence: heartbeats that stopped arriving. It cannot ever be sure. A silent node and a slow network look identical from here, so every detector is really choosing a point on a trade-off between speed (declare death fast) and accuracy (don't declare death on a node that is merely late). Two ways to make that choice: FixedDetector suspects after a constant silence. One number, set once. PhiDetector models the inter-arrival distribution it has been seeing and suspects when the current silence is improbable under that model. Hayashibara's phi-accrual detector, as shipped in Akka and Cassandra. Nothing here sleeps or uses a real clock. Arrival times are computed up front by `heartbeat_trace` and every detector is a pure function of the arrivals it has been fed plus the `now` you ask about. That is what makes the demo byte-identical on every run, and it is what makes it possible to ask "what would this detector have said at *exactly* t=340.0ms" instead of guessing. """ from __future__ import annotations import math import random import statistics from collections import deque HEARTBEAT_INTERVAL_MS = 100.0 class FixedDetector: """Suspects the node after `timeout_ms` of silence. No model, no memory.""" def __init__(self, timeout_ms: float): self.timeout_ms = timeout_ms self.last_arrival = 0.0 def heartbeat(self, arrival_ms: float) -> None: self.last_arrival = arrival_ms def suspects(self, now_ms: float) -> bool: return now_ms - self.last_arrival > self.timeout_ms class PhiDetector: """Hayashibara's phi-accrual detector. phi(now) = -log10 P(inter-arrival > elapsed), under a normal model fitted to the last `window` inter-arrival times. phi is a *suspicion level*, not a boolean: phi=1 means "1 chance in 10 this silence is innocent", phi=8 means 1 in 10^8. Convict above `threshold`. `min_stddev_ms` is a floor under the fitted sigma, and it is the most load-bearing line in this file. A detector that has been watching a calm network fits a very small sigma, which makes an ordinary hiccup a many-sigma event and convicts a healthy node. The floor is how the real implementations blunt that; see the commentary. The window is primed with `bootstrap` copies of the nominal interval so the detector has an opinion before it has seen any traffic — the same thing Akka does on join. """ def __init__( self, threshold: float = 8.0, window: int = 100, min_stddev_ms: float = 1.0, interval_ms: float = HEARTBEAT_INTERVAL_MS, bootstrap: int = 10, ): self.threshold = threshold self.min_stddev_ms = min_stddev_ms self.intervals: deque[float] = deque([interval_ms] * bootstrap, maxlen=window) self.last_arrival: float | None = None def heartbeat(self, arrival_ms: float) -> None: # The first heartbeat establishes a reference point and nothing else: # there is no *interval* until two have arrived. Recording one here # would put the node's join time into the window as if it were a # network delay. if self.last_arrival is not None: self.intervals.append(arrival_ms - self.last_arrival) self.last_arrival = arrival_ms def phi(self, now_ms: float) -> float: if self.last_arrival is None: return 0.0 elapsed = now_ms - self.last_arrival mu = statistics.fmean(self.intervals) sigma = max(statistics.pstdev(self.intervals), self.min_stddev_ms) # P(X > elapsed) for X ~ Normal(mu, sigma), via the complementary # error function. Clamped because the tail underflows to 0.0 long # before the silence gets interesting. p_later = 0.5 * math.erfc((elapsed - mu) / (sigma * math.sqrt(2.0))) return -math.log10(max(p_later, 1e-300)) def suspects(self, now_ms: float) -> bool: return self.phi(now_ms) > self.threshold def heartbeat_trace( seed: int, phases: list[tuple[int, float, float]], interval_ms: float = HEARTBEAT_INTERVAL_MS, ) -> tuple[list[float], float]: """Build one heartbeat stream, then kill the sender. `phases` is a list of (count, delay_mean_ms, delay_stddev_ms). The sender emits a heartbeat every `interval_ms` on the dot; the network adds a per-phase random delay. Arrivals are forced monotonic — this toy does not model reordering, and a detector that saw heartbeat 7 after heartbeat 8 would be answering a different question. Returns (arrival times, the time the first missing heartbeat was due). """ rng = random.Random(seed) arrivals: list[float] = [] send_at, last = 0.0, float("-inf") for count, mean, stddev in phases: for _ in range(count): delay = max(0.0, rng.gauss(mean, stddev)) arrival = max(send_at + delay, last + 0.001) arrivals.append(arrival) last = arrival send_at += interval_ms return arrivals, send_at def time_to_suspect(detector, after_ms: float, horizon_ms: float = 100_000.0) -> float: """First instant at or after `after_ms` when the detector suspects. Bisection is legitimate here only because suspicion is monotonic in `now` for both detectors: silence never becomes *less* alarming while it continues. That is true of the fixed detector by inspection and of phi because the normal tail is monotonically decreasing. """ lo, hi = after_ms, after_ms + horizon_ms for _ in range(200): mid = (lo + hi) / 2.0 if detector.suspects(mid): hi = mid else: lo = mid return hi def evaluate(detector, arrivals: list[float], crash_due_ms: float) -> tuple[int, float]: """Replay one trace against one detector. Returns (false alarms, detection latency in ms). A false alarm is a gap between two real heartbeats during which the detector convicted a node that was alive the whole time. Suspicion is monotonic in elapsed time, so the most alarming instant in a gap is the moment just before the next heartbeat lands: checking there is exact, not a sample. Detection latency is measured from when the *missing* heartbeat was due, not from the last one that arrived — the node is dead from the moment it fails to send, and a detector should not get credit for the network delay it already absorbed. """ false_alarms = 0 detector.heartbeat(arrivals[0]) for arrival in arrivals[1:]: if detector.suspects(arrival): false_alarms += 1 detector.heartbeat(arrival) detected_at = time_to_suspect(detector, arrivals[-1]) return false_alarms, detected_at - crash_due_ms