tephra 0.4.0

//! Reading a store that another process is writing.
//!
//! A [`Follower`] opens a data directory read-only and advances over the writer's committed
//! prefix on demand. It mutates nothing: it creates no directory, deletes no unfinished
//! segment, writes no `.idx`, and opens every file `O_RDONLY`, so the directory may be a
//! read-only mount and needs no write permission.
//!
//! # Why this is safe against a live writer
//!
//! The on-disk format was already built for it. Three properties carry the argument:
//!
//! - **Nothing is mutated in place or deleted.** There is no compaction, no retention, no
//!   free list and no tombstones, so a committed byte never changes and a follower's cached
//!   offsets and open descriptors stay valid indefinitely.
//! - **The commit rule is durable and shared.** Each batch ends in a commit marker, and a run
//!   counts only if every record in it validates by CRC *and* the run terminates in that
//!   marker. A follower applies the identical walk that crash recovery does, so it can never
//!   expose a record the writer would have rolled back.
//! - **The index is derived.** A `.idx` read while the writer is rewriting it fails its body
//!   CRC and is rebuilt from the log in memory, so no coordination is needed for it.
//!
//! Together these mean a follower always sees a **prefix**: gap-free, duplicate-free, and
//! only growing. It never sees a torn record or an uncommitted batch.
//!
//! # What a follower does not guarantee
//!
//! - **It is not a durability oracle.** A batch becomes visible when its commit marker
//!   reaches the page cache, which is just before the writer's `fsync` returns. A power loss
//!   in that window erases events a follower has already read. Closing the window would cost
//!   the writer a second fsync per batch on the path that is already its throughput ceiling.
//!   Treat a follower like an asynchronous replica: do not let it be the last word on whether
//!   something happened.
//! - **It lags.** The tip trails the writer by up to one [`refresh`](Follower::refresh)
//!   interval plus the scan.
//! - **It is same-host only.** Visibility relies on the writer's `write` reaching a shared
//!   page cache. That holds for a local filesystem; it does not hold for a network mount.
//! - **One writer, still.** A follower takes no lock, by design: a shared lock would conflict
//!   with the writer's exclusive one, so a follower that started first would block the writer.
//!   Correctness comes from the commit markers, not from exclusion.
//!
//! In-process, prefer the coordinator's own [`ReadHandle`]: it shares the writer's snapshot
//! with no lag and no second scan. A `Follower` is for the case where the writer is somewhere
//! else entirely.

use std::path::{Path, PathBuf};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Duration;

use thiserror::Error;

use crate::Position;
use crate::index::{IndexError, IndexSet};
use crate::log::set::{LogError, SegmentConfig, SegmentSet};
use crate::read::{ReadConfig, ReadCore, ReadHandle, Snapshot};

/// How to open a [`Follower`].
#[derive(Clone, Copy, Debug)]
pub struct FollowerConfig {
    /// Segment geometry. Only `header_size` is used: each segment's real size comes from the
    /// file, which is authoritative because segments are `fallocate`d at creation and never
    /// extended. A follower whose configured `segment_size` disagreed with the writer's would
    /// otherwise stop its scan mid-segment.
    pub segment: SegmentConfig,
    /// Read-path tuning for the handles this follower hands out.
    pub read: ReadConfig,
}

impl FollowerConfig {
    /// Defaults for a store whose segments use the standard header.
    pub fn new(segment: SegmentConfig) -> FollowerConfig {
        FollowerConfig {
            segment,
            read: ReadConfig::default(),
        }
    }
}

