cellos_supervisor/resolver_refresh/

ticker.rs

//! SEC-21 Phase 2 — continuous-ticker daemon mode.
//!
//! W2 SEC-21 (`super::ResolverRefresh::tick`) runs *one* refresh per cell
//! run, before the workload starts. Phase 2 keeps that loop running for the
//! lifetime of the cell so drift between runtime ticks produces events the
//! operator sees in real time, not just at boot.
//!
//! ## Design notes
//!
//! - The ticker owns its own [`super::ResolverState`]. The startup one-tick
//!   path in `supervisor.rs` uses a *separate* state. As a deliberate
//!   trade-off, the **first observation per hostname inside the continuous
//!   ticker** emits a baseline drift event with `previousDigest: empty`
//!   even when the startup tick already observed the same hostname. The
//!   alternative — sharing state across the startup tick and the
//!   continuous ticker — would entangle two lifetimes that don't compose
//!   cleanly; the duplication is bounded to one event per hostname per
//!   cell run.
//!
//! - The resolver function ([`super::ResolverFn`]) is sync. Calling it
//!   directly inside the tokio task would block a runtime worker for the
//!   duration of every blocking `getaddrinfo`-class lookup, so each
//!   per-tick batch resolution is wrapped in
//!   [`tokio::task::spawn_blocking`] — mirroring the W2 SEC-21 fix in
//!   `supervisor.rs::Supervisor::maybe_run_dns_authority_refresh`. The
//!   Phase 3 production wiring routes that closure to
//!   [`super::hickory_resolve::resolve_with_ttl`] via
//!   [`tokio::runtime::Handle::block_on`] so the ticker stays sync at the
//!   call site while still benefiting from the async resolver.
//!
//! - Shutdown is coordinated via a shared [`std::sync::atomic::AtomicBool`]
//!   the supervisor flips to `true` at cell destroy. The loop checks the
//!   flag at the top of every iteration AND races `shutdown` observation
//!   against the inter-tick sleep via `tokio::select!` so a teardown
//!   request returns within milliseconds, not at the next sleep boundary.
//!
//! - **Phase 3 — real upstream TTL.** The resolver function returns a
//!   [`super::ResolvedAnswer`] carrying both the target set and the
//!   minimum TTL the upstream nameserver advertised. The ticker forwards
//!   that TTL into [`super::ResolverRefresh::tick`], which clamps it to
//!   `refreshPolicy.minTtlSeconds` (the operator's DNS-rebinding floor)
//!   before stamping it into the `dns_authority_drift` event.

use std::collections::HashMap;
use std::io;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, SystemTime};

use cellos_core::{
    CloudEventV1, DnsRebindingPolicy, DnsRefreshPolicy, DnsResolver, DnsResolverDnssecPolicy,
};

use super::{
    RebindingState, ResolvedAnswer, ResolverRefresh, ResolverState, TrustAnchors,
    ValidatedResolvedAnswer,
};

/// Sync, send-and-share-able resolver closure shared by both the ticker
/// internals and the supervisor production wiring. Type-aliased so clippy
/// `type_complexity` stays quiet at the per-callsite level.
///
/// scope: returns [`ResolvedAnswer`] (targets + ttl_seconds +
/// resolver_addr). The production closure is a thin `Handle::block_on`
/// wrapper around [`super::hickory_resolve::resolve_with_ttl`]; tests build
/// fixed answers directly.
pub type SharedResolverFn = Arc<dyn Fn(&str) -> io::Result<ResolvedAnswer> + Send + Sync>;

/// SEC-21 Phase 3h — DNSSEC-validating resolver closure shared with the
/// ticker. Wraps [`super::resolve_with_ttl_validated`] in production; the
/// ticker tests pass synthetic closures that mint deterministic
/// [`ValidatedResolvedAnswer`] outcomes (Validated / Failed / Unsigned).
pub type SharedValidatedResolverFn =
    Arc<dyn Fn(&str) -> io::Result<ValidatedResolvedAnswer> + Send + Sync>;

/// Sink-shape consumer the ticker hands events to. Mirrors the proxy's
/// [`crate::dns_proxy::DnsQueryEmitter`] — synchronous on the call site so
/// the ticker loop never blocks on emit, with the bridging to async
/// [`cellos_core::ports::EventSink`] handled by the
/// [`super::sink_emitter::EventSinkEmitter`] adapter.
pub trait DriftEmitter: Send + Sync + 'static {
    fn emit(&self, event: CloudEventV1);
}

/// Snapshot of per-cell counters returned when the ticker exits.
///
/// Returned by the spawned task as its [`tokio::task::JoinHandle`] output
/// so the supervisor can log a summary at teardown. None of these counters
/// drive control flow — they are pure observability.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct TickerStats {
    /// Number of complete tick iterations executed.
    pub tick_count: u64,
    /// Number of `dns_authority_drift` events emitted across all ticks.
    pub events_emitted: u64,
    /// Number of `Err` returns from the injected resolver across all
    /// hostname lookups in all ticks.
    pub resolver_errors: u64,
}

/// Spawned-task handle the supervisor uses to coordinate shutdown.
///
/// On cell destroy the supervisor:
///
/// 1. Flips `shutdown` to `true`.
/// 2. Awaits `task` with a bounded timeout (the supervisor's call site
///    chooses 2s — matching the W4 SEAM-1 Phase 2b shutdown pattern in
///    [`crate::dns_proxy::spawn::DnsProxyHandle::join`]).
pub struct TickerHandle {
    /// Shared shutdown flag. Setting `true` signals the loop to exit at its
    /// next iteration. The loop also races this flag against the inter-tick
    /// sleep so a teardown request collapses worst-case shutdown latency.
    pub shutdown: Arc<AtomicBool>,
    /// JoinHandle for the spawned tokio task. The task always returns
    /// [`TickerStats`]; failures inside the loop are absorbed (resolver
    /// errors increment a counter, malformed events drop, etc.) so the
    /// task does not panic out from under the supervisor.
    pub task: tokio::task::JoinHandle<TickerStats>,
}

