osdns 0.2.0

//! Enforce-policy reconciliation: rebasing and reapplying an active lease's
//! desired overlay when an external actor changes the base DNS state.
//!
//! Reconciliation runs as a background worker while at least one lease on an
//! [`Enforce`](crate::ConflictPolicy)-policy manager is active, independent
//! of any public [`DnsManager::watch`](crate::DnsManager::watch)
//! subscription, which remains purely observational.
//!
//! # Event handling is state-aware and never drops a legitimate change
//!
//! Every watcher event for a resource owned by an active lease is fed to the
//! reconciler *before* the time-based suppression filter. The reconciler
//! reads the authoritative system state: when it still matches our applied
//! overlay (i.e. the event was generated by our own mutation), the pass is a
//! no-op - that is the suppression. When it does not, the change is real and
//! is acted on. Events that arrive while a resource is deferred or
//! rate-limited are coalesced into a pending set and processed as soon as the
//! resource becomes due; they are never discarded.
//!
//! # Rebase follows the normal transaction model
//!
//! An externally modified base is adopted with a full journal transaction:
//! capture the stable external base → persist/fsync `Prepared` with the new
//! base → apply the overlay → read back and verify → persist `Applied`. On
//! failure the pass rolls back to the new external base and defers. A
//! `Prepared` record is finalized only from retained backend-issued proof,
//! never because current DNS values happen to match `desired`. Orphaned crash
//! recovery reports a conflict.
//!
//! A feedback-loop circuit breaker bounds rebase attempts per resource; while
//! it is open, reconciliation is deferred (not dropped) until the cooldown
//! expires.

use std::collections::{HashMap, VecDeque};
use std::sync::mpsc::{self, RecvTimeoutError};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use crate::error::Error;
use crate::fault::TxPoint;
use crate::journal::Phase;
use crate::lease::LiveRecord;
use crate::manager::{INITIAL_POINTS, Inner};
use crate::normalize::NormalizedConfig;
use crate::ownership::ResourceId;
use crate::platform::PlatformSnapshot;
use crate::platform::ResourceStatus;

/// Two read-backs separated by this window must agree before the state is
/// considered stable and actionable.
pub(crate) const STABLE_WINDOW: Duration = Duration::from_millis(100);
/// Retries after an unstable state wait at least this long.
pub(crate) const UNSTABLE_RETRY: Duration = Duration::from_millis(200);
/// Retries after a backend error wait at least this long.
pub(crate) const ERROR_RETRY: Duration = Duration::from_millis(250);
/// Identity ambiguity is retried slowly so a transient native transition can
/// recover without turning an active Enforce lease into a hot retry loop.
pub(crate) const IDENTITY_RETRY: Duration = Duration::from_secs(5);
/// Minimum spacing between two full rebase transactions on one resource.
const BREAKER_WINDOW: Duration = Duration::from_secs(5);
const BREAKER_THRESHOLD: usize = 6;
const BREAKER_COOLDOWN: Duration = Duration::from_secs(2);
/// A resource whose reconciliation keeps hard-failing is dropped from the
/// pending set after this many consecutive errors (with a warning); the next
/// watcher event for it restarts the cycle.
const MAX_CONSECUTIVE_ERRORS: u32 = 10;

/// What a single reconciliation pass decided. Returned by the testing-only
/// entry point so Enforce semantics can be tested deterministically.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ReconcileOutcome {
    NoActiveLease,
    IdentityAmbiguous,
    StillOurs,
    Rebased,
    Deferred,
    #[allow(dead_code)]
    Failed,
}

#[derive(Debug, Clone)]
struct Pending {
    ready_at: Instant,
    consecutive_errors: u32,
}

#[derive(Default)]
struct BreakerState {
    attempts: VecDeque<Instant>,
    open_until: Option<Instant>,
}

#[derive(Default)]
pub(crate) struct Reconciler {
    pending: Mutex<HashMap<ResourceId, Pending>>,
    breaker: Mutex<HashMap<ResourceId, BreakerState>>,
}

impl Reconciler {
    #[cfg(feature = "test-util")]
    pub(crate) fn is_pending(&self, resource: &ResourceId) -> bool {
        self.pending
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .contains_key(resource)
    }
    /// Coalesces a watcher event into the pending set. An already-pending
    /// resource keeps its scheduled time: deferral windows (rate limit,
    /// circuit breaker, error backoff) are never bypassed by new events, and
    /// the deferred pass always reads the latest authoritative state.
    fn touch(&self, resource: ResourceId) {
        let mut pending = self
            .pending
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        pending.entry(resource).or_insert(Pending {
            ready_at: Instant::now(),
            consecutive_errors: 0,
        });
    }

