From 33d7085e97e700741bad3a9fe387f04fa4542254 Mon Sep 17 00:00:00 2001 From: NotAShelf Date: Thu, 30 Jul 2026 12:54:35 +0300 Subject: [PATCH 1/9] evix: replay completed initial evaluation events Signed-off-by: NotAShelf Change-Id: I8e42eec9749fa8b200bc49545a2987ff6a6a6964 --- crates/evix/src/error.rs | 5 ++ crates/evix/src/session.rs | 145 +++++++++++++++++++++++++++++++------ crates/evix/src/state.rs | 1 + 3 files changed, 129 insertions(+), 22 deletions(-) diff --git a/crates/evix/src/error.rs b/crates/evix/src/error.rs index f712415..9556928 100644 --- a/crates/evix/src/error.rs +++ b/crates/evix/src/error.rs @@ -7,6 +7,8 @@ pub enum Error { /// `Session::stream` was requested after the single-use stream had already /// started or completed. SessionStreamConsumed, + /// `Session::replay` was requested after the session's result changed. + SessionReplayUnavailable, /// A warm-graph operation was requested before initial evaluation completed. InitialEvaluationIncomplete { operation: &'static str }, /// A session operation requires completion, but evaluation is still running. @@ -59,6 +61,9 @@ impl fmt::Display for Error { Self::SessionStreamConsumed => { write!(f, "session stream has already been consumed") }, + Self::SessionReplayUnavailable => { + write!(f, "session has no replayable completed evaluation") + }, Self::InitialEvaluationIncomplete { operation } => { write!( f, diff --git a/crates/evix/src/session.rs b/crates/evix/src/session.rs index 886b573..215812b 100644 --- a/crates/evix/src/session.rs +++ b/crates/evix/src/session.rs @@ -37,11 +37,12 @@ use crate::{ /// later stream/query/diff calls report the stored failure instead of retrying /// implicitly. pub struct Session { - config: Config, - cancel: Arc, - state: Arc>, - completed: Arc, - initial: Arc>, + config: Config, + cancel: Arc, + state: Arc>, + completed: Arc, + initial: Arc>, + replay_available: AtomicBool, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -60,6 +61,7 @@ impl Session { state: Arc::new(RwLock::new(WarmState::default())), completed: Arc::new(Notify::new()), initial: Arc::new(Mutex::new(InitialEvaluation::Idle)), + replay_available: AtomicBool::new(true), }) } @@ -207,6 +209,7 @@ impl Session { /// fresh evaluation for each filesystem notification and diffs it against /// the previous warm graph. pub fn watch(&self) -> impl Stream> + '_ { + self.replay_available.store(false, Ordering::Release); let (tx, rx) = futures_mpsc::unbounded(); let config = self.config.clone(); let cancel = Arc::clone(&self.cancel); @@ -274,6 +277,7 @@ impl Session { &self, capacity: usize, ) -> impl Stream> + '_ { + self.replay_available.store(false, Ordering::Release); let (mut tx, rx) = futures_mpsc::channel(bounded_capacity(capacity)); let config = self.config.clone(); let cancel = Arc::clone(&self.cancel); @@ -372,6 +376,32 @@ impl Session { self.query_snapshot(filter).await } + /// Replay events from the completed initial evaluation without evaluating + /// Nix. + /// + /// # Errors + /// + /// Returns [`Error::InitialEvaluationIncomplete`] until the initial run + /// completes, or [`Error::SessionReplayUnavailable`] after `watch` or a + /// successful `diff_once` changes the session result. + pub async fn replay(&self) -> Result> { + let state = self.state.read().await; + if !self.replay_available.load(Ordering::Acquire) { + return Err(Error::SessionReplayUnavailable); + } + if !state.completed { + if let Some(error) = &state.error { + return Err(Error::EvaluationFailed { + message: error.clone(), + }); + } + return Err(Error::InitialEvaluationIncomplete { + operation: "replay", + }); + } + Ok(state.events.clone()) + } + /// Perform one full re-evaluation and diff it against the warm graph. pub async fn diff_once(&self) -> Result { let previous = { @@ -401,6 +431,7 @@ impl Session { state.errors = errors; state.completed = true; state.error = None; + self.replay_available.store(false, Ordering::Release); } Ok(diff) } @@ -459,20 +490,31 @@ async fn evaluate_initial( cancel: Arc, state: Arc>, completed: Arc, - on_event: F, + mut on_event: F, ) -> AnyhowResult where F: FnMut(Event) -> Fut + Send + 'static, Fut: Future>, { debug!("starting session evaluation"); - let result = run::evaluate_async(config, Arc::clone(&cancel), on_event).await; + let events = Arc::new(Mutex::new(Vec::new())); + let recorded_events = Arc::clone(&events); + let result = run::evaluate_async(config, Arc::clone(&cancel), move |event| { + recorded_events + .lock() + .expect("session event cache poisoned") + .push(event.clone()); + on_event(event) + }) + .await; match result { Ok((graph, errors, RunOutcome::Completed)) => { let mut state = state.write().await; state.graph = graph; state.errors = errors; + state.events = + events.lock().expect("session event cache poisoned").clone(); state.completed = true; state.error = None; completed.notify_waiters(); @@ -630,6 +672,62 @@ mod tests { })); } + #[test] + fn replay_returns_completed_events_without_starting_evaluation() { + tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .unwrap() + .block_on(async { + let session = Session::open(Config::default()).await.unwrap(); + { + let mut state = session.state.write().await; + state.events.push(Event::AttrSet { + attr: "jobs".into(), + attr_path: vec!["jobs".into()], + attrs: vec!["hello".into()], + }); + state.completed = true; + } + + let events = session.replay().await.unwrap(); + assert_eq!(events.len(), 1); + assert!( + matches!(&events[0], Event::AttrSet { attr, .. } if attr == "jobs") + ); + + session.replay_available.store(false, Ordering::Release); + assert!(matches!( + session.replay().await, + Err(Error::SessionReplayUnavailable) + )); + }); + } + + #[test] + fn replay_rechecks_availability_after_waiting_for_state() { + tokio::runtime::Builder::new_current_thread() + .build() + .unwrap() + .block_on(async { + let session = Session::open(Config::default()).await.unwrap(); + let mut state = session.state.write().await; + state.completed = true; + state.events.push(Event::AttrSet { + attr: "jobs".into(), + attr_path: vec!["jobs".into()], + attrs: vec![], + }); + + let mut replay = Box::pin(session.replay()); + assert!(futures_util::poll!(&mut replay).is_pending()); + session.replay_available.store(false, Ordering::Release); + drop(state); + + assert!(matches!(replay.await, Err(Error::SessionReplayUnavailable))); + }); + } + #[test] fn failed_initial_evaluation_is_reported_after_stream_consumed() { let runtime = tokio::runtime::Builder::new_current_thread() @@ -668,11 +766,12 @@ mod tests { .build() .unwrap(); let session = Session { - config: Config::default(), - cancel: Arc::new(AtomicBool::new(false)), - state: Arc::new(RwLock::new(WarmState::default())), - completed: Arc::new(Notify::new()), - initial: Arc::new(Mutex::new(InitialEvaluation::Finished)), + config: Config::default(), + cancel: Arc::new(AtomicBool::new(false)), + state: Arc::new(RwLock::new(WarmState::default())), + completed: Arc::new(Notify::new()), + initial: Arc::new(Mutex::new(InitialEvaluation::Finished)), + replay_available: AtomicBool::new(true), }; let mut stream = Box::pin(session.stream()); @@ -688,11 +787,12 @@ mod tests { .build() .unwrap(); let session = Session { - config: Config::default(), - cancel: Arc::new(AtomicBool::new(false)), - state: Arc::new(RwLock::new(WarmState::default())), - completed: Arc::new(Notify::new()), - initial: Arc::new(Mutex::new(InitialEvaluation::Finished)), + config: Config::default(), + cancel: Arc::new(AtomicBool::new(false)), + state: Arc::new(RwLock::new(WarmState::default())), + completed: Arc::new(Notify::new()), + initial: Arc::new(Mutex::new(InitialEvaluation::Finished)), + replay_available: AtomicBool::new(true), }; let mut stream = Box::pin(session.stream_bounded(1)); @@ -711,11 +811,12 @@ mod tests { #[test] fn cancel_sets_session_cancellation_flag() { let session = Session { - config: Config::default(), - cancel: Arc::new(AtomicBool::new(false)), - state: Arc::new(RwLock::new(WarmState::default())), - completed: Arc::new(Notify::new()), - initial: Arc::new(Mutex::new(InitialEvaluation::Idle)), + config: Config::default(), + cancel: Arc::new(AtomicBool::new(false)), + state: Arc::new(RwLock::new(WarmState::default())), + completed: Arc::new(Notify::new()), + initial: Arc::new(Mutex::new(InitialEvaluation::Idle)), + replay_available: AtomicBool::new(true), }; session.cancel(); diff --git a/crates/evix/src/state.rs b/crates/evix/src/state.rs index bfbe147..527520e 100644 --- a/crates/evix/src/state.rs +++ b/crates/evix/src/state.rs @@ -8,6 +8,7 @@ pub type EvalGraph = HashMap, Derivation>; pub struct WarmState { pub graph: EvalGraph, pub errors: Vec, + pub events: Vec, pub completed: bool, pub error: Option, } From 34a426c6cbba253b059c78dca5bc92aa4e70ff81 Mon Sep 17 00:00:00 2001 From: NotAShelf Date: Fri, 2 Oct 2026 21:04:14 +0300 Subject: [PATCH 2/9] daemon: expose cached evaluation replay through the CLI Signed-off-by: NotAShelf Change-Id: Ic801c5ec00560d18d48d2ace038b30c26a6a6964 --- crates/evix-cli/src/args.rs | 32 +++++++ crates/evix-cli/src/main.rs | 25 ++++-- crates/evix-cli/tests/eval.rs | 106 +++++++++++++++++++++++- crates/evix-daemon/src/lib.rs | 25 +++++- crates/evix-daemon/src/session_cache.rs | 92 ++++++++++++++++++-- crates/evix-daemon/src/tests.rs | 12 +-- crates/evix-protocol/src/lib.rs | 45 +++++++++- 7 files changed, 306 insertions(+), 31 deletions(-) diff --git a/crates/evix-cli/src/args.rs b/crates/evix-cli/src/args.rs index 2537946..419c1ba 100644 --- a/crates/evix-cli/src/args.rs +++ b/crates/evix-cli/src/args.rs @@ -61,6 +61,9 @@ enum Commands { socket: Option, #[pound(long)] no_daemon: bool, + /// Replay a completed matching daemon evaluation without evaluating Nix. + #[pound(long)] + replay: bool, }, #[pound(required_group = "input")] @@ -268,6 +271,7 @@ pub enum CommandPlan { config: Config, socket: Option, use_daemon: bool, + replay: bool, }, Watch { config: Config, @@ -346,7 +350,11 @@ fn command_plan(command: Commands) -> Result { gc_roots_dir, socket, no_daemon, + replay, } => { + if replay && no_daemon { + bail!("--replay requires the daemon") + } Ok(CommandPlan::Eval { config: config(EvalInput { flake, @@ -369,6 +377,7 @@ fn command_plan(command: Commands) -> Result { })?, socket, use_daemon: !no_daemon, + replay, }) }, Commands::Watch { @@ -864,6 +873,29 @@ mod tests { assert_eq!(config.item_timeout_seconds, 7); } + #[test] + fn eval_replay_requires_daemon() { + let (_, CommandPlan::Eval { replay, .. }) = + parse_plan_from(["eval", "--expr", "{}", "--replay"]) + .expect("parse replay eval plan") + else { + panic!("expected eval plan"); + }; + assert!(replay); + + let error = match parse_plan_from([ + "eval", + "--expr", + "{}", + "--replay", + "--no-daemon", + ]) { + Ok(_) => panic!("expected replay without daemon to fail"), + Err(error) => error.to_string(), + }; + assert!(error.contains("--replay requires the daemon")); + } + #[test] fn eval_rejects_zero_item_timeout_seconds() { let error = match parse_plan_from([ diff --git a/crates/evix-cli/src/main.rs b/crates/evix-cli/src/main.rs index 2dfdb1c..ee73155 100644 --- a/crates/evix-cli/src/main.rs +++ b/crates/evix-cli/src/main.rs @@ -60,13 +60,20 @@ fn run_plan(plan: CommandPlan) -> Result<()> { config, socket, use_daemon, + replay, } => { if use_daemon { - run_client_or_local( - daemon_request(Request::eval(&wire_config(&config)))?, - socket, - || run_local_eval(&config), - ) + let request = if replay { + Request::replay(&wire_config(&config)) + } else { + Request::eval(&wire_config(&config)) + }; + let request = daemon_request(request)?; + if replay { + run_daemon_only(request, socket) + } else { + run_client_or_local(request, socket, || run_local_eval(&config)) + } } else { run_local_eval(&config) } @@ -120,6 +127,7 @@ fn run_plan(plan: CommandPlan) -> Result<()> { fn daemon_request(request: Request) -> Result { Ok(match request { Request::Eval { config, .. } => Request::eval(&daemon_config(config)?), + Request::Replay { config, .. } => Request::replay(&daemon_config(config)?), Request::Watch { config, .. } => Request::watch(&daemon_config(config)?), Request::Query { config, filter, .. } => { Request::query(&daemon_config(config)?, &filter) @@ -237,6 +245,13 @@ fn run_daemon_request(mut stream: UnixStream, request: &Request) -> Result<()> { return Ok(()); } }, + Response::Replay { lines } => { + for line in lines { + if write_output_line(&mut stdout, &line)? == OutputWrite::Closed { + return Ok(()); + } + } + }, Response::Diff { diff } => { if write_output_line(&mut stdout, &evix_json::diff_line(&diff))? == OutputWrite::Closed diff --git a/crates/evix-cli/tests/eval.rs b/crates/evix-cli/tests/eval.rs index 8ade9e2..23e1a9b 100644 --- a/crates/evix-cli/tests/eval.rs +++ b/crates/evix-cli/tests/eval.rs @@ -1,9 +1,10 @@ use std::{ io::Read as _, net::{TcpListener, TcpStream}, - process::{Child, Command, Output, Stdio}, + path::{Path, PathBuf}, + process::{self, Child, Command, Output, Stdio}, thread, - time::{Duration, Instant}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; fn evix() -> Command { @@ -177,6 +178,57 @@ fn run_with_timeout(command: &mut Command, limit: Duration) -> Output { } } +#[test] +fn daemon_replay_matches_eval_and_diff_invalidates_it() { + let socket = temporary_socket_path(); + let _daemon = spawn_daemon(&socket); + wait_for_socket(&socket); + let socket = socket.to_str().expect("socket path is UTF-8"); + let expression = "{ recurseForDerivations = true; hello = { \ + recurseForDerivations = true; leaf = 1; }; }"; + + let initial = evix() + .args(["eval", "--socket", socket, "--expr", expression]) + .output() + .expect("run daemon evaluation"); + assert!( + initial.status.success(), + "stderr:\n{}", + String::from_utf8_lossy(&initial.stderr) + ); + + let replay = evix() + .args(["eval", "--replay", "--socket", socket, "--expr", expression]) + .output() + .expect("run replay"); + assert!( + replay.status.success(), + "stderr:\n{}", + String::from_utf8_lossy(&replay.stderr) + ); + assert_eq!(replay.stdout, initial.stdout); + + let diff = evix() + .args(["diff", "--socket", socket, "--expr", expression]) + .output() + .expect("run diff"); + assert!( + diff.status.success(), + "stderr:\n{}", + String::from_utf8_lossy(&diff.stderr) + ); + + let replay_after_diff = evix() + .args(["eval", "--replay", "--socket", socket, "--expr", expression]) + .output() + .expect("run replay after diff"); + assert!(!replay_after_diff.status.success()); + assert!( + String::from_utf8_lossy(&replay_after_diff.stderr) + .contains("no replayable completed evaluation") + ); +} + fn unused_loopback_endpoint() -> String { let listener = TcpListener::bind("127.0.0.1:0").expect("bind test port"); let addr = listener.local_addr().expect("read test port"); @@ -194,6 +246,24 @@ fn spawn_worker(endpoint: &str, token: &str) -> Child { .expect("spawn evix worker") } +fn spawn_daemon(socket: &Path) -> DaemonGuard { + DaemonGuard { + child: evix() + .args([ + "daemon", + "--foreground", + "--socket", + socket.to_str().expect("socket path is UTF-8"), + ]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn daemon"), + socket: socket.to_owned(), + } +} + fn wait_for_worker(endpoint: &str) { for _ in 0..100 { if TcpStream::connect(endpoint).is_ok() { @@ -204,7 +274,39 @@ fn wait_for_worker(endpoint: &str) { panic!("worker did not listen on {endpoint}"); } +fn wait_for_socket(socket: &Path) { + for _ in 0..100 { + if socket.exists() { + return; + } + thread::sleep(Duration::from_millis(50)); + } + panic!("daemon did not create socket at {}", socket.display()); +} + +fn temporary_socket_path() -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time before UNIX epoch") + .as_nanos(); + std::env::temp_dir() + .join(format!("evix-daemon-{}-{nanos}.sock", process::id())) +} + fn stop_worker(worker: &mut Child) { let _ = worker.kill(); let _ = worker.wait(); } + +struct DaemonGuard { + child: Child, + socket: PathBuf, +} + +impl Drop for DaemonGuard { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = std::fs::remove_file(&self.socket); + } +} diff --git a/crates/evix-daemon/src/lib.rs b/crates/evix-daemon/src/lib.rs index 96ef063..bf71a78 100644 --- a/crates/evix-daemon/src/lib.rs +++ b/crates/evix-daemon/src/lib.rs @@ -1,7 +1,7 @@ use std::{ env, fs::{self, OpenOptions}, - io::{BufRead, BufReader, Write}, + io::{BufRead, BufReader, BufWriter, Write}, os::{ fd::{AsRawFd as _, RawFd}, unix::{ @@ -33,6 +33,8 @@ use session_cache::DaemonState; const MAX_REQUEST_BYTES: usize = 16 * 1024 * 1024; const MAX_CONNECTIONS: usize = 64; +// Keep replay responses bounded while avoiding one protocol message per event. +const REPLAY_LINES_PER_RESPONSE: usize = 128; static SHUTDOWN_REQUESTED: AtomicBool = AtomicBool::new(false); static SHUTDOWN_SIGNAL_WRITE_FD: AtomicI32 = AtomicI32::new(-1); @@ -625,6 +627,9 @@ fn handle_connection( Request::Eval { config, .. } => { handle_eval(&state, &mut stream, config.into()).await }, + Request::Replay { config, .. } => { + handle_replay(&state, &mut stream, config.into()).await + }, Request::Watch { config, .. } => { handle_watch(&state, &mut stream, config.into()).await }, @@ -743,6 +748,21 @@ async fn handle_eval( write_response(stream, &Response::Done) } +async fn handle_replay( + state: &DaemonState, + stream: &mut UnixStream, + config: Config, +) -> Result<()> { + let lines = state.replay_lines(&config).await?; + let mut writer = BufWriter::new(stream); + for lines in lines.chunks(REPLAY_LINES_PER_RESPONSE) { + write_response(&mut writer, &Response::Replay { + lines: lines.to_vec(), + })?; + } + write_response(&mut writer, &Response::Done) +} + async fn handle_watch( state: &DaemonState, stream: &mut UnixStream, @@ -781,11 +801,12 @@ async fn handle_diff( let session = state.warm_session(&config)?; session.require_completed().await?; let diff = session.diff_once().await?; + state.clear_replay(&config)?; write_response(stream, &Response::diff(&diff))?; write_response(stream, &Response::Done) } -fn write_response(stream: &mut UnixStream, response: &Response) -> Result<()> { +fn write_response(stream: &mut W, response: &Response) -> Result<()> { serde_json::to_writer(&mut *stream, response)?; writeln!(stream)?; stream.flush()?; diff --git a/crates/evix-daemon/src/session_cache.rs b/crates/evix-daemon/src/session_cache.rs index 19c6d87..291d063 100644 --- a/crates/evix-daemon/src/session_cache.rs +++ b/crates/evix-daemon/src/session_cache.rs @@ -12,7 +12,13 @@ const MAX_SESSIONS: usize = 32; #[derive(Default)] pub(crate) struct DaemonState { - pub(crate) sessions: Mutex>>, + sessions: Mutex>, +} + +#[derive(Clone)] +struct CachedSession { + session: Arc, + replay_lines: Option>, } impl DaemonState { @@ -26,7 +32,10 @@ impl DaemonState { .sessions .lock() .expect("daemon session registry poisoned") - .insert(key, Arc::clone(&session)); + .insert(key, CachedSession { + session: Arc::clone(&session), + replay_lines: None, + }); Ok(session) } @@ -36,13 +45,67 @@ impl DaemonState { .sessions .lock() .expect("daemon session registry poisoned"); - sessions.get(&key).ok_or_else(|| { - anyhow::anyhow!( - "no warm session for requested daemon config; query/diff reuse a \ - session only when all daemon-protocol config values match a \ - completed eval or watch" - ) - }) + sessions + .get(&key) + .map(|cached| Arc::clone(&cached.session)) + .ok_or_else(|| { + anyhow::anyhow!( + "no warm session for requested daemon config; query/diff reuse a \ + session only when all daemon-protocol config values match a \ + completed eval or watch" + ) + }) + } + + pub(crate) async fn replay_lines( + &self, + config: &Config, + ) -> Result> { + let key = session_key(config)?; + let session = { + let mut sessions = self + .sessions + .lock() + .expect("daemon session registry poisoned"); + let cached = sessions.get_mut(&key).ok_or_else(|| { + anyhow::anyhow!("no replayable session for requested daemon config") + })?; + if let Some(lines) = &cached.replay_lines { + return Ok(Arc::clone(lines)); + } + Arc::clone(&cached.session) + }; + + let lines: Arc<[String]> = session + .replay() + .await? + .iter() + .map(evix::json::event_line) + .collect::>() + .into(); + let mut sessions = self + .sessions + .lock() + .expect("daemon session registry poisoned"); + if let Some(cached) = sessions.get_mut(&key) + && Arc::ptr_eq(&cached.session, &session) + { + cached.replay_lines = Some(Arc::clone(&lines)); + } + Ok(lines) + } + + pub(crate) fn clear_replay(&self, config: &Config) -> Result<()> { + let key = session_key(config)?; + if let Some(cached) = self + .sessions + .lock() + .expect("daemon session registry poisoned") + .get_mut(&key) + { + cached.replay_lines = None; + } + Ok(()) } } @@ -95,6 +158,17 @@ impl SessionRegistry { } } +impl SessionRegistry { + fn get_mut(&mut self, key: &str) -> Option<&mut T> { + if !self.sessions.contains_key(key) { + return None; + } + self.remove_order_entry(key); + self.order.push_back(key.to_owned()); + self.sessions.get_mut(key) + } +} + pub(crate) fn session_key(config: &Config) -> Result { serde_json::to_string(&SessionKeyConfig::from(config)) .context("serializing session key") diff --git a/crates/evix-daemon/src/tests.rs b/crates/evix-daemon/src/tests.rs index bece9f7..376cbf0 100644 --- a/crates/evix-daemon/src/tests.rs +++ b/crates/evix-daemon/src/tests.rs @@ -3,8 +3,6 @@ use std::{ time::{SystemTime, UNIX_EPOCH}, }; -use evix::Session; - use super::*; use crate::session_cache::{SessionRegistry, session_key}; @@ -208,13 +206,9 @@ fn warm_session_rejects_daemon_protocol_field_variants() { let runtime = Builder::new_current_thread().build().unwrap(); let state = DaemonState::default(); let base = Config::expr("{}"); - let session = - Arc::new(runtime.block_on(Session::open(base.clone())).unwrap()); - state - .sessions - .lock() - .expect("daemon session registry poisoned") - .insert(session_key(&base).unwrap(), Arc::clone(&session)); + runtime + .block_on(state.replace_session(base.clone())) + .unwrap(); let mut query_config = base.clone(); query_config.workers = 8; diff --git a/crates/evix-protocol/src/lib.rs b/crates/evix-protocol/src/lib.rs index eda812a..5928575 100644 --- a/crates/evix-protocol/src/lib.rs +++ b/crates/evix-protocol/src/lib.rs @@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize}; #[doc(hidden)] pub mod serde_config; -pub const DAEMON_PROTOCOL_VERSION: u32 = 1; +pub const DAEMON_PROTOCOL_VERSION: u32 = 2; pub const DEFAULT_ITEM_TIMEOUT_SECONDS: u64 = 30 * 60; /// Input source for a Nix evaluation. @@ -367,6 +367,11 @@ pub enum Request { protocol_version: u32, config: Config, }, + Replay { + #[serde(rename = "protocolVersion")] + protocol_version: u32, + config: Config, + }, Watch { #[serde(rename = "protocolVersion")] protocol_version: u32, @@ -395,6 +400,15 @@ impl Request { } } + /// Create an evaluation request that replays a completed daemon session. + #[must_use] + pub fn replay(config: &Config) -> Self { + Self::Replay { + protocol_version: DAEMON_PROTOCOL_VERSION, + config: config.clone(), + } + } + #[must_use] pub fn watch(config: &Config) -> Self { Self::Watch { @@ -425,6 +439,9 @@ impl Request { Self::Eval { protocol_version, .. } + | Self::Replay { + protocol_version, .. + } | Self::Watch { protocol_version, .. } @@ -468,10 +485,20 @@ impl std::error::Error for ProtocolVersionError {} #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "camelCase")] pub enum Response { - Event { event: Event }, - Diff { diff: Diff }, + Event { + event: Event, + }, + /// Cached newline-delimited event output from a replay request. + Replay { + lines: Vec, + }, + Diff { + diff: Diff, + }, Done, - Error { message: String }, + Error { + message: String, + }, } impl Response { @@ -512,6 +539,16 @@ mod tests { .expect("serialize request"); assert_eq!(value["protocolVersion"], DAEMON_PROTOCOL_VERSION); + assert!(value.get("replay").is_none()); + } + + #[test] + fn replay_request_serializes_as_its_own_operation() { + let value = serde_json::to_value(Request::replay(&Config::default())) + .expect("serialize replay request"); + + assert_eq!(value["type"], "replay"); + assert_eq!(value["protocolVersion"], DAEMON_PROTOCOL_VERSION); } #[test] From b5436e72b1edac1822b2d058e0a364a28d9bd7ef Mon Sep 17 00:00:00 2001 From: NotAShelf Date: Fri, 2 Oct 2026 21:05:01 +0300 Subject: [PATCH 3/9] build: tag 3.0.0 Signed-off-by: NotAShelf Change-Id: I2d54424f11d6ecf0302cb4f77f9ffef06a6a6964 --- Cargo.lock | 8 ++++---- Cargo.toml | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 590d347..416ec8f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -139,7 +139,7 @@ dependencies = [ [[package]] name = "evix" -version = "2.1.1" +version = "3.0.0" dependencies = [ "anyhow", "capnp", @@ -161,7 +161,7 @@ dependencies = [ [[package]] name = "evix-cli" -version = "2.1.1" +version = "3.0.0" dependencies = [ "anyhow", "evix", @@ -178,7 +178,7 @@ dependencies = [ [[package]] name = "evix-daemon" -version = "2.1.1" +version = "3.0.0" dependencies = [ "anyhow", "evix", @@ -195,7 +195,7 @@ dependencies = [ [[package]] name = "evix-protocol" -version = "2.1.1" +version = "3.0.0" dependencies = [ "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index fc3a461..41dff60 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,13 +8,13 @@ edition = "2024" license = "EUPL-1.2" repository = "https://github.com/manic-systems/evix" rust-version = "1.95.0" -version = "2.1.1" +version = "3.0.0" [workspace.dependencies] # Workspace components -evix = { path = "crates/evix", version = "2.1.1" } -evix-daemon = { path = "crates/evix-daemon", version = "2.1.1" } -evix-protocol = { path = "crates/evix-protocol", version = "2.1.1" } +evix = { path = "crates/evix", version = "3.0.0" } +evix-daemon = { path = "crates/evix-daemon", version = "3.0.0" } +evix-protocol = { path = "crates/evix-protocol", version = "3.0.0" } # Other dependencies anyhow = "1.0.104" From 58a97be596b454d544771a52da7313e32034fea7 Mon Sep 17 00:00:00 2001 From: NotAShelf Date: Fri, 2 Oct 2026 21:05:38 +0300 Subject: [PATCH 4/9] bench: include eval replay in results Signed-off-by: NotAShelf Change-Id: Idc64451478d1e163ac4308ffbb599a206a6a6964 --- bench/README.md | 7 +++++-- bench/bench.sh | 7 ++++--- bench/results.md | 18 +++++++++--------- 3 files changed, 18 insertions(+), 14 deletions(-) diff --git a/bench/README.md b/bench/README.md index 7301006..d41b6e8 100644 --- a/bench/README.md +++ b/bench/README.md @@ -53,5 +53,8 @@ $ bench/bench.sh 5 3 Runs `hyperfine` over evix at 1/4/8 local-only workers, evix remote-only distributed evaluation, evix mixed local+remote distributed evaluation, -daemon-backed cold prewarming, warm full-graph daemon queries, filtered warm -daemon queries, and nix-eval-jobs. Writes `bench/results.md`. +daemon-backed cold prewarming, warm replay of a completed daemon event stream, +warm full-graph daemon queries, and nix-eval-jobs. These commands produce the +same full result, so their relative timings are comparable. The `hyperfine` +warmup materializes replay's cached NDJSON, so the replay result measures +subsequent calls. Writes `bench/results.md`. diff --git a/bench/bench.sh b/bench/bench.sh index 66ef94a..6684148 100755 --- a/bench/bench.sh +++ b/bench/bench.sh @@ -2,7 +2,8 @@ # Wall-clock benchmark of evix evaluation vs nix-eval-jobs on a fixed fixture, # using hyperfine. Covers evix local-only evaluation, remote-only distributed # evaluation, mixed local+remote distributed evaluation, daemon prewarming, -# warm daemon graph queries, and nix-eval-jobs as the reference. +# warm daemon graph queries and event replays, and nix-eval-jobs as the +# reference. # # Usage: bench/bench.sh [breadth] [depth] (defaults: breadth=6 depth=3) set -euo pipefail @@ -44,7 +45,7 @@ for _ in $(seq 1 100); do [ -S "$sock" ] && break sleep 0.05 done -"$evix" eval --socket "$sock" --workers 4$args_str >/dev/null +"$evix" eval --socket "$sock" --workers 4"$args_str" >/dev/null cmds=( -n "evix local=1" "$evix eval --no-daemon --workers 1$args_str >/dev/null" @@ -53,8 +54,8 @@ cmds=( -n "evix distributed remote=4" "$evix eval --no-daemon --workers 0 --insecure-tokenless-remote --remote 127.0.0.1:$port $sys 4$args_str >/dev/null" -n "evix distributed local=4 remote=4" "$evix eval --no-daemon --workers 4 --insecure-tokenless-remote --remote 127.0.0.1:$port $sys 4$args_str >/dev/null" -n "evix daemon prewarm local=4" "$evix eval --socket $sock --workers 4$args_str >/dev/null" + -n "evix daemon warm replay local=4" "$evix eval --replay --socket $sock --workers 4$args_str >/dev/null" -n "evix daemon warm query full local=4" "$evix query --socket $sock --workers 4$args_str >/dev/null" - -n "evix daemon warm query n0 local=4" "$evix query --socket $sock --workers 4 --attr-prefix n0$args_str >/dev/null" ) if command -v nix-eval-jobs >/dev/null; then cmds+=(-n "nix-eval-jobs w=4" "nix-eval-jobs --gc-roots-dir $gc --workers 4 $fixture$fargs_str >/dev/null") diff --git a/bench/results.md b/bench/results.md index eb2b8ee..da0e8e2 100644 --- a/bench/results.md +++ b/bench/results.md @@ -1,11 +1,11 @@ | Command | Mean [ms] | Min [ms] | Max [ms] | Relative | |:---|---:|---:|---:|---:| -| `evix local=1` | 263.3 ± 6.5 | 258.1 | 274.6 | 9.31 ± 0.32 | -| `evix local=4` | 157.4 ± 4.5 | 151.0 | 163.2 | 5.57 ± 0.21 | -| `evix local=8` | 153.2 ± 2.4 | 151.0 | 156.9 | 5.42 ± 0.16 | -| `evix distributed remote=4` | 198.3 ± 2.9 | 194.6 | 201.1 | 7.02 ± 0.20 | -| `evix distributed local=4 remote=4` | 168.4 ± 11.1 | 156.9 | 186.7 | 5.96 ± 0.42 | -| `evix daemon prewarm local=4` | 197.0 ± 6.5 | 186.5 | 203.1 | 6.97 ± 0.29 | -| `evix daemon warm query full local=4` | 103.3 ± 0.7 | 102.4 | 104.2 | 3.65 ± 0.09 | -| `evix daemon warm query n0 local=4` | 28.3 ± 0.7 | 27.6 | 29.4 | 1.00 | -| `nix-eval-jobs w=4` | 147.5 ± 7.6 | 135.4 | 154.3 | 5.22 ± 0.30 | +| `evix local=1` | 840.0 ± 17.0 | 813.4 | 855.5 | 21.22 ± 0.52 | +| `evix local=4` | 424.4 ± 17.4 | 397.4 | 439.6 | 10.72 ± 0.46 | +| `evix local=8` | 437.7 ± 22.9 | 419.9 | 477.5 | 11.06 ± 0.60 | +| `evix distributed remote=4` | 571.2 ± 23.9 | 543.0 | 599.9 | 14.43 ± 0.64 | +| `evix distributed local=4 remote=4` | 446.7 ± 24.4 | 418.2 | 483.8 | 11.28 ± 0.64 | +| `evix daemon prewarm local=4` | 532.2 ± 6.5 | 525.5 | 542.8 | 13.44 ± 0.25 | +| `evix daemon warm replay local=4` | 39.6 ± 0.6 | 39.0 | 40.5 | 1.00 | +| `evix daemon warm query full local=4` | 286.3 ± 2.4 | 283.6 | 289.3 | 7.23 ± 0.12 | +| `nix-eval-jobs w=4` | 438.6 ± 4.3 | 435.3 | 446.0 | 11.08 ± 0.19 | From 23d6bf55d34cd91b3bfeea08d338928725ddc787 Mon Sep 17 00:00:00 2001 From: NotAShelf Date: Fri, 2 Oct 2026 21:18:15 +0300 Subject: [PATCH 5/9] docs: introduce incremental evaluation, briefly Signed-off-by: NotAShelf Change-Id: I3ea3252c6f8c205ec1e8a41bfa9349196a6a6964 --- README.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/README.md b/README.md index 83be60f..d62c413 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,21 @@ If an `evixd` socket is available, `eval` uses it and stores a warm session for later `query` or `diff` calls. If the daemon is not running, `eval` falls back to local evaluation. Use `--no-daemon` to force local evaluation. +`evix eval --replay` is an experimental, incremental-like [^1] evaluation model +that serves a snapshot of a completed daemon evaluation. A normal daemon-backed +`eval` records the NDJSON event stream; a later replay with the same evaluation +configuration writes those events again without starting Nix or querying the +warm graph. The snapshot is replaced by a new `eval`, and invalidated by `diff` +or `watch`. This system is **opt-in**. It does not check whether source files, +environment variables, or other impure inputs changed, so use ordinary `eval` +when a fresh result is important to its success. It requires a running daemon +and fails if there is no completed matching session. + +[^1]: I can almost hear you asking _"what the hell is incremental-like?"_. Well, + we basically lack _any_ kind of API on Nix's end to make truly incremental + evaluation. This is the _closest_ we can get to that while retaining + meaningful performance gains. + > [!NOTE] > Local/path flakes are checked against their `flake.lock`. Non-local flake refs > are still locked virtually inside each worker, so Evix warns for those inputs; @@ -214,6 +229,7 @@ Like `query`, `diff` requires an existing warm daemon session. | `--force-recurse` | Recurse into all attrsets, ignoring `recurseForDerivations` | | `--gc-roots-dir DIR` | Register GC root symlinks for evaluated derivations | | `--socket PATH` | Daemon socket path for daemon-backed commands | +| `--replay` | Replay a completed matching daemon evaluation | | `-v`, `--verbose` | Increase logging verbosity, repeat for trace logs | | `-q`, `--quiet` | Decrease logging verbosity, repeat to suppress more logs | From 31d920a12e220942cafa72263adcbcd161855c31 Mon Sep 17 00:00:00 2001 From: Amaan Qureshi Date: Fri, 2 Oct 2026 16:18:11 -0400 Subject: [PATCH 6/9] daemon: write each response with a single syscall --- crates/evix-daemon/src/lib.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/evix-daemon/src/lib.rs b/crates/evix-daemon/src/lib.rs index bf71a78..4f4dead 100644 --- a/crates/evix-daemon/src/lib.rs +++ b/crates/evix-daemon/src/lib.rs @@ -807,8 +807,10 @@ async fn handle_diff( } fn write_response(stream: &mut W, response: &Response) -> Result<()> { - serde_json::to_writer(&mut *stream, response)?; - writeln!(stream)?; + // serde_json::to_writer straight onto the socket costs a syscall per token. + let mut line = serde_json::to_vec(response)?; + line.push(b'\n'); + stream.write_all(&line)?; stream.flush()?; Ok(()) } From 8185fff38c775fb6b80dd2e44b85d2f9276c7da3 Mon Sep 17 00:00:00 2001 From: Amaan Qureshi Date: Fri, 2 Oct 2026 16:25:40 -0400 Subject: [PATCH 7/9] daemon: record replay from the streamed eval responses Recording in Session kept a second copy of every event for every session, including local evals and embedders. Serializing lazily on the first replay also let a concurrent diff land before the cache write, so later replays served pre-diff output. --- crates/evix-cli/src/main.rs | 7 -- crates/evix-daemon/src/lib.rs | 53 +++++---- crates/evix-daemon/src/session_cache.rs | 78 +++++++------ crates/evix-protocol/src/lib.rs | 16 +-- crates/evix/src/error.rs | 5 - crates/evix/src/session.rs | 145 ++++-------------------- crates/evix/src/state.rs | 1 - 7 files changed, 94 insertions(+), 211 deletions(-) diff --git a/crates/evix-cli/src/main.rs b/crates/evix-cli/src/main.rs index ee73155..d14698b 100644 --- a/crates/evix-cli/src/main.rs +++ b/crates/evix-cli/src/main.rs @@ -245,13 +245,6 @@ fn run_daemon_request(mut stream: UnixStream, request: &Request) -> Result<()> { return Ok(()); } }, - Response::Replay { lines } => { - for line in lines { - if write_output_line(&mut stdout, &line)? == OutputWrite::Closed { - return Ok(()); - } - } - }, Response::Diff { diff } => { if write_output_line(&mut stdout, &evix_json::diff_line(&diff))? == OutputWrite::Closed diff --git a/crates/evix-daemon/src/lib.rs b/crates/evix-daemon/src/lib.rs index 4f4dead..194a8eb 100644 --- a/crates/evix-daemon/src/lib.rs +++ b/crates/evix-daemon/src/lib.rs @@ -1,7 +1,7 @@ use std::{ env, fs::{self, OpenOptions}, - io::{BufRead, BufReader, BufWriter, Write}, + io::{BufRead, BufReader, Write}, os::{ fd::{AsRawFd as _, RawFd}, unix::{ @@ -29,12 +29,10 @@ mod connection_limit; mod session_cache; use connection_limit::ConnectionLimiter; -use session_cache::DaemonState; +use session_cache::{DaemonState, Replay}; const MAX_REQUEST_BYTES: usize = 16 * 1024 * 1024; const MAX_CONNECTIONS: usize = 64; -// Keep replay responses bounded while avoiding one protocol message per event. -const REPLAY_LINES_PER_RESPONSE: usize = 128; static SHUTDOWN_REQUESTED: AtomicBool = AtomicBool::new(false); static SHUTDOWN_SIGNAL_WRITE_FD: AtomicI32 = AtomicI32::new(-1); @@ -737,14 +735,27 @@ async fn handle_eval( stream: &mut UnixStream, config: Config, ) -> Result<()> { - let session = state.replace_session(config).await?; + let session = state.replace_session(config.clone()).await?; let mut events = session.stream_bounded(evix::DEFAULT_STREAM_BUFFER_CAPACITY); + let mut recorded = Some(Vec::new()); while let Some(event) = events.next().await { - match event { - Ok(event) => write_response(stream, &Response::event(&event))?, - Err(err) => write_response(stream, &Response::error(err.to_string()))?, + let response = match event { + Ok(event) => Response::event(&event), + Err(err) => { + recorded = None; + Response::error(err.to_string()) + }, + }; + let line = response_line(&response)?; + stream.write_all(&line)?; + if let Some(recorded) = &mut recorded { + recorded.extend_from_slice(&line); } } + + let replay = + recorded.map_or(Replay::Invalidated, |lines| Replay::Ready(lines.into())); + state.settle_replay(&config, &session, replay)?; write_response(stream, &Response::Done) } @@ -753,14 +764,8 @@ async fn handle_replay( stream: &mut UnixStream, config: Config, ) -> Result<()> { - let lines = state.replay_lines(&config).await?; - let mut writer = BufWriter::new(stream); - for lines in lines.chunks(REPLAY_LINES_PER_RESPONSE) { - write_response(&mut writer, &Response::Replay { - lines: lines.to_vec(), - })?; - } - write_response(&mut writer, &Response::Done) + stream.write_all(&state.replay(&config)?)?; + write_response(stream, &Response::Done) } async fn handle_watch( @@ -768,7 +773,8 @@ async fn handle_watch( stream: &mut UnixStream, config: Config, ) -> Result<()> { - let session = state.replace_session(config).await?; + let session = state.replace_session(config.clone()).await?; + state.settle_replay(&config, &session, Replay::Invalidated)?; let mut diffs = session.watch_bounded(evix::DEFAULT_STREAM_BUFFER_CAPACITY); while let Some(diff) = diffs.next().await { match diff { @@ -801,18 +807,21 @@ async fn handle_diff( let session = state.warm_session(&config)?; session.require_completed().await?; let diff = session.diff_once().await?; - state.clear_replay(&config)?; + state.settle_replay(&config, &session, Replay::Invalidated)?; write_response(stream, &Response::diff(&diff))?; write_response(stream, &Response::Done) } -fn write_response(stream: &mut W, response: &Response) -> Result<()> { +fn write_response(stream: &mut UnixStream, response: &Response) -> Result<()> { + stream.write_all(&response_line(response)?)?; + Ok(()) +} + +fn response_line(response: &Response) -> Result> { // serde_json::to_writer straight onto the socket costs a syscall per token. let mut line = serde_json::to_vec(response)?; line.push(b'\n'); - stream.write_all(&line)?; - stream.flush()?; - Ok(()) + Ok(line) } #[cfg(test)] mod tests; diff --git a/crates/evix-daemon/src/session_cache.rs b/crates/evix-daemon/src/session_cache.rs index 291d063..443f43b 100644 --- a/crates/evix-daemon/src/session_cache.rs +++ b/crates/evix-daemon/src/session_cache.rs @@ -4,7 +4,7 @@ use std::{ sync::{Arc, Mutex}, }; -use anyhow::{Context as _, Result}; +use anyhow::{Context as _, Result, bail}; use evix::{AutoArg, Config, Input, Remote, Session}; use serde::Serialize; @@ -17,8 +17,17 @@ pub(crate) struct DaemonState { #[derive(Clone)] struct CachedSession { - session: Arc, - replay_lines: Option>, + session: Arc, + replay: Replay, +} + +/// Newline-delimited [`Response::Event`](evix_protocol::Response::Event) +/// messages streamed by the session's initial evaluation. +#[derive(Clone)] +pub(crate) enum Replay { + Recording, + Ready(Arc<[u8]>), + Invalidated, } impl DaemonState { @@ -33,8 +42,8 @@ impl DaemonState { .lock() .expect("daemon session registry poisoned") .insert(key, CachedSession { - session: Arc::clone(&session), - replay_lines: None, + session: Arc::clone(&session), + replay: Replay::Recording, }); Ok(session) } @@ -57,53 +66,42 @@ impl DaemonState { }) } - pub(crate) async fn replay_lines( - &self, - config: &Config, - ) -> Result> { + pub(crate) fn replay(&self, config: &Config) -> Result> { let key = session_key(config)?; - let session = { - let mut sessions = self - .sessions - .lock() - .expect("daemon session registry poisoned"); - let cached = sessions.get_mut(&key).ok_or_else(|| { - anyhow::anyhow!("no replayable session for requested daemon config") - })?; - if let Some(lines) = &cached.replay_lines { - return Ok(Arc::clone(lines)); - } - Arc::clone(&cached.session) - }; - - let lines: Arc<[String]> = session - .replay() - .await? - .iter() - .map(evix::json::event_line) - .collect::>() - .into(); let mut sessions = self .sessions .lock() .expect("daemon session registry poisoned"); - if let Some(cached) = sessions.get_mut(&key) - && Arc::ptr_eq(&cached.session, &session) - { - cached.replay_lines = Some(Arc::clone(&lines)); + let Some(cached) = sessions.get(&key) else { + bail!("no replayable session for requested daemon config"); + }; + match cached.replay { + Replay::Recording => bail!("matching evaluation has not completed"), + Replay::Ready(lines) => Ok(lines), + Replay::Invalidated => { + bail!("session has no replayable completed evaluation") + }, } - Ok(lines) } - pub(crate) fn clear_replay(&self, config: &Config) -> Result<()> { + /// Leaves a replaced session or an invalidated replay alone, so a `diff` + /// that lands before the eval stream finishes still wins. + pub(crate) fn settle_replay( + &self, + config: &Config, + session: &Arc, + replay: Replay, + ) -> Result<()> { let key = session_key(config)?; - if let Some(cached) = self + let mut sessions = self .sessions .lock() - .expect("daemon session registry poisoned") - .get_mut(&key) + .expect("daemon session registry poisoned"); + if let Some(cached) = sessions.get_mut(&key) + && Arc::ptr_eq(&cached.session, session) + && !matches!(cached.replay, Replay::Invalidated) { - cached.replay_lines = None; + cached.replay = replay; } Ok(()) } diff --git a/crates/evix-protocol/src/lib.rs b/crates/evix-protocol/src/lib.rs index 5928575..ed186b1 100644 --- a/crates/evix-protocol/src/lib.rs +++ b/crates/evix-protocol/src/lib.rs @@ -485,20 +485,10 @@ impl std::error::Error for ProtocolVersionError {} #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "camelCase")] pub enum Response { - Event { - event: Event, - }, - /// Cached newline-delimited event output from a replay request. - Replay { - lines: Vec, - }, - Diff { - diff: Diff, - }, + Event { event: Event }, + Diff { diff: Diff }, Done, - Error { - message: String, - }, + Error { message: String }, } impl Response { diff --git a/crates/evix/src/error.rs b/crates/evix/src/error.rs index 9556928..f712415 100644 --- a/crates/evix/src/error.rs +++ b/crates/evix/src/error.rs @@ -7,8 +7,6 @@ pub enum Error { /// `Session::stream` was requested after the single-use stream had already /// started or completed. SessionStreamConsumed, - /// `Session::replay` was requested after the session's result changed. - SessionReplayUnavailable, /// A warm-graph operation was requested before initial evaluation completed. InitialEvaluationIncomplete { operation: &'static str }, /// A session operation requires completion, but evaluation is still running. @@ -61,9 +59,6 @@ impl fmt::Display for Error { Self::SessionStreamConsumed => { write!(f, "session stream has already been consumed") }, - Self::SessionReplayUnavailable => { - write!(f, "session has no replayable completed evaluation") - }, Self::InitialEvaluationIncomplete { operation } => { write!( f, diff --git a/crates/evix/src/session.rs b/crates/evix/src/session.rs index 215812b..886b573 100644 --- a/crates/evix/src/session.rs +++ b/crates/evix/src/session.rs @@ -37,12 +37,11 @@ use crate::{ /// later stream/query/diff calls report the stored failure instead of retrying /// implicitly. pub struct Session { - config: Config, - cancel: Arc, - state: Arc>, - completed: Arc, - initial: Arc>, - replay_available: AtomicBool, + config: Config, + cancel: Arc, + state: Arc>, + completed: Arc, + initial: Arc>, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -61,7 +60,6 @@ impl Session { state: Arc::new(RwLock::new(WarmState::default())), completed: Arc::new(Notify::new()), initial: Arc::new(Mutex::new(InitialEvaluation::Idle)), - replay_available: AtomicBool::new(true), }) } @@ -209,7 +207,6 @@ impl Session { /// fresh evaluation for each filesystem notification and diffs it against /// the previous warm graph. pub fn watch(&self) -> impl Stream> + '_ { - self.replay_available.store(false, Ordering::Release); let (tx, rx) = futures_mpsc::unbounded(); let config = self.config.clone(); let cancel = Arc::clone(&self.cancel); @@ -277,7 +274,6 @@ impl Session { &self, capacity: usize, ) -> impl Stream> + '_ { - self.replay_available.store(false, Ordering::Release); let (mut tx, rx) = futures_mpsc::channel(bounded_capacity(capacity)); let config = self.config.clone(); let cancel = Arc::clone(&self.cancel); @@ -376,32 +372,6 @@ impl Session { self.query_snapshot(filter).await } - /// Replay events from the completed initial evaluation without evaluating - /// Nix. - /// - /// # Errors - /// - /// Returns [`Error::InitialEvaluationIncomplete`] until the initial run - /// completes, or [`Error::SessionReplayUnavailable`] after `watch` or a - /// successful `diff_once` changes the session result. - pub async fn replay(&self) -> Result> { - let state = self.state.read().await; - if !self.replay_available.load(Ordering::Acquire) { - return Err(Error::SessionReplayUnavailable); - } - if !state.completed { - if let Some(error) = &state.error { - return Err(Error::EvaluationFailed { - message: error.clone(), - }); - } - return Err(Error::InitialEvaluationIncomplete { - operation: "replay", - }); - } - Ok(state.events.clone()) - } - /// Perform one full re-evaluation and diff it against the warm graph. pub async fn diff_once(&self) -> Result { let previous = { @@ -431,7 +401,6 @@ impl Session { state.errors = errors; state.completed = true; state.error = None; - self.replay_available.store(false, Ordering::Release); } Ok(diff) } @@ -490,31 +459,20 @@ async fn evaluate_initial( cancel: Arc, state: Arc>, completed: Arc, - mut on_event: F, + on_event: F, ) -> AnyhowResult where F: FnMut(Event) -> Fut + Send + 'static, Fut: Future>, { debug!("starting session evaluation"); - let events = Arc::new(Mutex::new(Vec::new())); - let recorded_events = Arc::clone(&events); - let result = run::evaluate_async(config, Arc::clone(&cancel), move |event| { - recorded_events - .lock() - .expect("session event cache poisoned") - .push(event.clone()); - on_event(event) - }) - .await; + let result = run::evaluate_async(config, Arc::clone(&cancel), on_event).await; match result { Ok((graph, errors, RunOutcome::Completed)) => { let mut state = state.write().await; state.graph = graph; state.errors = errors; - state.events = - events.lock().expect("session event cache poisoned").clone(); state.completed = true; state.error = None; completed.notify_waiters(); @@ -672,62 +630,6 @@ mod tests { })); } - #[test] - fn replay_returns_completed_events_without_starting_evaluation() { - tokio::runtime::Builder::new_current_thread() - .enable_time() - .build() - .unwrap() - .block_on(async { - let session = Session::open(Config::default()).await.unwrap(); - { - let mut state = session.state.write().await; - state.events.push(Event::AttrSet { - attr: "jobs".into(), - attr_path: vec!["jobs".into()], - attrs: vec!["hello".into()], - }); - state.completed = true; - } - - let events = session.replay().await.unwrap(); - assert_eq!(events.len(), 1); - assert!( - matches!(&events[0], Event::AttrSet { attr, .. } if attr == "jobs") - ); - - session.replay_available.store(false, Ordering::Release); - assert!(matches!( - session.replay().await, - Err(Error::SessionReplayUnavailable) - )); - }); - } - - #[test] - fn replay_rechecks_availability_after_waiting_for_state() { - tokio::runtime::Builder::new_current_thread() - .build() - .unwrap() - .block_on(async { - let session = Session::open(Config::default()).await.unwrap(); - let mut state = session.state.write().await; - state.completed = true; - state.events.push(Event::AttrSet { - attr: "jobs".into(), - attr_path: vec!["jobs".into()], - attrs: vec![], - }); - - let mut replay = Box::pin(session.replay()); - assert!(futures_util::poll!(&mut replay).is_pending()); - session.replay_available.store(false, Ordering::Release); - drop(state); - - assert!(matches!(replay.await, Err(Error::SessionReplayUnavailable))); - }); - } - #[test] fn failed_initial_evaluation_is_reported_after_stream_consumed() { let runtime = tokio::runtime::Builder::new_current_thread() @@ -766,12 +668,11 @@ mod tests { .build() .unwrap(); let session = Session { - config: Config::default(), - cancel: Arc::new(AtomicBool::new(false)), - state: Arc::new(RwLock::new(WarmState::default())), - completed: Arc::new(Notify::new()), - initial: Arc::new(Mutex::new(InitialEvaluation::Finished)), - replay_available: AtomicBool::new(true), + config: Config::default(), + cancel: Arc::new(AtomicBool::new(false)), + state: Arc::new(RwLock::new(WarmState::default())), + completed: Arc::new(Notify::new()), + initial: Arc::new(Mutex::new(InitialEvaluation::Finished)), }; let mut stream = Box::pin(session.stream()); @@ -787,12 +688,11 @@ mod tests { .build() .unwrap(); let session = Session { - config: Config::default(), - cancel: Arc::new(AtomicBool::new(false)), - state: Arc::new(RwLock::new(WarmState::default())), - completed: Arc::new(Notify::new()), - initial: Arc::new(Mutex::new(InitialEvaluation::Finished)), - replay_available: AtomicBool::new(true), + config: Config::default(), + cancel: Arc::new(AtomicBool::new(false)), + state: Arc::new(RwLock::new(WarmState::default())), + completed: Arc::new(Notify::new()), + initial: Arc::new(Mutex::new(InitialEvaluation::Finished)), }; let mut stream = Box::pin(session.stream_bounded(1)); @@ -811,12 +711,11 @@ mod tests { #[test] fn cancel_sets_session_cancellation_flag() { let session = Session { - config: Config::default(), - cancel: Arc::new(AtomicBool::new(false)), - state: Arc::new(RwLock::new(WarmState::default())), - completed: Arc::new(Notify::new()), - initial: Arc::new(Mutex::new(InitialEvaluation::Idle)), - replay_available: AtomicBool::new(true), + config: Config::default(), + cancel: Arc::new(AtomicBool::new(false)), + state: Arc::new(RwLock::new(WarmState::default())), + completed: Arc::new(Notify::new()), + initial: Arc::new(Mutex::new(InitialEvaluation::Idle)), }; session.cancel(); diff --git a/crates/evix/src/state.rs b/crates/evix/src/state.rs index 527520e..bfbe147 100644 --- a/crates/evix/src/state.rs +++ b/crates/evix/src/state.rs @@ -8,7 +8,6 @@ pub type EvalGraph = HashMap, Derivation>; pub struct WarmState { pub graph: EvalGraph, pub errors: Vec, - pub events: Vec, pub completed: bool, pub error: Option, } From a678438a4c56a84e7186496d26c250ddd886b4b9 Mon Sep 17 00:00:00 2001 From: Amaan Qureshi Date: Fri, 2 Oct 2026 16:27:57 -0400 Subject: [PATCH 8/9] bench: fix prewarm arguments and restore the filtered query --- bench/README.md | 7 +++---- bench/bench.sh | 3 ++- bench/results.md | 19 ++++++++++--------- 3 files changed, 15 insertions(+), 14 deletions(-) diff --git a/bench/README.md b/bench/README.md index d41b6e8..c1b09b9 100644 --- a/bench/README.md +++ b/bench/README.md @@ -54,7 +54,6 @@ $ bench/bench.sh 5 3 Runs `hyperfine` over evix at 1/4/8 local-only workers, evix remote-only distributed evaluation, evix mixed local+remote distributed evaluation, daemon-backed cold prewarming, warm replay of a completed daemon event stream, -warm full-graph daemon queries, and nix-eval-jobs. These commands produce the -same full result, so their relative timings are comparable. The `hyperfine` -warmup materializes replay's cached NDJSON, so the replay result measures -subsequent calls. Writes `bench/results.md`. +warm full-graph daemon queries, filtered warm daemon queries, and nix-eval-jobs. +Queries print only derivations, while evaluation and replay also print attrset +and error events. Writes `bench/results.md`. diff --git a/bench/bench.sh b/bench/bench.sh index 6684148..e3c73f8 100755 --- a/bench/bench.sh +++ b/bench/bench.sh @@ -45,7 +45,7 @@ for _ in $(seq 1 100); do [ -S "$sock" ] && break sleep 0.05 done -"$evix" eval --socket "$sock" --workers 4"$args_str" >/dev/null +"$evix" eval --socket "$sock" --workers 4 "${args[@]}" >/dev/null cmds=( -n "evix local=1" "$evix eval --no-daemon --workers 1$args_str >/dev/null" @@ -56,6 +56,7 @@ cmds=( -n "evix daemon prewarm local=4" "$evix eval --socket $sock --workers 4$args_str >/dev/null" -n "evix daemon warm replay local=4" "$evix eval --replay --socket $sock --workers 4$args_str >/dev/null" -n "evix daemon warm query full local=4" "$evix query --socket $sock --workers 4$args_str >/dev/null" + -n "evix daemon warm query n0 local=4" "$evix query --socket $sock --workers 4 --attr-prefix n0$args_str >/dev/null" ) if command -v nix-eval-jobs >/dev/null; then cmds+=(-n "nix-eval-jobs w=4" "nix-eval-jobs --gc-roots-dir $gc --workers 4 $fixture$fargs_str >/dev/null") diff --git a/bench/results.md b/bench/results.md index da0e8e2..7365bcc 100644 --- a/bench/results.md +++ b/bench/results.md @@ -1,11 +1,12 @@ | Command | Mean [ms] | Min [ms] | Max [ms] | Relative | |:---|---:|---:|---:|---:| -| `evix local=1` | 840.0 ± 17.0 | 813.4 | 855.5 | 21.22 ± 0.52 | -| `evix local=4` | 424.4 ± 17.4 | 397.4 | 439.6 | 10.72 ± 0.46 | -| `evix local=8` | 437.7 ± 22.9 | 419.9 | 477.5 | 11.06 ± 0.60 | -| `evix distributed remote=4` | 571.2 ± 23.9 | 543.0 | 599.9 | 14.43 ± 0.64 | -| `evix distributed local=4 remote=4` | 446.7 ± 24.4 | 418.2 | 483.8 | 11.28 ± 0.64 | -| `evix daemon prewarm local=4` | 532.2 ± 6.5 | 525.5 | 542.8 | 13.44 ± 0.25 | -| `evix daemon warm replay local=4` | 39.6 ± 0.6 | 39.0 | 40.5 | 1.00 | -| `evix daemon warm query full local=4` | 286.3 ± 2.4 | 283.6 | 289.3 | 7.23 ± 0.12 | -| `nix-eval-jobs w=4` | 438.6 ± 4.3 | 435.3 | 446.0 | 11.08 ± 0.19 | +| `evix local=1` | 118.8 ± 4.6 | 112.5 | 124.5 | 11.75 ± 0.81 | +| `evix local=4` | 62.1 ± 2.7 | 60.3 | 66.9 | 6.14 ± 0.44 | +| `evix local=8` | 58.1 ± 2.9 | 54.4 | 61.6 | 5.75 ± 0.43 | +| `evix distributed remote=4` | 82.3 ± 5.4 | 76.2 | 91.1 | 8.14 ± 0.71 | +| `evix distributed local=4 remote=4` | 61.8 ± 2.7 | 59.7 | 66.1 | 6.12 ± 0.44 | +| `evix daemon prewarm local=4` | 64.2 ± 1.2 | 62.8 | 65.4 | 6.35 ± 0.38 | +| `evix daemon warm replay local=4` | 12.9 ± 0.7 | 11.9 | 13.6 | 1.28 ± 0.10 | +| `evix daemon warm query full local=4` | 13.5 ± 0.3 | 13.2 | 14.0 | 1.34 ± 0.08 | +| `evix daemon warm query n0 local=4` | 10.1 ± 0.6 | 9.5 | 10.7 | 1.00 | +| `nix-eval-jobs w=4` | 80.5 ± 7.3 | 75.2 | 92.0 | 7.96 ± 0.85 | From a12e1c8adbb73336d49d0e9b4f8cdb837c5e8290 Mon Sep 17 00:00:00 2001 From: Amaan Qureshi Date: Fri, 2 Oct 2026 16:29:55 -0400 Subject: [PATCH 9/9] docs: note that replay skips gc root registration --- README.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index d62c413..9cf2237 100644 --- a/README.md +++ b/README.md @@ -101,7 +101,9 @@ configuration writes those events again without starting Nix or querying the warm graph. The snapshot is replaced by a new `eval`, and invalidated by `diff` or `watch`. This system is **opt-in**. It does not check whether source files, environment variables, or other impure inputs changed, so use ordinary `eval` -when a fresh result is important to its success. It requires a running daemon +when a fresh result is important to its success. Replay also skips GC root +registration, so with `--gc-roots-dir` the replayed derivations stay alive only +as long as the roots the original `eval` created. It requires a running daemon and fails if there is no completed matching session. [^1]: I can almost hear you asking _"what the hell is incremental-like?"_. Well,