xtask/

demo.rs

//! `cargo xtask demo` — the collaboration demo on one machine: a server, a
//! fresh log, and N editors connected to it.
//!
//! Doing this by hand is four terminals and a URL typed three times. What it
//! cannot do by hand at all is the interesting half: **freezing the server**.
//! Two people editing at human speed almost never have unacknowledged commits
//! in flight at the same moment, so the concurrent case — the one the whole
//! server-centric model exists for — is the one a manual demo never reaches.
//! `f` stops the server process, both editors keep authoring into their
//! queues, and `t` lets it sequence them.
//!
//! 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};

#[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 = "127.0.0.1:4100")]
    listen: String,

    /// Log to serve. Defaults to a scratch file that is 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
    /// (`RUST_LOG=edit=debug`).
    #[arg(long)]
    narrate: bool,

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

pub fn run(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))?;

    build(args)?;

    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(args, &db, &args.listen)?,
        clients: Vec::new(),
        db: scratch.then_some(db.clone()),
        frozen: false,
    };
    wait_for(addr).context("the server never accepted a connection")?;

    let url = format!("ws://{addr}/ws");
    for _ in 0..args.clients {
        session.clients.push(spawn_client(args, &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. `Drop` covers
/// every exit from [`console`] — including `?` — but not a signal: Ctrl-C in a
/// terminal reaches the whole process group, so the children take it too.
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(args: &DemoArgs) -> Result<()> {
    let mut argv = vec![
        "build",
        "-p",
        "blockworx",
        "--bin",
        "blockworx",
        "-p",
        "blockworx-server",
        "--bin",
        "blockworx-server",
    ];
    if args.release {
        argv.push("--release");
    }
    cargo(&argv).run().context("building the demo binaries")?;
    Ok(())
}

fn bin(args: &DemoArgs, name: &str) -> PathBuf {
    workspace_root()
        .join("target")
        .join(if args.release { "release" } else { "debug" })
        .join(name)
}

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

fn spawn_client(args: &DemoArgs, url: &str) -> Result<Child> {
    let mut cmd = Command::new(bin(args, "blockworx"));
    cmd.args(["--connect", url]).current_dir(workspace_root());
    if args.narrate {
        cmd.env("RUST_LOG", "blockworx=info,edit=debug");
    }
    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")
}