    fn defer(&self, resource: &ResourceId, delay: Duration, failed: bool) {
        let mut pending = self
            .pending
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let now = Instant::now();
        let entry = pending.entry(resource.clone()).or_insert(Pending {
            ready_at: now + delay,
            consecutive_errors: 0,
        });
        entry.ready_at = now + delay;
        if failed {
            entry.consecutive_errors += 1;
        }
    }

    fn remove(&self, resource: &ResourceId) {
        self.pending
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .remove(resource);
    }

    #[cfg(feature = "test-util")]
    pub(crate) fn clear(&self) {
        self.pending
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clear();
    }

    fn breaker_gate(&self, resource: &ResourceId) -> Option<Duration> {
        let mut breaker = self
            .breaker
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let state = breaker.entry(resource.clone()).or_default();
        let now = Instant::now();
        if let Some(open_until) = state.open_until {
            if now < open_until {
                return Some(open_until.saturating_duration_since(now));
            }
            state.open_until = None;
            state.attempts.clear();
        }
        while let Some(oldest) = state.attempts.front() {
            if now.duration_since(*oldest) > BREAKER_WINDOW {
                state.attempts.pop_front();
            } else {
                break;
            }
        }
        if state.attempts.len() >= BREAKER_THRESHOLD {
            state.open_until = Some(now + BREAKER_COOLDOWN);
            state.attempts.clear();
            return Some(BREAKER_COOLDOWN);
        }
        state.attempts.push_back(now);
        None
    }
}

impl Inner {
    pub(crate) fn lease_token(&self, resource: &ResourceId) -> Arc<Mutex<()>> {
        self.lease_tokens
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .entry(resource.clone())
            .or_default()
            .clone()
    }

    pub(crate) fn register_active(&self, record: Arc<Mutex<LiveRecord>>) {
        let resource = record
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .record
            .resource
            .clone();
        self.active
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .insert(resource, record);
    }

    pub(crate) fn unregister_active(&self, resource: &ResourceId) {
        self.active
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .remove(resource);
        self.lease_tokens
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .remove(resource);
    }

    /// Runs `f` with one live record locked against reconciliation: the
    /// per-resource token is held for the whole call so a reconcile pass
    /// cannot interleave with lease teardown or updates.
    pub(crate) fn with_live_record(
        &self,
        live: &Arc<Mutex<LiveRecord>>,
        f: impl FnOnce(&mut LiveRecord),
    ) {
        let resource = live
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .record
            .resource
            .clone();
        let token = self.lease_token(&resource);
        let _token_guard = token
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut guard = live.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        f(&mut guard);
    }

    pub(crate) fn reconcile_resource(
        &self,
        resource: &ResourceId,
        reconciler: &Reconciler,
    ) -> ReconcileOutcome {
        let Some(_entry) = self
            .active
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(resource)
            .cloned()
        else {
            reconciler.remove(resource);
            return ReconcileOutcome::NoActiveLease;
        };
        let token = self.lease_token(resource);
        let _token_guard = token
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        // Re-check after acquiring the token: the lease may have ended while
        // this pass waited for it.
        let Some(entry) = self
            .active
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(resource)
            .cloned()
        else {
            reconciler.remove(resource);
            return ReconcileOutcome::NoActiveLease;
        };

        let identity = entry
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .record
            .identity
            .clone();
        match self.backend.resource_status(&identity) {
            Ok(ResourceStatus::Gone | ResourceStatus::Replaced) => {
                let record = entry
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .record
                    .clone();
                if self.journal.remove(&record.lease_id, resource).is_ok() {
                    self.unregister_active(resource);
                    reconciler.remove(resource);
                    return ReconcileOutcome::NoActiveLease;
                }
                return ReconcileOutcome::Failed;
            }
            Ok(ResourceStatus::Ambiguous) => {
                reconciler.defer(resource, IDENTITY_RETRY, false);
                osdns_warn!(
                    resource = %resource,
                    "native resource identity is ambiguous; retaining the lease and journal without mutation"
                );
                return ReconcileOutcome::IdentityAmbiguous;
            }
            Ok(ResourceStatus::Same) => {}
            Err(_) => return ReconcileOutcome::Failed,
        }

        let outcome = self.reconcile_pass(resource, &entry, reconciler);
        match outcome {
            ReconcileOutcome::NoActiveLease
            | ReconcileOutcome::StillOurs
            | ReconcileOutcome::Rebased => {
                reconciler.remove(resource);
            }
            ReconcileOutcome::Deferred => {
                reconciler.defer(resource, UNSTABLE_RETRY, false);
            }
            ReconcileOutcome::IdentityAmbiguous => {
                // The identity branch already scheduled its deliberately slow
                // retry. Keep the lease registered and the journal intact.
            }
            ReconcileOutcome::Failed => {
                let pending = self
                    .reconciler
                    .pending
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                let exhausted = pending
                    .get(resource)
                    .is_some_and(|p| p.consecutive_errors + 1 >= MAX_CONSECUTIVE_ERRORS);
                drop(pending);
                if exhausted {
                    osdns_warn!(
                        resource = %resource,
                        "reconciliation keeps failing for this resource; dropping it from the pending set until the next watcher event"
                    );
                    reconciler.remove(resource);
                } else {
                    reconciler.defer(resource, ERROR_RETRY, true);
                }
            }
        }
        outcome
    }

