osdns 0.1.3

//! 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
//! mutation, read-back, or verification failure the pass rolls back safely to
//! the new external base and defers; a crash between `Prepared` and `Applied`
//! makes recovery roll the overlay back to the external base, which is
//! exactly the never-overwrite-external-state invariant.
//!
//! 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::{JournalRecord, Phase};
use crate::lease::LiveRecord;
use crate::manager::{INITIAL_POINTS, Inner};
use crate::ownership::ResourceId;
use crate::platform::PlatformSnapshot;

/// 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);
/// 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,
    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 {
    /// 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);
    }

    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);
    }

    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 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::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) => {
                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) => {
                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());
        let record = &mut guard.record;

        // A pending transaction whose overlay is now in place can be finalized.
        if record.phase == Phase::Prepared && self.backend.matches_desired(&second, &record.desired)
        {
            record.applied = Some(second.clone());
            record.phase = Phase::Applied;
            match self.journal.write(record) {
                Ok(()) => return ReconcileOutcome::StillOurs,
                Err(error) => {
                    osdns_warn!(
                        resource = %resource,
                        error = %error,
                        "reconciliation could not finalize the pending transaction; deferring"
                    );
                    record.applied = None;
                    record.phase = Phase::Prepared;
                    return ReconcileOutcome::Deferred;
                }
            }
        }

        // Still ours: the event came from our own mutation.
        if let Some(applied) = &record.applied
            && self.backend.equivalent(&second, applied)
        {
            return ReconcileOutcome::StillOurs;
        }
        if self.backend.matches_desired(&second, &record.desired) {
            return ReconcileOutcome::StillOurs;
        }

        // Overlay missing but base intact: a plain reapply, no rebase needed.
        if self.backend.equivalent(&second, &record.before) {
            self.suppressions.suppress(resource);
            match self.mutate_and_verify(resource, &record.desired, Some(&second), INITIAL_POINTS) {
                Ok(actual) => {
                    record.applied = Some(actual);
                    record.phase = Phase::Applied;
                    match self.journal.write(record) {
                        Ok(()) => return ReconcileOutcome::Rebased,
                        Err(error) => {
                            osdns_warn!(
                                resource = %resource,
                                error = %error,
                                "reapply succeeded but the journal could not be updated; deferring"
                            );
                            return ReconcileOutcome::Deferred;
                        }
                    }
                }
                Err(_) => return ReconcileOutcome::Deferred,
            }
        }

        // External modification of the base: full transactional rebase.
        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, record, &second, reconciler)
    }

    /// Adopts `external_base` as the new base and reapplies the desired
    /// overlay as a full journal transaction:
    ///
    /// 1. persist/fsync `Prepared` with `before = external_base`;
    /// 2. apply the overlay;
    /// 3. read back and verify;
    /// 4. persist `Applied` with the verified snapshot.
    ///
    /// On any failure the pass rolls back to `external_base` and defers. A
    /// crash between steps 1 and 4 leaves a `Prepared` record whose recovery
    /// restores `external_base`, never the pre-rebase overlay.
    #[allow(unused_variables)]
    fn rebase_transaction(
        &self,
        resource: &ResourceId,
        record: &mut JournalRecord,
        external_base: &PlatformSnapshot,
        reconciler: &Reconciler,
    ) -> ReconcileOutcome {
        self.suppressions.suppress(resource);
        let old = record.clone();

        record.before = external_base.clone();
        record.applied = None;
        record.phase = Phase::Prepared;
        if let Err(error) = self.journal.write(record) {
            osdns_warn!(
                resource = %resource,
                error = %error,
                "rebase could not persist the Prepared record; keeping the previous journal state"
            );
            *record = old;
            let _ = self.journal.write(record);
            return ReconcileOutcome::Deferred;
        }
        if let Err(error) = self.fire(TxPoint::AfterPrepared) {
            let _ = error;
            return ReconcileOutcome::Deferred;
        }

        match self.mutate_and_verify(
            resource,
            &record.desired,
            Some(external_base),
            INITIAL_POINTS,
        ) {
            Ok(actual) => {
                record.applied = Some(actual);
                record.phase = Phase::Applied;
                match self.journal.write(record) {
                    Ok(()) => {}
                    Err(error) => {
                        // The OS has the overlay but the journal only
                        // guarantees the external base: mark uncommitted so
                        // the next pass finalizes via read-back.
                        osdns_warn!(
                            resource = %resource,
                            error = %error,
                            "rebase applied the overlay but could not persist Applied; deferring"
                        );
                        record.applied = None;
                        record.phase = Phase::Prepared;
                        reconciler.defer(resource, UNSTABLE_RETRY, false);
                        return ReconcileOutcome::Deferred;
                    }
                }
                if let Err(error) = self.fire(TxPoint::AfterApplied) {
                    let _ = error;
                }
                ReconcileOutcome::Rebased
            }
            Err(_) => {
                // Prepared is already durable. Keep the external base even
                // if rollback fails or its readback is unavailable. The OS
                // may still contain our overlay; restoring the old journal
                // would let recovery erase the newly adopted external base.
                ReconcileOutcome::Deferred
            }
        }
    }
}

/// 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 {
                    inner.reconcile_resource(&resource, &inner.reconciler);
                }
            }
        })
        .map_err(|e| Error::Platform {
            backend: kind,
            message: format!("cannot spawn reconciler thread: {e}"),
        })?;
    Ok(tx)
}