/// Why a follower could not open or advance.
#[derive(Debug, Error)]
pub enum FollowerError {
    #[error(transparent)]
    Log(#[from] LogError),
    #[error(transparent)]
    Index(#[from] IndexError),
}

/// The mutable half, behind one mutex so `refresh` can take `&self` and a background poller
/// and a manual caller share exactly one code path.
struct Inner {
    set: SegmentSet,
    index: IndexSet,
    /// The highest position fed into the index, which is **not** the log set's tip.
    ///
    /// Tracked separately because the two advance in separate fallible steps. Deriving the
    /// catch-up cursor from the set instead would mean that a `catch_up` failing after the set
    /// advanced skips that range forever: the retry would recompute the cursor from the
    /// already-advanced set and never feed the events in between, leaving them readable by
    /// scan but invisible to every index query for the life of the process.
    fed_through: Position,
}

/// A read-only view of a store another process is writing.
///
/// Reads go through an ordinary [`ReadHandle`], so queries, backward reads and subscriptions
/// all behave exactly as they do against a writer. See the [module docs](self) for what a
/// follower does and does not guarantee.
pub struct Follower {
    inner: Mutex<Inner>,
    core: Arc<ReadCore>,
    read_config: ReadConfig,
    dir: PathBuf,
}

impl Follower {
    /// Opens `dir` read-only and catches up to whatever is committed there now.
    ///
    /// A directory with no readable segment is [`LogError::Uninitialized`], not an empty
    /// follower: a follower that raced the writer's very first segment would otherwise report
    /// "no events" for a store that is merely not ready yet, and on this layer an error must
    /// never look like end-of-stream. Retry until the writer has initialized the store.
    pub fn open(dir: impl AsRef<Path>, config: FollowerConfig) -> Result<Follower, FollowerError> {
        let set = SegmentSet::open_read_only(&dir, config.segment)?;
        let index = IndexSet::open_read_only(&set)?;
        let core = ReadCore::new(&set, &index);
        let fed_through = set.last_position();
        Ok(Follower {
            dir: set.dir().to_path_buf(),
            inner: Mutex::new(Inner {
                set,
                index,
                fed_through,
            }),
            core,
            read_config: config.read,
        })
    }

    /// A handle for reading at the follower's current tip.
    ///
    /// Cheap, and it never contends with a refresh: reads run on the caller's thread over the
    /// snapshot the last refresh published.
    pub fn reader(&self) -> ReadHandle {
        ReadHandle::new(Arc::clone(&self.core), self.read_config)
    }

    /// The last readable position.
    pub fn head(&self) -> Position {
        self.core.head()
    }

    /// The directory being followed.
    pub fn dir(&self) -> &Path {
        &self.dir
    }

    /// Advances to the writer's current committed prefix and returns the new tip.
    ///
    /// The publish order here mirrors the writer's own commit seam, and it is load-bearing:
    /// segments go out before the watermark, so a reader that observes the new watermark is
    /// guaranteed to already see the snapshot covering it.
    pub fn refresh(&self) -> Result<Position, FollowerError> {
        let mut guard = self.inner.lock().unwrap();
        let inner = &mut *guard;

        // The log may already be ahead of the index from a refresh that failed partway, so
        // every step below is driven by what it individually still owes, never by a tip
        // captured before the work started.
        let refreshed = inner.set.refresh()?;
        if inner.fed_through == refreshed.tip && self.core.head() == refreshed.tip {
            return Ok(refreshed.tip);
        }

        // Feed the index in position order, sealing at each boundary the log crossed. Only on
        // success does the cursor move, so a failure here is retried over the same range
        // rather than skipped.
        let index_sealed = inner.index.catch_up(&inner.set, inner.fed_through)?;
        inner.fed_through = refreshed.tip;

        // Republish if *either* side sealed. The writer can gate this on its own rollover
        // because there the index seals if and only if the log did; here they are separate
        // fallible steps, so an index seal can land in a refresh that saw no rollover, and
        // publishing a watermark against a snapshot that predates it would hand readers a tip
        // whose segments the snapshot does not contain.
        if refreshed.sealed_added > 0 || index_sealed {
            self.core
                .publish_segments(Snapshot::capture(&inner.set, &inner.index));
        }
        self.core.publish_watermark(refreshed.tip);
        self.core.wake();
        Ok(refreshed.tip)
    }

    /// Marks the store closed, so a parked [`Subscription`](crate::Subscription) observes the
    /// close and ends instead of waiting for a tip that will never move.
    pub fn close(&self) {
        self.core.close();
    }

    /// Refreshes on a background thread every `interval`.
    ///
    /// This is what makes subscriptions work against a follower: a subscription is already
    /// just repeated reads off an advancing watermark, and the poller is what advances it.
    /// Dropping the returned handle stops the thread and closes the follower.
    pub fn poll_every(self: &Arc<Self>, interval: Duration) -> FollowerPoller {
        let signal = Arc::new(Signal::default());
        let health = Arc::new(Mutex::new(PollerHealth::default()));

        let follower = Arc::clone(self);
        let thread_signal = Arc::clone(&signal);
        let thread_health = Arc::clone(&health);
        let join = thread::Builder::new()
            .name("tephra-follower".to_string())
            .spawn(move || {
                while !thread_signal.stopped() {
                    match follower.refresh() {
                        Ok(_) => {
                            let mut health = lock(&thread_health);
                            health.consecutive_failures = 0;
                            health.last_error = None;
                        }
                        Err(err) => {
                            // Most failures here are transient, typically a header the writer
                            // is in the middle of writing, so the loop keeps going rather than
                            // freezing every subscription on this follower. A permanent one
                            // (corruption mid-chain, a vanished segment) would otherwise be
                            // indistinguishable from an idle store, so it is recorded for
                            // `last_error` and `consecutive_failures` to surface.
                            #[cfg(feature = "tracing")]
                            tracing::warn!("follower refresh failed: {err}");
                            let mut health = lock(&thread_health);
                            health.consecutive_failures += 1;
                            health.last_error = Some(err.to_string());
                        }
                    }
                    thread_signal.wait(interval);
                }
            })
            .expect("spawn follower thread");

        FollowerPoller {
            signal,
            health,
            join: Some(join),
            follower: Arc::clone(self),
        }
    }
}

/// Recovers a mutex without propagating poisoning: a panic while reporting health must not
/// take the poller down with it.
fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
    mutex
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

/// Stop flag plus the condvar the poll loop waits on, so a stop is observed immediately
/// instead of after the current interval elapses.
#[derive(Default)]
struct Signal {
    stopped: Mutex<bool>,
    wake: Condvar,
}

impl Signal {
    fn stopped(&self) -> bool {
        *lock(&self.stopped)
    }

    fn stop(&self) {
        *lock(&self.stopped) = true;
        self.wake.notify_all();
    }

    /// Sleeps for at most `interval`, returning early once stopped.
    fn wait(&self, interval: Duration) {
        let guard = lock(&self.stopped);
        if *guard {
            return;
        }
        let _unused = self
            .wake
            .wait_timeout(guard, interval)
            .unwrap_or_else(|poisoned| poisoned.into_inner());
    }
}

/// What the background poller has been seeing.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct PollerHealth {
    /// Refreshes that have failed in a row. Zero after any success.
    pub consecutive_failures: u64,
    /// The most recent failure, or `None` if the last refresh succeeded.
    pub last_error: Option<String>,
    /// Set if the poll thread ended unexpectedly. Its subscriptions will never advance again.
    pub thread_died: bool,
}

/// A running background refresh. Dropping it stops the thread and closes the follower.
pub struct FollowerPoller {
    signal: Arc<Signal>,
    health: Arc<Mutex<PollerHealth>>,
    join: Option<JoinHandle<()>>,
    follower: Arc<Follower>,
}

impl FollowerPoller {
    /// What the poll loop has been seeing.
    ///
    /// A follower that cannot advance is otherwise indistinguishable from a store nobody is
    /// writing to: the tip simply stops moving and every subscription parks. Poll this to tell
    /// the two apart.
    pub fn health(&self) -> PollerHealth {
        let mut health = lock(&self.health).clone();
        health.thread_died = self
            .join
            .as_ref()
            .is_some_and(|join| join.is_finished() && !self.signal.stopped());
        health
    }

    /// Stops the thread and closes the follower, waking any parked subscription.
    pub fn stop(mut self) {
        self.shutdown();
    }

    fn shutdown(&mut self) {
        self.signal.stop();
        if let Some(join) = self.join.take()
            && join.join().is_err()
        {
            // The loop itself does not panic, so this means something below it did. Record it
            // rather than swallow it: the poller is gone either way.
            let mut health = lock(&self.health);
            health.thread_died = true;
        }
        // Without this a subscriber parked on the watermark waits forever: nothing else will
        // ever advance it once the poller is gone.
        self.follower.close();
    }
}

impl Drop for FollowerPoller {
    fn drop(&mut self) {
        self.shutdown();
    }
}