    #[allow(unused_variables)]
    fn reconcile_pass(
        &self,
        resource: &ResourceId,
        entry: &Arc<Mutex<LiveRecord>>,
        reconciler: &Reconciler,
    ) -> ReconcileOutcome {
        let first = match self.backend.readback(resource) {
            Ok(first) => first,
            Err(Error::ResourceGone { .. }) => {
                let record = entry
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .record
                    .clone();
                if self.journal.remove(&record.lease_id, resource).is_ok() {
                    self.unregister_active(resource);
                    return ReconcileOutcome::NoActiveLease;
                }
                return ReconcileOutcome::Failed;
            }
            Err(error) => {
                osdns_warn!(
                    resource = %resource,
                    error = %error,
                    "reconciliation could not read the current state; deferring"
                );
                return ReconcileOutcome::Deferred;
            }
        };
        std::thread::sleep(STABLE_WINDOW);
        let second = match self.backend.readback(resource) {
            Ok(second) => second,
            Err(Error::ResourceGone { .. }) => {
                let record = entry
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .record
                    .clone();
                if self.journal.remove(&record.lease_id, resource).is_ok() {
                    self.unregister_active(resource);
                    return ReconcileOutcome::NoActiveLease;
                }
                return ReconcileOutcome::Failed;
            }
            Err(error) => {
                osdns_warn!(
                    resource = %resource,
                    error = %error,
                    "reconciliation could not re-read the current state; deferring"
                );
                return ReconcileOutcome::Deferred;
            }
        };
        if !self.backend.equivalent(&first, &second) {
            return ReconcileOutcome::Deferred;
        }

        let mut guard = entry
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        match self.finalize_live(&mut guard, Some(&second)) {
            Ok(()) => {}
            Err(error) if error.is_external_modification() => {
                if guard.record.phase == Phase::Prepared
                    && !self.backend.equivalent(&second, &guard.record.before)
                {
                    return ReconcileOutcome::Deferred;
                }
            }
            Err(_) => return ReconcileOutcome::Deferred,
        }

        if let Some(applied) = &guard.record.applied
            && self.backend.owns_current(applied, &second)
        {
            return ReconcileOutcome::StillOurs;
        }

        if let Some(applied) = &guard.record.applied
            && self.backend.matches_desired(&second, &guard.record.desired)
            && !self.backend.owns_current(applied, &second)
        {
            return ReconcileOutcome::Deferred;
        }

        if self.backend.equivalent(&second, &guard.record.before) {
            let desired = guard.record.desired.clone();
            let expected = second.clone();
            drop(guard);
            return self.apply_overlay(resource, entry, &expected, &desired, &expected);
        }

        drop(guard);
        if let Some(cooldown) = reconciler.breaker_gate(resource) {
            osdns_warn!(
                resource = %resource,
                "reconciliation circuit breaker is open; deferring for {:?}",
                cooldown
            );
            return ReconcileOutcome::Deferred;
        }
        self.rebase_transaction(resource, entry, &second)
    }

    fn apply_overlay(
        &self,
        resource: &ResourceId,
        entry: &Arc<Mutex<LiveRecord>>,
        expected: &PlatformSnapshot,
        desired: &NormalizedConfig,
        rollback_to: &PlatformSnapshot,
    ) -> ReconcileOutcome {
        let mut residue = crate::manager::MutationResidue::new();
        let identity = lock_live(entry).record.identity.clone();
        match self.mutate_and_verify(
            &identity,
            expected,
            desired,
            Some(rollback_to),
            INITIAL_POINTS,
            &mut residue,
        ) {
            Ok(mutation) => self.commit_applied(resource, entry, mutation),
            Err(_) => {
                if residue.leftover.is_some() {
                    lock_live(entry).verified = residue.leftover;
                }
                ReconcileOutcome::Deferred
            }
        }
    }