/// Owned, owned-string configuration for one ticker — built by the
/// supervisor in async context and moved into the spawned task. All
/// borrowing-from-spec is resolved before spawn so the task's lifetime is
/// `'static`.
pub struct TickerConfig {
    /// Inter-tick interval. Resolved by the supervisor — typically
    /// `min(refreshPolicy.minTtlSeconds, 60).max(5)` — with a hard floor
    /// of 5s applied at the call site so a misconfigured spec cannot burn
    /// CPU here.
    pub interval: Duration,
    /// Refresh policy carried by the spec (`spec.authority.dnsAuthority.
    /// refreshPolicy`). `None` means "no floor / no ceiling, ttl-honor
    /// strategy", same as the startup tick.
    pub policy: Option<DnsRefreshPolicy>,
    /// SEC-21 Phase 3e — rebinding mitigation policy carried by the spec
    /// (`spec.authority.dnsAuthority.rebindingPolicy`). `None` means "no
    /// per-hostname response-IP tracking" — the ticker emits only the
    /// standard `dns_authority_drift` events. When `Some`, the ticker
    /// owns a per-cell [`RebindingState`] across ticks and emits
    /// `dns_authority_rebind_threshold` / `dns_authority_rebind_rejected`
    /// events as the operator-declared cap / allowlist is breached.
    pub rebinding_policy: Option<DnsRebindingPolicy>,
    /// Declared resolvers. The first entry's `resolverId` is stamped into
    /// every emitted event.
    pub resolvers: Vec<DnsResolver>,
    /// Hostnames the ticker may refresh — typically the
    /// `dnsAuthority.hostnameAllowlist` ∪ egress-rule hosts, resolved by
    /// the supervisor at predicate time so the ticker has nothing to
    /// re-derive.
    pub hostnames: Vec<String>,
    /// Optional `keysetId` to stamp into emitted events.
    pub keyset_id: Option<String>,
    /// Optional `issuerKid` to stamp into emitted events.
    pub issuer_kid: Option<String>,
    /// Optional policy-bundle digest (`sha256:<hex>`).
    pub policy_digest: Option<String>,
    /// Optional pass-through correlation id.
    pub correlation_id: Option<String>,
    /// CloudEvent `source` field.
    pub source: String,
    /// Cell id for event payloads.
    pub cell_id: String,
    /// Run id for event payloads.
    pub run_id: String,
    /// SEC-21 Phase 3h — opt-in DNSSEC validation policy. `None`
    /// preserves P3a/P3e behaviour exactly: the ticker calls the plain
    /// `SharedResolverFn` and never emits `dns_authority_dnssec_failed`
    /// events. When `Some`, the ticker uses
    /// [`Self::validated_resolver`] (which MUST also be set) and tags
    /// `dnssec_status` on every emitted `dns_authority_drift` event.
    /// Carries the EFFECTIVE per-tick policy — supervisors with a
    /// heterogeneous `dnsAuthority.resolvers[]` set today pick a single
    /// policy (the first opt-in resolver's) per ticker; multi-policy
    /// per-resolver routing is a future slice.
    pub dnssec_policy: Option<DnsResolverDnssecPolicy>,
    /// Trust anchors loaded from env / spec / IANA-default. Used purely
    /// for stamping the source descriptor into `dns_authority_dnssec_failed`
    /// event payloads. Hickory 0.24 limitation: the resolver does not
    /// accept custom anchors via public API; see [`super::dnssec`].
    pub trust_anchors: Option<TrustAnchors>,
    /// SEC-21 Phase 3h — DNSSEC-validating resolver closure. MUST be
    /// set when `dnssec_policy` is `Some`; ignored otherwise. The
    /// supervisor wires this to [`super::resolve_with_ttl_validated`]
    /// in production; tests pass synthetic closures.
    pub validated_resolver: Option<SharedValidatedResolverFn>,
}

/// Spawn the continuous ticker on the current tokio runtime.
///
/// Runs forever (`shutdown.load() == false`) on `cfg.interval` cadence,
/// invoking the injected `resolver` for each hostname inside
/// `tokio::task::spawn_blocking`. Each per-hostname Ok/Err is fed to a
/// [`super::ResolverRefresh::tick`] call against the ticker's *own*
/// [`super::ResolverState`]; resulting events are dispatched through
/// `emitter`.
///
/// **Tokio-context constraint:** must be called from inside a tokio
/// runtime (multi-thread or current-thread). The returned handle's
/// `task` lives on the runtime that was current at spawn time.
pub fn spawn_continuous_ticker(
    cfg: TickerConfig,
    emitter: Arc<dyn DriftEmitter>,
    resolver: SharedResolverFn,
) -> TickerHandle {
    let shutdown = Arc::new(AtomicBool::new(false));
    let shutdown_for_task = shutdown.clone();

    let task =
        tokio::spawn(
            async move { run_ticker_loop(cfg, emitter, resolver, shutdown_for_task).await },
        );

    TickerHandle { shutdown, task }
}

