diff --git a/crates/tty7-core/src/core/workspace_store.rs b/crates/tty7-core/src/core/workspace_store.rs index 4c995760..6221695a 100644 --- a/crates/tty7-core/src/core/workspace_store.rs +++ b/crates/tty7-core/src/core/workspace_store.rs @@ -72,6 +72,20 @@ pub const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024; /// error rather than grow the file until the disk fills. pub const MAX_WORKSPACES: usize = 1024; +/// Ceiling on the whole document, which is what a single `WorkspaceList` reply +/// has to fit into. +/// +/// The per-record and per-count ceilings above are independent of each other, +/// and their product is 4 GiB — sixty-four times the frame limit. Seventeen +/// accepted `WorkspacePut`s of a maximal record are enough to put the array +/// past it, and from then on *every* `WorkspaceList` on the machine is a reply +/// that cannot be encoded: every client shows an empty workspace list, and the +/// only repair is editing the file by hand. So the total is bounded where it is +/// actually known — at the save — with room to spare under +/// [`MAX_FRAME`](crate::daemon::protocol::MAX_FRAME), since what is measured +/// here is the pretty-printed form and the wire carries the compact one. +pub const MAX_STORE_BYTES: usize = 32 * 1024 * 1024; + /// Ceiling on a record key, which is a workspace uuid in every non-hostile /// case. const MAX_ID_BYTES: usize = 128; @@ -172,6 +186,23 @@ struct State { /// because the file's array order is what a client lists, and a hash map /// would reshuffle the picker on every save for no reason. records: Vec<(String, Value)>, + /// `(mtime, len)` of the file as this snapshot last saw it, or `None` when + /// there was no file. + /// + /// This store is not always the only writer. The design's answer is one + /// server per machine, and `tty7-server --stdio` now starts the daemon + /// rather than serving in-process for exactly that reason — but an explicit + /// `--serve`, or a daemon that could not be started, still leaves two + /// processes over one file. `persist` writes the *whole* document, so + /// without noticing that the file moved underneath it, the second to save + /// silently drops everything the first did. + stamp: Option<(std::time::SystemTime, u64)>, +} + +/// The file's identity as far as [`State::stamp`] is concerned. +fn stamp_of(path: &Path) -> Option<(std::time::SystemTime, u64)> { + let meta = std::fs::metadata(path).ok()?; + Some((meta.modified().ok()?, meta.len())) } impl WorkspaceStore { @@ -186,9 +217,10 @@ impl WorkspaceStore { pub fn open(path: impl Into) -> Arc { let path = path.into(); let records = load_records(&path); + let stamp = stamp_of(&path); Arc::new(WorkspaceStore { + state: Mutex::new(State { records, stamp }), path, - state: Mutex::new(State { records }), attachments: Mutex::new(Vec::new()), subscribers: Mutex::new(Vec::new()), next_subscriber: AtomicU64::new(1), @@ -302,7 +334,7 @@ impl WorkspaceStore { Undo::Remove(st.records.len() - 1) } }; - if let Err(e) = self.persist(&st) { + if let Err(e) = self.persist(&st, true) { match undo { Undo::Restore(i, old) => st.records[i].1 = old, Undo::Remove(i) => { @@ -311,6 +343,7 @@ impl WorkspaceStore { } return Err(e); } + self.restamp(&mut st); } self.notify(id, origin); @@ -329,10 +362,11 @@ impl WorkspaceStore { return Ok(false); }; let removed = st.records.remove(i); - if let Err(e) = self.persist(&st) { + if let Err(e) = self.persist(&st, false) { st.records.insert(i, removed); return Err(e); } + self.restamp(&mut st); } // The attachment goes with it: nothing can be attached to a workspace // that no longer exists, and leaving the entry would have M6 report a @@ -440,7 +474,24 @@ impl WorkspaceStore { // in-memory state is still a valid state (the undo path restores it // before returning) and the file is either the old or the new one, so // carrying on is strictly better than taking the server down. - self.state.lock().unwrap_or_else(|e| e.into_inner()) + let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner()); + + // Re-read when the file moved under us. Cheap — one `stat` — and it is + // what keeps a second writer's changes from being overwritten by this + // store's whole-document save, since the base we mutate is then theirs + // rather than a snapshot from before their write. It also lets a read + // see their changes at all: `notify` reaches subscribers in *this* + // process only. + let on_disk = stamp_of(&self.path); + if on_disk != st.stamp { + log::debug!( + "{} changed underneath this store; re-reading", + self.path.display() + ); + st.records = load_records(&self.path); + st.stamp = on_disk; + } + st } fn attachments_locked(&self) -> std::sync::MutexGuard<'_, Vec<(String, Attachment)>> { @@ -453,7 +504,14 @@ impl WorkspaceStore { /// [`Workspaces`](crate::core::session::Workspaces) parses, so this file is /// readable by the same code that reads a client's `session.json` and a /// human can diff the two. - fn persist(&self, st: &State) -> io::Result<()> { + /// Write the whole document. + /// + /// `bounded` asks for [`MAX_STORE_BYTES`] to be enforced. Set by the paths + /// that *grow* the file and clear by the ones that shrink it: a store that + /// came up holding an over-large file — written by an older build, or by + /// hand — must still be able to delete its way back under the limit rather + /// than refusing every operation including the repair. + fn persist(&self, st: &State, bounded: bool) -> io::Result<()> { #[derive(Serialize)] struct Doc<'a> { workspaces: Vec<&'a Value>, @@ -462,11 +520,28 @@ impl WorkspaceStore { workspaces: st.records.iter().map(|(_, v)| v).collect(), }; let bytes = serde_json::to_vec_pretty(&doc).map_err(io::Error::other)?; + if bounded && bytes.len() > MAX_STORE_BYTES { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!( + "the workspace store would be {} bytes; the limit is {MAX_STORE_BYTES}, \ + which is what one WorkspaceList reply has to fit into", + bytes.len() + ), + )); + } if let Some(parent) = self.path.parent() { std::fs::create_dir_all(parent)?; } crate::core::config::write_atomic(&self.path, &bytes) } + + /// Record the file's identity after this store wrote it, so the next + /// [`WorkspaceStore::locked`] does not mistake its own save for someone + /// else's and re-read it. + fn restamp(&self, st: &mut State) { + st.stamp = stamp_of(&self.path); + } } /// How to undo a mutation whose write failed. @@ -632,6 +707,102 @@ mod tests { }) } + /// Two stores over one file — an explicit `--serve` alongside a daemon, or + /// a daemon that could not be started — must not silently undo each other. + /// + /// `persist` writes the whole document, so a store that mutates a snapshot + /// taken before the other's write puts that stale snapshot back. This is + /// how a workspace rename made on the laptop vanishes the next time the + /// desktop reorders a tab, with nothing reported to either. + #[test] + fn a_second_writer_does_not_get_overwritten_by_a_stale_snapshot() { + let dir = tempfile::TempDir::new().unwrap(); + let path = dir.path().join(STORE_FILE); + let first = WorkspaceStore::open(&path); + let second = WorkspaceStore::open(&path); + + first.put("w1", record("w1", "one"), None).unwrap(); + first.put("w2", record("w2", "two"), None).unwrap(); + + // `second` last read the file when it was empty. It has to notice. + second + .put("w2", record("w2", "two, renamed"), None) + .unwrap(); + + let names: Vec = WorkspaceStore::open(&path) + .list() + .iter() + .map(|r| r["name"].as_str().unwrap_or_default().to_string()) + .collect(); + assert_eq!( + names, + ["one", "two, renamed"], + "the second writer's save dropped what the first had written" + ); + } + + /// The same, one layer down: a read sees another process's write, because + /// `notify` only ever reaches subscribers inside this process. + #[test] + fn a_read_sees_a_change_another_store_made_to_the_file() { + let dir = tempfile::TempDir::new().unwrap(); + let path = dir.path().join(STORE_FILE); + let reader = WorkspaceStore::open(&path); + let writer = WorkspaceStore::open(&path); + + assert!(reader.get("w1").is_none()); + writer.put("w1", record("w1", "one"), None).unwrap(); + assert_eq!( + reader.get("w1").map(|r| r["name"].clone()), + Some(serde_json::json!("one")), + "a read answered from a snapshot older than the file" + ); + } + + /// The per-record and per-count ceilings do not bound their product, so the + /// document is bounded where it is known — at the save. + /// + /// Past `MAX_FRAME` the store is not merely large, it is unreadable: every + /// `WorkspaceList` becomes a reply that cannot be encoded, so every client + /// shows an empty list and the only repair is editing the file by hand. + #[test] + fn a_put_that_would_outgrow_one_reply_is_refused_and_undone() { + let (store, _dir) = store(); + // Records big enough that a handful crosses the limit, and small enough + // that the test stays quick. + let chunk = "x".repeat(2 * 1024 * 1024); + let big = |id: &str| { + let mut r = record(id, "big"); + r["padding"] = Value::String(chunk.clone()); + r + }; + + let mut accepted = 0; + let refusal = loop { + let id = format!("w{accepted}"); + match store.put(&id, big(&id), None) { + Ok(()) => accepted += 1, + Err(e) => break e, + } + assert!(accepted < 64, "the total was never bounded"); + }; + assert_eq!(refusal.kind(), io::ErrorKind::InvalidInput); + assert!( + refusal.to_string().contains("WorkspaceList"), + "the refusal has to say what the limit is for: {refusal}" + ); + + // Refused, not half-applied: the record that did not fit is not in the + // store and is not in the file. + assert_eq!(store.len(), accepted); + assert!(store.get(&format!("w{accepted}")).is_none()); + assert_eq!(WorkspaceStore::open(store.path()).len(), accepted); + + // And a delete still works, so a store that came up over the limit can + // be repaired rather than being wedged. + assert!(store.delete("w0", None).unwrap()); + } + // ── The basics ────────────────────────────────────────────────────────── #[test] diff --git a/crates/tty7-core/src/host/server.rs b/crates/tty7-core/src/host/server.rs index 61f2c18a..6abd5779 100644 --- a/crates/tty7-core/src/host/server.rs +++ b/crates/tty7-core/src/host/server.rs @@ -144,6 +144,21 @@ impl Services { #[derive(Default)] pub struct AttachRegistry { live: Mutex>, + /// Held across *both* tables for the length of one handover. + /// + /// A takeover moves two things that live in different places: this + /// registry's handles, and the `WorkspaceStore`'s record. Each is + /// internally locked, and that is not enough — two clients attaching to one + /// workspace at the same moment can each win a different table, after which + /// the store names a session the registry has already evicted and no + /// `detach` can ever clear it, because the token no longer matches. From + /// then on the workspace reports a takeover against a client that + /// disconnected hours ago. + /// + /// Coarse on purpose: attach and detach happen once per workspace opened or + /// closed, so serializing them costs nothing worth measuring. Always the + /// outermost lock of the two, and never held while writing to a peer. + handover: Mutex<()>, } struct Live { @@ -182,6 +197,11 @@ struct Evicted { } impl AttachRegistry { + /// Take the handover lock. See [`AttachRegistry::handover`]. + fn handover(&self) -> std::sync::MutexGuard<'_, ()> { + self.handover.lock().unwrap_or_else(|e| e.into_inner()) + } + /// Who holds `workspace` — `(token, hostname)`. Diagnostics and tests. pub fn holder(&self, workspace: &str) -> Option<(String, String)> { self.locked() @@ -244,6 +264,14 @@ impl AttachRegistry { evicted } + /// Forget `workspace` whoever holds it — the workspace itself is gone. + /// + /// Unconditional, unlike [`AttachRegistry::release`]: a delete is not one + /// session giving something up, it is the thing ceasing to exist. + fn forget_workspace(&self, workspace: &str) { + self.locked().retain(|l| l.workspace != workspace); + } + /// Release `workspace`, but only if `conn` still holds it. `false` means it /// had already been taken over, which is success as far as the caller is /// concerned — and the reason releasing is conditional at all. @@ -520,13 +548,21 @@ fn attach_workspace( dedicated: bool, ) -> io::Result> { let store = conn.workspaces()?; - let displaced = store.attach( - workspace, - Attachment::new(conn.holder.token.clone(), conn.holder.hostname.clone()), - ); - let evicted = conn - .attachments - .claim(workspace, conn.id, &conn.holder, dedicated); + let (displaced, evicted) = { + // Both tables move under one lock. Held only across the two moves — + // the notice below goes out with nothing held, because writing to a + // peer that has stopped reading must not hold up the next client's + // attach. + let _handover = conn.attachments.handover(); + let displaced = store.attach( + workspace, + Attachment::new(conn.holder.token.clone(), conn.holder.hostname.clone()), + ); + let evicted = conn + .attachments + .claim(workspace, conn.id, &conn.holder, dedicated); + (displaced, evicted) + }; if let Some(evicted) = evicted { log::info!( @@ -562,6 +598,7 @@ fn attach_workspace( /// then tidied up must not evict the client that took over from it. fn detach_workspace(conn: &Arc, workspace: &str) -> io::Result { let store = conn.workspaces()?; + let _handover = conn.attachments.handover(); let released = conn.attachments.release(workspace, conn.id); let forgotten = store.detach(workspace, &conn.holder.token); Ok(released || forgotten) @@ -814,7 +851,19 @@ fn run_request( ControlRequest::WorkspaceDelete { id } => { // Deleting what is not there is success — a delete that raced // another client's delete has got what it asked for. - conn.workspaces()?.delete(&id, conn.workspace_origin)?; + let store = conn.workspaces()?; + { + // The store drops its own attachment on delete; the registry + // has to be told, and under the same lock, or the two disagree + // with no race needed at all. Left behind, the stale `Live` + // entry means the *next* client to attach a workspace with this + // id evicts a session nobody displaced — and, that entry being + // dedicated, closes its whole link, taking every other + // workspace on it down too. + let _handover = conn.attachments.handover(); + store.delete(&id, conn.workspace_origin)?; + conn.attachments.forget_workspace(&id); + } (ReplyOk::Unit, Vec::new()) } @@ -894,6 +943,7 @@ impl Conn { /// earlier is already gone from the registry and is not touched — the exact /// case the store's token check exists for, seen from the other side. fn release_all_workspaces(&self) { + let _handover = self.attachments.handover(); let released = self.attachments.release_conn(self.id); let Some(store) = self.workspaces.as_ref() else { return; @@ -3015,6 +3065,98 @@ mod tests { ); } + /// Deleting a workspace clears it from *both* tables. + /// + /// The store drops its own attachment on delete. If the registry keeps its + /// handle, the two disagree with no race needed, and the next client to + /// attach that id evicts a session nobody displaced — closing its whole + /// link, since a dedicated entry takes every other workspace on that + /// connection down with it. + #[test] + fn deleting_a_workspace_clears_both_attachment_tables() { + let (services, _dir) = workspace_services(); + let registry = Arc::clone(&services.attachments); + let store = services.workspaces.clone().unwrap(); + + let ((mut laptop, _), _l) = + raw_hello(services.clone(), hello_for("w", "tok-laptop", "laptop")); + await_holder(®istry, "w", "laptop"); + assert!(store.attachment("w").is_some()); + + ask( + &mut laptop, + 1, + ControlRequest::WorkspacePut { + id: "w".to_string(), + json: ws_record("w", "the workspace"), + }, + ); + let reply = ask( + &mut laptop, + 2, + ControlRequest::WorkspaceDelete { + id: "w".to_string(), + }, + ); + assert!( + matches!(reply, ControlReply::Ok(ReplyOk::Unit)), + "{reply:?}" + ); + + assert!( + store.attachment("w").is_none(), + "the store still names a holder for a workspace that is gone" + ); + assert!( + registry.holder("w").is_none(), + "the registry still holds a workspace that is gone" + ); + } + + /// The store's record and the registry's handle move under **one** lock. + /// + /// They are separate tables with separate locks, and taking them one after + /// the other is not enough: two clients attaching the same workspace at the + /// same instant can each win a different one, after which the store names a + /// session the registry has already evicted. No `detach` can clear it — its + /// token no longer matches — so from then on the workspace reports a + /// takeover against a client that disconnected hours ago. + /// + /// Held from the test rather than raced, because the window is a few + /// instructions wide and a racing test passes against the broken ordering + /// far more often than it fails. Holding the handover proves the stronger + /// thing anyway: with it held, an attach reaches *neither* table. + #[test] + fn an_attach_moves_both_tables_under_one_lock() { + let (services, _dir) = workspace_services(); + let registry = Arc::clone(&services.attachments); + let store = services.workspaces.clone().unwrap(); + + let held = registry.handover(); + // The handshake replies before the attach, so this returns rather than + // blocking on the lock we are holding. + let ((_laptop, _ok), _served) = + raw_hello(services.clone(), hello_for("w", "tok-laptop", "laptop")); + + std::thread::sleep(Duration::from_millis(150)); + assert!( + registry.holder("w").is_none(), + "the registry was moved while a handover was in flight" + ); + assert!( + store.attachment("w").is_none(), + "the store was moved while a handover was in flight" + ); + + drop(held); + await_holder(®istry, "w", "laptop"); + assert_eq!( + store.attachment("w").map(|a| a.token).as_deref(), + Some("tok-laptop"), + "both tables have to name the same session once the handover is done" + ); + } + /// The other half of that rule. A client holds **one connection per /// machine**, so closing the link on a takeover would drop windows nobody /// preempted; the push still goes out, the link stays up. diff --git a/crates/tty7-server/src/main.rs b/crates/tty7-server/src/main.rs index 49f1cd43..65651348 100644 --- a/crates/tty7-server/src/main.rs +++ b/crates/tty7-server/src/main.rs @@ -163,6 +163,7 @@ fn run_stdio(args: &[String]) -> io::Result<()> { { use std::os::unix::net::UnixStream; use tty7_core::daemon::duplex::StdioDuplex; + use tty7_core::daemon::spawn; use tty7_core::host::local::LocalHost; use tty7_core::host::server; @@ -201,10 +202,43 @@ fn run_stdio(args: &[String]) -> io::Result<()> { Err(e) if force_bridge => return Err(e), Err(e) => { log_stderr(format_args!( - "no control server at {} ({e}); serving in this process", + "no control server at {} ({e})", sock.display() )); - None + // One control server per machine, started if nobody has — + // the same rule `bridge_panes` follows one dialect over, + // and for the same reason. Two `--stdio` sessions both + // falling through to serving in-process would each hold + // their own `WorkspaceStore` over the one file, and + // `persist` writes the whole document: the second to save + // silently drops the first's changes. Their attachment + // registries would be separate too, which makes design + // §10's takeover a no-op between them — both clients would + // hold the same workspace and neither would be told. + // + // Not attempted when the caller named a socket: starting a + // daemon binds the machine's default endpoint, not theirs, + // so it would be a daemon nobody asked for and nobody uses. + if may_start_daemon(args) { + match spawn::ensure_running() + .map_err(io::Error::other) + .and_then(|()| UnixStream::connect(&sock)) + { + Ok(s) => { + log_stderr(format_args!("started one; bridging to it")); + Some(s) + } + Err(e) => { + log_stderr(format_args!( + "could not start one ({e}); serving in this process" + )); + None + } + } + } else { + log_stderr(format_args!("serving in this process")); + None + } } } }; @@ -358,6 +392,17 @@ fn control_services() -> tty7_core::host::server::Services { } /// `--flag ` or `--flag=`, first occurrence wins. +/// Whether a failed control probe may start the machine's daemon. +/// +/// Only when the caller did not name a socket. `--control-sock` says "this +/// endpoint", and `spawn::ensure_running` binds the machine's default one — so +/// starting a daemon there would leave a process nobody asked for and nobody +/// reaches. It is also what keeps the test suite, and any `--config-dir` +/// isolation built on it, from spraying daemons across a developer's machine. +fn may_start_daemon(args: &[String]) -> bool { + flag_value(args, "--control-sock").is_none() +} + fn flag_value(args: &[String], flag: &str) -> Option { let with_eq = format!("{flag}="); let mut it = args.iter(); @@ -385,3 +430,23 @@ fn apply_config_dir_arg(args: &[String]) { tty7_core::core::config::set_config_dir(path.into()); } } + +#[cfg(test)] +mod tests { + use super::*; + + fn argv(args: &[&str]) -> Vec { + args.iter().map(|a| a.to_string()).collect() + } + + /// A named socket suppresses the daemon start, in both spellings of the + /// flag. Without this guard every `--stdio` in the test suite that points + /// at a temp socket would start a real daemon on the developer's machine. + #[test] + fn a_named_control_socket_suppresses_starting_a_daemon() { + assert!(may_start_daemon(&argv(&[]))); + assert!(may_start_daemon(&argv(&["--serve"]))); + assert!(!may_start_daemon(&argv(&["--control-sock", "/tmp/x.sock"]))); + assert!(!may_start_daemon(&argv(&["--control-sock=/tmp/x.sock"]))); + } +}