    #[allow(unused_variables)]
    fn commit_applied(
        &self,
        resource: &ResourceId,
        entry: &Arc<Mutex<LiveRecord>>,
        mutation: crate::platform::VerifiedMutation,
    ) -> ReconcileOutcome {
        let persisted = mutation.persist();
        let mut live = lock_live(entry);
        live.record.applied = Some(persisted.clone());
        live.record.phase = Phase::Applied;
        match self.journal.write(&live.record) {
            Ok(()) => {
                live.verified = None;
                drop(live);
                let _ = self.fire(TxPoint::AfterApplied);
                ReconcileOutcome::Rebased
            }
            Err(error) => {
                osdns_warn!(
                    resource = %resource,
                    error = %error,
                    "mutation was verified but the Applied journal write failed; deferring"
                );
                live.verified = mutation
                    .proof
                    .or(Some(crate::platform::OwnershipProof::issued(persisted)));
                live.record.applied = None;
                live.record.phase = Phase::Prepared;
                ReconcileOutcome::Deferred
            }
        }
    }

    #[allow(unused_variables)]
    fn rebase_transaction(
        &self,
        resource: &ResourceId,
        entry: &Arc<Mutex<LiveRecord>>,
        external_base: &PlatformSnapshot,
    ) -> ReconcileOutcome {
        self.suppressions.suppress(resource);
        {
            let mut live = lock_live(entry);
            let old = live.record.clone();
            let old_verified = live.verified.clone();
            live.record.before = external_base.clone();
            live.record.applied = None;
            live.record.phase = Phase::Prepared;
            live.verified = None;
            if let Err(error) = self.journal.write(&live.record) {
                osdns_warn!(
                    resource = %resource,
                    error = %error,
                    "rebase could not persist the Prepared record; keeping the previous journal state"
                );
                live.record = old;
                live.verified = old_verified;
                let _ = self.journal.write(&live.record);
                return ReconcileOutcome::Deferred;
            }
        }
        if self.fire(TxPoint::AfterPrepared).is_err() {
            return ReconcileOutcome::Deferred;
        }

        let desired = lock_live(entry).record.desired.clone();
        self.apply_overlay(resource, entry, external_base, &desired, external_base)
    }
}

fn lock_live(entry: &Arc<Mutex<LiveRecord>>) -> std::sync::MutexGuard<'_, LiveRecord> {
    entry
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

/// Spawns the reconciliation worker for an Enforce-policy manager and returns
/// the feed used to enqueue resources from watcher events.
///
/// The worker coalesces pending resources and processes each as soon as it is
/// due; events that arrive during a defer window update the pending entry
/// instead of being dropped.
pub(crate) fn spawn_reconciler(inner: Arc<Inner>) -> Result<mpsc::Sender<ResourceId>, Error> {
    let kind = inner.backend.kind();
    let (tx, rx) = mpsc::channel::<ResourceId>();
    thread::Builder::new()
        .name("osdns-reconciler".to_string())
        .spawn(move || {
            loop {
                let now = Instant::now();
                let due: Vec<ResourceId> = {
                    let pending = inner
                        .reconciler
                        .pending
                        .lock()
                        .unwrap_or_else(|poisoned| poisoned.into_inner());
                    pending
                        .iter()
                        .filter(|(_, p)| p.ready_at <= now)
                        .map(|(k, _)| k.clone())
                        .collect()
                };
                if due.is_empty() {
                    let timeout = {
                        let pending = inner
                            .reconciler
                            .pending
                            .lock()
                            .unwrap_or_else(|poisoned| poisoned.into_inner());
                        let now = Instant::now();
                        pending
                            .values()
                            .map(|p| p.ready_at.saturating_duration_since(now))
                            .min()
                    };
                    match rx.recv_timeout(timeout.unwrap_or(Duration::from_secs(3600))) {
                        Ok(resource) => inner.reconciler.touch(resource),
                        Err(RecvTimeoutError::Timeout) => {}
                        Err(RecvTimeoutError::Disconnected) => break,
                    }
                    continue;
                }
                for resource in due {
                    if inner.enforce_parked() {
                        continue;
                    }
                    inner.reconcile_resource(&resource, &inner.reconciler);
                }
            }
        })
        .map_err(|e| Error::Platform {
            backend: kind,
            message: format!("cannot spawn reconciler thread: {e}"),
        })?;
    Ok(tx)
}