xtask/

collab.rs

//! Running the collaboration stack on this machine.
//!
//! Three commands, because they are wanted at three granularities:
//!
//! - `server` and `client` are the primitives — one process each, in the
//!   foreground, so they compose across terminals and take their defaults
//!   from each other (a bare `server` and a bare `client` find one another).
//!   In its own terminal the server also gets freeze and thaw for free:
//!   Ctrl-Z stops it, `fg` resumes it, no tooling involved. Nothing is
//!   spawned in the background, so nothing can be left behind.
//! - `demo` is both at once for people who do not want three terminals: a
//!   scratch log, N editors, and a console that stops and continues the
//!   server without also stopping the terminal it is typed in.
//!
//! Freezing is the point of all this. Two people editing at human speed
//! almost never hold unacknowledged commits at the same moment, so the
//! concurrent case — the one the server-centric model exists for — is the one
//! a live demo never reaches by accident. Stopping the server makes it
//! reachable on purpose.
//!
//! Freeze only once the editors say `live` in their titles: a stopped server
//! does not drop the connection, but it cannot complete a handshake either,
//! so freezing before they connect leaves them `[disconnected]` — and nothing
//! reconnects them, because resume-on-reconnect is not built yet (R3 in
//! `docs/collab-migration-playbook.md`).

use std::io::{BufRead, Write};
use std::net::{SocketAddr, TcpStream};
use std::path::PathBuf;
use std::process::{Child, Command};
use std::time::{Duration, Instant};

use anyhow::{Context, Result, bail};
use clap::Args;

use crate::{cargo, workspace_root};

/// Where a bare `server` listens and a bare `client` looks, so the two
/// compose with no flags at all.
const LISTEN: &str = "127.0.0.1:4100";

/// The log a bare `server` serves: disposable, ignored by git, and the same
/// one across runs so a document survives a restart.
fn default_db() -> PathBuf {
    workspace_root().join("target").join("blockworx.db")
}

fn ws_url(listen: &str) -> String {
    format!("ws://{listen}/ws")
}

/// Which build the demo spawns — it names both the cargo flag and the
/// directory the binaries are found in, so the two cannot disagree.
#[derive(Clone, Copy)]
enum Profile {
    Debug,
    Release,
}

impl From<bool> for Profile {
    fn from(release: bool) -> Self {
        if release {
            Profile::Release
        } else {
            Profile::Debug
        }
    }
}

impl Profile {
    fn dir(self) -> &'static str {
        match self {
            Profile::Debug => "debug",
            Profile::Release => "release",
        }
    }

    fn flag(self) -> Option<&'static str> {
        matches!(self, Profile::Release).then_some("--release")
    }
}

/// Whether an editor narrates its document mutations on the console.
#[derive(Clone, Copy)]
enum Narration {
    Quiet,
    Narrating,
}

impl From<bool> for Narration {
    fn from(narrate: bool) -> Self {
        if narrate {
            Narration::Narrating
        } else {
            Narration::Quiet
        }
    }
}

impl Narration {
    fn filter(self) -> Option<&'static str> {
        matches!(self, Narration::Narrating).then_some("blockworx=info,edit=debug")
    }
}

#[derive(Args)]
pub struct ServerArgs {
    /// Log to serve. Created if missing; kept across runs. Defaults to
    /// `target/blockworx.db`.
    #[arg(long)]
    db: Option<PathBuf>,

    /// Address to listen on.
    #[arg(long, default_value = LISTEN)]
    listen: String,

    /// Build and run optimized.
    #[arg(long)]
    release: bool,
}

#[derive(Args)]
pub struct ClientArgs {
    /// Server to edit against. Defaults to a `server` on this machine.
    #[arg(long)]
    connect: Option<String>,

    /// Narrate every document mutation on the console
    /// (`RUST_LOG=edit=debug`).
    #[arg(long)]
    narrate: bool,

    /// Build and run optimized.
    #[arg(long)]
    release: bool,

