diff --git a/Cargo.lock b/Cargo.lock index 65dddb3c..ccd2dc37 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -386,6 +386,15 @@ dependencies = [ "itertools", ] +[[package]] +name = "crossbeam-channel" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-deque" version = "0.8.6" @@ -468,6 +477,12 @@ dependencies = [ "syn", ] +[[package]] +name = "deranged" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" + [[package]] name = "derive_builder" version = "0.20.2" @@ -550,7 +565,7 @@ checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" [[package]] name = "embers-cli" -version = "0.1.1" +version = "0.1.2" dependencies = [ "assert_cmd", "base64", @@ -573,7 +588,7 @@ dependencies = [ [[package]] name = "embers-client" -version = "0.1.1" +version = "0.1.2" dependencies = [ "async-trait", "base64", @@ -594,17 +609,18 @@ dependencies = [ [[package]] name = "embers-core" -version = "0.1.1" +version = "0.1.2" dependencies = [ "serde", "thiserror 2.0.18", "tracing", + "tracing-appender", "tracing-subscriber", ] [[package]] name = "embers-protocol" -version = "0.1.1" +version = "0.1.2" dependencies = [ "embers-core", "flatbuffers", @@ -614,7 +630,7 @@ dependencies = [ [[package]] name = "embers-server" -version = "0.1.1" +version = "0.1.2" dependencies = [ "alacritty_terminal", "base64", @@ -631,7 +647,7 @@ dependencies = [ [[package]] name = "embers-test-support" -version = "0.1.1" +version = "0.1.2" dependencies = [ "assert_cmd", "embers-core", @@ -1003,6 +1019,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-conv" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" + [[package]] name = "num-modular" version = "0.6.1" @@ -1212,6 +1234,12 @@ dependencies = [ "winreg", ] +[[package]] +name = "powerfmt" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -1699,6 +1727,12 @@ version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" +[[package]] +name = "symlink" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" + [[package]] name = "syn" version = "2.0.117" @@ -1787,6 +1821,36 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "time" +version = "0.3.49" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "711a53c2d47bbd818258c498c8dbfe186a2526c631495cfe7e078567f86b8469" +dependencies = [ + "deranged", + "num-conv", + "powerfmt", + "serde_core", + "time-core", + "time-macros", +] + +[[package]] +name = "time-core" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" + +[[package]] +name = "time-macros" +version = "0.2.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71c652a3727a9cbb9a02f707f530b618ce00d0ccd762009c8c23bd191df3c17d" +dependencies = [ + "num-conv", + "time-core", +] + [[package]] name = "tiny-keccak" version = "2.0.2" @@ -1844,6 +1908,19 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-appender" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c" +dependencies = [ + "crossbeam-channel", + "symlink", + "thiserror 2.0.18", + "time", + "tracing-subscriber", +] + [[package]] name = "tracing-attributes" version = "0.1.31" diff --git a/Cargo.toml b/Cargo.toml index e6160d25..ed2538c9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -43,6 +43,7 @@ tempfile = "3" thiserror = "2" tokio = { version = "1", features = ["fs", "io-std", "io-util", "macros", "net", "process", "rt-multi-thread", "signal", "sync", "time"] } tracing = "0.1" +tracing-appender = "0.2" tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } unicode-segmentation = "1.12" unicode-width = "0.2" diff --git a/README.md b/README.md index 921e7f71..f89ac4a1 100644 --- a/README.md +++ b/README.md @@ -87,6 +87,44 @@ Useful subcommand groups include: - `node`: `node zoom`, `node swap`, `node break`, `node join-buffer`, `node move-before`, `node move-after` - `popup`: `display-popup`, `kill-popup` +## Logging + +The tracing filter is resolved from the first of these that is set, highest +precedence first: `--log ` (alias `--log-level`), `-v`/`-vv`, +`EMBERS_LOG`, `RUST_LOG`, then the default of `info`. `` accepts the full +`tracing` env-filter syntax (a bare level like `debug`, or per-target directives +like `embers_server=trace,info`). + +```sh +embers --log-level debug list-sessions +EMBERS_LOG=embers_server=trace,info embers +``` + +Foreground commands log to stderr. The background server writes to a +daily-rotating file in the socket's directory, named +`embers-server..log`, retaining the most recent 7 days. The launching +command's filter is propagated to the server, and server panics are recorded in +the same log. + +## Resource Limits + +The server enforces operator-tunable ceilings so a runaway client cannot exhaust +host resources. Each is overridable via an environment variable: + +- `EMBERS_MAX_SESSIONS` (default `256`) +- `EMBERS_MAX_BUFFERS` (default `2048`) — each buffer owns a PTY-backed process plus scrollback, so this is the dominant resource bound +- `EMBERS_MAX_SCROLLBACK_LINES` (default `10000`) + +Requests that would exceed a limit are rejected with an error naming the limit; +existing sessions are unaffected. + +```sh +EMBERS_MAX_BUFFERS=256 EMBERS_MAX_SCROLLBACK_LINES=2000 embers +``` + +See [`docs/configuration.md`](docs/configuration.md) for the full reference of +operational environment variables and flags. + ## Configuration Embers loads configuration in this order: @@ -108,6 +146,10 @@ embers --config ./config.rhai The generated config API reference lives in [`docs/config-api`](docs/config-api/index.md), with a rendered mdBook copy in [`docs/config-api-book`](docs/config-api-book/index.html). +Operational configuration (socket path, logging, and resource limits via +environment variables and flags) is documented in +[`docs/configuration.md`](docs/configuration.md). + ## Development Run the test suite: diff --git a/crates/embers-cli/src/bin/embers-cli.rs b/crates/embers-cli/src/bin/embers-cli.rs index 4e7dec3f..14fa96da 100644 --- a/crates/embers-cli/src/bin/embers-cli.rs +++ b/crates/embers-cli/src/bin/embers-cli.rs @@ -1,11 +1,15 @@ use clap::Parser; -use embers_cli::{Cli, run}; +use embers_cli::{Cli, Command, run}; use embers_core::init_tracing; #[tokio::main] async fn main() { let cli = Cli::parse(); - init_tracing(&cli.log_filter()); + // The detached server sets up its own rotating file logger in `run_server`; + // every other invocation logs to stderr here. + if !matches!(cli.command, Some(Command::Serve)) { + init_tracing(&cli.log_filter()); + } if let Err(error) = run(cli).await { eprintln!("{}", format_error_chain(&error)); diff --git a/crates/embers-cli/src/lib.rs b/crates/embers-cli/src/lib.rs index bd878d83..a9e49af7 100644 --- a/crates/embers-cli/src/lib.rs +++ b/crates/embers-cli/src/lib.rs @@ -32,6 +32,9 @@ use embers_server::{SOCKET_ENV_VAR, Server, ServerConfig}; use tokio::time::{Duration, sleep}; use tracing::warn; +/// Environment variable selecting the tracing filter (e.g. `info`, `embers=debug`). +pub const EMBERS_LOG_ENV_VAR: &str = "EMBERS_LOG"; + #[derive(Debug, Parser)] #[command(name = "embers", about = "headless terminal multiplexer for embers")] pub struct Cli { @@ -39,7 +42,12 @@ pub struct Cli { pub socket: Option, #[arg(long, global = true)] pub config: Option, - #[arg(long, global = true, value_name = "FILTER")] + #[arg( + long, + visible_alias = "log-level", + global = true, + value_name = "FILTER" + )] pub log: Option, #[arg(short = 'v', long = "verbose", global = true, action = clap::ArgAction::Count)] pub verbose: u8, @@ -57,7 +65,7 @@ impl Cli { 1 => return "debug".to_owned(), _ => return "trace".to_owned(), } - if let Some(filter) = std::env::var("EMBERS_LOG") + if let Some(filter) = std::env::var(EMBERS_LOG_ENV_VAR) .ok() .filter(|value| !value.trim().is_empty()) { @@ -1038,6 +1046,7 @@ async fn execute_command(connection: &mut CliConnection, command: Command) -> Re } pub async fn run(cli: Cli) -> Result<()> { + let log_filter = cli.log_filter(); let Cli { socket, config, @@ -1066,7 +1075,7 @@ pub async fn run(cli: Cli) -> Result<()> { match command { None => { - ensure_server_process(&socket).await?; + ensure_server_process(&socket, &log_filter).await?; interactive::run(socket, None, config).await } Some(Command::Attach { target }) => { @@ -1082,12 +1091,12 @@ pub async fn run(cli: Cli) -> Result<()> { target, all_sessions, }) => { - ensure_server_process(&socket).await?; + ensure_server_process(&socket, &log_filter).await?; automation::run(socket, target, all_sessions).await } - Some(Command::Serve) => run_server(socket).await, + Some(Command::Serve) => run_server(socket, &log_filter).await, Some(command) => { - ensure_server_process(&socket).await?; + ensure_server_process(&socket, &log_filter).await?; let output = execute(&socket, command).await?; if !output.is_empty() { println!("{output}"); @@ -1196,7 +1205,7 @@ async fn server_is_available(socket_path: &Path) -> bool { CliConnection::connect(socket_path).await.is_ok() } -async fn ensure_server_process(socket_path: &Path) -> Result<()> { +async fn ensure_server_process(socket_path: &Path, log_filter: &str) -> Result<()> { if server_is_available(socket_path).await { return Ok(()); } @@ -1208,6 +1217,10 @@ async fn ensure_server_process(socket_path: &Path) -> Result<()> { .arg("__serve") .arg("--socket") .arg(socket_path) + // Propagate the resolved filter so a `--log`/`-v` flag reaches the + // detached server, not just the inherited EMBERS_LOG/RUST_LOG env. The + // server writes its own rotating log file, so its stdio is discarded. + .env(EMBERS_LOG_ENV_VAR, log_filter) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()) @@ -1242,8 +1255,15 @@ async fn ensure_server_process(socket_path: &Path) -> Result<()> { } } -async fn run_server(socket_path: PathBuf) -> Result<()> { +async fn run_server(socket_path: PathBuf, log_filter: &str) -> Result<()> { ensure_socket_parent(&socket_path)?; + // The detached server logs to a daily-rotating file next to the socket. Set + // this up before anything else so startup is captured, then route panics + // through tracing so a crash lands in the same log instead of a dead stderr. + if let Some(dir) = socket_path.parent() { + embers_core::init_server_tracing(log_filter, dir)?; + install_server_panic_hook(); + } let secure_parent = socket_path .parent() .is_some_and(|parent| parent == default_runtime_dir().as_path()); @@ -1253,6 +1273,17 @@ async fn run_server(socket_path: PathBuf) -> Result<()> { handle.shutdown().await } +/// Routes panics in the detached server through `tracing` (chained after the +/// default hook) so a crash is recorded in the rotating log file, which is the +/// server's only output sink once its stdio is discarded. +fn install_server_panic_hook() { + let previous = std::panic::take_hook(); + std::panic::set_hook(Box::new(move |info| { + tracing::error!(target: "panic", "server panicked: {info}"); + previous(info); + })); +} + fn ensure_socket_parent(socket_path: &Path) -> Result<()> { let Some(parent) = socket_path.parent() else { return Ok(()); diff --git a/crates/embers-cli/src/main.rs b/crates/embers-cli/src/main.rs index eca5d5f7..86e76128 100644 --- a/crates/embers-cli/src/main.rs +++ b/crates/embers-cli/src/main.rs @@ -1,12 +1,16 @@ use clap::Parser; -use embers_cli::{Cli, run}; +use embers_cli::{Cli, Command, run}; use embers_core::init_tracing; #[tokio::main] async fn main() { - init_tracing("info"); - let cli = Cli::parse(); + // The detached server sets up its own rotating file logger in `run_server`; + // every other invocation logs to stderr here. + if !matches!(cli.command, Some(Command::Serve)) { + init_tracing(&cli.log_filter()); + } + if let Err(error) = run(cli).await { eprintln!("{error}"); std::process::exit(1); diff --git a/crates/embers-cli/tests/interactive.rs b/crates/embers-cli/tests/interactive.rs index eb34df84..f18460f5 100644 --- a/crates/embers-cli/tests/interactive.rs +++ b/crates/embers-cli/tests/interactive.rs @@ -300,6 +300,27 @@ async fn wait_for_socket(socket_path: &Path) { panic!("timed out waiting for socket {}", socket_path.display()); } +fn interactive_startup_tempdir() -> tempfile::TempDir { + #[cfg(unix)] + { + tempfile::Builder::new() + .prefix("emb.") + .tempdir_in("/tmp") + .expect("tempdir") + } + + #[cfg(not(unix))] + { + tempfile::tempdir().expect("tempdir") + } +} + +fn interactive_socket_path(tempdir: &tempfile::TempDir) -> PathBuf { + // Keep the runtime keeper socket under macOS's Unix socket path limit even + // when the ambient TMPDIR is deeply nested. + tempdir.path().join("mux.sock") +} + async fn populate_scrollback_or_wait(harness: &mut PtyHarness, lines: usize) { harness .write_all("echo READY\r") @@ -604,8 +625,8 @@ async fn embers_without_subcommand_starts_server_and_client() { if !require_pty() { return; } - let tempdir = tempfile::tempdir().expect("tempdir"); - let socket_path = tempdir.path().join("embers.sock"); + let tempdir = interactive_startup_tempdir(); + let socket_path = interactive_socket_path(&tempdir); let socket_arg = socket_path.to_string_lossy().into_owned(); let (_spawned, mut harness) = spawn_embers(&["--socket", &socket_arg], socket_path.clone()); @@ -830,8 +851,8 @@ async fn page_up_enters_local_scrollback() { if !require_pty() { return; } - let tempdir = tempfile::tempdir().expect("tempdir"); - let socket_path = tempdir.path().join("embers.sock"); + let tempdir = interactive_startup_tempdir(); + let socket_path = interactive_socket_path(&tempdir); let socket_arg = socket_path.to_string_lossy().into_owned(); let (_spawned, mut harness) = spawn_embers(&["--socket", &socket_arg], socket_path.clone()); @@ -854,8 +875,8 @@ async fn local_selection_yank_emits_osc52_clipboard_sequence() { if !require_pty() { return; } - let tempdir = tempfile::tempdir().expect("tempdir"); - let socket_path = tempdir.path().join("embers.sock"); + let tempdir = interactive_startup_tempdir(); + let socket_path = interactive_socket_path(&tempdir); let socket_arg = socket_path.to_string_lossy().into_owned(); let (_spawned, mut harness) = spawn_embers(&["--socket", &socket_arg], socket_path.clone()); diff --git a/crates/embers-core/Cargo.toml b/crates/embers-core/Cargo.toml index fff22825..8d3e6cd7 100644 --- a/crates/embers-core/Cargo.toml +++ b/crates/embers-core/Cargo.toml @@ -14,4 +14,5 @@ doctest = false serde.workspace = true thiserror.workspace = true tracing.workspace = true +tracing-appender.workspace = true tracing-subscriber.workspace = true diff --git a/crates/embers-core/src/diagnostics.rs b/crates/embers-core/src/diagnostics.rs index 9832c4a3..8aa2bd79 100644 --- a/crates/embers-core/src/diagnostics.rs +++ b/crates/embers-core/src/diagnostics.rs @@ -1,11 +1,21 @@ +use std::io; +use std::path::Path; use std::sync::Once; use std::sync::atomic::{AtomicU64, Ordering}; use tracing::{Span, span}; +use tracing_appender::rolling::{RollingFileAppender, Rotation}; use tracing_subscriber::EnvFilter; use crate::{NodeId, RequestId}; +/// Filename prefix for the detached server's rolling log files. +pub const SERVER_LOG_FILE_PREFIX: &str = "embers-server"; +/// Filename suffix (extension) for the detached server's rolling log files. +pub const SERVER_LOG_FILE_SUFFIX: &str = "log"; +/// Number of rotated daily log files to retain before the oldest is pruned. +const SERVER_LOG_MAX_FILES: usize = 7; + static REQUEST_IDS: AtomicU64 = AtomicU64::new(1); static TRACING: Once = Once::new(); static TEST_TRACING: Once = Once::new(); @@ -52,6 +62,40 @@ pub fn init_tracing(default_filter: &str) { }); } +/// Initializes tracing for the detached server, writing to a daily-rotating log +/// file in `directory` (named `embers-server..log`) and retaining the most +/// recent [`SERVER_LOG_MAX_FILES`] days. ANSI styling is disabled so the file +/// stays plain text. +/// +/// Returns an error if the rotating appender can't be created. The detached +/// server's stdio is discarded, so the log file is its only usable sink; failing +/// startup loudly is preferable to running a daemon whose logs — and the panic +/// traces this hardening installs — silently vanish into the dead stderr. +pub fn init_server_tracing(default_filter: &str, directory: &Path) -> io::Result<()> { + let mut result = Ok(()); + TRACING.call_once(|| { + match RollingFileAppender::builder() + .rotation(Rotation::DAILY) + .filename_prefix(SERVER_LOG_FILE_PREFIX) + .filename_suffix(SERVER_LOG_FILE_SUFFIX) + .max_log_files(SERVER_LOG_MAX_FILES) + .build(directory) + { + Ok(appender) => { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::new(default_filter)) + .with_target(false) + .with_ansi(false) + .compact() + .with_writer(appender) + .try_init(); + } + Err(error) => result = Err(io::Error::other(error.to_string())), + } + }); + result +} + pub fn init_test_tracing() { TEST_TRACING.call_once(|| { let _ = tracing_subscriber::fmt() diff --git a/crates/embers-core/src/lib.rs b/crates/embers-core/src/lib.rs index 1e28088b..7e1d7c11 100644 --- a/crates/embers-core/src/lib.rs +++ b/crates/embers-core/src/lib.rs @@ -6,8 +6,8 @@ pub mod metadata; pub mod snapshot; pub use diagnostics::{ - RequestContext, format_focus_path, format_tree_dump, init_test_tracing, init_tracing, - new_request_id, request_span, + RequestContext, format_focus_path, format_tree_dump, init_server_tracing, init_test_tracing, + init_tracing, new_request_id, request_span, }; pub use error::{ErrorCode, MuxError, Result, WireError}; pub use geometry::{FloatGeometry, Point, PtySize, Rect, Size, SplitDirection}; diff --git a/crates/embers-server/src/buffer_runtime.rs b/crates/embers-server/src/buffer_runtime.rs index 2eb2dd2f..725ac25c 100644 --- a/crates/embers-server/src/buffer_runtime.rs +++ b/crates/embers-server/src/buffer_runtime.rs @@ -629,9 +629,13 @@ impl KeeperConnection { impl KeeperSurface { fn new(size: PtySize) -> Self { + // The keeper inherits the server's environment, so resolving the + // scrollback ceiling here applies a single operator-configured value to + // this buffer (see `MAX_SCROLLBACK_LINES_ENV_VAR`). + let max_scrollback_lines = crate::config::ResourceLimits::from_env().max_scrollback_lines; Self { router: RawByteRouter, - backend: Box::new(AlacrittyTerminalBackend::new(size)), + backend: Box::new(AlacrittyTerminalBackend::new(size, max_scrollback_lines)), size, } } @@ -865,13 +869,15 @@ pub fn run_runtime_keeper(cli: RuntimeKeeperCli) -> Result<()> { let pair = match pair { Some(pair) => pair, None => { - // Safe: each failed retry stores the PTY allocation error before continuing. - let error = - last_error.expect("openpty retry loop must capture an error before failing"); - return Err(MuxError::pty(format!( - "failed to openpty after {} attempts: {error}", - KEEPER_PTY_MAX_RETRIES + 1 - ))); + // Each failed retry stores the PTY allocation error before continuing, so + // last_error is normally populated; fall back to a generic message if not + // rather than panicking on this hot spawn path. + let attempts = KEEPER_PTY_MAX_RETRIES + 1; + let detail = match last_error { + Some(error) => format!("failed to openpty after {attempts} attempts: {error}"), + None => format!("failed to openpty after {attempts} attempts"), + }; + return Err(MuxError::pty(detail)); } }; @@ -1142,11 +1148,10 @@ impl KeeperRuntime { { return Err(MuxError::conflict("buffer pipe is already running")); } - *pipe = Some(KeeperPipe::spawn(command, cwd, env)?); - Ok(pipe - .as_mut() - .expect("pipe slot populated after spawn") - .status()) + let mut spawned = KeeperPipe::spawn(command, cwd, env)?; + let status = spawned.status(); + *pipe = Some(spawned); + Ok(status) } fn stop_pipe(&self) -> Result { diff --git a/crates/embers-server/src/config.rs b/crates/embers-server/src/config.rs index 551d7e26..1208831d 100644 --- a/crates/embers-server/src/config.rs +++ b/crates/embers-server/src/config.rs @@ -4,12 +4,79 @@ use std::path::PathBuf; pub const SOCKET_ENV_VAR: &str = "EMBERS_SOCKET"; +/// Environment variable overriding [`ResourceLimits::max_sessions`]. +pub const MAX_SESSIONS_ENV_VAR: &str = "EMBERS_MAX_SESSIONS"; +/// Environment variable overriding [`ResourceLimits::max_buffers`]. +pub const MAX_BUFFERS_ENV_VAR: &str = "EMBERS_MAX_BUFFERS"; +/// Environment variable overriding [`ResourceLimits::max_scrollback_lines`]. +/// +/// This is read by the runtime keeper process when constructing a terminal +/// backend, which inherits the server's environment, so a single value applies +/// to both processes. +pub const MAX_SCROLLBACK_LINES_ENV_VAR: &str = "EMBERS_MAX_SCROLLBACK_LINES"; + +/// Default ceiling on concurrently live sessions. +pub const DEFAULT_MAX_SESSIONS: usize = 256; +/// Default ceiling on concurrently live buffers. Each buffer owns a PTY-backed +/// child process plus scrollback, so this is the dominant resource bound. +pub const DEFAULT_MAX_BUFFERS: usize = 2048; +/// Default scrollback retained per buffer. Combined with [`DEFAULT_MAX_BUFFERS`] +/// this bounds worst-case server memory. +pub const DEFAULT_MAX_SCROLLBACK_LINES: usize = 10_000; + +/// Operator-tunable ceilings that prevent a client from exhausting server +/// resources by creating unbounded sessions, buffers, or scrollback. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ResourceLimits { + pub max_sessions: usize, + pub max_buffers: usize, + pub max_scrollback_lines: usize, +} + +impl Default for ResourceLimits { + fn default() -> Self { + Self { + max_sessions: DEFAULT_MAX_SESSIONS, + max_buffers: DEFAULT_MAX_BUFFERS, + max_scrollback_lines: DEFAULT_MAX_SCROLLBACK_LINES, + } + } +} + +impl ResourceLimits { + /// Build limits from defaults, applying any environment-variable overrides. + /// A value of `0` or an unparseable value falls back to the default. + pub fn from_env() -> Self { + let mut limits = Self::default(); + if let Some(value) = parse_limit_env(MAX_SESSIONS_ENV_VAR) { + limits.max_sessions = value; + } + if let Some(value) = parse_limit_env(MAX_BUFFERS_ENV_VAR) { + limits.max_buffers = value; + } + if let Some(value) = parse_limit_env(MAX_SCROLLBACK_LINES_ENV_VAR) { + limits.max_scrollback_lines = value; + } + limits + } +} + +/// Read a positive `usize` from `var`. Returns `None` when unset, empty, zero, +/// or unparseable, so callers keep their default. +fn parse_limit_env(var: &str) -> Option { + std::env::var(var) + .ok() + .and_then(|value| value.trim().parse::().ok()) + .filter(|value| *value > 0) +} + #[derive(Clone, Debug, PartialEq, Eq)] pub struct ServerConfig { pub socket_path: PathBuf, pub workspace_path: PathBuf, pub runtime_dir: PathBuf, pub buffer_env: BTreeMap, + pub limits: ResourceLimits, } impl ServerConfig { @@ -26,6 +93,38 @@ impl ServerConfig { workspace_path, runtime_dir, buffer_env, + limits: ResourceLimits::from_env(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resource_limits_default_to_documented_constants() { + let limits = ResourceLimits::default(); + assert_eq!(limits.max_sessions, DEFAULT_MAX_SESSIONS); + assert_eq!(limits.max_buffers, DEFAULT_MAX_BUFFERS); + assert_eq!(limits.max_scrollback_lines, DEFAULT_MAX_SCROLLBACK_LINES); + } + + #[test] + fn parse_limit_env_rejects_zero_empty_and_garbage() { + // SAFETY: single-threaded test; we set and remove the var within it. + let var = "EMBERS_TEST_PARSE_LIMIT_ENV"; + for value in ["0", "", " ", "nope", "-5"] { + unsafe { std::env::set_var(var, value) }; + assert_eq!( + parse_limit_env(var), + None, + "value {value:?} should be rejected" + ); } + unsafe { std::env::set_var(var, " 42 ") }; + assert_eq!(parse_limit_env(var), Some(42)); + unsafe { std::env::remove_var(var) }; + assert_eq!(parse_limit_env(var), None); } } diff --git a/crates/embers-server/src/lib.rs b/crates/embers-server/src/lib.rs index 8b3c5c37..a7f0dd71 100644 --- a/crates/embers-server/src/lib.rs +++ b/crates/embers-server/src/lib.rs @@ -13,7 +13,7 @@ pub use buffer_runtime::{ BufferRuntimePipeStopReason, BufferRuntimeStatus, BufferRuntimeUpdate, RuntimeKeeperCli, run_runtime_keeper, }; -pub use config::{SOCKET_ENV_VAR, ServerConfig}; +pub use config::{ResourceLimits, SOCKET_ENV_VAR, ServerConfig}; pub use model::{ Buffer, BufferAttachment, BufferPipe, BufferPipeState, BufferPipeStopReason, BufferState, BufferViewNode, BufferViewState, ExitedBuffer, FloatingWindow, InterruptedBuffer, Node, diff --git a/crates/embers-server/src/server.rs b/crates/embers-server/src/server.rs index 05d012da..d53cc9ac 100644 --- a/crates/embers-server/src/server.rs +++ b/crates/embers-server/src/server.rs @@ -29,8 +29,9 @@ use tokio::net::UnixListener; use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; use tokio::sync::{Mutex, Notify, mpsc, oneshot, watch}; use tokio::task::JoinHandle; -use tracing::{Instrument, debug, error, info}; +use tracing::{Instrument, debug, error, info, warn}; +use crate::config::ResourceLimits; use crate::model::{ BufferKind, BufferPipe, BufferPipeState, BufferPipeStopReason, HelperBufferScope, Node, }; @@ -44,6 +45,18 @@ use crate::{ ServerConfig, ServerState, TabEntry, }; +/// Capacity of each client's outbound envelope queue. A client that cannot keep +/// up (suspended terminal, stalled socket) fills this queue; rather than letting +/// it grow unbounded and exhaust the shared daemon, [`Runtime::broadcast`] drops +/// such a client. Sized to absorb normal render bursts while bounding per-client +/// memory. +/// +/// Note the response path in [`handle_connection`] does not drop on a full queue: +/// a response is awaited by the client, so it backpressures (awaits capacity) and +/// only fails on a genuinely closed channel. Only best-effort events, which a +/// reattached client recovers via resync, are dropped when the queue is full. +const OUTBOUND_CHANNEL_CAPACITY: usize = 1024; + #[derive(Debug)] pub struct Server { config: ServerConfig, @@ -67,6 +80,7 @@ impl Server { self.config.workspace_path.clone(), self.config.runtime_dir.clone(), self.config.buffer_env.clone(), + self.config.limits.clone(), )); runtime.restore_buffer_runtimes().await?; let listener = UnixListener::bind(&self.config.socket_path)?; @@ -88,7 +102,7 @@ impl Server { let (stream, _) = result?; let connection_id = runtime.next_connection_id.fetch_add(1, Ordering::Relaxed); let (reader, writer) = stream.into_split(); - let (outbound_tx, outbound_rx) = mpsc::unbounded_channel(); + let (outbound_tx, outbound_rx) = mpsc::channel(OUTBOUND_CHANNEL_CAPACITY); let (shutdown_tx, shutdown_rx) = oneshot::channel(); let (stopped_tx, stopped_rx) = oneshot::channel(); runtime @@ -260,7 +274,7 @@ impl ShutdownSignal { struct Subscription { connection_id: u64, session_id: Option, - sender: mpsc::UnboundedSender, + sender: mpsc::Sender, } struct ClientConnection { @@ -292,6 +306,7 @@ struct Runtime { workspace_path: PathBuf, runtime_dir: PathBuf, buffer_env: BTreeMap, + limits: ResourceLimits, subscriptions: Mutex>, clients: Mutex>, next_connection_id: AtomicU64, @@ -368,6 +383,7 @@ impl Runtime { workspace_path: PathBuf, runtime_dir: PathBuf, buffer_env: BTreeMap, + limits: ResourceLimits, ) -> Self { Self { state: Mutex::new(state), @@ -377,6 +393,7 @@ impl Runtime { workspace_path, runtime_dir, buffer_env, + limits, subscriptions: Mutex::new(BTreeMap::new()), clients: Mutex::new(BTreeMap::new()), next_connection_id: AtomicU64::new(1), @@ -400,9 +417,11 @@ impl Runtime { } fn take_buffer_shutdown_intent(&self, buffer_id: BufferId) -> bool { + // Recover from poisoning: this set is plain bookkeeping and must not + // bring down the server if some other task panicked while holding it. self.buffer_shutdown_intents .lock() - .expect("buffer shutdown intent lock") + .unwrap_or_else(|poisoned| poisoned.into_inner()) .remove(&buffer_id) } @@ -540,7 +559,7 @@ impl Runtime { async fn dispatch_request( self: &Arc, connection_id: u64, - outbound: &mpsc::UnboundedSender, + outbound: &mpsc::Sender, request: ClientMessage, ) -> ( ServerResponse, @@ -745,6 +764,20 @@ impl Runtime { match request { SessionRequest::Create { request_id, name } => { + if state.sessions.len() >= self.limits.max_sessions { + return ( + error_response( + Some(request_id), + ErrorCode::Conflict, + limit_reached_message( + "session", + self.limits.max_sessions, + crate::config::MAX_SESSIONS_ENV_VAR, + ), + ), + Vec::new(), + ); + } let session_id = state.create_session(name); match session_snapshot(&state, session_id) { Ok(snapshot) => ( @@ -909,6 +942,20 @@ impl Runtime { let buffer_id = { let mut state = self.state.lock().await; + if state.buffers.len() >= self.limits.max_buffers { + return ( + error_response( + Some(request_id), + ErrorCode::Conflict, + limit_reached_message( + "buffer", + self.limits.max_buffers, + crate::config::MAX_BUFFERS_ENV_VAR, + ), + ), + Vec::new(), + ); + } state.create_buffer_with_env( title.unwrap_or_else(|| "buffer".to_owned()), command, @@ -2243,7 +2290,9 @@ impl Runtime { } clients .get_mut(&client_id) - .expect("client existence was checked above") + .ok_or_else(|| { + MuxError::not_found(format!("client {client_id} was not found")) + })? .current_session_id = Some(session_id); let subscriptions = self.subscriptions.lock().await; let mut subscribed_all_sessions = false; @@ -2384,6 +2433,13 @@ impl Runtime { let helper_title = format!("{} history", source_title); let helper_buffer_id = { let mut state = self.state.lock().await; + if state.buffers.len() >= self.limits.max_buffers { + return Err(MuxError::conflict(limit_reached_message( + "buffer", + self.limits.max_buffers, + crate::config::MAX_BUFFERS_ENV_VAR, + ))); + } let helper_buffer_id = state.create_helper_buffer( helper_title.clone(), source_buffer_id, @@ -2675,16 +2731,12 @@ impl Runtime { async fn record_buffer_update(&self, buffer_id: BufferId, update: BufferRuntimeUpdate) { let (render_invalidated, pipe_event) = { let mut state = self.state.lock().await; - let Some(existing) = state.buffers.get(&buffer_id) else { + let Some(buffer) = state.buffers.get_mut(&buffer_id) else { return; }; - let previous_pipe = existing.pipe.clone(); + let previous_pipe = buffer.pipe.clone(); let mut render_invalidated = false; { - let buffer = state - .buffers - .get_mut(&buffer_id) - .expect("buffer still exists while update is applied"); let sequence_advanced = update.sequence > buffer.last_snapshot_seq; let sequence_current = update.sequence >= buffer.last_snapshot_seq; if sequence_advanced { @@ -2839,7 +2891,7 @@ impl Runtime { let mut shutdown_intents = self .buffer_shutdown_intents .lock() - .expect("buffer shutdown intent lock"); + .unwrap_or_else(|poisoned| poisoned.into_inner()); runtimes .iter() .map(|(&buffer_id, runtime)| { @@ -2876,13 +2928,26 @@ impl Runtime { .session_id .is_some_and(|session_id| event_session_ids.contains(&session_id)); - if event_matches - && subscription + if event_matches { + match subscription .sender - .send(ServerEnvelope::Event(event.clone())) - .is_err() - { - return false; + .try_send(ServerEnvelope::Event(event.clone())) + { + Ok(()) => {} + // Client is gone: drop the subscription. + Err(mpsc::error::TrySendError::Closed(_)) => return false, + // Client cannot keep up: detach it rather than let + // its queue grow without bound and exhaust the + // daemon. It can reattach and resync from a fresh + // snapshot. + Err(mpsc::error::TrySendError::Full(_)) => { + warn!( + connection_id = subscription.connection_id, + "client outbound queue full; detaching slow client" + ); + return false; + } + } } } true @@ -3087,7 +3152,7 @@ async fn handle_connection( runtime: Arc, connection_id: u64, mut reader: OwnedReadHalf, - outbound: mpsc::UnboundedSender, + outbound: mpsc::Sender, mut shutdown: oneshot::Receiver<()>, ) -> Result { let mut server_shutdown = runtime.shutdown.subscribe(); @@ -3113,6 +3178,7 @@ async fn handle_connection( Some(frame.request_id), ProtocolError::UnexpectedFrameType(frame.frame_type), ))) + .await .is_err() { return Err(MuxError::transport("connection writer closed")); @@ -3131,6 +3197,7 @@ async fn handle_connection( actual: request.request_id(), }, ))) + .await .is_err() { return Err(MuxError::transport("connection writer closed")); @@ -3145,6 +3212,7 @@ async fn handle_connection( Some(frame.request_id), error, ))) + .await .is_err() { return Err(MuxError::transport("connection writer closed")); @@ -3159,7 +3227,11 @@ async fn handle_connection( .instrument(request_span("handle_request", request_id)) .await; - if outbound.send(ServerEnvelope::Response(response)).is_err() { + if outbound + .send(ServerEnvelope::Response(response)) + .await + .is_err() + { return Err(MuxError::transport("connection writer closed")); } let retired_session_ids = closed_session_ids(&events); @@ -3175,7 +3247,7 @@ async fn handle_connection( async fn write_loop( mut writer: OwnedWriteHalf, - mut outbound: mpsc::UnboundedReceiver, + mut outbound: mpsc::Receiver, ) -> Result<()> { while let Some(envelope) = outbound.recv().await { let payload = encode_server_envelope(&envelope).map_err(protocol_error_to_mux)?; @@ -3319,6 +3391,13 @@ fn protocol_error_response(request_id: Option, error: ProtocolError) error_response(request_id, ErrorCode::ProtocolViolation, error.to_string()) } +/// Rejection message shared by the session/buffer creation paths when a resource +/// ceiling is hit. `resource` is the singular noun ("session"/"buffer"); `limit` +/// and `env_var` name the active ceiling and the override knob. +fn limit_reached_message(resource: &str, limit: usize, env_var: &str) -> String { + format!("{resource} limit reached ({limit}); close existing {resource}s or raise {env_var}") +} + fn mux_error_response(request_id: Option, error: MuxError) -> ServerResponse { let (code, message) = match error { MuxError::Wire(wire) => (wire.code, wire.message), @@ -3446,15 +3525,21 @@ mod tests { use std::path::PathBuf; use std::sync::Arc; - use embers_core::{ActivityState, FloatGeometry, MuxError, RequestId, SplitDirection}; + use embers_core::{ + ActivityState, ErrorCode, FloatGeometry, MuxError, RequestId, SplitDirection, + }; use embers_protocol::{ - BufferHistoryPlacement, BufferHistoryScope, InputRequest, NodeBreakDestination, - NodeJoinPlacement, NodeRequest, ServerEnvelope, ServerEvent, ServerResponse, + BufferHistoryPlacement, BufferHistoryScope, BufferRequest, InputRequest, + NodeBreakDestination, NodeJoinPlacement, NodeRequest, ServerEnvelope, ServerEvent, + ServerResponse, SessionRequest, }; use tempfile::tempdir; use tokio::sync::mpsc; - use super::{Runtime, ShutdownSignal, Subscription, wait_for_shutdown}; + use super::{ + OUTBOUND_CHANNEL_CAPACITY, Runtime, ShutdownSignal, Subscription, wait_for_shutdown, + }; + use crate::ResourceLimits; use crate::model::HelperBufferScope; use crate::{ BufferRuntimePipeStatus, BufferRuntimePipeStopReason, BufferRuntimeUpdate, BufferState, @@ -3514,6 +3599,7 @@ mod tests { tempdir.path().join("workspace.json"), tempdir.path().join("runtime"), BTreeMap::new(), + ResourceLimits::default(), ), session_id, floating_id, @@ -3577,6 +3663,7 @@ mod tests { tempdir.path().join("workspace.json"), tempdir.path().join("runtime"), BTreeMap::new(), + ResourceLimits::default(), ), session_id, floating_id, @@ -3603,6 +3690,7 @@ mod tests { PathBuf::from("workspace"), PathBuf::from("runtime"), BTreeMap::new(), + ResourceLimits::default(), ); runtime .buffer_shutdown_intents @@ -3615,6 +3703,140 @@ mod tests { assert!(!runtime.take_buffer_shutdown_intent(embers_core::BufferId(2))); } + #[tokio::test] + async fn create_session_is_rejected_when_session_limit_reached() { + let runtime = Runtime::new( + ServerState::new(), + PathBuf::from("server.sock"), + PathBuf::from("workspace"), + PathBuf::from("runtime"), + BTreeMap::new(), + ResourceLimits { + max_sessions: 1, + ..ResourceLimits::default() + }, + ); + + let (first, _events) = runtime + .dispatch_session(SessionRequest::Create { + request_id: RequestId(1), + name: "alpha".to_owned(), + }) + .await; + assert!(matches!(first, ServerResponse::SessionSnapshot(_))); + + let (second, events) = runtime + .dispatch_session(SessionRequest::Create { + request_id: RequestId(2), + name: "beta".to_owned(), + }) + .await; + assert!(events.is_empty()); + match second { + ServerResponse::Error(response) => { + assert_eq!(response.error.code, ErrorCode::Conflict); + } + other => panic!("expected conflict error, got {other:?}"), + } + assert_eq!(runtime.state.lock().await.sessions.len(), 1); + } + + #[tokio::test] + async fn create_buffer_is_rejected_when_buffer_limit_reached() { + let runtime = Arc::new(Runtime::new( + ServerState::new(), + PathBuf::from("server.sock"), + PathBuf::from("workspace"), + PathBuf::from("runtime"), + BTreeMap::new(), + ResourceLimits { + max_buffers: 1, + ..ResourceLimits::default() + }, + )); + + // Pre-seed the single allowed buffer directly so the dispatch guard + // rejects the next create before any PTY process is spawned. + runtime + .state + .lock() + .await + .create_buffer("existing", vec!["/bin/sh".to_owned()], None); + + let (response, events) = runtime + .dispatch_buffer( + 1, + BufferRequest::Create { + request_id: RequestId(1), + title: None, + command: vec!["/bin/sh".to_owned()], + cwd: None, + env: BTreeMap::new(), + }, + ) + .await; + assert!(events.is_empty()); + match response { + ServerResponse::Error(response) => { + assert_eq!(response.error.code, ErrorCode::Conflict); + } + other => panic!("expected conflict error, got {other:?}"), + } + assert_eq!(runtime.state.lock().await.buffers.len(), 1); + } + + #[tokio::test] + async fn open_history_buffer_is_rejected_when_buffer_limit_reached() { + let tempdir = tempdir().expect("tempdir"); + let mut state = ServerState::new(); + let session_id = state.create_session("alpha"); + // A helper source carries its lines inline, so opening its history does + // not need a live PTY capture and exercises the limit guard directly. + let base_buffer_id = state.create_buffer("base", vec!["/bin/sh".to_owned()], None); + let source_buffer_id = state + .create_helper_buffer( + "history", + base_buffer_id, + HelperBufferScope::Full, + None, + vec!["line-1".to_owned()], + ) + .expect("create source helper buffer"); + let leaf = state + .create_buffer_view(session_id, source_buffer_id) + .expect("create leaf"); + state + .add_root_tab(session_id, "main", leaf) + .expect("attach leaf"); + + let buffer_count = state.buffers.len(); + let runtime = Arc::new(Runtime::new( + state, + tempdir.path().join("server.sock"), + tempdir.path().join("workspace.json"), + tempdir.path().join("runtime"), + BTreeMap::new(), + ResourceLimits { + max_buffers: buffer_count, + ..ResourceLimits::default() + }, + )); + + let error = runtime + .open_history_buffer( + 1, + None, + source_buffer_id, + BufferHistoryScope::Full, + BufferHistoryPlacement::Tab, + ) + .await + .expect_err("helper buffer creation must respect the buffer limit"); + + assert!(matches!(error, MuxError::Conflict(_)), "got {error:?}"); + assert_eq!(runtime.state.lock().await.buffers.len(), buffer_count); + } + #[tokio::test] async fn record_buffer_update_ignores_stale_sequences() { let runtime = Runtime::new( @@ -3623,6 +3845,7 @@ mod tests { PathBuf::from("workspace"), PathBuf::from("runtime"), BTreeMap::new(), + ResourceLimits::default(), ); let buffer_id = { let mut state = runtime.state.lock().await; @@ -3635,7 +3858,7 @@ mod tests { buffer.activity = ActivityState::Activity; buffer_id }; - let (sender, mut receiver) = mpsc::unbounded_channel(); + let (sender, mut receiver) = mpsc::channel(OUTBOUND_CHANNEL_CAPACITY); runtime.subscriptions.lock().await.insert( 1, Subscription { @@ -3706,6 +3929,7 @@ mod tests { PathBuf::from("workspace"), PathBuf::from("runtime"), BTreeMap::new(), + ResourceLimits::default(), ); let buffer_id = { let mut state = runtime.state.lock().await; @@ -3717,7 +3941,7 @@ mod tests { buffer.last_snapshot_seq = 5; buffer_id }; - let (sender, mut receiver) = mpsc::unbounded_channel(); + let (sender, mut receiver) = mpsc::channel(OUTBOUND_CHANNEL_CAPACITY); runtime.subscriptions.lock().await.insert( 1, Subscription { @@ -3764,6 +3988,7 @@ mod tests { PathBuf::from("workspace"), PathBuf::from("runtime"), BTreeMap::new(), + ResourceLimits::default(), ); let buffer_id = { let mut state = runtime.state.lock().await; @@ -3775,7 +4000,7 @@ mod tests { buffer.last_snapshot_seq = 5; buffer_id }; - let (sender, mut receiver) = mpsc::unbounded_channel(); + let (sender, mut receiver) = mpsc::channel(OUTBOUND_CHANNEL_CAPACITY); runtime.subscriptions.lock().await.insert( 1, Subscription { @@ -3831,6 +4056,7 @@ mod tests { PathBuf::from("workspace"), PathBuf::from("runtime"), BTreeMap::new(), + ResourceLimits::default(), ); let buffer_id = { let mut state = runtime.state.lock().await; @@ -3887,6 +4113,7 @@ mod tests { tempdir.path().join("workspace.json"), tempdir.path().join("runtime"), BTreeMap::new(), + ResourceLimits::default(), )); runtime @@ -3923,6 +4150,7 @@ mod tests { tempdir.path().join("workspace.json"), tempdir.path().join("runtime"), BTreeMap::new(), + ResourceLimits::default(), )); runtime @@ -3947,6 +4175,7 @@ mod tests { tempdir.path().join("workspace.json"), tempdir.path().join("runtime"), BTreeMap::new(), + ResourceLimits::default(), ); let error = runtime @@ -3993,6 +4222,7 @@ mod tests { tempdir.path().join("workspace.json"), tempdir.path().join("runtime"), BTreeMap::new(), + ResourceLimits::default(), )); let (response, events) = runtime @@ -4066,6 +4296,7 @@ mod tests { tempdir.path().join("workspace.json"), tempdir.path().join("runtime"), BTreeMap::new(), + ResourceLimits::default(), ); let (response, events) = runtime diff --git a/crates/embers-server/src/state.rs b/crates/embers-server/src/state.rs index dac06ef7..2b8d657d 100644 --- a/crates/embers-server/src/state.rs +++ b/crates/embers-server/src/state.rs @@ -154,10 +154,11 @@ impl ServerState { } } - let safe_next_session_id = next_id_after_max(sessions.keys().map(|id| id.0)); - let safe_next_buffer_id = next_id_after_max(buffers.keys().map(|id| id.0)); - let safe_next_node_id = next_id_after_max(nodes.keys().map(|id| id.0)); - let safe_next_floating_id = next_id_after_max(floating.keys().map(|id| id.0)); + let safe_next_session_id = checked_next_id(sessions.keys().map(|id| id.0), "session")?; + let safe_next_buffer_id = checked_next_id(buffers.keys().map(|id| id.0), "buffer")?; + let safe_next_node_id = checked_next_id(nodes.keys().map(|id| id.0), "node")?; + let safe_next_floating_id = + checked_next_id(floating.keys().map(|id| id.0), "floating window")?; let state = Self { sessions, @@ -180,10 +181,16 @@ impl ServerState { buffers: self.buffers.values().map(persisted_buffer).collect(), nodes: self.nodes.values().map(persisted_node).collect(), floating: self.floating.values().map(persisted_floating).collect(), - next_session_id: next_id_after_max(self.sessions.keys().map(|id| id.0)), - next_buffer_id: next_id_after_max(self.buffers.keys().map(|id| id.0)), - next_node_id: next_id_after_max(self.nodes.keys().map(|id| id.0)), - next_floating_id: next_id_after_max(self.floating.keys().map(|id| id.0)), + // In-memory state cannot reach an exhausted id space (allocation would + // have failed long before), so serialize a saturating hint; if it ever + // did occur, the next load rejects it via `checked_next_id`. + next_session_id: next_id_after_max(self.sessions.keys().map(|id| id.0)) + .unwrap_or(u64::MAX), + next_buffer_id: next_id_after_max(self.buffers.keys().map(|id| id.0)) + .unwrap_or(u64::MAX), + next_node_id: next_id_after_max(self.nodes.keys().map(|id| id.0)).unwrap_or(u64::MAX), + next_floating_id: next_id_after_max(self.floating.keys().map(|id| id.0)) + .unwrap_or(u64::MAX), } } @@ -1587,18 +1594,21 @@ impl ServerState { let title = self.buffer(buffer_id)?.title.clone(); let tabs_id = if matches!(self.node(node_id)?, Node::Tabs(_)) { node_id - } else if matches!( - self.node_parent(node_id)? - .map(|id| self.node(id)) - .transpose()?, - Some(Node::Tabs(_)) - ) { - self.node_parent(node_id)?.expect("checked parent exists") } else { - let tabs_id = - self.wrap_node_in_tabs(node_id, self.default_tab_title(node_id)?)?; - created_tabs_wrapper = Some(tabs_id); - tabs_id + let parent_id = self.node_parent(node_id)?; + let parent_is_tabs = matches!( + parent_id.map(|id| self.node(id)).transpose()?, + Some(Node::Tabs(_)) + ); + match parent_id.filter(|_| parent_is_tabs) { + Some(parent_id) => parent_id, + None => { + let tabs_id = + self.wrap_node_in_tabs(node_id, self.default_tab_title(node_id)?)?; + created_tabs_wrapper = Some(tabs_id); + tabs_id + } + } }; let insert_index = { let tabs = match self.node(tabs_id)? { @@ -2985,17 +2995,28 @@ impl ServerState { } } -fn next_id_after_max(ids: impl Iterator) -> u64 { +/// Smallest id strictly greater than every id in `ids`, or `1` when `ids` is empty. +/// +/// Returns `None` when the id space is exhausted: the maximum id is `u64::MAX`, +/// leaving no room to allocate a new id. Callers on the persistence-load boundary +/// surface this as an error instead of allowing it to wrap or panic. +fn next_id_after_max(ids: impl Iterator) -> Option { match ids.max() { - Some(max) => max.checked_add(1).unwrap_or_else(|| { - panic!( - "next_id_after_max allocator exhaustion: restored max id == u64::MAX, cannot allocate a new id" - ) - }), - None => 1, + Some(max) => max.checked_add(1), + None => Some(1), } } +/// Like [`next_id_after_max`], but turns id-space exhaustion into a descriptive +/// error naming the entity kind, for use when restoring a persisted workspace. +fn checked_next_id(ids: impl Iterator, kind: &str) -> Result { + next_id_after_max(ids).ok_or_else(|| { + MuxError::internal(format!( + "persisted workspace exhausts the {kind} id space (highest id is u64::MAX); cannot allocate new ids" + )) + }) +} + #[cfg(test)] mod tests { use super::*; @@ -3057,4 +3078,32 @@ mod tests { assert!(message.contains("floating window")); assert!(message.contains("not referenced by session")); } + + #[test] + fn next_id_after_max_reports_exhaustion_instead_of_panicking() { + assert_eq!(next_id_after_max(std::iter::empty()), Some(1)); + assert_eq!(next_id_after_max([3u64, 7, 5].into_iter()), Some(8)); + assert_eq!(next_id_after_max([u64::MAX].into_iter()), None); + } + + #[test] + fn from_persisted_rejects_exhausted_buffer_id_space() { + let mut state = ServerState::new(); + state.create_session("main"); + let buffer_id = state.create_buffer("shell", vec!["/bin/sh".to_owned()], None); + + let mut workspace = state.to_persisted(); + workspace + .buffers + .iter_mut() + .find(|buffer| buffer.id == buffer_id.0) + .expect("buffer persists") + .id = u64::MAX; + + let error = ServerState::from_persisted(workspace) + .expect_err("exhausted buffer id space should be rejected"); + let message = error.to_string(); + assert!(message.contains("buffer")); + assert!(message.contains("id space")); + } } diff --git a/crates/embers-server/src/terminal_backend.rs b/crates/embers-server/src/terminal_backend.rs index 796569da..99d72e93 100644 --- a/crates/embers-server/src/terminal_backend.rs +++ b/crates/embers-server/src/terminal_backend.rs @@ -10,7 +10,6 @@ use embers_core::{ ActivityState, CursorPosition, CursorShape, CursorState, PtySize, SnapshotLine, TerminalModes, TerminalSnapshot, }; -use tracing::error; #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct BackendMetadata { @@ -114,10 +113,13 @@ impl BackendEventProxy { impl EventListener for BackendEventProxy { fn send_event(&self, event: Event) { - let Ok(mut state) = self.state.lock() else { - error!(?event, "backend event lock poisoned"); - return; - }; + // Recover from a poisoned lock rather than dropping the update: the event + // state is plain data, and silently skipping title/bell writes would + // desync from the metadata/take_activity readers, which both recover. + let mut state = self + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); match event { Event::Title(title) => state.title = Some(title), @@ -149,14 +151,17 @@ impl Dimensions for BackendSize { } impl AlacrittyTerminalBackend { - pub fn new(size: PtySize) -> Self { + /// `max_scrollback_lines` is the per-buffer scrollback ceiling, resolved by + /// the caller (the runtime keeper) so this emulation layer stays independent + /// of server configuration and env. + pub fn new(size: PtySize, max_scrollback_lines: usize) -> Self { let events = Arc::new(Mutex::new(BackendEventState::default())); let dimensions = BackendSize { columns: size.cols as usize, screen_lines: size.rows as usize, }; let config = Config { - scrolling_history: 10_000, + scrolling_history: max_scrollback_lines, ..Config::default() }; @@ -317,7 +322,12 @@ impl TerminalBackend for AlacrittyTerminalBackend { } fn metadata(&self) -> BackendMetadata { - let state = self.events.lock().expect("backend event lock"); + // Recover from a poisoned lock rather than crashing the backend: the + // event state is plain data and a panic here would take down the buffer. + let state = self + .events + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); let modes = self.terminal_modes(); BackendMetadata { title: state.title.clone(), @@ -332,7 +342,10 @@ impl TerminalBackend for AlacrittyTerminalBackend { } fn take_activity(&mut self) -> ActivityState { - let mut state = self.events.lock().expect("backend event lock"); + let mut state = self + .events + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); if std::mem::take(&mut state.bell_pending) { ActivityState::Bell } else { @@ -365,8 +378,13 @@ mod tests { AlacrittyTerminalBackend, BackendDamage, BackendMetadata, BackendScrollbackSlice, RawByteRouter, TerminalBackend, }; + use crate::config::DEFAULT_MAX_SCROLLBACK_LINES; use embers_core::{ActivityState, CursorShape, PtySize, TerminalSnapshot}; + fn backend(size: PtySize) -> AlacrittyTerminalBackend { + AlacrittyTerminalBackend::new(size, DEFAULT_MAX_SCROLLBACK_LINES) + } + #[derive(Default)] struct StubBackend { ingested: Vec, @@ -429,7 +447,7 @@ mod tests { #[test] fn visible_snapshot_extracts_plain_text_lines() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(8, 3)); + let mut backend = backend(PtySize::new(8, 3)); let _ = backend.take_damage(); backend.ingest_bytes(b"hello\r\nworld"); @@ -447,7 +465,7 @@ mod tests { #[test] fn carriage_return_overwrites_cells_without_advancing_the_row() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(8, 2)); + let mut backend = backend(PtySize::new(8, 2)); let _ = backend.take_damage(); backend.ingest_bytes(b"hello\rHEY"); @@ -458,7 +476,7 @@ mod tests { #[test] fn automatic_wrap_moves_following_bytes_to_the_next_row() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(4, 2)); + let mut backend = backend(PtySize::new(4, 2)); let _ = backend.take_damage(); backend.ingest_bytes(b"abcdX"); @@ -469,7 +487,7 @@ mod tests { #[test] fn erase_in_line_clears_trailing_cells_from_the_cursor() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(6, 1)); + let mut backend = backend(PtySize::new(6, 1)); let _ = backend.take_damage(); backend.ingest_bytes(b"abcdef\rabc\x1b[K"); @@ -480,7 +498,7 @@ mod tests { #[test] fn clear_screen_resets_visible_cells_before_new_output() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(6, 2)); + let mut backend = backend(PtySize::new(6, 2)); let _ = backend.take_damage(); backend.ingest_bytes(b"one\r\ntwo\x1b[2J\x1b[Hdone"); @@ -491,7 +509,7 @@ mod tests { #[test] fn scrollback_capture_preserves_history_beyond_viewport() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(6, 2)); + let mut backend = backend(PtySize::new(6, 2)); let _ = backend.take_damage(); backend.ingest_bytes(b"one\r\ntwo\r\nthree\r\nfour"); @@ -509,7 +527,7 @@ mod tests { #[test] fn scrollback_slice_returns_requested_window() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(6, 2)); + let mut backend = backend(PtySize::new(6, 2)); let _ = backend.take_damage(); backend.ingest_bytes(b"one\r\ntwo\r\nthree\r\nfour"); @@ -522,7 +540,7 @@ mod tests { #[test] fn damage_can_be_read_and_reset() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(6, 2)); + let mut backend = backend(PtySize::new(6, 2)); assert!(matches!(backend.take_damage(), BackendDamage::Full)); assert!(!matches!(backend.take_damage(), BackendDamage::Full)); @@ -534,7 +552,7 @@ mod tests { #[test] fn metadata_surfaces_terminal_modes_and_title() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(10, 2)); + let mut backend = backend(PtySize::new(10, 2)); let _ = backend.take_damage(); backend.ingest_bytes(b"\x1b]0;embers\x07\x1b[?1049h\x1b[?1000h\x1b[?1004h\x1b[?2004h"); @@ -549,7 +567,7 @@ mod tests { #[test] fn metadata_mode_flags_clear_when_disable_sequences_arrive() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(10, 2)); + let mut backend = backend(PtySize::new(10, 2)); let _ = backend.take_damage(); backend.ingest_bytes(b"\x1b[?1049h\x1b[?1000h\x1b[?1004h\x1b[?2004h"); @@ -569,7 +587,7 @@ mod tests { #[test] fn bell_activity_is_consumed_separately_from_metadata() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(10, 2)); + let mut backend = backend(PtySize::new(10, 2)); let _ = backend.take_damage(); backend.ingest_bytes(b"\x1b]0;embers\x07\x07"); @@ -599,7 +617,7 @@ mod tests { #[test] fn alternate_screen_visible_snapshot_tracks_active_screen_and_restores_primary_screen() { - let mut backend = AlacrittyTerminalBackend::new(PtySize::new(20, 4)); + let mut backend = backend(PtySize::new(20, 4)); let _ = backend.take_damage(); backend.ingest_bytes(b"main-one\r\nmain-two"); diff --git a/docs/configuration.md b/docs/configuration.md new file mode 100644 index 00000000..430c49e5 --- /dev/null +++ b/docs/configuration.md @@ -0,0 +1,90 @@ +# Operational Configuration + +This document covers the environment variables and command-line flags that +configure the Embers server and client at runtime. For the scriptable Rhai +config API (key bindings, modes, actions, themes, tab bars, hooks) see +[`config-api`](config-api/index.md). + +## Environment variables + +| Variable | Default | Applies to | Description | +| --- | --- | --- | --- | +| `EMBERS_SOCKET` | platform runtime socket (see below) | client + server | Path to the control socket. Overridden by `--socket`. | +| `EMBERS_CONFIG` | platform config dir | client | Path to the Rhai config file. Overridden by `--config`. | +| `EMBERS_LOG` | `info` | client + server | Tracing filter (e.g. `info`, `debug`, `embers=trace,info`). Overridden by `--log`/`-v`. Falls back to `RUST_LOG`. | +| `RUST_LOG` | unset | client + server | Standard tracing filter, used when `EMBERS_LOG` is unset. | +| `EMBERS_MAX_SESSIONS` | `256` | server | Ceiling on concurrently live sessions. | +| `EMBERS_MAX_BUFFERS` | `2048` | server | Ceiling on concurrently live buffers. Each buffer owns a PTY-backed child process plus scrollback, so this is the dominant resource bound. | +| `EMBERS_MAX_SCROLLBACK_LINES` | `10000` | server | Scrollback lines retained per buffer. | + +A limit set to `0`, empty, or an unparseable value is ignored, so the default +applies. + +## Socket path + +When `--socket`/`EMBERS_SOCKET` is not set, the socket is resolved to the first +available of: + +1. `$XDG_RUNTIME_DIR/embers/embers.sock`, when `XDG_RUNTIME_DIR` is set +2. `/run/user//embers/embers.sock`, when available +3. `/tmp/embers-/embers.sock`, as a fallback + +The default runtime directory is created with `0700` permissions and the socket +with `0600`, so other users on the host cannot reach it. + +## Logging + +The tracing filter is resolved from the first of these that is set, highest +precedence first: + +1. `--log ` (alias `--log-level`) +2. `-v` (`debug`) or `-vv` (`trace`) +3. `EMBERS_LOG` +4. `RUST_LOG` +5. `info` (default) + +`` accepts the full +[`tracing` env-filter](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/filter/struct.EnvFilter.html) +syntax, e.g. a bare level (`debug`) or per-target directives +(`embers_server=trace,info`). + +```sh +embers --log-level debug list-sessions +EMBERS_LOG=embers_server=trace,info embers +``` + +### Client vs. server logs + +Foreground invocations (interactive client, one-shot CLI commands) log to +`stderr`. + +The background server has no terminal, so it writes to a **daily-rotating log +file** in the socket's directory: + +- File name: `embers-server..log` (e.g. `embers-server.2026-06-20.log`) +- Rotation: daily +- Retention: the most recent 7 files; older files are pruned automatically +- Format: plain text (no ANSI styling) + +For the default socket this is, for example, +`$XDG_RUNTIME_DIR/embers/embers-server..log`. A `--log`/`-v` flag passed to +the launching command is propagated to the detached server via `EMBERS_LOG`, so +the same filter applies to both. Server panics are routed through tracing into +the same log file. + +## Resource limits + +The server enforces operator-tunable ceilings so a misbehaving or runaway client +cannot exhaust host resources by creating unbounded sessions, buffers, or +scrollback. When a limit is reached the offending request is rejected with an +error naming the limit and its override variable; existing sessions are +unaffected. + +Worst-case server memory from scrollback is bounded by roughly +`max_buffers × max_scrollback_lines × line_width`, so the buffer and scrollback +limits can be tuned together to fit a deployment. + +```sh +# Tighter limits for a constrained host +EMBERS_MAX_BUFFERS=256 EMBERS_MAX_SCROLLBACK_LINES=2000 embers +```