liminal-server 0.23.1

use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex, Weak};
use std::thread::JoinHandle;
use std::time::Instant;

use super::handler::{ProductionParticipantHandler, elapsed_ms};
use crate::server::participant::ParticipantSemanticError;

#[derive(Debug, Default)]
pub(super) struct WarmWorker(Mutex<WorkerState>);

#[derive(Debug, Default)]
struct WorkerState {
    started: bool,
    stopped: bool,
    thread: Option<JoinHandle<()>>,
    failure: Option<String>,
}

impl WarmWorker {
    fn start(
        &self,
        handler: Weak<ProductionParticipantHandler>,
    ) -> Result<(), ParticipantSemanticError> {
        let mut worker = self
            .0
            .lock()
            .map_err(|_| worker_error("participant warm start lock poisoned"))?;
        if worker.started || worker.stopped {
            return Err(worker_error("participant warm already started"));
        }
        worker.started = true;
        worker.thread = Some(
            std::thread::Builder::new()
                .name("liminal-participant-warm".to_owned())
                .spawn(move || warm_in_background(&handler))
                .map_err(|error| {
                    worker_error(&format!("participant warm spawn failed: {error}"))
                })?,
        );
        drop(worker);
        Ok(())
    }

    fn join(&self) -> Result<(), ParticipantSemanticError> {
        let mut worker = self
            .0
            .lock()
            .map_err(|_| worker_error("participant warm join lock poisoned"))?;
        worker.stopped = true;
        if let Some(failure) = &worker.failure {
            return Err(worker_error(failure));
        }
        if let Some(thread) = worker.thread.take()
            && thread.join().is_err()
        {
            let message = "participant warm panicked during shutdown";
            worker.failure = Some(message.to_owned());
            return Err(worker_error(message));
        }
        drop(worker);
        Ok(())
    }
}

fn worker_error(message: &str) -> ParticipantSemanticError {
    ParticipantSemanticError::Internal {
        message: message.to_owned(),
    }
}

/// The server's hold on the participant warm: stopping it ends the warm at
/// its next conversation boundary; joining releases its active store owner.
#[derive(Clone, Debug)]
pub(in crate::server) struct ParticipantWarm {
    handler: Weak<ProductionParticipantHandler>,
    worker: Arc<WarmWorker>,
}

impl ParticipantWarm {
    pub(in crate::server) fn join(&self) -> Result<(), ParticipantSemanticError> {
        self.stop();
        self.worker.join()
    }

    /// Stops the warm. A conversation already restoring finishes its unit,
    /// and every touch waiting on that unit is woken and answered at once.
    pub(in crate::server) fn stop(&self) {
        if let Some(handler) = self.handler.upgrade() {
            handler.warm_stop.store(true, Ordering::SeqCst);
            if let Err(error) = handler.pending.wake_waiters() {
                tracing::error!(
                    error = %error,
                    "a server stop could not wake the touches waiting on a restore"
                );
            }
        }
    }

    /// Finishes the warm on the caller's thread and answers how many
    /// conversations refused. Each conversation restores once behind its own
    /// gate, so the store work is the same however the warm thread and this
    /// call split it; a measurement that must not race the warm calls this
    /// before it reads.
    ///
    /// # Errors
    ///
    /// Only a poisoned pending-restore lock.
    pub(in crate::server) fn finish(&self) -> Result<usize, ParticipantSemanticError> {
        self.handler.upgrade().map_or(Ok(0), |handler| {
            handler.restore_remaining("a measurement waiting for the warm")
        })
    }

    /// The clean stop's capacity snapshot (DUR-020 R3), written after the
    /// connections have closed and before the final flush.
    pub(in crate::server) fn record_clean_stop(&self) {
        if let Some(handler) = self.handler.upgrade() {
            handler.record_clean_stop_capacity();
        }
    }
}

impl ProductionParticipantHandler {
    /// Gives the handler its own weak reference, which the warm thread
    /// holds, and answers the server's stop handle for that warm.
    ///
    /// # Errors
    ///
    /// Refuses a second installation by name.
    pub(in crate::server) fn install_warm(
        self: &Arc<Self>,
    ) -> Result<ParticipantWarm, ParticipantSemanticError> {
        self.self_ref.set(Arc::downgrade(self)).map_err(|_| {
            ParticipantSemanticError::Internal {
                message: "participant warm was already installed".to_owned(),
            }
        })?;
        Ok(ParticipantWarm {
            handler: Arc::downgrade(self),
            worker: Arc::clone(&self.warm_worker),
        })
    }