    /// Arguments forwarded to the editor after `--`.
    #[arg(last = true)]
    app_args: Vec<String>,
}

#[derive(Args)]
pub struct DemoArgs {
    /// How many editors to open on the log.
    #[arg(long, default_value_t = 2)]
    clients: usize,

    /// Address the server listens on.
    #[arg(long, default_value = LISTEN)]
    listen: String,

    /// Log to serve. Defaults to a scratch file removed on exit; naming one
    /// keeps it, so a second run resumes the same document.
    #[arg(long)]
    db: Option<PathBuf>,

    /// Narrate every document mutation in each editor's console.
    #[arg(long)]
    narrate: bool,

    /// Build and run optimized.
    #[arg(long)]
    release: bool,
}

/// One server in the foreground. Ctrl-Z and `fg` are freeze and thaw.
pub fn server(args: &ServerArgs) -> Result<()> {
    let db = args.db.clone().unwrap_or_else(default_db);
    if let Some(parent) = db.parent() {
        std::fs::create_dir_all(parent).ok();
    }
    eprintln!("  {} on {}", db.display(), ws_url(&args.listen));
    eprintln!("  Ctrl-Z freezes it (editors queue their commits), `fg` thaws it.");
    eprintln!();

    let db = db.to_string_lossy().into_owned();
    let mut argv = vec!["run", "-p", "blockworx-server", "--bin", "blockworx-server"];
    argv.extend(Profile::from(args.release).flag());
    argv.extend(["--", &db, "--listen", &args.listen]);
    cargo(&argv).run().context("running blockworx-server")?;
    Ok(())
}

/// One editor in the foreground, connected to a server.
pub fn client(args: &ClientArgs) -> Result<()> {
    let url = args.connect.clone().unwrap_or_else(|| ws_url(LISTEN));
    eprintln!("  editing {url}");
    eprintln!();

    let mut argv = vec!["run", "-p", "blockworx", "--bin", "blockworx"];
    argv.extend(Profile::from(args.release).flag());
    argv.extend(["--", "--connect", &url]);
    argv.extend(args.app_args.iter().map(String::as_str));

    let mut command = cargo(&argv);
    if let Some(filter) = Narration::from(args.narrate).filter() {
        command = command.env("RUST_LOG", filter);
    }
    command.run().context("running blockworx")?;
    Ok(())
}

/// A server and N editors from one terminal, with a console that freezes the
/// server without also stopping the terminal it is typed in.
pub fn demo(args: &DemoArgs) -> Result<()> {
    if args.clients == 0 {
        bail!("--clients must be at least 1");
    }
    let addr: SocketAddr = args
        .listen
        .parse()
        .with_context(|| format!("--listen {} is not an address", args.listen))?;

    let profile = Profile::from(args.release);
    let narration = Narration::from(args.narrate);
    build(profile)?;

    let scratch = args.db.is_none();
    let db = args.db.clone().unwrap_or_else(|| {
        workspace_root()
            .join("target")
            .join("demo-scratch.blockworx.db")
    });
    if scratch && db.exists() {
        std::fs::remove_file(&db).ok();
    }

    let mut session = Session {
        server: spawn_server(profile, &db, &args.listen)?,
        clients: Vec::new(),
        db: scratch.then(|| db.clone()),
        frozen: false,
    };
    wait_for(addr).context("the server never accepted a connection")?;

    let url = ws_url(&args.listen);
    for _ in 0..args.clients {
        session
            .clients
            .push(spawn_client(profile, narration, &url)?);
    }

    eprintln!();
    eprintln!("  {} editors on {}", args.clients, db.display());
    eprintln!("  serving {url}");
    eprintln!();
    eprintln!("  Edit in either window; the other follows. To see the case hands");
    eprintln!("  cannot reach, wait for both titles to say `live`, then freeze,");
    eprintln!("  edit the same block in both, and thaw: the later commit wins and");
    eprintln!("  both canvases agree.");
    eprintln!();
    console(&mut session)
}