/// The actual loop, factored out for testability.
async fn run_ticker_loop(
    cfg: TickerConfig,
    emitter: Arc<dyn DriftEmitter>,
    resolver: SharedResolverFn,
    shutdown: Arc<AtomicBool>,
) -> TickerStats {
    let mut stats = TickerStats::default();
    let mut state = ResolverState::new();
    // SEC-21 Phase 3e — per-cell rebinding state lives ON the ticker so
    // observations persist across ticks for the lifetime of the cell. A
    // fresh state is created on every ticker spawn, matching the
    // ResolverState lifecycle. When `cfg.rebinding_policy is None`, this
    // state is never read or written by `tick_with_rebinding` (the
    // method short-circuits the rebinding-aware path).
    let mut rebinding_state = RebindingState::new();

    // Quick exit if the supervisor handed us nothing to refresh — saves
    // spinning the loop forever for a misconfigured cell.
    if cfg.hostnames.is_empty() {
        return stats;
    }

    loop {
        if shutdown.load(Ordering::SeqCst) {
            break;
        }

        // SEC-21 Phase 3h decision: when the operator has wired DNSSEC,
        // pre-resolve through the validating closure so the per-tick
        // refresher receives the validation discriminator alongside the
        // standard answer. When DNSSEC is off, take the unchanged P3a
        // pre-resolution path. Both branches honour the same
        // `spawn_blocking` discipline.
        let dnssec_active = cfg.dnssec_policy.is_some() && cfg.validated_resolver.is_some();
        let validated_resolved: HashMap<String, io::Result<ValidatedResolvedAnswer>> =
            if dnssec_active {
                let hostnames_for_resolve = cfg.hostnames.clone();
                let validated = cfg.validated_resolver.as_ref().unwrap().clone();
                match tokio::task::spawn_blocking(move || {
                    let mut out: HashMap<String, io::Result<ValidatedResolvedAnswer>> =
                        HashMap::new();
                    for hostname in &hostnames_for_resolve {
                        out.insert(hostname.clone(), validated(hostname));
                    }
                    out
                })
                .await
                {
                    Ok(map) => map,
                    Err(_) => {
                        if !sleep_or_shutdown(cfg.interval, &shutdown).await {
                            break;
                        }
                        continue;
                    }
                }
            } else {
                HashMap::new()
            };

        // Resolve every hostname on a blocking thread. Mirrors the W2
        // SEC-21 startup-tick path — never call sync resolvers on the
        // runtime thread. Phase 3: the closure itself dispatches to the
        // hickory-resolver async helper via `Handle::block_on`, but that
        // helper is fast-running per query and we still want the
        // spawn_blocking isolation to keep the runtime workers free.
        //
        // scope: when DNSSEC is active, the validated_resolved map is
        // the source of truth and we SKIP this unvalidated pre-resolve
        // entirely (the unvalidated closure may even be a panicking stub
        // in tests that prove the validated path is exclusively used).
        let resolved: HashMap<String, io::Result<ResolvedAnswer>> = if dnssec_active {
            HashMap::new()
        } else {
            let hostnames_for_resolve = cfg.hostnames.clone();
            let resolver_for_blocking = resolver.clone();
            let join_result = tokio::task::spawn_blocking(move || {
                let mut out: HashMap<String, io::Result<ResolvedAnswer>> = HashMap::new();
                for hostname in &hostnames_for_resolve {
                    out.insert(hostname.clone(), resolver_for_blocking(hostname));
                }
                out
            })
            .await;

            match join_result {
                Ok(map) => map,
                Err(_) => {
                    // spawn_blocking join error (panic / cancel). Skip
                    // this tick — counter not bumped because we did no
                    // useful work; loop continues so a transient runtime
                    // hiccup does not silently kill the ticker.
                    if !sleep_or_shutdown(cfg.interval, &shutdown).await {
                        break;
                    }
                    continue;
                }
            }
        };

        // Count resolver errors before we hand the map to tick().
        // When DNSSEC is active, the *validated* map is the source of
        // truth for error counting — the unvalidated map is unused.
        if dnssec_active {
            for v in validated_resolved.values() {
                if v.is_err() {
                    stats.resolver_errors = stats.resolver_errors.saturating_add(1);
                }
            }
        } else {
            for v in resolved.values() {
                if v.is_err() {
                    stats.resolver_errors = stats.resolver_errors.saturating_add(1);
                }
            }
        }

        let resolver_for_tick = |hostname: &str| -> io::Result<ResolvedAnswer> {
            match resolved.get(hostname) {
                Some(Ok(answer)) => Ok(answer.clone()),
                Some(Err(e)) => Err(io::Error::new(e.kind(), e.to_string())),
                None => Err(io::Error::other(
                    "ticker: hostname missing from pre-resolved map",
                )),
            }
        };

        let refresher = ResolverRefresh {
            policy: cfg.policy.as_ref(),
            rebinding_policy: cfg.rebinding_policy.as_ref(),
            resolvers: cfg.resolvers.as_slice(),
            hostnames: cfg.hostnames.as_slice(),
            keyset_id: cfg.keyset_id.as_deref(),
            issuer_kid: cfg.issuer_kid.as_deref(),
            policy_digest: cfg.policy_digest.as_deref(),
            correlation_id: cfg.correlation_id.as_deref(),
            source: Some(cfg.source.as_str()),
            // SEC-21 Phase 3h — DNSSEC validation policy + loaded
            // trust anchors. Both default to `None` (P3a/P3e behaviour
            // unchanged); the supervisor sets these in
            // `pick_dnssec_resolver_policy` when at least one resolver
            // has opted in via spec.
            dnssec_policy: cfg.dnssec_policy.as_ref(),
            trust_anchors: cfg.trust_anchors.as_ref(),
        };

        let events = if dnssec_active {
            // SEC-21 Phase 3h — drive the validating refresher path so
            // the per-tick decision tree can emit
            // `dns_authority_dnssec_failed` and tag drift events with
            // `dnssec_status`.
            refresher.tick_with_dnssec(
                &mut state,
                &mut rebinding_state,
                &validated_resolved,
                SystemTime::now(),
                &cfg.cell_id,
                &cfg.run_id,
            )
        } else {
            refresher.tick_with_rebinding(
                &mut state,
                &mut rebinding_state,
                &resolver_for_tick,
                SystemTime::now(),
                &cfg.cell_id,
                &cfg.run_id,
            )
        };

        for ev in events {
            stats.events_emitted = stats.events_emitted.saturating_add(1);
            emitter.emit(ev);
        }

        stats.tick_count = stats.tick_count.saturating_add(1);

        if !sleep_or_shutdown(cfg.interval, &shutdown).await {
            break;
        }
    }

    stats
}