    /// Starts the warm on its own thread, once the incarnation's repair is
    /// owed, so every conversation it restores is repaired by its own unit.
    /// The listener never waits for it. A handler with no installed warm (a
    /// test that builds the handler alone) starts none: its conversations
    /// restore on their first touch.
    pub(super) fn start_warm(&self) {
        let Some(handler) = self.self_ref.get().cloned() else {
            return;
        };
        if let Err(error) = self.warm_worker.start(handler) {
            tracing::error!(%error, "participant warm thread could not start");
        }
    }
}

/// The warm loop: restores each pending conversation in id order, holding the
/// handler only for one conversation at a time, until none is pending, the
/// server stops it, or the server drops the handler.
fn warm_in_background(handler: &Weak<ProductionParticipantHandler>) {
    let Some(owner) = handler.upgrade() else {
        return;
    };
    let remaining = match owner.pending.remaining() {
        Ok(remaining) => remaining,
        Err(error) => {
            tracing::error!(error = %error, "participant warm could not list pending conversations");
            return;
        }
    };
    drop(owner);
    let total = remaining.len();
    let started = Instant::now();
    tracing::info!(
        step = "participant warm",
        conversations = total,
        "participant warm started"
    );
    let mut refused = Vec::new();
    for (index, conversation_id) in remaining.into_iter().enumerate() {
        let Some(owner) = handler.upgrade() else {
            return;
        };
        if owner.warm_stop.load(Ordering::SeqCst) {
            tracing::info!(
                step = "participant warm",
                done = index,
                total,
                elapsed_ms = elapsed_ms(started),
                "participant warm stopped with the server; the rest restore on the next start"
            );
            return;
        }
        if let Err(refusal) = owner.ensure_restored(conversation_id) {
            refused.push((conversation_id, refusal.to_string()));
        }
    }
    tracing::info!(
        step = "participant warm",
        conversations = total,
        unloadable = refused.len(),
        refused = ?refused,
        elapsed_ms = elapsed_ms(started),
        "participant warm finished"
    );
    verify_restored_histories(handler);
}

/// After the warm: cross-checks every restored conversation's identities
/// against its checkpoint's history index, the check the restore leaves off
/// its path so that it reads the same at any history length (Waffles, 11
/// October 09:14: nothing is skipped, only moved). A mismatch is logged by
/// name and counted, and the conversation keeps being served. A root
/// verified on an earlier boot is recorded, and only what changed since it
/// is read (Waffles, 12 October 07:46: the full check re-read 50,011
/// identities in 164 s on every boot, growing with the history). The
/// finished line names how many identities were covered, how many were read
/// from the index, and which conversations disagreed.
fn verify_restored_histories(handler: &Weak<ProductionParticipantHandler>) {
    let Some(owner) = handler.upgrade() else {
        return;
    };
    let restored = match owner.restored_conversation_ids() {
        Ok(restored) => restored,
        Err(error) => {
            tracing::error!(error = %error, "participant history check could not list restored conversations");
            return;
        }
    };
    drop(owner);
    let started = Instant::now();
    let mut identities = 0_usize;
    let mut read = 0_usize;
    let mut mismatched = Vec::new();
    for conversation_id in restored {
        let Some(owner) = handler.upgrade() else {
            return;
        };
        if owner.warm_stop.load(Ordering::SeqCst) {
            return;
        }
        let checked = match owner.verify_identity_history(conversation_id) {
            Ok(checked) => checked,
            Err(error) => {
                // Logged by name and counted, never contained (Waffles' ruling
                // of 11 October 09:14): on 11 October a705852d contained the
                // live room on this check while the restore itself was sound.
                tracing::error!(
                    step = "participant history check",
                    conversation_id = ?conversation_id,
                    error = %error,
                    "a restored conversation's identities disagree with its checkpoint's history index"
                );
                mismatched.push(conversation_id);
                continue;
            }
        };
        identities = identities.saturating_add(checked.identities);
        read = read.saturating_add(checked.checked);
    }
    tracing::info!(
        step = "participant history check",
        identities,
        read,
        mismatches = mismatched.len(),
        mismatched = ?mismatched,
        elapsed_ms = elapsed_ms(started),
        "history cross-check finished"
    );
}