diff --git a/src/cli.rs b/src/cli.rs index 1db3dec..f6afad5 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -13,6 +13,24 @@ use serde_json::Value; use std::path::{Path, PathBuf}; use std::time::Duration; +/// What `memd up` should do given the two health signals. The MCP endpoint +/// answering is not enough: a wedged daemon (process alive, Meilisearch +/// unhealthy) must be restarted, not reported as "already running". +#[derive(Debug, PartialEq)] +enum UpAction { + AlreadyRunning, + RestartWedged, + Start, +} + +fn up_action(mcp_up: bool, meili_up: bool) -> UpAction { + match (mcp_up, meili_up) { + (true, true) => UpAction::AlreadyRunning, + (true, false) => UpAction::RestartWedged, + (false, _) => UpAction::Start, + } +} + /// Start the daemon. In foreground mode this *is* the daemon; otherwise it /// installs the launchd service (macOS) or spawns a detached process. pub async fn up(foreground: bool) -> Result<()> { @@ -21,12 +39,21 @@ pub async fn up(foreground: bool) -> Result<()> { } let cfg = Config::load_or_init()?; - if daemon_healthy(&cfg).await { - println!( - "memd is already running (MCP on http://{}:{}).", - cfg.mcp.host, cfg.mcp.port - ); - return Ok(()); + let svc = MemoryService::from_config(&cfg); + match up_action(daemon_healthy(&cfg).await, svc.client().is_healthy().await) { + UpAction::AlreadyRunning => { + println!( + "memd is already running (MCP on http://{}:{}).", + cfg.mcp.host, cfg.mcp.port + ); + return Ok(()); + } + UpAction::RestartWedged => { + println!("memd is running but Meilisearch is not responding — restarting the daemon…"); + down().await?; + tokio::time::sleep(Duration::from_millis(500)).await; + } + UpAction::Start => {} } if cfg!(target_os = "macos") { @@ -36,10 +63,10 @@ pub async fn up(foreground: bool) -> Result<()> { println!("Started memd daemon in the background."); } - // Wait for the MCP endpoint to come up. + // Wait for both the MCP endpoint and Meilisearch to come up. print!("Waiting for daemon to become healthy"); for _ in 0..120 { - if daemon_healthy(&cfg).await { + if daemon_healthy(&cfg).await && svc.client().is_healthy().await { println!( "\nmemd is up. MCP: http://{}:{}/mcp", cfg.mcp.host, cfg.mcp.port @@ -1118,3 +1145,26 @@ async fn restart_service(cfg: &Config) -> Result<()> { } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn up_reports_already_running_only_when_fully_healthy() { + assert_eq!(up_action(true, true), UpAction::AlreadyRunning); + } + + #[test] + fn up_restarts_when_mcp_is_up_but_meilisearch_is_down() { + // The wedged state: daemon process alive, engine dead. "already + // running" here would leave memory broken. + assert_eq!(up_action(true, false), UpAction::RestartWedged); + } + + #[test] + fn up_starts_when_daemon_is_down() { + assert_eq!(up_action(false, false), UpAction::Start); + assert_eq!(up_action(false, true), UpAction::Start); + } +} diff --git a/src/crawler/mod.rs b/src/crawler/mod.rs index 75619b8..fade5b8 100644 --- a/src/crawler/mod.rs +++ b/src/crawler/mod.rs @@ -41,6 +41,7 @@ pub async fn scan(cfg: &Config, svc: &MemoryService) -> Result { let mut summary = CrawlSummary::default(); let mut seen: HashSet = HashSet::new(); let mut batch: Vec = Vec::new(); + let mut warn_limit = crate::logging::LogLimiter::new(10); const BATCH: usize = 200; for root in cfg.expand_roots() { @@ -93,7 +94,11 @@ pub async fn scan(cfg: &Config, svc: &MemoryService) -> Result { } Ok(None) => summary.skipped += 1, Err(e) => { - tracing::warn!("preparing {path_str} failed: {e}"); + // Cap the per-file noise: with the engine down, a + // scan can fail on every single file. + if warn_limit.should_log() { + tracing::warn!("preparing {path_str} failed: {e}"); + } summary.errors += 1; } } @@ -104,6 +109,12 @@ pub async fn scan(cfg: &Config, svc: &MemoryService) -> Result { } } flush(svc, &mut batch, &mut summary).await; + if warn_limit.suppressed() > 0 { + tracing::warn!( + "{} more file preparation failures not logged individually", + warn_limit.suppressed() + ); + } // Deletions: drop crawler docs whose source file no longer matches. summary.deleted = reconcile_deletions(svc, &seen).await.unwrap_or(0); diff --git a/src/daemon.rs b/src/daemon.rs index 958b081..627b803 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -4,7 +4,7 @@ use crate::config::Config; use crate::memory::MemoryService; -use crate::{crawler, mcp, meili, paths, update}; +use crate::{crawler, logging, mcp, meili, paths, update}; use anyhow::{Context, Result}; use std::time::Duration; @@ -31,8 +31,10 @@ pub async fn serve() -> Result<()> { update::engine::Applied::None => {} } - // 1. Start the managed Meilisearch child process. + // 1. Start the managed Meilisearch child process. Its output flows through + // our size-capped log writer, not an unbounded supervisor redirect. let mut child = meili::spawn(&cfg).await?; + meili::forward_output(&mut child); let svc = MemoryService::from_config(&cfg); // 2. Wait for health, then ensure the index + local embedder. @@ -47,7 +49,30 @@ pub async fn serve() -> Result<()> { .context("ensuring memory_events index")?; tracing::info!("Meilisearch ready; indexes configured"); - // 3. Crawler + watcher in the background. + // Restart channel: the updater and the health watchdog both use it. + let (restart_tx, mut restart_rx) = tokio::sync::mpsc::channel::<&'static str>(1); + + // 3. Watchdog: the child staying alive is not enough — Meilisearch can + // wedge (process up, every request failing, e.g. after disk pressure). + // Restart the daemon when health checks fail long enough. + let watchdog_client = svc.client().clone(); + let watchdog_tx = restart_tx.clone(); + let watchdog = tokio::spawn(async move { + let mut detector = meili::WedgeDetector::new(4); + let mut ticks = tokio::time::interval(Duration::from_secs(30)); + ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticks.tick().await; // skip the immediate first tick + loop { + ticks.tick().await; + if detector.observe(watchdog_client.is_healthy().await) { + tracing::error!("Meilisearch is wedged (alive but unhealthy) — restarting"); + let _ = watchdog_tx.send("meilisearch wedged").await; + return; + } + } + }); + + // 4. Crawler + watcher in the background. let crawl_cfg = cfg.clone(); let crawl_svc = svc.clone(); let crawler_task = tokio::spawn(async move { @@ -56,7 +81,7 @@ pub async fn serve() -> Result<()> { } }); - // 4. HTTP MCP endpoint. + // 5. HTTP MCP endpoint. let addr = format!("{}:{}", cfg.mcp.host, cfg.mcp.port); let listener = tokio::net::TcpListener::bind(&addr) .await @@ -70,20 +95,20 @@ pub async fn serve() -> Result<()> { } }); - // 5. Daily auto-update check; a prepared update requests a restart. - let (restart_tx, mut restart_rx) = tokio::sync::mpsc::channel::<&'static str>(1); + // 6. Daily auto-update check; a prepared update requests a restart. let upd_cfg = cfg.clone(); let updater = tokio::spawn(async move { update::run_loop(upd_cfg, restart_tx).await; }); - // 6. Wait for a shutdown signal, child exit, or a restart request. + // 7. Wait for a shutdown signal, child exit, or a restart request. let restart = shutdown_signal(&mut child, &mut restart_rx).await; tracing::info!("shutting down"); crawler_task.abort(); server.abort(); updater.abort(); + watchdog.abort(); let _ = child.kill().await; let _ = std::fs::remove_file(paths::pid_file()?); @@ -151,13 +176,15 @@ async fn shutdown_signal( } } -/// Configure tracing to append to the daemon log file. The returned guard must -/// be kept alive for logs to flush. +/// Configure tracing to append to the daemon log file, size-capped: the file +/// is rotated to `memd.log.1` when it outgrows [`logging::MAX_LOG_BYTES`], so +/// the log can never fill the disk again. The returned guard must be kept +/// alive for logs to flush. fn init_logging() -> Result { let log_path = paths::log_file()?; - let dir = log_path.parent().unwrap().to_path_buf(); - let file_name = log_path.file_name().unwrap().to_string_lossy().to_string(); - let appender = tracing_appender::rolling::never(dir, file_name); + // Bound a log left oversized by a previous run before opening it. + logging::rotate_if_oversized(&log_path, logging::MAX_LOG_BYTES)?; + let appender = logging::SizeRotatingWriter::new(log_path, logging::MAX_LOG_BYTES)?; let (writer, guard) = tracing_appender::non_blocking(appender); use tracing_subscriber::EnvFilter; diff --git a/src/logging.rs b/src/logging.rs new file mode 100644 index 0000000..96b2b2f --- /dev/null +++ b/src/logging.rs @@ -0,0 +1,211 @@ +//! Daemon log management: a size-capped rotating file writer and helpers to +//! keep `memd.log` bounded. Rotation keeps exactly one previous file +//! (`memd.log.1`), so disk use never exceeds ~2× the cap. + +use anyhow::Result; +use std::io::Write; +use std::path::{Path, PathBuf}; + +/// Maximum size of the active log file before it is rotated. +pub const MAX_LOG_BYTES: u64 = 50 * 1024 * 1024; + +/// Rename `path` to `path.1` (replacing any previous rotation). +fn rotate(path: &Path) -> std::io::Result<()> { + let mut rotated = path.as_os_str().to_owned(); + rotated.push(".1"); + std::fs::rename(path, PathBuf::from(rotated)) +} + +/// Rotate `path` out of the way if it has grown past `max_bytes`. Called once +/// at daemon startup, before any writer opens the file, so a log left oversized +/// by a previous run (or by the pre-rotation era) is bounded immediately. +pub fn rotate_if_oversized(path: &Path, max_bytes: u64) -> Result<()> { + if let Ok(meta) = std::fs::metadata(path) + && meta.len() > max_bytes + { + rotate(path)?; + } + Ok(()) +} + +/// A `Write` impl that appends to `path` and rotates it to `path.1` once +/// `max_bytes` is exceeded, then continues in a fresh file. It owns its file +/// handle and re-opens after rotation, so rotation works even while writing. +pub struct SizeRotatingWriter { + path: PathBuf, + max_bytes: u64, + file: std::fs::File, + written: u64, +} + +impl SizeRotatingWriter { + pub fn new(path: PathBuf, max_bytes: u64) -> Result { + let file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&path)?; + let written = file.metadata().map(|m| m.len()).unwrap_or(0); + Ok(Self { + path, + max_bytes, + file, + written, + }) + } + + fn rotate_and_reopen(&mut self) -> std::io::Result<()> { + self.file.flush()?; + rotate(&self.path)?; + self.file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&self.path)?; + self.written = 0; + Ok(()) + } +} + +impl Write for SizeRotatingWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if self.written >= self.max_bytes { + // Best-effort: a failed rotation must not take logging down. + let _ = self.rotate_and_reopen(); + } + let n = self.file.write(buf)?; + self.written += n as u64; + Ok(n) + } + + fn flush(&mut self) -> std::io::Result<()> { + self.file.flush() + } +} + +/// Caps how many times a repetitive event is logged. The first `max` calls to +/// [`LogLimiter::should_log`] return true; everything after is counted as +/// suppressed so the caller can emit one roll-up line at the end. +pub struct LogLimiter { + max: usize, + seen: usize, +} + +impl LogLimiter { + pub fn new(max: usize) -> Self { + Self { max, seen: 0 } + } + + pub fn should_log(&mut self) -> bool { + self.seen += 1; + self.seen <= self.max + } + + /// How many events were suppressed (0 while under the cap). + pub fn suppressed(&self) -> usize { + self.seen.saturating_sub(self.max) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tmp_dir(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("memd-logging-test-{name}")); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + #[test] + fn writer_rotates_when_cap_exceeded() { + let dir = tmp_dir("rotate"); + let log = dir.join("memd.log"); + let mut w = SizeRotatingWriter::new(log.clone(), 100).unwrap(); + + // 20 lines of 10 bytes = 200 bytes: must rotate at least once. + for _ in 0..20 { + w.write_all(b"0123456789").unwrap(); + } + w.flush().unwrap(); + + let rotated = dir.join("memd.log.1"); + assert!(rotated.exists(), "expected {} to exist", rotated.display()); + assert!( + std::fs::metadata(&log).unwrap().len() <= 110, + "active log should have been reset by rotation" + ); + } + + #[test] + fn writer_keeps_only_one_rotated_file() { + let dir = tmp_dir("keep-one"); + let log = dir.join("memd.log"); + let mut w = SizeRotatingWriter::new(log.clone(), 50).unwrap(); + + for _ in 0..40 { + w.write_all(b"0123456789").unwrap(); + } + w.flush().unwrap(); + + // Multiple rotations happened; only .1 may exist. + assert!(dir.join("memd.log.1").exists()); + assert!(!dir.join("memd.log.2").exists()); + assert!(!dir.join("memd.log.1.1").exists()); + } + + #[test] + fn writer_counts_preexisting_bytes_toward_cap() { + let dir = tmp_dir("preexisting"); + let log = dir.join("memd.log"); + std::fs::write(&log, vec![b'x'; 90]).unwrap(); + + let mut w = SizeRotatingWriter::new(log.clone(), 100).unwrap(); + w.write_all(b"0123456789").unwrap(); // reaches 100 + w.write_all(b"after").unwrap(); // must land in a fresh file + w.flush().unwrap(); + + assert!(dir.join("memd.log.1").exists()); + assert_eq!(std::fs::read_to_string(&log).unwrap(), "after"); + } + + #[test] + fn rotate_if_oversized_moves_large_file() { + let dir = tmp_dir("startup"); + let log = dir.join("memd.log"); + std::fs::write(&log, vec![b'x'; 200]).unwrap(); + + rotate_if_oversized(&log, 100).unwrap(); + + assert!(!log.exists()); + assert_eq!( + std::fs::metadata(dir.join("memd.log.1")).unwrap().len(), + 200 + ); + } + + #[test] + fn rotate_if_oversized_leaves_small_and_missing_files_alone() { + let dir = tmp_dir("startup-small"); + let log = dir.join("memd.log"); + std::fs::write(&log, b"tiny").unwrap(); + + rotate_if_oversized(&log, 100).unwrap(); + assert!(log.exists()); + assert!(!dir.join("memd.log.1").exists()); + + // A missing file is fine too. + rotate_if_oversized(&dir.join("nope.log"), 100).unwrap(); + } + + #[test] + fn limiter_allows_first_n_then_suppresses() { + let mut l = LogLimiter::new(3); + assert!(l.should_log()); + assert!(l.should_log()); + assert!(l.should_log()); + assert_eq!(l.suppressed(), 0); + assert!(!l.should_log()); + assert!(!l.should_log()); + assert_eq!(l.suppressed(), 2); + } +} diff --git a/src/main.rs b/src/main.rs index ac95d80..91ef3eb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,6 +11,7 @@ mod crawler; mod daemon; mod history; mod launchd; +mod logging; mod mcp; mod meili; mod memory; diff --git a/src/meili/mod.rs b/src/meili/mod.rs index 3c9105e..aba9726 100644 --- a/src/meili/mod.rs +++ b/src/meili/mod.rs @@ -104,12 +104,123 @@ pub async fn download_binary(version: &str) -> Result { /// Spawn the managed Meilisearch as a child process. /// -/// `stdout`/`stderr` are inherited so the daemon's log redirection captures -/// them. The caller owns the returned [`Child`] and its lifecycle. +/// `stdout`/`stderr` are piped; callers must drain them (see +/// [`forward_output`]) or the child blocks once the pipe buffer fills. The +/// caller owns the returned [`Child`] and its lifecycle. pub async fn spawn(cfg: &Config) -> Result { spawn_with_import(cfg, None).await } +/// Command-line arguments for the managed Meilisearch. `--log-level WARN` +/// keeps it from logging every HTTP request — at INFO an unhealthy instance +/// once grew the daemon log to 9 GB. +fn build_args( + cfg: &Config, + db: &std::path::Path, + dumps: &std::path::Path, + snapshots: &std::path::Path, + import_dump: Option<&std::path::Path>, +) -> Vec { + let addr = format!("{}:{}", cfg.meilisearch.host, cfg.meilisearch.port); + let mut args: Vec = vec![ + "--db-path".into(), + db.into(), + "--dump-dir".into(), + dumps.into(), + "--snapshot-dir".into(), + snapshots.into(), + "--http-addr".into(), + addr.into(), + "--master-key".into(), + cfg.meilisearch.master_key.clone().into(), + "--no-analytics".into(), + "--env".into(), + "production".into(), + "--log-level".into(), + "WARN".into(), + ]; + if let Some(dump) = import_dump { + args.push("--import-dump".into()); + args.push(dump.into()); + } + args +} + +/// Drain the child's stdout/stderr into our tracing log so Meilisearch output +/// goes through the daemon's size-capped writer instead of an unbounded +/// supervisor redirect. Must be called once per spawned child. +pub fn forward_output(child: &mut Child) { + use tokio::io::{AsyncBufReadExt, BufReader}; + let streams: [Option>; 2] = [ + child + .stdout + .take() + .map(|s| Box::new(s) as Box), + child + .stderr + .take() + .map(|s| Box::new(s) as Box), + ]; + for stream in streams.into_iter().flatten() { + tokio::spawn(async move { + let mut lines = BufReader::new(stream).lines(); + while let Ok(Some(line)) = lines.next_line().await { + tracing::info!(target: "meilisearch", "{}", strip_ansi(&line)); + } + }); + } +} + +/// Remove ANSI SGR escape sequences (`ESC [ … m`) — Meilisearch colors its +/// log lines even when writing to a pipe, and raw escapes would litter our +/// log file. +fn strip_ansi(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + let mut chars = s.chars(); + while let Some(c) = chars.next() { + if c == '\x1b' { + // Skip "[m"; tolerate a bare ESC by dropping just it. + if chars.clone().next() == Some('[') { + for c2 in chars.by_ref() { + if c2 == 'm' { + break; + } + } + } + } else { + out.push(c); + } + } + out +} + +/// Detects a wedged engine: the process is alive but health checks keep +/// failing. Fires once `threshold` consecutive checks have failed; a healthy +/// check resets the streak. +pub struct WedgeDetector { + threshold: u32, + consecutive: u32, +} + +impl WedgeDetector { + pub fn new(threshold: u32) -> Self { + Self { + threshold, + consecutive: 0, + } + } + + /// Record one health-check result. Returns true while wedged. + pub fn observe(&mut self, healthy: bool) -> bool { + if healthy { + self.consecutive = 0; + } else { + self.consecutive = self.consecutive.saturating_add(1); + } + self.consecutive >= self.threshold + } +} + /// Like [`spawn`], but optionally boot with `--import-dump` (used by engine /// migration; requires an empty database directory). pub async fn spawn_with_import( @@ -129,28 +240,17 @@ pub async fn spawn_with_import( let snapshots = data.join("snapshots"); std::fs::create_dir_all(&snapshots)?; - let addr = format!("{}:{}", cfg.meilisearch.host, cfg.meilisearch.port); - tracing::info!("starting Meilisearch on {addr}"); - - let mut cmd = Command::new(&bin); - cmd.current_dir(&data) - .arg("--db-path") - .arg(&db) - .arg("--dump-dir") - .arg(&dumps) - .arg("--snapshot-dir") - .arg(&snapshots) - .arg("--http-addr") - .arg(&addr) - .arg("--master-key") - .arg(&cfg.meilisearch.master_key) - .arg("--no-analytics") - .arg("--env") - .arg("production"); - if let Some(dump) = import_dump { - cmd.arg("--import-dump").arg(dump); - } - let child = cmd + tracing::info!( + "starting Meilisearch on {}:{}", + cfg.meilisearch.host, + cfg.meilisearch.port + ); + + let child = Command::new(&bin) + .current_dir(&data) + .args(build_args(cfg, &db, &dumps, &snapshots, import_dump)) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) .kill_on_drop(true) .spawn() .with_context(|| format!("spawning {}", bin.display()))?; @@ -168,3 +268,66 @@ pub async fn wait_healthy(client: &MeiliClient, timeout: Duration) -> Result<()> } bail!("Meilisearch did not become healthy within {:?}", timeout) } + +#[cfg(test)] +mod tests { + use super::*; + use std::path::Path; + + #[test] + fn wedge_detector_fires_after_threshold_consecutive_failures() { + let mut d = WedgeDetector::new(3); + assert!(!d.observe(false)); + assert!(!d.observe(false)); + assert!(d.observe(false), "third consecutive failure = wedged"); + assert!(d.observe(false), "stays wedged while failures continue"); + } + + #[test] + fn wedge_detector_resets_on_healthy() { + let mut d = WedgeDetector::new(2); + assert!(!d.observe(false)); + assert!(!d.observe(true), "healthy check resets the streak"); + assert!(!d.observe(false)); + assert!(d.observe(false)); + } + + #[test] + fn spawn_args_cap_meilisearch_log_level() { + let cfg = Config::default(); + let args = build_args( + &cfg, + Path::new("/db"), + Path::new("/dumps"), + Path::new("/snapshots"), + None, + ); + let pos = args + .iter() + .position(|a| a == "--log-level") + .expect("--log-level must be passed so Meilisearch does not log every request"); + assert_eq!(args[pos + 1], "WARN"); + } + + #[test] + fn strip_ansi_removes_color_codes_and_keeps_text() { + assert_eq!( + strip_ansi("\x1b[2m2026-07-15T23:48:05Z\x1b[0m \x1b[33m WARN\x1b[0m boom"), + "2026-07-15T23:48:05Z WARN boom" + ); + assert_eq!(strip_ansi("plain text"), "plain text"); + } + + #[test] + fn spawn_args_include_import_dump_when_given() { + let cfg = Config::default(); + let args = build_args( + &cfg, + Path::new("/db"), + Path::new("/dumps"), + Path::new("/snapshots"), + Some(Path::new("/dumps/x.dump")), + ); + assert!(args.iter().any(|a| a == "--import-dump")); + } +} diff --git a/src/update/engine.rs b/src/update/engine.rs index 96336bf..b49a761 100644 --- a/src/update/engine.rs +++ b/src/update/engine.rs @@ -217,6 +217,7 @@ async fn import_and_verify(cfg: &Config, m: &Migration) -> Result<()> { let mut new_cfg = cfg.clone(); new_cfg.meilisearch.version = m.to.clone(); let mut child = meili::spawn_with_import(&new_cfg, Some(&m.dump_path)).await?; + meili::forward_output(&mut child); let client = MeiliClient::new(new_cfg.meili_url(), new_cfg.meilisearch.master_key.clone()); let result = async {