/// Clamp an operator-supplied tick interval (in seconds) to the
/// minimum-5s floor the ticker enforces. Pure helper, exposed so the
/// supervisor wiring + the property test can both call into it.
pub fn clamp_tick_interval_secs(secs: u64) -> u64 {
    secs.max(5)
}

/// Sleep for `interval` OR return early when `shutdown` flips to `true`.
///
/// Returns `true` when the interval elapsed normally, `false` when
/// shutdown was requested mid-sleep. Polls the shutdown flag on a
/// 50ms granularity so worst-case shutdown latency is ~50ms even when
/// `interval` is large (e.g. 60s).
async fn sleep_or_shutdown(interval: Duration, shutdown: &AtomicBool) -> bool {
    // Poll-based wake. A `tokio::sync::Notify` would be tighter but adds
    // a second synchronization primitive on top of the AtomicBool the
    // supervisor already owns; the proxy spawn module took the same
    // poll-on-AtomicBool route for the same reason.
    let poll_step = Duration::from_millis(50);
    let deadline = std::time::Instant::now() + interval;
    loop {
        if shutdown.load(Ordering::SeqCst) {
            return false;
        }
        let now = std::time::Instant::now();
        if now >= deadline {
            return true;
        }
        let remaining = deadline - now;
        tokio::time::sleep(remaining.min(poll_step)).await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
    use std::sync::atomic::AtomicU64;
    use std::sync::Mutex;

    /// Build a Phase-3-shaped answer. The ticker tests don't probe TTL
    /// semantics directly — that's covered in `super::tests` for the clamp
    /// and in [`super::hickory_resolve::tests`] for the wire-format min-TTL —
    /// so we default to ttl=0 here.
    fn answer(targets: Vec<String>) -> ResolvedAnswer {
        ResolvedAnswer {
            targets,
            ttl_seconds: 0,
            resolver_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 53),
        }
    }

    #[derive(Default)]
    struct CollectingEmitter {
        events: Mutex<Vec<CloudEventV1>>,
    }
    impl DriftEmitter for CollectingEmitter {
        fn emit(&self, event: CloudEventV1) {
            self.events.lock().unwrap().push(event);
        }
    }

    fn one_resolver() -> Vec<DnsResolver> {
        vec![DnsResolver {
            resolver_id: "resolver-doh-cloudflare".into(),
            endpoint: "https://1.1.1.1/dns-query".into(),
            protocol: cellos_core::DnsResolverProtocol::Doh,
            trust_kid: None,
            dnssec: None,
        }]
    }

    fn base_cfg(hostnames: Vec<String>, interval: Duration) -> TickerConfig {
        TickerConfig {
            interval,
            policy: Some(DnsRefreshPolicy {
                min_ttl_seconds: Some(0),
                max_stale_seconds: None,
                strategy: None,
            }),
            rebinding_policy: None,
            resolvers: one_resolver(),
            hostnames,
            keyset_id: Some("keyset-test".into()),
            issuer_kid: Some("kid-test".into()),
            policy_digest: None,
            correlation_id: None,
            source: "cellos-supervisor-test".into(),
            cell_id: "cell-A".into(),
            run_id: "run-A".into(),
            // SEC-21 Phase 3h — DNSSEC opt-out for the baseline test
            // helper. The Phase 3h-specific tests build their own
            // TickerConfig with these fields populated.
            dnssec_policy: None,
            trust_anchors: None,
            validated_resolver: None,
        }
    }

    /// Inject a resolver whose answers cycle through a fixed sequence —
    /// each call advances by one, the last entry repeats forever.
    fn cycling_resolver(sequence: Vec<Vec<String>>) -> SharedResolverFn {
        let counter = Arc::new(AtomicU64::new(0));
        let seq = Arc::new(sequence);
        Arc::new(move |_h: &str| {
            let idx = counter.fetch_add(1, Ordering::SeqCst) as usize;
            let pick = if idx >= seq.len() {
                seq.last().cloned().unwrap_or_default()
            } else {
                seq[idx].clone()
            };
            Ok(answer(pick))
        })
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_emits_drift_when_targets_change_between_ticks() {
        let cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(100));
        let emitter = Arc::new(CollectingEmitter::default());
        let resolver = cycling_resolver(vec![
            vec!["1.1.1.1".into()],
            vec!["1.0.0.1".into()],
            vec!["1.0.0.1".into()],
        ]);

        let handle = spawn_continuous_ticker(cfg, emitter.clone(), resolver);
        // Two ticks at 100ms — sleep 250ms to comfortably observe both.
        tokio::time::sleep(Duration::from_millis(250)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let stats = tokio::time::timeout(Duration::from_secs(1), handle.task)
            .await
            .expect("ticker join timeout")
            .expect("ticker task panicked");

        let events = emitter.events.lock().unwrap();
        assert!(
            events.len() >= 2,
            "expected baseline + change drift events, got {}",
            events.len()
        );
        assert!(stats.tick_count >= 2);
        assert!(stats.events_emitted >= 2);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_silent_when_targets_stable() {
        let cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(50));
        let emitter = Arc::new(CollectingEmitter::default());
        // Stable answer across all ticks.
        let resolver: SharedResolverFn =
            Arc::new(|_h: &str| Ok(answer(vec!["203.0.113.10".into()])));

        let handle = spawn_continuous_ticker(cfg, emitter.clone(), resolver);
        tokio::time::sleep(Duration::from_millis(250)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task)
            .await
            .expect("ticker join timeout");

        let events = emitter.events.lock().unwrap();
        assert_eq!(
            events.len(),
            1,
            "stable targets must emit exactly one baseline event, got {}",
            events.len()
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_respects_shutdown_promptly() {
        let cfg = base_cfg(vec!["api.example.com".into()], Duration::from_secs(10));
        let emitter = Arc::new(CollectingEmitter::default());
        let resolver: SharedResolverFn = Arc::new(|_h: &str| Ok(answer(vec!["1.1.1.1".into()])));

        let handle = spawn_continuous_ticker(cfg, emitter, resolver);
        // Let the first tick fire, then immediately shutdown — the loop
        // should be in `sleep_or_shutdown` waiting on a 10s deadline,
        // and the 50ms poll granularity means we collapse to ~50ms.
        tokio::time::sleep(Duration::from_millis(80)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let started = std::time::Instant::now();
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task)
            .await
            .expect("ticker did not honour shutdown within 1s");
        let elapsed = started.elapsed();
        assert!(
            elapsed < Duration::from_millis(500),
            "shutdown took {elapsed:?}, expected <500ms"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_respects_floor_interval_min() {
        // The ≥5s floor is enforced at the supervisor call site, not in
        // the ticker module — the ticker honours whatever interval it is
        // handed. This test pins the supervisor-side helper that performs
        // the clamp so a future refactor of the call site cannot silently
        // remove the floor.
        let floor = crate::resolver_refresh::ticker::clamp_tick_interval_secs(1);
        assert!(floor >= 5, "tick interval floor must be >=5s; got {floor}");
        let unbounded = crate::resolver_refresh::ticker::clamp_tick_interval_secs(120);
        assert_eq!(unbounded, 120, "values >=floor must pass through untouched");
        let zero = crate::resolver_refresh::ticker::clamp_tick_interval_secs(0);
        assert!(zero >= 5, "zero must clamp up to floor");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_handles_resolver_failure_gracefully() {
        let cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(50));
        let emitter = Arc::new(CollectingEmitter::default());
        let resolver: SharedResolverFn = Arc::new(|_h: &str| Err(io::Error::other("transient")));

        let handle = spawn_continuous_ticker(cfg, emitter.clone(), resolver);
        tokio::time::sleep(Duration::from_millis(250)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let stats = tokio::time::timeout(Duration::from_secs(1), handle.task)
            .await
            .expect("ticker join timeout")
            .expect("ticker task panicked");

        let events = emitter.events.lock().unwrap();
        assert!(
            events.is_empty(),
            "resolver failures must not emit drift, got {} events",
            events.len()
        );
        assert!(
            stats.resolver_errors >= 1,
            "resolver_errors counter should reflect the failures, got {}",
            stats.resolver_errors
        );
        assert!(
            stats.tick_count >= 1,
            "ticker must keep running across resolver errors"
        );
    }

    // ============================================================
    // SEC-21 Phase 3e — DNS rebinding closure tests.
    //
    // These exercise the `rebinding_policy` field threaded through
    // `TickerConfig` → `ResolverRefresh::tick_with_rebinding`. The unit
    // tests in `super::tests` cover the per-tick decision math; these
    // tests pin the END-TO-END ticker behaviour so a future refactor
    // can't silently disconnect the policy from event emission.
    // ============================================================

    fn count_events_of(events: &[CloudEventV1], suffix: &str) -> usize {
        events.iter().filter(|e| e.ty.ends_with(suffix)).count()
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_no_rebinding_events_when_policy_is_none() {
        // Baseline contract: `rebinding_policy: None` → ticker emits
        // only the standard `dns_authority_drift` events. A future
        // refactor that accidentally wires the rebinding code path to
        // fire unconditionally will trip this test.
        let mut cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(80));
        cfg.rebinding_policy = None;
        let emitter = Arc::new(CollectingEmitter::default());
        // 5 ticks worth of churning IPs — rebind events would normally
        // fire for at least the first 4-5 distinct IPs.
        let resolver = cycling_resolver(vec![
            vec!["1.0.0.1".into()],
            vec!["1.0.0.2".into()],
            vec!["1.0.0.3".into()],
            vec!["1.0.0.4".into()],
            vec!["1.0.0.5".into()],
        ]);

        let handle = spawn_continuous_ticker(cfg, emitter.clone(), resolver);
        tokio::time::sleep(Duration::from_millis(250)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task).await;

        let events = emitter.events.lock().unwrap();
        assert_eq!(
            count_events_of(&events, "dns_authority_rebind_threshold"),
            0,
            "no rebinding policy → no threshold events"
        );
        assert_eq!(
            count_events_of(&events, "dns_authority_rebind_rejected"),
            0,
            "no rebinding policy → no rejected events"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_threshold_only_audit_mode_emits_threshold_events() {
        // Audit-only: `reject_on_rebind=false`. Cap at 2 distinct IPs.
        // The ticker walks 4 distinct IPs across 4 ticks; once the
        // cumulative count exceeds 2, every subsequent tick that adds
        // a novel IP fires a threshold event.
        let mut cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(60));
        cfg.rebinding_policy = Some(DnsRebindingPolicy {
            response_ip_allowlist: Vec::new(),
            max_novel_ips_per_hostname: 2,
            reject_on_rebind: false,
        });
        cfg.policy_digest =
            Some("sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".into());
        let emitter = Arc::new(CollectingEmitter::default());
        let resolver = cycling_resolver(vec![
            vec!["1.0.0.1".into()],
            vec!["1.0.0.2".into()],
            vec!["1.0.0.3".into()],
            vec!["1.0.0.4".into()],
        ]);

        let handle = spawn_continuous_ticker(cfg, emitter.clone(), resolver);
        tokio::time::sleep(Duration::from_millis(280)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task).await;

        let events = emitter.events.lock().unwrap();
        let threshold = count_events_of(&events, "dns_authority_rebind_threshold");
        let rejected = count_events_of(&events, "dns_authority_rebind_rejected");
        assert!(
            threshold >= 2,
            "audit-only mode must emit at least 2 threshold events when cap is breached over multiple ticks; got {threshold}"
        );
        assert_eq!(
            rejected, 0,
            "no allowlist set → no rejected events; got {rejected}"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_reject_on_rebind_filters_drift_targets() {
        // Enforce mode: cap=2, reject_on_rebind=true. Resolver returns a
        // cumulative-style answer that grows on each tick (mirrors a CDN
        // legitimately serving multiple A records, with an attacker IP
        // appended on tick 2). After the cap fills, the new IP must be
        // filtered out of the workload-visible target set.
        let mut cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(60));
        cfg.rebinding_policy = Some(DnsRebindingPolicy {
            response_ip_allowlist: Vec::new(),
            max_novel_ips_per_hostname: 2,
            reject_on_rebind: true,
        });
        cfg.policy_digest =
            Some("sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".into());
        let emitter = Arc::new(CollectingEmitter::default());
        // Tick 1: two legitimate CDN IPs (fills cap=2 exactly).
        // Tick 2: same two PLUS an attacker IP — must be filtered.
        let resolver = cycling_resolver(vec![
            vec!["1.0.0.1".into(), "1.0.0.2".into()],
            vec!["1.0.0.1".into(), "1.0.0.2".into(), "198.51.100.7".into()],
        ]);

        let handle = spawn_continuous_ticker(cfg, emitter.clone(), resolver);
        tokio::time::sleep(Duration::from_millis(220)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task).await;

        let events = emitter.events.lock().unwrap();
        let drift_events: Vec<_> = events
            .iter()
            .filter(|e| e.ty.ends_with("dns_authority_drift"))
            .collect();
        assert!(
            !drift_events.is_empty(),
            "drift events must still fire (rejection only filters targets, not the drift signal)"
        );
        let last = drift_events.last().unwrap();
        let data = last.data.as_ref().expect("data");
        let current: Vec<&str> = data["currentTargets"]
            .as_array()
            .unwrap()
            .iter()
            .map(|v| v.as_str().unwrap())
            .collect();
        assert!(
            current.contains(&"1.0.0.1"),
            "first legitimate IP must survive rejection: {current:?}"
        );
        assert!(
            current.contains(&"1.0.0.2"),
            "second legitimate IP must survive rejection: {current:?}"
        );
        assert!(
            !current.contains(&"198.51.100.7"),
            "attacker IP beyond cap=2 must be filtered when reject_on_rebind=true: {current:?}"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_allowlist_only_emits_rejected_events() {
        // Allowlist set, cap large, reject=false. Resolver returns one
        // IP NOT in the allowlist on every tick → one rejected event
        // per tick (deduped per-tick — no flapping when the same IP
        // repeats). No threshold events because the cap isn't tight.
        let mut cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(60));
        cfg.rebinding_policy = Some(DnsRebindingPolicy {
            response_ip_allowlist: vec!["api.example.com:1.1.1.1".into()],
            max_novel_ips_per_hostname: 100,
            reject_on_rebind: false,
        });
        cfg.policy_digest =
            Some("sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".into());
        let emitter = Arc::new(CollectingEmitter::default());
        let resolver: SharedResolverFn =
            Arc::new(|_h: &str| Ok(answer(vec!["198.51.100.7".into()])));

        let handle = spawn_continuous_ticker(cfg, emitter.clone(), resolver);
        tokio::time::sleep(Duration::from_millis(220)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task).await;

        let events = emitter.events.lock().unwrap();
        let threshold = count_events_of(&events, "dns_authority_rebind_threshold");
        let rejected = count_events_of(&events, "dns_authority_rebind_rejected");
        // The first tick observes the disallowed IP; subsequent ticks
        // observe the SAME IP (cumulative count stays at 1, well below
        // the 100 cap) — but the allowlist check fires every tick.
        assert!(
            rejected >= 1,
            "allowlist violation must fire at least one rejected event"
        );
        assert_eq!(
            threshold, 0,
            "cap is far above the IP count → no threshold events; got {threshold}"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_combined_threshold_and_allowlist_both_fire() {
        // Both policies active: tight cap AND allowlist. Resolver
        // returns IPs that violate both. Assert both event types fire.
        let mut cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(60));
        cfg.rebinding_policy = Some(DnsRebindingPolicy {
            response_ip_allowlist: vec!["api.example.com:1.1.1.1".into()],
            max_novel_ips_per_hostname: 1,
            reject_on_rebind: false,
        });
        cfg.policy_digest =
            Some("sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".into());
        let emitter = Arc::new(CollectingEmitter::default());
        // Tick 1: 1.1.1.1 (in allowlist, fills cap).
        // Tick 2: 198.51.100.7 (not in allowlist, exceeds cap).
        let resolver = cycling_resolver(vec![vec!["1.1.1.1".into()], vec!["198.51.100.7".into()]]);

        let handle = spawn_continuous_ticker(cfg, emitter.clone(), resolver);
        tokio::time::sleep(Duration::from_millis(220)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task).await;

        let events = emitter.events.lock().unwrap();
        let threshold = count_events_of(&events, "dns_authority_rebind_threshold");
        let rejected = count_events_of(&events, "dns_authority_rebind_rejected");
        assert!(
            threshold >= 1,
            "second tick exceeds cap=1 → threshold event expected"
        );
        assert!(
            rejected >= 1,
            "second tick IP is not in allowlist → rejected event expected"
        );
    }

    // ============================================================
    // SEC-21 Phase 3h — DNSSEC validation ticker tests.
    //
    // These exercise the `dnssec_policy` + `validated_resolver`
    // wiring threaded through `TickerConfig` and pin the load-bearing
    // properties:
    //
    //   1. `dns_authority_dnssec_failed` events fire when the validator
    //      reports Failed/Unsigned and the policy is active (audit OR
    //      enforce mode — the event ALWAYS fires; failClosed only
    //      changes whether the answer is dropped).
    //   2. `failClosed=true` empties `currentTargets` in the drift
    //      event so the workload-facing target set never sees the
    //      attacker IPs.
    //   3. `dnssec_status` is stamped on every drift event in DNSSEC
    //      mode (validated / validation_failed / unsigned) so the SIEM
    //      can correlate without parsing the dnssec_failed event.
    // ============================================================

    fn cycling_validated_resolver(
        sequence: Vec<crate::resolver_refresh::ValidatedResolvedAnswer>,
    ) -> SharedValidatedResolverFn {
        let counter = Arc::new(AtomicU64::new(0));
        let seq = Arc::new(sequence);
        Arc::new(move |_h: &str| {
            let idx = counter.fetch_add(1, Ordering::SeqCst) as usize;
            Ok(if idx >= seq.len() {
                seq.last().cloned().unwrap_or_else(|| {
                    crate::resolver_refresh::ValidatedResolvedAnswer {
                        answer: ResolvedAnswer {
                            targets: Vec::new(),
                            ttl_seconds: 0,
                            resolver_addr: SocketAddr::new(
                                IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
                                53,
                            ),
                        },
                        validation: crate::resolver_refresh::DnssecValidationResult::Unsigned,
                    }
                })
            } else {
                seq[idx].clone()
            })
        })
    }

    fn validated_with(
        targets: Vec<&str>,
        validation: crate::resolver_refresh::DnssecValidationResult,
    ) -> crate::resolver_refresh::ValidatedResolvedAnswer {
        crate::resolver_refresh::ValidatedResolvedAnswer {
            answer: ResolvedAnswer {
                targets: targets.into_iter().map(String::from).collect(),
                ttl_seconds: 60,
                resolver_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 53),
            },
            validation,
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dnssec_failed_event_emitted_when_validate_true_failclosed_false() {
        // Audit-only mode: validate=true, failClosed=false. Validator
        // reports Failed → ticker MUST emit dns_authority_dnssec_failed
        // BUT keep the answer (drift fires with the original targets,
        // dnssec_status=validation_failed).
        let mut cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(70));
        cfg.dnssec_policy = Some(DnsResolverDnssecPolicy {
            validate: true,
            fail_closed: false,
            trust_anchors_path: None,
        });
        cfg.trust_anchors = Some(crate::resolver_refresh::TrustAnchors::iana_default());
        cfg.validated_resolver = Some(cycling_validated_resolver(vec![validated_with(
            vec!["1.0.0.1"],
            crate::resolver_refresh::DnssecValidationResult::Failed {
                reason: "synthetic-bogus".to_string(),
            },
        )]));
        let emitter = Arc::new(CollectingEmitter::default());

        let handle = spawn_continuous_ticker(
            cfg,
            emitter.clone(),
            Arc::new(|_h: &str| Ok(answer(vec!["1.0.0.1".into()]))),
        );
        tokio::time::sleep(Duration::from_millis(220)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task).await;

        let events = emitter.events.lock().unwrap();
        let dnssec_failed = count_events_of(&events, "dns_authority_dnssec_failed");
        assert!(
            dnssec_failed >= 1,
            "audit-only DNSSEC failure must fire dns_authority_dnssec_failed; got {dnssec_failed}"
        );
        // Drift event is still present because the answer was kept.
        let drift_events: Vec<_> = events
            .iter()
            .filter(|e| e.ty.ends_with("dns_authority_drift"))
            .collect();
        assert!(
            !drift_events.is_empty(),
            "audit-only mode keeps the answer; drift must still fire"
        );
        // Drift must carry currentTargets (audit-only does not drop).
        let last_drift = drift_events.last().unwrap();
        let data = last_drift.data.as_ref().expect("data");
        let current: Vec<&str> = data["currentTargets"]
            .as_array()
            .unwrap()
            .iter()
            .map(|v| v.as_str().unwrap())
            .collect();
        assert!(
            current.contains(&"1.0.0.1"),
            "audit-only mode preserves the unvalidated answer in the drift event: {current:?}"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dnssec_drops_answer_when_failclosed_true() {
        // Enforce mode: validate=true, failClosed=true. Validator
        // reports Failed → ticker MUST drop the answer (drift's
        // currentTargets is empty) AND fire dns_authority_dnssec_failed.
        let mut cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(70));
        cfg.dnssec_policy = Some(DnsResolverDnssecPolicy {
            validate: true,
            fail_closed: true,
            trust_anchors_path: None,
        });
        cfg.trust_anchors = Some(crate::resolver_refresh::TrustAnchors::iana_default());
        cfg.validated_resolver = Some(cycling_validated_resolver(vec![validated_with(
            vec!["198.51.100.7"],
            crate::resolver_refresh::DnssecValidationResult::Failed {
                reason: "synthetic-bogus".to_string(),
            },
        )]));
        let emitter = Arc::new(CollectingEmitter::default());

        let handle = spawn_continuous_ticker(
            cfg,
            emitter.clone(),
            Arc::new(|_h: &str| Ok(answer(vec!["198.51.100.7".into()]))),
        );
        tokio::time::sleep(Duration::from_millis(220)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task).await;

        let events = emitter.events.lock().unwrap();
        let dnssec_failed = count_events_of(&events, "dns_authority_dnssec_failed");
        assert!(
            dnssec_failed >= 1,
            "enforce DNSSEC failure must fire dns_authority_dnssec_failed; got {dnssec_failed}"
        );

        let drift_events: Vec<_> = events
            .iter()
            .filter(|e| e.ty.ends_with("dns_authority_drift"))
            .collect();
        // failClosed=true: the answer is dropped, but the drift event
        // still fires (with empty currentTargets) so the SIEM observes
        // the workload-visible state.
        if let Some(last) = drift_events.last() {
            let data = last.data.as_ref().expect("data");
            let current: Vec<&str> = data["currentTargets"]
                .as_array()
                .unwrap()
                .iter()
                .map(|v| v.as_str().unwrap())
                .collect();
            assert!(
                !current.contains(&"198.51.100.7"),
                "failClosed=true MUST drop the attacker IP from drift currentTargets: {current:?}"
            );
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dnssec_status_field_set_in_drift_event() {
        // When DNSSEC is active and the validator reports Validated,
        // the drift event MUST carry dnssec_status="validated" so the
        // SIEM can correlate without parsing dns_authority_dnssec_failed
        // (which won't fire on the success path).
        let mut cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(70));
        cfg.dnssec_policy = Some(DnsResolverDnssecPolicy {
            validate: true,
            fail_closed: false,
            trust_anchors_path: None,
        });
        cfg.trust_anchors = Some(crate::resolver_refresh::TrustAnchors::iana_default());
        cfg.validated_resolver = Some(cycling_validated_resolver(vec![validated_with(
            vec!["1.1.1.1"],
            crate::resolver_refresh::DnssecValidationResult::Validated {
                algorithm: "RSASHA256".to_string(),
                key_tag: 19036,
            },
        )]));
        let emitter = Arc::new(CollectingEmitter::default());

        let handle = spawn_continuous_ticker(
            cfg,
            emitter.clone(),
            Arc::new(|_h: &str| Ok(answer(vec!["1.1.1.1".into()]))),
        );
        tokio::time::sleep(Duration::from_millis(180)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task).await;

        let events = emitter.events.lock().unwrap();
        // No dnssec_failed event on the success path.
        assert_eq!(
            count_events_of(&events, "dns_authority_dnssec_failed"),
            0,
            "Validated path must not emit dns_authority_dnssec_failed"
        );
        // Drift event(s) carry dnssec_status: "validated".
        let drift_events: Vec<_> = events
            .iter()
            .filter(|e| e.ty.ends_with("dns_authority_drift"))
            .collect();
        assert!(
            !drift_events.is_empty(),
            "drift must fire on first observation"
        );
        let data = drift_events[0].data.as_ref().expect("data");
        assert_eq!(
            data["dnssecStatus"], "validated",
            "drift in DNSSEC mode must stamp dnssecStatus=validated"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ticker_novel_ip_within_cap_does_not_emit_threshold() {
        // Cap=4 (the default), resolver returns 3 distinct IPs over 3
        // ticks — cumulative 3, never exceeds. No threshold events.
        let mut cfg = base_cfg(vec!["api.example.com".into()], Duration::from_millis(60));
        cfg.rebinding_policy = Some(DnsRebindingPolicy {
            response_ip_allowlist: Vec::new(),
            max_novel_ips_per_hostname: 4,
            reject_on_rebind: false,
        });
        cfg.policy_digest =
            Some("sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".into());
        let emitter = Arc::new(CollectingEmitter::default());
        let resolver = cycling_resolver(vec![
            vec!["1.0.0.1".into()],
            vec!["1.0.0.2".into()],
            vec!["1.0.0.3".into()],
        ]);

        let handle = spawn_continuous_ticker(cfg, emitter.clone(), resolver);
        tokio::time::sleep(Duration::from_millis(220)).await;
        handle.shutdown.store(true, Ordering::SeqCst);
        let _ = tokio::time::timeout(Duration::from_secs(1), handle.task).await;

        let events = emitter.events.lock().unwrap();
        let threshold = count_events_of(&events, "dns_authority_rebind_threshold");
        assert_eq!(
            threshold, 0,
            "3 distinct IPs under cap=4 must not fire threshold; got {threshold}"
        );
    }
}