liminal-server 0.21.1

//! DUR-021: the spine's reclaim pass runs behind the open door. The store
//! opens with its reclaim held; the runtime releases it once the listeners
//! are bound; every pass ending is logged with its figures by a watch thread
//! that ends when the store closes; and what each shard resumed after an
//! unclean stop is on the log. Nothing here quiesces a pass: that is an
//! operator's act on the health route, never a boot's or a stop's.

use std::sync::{Arc, Mutex, PoisonError};
use std::time::Instant;

use haematite::db::ShardReclaimAnswer;
use haematite::{Database, EventStore, ReclaimPassOutcome};

/// The watch thread's name.
pub const WATCH_THREAD: &str = "liminal-spine-reclaim-watch";

/// The release thread's name.
pub const RELEASE_THREAD: &str = "liminal-spine-reclaim-release";

/// Releases the held reclaim ([`release`]) on a thread of its own, started
/// the moment the listeners are bound, and returns at once. The release asks
/// every shard for its status and waits for each answer, and a shard answers
/// only between two of its reclaim steps: on 12 October the seats' server
/// waited 1,625 ms for shard 4's one step (1,538 ms) before it logged
/// started and started its services. Nothing the start does needs the
/// answers, so nothing on the start path waits for them. The thread holds
/// the store for the release alone, at most one step per shard.
pub fn release_after_bind(spine: &Arc<EventStore>, began: Instant) {
    let spine = Arc::clone(spine);
    off_the_start_path(move || release(&spine, began));
}

/// Runs `work` on the release thread; a thread that cannot start runs it
/// here, said by name, so the release is never lost, only waited for.
pub fn off_the_start_path<F: FnOnce() + Send + 'static>(work: F) {
    let shared = Arc::new(Mutex::new(Some(work)));
    let taken = Arc::clone(&shared);
    let spawned = std::thread::Builder::new()
        .name(RELEASE_THREAD.to_owned())
        .spawn(move || run_once(&taken));
    if let Err(error) = spawned {
        tracing::warn!(
            error = %error,
            "spine reclaim release thread did not start; released on the start path"
        );
        run_once(&shared);
    }
}

/// Takes the work out and runs it, once, whichever side gets it.
fn run_once<F: FnOnce()>(work: &Mutex<Option<F>>) {
    let taken = work.lock().unwrap_or_else(PoisonError::into_inner).take();
    if let Some(work) = taken {
        work();
    }
}

/// Arms the pass watch at open with a capacity of two reports per shard and
/// drains it on a thread of its own that logs each ended pass. The thread
/// holds the receiver only, so it ends when the store closes and never holds
/// the store open. A refused watch is logged and the boot goes on: no pass
/// is held back by its own reporting.
pub fn arm_watch(database: &Database) {
    let capacity = database.shard_count().saturating_mul(2).max(1);
    let watch = match database.watch_reclaim_passes(capacity) {
        Ok(watch) => watch,
        Err(error) => {
            tracing::warn!(error = %error, "spine reclaim pass watch refused");
            return;
        }
    };
    let armed_at = Instant::now();
    let spawned = std::thread::Builder::new()
        .name(WATCH_THREAD.to_owned())
        .spawn(move || {
            while let Ok(report) = watch.recv() {
                let (outcome, failed_reason) = match &report.outcome {
                    ReclaimPassOutcome::Passed => ("passed", None),
                    ReclaimPassOutcome::Failed { reason } => ("failed", Some(reason.clone())),
                };
                tracing::info!(
                    shard = report.shard_id,
                    outcome,
                    passes_finished = report.passes_finished,
                    // Time since this watch was armed at boot, not what the pass
                    // took: haematite 0.18.5's report carries no start.
                    since_start_ms =
                        u64::try_from(armed_at.elapsed().as_millis()).unwrap_or(u64::MAX),
                    unmarked_bytes = report.unmarked_bytes,
                    retained_bytes = report.retained_bytes,
                    peak_bytes = report.peak,
                    ended_at_ms = report.ended_at_ms,
                    failed_steps = failed_reason,
                    "spine reclaim pass ended"
                );
            }
        });
    if let Err(error) = spawned {
        tracing::warn!(error = %error, "spine reclaim pass watch thread did not start");
    }
}

/// Releases the held reclaim once the listeners are bound (R1) and logs what
/// each shard resumed (R3). A second release is a no-op by the store's own
/// word. A refused release is logged by name and the server serves on: the
/// pass is owed, not lost, and the next open resumes it.
pub fn release(spine: &EventStore, began: Instant) {
    let database = spine.database();
    if let Err(error) = database.release_reclaim() {
        tracing::warn!(error = %error, "spine reclaim release refused");
        return;
    }
    let status = database.reclaim_status();
    let phases: Vec<String> = status
        .shards
        .iter()
        .map(|shard| match shard {
            ShardReclaimAnswer::Answered { shard, status } => {
                format!("{shard}:{:?}", status.phase)
            }
            ShardReclaimAnswer::NotAnswering { shard, reason } => {
                format!("{shard}:not-answering({reason})")
            }
        })
        .collect();
    tracing::info!(
        boot_elapsed_ms = u64::try_from(began.elapsed().as_millis()).unwrap_or(u64::MAX),
        phases = %phases.join(" "),
        "spine reclaim released"
    );
    for shard in &status.shards {
        if let ShardReclaimAnswer::Answered { shard, status } = shard {
            log_resume(*shard, status.resume.as_deref());
        }
    }
}

/// The note haematite gives a shard that took up a pass a stop left
/// (`reclaim_resume_taken: N marked, ...`). Any other note (a record found
/// corrupt, ignored or already completed) is not a resumed pass.
const RESUME_TAKEN: &str = "reclaim_resume_taken";

/// Says, for every shard, whether it took up a pass a stop left or starts
/// from nothing (R3), so an operator reads on the log that an unclean stop
/// lost no pass, and a refused record is never read as a resumed one.
fn log_resume(shard: usize, resume: Option<&str>) {
    match resume {
        Some(note) if note.starts_with(RESUME_TAKEN) => {
            tracing::info!(shard, resume = note, "spine reclaim resumed");
        }
        note => tracing::info!(
            shard,
            resume = note.unwrap_or("none"),
            "spine reclaim starts from nothing"
        ),
    }
}