/// Children plus what has to happen to them on the way out.
///
/// Two exits are covered and one is not, which is worth knowing before
/// leaving a `demo` running: `q` (and any `?` out of [`console`]) runs `Drop`,
/// and Ctrl-C reaches the whole process group so the children take it
/// directly. A **SIGTERM to xtask itself skips `Drop` and orphans them** —
/// measured, not assumed. Closing a signal that narrow costs either a signal
/// crate or `PR_SET_PDEATHSIG`, and xtask's entire dependency list is three
/// build helpers; `server` and `client` spawn nothing in the background and
/// have no such hole.
struct Session {
    server: Child,
    clients: Vec<Child>,
    db: Option<PathBuf>,
    frozen: bool,
}

impl Drop for Session {
    fn drop(&mut self) {
        if self.frozen {
            // A stopped process ignores SIGTERM until it runs again.
            signal(&self.server, "-CONT");
        }
        for child in self.clients.iter_mut().chain([&mut self.server]) {
            child.kill().ok();
            child.wait().ok();
        }
        if let Some(db) = &self.db {
            std::fs::remove_file(db).ok();
        }
    }
}

fn console(session: &mut Session) -> Result<()> {
    let stdin = std::io::stdin();
    loop {
        eprint!(
            "{}  [f]reeze [t]haw [q]uit > ",
            if session.frozen { "frozen" } else { "live  " }
        );
        std::io::stderr().flush().ok();

        let mut line = String::new();
        if stdin.lock().read_line(&mut line)? == 0 {
            return Ok(());
        }
        match line.trim() {
            "f" | "freeze" if !session.frozen => {
                signal(&session.server, "-STOP");
                session.frozen = true;
                eprintln!("  server stopped — edits now queue in each editor, unacknowledged");
            }
            "t" | "thaw" if session.frozen => {
                signal(&session.server, "-CONT");
                session.frozen = false;
                eprintln!("  server running — the queues flush and the titles catch up");
            }
            "f" | "t" | "freeze" | "thaw" => eprintln!("  already there"),
            "q" | "quit" | "" => return Ok(()),
            other => eprintln!("  don't know {other:?}"),
        }
    }
}

fn build(profile: Profile) -> Result<()> {
    let mut argv = vec![
        "build",
        "-p",
        "blockworx",
        "--bin",
        "blockworx",
        "-p",
        "blockworx-server",
        "--bin",
        "blockworx-server",
    ];
    argv.extend(profile.flag());
    cargo(&argv).run().context("building the demo binaries")?;
    Ok(())
}

fn bin(profile: Profile, name: &str) -> PathBuf {
    workspace_root()
        .join("target")
        .join(profile.dir())
        .join(name)
}

fn spawn_server(profile: Profile, db: &PathBuf, listen: &str) -> Result<Child> {
    Command::new(bin(profile, "blockworx-server"))
        .arg(db)
        .args(["--listen", listen])
        .current_dir(workspace_root())
        .spawn()
        .context("starting blockworx-server")
}

fn spawn_client(profile: Profile, narration: Narration, url: &str) -> Result<Child> {
    let mut cmd = Command::new(bin(profile, "blockworx"));
    cmd.args(["--connect", url]).current_dir(workspace_root());
    if let Some(filter) = narration.filter() {
        cmd.env("RUST_LOG", filter);
    }
    cmd.spawn().context("starting blockworx")
}

/// `kill` rather than a signal crate: xtask's whole dependency list is three
/// build helpers, and this is the only place that needs one.
fn signal(child: &Child, sig: &str) {
    Command::new("kill")
        .args([sig, &child.id().to_string()])
        .status()
        .ok();
}

fn wait_for(addr: SocketAddr) -> Result<()> {
    let deadline = Instant::now() + Duration::from_secs(20);
    while Instant::now() < deadline {
        if TcpStream::connect_timeout(&addr, Duration::from_millis(200)).is_ok() {
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(100));
    }
    bail!("{addr} did not come up within 20s")
}