feat: in-flight query coalescing with COALESCED path (#20)

* feat: in-flight query coalescing for recursive resolver

When multiple queries for the same (domain, qtype) arrive concurrently
and all miss the cache, only the first triggers recursive resolution.
Subsequent queries wait on a broadcast channel for the result.

Prevents thundering herd where N concurrent cache misses each
independently walk the full NS chain, compounding timeouts.

Uses InflightGuard (Drop impl) to guarantee map cleanup on
panic/cancellation — prevents permanent SERVFAIL poisoning.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* style: add InflightMap type alias for clippy

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat: add COALESCED query path and coalescing tests

Followers in the inflight coalescing path now log as COALESCED instead
of RECURSIVE, making it visible in the dashboard when queries were
deduplicated vs independently resolved. Adds 10 tests covering
InflightGuard cleanup, broadcast mechanics, and concurrent handle_query
coalescing through a mock TCP DNS server.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* style: cargo fmt

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: extract acquire_inflight, rewrite tests against real code

Move Disposition enum and inflight acquisition logic into a standalone
acquire_inflight() function. Rewrite 4 tests that were exercising tokio
primitives to call the real coalescing code path instead.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

src/srtt.rs

@@ -108,6 +108,13 @@ impl SrttCache {
         self.entries.is_empty()
     }
 
+    #[cfg(test)]
+    fn set_updated_at(&mut self, ip: IpAddr, at: Instant) {
+        if let Some(entry) = self.entries.get_mut(&ip) {
+            entry.updated_at = at;
+        }
+    }
+
     fn maybe_evict(&mut self) {
         if self.entries.len() < MAX_ENTRIES {
             return;
@@ -203,6 +210,99 @@ mod tests {
         assert_eq!(addrs, original);
     }
 
+    fn age(secs: u64) -> Instant {
+        Instant::now() - std::time::Duration::from_secs(secs)
+    }
+
+    /// Cache with ip(1) saturated at FAILURE_PENALTY_MS
+    fn saturated_penalty_cache() -> SrttCache {
+        let mut cache = SrttCache::new(true);
+        for _ in 0..30 {
+            cache.record_rtt(ip(1), FAILURE_PENALTY_MS, false);
+        }
+        cache
+    }
+
+    #[test]
+    fn no_decay_within_threshold() {
+        let mut cache = SrttCache::new(true);
+        cache.record_rtt(ip(1), 5000, false);
+        cache.set_updated_at(ip(1), age(DECAY_AFTER_SECS));
+        assert_eq!(cache.get(ip(1)), cache.entries[&ip(1)].srtt_ms);
+    }
+
+    #[test]
+    fn one_decay_period() {
+        let mut cache = saturated_penalty_cache();
+        let raw = cache.entries[&ip(1)].srtt_ms;
+        cache.set_updated_at(ip(1), age(DECAY_AFTER_SECS + 1));
+        let expected = (raw + INITIAL_SRTT_MS) / 2;
+        assert_eq!(cache.get(ip(1)), expected);
+    }
+
+    #[test]
+    fn multiple_decay_periods() {
+        let mut cache = saturated_penalty_cache();
+        let raw = cache.entries[&ip(1)].srtt_ms;
+        cache.set_updated_at(ip(1), age(DECAY_AFTER_SECS * 4 + 1));
+        let mut expected = raw;
+        for _ in 0..4 {
+            expected = (expected + INITIAL_SRTT_MS) / 2;
+        }
+        assert_eq!(cache.get(ip(1)), expected);
+    }
+
+    #[test]
+    fn decay_caps_at_8_periods() {
+        // 9 periods and 100 periods should produce the same result (capped at 8)
+        let mut cache_a = saturated_penalty_cache();
+        let mut cache_b = saturated_penalty_cache();
+        cache_a.set_updated_at(ip(1), age(DECAY_AFTER_SECS * 9 + 1));
+        cache_b.set_updated_at(ip(1), age(DECAY_AFTER_SECS * 100));
+        assert_eq!(cache_a.get(ip(1)), cache_b.get(ip(1)));
+    }
+
+    #[test]
+    fn decay_converges_toward_initial() {
+        let mut cache = saturated_penalty_cache();
+        cache.set_updated_at(ip(1), age(DECAY_AFTER_SECS * 100));
+        let decayed = cache.get(ip(1));
+        let diff = decayed.abs_diff(INITIAL_SRTT_MS);
+        assert!(
+            diff < 25,
+            "expected near INITIAL_SRTT_MS, got {} (diff={})",
+            decayed,
+            diff
+        );
+    }
+
+    #[test]
+    fn record_rtt_applies_decay_before_ewma() {
+        let mut cache = saturated_penalty_cache();
+        cache.set_updated_at(ip(1), age(DECAY_AFTER_SECS * 8));
+        cache.record_rtt(ip(1), 50, false);
+        let srtt = cache.get(ip(1));
+        // Without decay-before-EWMA, result would be ~(5000*7+50)/8 ≈ 4381
+        assert!(srtt < 500, "expected decay before EWMA, got srtt={}", srtt);
+    }
+
+    #[test]
+    fn decay_reranks_stale_failures() {
+        let mut cache = saturated_penalty_cache();
+        for _ in 0..30 {
+            cache.record_rtt(ip(2), 300, false);
+        }
+        let mut addrs = vec![sock(1), sock(2)];
+        cache.sort_by_rtt(&mut addrs);
+        assert_eq!(addrs, vec![sock(2), sock(1)]);
+
+        // Age server 1 so it decays toward INITIAL (200ms) — below server 2's 300ms
+        cache.set_updated_at(ip(1), age(DECAY_AFTER_SECS * 100));
+        let mut addrs = vec![sock(1), sock(2)];
+        cache.sort_by_rtt(&mut addrs);
+        assert_eq!(addrs, vec![sock(1), sock(2)]);
+    }
+
     #[test]
     fn eviction_removes_oldest() {
         let mut cache = SrttCache::new(true);