frame_cli/dev/

session.rs

//! The dev session loop: watcher events → classification → coalescing →
//! builds → candidate pushes, around ONE wait.
//!
//! Ruling B, structurally: when no quiet deadline is armed the loop waits
//! in a BLOCKING `recv()` — zero-edit idle admits zero timer work. The
//! only bounded wait in the loop is the semantic edit-quiet deadline,
//! armed exclusively at the [`TimerChoke`] — the named debounce choke
//! point — in response to an observed relevant event, and its expiry
//! starts already-known dirty work through the coalescer; nothing here
//! inspects the filesystem to discover whether work exists. The
//! `cfg(test)` counter on the choke is R6's zero-edit idle authority.

use std::path::PathBuf;
use std::sync::mpsc::{Receiver, RecvTimeoutError};
use std::time::{Duration, Instant};

use crate::dev::build::BuildError;
use crate::dev::debounce::{BuildGeneration, Coalescer, Directive};
use crate::dev::watch::{Classification, WatchedSet};

/// Everything that can wake the session loop. The watcher callback and
/// the build worker send these; the loop is the only receiver.
pub enum LoopInput {
    /// One watcher event (or the backend's error, which classifies as
    /// desynchronization).
    Fs(notify::Result<notify::Event>),
    /// A candidate build finished (successfully or not).
    BuildCompleted {
        /// The build generation that completed.
        generation: BuildGeneration,
        /// The build verdict; failures carry verbatim diagnostics.
        result: Result<(), BuildError>,
    },
    /// The node child exited or its connection died (S3): the loop stops
    /// watching and reports NODE FAILED.
    NodeFailed {
        /// Exit/tombstone/connection evidence, rendered.
        detail: String,
    },
    /// Ctrl-C / SIGTERM: the operator asked for the ordered teardown.
    Interrupted,
}

/// The effects the loop commands; the real implementation snapshots,
/// builds, and pushes over S1 — tests record and script. Every method is
/// commanded by exactly one [`Directive`] (or the halt paths).
pub trait SessionEffects {
    /// Launch one build of the CURRENT content at this generation; its
    /// completion arrives as [`LoopInput::BuildCompleted`].
    fn start_build(&mut self, generation: BuildGeneration);
    /// A completed build was superseded before activation; drop its
    /// candidate.
    fn discard_stale(&mut self, generation: BuildGeneration);
    /// Push the completed candidate through S1 and report the outcome.
    ///
    /// # Errors
    ///
    /// The node connection is gone (the S3 fate), rendered.
    fn push_candidate(&mut self, generation: BuildGeneration) -> Result<(), String>;
    /// A build failed: report `RELOAD FAILED — RUNNING LAST GOOD` with
    /// the verbatim diagnostics.
    fn report_build_failure(&mut self, generation: BuildGeneration, error: &BuildError);
    /// The watcher lost sync: re-establish the subscription after the
    /// forced snapshot rebuild the coalescer already scheduled.
    ///
    /// # Errors
    ///
    /// Re-subscription failed: watching halts loudly, node on last good.
    fn resubscribe(&mut self) -> Result<(), String>;
}

/// Why the session ended. Every variant is terminal and loud.
#[derive(Debug, PartialEq, Eq)]
pub enum SessionExit {
    /// The node failed (child exit / connection death): typed report,
    /// watcher stopped, nonzero exit — never auto-restart.
    NodeFailed {
        /// Exit/tombstone/connection evidence.
        detail: String,
    },
    /// A watched root disappeared (deleted or moved out): watching cannot
    /// continue; the path and recovery are named.
    WatchRootGone {
        /// The path that disappeared.
        path: PathBuf,
    },
    /// Watcher re-subscription after a desynchronization failed: watching
    /// halts loudly while the node serves last good.
    ResubscribeFailed {
        /// The typed failure, rendered.
        detail: String,
    },
    /// Ctrl-C / SIGTERM: ordered teardown, exit zero.
    Interrupted,
    /// Every input sender is gone (teardown).
    InputsClosed,
}

/// THE timer-admission choke point: the only place the loop converts an
/// [`Directive::ArmQuietDeadline`] into a real bounded wait. The counter
/// is the R6 zero-edit-idle authority: established RUNNING CURRENT, held
/// idle, it reads zero.
#[derive(Default)]
pub struct TimerChoke {
    armed_until: Option<Instant>,
    /// How many times a deadline was ever armed (test observability).
    #[cfg(test)]
    pub armed_count: usize,
}

impl TimerChoke {
    fn arm(&mut self, deadline: Instant) {
        self.armed_until = Some(deadline);
        #[cfg(test)]
        {
            self.armed_count += 1;
        }
    }

    fn disarm(&mut self) {
        self.armed_until = None;
    }
}

/// One dev session's loop state.
pub struct Session<E: SessionEffects> {
    set: WatchedSet,
    coalescer: Coalescer,
    choke: TimerChoke,
    effects: E,
}

impl<E: SessionEffects> Session<E> {
    /// A session over the given watched set and quiet window.
    pub fn new(set: WatchedSet, quiet: Duration, effects: E) -> Self {
        Self {
            set,
            coalescer: Coalescer::new(quiet),
            choke: TimerChoke::default(),
            effects,
        }
    }

    /// The effects, for post-run inspection in tests.
    pub fn effects(&self) -> &E {
        &self.effects
    }

    /// The choke, for the zero-edit-idle witness.
    #[cfg(test)]
    pub fn choke(&self) -> &TimerChoke {
        &self.choke
    }

    /// Runs the loop until a terminal exit. See the module doc for the
    /// wait discipline.
    pub fn run(&mut self, inputs: &Receiver<LoopInput>) -> SessionExit {
        loop {
            let input = match self.choke.armed_until {
                // Idle or building with nothing armed: BLOCK. No timer
                // exists here.
                None => match inputs.recv() {
                    Ok(input) => input,
                    Err(_) => return SessionExit::InputsClosed,
                },
                // THE semantic quiet deadline, and nothing else.
                Some(deadline) => {
                    let now = Instant::now();
                    if now >= deadline {
                        self.choke.disarm();
                        let directive = self.coalescer.quiet_deadline_elapsed(now);
                        self.execute(directive);
                        continue;
                    }
                    match inputs.recv_timeout(deadline - now) {
                        Ok(input) => input,
                        Err(RecvTimeoutError::Timeout) => {
                            self.choke.disarm();
                            let directive = self.coalescer.quiet_deadline_elapsed(Instant::now());
                            self.execute(directive);
                            continue;
                        }
                        Err(RecvTimeoutError::Disconnected) => return SessionExit::InputsClosed,
                    }
                }
            };
            if let Some(exit) = self.handle(input) {
                return exit;
            }
        }
    }

    fn handle(&mut self, input: LoopInput) -> Option<SessionExit> {
        match input {
            LoopInput::Fs(Ok(event)) => self.handle_fs(&event),
            // A backend ERROR is loss: the backend no longer promises
            // completeness — same recovery as the rescan flag.
            LoopInput::Fs(Err(_error)) => self.handle_desync(),
            LoopInput::BuildCompleted { generation, result } => {
                if let Err(error) = &result {
                    self.effects.report_build_failure(generation, error);
                }
                let succeeded = result.is_ok();
                let [first, second] = self.coalescer.build_completed(generation, Instant::now());
                for directive in [first, second] {
                    // A failed build's candidate must never be pushed:
                    // Promote becomes the failure report already made.
                    if matches!(directive, Directive::PromoteCandidate(_)) && !succeeded {
                        continue;
                    }
                    if let Some(exit) = self.execute_terminal(directive) {
                        return Some(exit);
                    }
                }
                None
            }
            LoopInput::NodeFailed { detail } => Some(SessionExit::NodeFailed { detail }),
            LoopInput::Interrupted => Some(SessionExit::Interrupted),
        }
    }

    fn handle_fs(&mut self, event: &notify::Event) -> Option<SessionExit> {
        match self.set.classify(event) {
            Classification::Irrelevant => None,
            Classification::Dirtying => {
                let directive = self.coalescer.relevant_event(Instant::now());
                self.execute(directive);
                None
            }
            Classification::RootAffected(path) => {
                // The ONE event-driven existence check: an atomic editor
                // replace leaves the path present (dirtying); a true
                // disappearance is the typed halted state.
                if path.exists() {
                    let directive = self.coalescer.relevant_event(Instant::now());
                    self.execute(directive);
                    None
                } else {
                    Some(SessionExit::WatchRootGone { path })
                }
            }
            Classification::Desynchronized => self.handle_desync(),
        }
    }

    fn handle_desync(&mut self) -> Option<SessionExit> {
        let directive = self.coalescer.watcher_desynchronized(Instant::now());
        self.execute(directive);
        match self.effects.resubscribe() {
            Ok(()) => None,
            Err(detail) => Some(SessionExit::ResubscribeFailed { detail }),
        }
    }

    fn execute(&mut self, directive: Directive) {
        let _ = self.execute_terminal(directive);
    }

    fn execute_terminal(&mut self, directive: Directive) -> Option<SessionExit> {
        match directive {
            Directive::None => None,
            Directive::ArmQuietDeadline(deadline) => {
                self.choke.arm(deadline);
                None
            }
            Directive::StartBuild(generation) => {
                self.effects.start_build(generation);
                None
            }
            Directive::DiscardStale(generation) => {
                self.effects.discard_stale(generation);
                None
            }
            Directive::PromoteCandidate(generation) => {
                match self.effects.push_candidate(generation) {
                    Ok(()) => None,
                    Err(detail) => Some(SessionExit::NodeFailed { detail }),
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::expect_used)]

    use super::{LoopInput, Session, SessionEffects, SessionExit};
    use crate::dev::build::BuildError;
    use crate::dev::debounce::BuildGeneration;
    use crate::dev::watch::WatchedSet;
    use notify::EventKind;
    use notify::event::{CreateKind, Event};
    use std::path::{Path, PathBuf};
    use std::sync::mpsc;
    use std::time::Duration;

    const QUIET: Duration = Duration::from_millis(20);

    #[derive(Default)]
    struct Recording {
        started: Vec<u64>,
        discarded: Vec<u64>,
        pushed: Vec<u64>,
        failures: Vec<u64>,
        resubscribes: usize,
    }

    impl SessionEffects for Recording {
        fn start_build(&mut self, generation: BuildGeneration) {
            self.started.push(generation.0);
        }
        fn discard_stale(&mut self, generation: BuildGeneration) {
            self.discarded.push(generation.0);
        }
        fn push_candidate(&mut self, generation: BuildGeneration) -> Result<(), String> {
            self.pushed.push(generation.0);
            Ok(())
        }
        fn report_build_failure(&mut self, generation: BuildGeneration, _error: &BuildError) {
            self.failures.push(generation.0);
        }
        fn resubscribe(&mut self) -> Result<(), String> {
            self.resubscribes += 1;
            Ok(())
        }
    }

    fn session() -> Session<Recording> {
        Session::new(
            WatchedSet::of_project(Path::new("/proj")),
            QUIET,
            Recording::default(),
        )
    }

    fn dirtying_event() -> LoopInput {
        LoopInput::Fs(Ok(Event::new(EventKind::Create(CreateKind::File))
            .add_path(PathBuf::from("/proj/component/src/app.gleam"))))
    }

    fn build_done(generation: u64) -> LoopInput {
        LoopInput::BuildCompleted {
            generation: BuildGeneration(generation),
            result: Ok(()),
        }
    }

    /// One edit: the deadline arms once, one build starts on its expiry,
    /// the clean completion is pushed. End to end through the real wait.
    #[test]
    fn one_edit_one_build_one_push() {
        let (sender, receiver) = mpsc::channel();
        let mut session = session();
        sender.send(dirtying_event()).expect("send");
        std::thread::spawn(move || {
            // Give the quiet deadline room to expire, then complete the
            // build the expiry started, then hang up (loop exits).
            std::thread::sleep(Duration::from_millis(60));
            let _ = sender.send(build_done(1));
        });
        let exit = session.run(&receiver);
        assert_eq!(exit, SessionExit::InputsClosed);
        assert_eq!(session.effects().started, vec![1]);
        assert_eq!(session.effects().pushed, vec![1]);
        assert_eq!(session.choke().armed_count, 1, "one edit, one arm");
    }

    /// Zero-edit idle: inputs close without any event — the choke was
    /// never armed. (The loop sat in a BLOCKING recv; the counter is the
    /// authority that no timer work was admitted.)
    #[test]
    fn zero_edit_idle_arms_zero_timers() {
        let (sender, receiver) = mpsc::channel();
        let mut session = session();
        drop(sender);
        let exit = session.run(&receiver);
        assert_eq!(exit, SessionExit::InputsClosed);
        assert_eq!(
            session.choke().armed_count,
            0,
            "idle admits ZERO timer work"
        );
        assert!(session.effects().started.is_empty());
    }

    /// A failed build reports diagnostics and is NEVER pushed; the next
    /// clean generation follows the ordinary path (R4's recovery shape).
    #[test]
    fn failed_build_reports_and_never_pushes() {
        let (sender, receiver) = mpsc::channel();
        let mut session = session();
        sender.send(dirtying_event()).expect("send");
        std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(60));
            let _ = sender.send(LoopInput::BuildCompleted {
                generation: BuildGeneration(1),
                result: Err(BuildError::MissingBeam {
                    path: PathBuf::from("/x"),
                }),
            });
        });
        let exit = session.run(&receiver);
        assert_eq!(exit, SessionExit::InputsClosed);
        assert_eq!(session.effects().failures, vec![1]);
        assert!(session.effects().pushed.is_empty(), "failures never push");
    }

    /// A deleted watch root is the typed halted exit naming the path; an
    /// atomic replace (path still present) is ordinary dirtying. Uses a
    /// real scratch path for the existence check.
    #[test]
    fn missing_root_halts_and_present_root_dirties() {
        let (sender, receiver) = mpsc::channel();
        let mut session = session();
        let gone = PathBuf::from("/proj/component/src");
        sender
            .send(LoopInput::Fs(Ok(Event::new(EventKind::Remove(
                notify::event::RemoveKind::Folder,
            ))
            .add_path(gone.clone()))))
            .expect("send");
        let exit = session.run(&receiver);
        assert_eq!(exit, SessionExit::WatchRootGone { path: gone });
    }

    /// Node failure is terminal, typed, and never restarts anything.
    #[test]
    fn node_failure_is_terminal() {
        let (sender, receiver) = mpsc::channel();
        let mut session = session();
        sender
            .send(LoopInput::NodeFailed {
                detail: "child exited: signal 9".to_owned(),
            })
            .expect("send");
        let exit = session.run(&receiver);
        assert_eq!(
            exit,
            SessionExit::NodeFailed {
                detail: "child exited: signal 9".to_owned()
            }
        );
        assert!(session.effects().started.is_empty());
    }

    /// Watcher loss forces the immediate snapshot rebuild and one
    /// re-subscription; a failing re-subscription halts loudly.
    #[test]
    fn desync_rebuilds_and_resubscribes() {
        let (sender, receiver) = mpsc::channel();
        let mut session = session();
        sender
            .send(LoopInput::Fs(Ok(
                Event::new(EventKind::Any).set_flag(notify::event::Flag::Rescan)
            )))
            .expect("send");
        drop(sender);
        let exit = session.run(&receiver);
        assert_eq!(exit, SessionExit::InputsClosed);
        assert_eq!(session.effects().started, vec![1], "forced snapshot build");
        assert_eq!(session.effects().resubscribes, 1);
    }
}