diff --git a/Cargo.lock b/Cargo.lock index 28b53448..635a2cf0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9606,7 +9606,6 @@ dependencies = [ name = "tty7-server" version = "26.7.6" dependencies = [ - "serde_json", "tempfile", "tty7-core", ] diff --git a/crates/tty7-core/src/core/machine.rs b/crates/tty7-core/src/core/machine.rs index b4f9e1d0..5c985b1d 100644 --- a/crates/tty7-core/src/core/machine.rs +++ b/crates/tty7-core/src/core/machine.rs @@ -2,9 +2,10 @@ //! //! # What this replaces, and why //! -//! [`crate::core::workspace_store`] is the previous design: an *opaque* record -//! store, where the client owned the schema and the server filed JSON blobs it -//! never read. That shape was right when there was exactly one writer (the GUI) +//! The previous design (`core::workspace_store`, since deleted) was an +//! *opaque* record store, where the client owned the schema and the server +//! filed JSON blobs it never read. That shape was right when there was +//! exactly one writer (the GUI) //! and the server's only job was to make a laptop's layout visible from a //! desktop. It stops being right the moment two clients — a GUI and a CLI, or //! two GUIs — write concurrently: whole-record `Put` is last-writer-wins, and @@ -69,18 +70,23 @@ use serde::{Deserialize, Serialize}; use crate::core::cli_agent::CLIAgent; use crate::core::session::WorkspaceId; -use crate::core::workspace_store::Attachment; use crate::daemon::protocol::NativeSshSpec; -/// The file's name under the data directory ([`crate::core::workspace_store::DATA_DIR_ENV`] -/// resolves where that is). +/// The file's name under the data directory ([`DATA_DIR_ENV`] resolves where +/// that is). /// -/// Deliberately **not** `workspaces.json`: that file's document is the retired -/// opaque-record store, whose reader quarantines anything it cannot parse. A -/// build downgraded across this refactor must find its old file untouched, and -/// this build's tree must not be "repaired" away by the old reader. +/// Deliberately **not** `workspaces.json`: that name belonged to the retired +/// opaque-record store, whose reader quarantined anything it could not parse. +/// A build downgraded across that refactor must find its old file untouched, +/// and this build's tree must not be "repaired" away by the old reader. pub const MACHINE_FILE: &str = "machine.json"; +/// Overrides where the machine's data directory lives. Set by tests and by a +/// second server on a shared box — the same escape hatch +/// [`CONTROL_SOCK_ENV`](crate::host::server::CONTROL_SOCK_ENV) is for the +/// socket. +pub const DATA_DIR_ENV: &str = "TTY7_DATA_DIR"; + /// Ceiling on workspaces, carried over from the old store: a client looping on /// "create workspace" should hit a named error rather than grow the file until /// the disk fills. @@ -156,6 +162,41 @@ pub struct Machine { pub panes: Vec, } +/// Who is currently attached to a workspace. +/// +/// **Data only.** The takeover behaviour — push `Preempted { by }` to the old +/// session, close its streams, offer a take-back button — lives in the control +/// server. What is here is the record that machinery needs to exist before it +/// can be written: the random token that tells two connections from the same +/// client apart, and the hostname that fills in "already open on ". Both +/// arrive in the [`ControlHello`](crate::daemon::control::ControlHello). +/// +/// **Never persisted** (the field carrying it is `#[serde(skip)]`): an +/// attachment describes a live connection; after a server restart there are +/// none, and a stale one on disk would report a takeover against a client +/// that no longer exists. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Attachment { + /// The client's per-session random token, from `ControlHello::client_token`. + pub token: String, + /// The client machine's hostname, shown to the user in the preempted + /// window's status bar. + pub hostname: String, + /// Unix seconds when the attach happened. + pub since: u64, +} + +impl Attachment { + /// An attachment stamped now. + pub fn new(token: impl Into, hostname: impl Into) -> Attachment { + Attachment { + token: token.into(), + hostname: hostname.into(), + since: unix_now(), + } + } +} + /// One workspace: a named group of tabs. The unit a window shows and a client /// attaches to. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -1471,11 +1512,44 @@ fn quarantine(path: &Path) { } } -/// `/machine.json` — beside the old store's `workspaces.json`, under -/// the same directory resolution ([`crate::core::workspace_store`] documents -/// the order). +/// `/machine.json`. +/// +/// | Order | Directory | Why | +/// |---|---|---| +/// | 1 | `$TTY7_DATA_DIR` | Explicit wins; how tests and a second server get their own file | +/// | 2 | `$XDG_DATA_HOME/tty7` | The location the design names, spelled the way XDG spells it | +/// | 3 | `$HOME/.local/share/tty7` | No `XDG_DATA_HOME` — the literal fallback path | +/// +/// Deliberately **not** under the config dir. `views.json` there is the +/// *client's* view state, and a box that is both someone's laptop and someone +/// else's remote must keep the two files apart or one role would overwrite the +/// other's idea of which workspaces exist. pub fn default_machine_path() -> io::Result { - crate::core::workspace_store::default_store_path().map(|p| p.with_file_name(MACHINE_FILE)) + Ok(data_dir()?.join(MACHINE_FILE)) +} + +fn data_dir() -> io::Result { + if let Some(explicit) = std::env::var_os(DATA_DIR_ENV).filter(|v| !v.is_empty()) { + return Ok(PathBuf::from(explicit)); + } + #[cfg(not(windows))] + let base = env_dir("XDG_DATA_HOME") + .or_else(|| env_dir("HOME").map(|h| h.join(".local").join("share"))); + #[cfg(windows)] + let base = env_dir("LOCALAPPDATA") + .or_else(|| env_dir("USERPROFILE").map(|h| h.join(".local").join("share"))); + + base.map(|b| b.join("tty7")).ok_or_else(|| { + io::Error::other(format!( + "no home directory to place {MACHINE_FILE} in; set {DATA_DIR_ENV}" + )) + }) +} + +fn env_dir(key: &str) -> Option { + std::env::var_os(key) + .filter(|v| !v.is_empty()) + .map(PathBuf::from) } fn unix_now() -> u64 { @@ -2115,6 +2189,34 @@ mod tests { ); } + /// An attachment is a field of its workspace, so deleting the workspace + /// takes it along — there is no table it could go stale in. The retired + /// record store kept a separate attachment list and had to clear it by + /// hand; this pins the structural guarantee that replaced that code. + #[test] + fn an_attachment_dies_with_its_workspace() { + let (store, _dir, ws, _tab) = store_with_tab(); + store.attach(ws, Attachment::new("tok", "laptop")); + assert!(store.attachment(ws).is_some()); + store.workspace_delete(ws, None).unwrap(); + assert_eq!(store.attachment(ws), None); + } + + /// The default path ends at the documented file under the data directory — + /// the resolution the retired record store defined and the tree inherited. + #[test] + fn the_default_path_ends_at_the_documented_file() { + match default_machine_path() { + Ok(path) => assert_eq!( + path.file_name().and_then(|n| n.to_str()), + Some(MACHINE_FILE) + ), + // No home at all (a bare CI container): the error names the + // escape hatch rather than being a mystery. + Err(e) => assert!(e.to_string().contains(DATA_DIR_ENV)), + } + } + // ── Corruption ───────────────────────────────────────────────────────── #[test] diff --git a/crates/tty7-core/src/core/mod.rs b/crates/tty7-core/src/core/mod.rs index f749aa70..701e200b 100644 --- a/crates/tty7-core/src/core/mod.rs +++ b/crates/tty7-core/src/core/mod.rs @@ -31,5 +31,4 @@ pub mod shells; pub mod ssh_profile; pub mod threads; pub mod window_state; -pub mod workspace_store; pub mod worktree; diff --git a/crates/tty7-core/src/core/workspace_store.rs b/crates/tty7-core/src/core/workspace_store.rs deleted file mode 100644 index 4fbbf044..00000000 --- a/crates/tty7-core/src/core/workspace_store.rs +++ /dev/null @@ -1,1168 +0,0 @@ -//! The **remote** side of the storage split: the machine's own -//! `~/.local/share/tty7/workspaces.json`, and the one writer to it. -//! -//! # Which half of the split this is -//! -//! | Lives | Holds | Because | -//! |---|---|---| -//! | **Here**, on the machine the panes run on | The workspace list and names, the tab/pane tree, each pane's cwd / `pane_id` / agent, `last_active` | Connect from another laptop and you must see the same thing. This is a fact about the machine | -//! | The **client**'s `session.json` | Which host's which workspaces this client has opened, window geometry, the `open` flag | It is *this client's* view state. Closing a window at the office must not hide the workspace from the laptop at home | -//! -//! [`Workspace::to_remote_json`](crate::core::session::Workspace::to_remote_json) -//! is the client's half of that contract; this module is the server's. -//! -//! # Records are opaque on purpose -//! -//! A record is a [`serde_json::Value`], not a parsed -//! [`Workspace`](crate::core::session::Workspace). The server is a store, not a -//! participant: the client owns the schema, and a client newer than the server -//! it is talking to is the *normal* case (the server is installed once and then -//! left alone for months, auto-install notwithstanding). Parsing here -//! would mean a field the server has never heard of is dropped on the next -//! write — silent data loss whose only symptom is a setting that will not -//! stick. -//! -//! What the store does insist on is the shape it has to index by: a record is a -//! JSON object, and its `id` agrees with the key it was filed under. Those two -//! are what keep the file's array parseable as -//! [`Workspaces`](crate::core::session::Workspaces) by anything that wants the -//! typed view. -//! -//! # Concurrency -//! -//! Several control connections can be writing at once — two of the user's own -//! machines, or one machine reconnecting while the old link has not yet -//! noticed. One mutex covers the record list *and* the file write, so the -//! on-disk order is the in-memory order and no interleaving can produce a file -//! that never existed as a state. The write is atomic -//! ([`write_atomic`](crate::core::config::write_atomic)), so a crash mid-save -//! leaves the old file rather than half of the new one, and a write that fails -//! rolls the memory back rather than leaving the two out of step. -//! -//! Change notifications ([`WorkspaceStore::subscribe`]) are delivered -//! **outside** the lock, and the server's callback only enqueues — a peer that -//! has stopped reading its socket must not be able to stall another peer's -//! `WorkspacePut`. - -use std::io; -use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; - -use serde::{Deserialize, Serialize}; -use serde_json::Value; - -/// The file's name under the data directory. -pub const STORE_FILE: &str = "workspaces.json"; - -/// Overrides where the store lives. Set by tests and by a second server on a -/// shared box — the same escape hatch -/// [`CONTROL_SOCK_ENV`](crate::host::server::CONTROL_SOCK_ENV) is for the -/// socket. -pub const DATA_DIR_ENV: &str = "TTY7_DATA_DIR"; - -/// Ceiling on one record. A workspace with a hundred tabs is a few tens of -/// kilobytes; this is four megabytes, so it only ever catches a client that has -/// gone wrong. Without it a single `WorkspacePut` could pin the file — and the -/// memory holding it — at the 64 MiB frame limit. -pub const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024; - -/// Ceiling on records. Same reasoning one level up: a user has tens of -/// workspaces, and a client looping on "create workspace" should hit a named -/// 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; - -// --------------------------------------------------------------------------- -// Attachment bookkeeping (the data half of M6's takeover) -// --------------------------------------------------------------------------- - -/// Who is currently attached to a workspace. -/// -/// **Data only.** The takeover — push `Preempted { by }` to the old -/// session, close its streams, offer a [抢回] button — is M6's, and none of it -/// is here. What is here is the record that machinery needs to exist before it -/// can be written: the random token that tells two connections from the same -/// client apart, and the hostname that fills in "已在 <主机名> 上打开". Both -/// arrive in the [`ControlHello`](crate::daemon::control::ControlHello). -/// -/// **Never persisted.** An attachment describes a live connection; after a -/// server restart there are none, and a stale one on disk would make M6 report -/// a takeover against a client that no longer exists. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct Attachment { - /// The client's per-session random token, from `ControlHello::client_token`. - pub token: String, - /// The client machine's hostname, shown to the user in the preempted - /// window's status bar. - pub hostname: String, - /// Unix seconds when the attach happened. - pub since: u64, -} - -impl Attachment { - /// An attachment stamped now. - pub fn new(token: impl Into, hostname: impl Into) -> Attachment { - Attachment { - token: token.into(), - hostname: hostname.into(), - since: unix_now(), - } - } -} - -// --------------------------------------------------------------------------- -// Subscriptions -// --------------------------------------------------------------------------- - -/// Identifies one subscriber, so a writer can be told apart from the clients it -/// is notifying. -#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] -pub struct SubscriberId(pub u64); - -/// What a subscriber is told: the id of the workspace that changed. -/// -/// Deliberately not the new contents. The event is a hint to refetch, so a -/// client that missed three of them is in the same state as one that saw all -/// three — which is what makes dropping a notification safe when a peer is -/// behind. -pub type Notify = Arc; - -/// A live subscription. Dropping it unsubscribes, so a connection's teardown -/// cannot leave a callback pointing at a sink nobody is reading. -pub struct Subscription { - store: Arc, - id: SubscriberId, -} - -impl Subscription { - /// This subscriber's id — pass it as the `origin` of your own writes so you - /// are not told about changes you made yourself. - pub fn id(&self) -> SubscriberId { - self.id - } -} - -impl Drop for Subscription { - fn drop(&mut self) { - self.store.unsubscribe(self.id); - } -} - -// --------------------------------------------------------------------------- -// The store -// --------------------------------------------------------------------------- - -/// The machine's workspace records, and the file they are persisted to. -pub struct WorkspaceStore { - path: PathBuf, - state: Mutex, - /// Separate from `state` on purpose: attaching is not a change to the - /// layout, does not write the file, and must not queue behind one. - attachments: Mutex>, - subscribers: Mutex>, - next_subscriber: AtomicU64, -} - -struct State { - /// Insertion-ordered `(id, record)`. A `Vec` rather than a `HashMap` - /// 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 { - /// Open the store at `path`, reading whatever is there. - /// - /// Infallible by design, exactly like - /// [`Workspaces::load`](crate::core::session::Workspaces::load): a machine - /// whose workspace file is missing or unreadable must still serve files and - /// panes. A file that does not parse is copied aside as - /// `workspaces.json.corrupt` before anything can overwrite it, so "the - /// store came up empty" is recoverable by hand rather than terminal. - 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, - attachments: Mutex::new(Vec::new()), - subscribers: Mutex::new(Vec::new()), - next_subscriber: AtomicU64::new(1), - }) - } - - /// Open the store at [`default_store_path`]. - pub fn shared() -> io::Result> { - Ok(WorkspaceStore::open(default_store_path()?)) - } - - /// Where this store is persisted. - pub fn path(&self) -> &Path { - &self.path - } - - // ----- reads ----------------------------------------------------------- - - /// Every record, in file order. Answers - /// [`WorkspaceList`](crate::daemon::control::ControlRequest::WorkspaceList). - pub fn list(&self) -> Vec { - self.locked() - .records - .iter() - .map(|(_, v)| v.clone()) - .collect() - } - - /// One record. `None` means no such workspace, which the server turns into - /// a `NotFound` — distinguishable from a workspace that exists and is - /// empty, which a `null` payload would not be. - pub fn get(&self, id: &str) -> Option { - self.locked() - .records - .iter() - .find(|(k, _)| k == id) - .map(|(_, v)| v.clone()) - } - - /// How many records are on file. - pub fn len(&self) -> usize { - self.locked().records.len() - } - - pub fn is_empty(&self) -> bool { - self.len() == 0 - } - - // ----- writes ---------------------------------------------------------- - - /// File `record` under `id`, replacing any record already there, and - /// persist. - /// - /// `origin` is the subscriber that asked for the change, so it is not - /// notified of its own write; `None` notifies everyone. - /// - /// The record's `id` field, if present, must agree with `id` — a mismatch - /// would put the file's typed view at odds with the store's key, and the - /// next client to read the array would see a workspace under the wrong - /// identity. When absent it is filled in, so a client that only sent the - /// body still produces a well-formed file. - pub fn put(&self, id: &str, mut record: Value, origin: Option) -> io::Result<()> { - check_id(id)?; - let obj = record.as_object_mut().ok_or_else(|| { - io::Error::new( - io::ErrorKind::InvalidInput, - "a workspace record must be a JSON object", - ) - })?; - match obj.get("id") { - Some(Value::String(existing)) if existing == id => {} - Some(other) => { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - format!("workspace record carries id {other} but was filed under {id}"), - )); - } - None => { - obj.insert("id".to_string(), Value::String(id.to_string())); - } - } - - let encoded = serde_json::to_vec(&record).map_err(io::Error::other)?; - if encoded.len() > MAX_RECORD_BYTES { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - format!( - "workspace record is {} bytes; the limit is {MAX_RECORD_BYTES}", - encoded.len() - ), - )); - } - - { - let mut st = self.locked(); - let existing = st.records.iter().position(|(k, _)| k == id); - if existing.is_none() && st.records.len() >= MAX_WORKSPACES { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - format!("this machine already holds {MAX_WORKSPACES} workspaces"), - )); - } - - // Mutate, persist, and undo precisely if the disk said no — the - // in-memory state is what every later read answers from, so it must - // never claim something the file does not. - let undo = match existing { - Some(i) => Undo::Restore(i, std::mem::replace(&mut st.records[i].1, record)), - None => { - st.records.push((id.to_string(), record)); - Undo::Remove(st.records.len() - 1) - } - }; - if let Err(e) = self.persist(&st, true) { - match undo { - Undo::Restore(i, old) => st.records[i].1 = old, - Undo::Remove(i) => { - st.records.remove(i); - } - } - return Err(e); - } - self.restamp(&mut st); - } - - self.notify(id, origin); - Ok(()) - } - - /// Forget a workspace. `false` means there was nothing to forget, which is - /// still success: a delete that raced another client's delete has got what - /// it asked for, and reporting an error would make the client retry - /// something already done. - pub fn delete(&self, id: &str, origin: Option) -> io::Result { - check_id(id)?; - { - let mut st = self.locked(); - let Some(i) = st.records.iter().position(|(k, _)| k == id) else { - return Ok(false); - }; - let removed = st.records.remove(i); - 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 - // takeover against a ghost. - self.attachments_locked().retain(|(k, _)| k != id); - self.notify(id, origin); - Ok(true) - } - - // ----- attachment (M6's data, not M6's behaviour) ---------------------- - - /// Record `who` as the workspace's current session and answer whoever held - /// it before. - /// - /// The previous holder is **the thing M6 acts on**: a `Some` return is - /// exactly the takeover case, and the caller is the one that pushes - /// `Preempted { by }` and closes the old streams. This function does - /// neither — it only makes the fact available. - pub fn attach(&self, workspace: &str, who: Attachment) -> Option { - let mut slots = self.attachments_locked(); - match slots.iter_mut().find(|(k, _)| k == workspace) { - Some((_, current)) => Some(std::mem::replace(current, who)), - None => { - slots.push((workspace.to_string(), who)); - None - } - } - } - - /// Who is attached to `workspace`, if anyone. - pub fn attachment(&self, workspace: &str) -> Option { - self.attachments_locked() - .iter() - .find(|(k, _)| k == workspace) - .map(|(_, a)| a.clone()) - } - - /// Release `workspace`, but **only if `token` still holds it**. - /// - /// The token check is the whole point. A preempted client tears its - /// connection down *after* the new one has attached, and an unconditional - /// release would have that teardown evict the client that just took over — - /// leaving the workspace looking free while a live window is on it. - pub fn detach(&self, workspace: &str, token: &str) -> bool { - let mut slots = self.attachments_locked(); - let before = slots.len(); - slots.retain(|(k, a)| !(k == workspace && a.token == token)); - slots.len() != before - } - - /// Every live attachment, for diagnostics. - pub fn attachments(&self) -> Vec<(String, Attachment)> { - self.attachments_locked().clone() - } - - // ----- change notification --------------------------------------------- - - /// Be told when a record changes. Dropping the returned [`Subscription`] - /// unsubscribes. - /// - /// `f` **must not block**: it runs on the thread of whichever connection - /// made the change, so a callback that waited on a slow peer's socket would - /// let one stalled client hold up everyone else's writes. The control - /// server's callback enqueues onto a bounded channel and returns. - pub fn subscribe(self: &Arc, f: Notify) -> Subscription { - let id = SubscriberId(self.next_subscriber.fetch_add(1, Ordering::Relaxed)); - self.subscribers - .lock() - .unwrap_or_else(|e| e.into_inner()) - .push((id, f)); - Subscription { - store: Arc::clone(self), - id, - } - } - - fn unsubscribe(&self, id: SubscriberId) { - self.subscribers - .lock() - .unwrap_or_else(|e| e.into_inner()) - .retain(|(sid, _)| *sid != id); - } - - /// Fan a change out, skipping the subscriber that caused it. - /// - /// Called with no lock held: a callback is other people's code, and holding - /// the store's mutex across it would make every future write hostage to it. - fn notify(&self, id: &str, origin: Option) { - let subscribers: Vec<(SubscriberId, Notify)> = self - .subscribers - .lock() - .unwrap_or_else(|e| e.into_inner()) - .clone(); - for (sid, f) in subscribers { - if Some(sid) != origin { - f(id); - } - } - } - - // ----- internals ------------------------------------------------------- - - fn locked(&self) -> std::sync::MutexGuard<'_, State> { - // A poisoned lock means a panic between a mutation and its write. The - // 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. - 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)>> { - self.attachments.lock().unwrap_or_else(|e| e.into_inner()) - } - - /// Serialize the whole file and replace it atomically. - /// - /// The document is `{"workspaces": [...]}` — the identical shape - /// [`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. - /// 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>, - } - let doc = Doc { - 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. -enum Undo { - Restore(usize, Value), - Remove(usize), -} - -/// A key has to be something that can key a JSON object and appear in a log -/// line. It is never used to build a path, so this is a sanity check rather -/// than a security boundary. -fn check_id(id: &str) -> io::Result<()> { - if id.is_empty() { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "a workspace id must not be empty", - )); - } - if id.len() > MAX_ID_BYTES { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - format!("a workspace id must be at most {MAX_ID_BYTES} bytes"), - )); - } - if id.chars().any(char::is_control) { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "a workspace id must not contain control characters", - )); - } - Ok(()) -} - -/// Read the file, keeping whatever is well-formed. -/// -/// One unparseable *record* costs that record, not the file: a client that -/// wrote something odd should not make the user's other twelve workspaces -/// disappear. An unparseable *file* is quarantined and the store comes up -/// empty. -fn load_records(path: &Path) -> Vec<(String, Value)> { - let text = match std::fs::read_to_string(path) { - Ok(t) => t, - Err(e) if e.kind() == io::ErrorKind::NotFound => return Vec::new(), - Err(e) => { - log::warn!("could not read {}: {e}; starting empty", path.display()); - return Vec::new(); - } - }; - - let value: Value = match serde_json::from_str(crate::core::config::strip_bom(&text)) { - Ok(v) => v, - Err(e) => { - log::warn!("{} does not parse ({e}); quarantining it", path.display()); - quarantine(path); - return Vec::new(); - } - }; - let Some(array) = value.get("workspaces").and_then(Value::as_array) else { - log::warn!( - "{} has no `workspaces` array; quarantining it", - path.display() - ); - quarantine(path); - return Vec::new(); - }; - - let mut records: Vec<(String, Value)> = Vec::with_capacity(array.len()); - for record in array { - let Some(id) = record.get("id").and_then(Value::as_str) else { - log::warn!("dropping a workspace record with no string `id`"); - continue; - }; - if check_id(id).is_err() { - log::warn!("dropping a workspace record with an unusable id"); - continue; - } - if records.iter().any(|(k, _)| k == id) { - log::warn!("dropping a duplicate record for workspace {id}"); - continue; - } - records.push((id.to_string(), record.clone())); - } - records -} - -/// Copy a file we are about to stop honouring somewhere the user can find it. -/// Best effort: failing to make the backup is not a reason to refuse to start. -fn quarantine(path: &Path) { - let aside = path.with_extension("json.corrupt"); - match std::fs::copy(path, &aside) { - Ok(_) => log::warn!("the previous contents were kept at {}", aside.display()), - Err(e) => log::warn!("could not keep a copy at {}: {e}", aside.display()), - } -} - -/// `/workspaces.json`. -/// -/// | Order | Directory | Why | -/// |---|---|---| -/// | 1 | `$TTY7_DATA_DIR` | Explicit wins; how tests and a second server get their own file | -/// | 2 | `$XDG_DATA_HOME/tty7` | The location the design names, spelled the way XDG spells it | -/// | 3 | `$HOME/.local/share/tty7` | No `XDG_DATA_HOME` — the literal fallback path | -/// -/// Deliberately **not** under the config dir. `session.json` there is the -/// *client's* view state, and a box that is both someone's laptop and someone -/// else's remote must keep the two files apart or one role would overwrite the -/// other's idea of which workspaces exist. -pub fn default_store_path() -> io::Result { - Ok(data_dir()?.join(STORE_FILE)) -} - -fn data_dir() -> io::Result { - if let Some(explicit) = std::env::var_os(DATA_DIR_ENV).filter(|v| !v.is_empty()) { - return Ok(PathBuf::from(explicit)); - } - #[cfg(not(windows))] - let base = env_dir("XDG_DATA_HOME") - .or_else(|| env_dir("HOME").map(|h| h.join(".local").join("share"))); - #[cfg(windows)] - let base = env_dir("LOCALAPPDATA") - .or_else(|| env_dir("USERPROFILE").map(|h| h.join(".local").join("share"))); - - base.map(|b| b.join("tty7")).ok_or_else(|| { - io::Error::other(format!( - "no home directory to place {STORE_FILE} in; set {DATA_DIR_ENV}" - )) - }) -} - -fn env_dir(key: &str) -> Option { - std::env::var_os(key) - .filter(|v| !v.is_empty()) - .map(PathBuf::from) -} - -fn unix_now() -> u64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::core::session::WorkspaceId; - use std::sync::atomic::AtomicUsize; - - fn store() -> (Arc, tempfile::TempDir) { - let dir = tempfile::TempDir::new().unwrap(); - let store = WorkspaceStore::open(dir.path().join(STORE_FILE)); - (store, dir) - } - - fn record(id: &str, name: &str) -> Value { - serde_json::json!({ - "id": id, - "name": name, - "session": {"active": 0, "tabs": [ - {"pane": {"Leaf": {"cwd": "/home/me/proj", "pane_id": 7}}} - ]}, - "last_active": 1_753_600_000u64, - }) - } - - /// 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] - fn a_missing_file_is_an_empty_store_not_an_error() { - let (store, _dir) = store(); - assert!(store.is_empty()); - assert!(store.list().is_empty()); - assert_eq!(store.get("nope"), None); - // And deleting nothing is success, not an error. - assert!(!store.delete("nope", None).unwrap()); - } - - #[test] - fn put_get_list_delete_round_trip_through_the_file() { - let (store, dir) = store(); - store.put("a", record("a", "api"), None).unwrap(); - store.put("b", record("b", "web"), None).unwrap(); - assert_eq!(store.len(), 2); - assert_eq!(store.get("a").unwrap()["name"], "api"); - - // A second store over the same path sees it: the file is the authority, - // which is the entire reason this lives on the remote. - let reopened = WorkspaceStore::open(dir.path().join(STORE_FILE)); - assert_eq!(reopened.len(), 2); - assert_eq!(reopened.get("b").unwrap()["name"], "web"); - // File order is list order. - let names: Vec = reopened - .list() - .iter() - .map(|v| v["name"].as_str().unwrap().to_string()) - .collect(); - assert_eq!(names, vec!["api", "web"]); - - assert!(store.delete("a", None).unwrap()); - assert_eq!( - WorkspaceStore::open(dir.path().join(STORE_FILE)).len(), - 1, - "a delete has to reach the disk, not just the map" - ); - } - - #[test] - fn replacing_a_record_keeps_its_place_in_the_list() { - let (store, _dir) = store(); - for id in ["a", "b", "c"] { - store.put(id, record(id, id), None).unwrap(); - } - store.put("a", record("a", "renamed"), None).unwrap(); - let ids: Vec = store - .list() - .iter() - .map(|v| v["id"].as_str().unwrap().to_string()) - .collect(); - assert_eq!(ids, vec!["a", "b", "c"], "a rename must not reshuffle"); - assert_eq!(store.get("a").unwrap()["name"], "renamed"); - } - - // ── Validation ────────────────────────────────────────────────────────── - - #[test] - fn a_record_must_be_an_object_whose_id_agrees_with_its_key() { - let (store, _dir) = store(); - let kinds = |e: io::Error| e.kind(); - assert_eq!( - store - .put("a", serde_json::json!([1, 2, 3]), None) - .map_err(kinds), - Err(io::ErrorKind::InvalidInput) - ); - assert_eq!( - store - .put("a", serde_json::json!({"id": "b"}), None) - .map_err(kinds), - Err(io::ErrorKind::InvalidInput), - "filing b's record under a would put the key and the file at odds" - ); - assert_eq!( - store.put("", record("", "x"), None).map_err(kinds), - Err(io::ErrorKind::InvalidInput) - ); - assert!(store.is_empty(), "a rejected put must not be half-applied"); - - // A body with no id is completed rather than refused: the key is the - // authority and the file still ends up well-formed. - store - .put("a", serde_json::json!({"name": "api"}), None) - .unwrap(); - assert_eq!(store.get("a").unwrap()["id"], "a"); - } - - #[test] - fn oversized_and_overnumerous_records_are_refused_by_name() { - let (store, _dir) = store(); - let huge = serde_json::json!({"name": "x".repeat(MAX_RECORD_BYTES + 16)}); - assert_eq!( - store.put("a", huge, None).unwrap_err().kind(), - io::ErrorKind::InvalidInput - ); - assert!(store.is_empty()); - } - - // ── Corruption ────────────────────────────────────────────────────────── - - #[test] - fn a_corrupt_file_is_quarantined_rather_than_overwritten() { - let dir = tempfile::TempDir::new().unwrap(); - let path = dir.path().join(STORE_FILE); - std::fs::write(&path, b"{ this is not json").unwrap(); - - let store = WorkspaceStore::open(&path); - assert!( - store.is_empty(), - "an unparseable file yields an empty store" - ); - // The user's bytes are still recoverable after the store overwrites the - // original. - store.put("a", record("a", "api"), None).unwrap(); - let aside = std::fs::read_to_string(path.with_extension("json.corrupt")).unwrap(); - assert_eq!(aside, "{ this is not json"); - } - - #[test] - fn one_bad_record_does_not_cost_the_others() { - let dir = tempfile::TempDir::new().unwrap(); - let path = dir.path().join(STORE_FILE); - std::fs::write( - &path, - br#"{"workspaces":[ - {"id":"a","name":"api"}, - {"name":"no id at all"}, - {"id":42}, - {"id":"a","name":"duplicate"}, - {"id":"b","name":"web"} - ]}"#, - ) - .unwrap(); - let store = WorkspaceStore::open(&path); - assert_eq!(store.len(), 2); - assert_eq!(store.get("a").unwrap()["name"], "api", "the first wins"); - assert_eq!(store.get("b").unwrap()["name"], "web"); - } - - #[test] - fn a_utf8_bom_does_not_empty_the_store() { - let dir = tempfile::TempDir::new().unwrap(); - let path = dir.path().join(STORE_FILE); - std::fs::write(&path, "\u{FEFF}{\"workspaces\":[{\"id\":\"a\"}]}").unwrap(); - assert_eq!(WorkspaceStore::open(&path).len(), 1); - } - - // ── Notification ──────────────────────────────────────────────────────── - - #[test] - fn a_change_notifies_every_subscriber_but_its_author() { - let (store, _dir) = store(); - let heard_by_a = Arc::new(Mutex::new(Vec::::new())); - let heard_by_b = Arc::new(Mutex::new(Vec::::new())); - let sink = |log: &Arc>>| { - let log = Arc::clone(log); - Arc::new(move |id: &str| log.lock().unwrap().push(id.to_string())) as Notify - }; - let a = store.subscribe(sink(&heard_by_a)); - let _b = store.subscribe(sink(&heard_by_b)); - - // A writes: B hears about it, A does not hear its own change. - store.put("w1", record("w1", "one"), Some(a.id())).unwrap(); - assert!(heard_by_a.lock().unwrap().is_empty()); - assert_eq!(&*heard_by_b.lock().unwrap(), &["w1".to_string()]); - - // A delete is a change too, and a write with no origin reaches all. - store.delete("w1", Some(a.id())).unwrap(); - store.put("w2", record("w2", "two"), None).unwrap(); - assert_eq!(&*heard_by_a.lock().unwrap(), &["w2".to_string()]); - assert_eq!( - &*heard_by_b.lock().unwrap(), - &["w1".to_string(), "w1".to_string(), "w2".to_string()] - ); - - // Deleting nothing changed nothing, so it says nothing. - let before = heard_by_b.lock().unwrap().len(); - assert!(!store.delete("gone", None).unwrap()); - assert_eq!(heard_by_b.lock().unwrap().len(), before); - } - - #[test] - fn dropping_a_subscription_stops_the_notifications() { - let (store, _dir) = store(); - let count = Arc::new(AtomicUsize::new(0)); - let seen = Arc::clone(&count); - let sub = store.subscribe(Arc::new(move |_| { - seen.fetch_add(1, Ordering::SeqCst); - })); - store.put("a", record("a", "x"), None).unwrap(); - assert_eq!(count.load(Ordering::SeqCst), 1); - drop(sub); - store.put("b", record("b", "y"), None).unwrap(); - assert_eq!( - count.load(Ordering::SeqCst), - 1, - "a torn-down connection must not still be written to" - ); - } - - /// A rejected put changed nothing, so it must not claim otherwise. - #[test] - fn a_failed_put_notifies_nobody() { - let (store, _dir) = store(); - let count = Arc::new(AtomicUsize::new(0)); - let seen = Arc::clone(&count); - let _sub = store.subscribe(Arc::new(move |_| { - seen.fetch_add(1, Ordering::SeqCst); - })); - store - .put("a", serde_json::json!("not an object"), None) - .ok(); - assert_eq!(count.load(Ordering::SeqCst), 0); - } - - // ── Concurrency ───────────────────────────────────────────────────────── - - /// Several connections writing at once is the normal case, not the - /// pathological one. Every write must land, and the file must end up as a - /// state that actually existed — not a half-written interleaving. - #[test] - fn concurrent_writers_all_land_and_the_file_stays_whole() { - let dir = tempfile::TempDir::new().unwrap(); - let path = dir.path().join(STORE_FILE); - let store = WorkspaceStore::open(&path); - - let threads: Vec<_> = (0..8) - .map(|t| { - let store = Arc::clone(&store); - std::thread::spawn(move || { - for i in 0..25 { - let id = format!("w{t}-{i}"); - store.put(&id, record(&id, "x"), None).unwrap(); - } - }) - }) - .collect(); - for t in threads { - t.join().unwrap(); - } - - assert_eq!(store.len(), 200); - // And the file on disk agrees, which is the part a torn write would - // fail: it would not parse at all. - let reopened = WorkspaceStore::open(&path); - assert_eq!(reopened.len(), 200); - for t in 0..8 { - assert!(reopened.get(&format!("w{t}-24")).is_some()); - } - } - - /// The same workspace written from two connections: last writer wins, and - /// the loser's record is gone rather than merged into a hybrid neither - /// client asked for. - #[test] - fn concurrent_writes_to_one_record_are_last_writer_wins() { - let (store, _dir) = store(); - let a = Arc::clone(&store); - let b = Arc::clone(&store); - let ta = std::thread::spawn(move || { - for _ in 0..200 { - a.put("w", record("w", "from-a"), None).unwrap(); - } - }); - let tb = std::thread::spawn(move || { - for _ in 0..200 { - b.put("w", record("w", "from-b"), None).unwrap(); - } - }); - ta.join().unwrap(); - tb.join().unwrap(); - assert_eq!(store.len(), 1); - let name = store.get("w").unwrap()["name"] - .as_str() - .unwrap() - .to_string(); - assert!(name == "from-a" || name == "from-b", "{name}"); - } - - // ── Attachment (M6's data) ────────────────────────────────────────────── - - #[test] - fn attaching_reports_the_session_it_displaced() { - let (store, _dir) = store(); - assert_eq!(store.attachment("w"), None); - - let laptop = Attachment::new("tok-1", "laptop"); - assert_eq!( - store.attach("w", laptop.clone()), - None, - "nothing to preempt" - ); - assert_eq!(store.attachment("w"), Some(laptop.clone())); - - // The second client's attach hands back the first — the exact fact M6's - // takeover acts on. - let desktop = Attachment::new("tok-2", "desktop"); - assert_eq!(store.attach("w", desktop.clone()), Some(laptop.clone())); - assert_eq!(store.attachment("w"), Some(desktop)); - - // The preempted client tearing down afterwards must not evict the new - // owner: its token no longer holds the workspace. - assert!(!store.detach("w", &laptop.token)); - assert_eq!(store.attachment("w").unwrap().hostname, "desktop"); - assert!(store.detach("w", "tok-2")); - assert_eq!(store.attachment("w"), None); - } - - #[test] - fn attachments_are_scoped_to_a_workspace_and_die_with_it() { - let (store, _dir) = store(); - store.put("w", record("w", "one"), None).unwrap(); - store.attach("w", Attachment::new("tok", "laptop")); - store.attach("other", Attachment::new("tok", "laptop")); - assert_eq!(store.attachments().len(), 2); - - store.delete("w", None).unwrap(); - assert_eq!(store.attachment("w"), None); - assert!(store.attachment("other").is_some()); - } - - /// Attachments describe live connections, so they must not outlive the - /// process that held them. - #[test] - fn attachments_are_never_written_to_the_file() { - let dir = tempfile::TempDir::new().unwrap(); - let path = dir.path().join(STORE_FILE); - let store = WorkspaceStore::open(&path); - store.put("w", record("w", "one"), None).unwrap(); - store.attach("w", Attachment::new("secret-token", "laptop")); - - let text = std::fs::read_to_string(&path).unwrap(); - assert!(!text.contains("secret-token"), "{text}"); - assert!(!text.contains("laptop"), "{text}"); - assert_eq!( - WorkspaceStore::open(&path).attachment("w"), - None, - "a restarted server has no attached clients" - ); - } - - // ── Path resolution ───────────────────────────────────────────────────── - - #[test] - fn the_store_path_ends_at_the_documented_file() { - // `TTY7_DATA_DIR` is process-global, so this only asserts the shape the - // resolution produces rather than setting the variable under other - // tests running beside it. - let p = WorkspaceStore::open(PathBuf::from("/srv/data/tty7").join(STORE_FILE)); - assert!(p.path().ends_with("tty7/workspaces.json")); - } - - #[test] - fn a_workspace_id_is_a_usable_store_key() { - let (store, _dir) = store(); - let id = WorkspaceId::new().to_string(); - store.put(&id, record(&id, "api"), None).unwrap(); - assert!(store.get(&id).is_some()); - } -} diff --git a/crates/tty7-core/src/daemon/control.rs b/crates/tty7-core/src/daemon/control.rs index 3593db5b..c97c9f38 100644 --- a/crates/tty7-core/src/daemon/control.rs +++ b/crates/tty7-core/src/daemon/control.rs @@ -90,7 +90,7 @@ use super::protocol::{MAX_FRAME, read_frame, write_frame}; /// comparing dialect numbers, so a capability that doesn't move the number is a /// capability the far machine never gets. A [`feature`] string is the right /// answer only for something two current servers can genuinely disagree about -/// (the workspace store, which depends on how the server was started); "this +/// (the machine tree, which depends on how the server was started); "this /// build knows the request and older ones don't" is what the number is for. /// /// ## History @@ -184,18 +184,19 @@ pub mod feature { pub const CONTROL: &str = "control"; /// Serves [`super::ControlRequest`]'s filesystem and git methods — i.e. can /// back a remote `Host`. Distinct from [`CONTROL`] because a peer could - /// speak the dialect while exposing only the workspace store. + /// speak the dialect while exposing only the workspace tree. pub const HOST_RPC: &str = "host-rpc"; - /// Serves the `Workspace*` requests. - pub const WORKSPACE_STORE: &str = "workspace-store"; + // `"workspace-store"` is a burned name: it advertised the retired + // opaque-record scheme (verbs `workspace_list` / `workspace_get` / + // `workspace_put` / `workspace_delete`, event `workspace_changed`), all of + // which are burned with it. Never re-advertise or re-mint any of them with + // a different meaning. /// Serves the machine-owned workspace tree: the `MachineGet` / /// `WorkspaceTree` pulls, the semantic tree operations, and the /// [`super::ControlEvent::Layout`] pushes. Advertised only when the server /// actually carries a [`crate::core::machine::MachineStore`], so a client /// learns from the handshake whether the tree verbs are worth a round - /// trip. Distinct from [`WORKSPACE_STORE`], which is the retired - /// opaque-record scheme this one replaces — the two coexist while clients - /// migrate. + /// trip. pub const MACHINE_TREE: &str = "machine-tree"; /// Can be launched as `--stdio` and bridge its own stdin/stdout to the /// machine-local socket (the fallback when `AllowStreamLocalForwarding` is @@ -345,18 +346,10 @@ pub enum ControlRequest { id: u64, }, - // ----- workspace store (M5; the slots exist, the server doesn't yet) ----- - WorkspaceList, - WorkspaceGet { - id: String, - }, - WorkspacePut { - id: String, - json: serde_json::Value, - }, - WorkspaceDelete { - id: String, - }, + // The opaque record store's verbs — `workspace_list` / `workspace_get` / + // `workspace_put` / `workspace_delete` — lived here until the machine tree + // below replaced them. Their serde names are burned (see `feature`); do + // not re-mint them with a different meaning. // ----- attachment (M6's takeover) --------------------------------------- /// Claim a workspace for this connection's session, taking it over from @@ -382,10 +375,10 @@ pub enum ControlRequest { }, // ----- machine tree (the daemon-owned structure) ------------------------ - // The semantic replacement for the opaque `Workspace*` record verbs above: - // instead of a whole-record `Put` (last-writer-wins the moment two clients - // write), each operation names its edit, the server validates it against - // the tree it owns, and everyone else hears an incremental + // The semantic replacement for the retired opaque record verbs: instead of + // a whole-record `Put` (last-writer-wins the moment two clients write), + // each operation names its edit, the server validates it against the tree + // it owns, and everyone else hears an incremental // [`ControlEvent::Layout`]. Positions cross as `u64` for the same reason // `Search`'s limits do: a 32-bit server clamps rather than wraps. /// The whole tree — every workspace, tab and pane record on the machine. @@ -412,7 +405,8 @@ pub enum ControlRequest { name: Option, }, /// Forget a tree workspace and everything under it. Named `Remove` because - /// `WorkspaceDelete` above is taken by the retired record store's verb. + /// `WorkspaceDelete` was the retired record store's verb, and its serde + /// name stays burned. WorkspaceRemove { workspace: WorkspaceId, }, @@ -534,16 +528,12 @@ impl ControlRequest { // spawns `wsl.exe -l -q`, which is slow enough to deserve the same // budget as git. Shells => Duration::from_secs(20), - WorkspaceList | WorkspaceGet { .. } | WorkspacePut { .. } | WorkspaceDelete { .. } => { - Duration::from_secs(10) - } // An attach is bookkeeping plus at most one push to a peer that may // be wedged — the push is `try`-shaped on the server, so this only // has to cover a slow link, not a slow client. WorkspaceAttach { .. } | WorkspaceDetach { .. } => Duration::from_secs(10), // Tree operations are a locked mutation plus one small file write, - // so the budget is the record verbs': it covers a slow disk, not - // slow work. + // so the budget covers a slow disk, not slow work. MachineGet | WorkspaceTree { .. } | WorkspaceCreate { .. } @@ -621,8 +611,6 @@ pub enum ReplyOk { WatchId(u64), /// [`ControlRequest::Shells`]: what that machine can launch. Shells(ShellInventory), - /// The workspace store's payload (M5). - Json(serde_json::Value), /// [`ControlRequest::WorkspaceAttach`] succeeded. `took_over_from` names the /// machine whose session was displaced, so the client that *did* the taking /// can say so — only the notice going the other way is specified, @@ -782,9 +770,8 @@ pub enum ControlEvent { workspace: String, by: String, }, - WorkspaceChanged { - id: String, - }, + // `workspace_changed` was the retired record store's change notice; its + // serde name is burned along with the record verbs. /// One incremental change to one tree workspace on this machine — the /// push half of the machine-tree verbs. The writer never receives its own /// operation back (origin exclusion, so an optimistically-applied edit is @@ -793,7 +780,7 @@ pub enum ControlEvent { /// [`ControlRequest::WorkspaceTree`]. /// /// `workspace` is the [`WorkspaceId`] rendered as a string, matching how - /// `Preempted` and `WorkspaceChanged` name theirs. + /// `Preempted` names its. Layout { workspace: String, delta: LayoutDelta, @@ -805,7 +792,7 @@ pub enum ControlEvent { /// [`RemoteHost`](crate::host::remote::RemoteHost) routes `Watch` and /// `WatchOverflow` into the subscription that asked for them, because those /// belong to a caller that is still holding a `WatchSub`. The rest — -/// `Preempted`, `PaneExited`, `AgentStatus`, `WorkspaceChanged` — are about a +/// `Preempted`, `PaneExited`, `AgentStatus`, `Layout` — are about a /// *window*, and the host layer has no window. /// /// A process-wide observer rather than a parameter on `connect_with` because @@ -2001,13 +1988,6 @@ mod tests { dirs: vec!["/home/me/proj".into(), "/home/me/proj/src".into()], }, ControlRequest::WatchClose { id: 7 }, - ControlRequest::WorkspaceList, - ControlRequest::WorkspaceGet { id: "w1".into() }, - ControlRequest::WorkspacePut { - id: "w1".into(), - json: serde_json::json!({ "tabs": [1, 2, 3] }), - }, - ControlRequest::WorkspaceDelete { id: "w1".into() }, ] } @@ -2054,7 +2034,6 @@ mod tests { stderr: vec![0x00, 0xff, 0xfe, b'\n'], })), ControlReply::Ok(ReplyOk::WatchId(42)), - ControlReply::Ok(ReplyOk::Json(serde_json::json!({ "a": [1, null] }))), ControlReply::Err(WireError::new(WireErrorKind::NotFound, "no such file")), ControlReply::Err(WireError::new( WireErrorKind::PermissionDenied, @@ -2102,7 +2081,6 @@ mod tests { workspace: "w1".into(), by: "other-laptop".into(), }, - ControlEvent::WorkspaceChanged { id: "w1".into() }, ] } @@ -2712,16 +2690,6 @@ mod tests { }, s(20), ), - (R::WorkspaceList, s(10)), - (R::WorkspaceGet { id: "w".into() }, s(10)), - ( - R::WorkspacePut { - id: "w".into(), - json: serde_json::Value::Null, - }, - s(10), - ), - (R::WorkspaceDelete { id: "w".into() }, s(10)), ]; assert_eq!( cases.len(), @@ -2760,7 +2728,9 @@ mod tests { assert_eq!(ok.home, "/home/me"); assert!(ok.has_feature(feature::CONTROL)); assert!(ok.has_feature(feature::HOST_RPC)); - assert!(!ok.has_feature(feature::WORKSPACE_STORE)); + // The retired record store's bit is a burned name and must + // never come back. + assert!(!ok.has_feature("workspace-store")); } other => panic!("expected HelloOk, got {other:?}"), } diff --git a/crates/tty7-core/src/daemon/router.rs b/crates/tty7-core/src/daemon/router.rs index 862c54d6..e94fb83e 100644 --- a/crates/tty7-core/src/daemon/router.rs +++ b/crates/tty7-core/src/daemon/router.rs @@ -142,7 +142,7 @@ const REPLY_TIMEOUT: Duration = Duration::from_secs(240); #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum RouteChannel { - /// Host RPC, the workspace store, event pushes — `daemon::control`. + /// Host RPC, the machine tree, event pushes — `daemon::control`. #[default] Control, /// One pane: `Spawn`/`Attach`/`Input`/`Output` — `daemon::protocol`. diff --git a/crates/tty7-core/src/daemon/server.rs b/crates/tty7-core/src/daemon/server.rs index 3e60b712..cc236ebf 100644 --- a/crates/tty7-core/src/daemon/server.rs +++ b/crates/tty7-core/src/daemon/server.rs @@ -226,37 +226,22 @@ pub fn run_daemon() -> anyhow::Result<()> { /// What this machine offers over a control connection, beyond its filesystem. /// -/// The workspace store is why a daemon serves control at all: the workspace +/// The machine tree is why a daemon serves control at all: the workspace /// list, the tab/pane tree and each pane's facts live on **the machine the /// panes run on**, so that every client of this machine — the GUI on it, a /// laptop across the world — sees the same thing. Clients keep only their own /// view state. /// /// A machine with no home directory to place the file in still serves files -/// and panes — it simply omits `workspace-store` from its capabilities, and -/// clients see the same "does not serve the workspace store" answer a server +/// and panes — it simply omits `machine-tree` from its capabilities, and +/// clients see the same "does not serve the machine tree" answer a server /// without one has always given. pub fn control_services() -> crate::host::server::Services { use crate::core::machine::MachineStore; - use crate::core::workspace_store::WorkspaceStore; // Reported on stderr as well as the log, like the socket line in // [`run_daemon`]: on a headless box the log file is off by default, and - // "which stores does this daemon actually serve" is the first question a + // "does this daemon actually serve the tree" is the first question a // capability mismatch raises. - let services = match WorkspaceStore::shared() { - Ok(store) => { - eprintln!("workspace store at {}", store.path().display()); - crate::host::server::Services::with_workspaces(store) - } - Err(e) => { - eprintln!("no workspace store ({e}); serving files and panes only"); - crate::host::server::Services::none() - } - }; - // The machine tree rides alongside the record store while clients migrate - // from whole-record `Put` to the semantic operations; both resolve their - // file under the same data directory, so a machine that can hold one can - // hold the other. match MachineStore::shared() { Ok(machine) => { eprintln!("machine tree at {}", machine.path().display()); @@ -265,11 +250,11 @@ pub fn control_services() -> crate::host::server::Services { // what a client revives from is what the machine saw, not what // some client last remembered to write. crate::core::machine::publish_observations(&machine); - services.and_machine(machine) + crate::host::server::Services::with_machine(machine) } Err(e) => { - eprintln!("no machine tree ({e}); its verbs stay unserved"); - services + eprintln!("no machine tree ({e}); serving files and panes only"); + crate::host::server::Services::none() } } } diff --git a/crates/tty7-core/src/host/remote.rs b/crates/tty7-core/src/host/remote.rs index f7ec17db..42677b29 100644 --- a/crates/tty7-core/src/host/remote.rs +++ b/crates/tty7-core/src/host/remote.rs @@ -162,7 +162,7 @@ impl RemoteHost { } /// The underlying connection, for callers that need to speak control - /// directly (the workspace store, once it exists). + /// directly (the machine-tree verbs). pub fn client(&self) -> &Arc { &self.client } diff --git a/crates/tty7-core/src/host/server.rs b/crates/tty7-core/src/host/server.rs index ccf9e1c1..778bcc5a 100644 --- a/crates/tty7-core/src/host/server.rs +++ b/crates/tty7-core/src/host/server.rs @@ -46,8 +46,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Condvar, Mutex}; use std::time::Duration; -use crate::core::machine::{self, MachineStore}; -use crate::core::workspace_store::{Attachment, SubscriberId, Subscription, WorkspaceStore}; +use crate::core::machine::{self, Attachment, MachineStore}; use crate::daemon::control::{ CONTROL_VERSION, ControlClientMsg, ControlEvent, ControlHello, ControlHelloOk, ControlReply, ControlRequest, ControlServerMsg, LinkShutdown, ReplyOk, WATCH_BURST_CAP, WireError, @@ -73,21 +72,11 @@ pub const WORKER_LINGER: Duration = Duration::from_secs(10); /// would each be answered long after the client's own deadline gave up on them. pub const MAX_QUEUED: usize = 1024; -/// `WorkspaceChanged` pushes one connection will let pile up before it starts -/// dropping them. -/// -/// Dropping is safe here in a way it is not for a watch batch: the event says -/// only "workspace `id` changed, refetch", so a client that has one queued -/// already learns everything a second one would tell it. The cap exists so a -/// peer that has stopped reading its socket cannot turn another client's -/// `WorkspacePut` into unbounded memory. -pub const WORKSPACE_EVENT_QUEUE: usize = 64; - /// `Layout` deltas one connection will let queue before it starts dropping. /// -/// Unlike a `WorkspaceChanged` push, a delta is *not* self-superseding — a -/// dropped one leaves the peer's picture of the tree wrong until its next full -/// pull. The cap is still right, for the same reason as the watch caps: a peer +/// A delta is *not* self-superseding — a dropped one leaves the peer's +/// picture of the tree wrong until its next full pull. The cap is still +/// right, for the same reason as the watch caps: a peer /// that has stopped reading its socket must not turn another client's edit /// into unbounded server memory. What makes the drop survivable is that such a /// peer is already inside [`crate::daemon::control::KEEPALIVE_DEAD_AFTER`] of @@ -104,24 +93,15 @@ pub const LAYOUT_EVENT_QUEUE: usize = 1024; /// /// Separate from the `SharedHost` argument because the two are genuinely /// independent roles, and the handshake says so: a box can back a remote -/// workspace's file tree without owning any workspace records (that is every -/// server today, and it is what [`Services::default`] produces), and the -/// `workspace-store` capability bit is advertised only when this actually -/// carries a store. A client therefore learns from the handshake whether asking -/// is worth a round trip. +/// workspace's file tree without owning any workspace tree (which is what +/// [`Services::default`] produces), and the `machine-tree` capability bit is +/// advertised only when this actually carries one. A client therefore learns +/// from the handshake whether asking is worth a round trip. #[derive(Clone, Default)] pub struct Services { - /// The machine's workspace records. `None` answers every `Workspace*` - /// request with "this server does not serve the workspace store" — the same - /// answer a build from before M5 gives. - pub workspaces: Option>, /// The machine's own workspace *tree* — the daemon-owned structure the - /// semantic operations edit, replacing the opaque record store above. - /// Carried separately while the two schemes coexist: clients that still - /// speak whole-record `Put` keep working against `workspaces`, and the - /// `machine-tree` capability bit is advertised only when this is here. - /// `None` answers every tree verb with "this server does not serve the - /// machine tree". + /// semantic operations edit. `None` answers every tree verb with "this + /// server does not serve the machine tree". pub machine: Option>, /// Who currently holds each workspace, and how to reach them. Shared across /// every connection this server accepts — that sharing *is* the takeover: @@ -131,27 +111,18 @@ pub struct Services { } impl Services { - /// Host RPC only, no workspace store. + /// Host RPC only, no machine tree. pub fn none() -> Services { Services::default() } - /// Host RPC plus the workspace store. - pub fn with_workspaces(store: Arc) -> Services { + /// Host RPC plus the machine tree. + pub fn with_machine(store: Arc) -> Services { Services { - workspaces: Some(store), - machine: None, + machine: Some(store), attachments: Arc::new(AttachRegistry::default()), } } - - /// `self`, also serving the machine tree. Builder-shaped because the tree - /// rides alongside whatever else the server carries — a store, or nothing - /// but host RPC — rather than replacing it. - pub fn and_machine(mut self, store: Arc) -> Services { - self.machine = Some(store); - self - } } // --------------------------------------------------------------------------- @@ -160,11 +131,11 @@ impl Services { /// The live half of the attachment record. /// -/// [`Attachment`](crate::core::workspace_store::Attachment) in the store is the +/// [`Attachment`](crate::core::machine::Attachment) in the machine tree is the /// *data* — token, hostname, since — and answers "who holds this workspace". /// This is the *handles*: the sink a `Preempted` push goes out on and the /// shutdown that closes the displaced session's link. They are separate because -/// the store lives in `core` and knows nothing about sockets, and because an +/// the tree lives in `core` and knows nothing about sockets, and because an /// attachment must never be written to the file (a stale one on disk would have /// the server report a takeover against a client that no longer exists). /// @@ -178,7 +149,7 @@ pub struct AttachRegistry { /// 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 + /// registry's handles, and the `MachineStore`'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 @@ -422,12 +393,10 @@ where } }; - // Subscribed before the first request is read, so a change another client - // makes while this one is still listing cannot slip through the gap. - let workspace_sub = subscribe_workspaces(&services, &sink); - // Same rule, and the same gap, for the machine tree's deltas: a full pull - // issued after this point can race a delta (the client tolerates that), - // but an edit can never fall between subscription and first read. + // Subscribed before the first request is read, so an edit another client + // makes while this one is still pulling cannot slip through the gap: a + // full pull issued after this point can race a delta (the client tolerates + // that), but an edit can never fall between subscription and first read. let machine_sub = subscribe_machine(&services, &sink); let conn = Arc::new(Conn { @@ -438,8 +407,6 @@ where deferred_watches: Mutex::new(HashMap::new()), next_watch: AtomicU64::new(1), pool: Pool::new(), - workspaces: services.workspaces.clone(), - workspace_origin: workspace_sub.as_ref().map(Subscription::id), machine: services.machine.clone(), machine_origin: machine_sub.as_ref().map(machine::Subscription::id), attachments: Arc::clone(&services.attachments), @@ -467,7 +434,7 @@ where // Teardown, in the order that makes each step meaningful: stop accepting // work, drop the watches (which stops the pushes and releases the OS - // watchers), release anything this session still holds, drop the workspace + // watchers), release anything this session still holds, drop the machine // subscription (which ends its forwarder), then close the link so anything // still writing fails fast rather than blocking on a peer that is gone. conn.pool.close(); @@ -476,7 +443,6 @@ where .unwrap_or_else(|e| e.into_inner()) .clear(); conn.release_all_workspaces(); - drop(workspace_sub); drop(machine_sub); sink.retire(); let _ = shutdown.shutdown_link(); @@ -529,18 +495,15 @@ fn handshake( }; // Advertised from what this server actually carries, not from what the - // build can do. A client that sees `workspace-store` missing knows not to + // build can do. A client that sees `machine-tree` missing knows not to // spend a round trip asking, and — the case that matters — a machine - // serving only a file tree does not claim to own workspace records it has + // serving only a file tree does not claim to own a workspace tree it has // no file for. let mut features = vec![ feature::CONTROL.to_string(), feature::HOST_RPC.to_string(), feature::STDIO_BRIDGE.to_string(), ]; - if services.workspaces.is_some() { - features.push(feature::WORKSPACE_STORE.to_string()); - } if services.machine.is_some() { features.push(feature::MACHINE_TREE.to_string()); } @@ -574,7 +537,7 @@ static NEXT_CONN: AtomicU64 = AtomicU64::new(1); /// The takeover, server side: claim `workspace` for this connection and /// tell whoever held it. /// -/// The order is the whole behaviour. The store's record moves first (so a +/// The order is the whole behaviour. The tree's record moves first (so a /// concurrent `attachment()` never shows the workspace as free), the registry's /// handles move under one lock, and only then is the displaced session told — /// outside every lock, because writing to a peer that has stopped reading must @@ -590,38 +553,34 @@ fn attach_workspace( workspace: &str, dedicated: bool, ) -> io::Result> { - // The attach verbs predate the machine tree, so the id arrives as a - // string; while the record store and the tree coexist, the attachment's - // data half lands on **whichever of the two this server carries** (both, - // on a full daemon — they describe the same workspace). A server with - // neither answers exactly what a store-less server always has. - if conn.workspaces.is_none() && conn.machine.is_none() { + // The attach verbs predate the typed tree, so the id arrives as a string. + // The data half of the attachment lives in the machine tree; a server + // without one answers the refusal a tree-less server always has. + if conn.machine.is_none() { return Err(io::Error::other( - "this server does not serve the workspace store", + "this server does not serve the machine tree", )); } let tree_id: Option = workspace.parse().ok(); let (displaced, evicted) = { - // Every table moves under one lock. Held only across the moves — + // Both tables move under one lock. Held only across the 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 attachment = Attachment::new(conn.holder.token.clone(), conn.holder.hostname.clone()); - let displaced_record = conn - .workspaces - .as_ref() - .and_then(|store| store.attach(workspace, attachment.clone())); - let displaced_tree = match (&conn.machine, tree_id) { + // A workspace the tree does not list (or an id that is not a uuid) + // records no data half; the registry's live handles still move, so + // the takeover behaviour is identical either way, and the tree's + // record appears the moment the workspace does. + let displaced = match (&conn.machine, tree_id) { (Some(machine), Some(id)) => machine.attach(id, attachment), _ => None, }; let evicted = conn .attachments .claim(workspace, conn.id, &conn.holder, dedicated); - // On a server carrying both, the two answers name the same session; - // the record store's wins only in the sense that it is asked first. - (displaced_record.or(displaced_tree), evicted) + (displaced, evicted) }; if let Some(evicted) = evicted { @@ -653,26 +612,22 @@ fn attach_workspace( /// Release `workspace` if this connection still holds it. /// -/// Token-checked in the store *and* connection-checked in the registry, which +/// Token-checked in the tree *and* connection-checked in the registry, which /// are the same guard seen from both halves: a session that was preempted and /// then tidied up must not evict the client that took over from it. fn detach_workspace(conn: &Arc, workspace: &str) -> io::Result { - if conn.workspaces.is_none() && conn.machine.is_none() { + if conn.machine.is_none() { return Err(io::Error::other( - "this server does not serve the workspace store", + "this server does not serve the machine tree", )); } let _handover = conn.attachments.handover(); let released = conn.attachments.release(workspace, conn.id); - let forgotten = conn - .workspaces - .as_ref() - .is_some_and(|store| store.detach(workspace, &conn.holder.token)); - let forgotten_tree = match (&conn.machine, workspace.parse().ok()) { + let forgotten = match (&conn.machine, workspace.parse().ok()) { (Some(machine), Some(id)) => machine.detach(id, &conn.holder.token), _ => false, }; - Ok(released || forgotten || forgotten_tree) + Ok(released || forgotten) } /// This machine's home directory, for the handshake's `home` field — the value @@ -898,50 +853,6 @@ fn run_request( (ReplyOk::Unit, Vec::new()) } - // ----- workspace store ----------------------------------------------- - // Records cross as opaque JSON: the client owns the schema, and a - // server that parsed them would drop any field it was too old to know - // about on the next write. See `core::workspace_store`. - ControlRequest::WorkspaceList => ( - ReplyOk::Json(serde_json::Value::Array(conn.workspaces()?.list())), - Vec::new(), - ), - ControlRequest::WorkspaceGet { id } => { - // `NotFound` rather than a `null` payload: "there is no such - // workspace" and "there is one and it is empty" are different - // answers, and a client that conflated them would helpfully - // overwrite a record it failed to read. - let record = conn.workspaces()?.get(&id).ok_or_else(|| { - io::Error::new( - io::ErrorKind::NotFound, - format!("no workspace {id} on this machine"), - ) - })?; - (ReplyOk::Json(record), Vec::new()) - } - ControlRequest::WorkspacePut { id, json } => { - conn.workspaces()?.put(&id, json, conn.workspace_origin)?; - (ReplyOk::Unit, Vec::new()) - } - ControlRequest::WorkspaceDelete { id } => { - // Deleting what is not there is success — a delete that raced - // another client's delete has got what it asked for. - 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()) - } - // ----- attachment (D8) ----------------------------------- ControlRequest::WorkspaceAttach { id } => ( ReplyOk::Attached { @@ -986,8 +897,8 @@ fn run_request( ControlRequest::WorkspaceRemove { workspace } => { let store = conn.machine()?; let panes = { - // Same discipline as `WorkspaceDelete` above: the attach - // registry forgets the workspace under the handover lock, or a + // The tree drops its own attachment with the workspace; the + // attach registry forgets it under the handover lock, or a // stale dedicated entry would one day close an innocent link. let _handover = conn.attachments.handover(); let panes = store.workspace_delete(workspace, conn.machine_origin)?; @@ -1147,16 +1058,10 @@ struct Conn { deferred_watches: Mutex>)>>, next_watch: AtomicU64, pool: Pool, - /// The machine's workspace records, when this server serves them. - workspaces: Option>, - /// This connection's subscriber id, so its own writes do not come back to - /// it as `WorkspaceChanged` pushes. `None` when there is no store. - workspace_origin: Option, /// The machine's workspace tree, when this server serves it. machine: Option>, - /// This connection's tree-subscriber id — the same origin-exclusion role - /// `workspace_origin` plays for the record store, so a tree operation's - /// own `Layout` delta never comes back to its writer. + /// This connection's tree-subscriber id — origin exclusion, so a tree + /// operation's own `Layout` delta never comes back to its writer. machine_origin: Option, /// Shared with every other connection this server accepts — see /// [`AttachRegistry`]. @@ -1169,17 +1074,6 @@ struct Conn { } impl Conn { - /// The workspace store, or the error a server without one answers. - /// - /// The message is deliberately the one the unimplemented slots gave before - /// M5: a client talking to a file-tree-only server must get the same answer - /// whether that server predates the store or simply was not given one. - fn workspaces(&self) -> io::Result<&Arc> { - self.workspaces - .as_ref() - .ok_or_else(|| io::Error::other("this server does not serve the workspace store")) - } - /// The machine tree, or the refusal a server not carrying one answers. /// The client's cue is the `machine-tree` capability bit; this is the /// answer for one that asked anyway. @@ -1193,14 +1087,11 @@ impl Conn { /// /// Connection-scoped, so a workspace that was taken over from this session /// 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. + /// case the tree'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); for workspace in released { - if let Some(store) = self.workspaces.as_ref() { - store.detach(&workspace, &self.holder.token); - } if let (Some(machine), Some(id)) = (&self.machine, workspace.parse().ok()) { machine.detach(id, &self.holder.token); } @@ -1259,7 +1150,7 @@ impl Conn { ControlServerMsg::Response { req_id, reply } }; // Encoded before anything is written, so a reply this server cannot put - // on the wire — a `SearchHit` whose path is not UTF-8, a `WorkspaceList` + // on the wire — a `SearchHit` whose path is not UTF-8, a `MachineGet` // grown past `MAX_FRAME` — becomes an error the client *receives*. // Dropping it instead leaves the client waiting out the request's whole // deadline (20s for a search, and again on the next keystroke) for a @@ -1361,33 +1252,14 @@ impl Conn { } } -/// Subscribe this connection to the workspace store's changes, if there is one. -/// -/// Two hops rather than one, and the split is the point. The store's callback -/// runs on the thread of *whichever connection made the change*, so it does -/// nothing but enqueue; the forwarder thread is what actually writes, and a -/// peer that has stopped reading stalls only its own forwarder. Calling -/// `Sink::send` straight from the callback would have one wedged client hold up -/// every other client's `WorkspacePut`. -fn subscribe_workspaces(services: &Services, sink: &Arc) -> Option { - let store = services.workspaces.as_ref()?; - let (tx, rx) = smol::channel::bounded::(WORKSPACE_EVENT_QUEUE); - let subscription = store.subscribe(Arc::new(move |id: &str| { - // Never blocks. A full queue means this peer is already behind on a - // signal that only says "refetch", and the notice sitting in the queue - // says it just as well. - let _ = tx.try_send(id.to_string()); - })); - spawn_workspace_forwarder(rx, Arc::clone(sink)); - Some(subscription) -} - /// Subscribe this connection to the machine tree's deltas, if there is one. /// -/// The same two-hop shape as [`subscribe_workspaces`], for the same reason: -/// the store's callback runs on the writing connection's thread, so it only -/// enqueues, and a peer that has stopped reading stalls nothing but its own -/// forwarder. The queue-full case is documented on [`LAYOUT_EVENT_QUEUE`]. +/// Two hops rather than one, and the split is the point. The store's callback +/// runs on the thread of *whichever connection made the change*, so it only +/// enqueues; the forwarder thread is what actually writes, and a peer that has +/// stopped reading stalls nothing but its own forwarder. Calling `Sink::send` +/// straight from the callback would have one wedged client hold up every other +/// client's edit. The queue-full case is documented on [`LAYOUT_EVENT_QUEUE`]. fn subscribe_machine(services: &Services, sink: &Arc) -> Option { let store = services.machine.as_ref()?; let (tx, rx) = smol::channel::bounded::<(String, machine::LayoutDelta)>(LAYOUT_EVENT_QUEUE); @@ -1402,8 +1274,11 @@ fn subscribe_machine(services: &Services, sink: &Arc) -> Option, sink: Arc, @@ -1423,27 +1298,6 @@ fn spawn_layout_forwarder( } } -/// Relay workspace changes to the peer as `WorkspaceChanged` pushes. -/// -/// Ends on its own when the `Subscription` is dropped: that removes the closure -/// holding the sender, which closes the channel. Same shape, and the same -/// reason, as [`spawn_watch_forwarder`]. -fn spawn_workspace_forwarder(rx: smol::channel::Receiver, sink: Arc) { - let spawned = std::thread::Builder::new() - .name("tty7-control-workspace".into()) - .spawn(move || { - while let Ok(id) = rx.recv_blocking() { - let event = ControlEvent::WorkspaceChanged { id }; - if sink.send(&ControlServerMsg::Event(event)).is_err() { - return; - } - } - }); - if let Err(e) = spawned { - log::warn!("could not start the workspace-change forwarder: {e}"); - } -} - /// Relay one subscription's batches to the peer as `CONTROL_EVENT` pushes. /// /// The host has already coalesced and deduplicated within its window, so this @@ -1873,10 +1727,9 @@ mod sock { /// [`serve_listener`], with the extra services every connection gets. /// - /// One [`WorkspaceStore`](crate::core::workspace_store::WorkspaceStore) - /// shared by every connection, which is what makes a change on one visible - /// to the others: two stores over one file would each believe their own - /// copy and the last save would win silently. + /// One [`MachineStore`] shared by every connection, which is what makes a + /// change on one visible to the others: two stores over one file would + /// each believe their own copy and the last save would win silently. pub fn serve_listener_with(listener: UnixListener, host: SharedHost, services: Services) { for stream in listener.incoming() { match stream { @@ -2205,10 +2058,27 @@ mod tests { } } + /// Services carrying a fresh machine tree — the shape every attach and + /// takeover test runs against, because the tree is where the attachment's + /// data half lives. fn workspace_services() -> (Services, tempfile::TempDir) { let dir = tempfile::TempDir::new().unwrap(); - let store = WorkspaceStore::open(dir.path().join("workspaces.json")); - (Services::with_workspaces(store), dir) + let store = MachineStore::open(dir.path().join(machine::MACHINE_FILE)); + (Services::with_machine(store), dir) + } + + /// A workspace created in `services`' tree, as the string id the attach + /// verbs carry. The tree only records an attachment for a workspace it + /// lists, so the takeover tests attach to a real one. + fn tree_workspace(services: &Services) -> String { + services + .machine + .as_ref() + .expect("workspace_services always carries a tree") + .workspace_create(None, None, None) + .expect("an empty tree accepts a workspace") + .id + .to_string() } // ----------------------------------------------------------------------- @@ -2689,18 +2559,17 @@ mod tests { } } - /// The workspace store's request slots exist on the wire (so M5 is additive) - /// but this server does not serve them, and says so instead of answering - /// with something that looks like an empty store. + /// A server not carrying the machine tree says so instead of answering + /// with something that looks like an empty machine. #[test] - fn the_workspace_store_is_declined_not_faked() { + fn the_machine_tree_is_declined_not_faked() { let p = pair(); let err = p .host .client() - .call(ControlRequest::WorkspaceList) + .call(ControlRequest::MachineGet) .unwrap_err(); - assert!(err.to_string().contains("workspace store"), "{err}"); + assert!(err.to_string().contains("machine tree"), "{err}"); } // ----------------------------------------------------------------------- @@ -3135,29 +3004,11 @@ mod tests { } // ----------------------------------------------------------------------- - // The workspace store + // Raw-wire helpers // ----------------------------------------------------------------------- - /// A store on a temp file, plus the directory keeping it alive. - fn temp_store() -> (Arc, tempfile::TempDir) { - let dir = tempfile::TempDir::new().unwrap(); - let store = WorkspaceStore::open(dir.path().join("workspaces.json")); - (store, dir) - } - - fn ws_record(id: &str, name: &str) -> serde_json::Value { - serde_json::json!({ - "id": id, - "name": name, - "session": {"active": 0, "tabs": [ - {"pane": {"Leaf": {"cwd": "/home/me/proj", "pane_id": 3}}} - ]}, - "last_active": 1_753_600_000u64, - }) - } - /// Issue one request and return its reply, ignoring any pushes that arrive - /// first — a `WorkspaceChanged` from another connection can legitimately + /// first — a `Layout` delta from another connection can legitimately /// interleave with this one's reply. fn ask(client: &mut UnixStream, req_id: u64, req: ControlRequest) -> ControlReply { ControlClientMsg::Request { req_id, req } @@ -3175,268 +3026,6 @@ mod tests { } } - fn ok_json(reply: ControlReply) -> serde_json::Value { - match reply { - ControlReply::Ok(ReplyOk::Json(v)) => v, - other => panic!("expected a Json reply, got {other:?}"), - } - } - - /// A server with no store answers the four slots the way it always has, and - /// says so in the handshake so a client need not ask to find out. - #[test] - fn a_server_without_a_store_advertises_nothing_and_refuses_politely() { - let (mut client, peer) = raw(); - assert!(!peer.has_feature(feature::WORKSPACE_STORE)); - assert!(peer.has_feature(feature::HOST_RPC)); - - match ask(&mut client, 1, ControlRequest::WorkspaceList) { - ControlReply::Err(e) => { - assert_eq!(e.kind, WireErrorKind::Other); - assert!( - e.msg.contains("does not serve the workspace store"), - "{e:?}" - ); - } - other => panic!("expected an error, got {other:?}"), - } - // And it is still a perfectly good file server afterwards: an - // unsupported request must not poison the connection. - assert!(matches!( - ask(&mut client, 2, ControlRequest::Ping), - ControlReply::Ok(ReplyOk::Pong) - )); - } - - /// The four RPCs, end to end over the wire, against a real file. - #[test] - fn the_four_workspace_rpcs_round_trip_over_the_wire() { - let (store, dir) = temp_store(); - let (mut client, peer) = raw_with(Services::with_workspaces(Arc::clone(&store))); - assert!(peer.has_feature(feature::WORKSPACE_STORE)); - - // Empty to begin with. - assert_eq!( - ok_json(ask(&mut client, 1, ControlRequest::WorkspaceList)), - serde_json::json!([]) - ); - - // Put two. - for (i, (id, name)) in [("w-a", "api"), ("w-b", "web")].iter().enumerate() { - assert!(matches!( - ask( - &mut client, - 10 + i as u64, - ControlRequest::WorkspacePut { - id: (*id).to_string(), - json: ws_record(id, name), - }, - ), - ControlReply::Ok(ReplyOk::Unit) - )); - } - - // Get one back, exactly as it was written. - let got = ok_json(ask( - &mut client, - 20, - ControlRequest::WorkspaceGet { - id: "w-a".to_string(), - }, - )); - assert_eq!(got, ws_record("w-a", "api")); - - // List answers an array in file order. - let listed = ok_json(ask(&mut client, 21, ControlRequest::WorkspaceList)); - let ids: Vec<&str> = listed - .as_array() - .unwrap() - .iter() - .map(|v| v["id"].as_str().unwrap()) - .collect(); - assert_eq!(ids, vec!["w-a", "w-b"]); - - // A missing id is `NotFound`, not an empty payload. - match ask( - &mut client, - 22, - ControlRequest::WorkspaceGet { - id: "nope".to_string(), - }, - ) { - ControlReply::Err(e) => assert_eq!(e.kind, WireErrorKind::NotFound), - other => panic!("expected NotFound, got {other:?}"), - } - - // A record whose id disagrees with its key is refused. - match ask( - &mut client, - 23, - ControlRequest::WorkspacePut { - id: "w-a".to_string(), - json: ws_record("w-b", "confused"), - }, - ) { - ControlReply::Err(e) => assert_eq!(e.kind, WireErrorKind::InvalidInput), - other => panic!("expected InvalidInput, got {other:?}"), - } - - // Delete, twice — the second is still success. - for req_id in [30, 31] { - assert!(matches!( - ask( - &mut client, - req_id, - ControlRequest::WorkspaceDelete { - id: "w-a".to_string(), - }, - ), - ControlReply::Ok(ReplyOk::Unit) - )); - } - - // The file on the server's disk is the authority, and it agrees. - let text = std::fs::read_to_string(dir.path().join("workspaces.json")).unwrap(); - assert!(text.contains("w-b"), "{text}"); - assert!(!text.contains("w-a"), "{text}"); - assert_eq!(store.len(), 1); - } - - /// **What the event exists for.** Two clients on one machine: a change made - /// by one has to reach the other, and must not come back to its author as - /// news it already has. - #[test] - fn a_change_reaches_the_other_client_and_not_its_author() { - let (store, _dir) = temp_store(); - let services = Services::with_workspaces(Arc::clone(&store)); - let (mut writer, _) = raw_with(services.clone()); - let (mut listener, _) = raw_with(services); - - assert!(matches!( - ask( - &mut writer, - 1, - ControlRequest::WorkspacePut { - id: "w".to_string(), - json: ws_record("w", "api"), - }, - ), - ControlReply::Ok(ReplyOk::Unit) - )); - - // The listener is told which workspace to refetch. - listener - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); - match ControlServerMsg::read(&mut listener).unwrap() { - ControlServerMsg::Event(ControlEvent::WorkspaceChanged { id }) => { - assert_eq!(id, "w"); - } - other => panic!("expected a WorkspaceChanged push, got {other:?}"), - } - - // A delete is a change too. - assert!(matches!( - ask( - &mut writer, - 2, - ControlRequest::WorkspaceDelete { - id: "w".to_string(), - }, - ), - ControlReply::Ok(ReplyOk::Unit) - )); - match ControlServerMsg::read(&mut listener).unwrap() { - ControlServerMsg::Event(ControlEvent::WorkspaceChanged { id }) => assert_eq!(id, "w"), - other => panic!("expected a WorkspaceChanged push, got {other:?}"), - } - - // The author heard nothing about either of its own writes: its next - // frame is the reply to a fresh request, not a backlog of echoes. - writer - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); - ControlClientMsg::Request { - req_id: 3, - req: ControlRequest::Ping, - } - .encode(&mut writer) - .unwrap(); - writer.flush().unwrap(); - match ControlServerMsg::read(&mut writer).unwrap() { - ControlServerMsg::Response { req_id: 3, reply } => { - assert!(matches!(reply, ControlReply::Ok(ReplyOk::Pong))); - } - other => panic!("the author was pushed its own change: {other:?}"), - } - } - - /// A subscription is a connection's resource like any other: when the - /// connection ends, the store must stop holding a callback into its sink. - #[test] - fn a_closed_connection_stops_being_a_subscriber() { - let (store, _dir) = temp_store(); - let (server, client) = UnixStream::pair().unwrap(); - let served = { - let store = Arc::clone(&store); - std::thread::spawn(move || { - let _ = serve_with(server, LocalHost::new(), Services::with_workspaces(store)); - }) - }; - // Handshake, then hang up. - let mut client = client; - ControlClientMsg::Hello(ControlHello::host_rpc("t", "h")) - .encode(&mut client) - .unwrap(); - client.flush().unwrap(); - let _ = ControlServerMsg::read(&mut client).unwrap(); - drop(client); - served.join().unwrap(); - - // The store still works, and writing to it does not try to reach a sink - // that is gone. (A leaked subscriber would show up as a `BrokenPipe` - // log rather than a failure, so the assertion is that the put succeeds - // and the record lands.) - store.put("w", ws_record("w", "api"), None).unwrap(); - assert_eq!(store.len(), 1); - } - - /// A store shared by many connections writing at once: the server has to be - /// as safe as the store is, and no request may be lost or answered twice. - #[test] - fn concurrent_connections_can_all_write_the_store() { - let (store, _dir) = temp_store(); - let services = Services::with_workspaces(Arc::clone(&store)); - - let writers: Vec<_> = (0..6) - .map(|c| { - let services = services.clone(); - std::thread::spawn(move || { - let (mut client, _) = raw_with(services); - for i in 0..10 { - let id = format!("c{c}-{i}"); - let reply = ask( - &mut client, - i as u64 + 1, - ControlRequest::WorkspacePut { - id: id.clone(), - json: ws_record(&id, "x"), - }, - ); - assert!( - matches!(reply, ControlReply::Ok(ReplyOk::Unit)), - "{reply:?}" - ); - } - }) - }) - .collect(); - for w in writers { - w.join().unwrap(); - } - assert_eq!(store.len(), 60); - } - // ----------------------------------------------------------------------- // Attachment and takeover (D8) // ----------------------------------------------------------------------- @@ -3452,13 +3041,14 @@ mod tests { fn a_second_client_takes_the_workspace_and_the_first_is_told() { let (services, _dir) = workspace_services(); let registry = Arc::clone(&services.attachments); + let w = tree_workspace(&services); let ((mut laptop, _), _laptop_served) = - raw_hello(services.clone(), hello_for("w", "tok-laptop", "laptop")); - await_holder(®istry, "w", "laptop"); + raw_hello(services.clone(), hello_for(&w, "tok-laptop", "laptop")); + await_holder(®istry, &w, "laptop"); let ((mut desktop, _), _desktop_served) = - raw_hello(services.clone(), hello_for("w", "tok-desktop", "desktop")); + raw_hello(services.clone(), hello_for(&w, "tok-desktop", "desktop")); // The displaced session hears who took it, and which workspace: one // connection can carry several, so a push without the id would leave the @@ -3466,21 +3056,21 @@ mod tests { assert_eq!( await_preempted(&mut laptop), Some(ControlEvent::Preempted { - workspace: "w".to_string(), + workspace: w.clone(), by: "desktop".to_string(), }) ); - assert_eq!(registry.holder("w").map(|(_, h)| h), Some("desktop".into())); + assert_eq!(registry.holder(&w).map(|(_, h)| h), Some("desktop".into())); assert_eq!( services - .workspaces + .machine .as_ref() .unwrap() - .attachment("w") + .attachment(w.parse().unwrap()) .unwrap() .hostname, "desktop", - "the store's record moves with the live handles" + "the tree's record moves with the live handles" ); // And the newcomer is told what it took over from — a takeover the new @@ -3488,7 +3078,7 @@ mod tests { let (reply, _) = round_trip( &mut desktop, 1, - ControlRequest::WorkspaceAttach { id: "w".into() }, + ControlRequest::WorkspaceAttach { id: w.clone() }, ); assert_eq!( reply, @@ -3504,11 +3094,12 @@ mod tests { #[test] fn a_dedicated_connection_is_closed_when_its_workspace_is_taken() { let (services, _dir) = workspace_services(); + let w = tree_workspace(&services); let ((mut laptop, _), _l) = - raw_hello(services.clone(), hello_for("w", "tok-laptop", "laptop")); - await_holder(&services.attachments, "w", "laptop"); + raw_hello(services.clone(), hello_for(&w, "tok-laptop", "laptop")); + await_holder(&services.attachments, &w, "laptop"); let ((_desktop, _), _d) = - raw_hello(services.clone(), hello_for("w", "tok-desktop", "desktop")); + raw_hello(services.clone(), hello_for(&w, "tok-desktop", "desktop")); assert!(await_preempted(&mut laptop).is_some()); // The push comes first and the close after: the notice is useless if it @@ -3521,59 +3112,51 @@ mod tests { ); } - /// Deleting a workspace clears it from *both* tables. + /// Removing 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. + /// The tree drops its own attachment with the workspace. 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() { + fn removing_a_workspace_clears_both_attachment_tables() { let (services, _dir) = workspace_services(); let registry = Arc::clone(&services.attachments); - let store = services.workspaces.clone().unwrap(); + let machine = services.machine.clone().unwrap(); + let w = tree_workspace(&services); + let id: crate::core::session::WorkspaceId = w.parse().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()); + raw_hello(services.clone(), hello_for(&w, "tok-laptop", "laptop")); + await_holder(®istry, &w, "laptop"); + assert!(machine.attachment(id).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(), - }, + 1, + ControlRequest::WorkspaceRemove { workspace: id }, ); assert!( - matches!(reply, ControlReply::Ok(ReplyOk::Unit)), + matches!(reply, ControlReply::Ok(ReplyOk::Panes(_))), "{reply:?}" ); assert!( - store.attachment("w").is_none(), - "the store still names a holder for a workspace that is gone" + machine.attachment(id).is_none(), + "the tree still names a holder for a workspace that is gone" ); assert!( - registry.holder("w").is_none(), + 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. + /// The tree'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 + /// same instant can each win a different one, after which the tree 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. @@ -3586,28 +3169,30 @@ mod tests { 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 machine = services.machine.clone().unwrap(); + let w = tree_workspace(&services); + let id: crate::core::session::WorkspaceId = w.parse().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")); + raw_hello(services.clone(), hello_for(&w, "tok-laptop", "laptop")); std::thread::sleep(Duration::from_millis(150)); assert!( - registry.holder("w").is_none(), + 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" + machine.attachment(id).is_none(), + "the tree was moved while a handover was in flight" ); drop(held); - await_holder(®istry, "w", "laptop"); + await_holder(®istry, &w, "laptop"); assert_eq!( - store.attachment("w").map(|a| a.token).as_deref(), + machine.attachment(id).map(|a| a.token).as_deref(), Some("tok-laptop"), "both tables have to name the same session once the handover is done" ); @@ -3620,16 +3205,18 @@ mod tests { fn a_shared_connection_survives_losing_one_of_its_workspaces() { let (services, _dir) = workspace_services(); let registry = Arc::clone(&services.attachments); + let w1 = tree_workspace(&services); + let w2 = tree_workspace(&services); let ((mut laptop, _), _l) = raw_hello( services.clone(), ControlHello::host_rpc("tok-laptop", "laptop"), ); - for (i, id) in ["w1", "w2"].iter().enumerate() { + for (i, id) in [&w1, &w2].iter().enumerate() { let (reply, _) = round_trip( &mut laptop, i as u64 + 1, - ControlRequest::WorkspaceAttach { id: (*id).into() }, + ControlRequest::WorkspaceAttach { id: (*id).clone() }, ); assert_eq!( reply, @@ -3641,11 +3228,11 @@ mod tests { assert_eq!(registry.len(), 2); let ((_desktop, _), _d) = - raw_hello(services.clone(), hello_for("w1", "tok-desktop", "desktop")); + raw_hello(services.clone(), hello_for(&w1, "tok-desktop", "desktop")); assert_eq!( await_preempted(&mut laptop), Some(ControlEvent::Preempted { - workspace: "w1".to_string(), + workspace: w1.clone(), by: "desktop".to_string(), }) ); @@ -3654,11 +3241,11 @@ mod tests { let (reply, _) = round_trip(&mut laptop, 9, ControlRequest::Ping); assert_eq!(reply, ControlReply::Ok(ReplyOk::Pong)); assert_eq!( - registry.holder("w2").map(|(t, _)| t), + registry.holder(&w2).map(|(t, _)| t), Some("tok-laptop".into()) ); assert_eq!( - registry.holder("w1").map(|(t, _)| t), + registry.holder(&w1).map(|(t, _)| t), Some("tok-desktop".into()) ); } @@ -3670,7 +3257,9 @@ mod tests { fn a_displaced_session_tidying_up_does_not_evict_the_new_owner() { let (services, _dir) = workspace_services(); let registry = Arc::clone(&services.attachments); - let store = Arc::clone(services.workspaces.as_ref().unwrap()); + let machine = Arc::clone(services.machine.as_ref().unwrap()); + let w = tree_workspace(&services); + let other = tree_workspace(&services); let ((mut laptop, _), _l) = raw_hello( services.clone(), @@ -3678,30 +3267,33 @@ mod tests { ); // Two workspaces on one link, which is what a client with two windows // on one machine has — and what keeps this link up once `w` is taken. - for (i, id) in ["w", "other"].iter().enumerate() { + for (i, id) in [&w, &other].iter().enumerate() { round_trip( &mut laptop, i as u64 + 1, - ControlRequest::WorkspaceAttach { id: (*id).into() }, + ControlRequest::WorkspaceAttach { id: (*id).clone() }, ); } let ((_desktop, _), _d) = - raw_hello(services.clone(), hello_for("w", "tok-desktop", "desktop")); + raw_hello(services.clone(), hello_for(&w, "tok-desktop", "desktop")); assert!(await_preempted(&mut laptop).is_some()); // The laptop, which no longer holds anything, tidies up. let (reply, _) = round_trip( &mut laptop, 3, - ControlRequest::WorkspaceDetach { id: "w".into() }, + ControlRequest::WorkspaceDetach { id: w.clone() }, ); assert_eq!(reply, ControlReply::Ok(ReplyOk::Unit)); assert_eq!( - registry.holder("w").map(|(t, _)| t), + registry.holder(&w).map(|(t, _)| t), Some("tok-desktop".into()), "the displaced session must not release what it no longer holds" ); - assert_eq!(store.attachment("w").unwrap().token, "tok-desktop"); + assert_eq!( + machine.attachment(w.parse().unwrap()).unwrap().token, + "tok-desktop" + ); } /// A connection ending gives its workspaces back, so the next client does @@ -3710,22 +3302,22 @@ mod tests { fn a_closed_connection_releases_what_it_held() { let (services, _dir) = workspace_services(); let registry = Arc::clone(&services.attachments); - let store = Arc::clone(services.workspaces.as_ref().unwrap()); + let machine = Arc::clone(services.machine.as_ref().unwrap()); + let w = tree_workspace(&services); { - let ((client, _), served) = - raw_hello(services.clone(), hello_for("w", "tok", "laptop")); - await_holder(®istry, "w", "laptop"); + let ((client, _), served) = raw_hello(services.clone(), hello_for(&w, "tok", "laptop")); + await_holder(®istry, &w, "laptop"); drop(client); served.join().unwrap(); } assert!(registry.is_empty(), "the registry outlived the connection"); - assert_eq!(store.attachment("w"), None); + assert_eq!(machine.attachment(w.parse().unwrap()), None); } - /// A server with no workspace store has no workspaces to attach to, and + /// A server with no machine tree has no workspaces to attach to, and /// says so rather than pretending the claim succeeded. #[test] - fn attaching_to_a_server_without_a_store_is_an_error() { + fn attaching_to_a_server_without_a_tree_is_an_error() { let (mut client, _) = raw_with(Services::none()); let (reply, _) = round_trip( &mut client, diff --git a/crates/tty7-server/Cargo.toml b/crates/tty7-server/Cargo.toml index 11f2fa13..d675dd0a 100644 --- a/crates/tty7-server/Cargo.toml +++ b/crates/tty7-server/Cargo.toml @@ -26,10 +26,5 @@ tty7-core = { path = "../tty7-core" } # Sandboxes for the suite: an empty directory per case, removed on drop. The # server is on this machine, so a local temp dir is a path in its namespace. tempfile = "3" -# Workspace records cross the control wire as opaque JSON, so -# `tests/workspace_store.rs` has to build and read one. Dev-only: the binary -# itself still depends on nothing but `tty7-core`. -serde_json.workspace = true - [lints] workspace = true diff --git a/crates/tty7-server/src/main.rs b/crates/tty7-server/src/main.rs index 8aebdaed..f9a729ad 100644 --- a/crates/tty7-server/src/main.rs +++ b/crates/tty7-server/src/main.rs @@ -218,7 +218,7 @@ fn run_stdio(args: &[String]) -> io::Result<()> { // 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 + // their own `MachineStore` 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 diff --git a/crates/tty7-server/tests/machine_tree.rs b/crates/tty7-server/tests/machine_tree.rs index ea30d9da..91b091e0 100644 --- a/crates/tty7-server/tests/machine_tree.rs +++ b/crates/tty7-server/tests/machine_tree.rs @@ -1,10 +1,9 @@ //! The machine-owned workspace tree, end to end against a real `tty7-server` //! child process. //! -//! Same shape and the same reasoning as `workspace_store.rs`, one architecture -//! over: the client is the shipped `ControlClient`, the wire is the control -//! dialect over real pipes, and the server is the shipped binary owning its -//! tree in a file. What the process boundary buys here specifically: +//! The client is the shipped `ControlClient`, the wire is the control dialect +//! over real pipes, and the server is the shipped binary owning its tree in a +//! file. What the process boundary buys here specifically: //! //! | | Why an in-process store would not do | //! |---|---| @@ -330,7 +329,7 @@ fn an_operation_from_one_client_reaches_the_other_as_a_delta() { server::serve_listener_with( listener, LocalHost::new(), - server::Services::none().and_machine(machine), + server::Services::with_machine(machine), ) }); } @@ -427,7 +426,7 @@ fn attachment_rides_the_tree_when_no_record_store_is_served() { server::serve_listener_with( listener, LocalHost::new(), - server::Services::none().and_machine(machine), + server::Services::with_machine(machine), ) }); } diff --git a/crates/tty7-server/tests/stdio_conformance.rs b/crates/tty7-server/tests/stdio_conformance.rs index 2dc774d0..622826ee 100644 --- a/crates/tty7-server/tests/stdio_conformance.rs +++ b/crates/tty7-server/tests/stdio_conformance.rs @@ -93,7 +93,7 @@ fn stdio_host() -> (SharedHost, TempSandbox) { // bridge to *that* would be testing their machine's state instead of // this build. .args(["--stdio", "--serve"]) - // The server opens its workspace store at startup. None of these cases + // The server opens its machine tree at startup. None of these cases // touch it, but pointing it at the sandbox keeps forty-six child // processes off the developer's real `~/.local/share/tty7`. .env("TTY7_DATA_DIR", sandbox.path()) diff --git a/crates/tty7-server/tests/workspace_store.rs b/crates/tty7-server/tests/workspace_store.rs deleted file mode 100644 index df750aa1..00000000 --- a/crates/tty7-server/tests/workspace_store.rs +++ /dev/null @@ -1,543 +0,0 @@ -//! The workspace store, end to end against a real `tty7-server` child process. -//! -//! Same shape and the same reasoning as [`stdio_conformance`]: the client is -//! the shipped `ControlClient`, the wire is the control dialect over real -//! pipes, and the server is the shipped binary keeping its records in a file it -//! owns. What this file adds is the half the conformance suite cannot reach — -//! the store is not a `Host` method, so no amount of `read_dir` parity proves -//! that `WorkspacePut` reached a disk or that another client heard about it. -//! -//! The three things worth a process boundary: -//! -//! | | Why an in-process socket pair would not do | -//! |---|---| -//! | The record is on **the server's** disk | The whole storage split is "the machine is the authority". A store in the test's own address space proves nothing about that | -//! | `workspace-store` is advertised only when served | The capability bit is built from what the *binary* wires up, and that wiring lives in `main.rs` | -//! | A change reaches the **other** connection | Two clients, one server process, one file — the configuration the user actually has when their laptop and their desktop are both connected | -//! -//! Every case gets its own `$TTY7_DATA_DIR`, so no case can be explained by -//! another's leftovers and nothing here can touch the developer's real -//! `~/.local/share/tty7/workspaces.json`. - -// Unix-only, for the same reason as `stdio_conformance.rs`: the server under -// test is a `--stdio` child, and two of the cases stand up a control socket. -#![cfg(unix)] - -use std::io; -use std::path::{Path, PathBuf}; -use std::process::{Child, Command, Stdio}; -use std::sync::{Arc, Mutex}; -use std::time::{Duration, Instant}; - -use tty7_core::core::workspace_store::{STORE_FILE, WorkspaceStore}; -use tty7_core::daemon::control::{ - ControlClient, ControlEvent, ControlHello, ControlRequest, LinkShutdown, ReplyOk, feature, -}; -use tty7_core::host::local::LocalHost; -use tty7_core::host::server; - -/// The child, and the only way to end it — see `stdio_conformance.rs` for why a -/// `LinkShutdown` is what reaps a process-backed link. -struct ServerProcess { - child: Mutex>, -} - -impl LinkShutdown for ServerProcess { - fn shutdown_link(&self) -> io::Result<()> { - let Some(mut child) = self.child.lock().unwrap_or_else(|e| e.into_inner()).take() else { - return Ok(()); - }; - let _ = child.kill(); - let _ = child.wait(); - Ok(()) - } -} - -/// One connected client: the RPC channel, plus everything the server pushed to -/// it. -struct Client { - control: ControlClient, - events: Arc>>, - peer_features: Vec, -} - -impl Client { - /// Wait for a `WorkspaceChanged` naming `id`, or fail saying what did - /// arrive. Polled rather than blocked on a channel because the event and - /// the reply that caused it race by construction. - fn expect_changed(&self, id: &str) { - let deadline = Instant::now() + Duration::from_secs(10); - loop { - let seen = self - .events - .lock() - .unwrap_or_else(|e| e.into_inner()) - .clone(); - if seen - .iter() - .any(|e| matches!(e, ControlEvent::WorkspaceChanged { id: got } if got == id)) - { - return; - } - assert!( - Instant::now() < deadline, - "no WorkspaceChanged for {id}; saw {seen:?}" - ); - std::thread::sleep(Duration::from_millis(20)); - } - } - - /// Wait for the takeover notice naming `workspace` and `by`. - fn expect_preempted(&self, workspace: &str, by: &str) { - let deadline = Instant::now() + Duration::from_secs(10); - loop { - let seen = self - .events - .lock() - .unwrap_or_else(|e| e.into_inner()) - .clone(); - if seen.iter().any(|e| { - matches!(e, ControlEvent::Preempted { workspace: w, by: b } - if w == workspace && b == by) - }) { - return; - } - assert!( - Instant::now() < deadline, - "no Preempted for {workspace} by {by}; saw {seen:?}" - ); - std::thread::sleep(Duration::from_millis(20)); - } - } - - fn changed_count(&self) -> usize { - self.events - .lock() - .unwrap_or_else(|e| e.into_inner()) - .iter() - .filter(|e| matches!(e, ControlEvent::WorkspaceChanged { .. })) - .count() - } -} - -/// Start a `tty7-server --stdio --serve` whose store lives in `data_dir`, and -/// connect a client to it. -/// -/// `--serve` rather than letting the mode be probed: a developer running these -/// tests may well have a real `tty7-server --daemon` up, and bridging to *that* -/// would be testing their machine's state — and, here, writing to their real -/// workspace file. -fn connect(data_dir: &Path, token: &str) -> Client { - let mut child = Command::new(env!("CARGO_BIN_EXE_tty7-server")) - .args(["--stdio", "--serve"]) - .env("TTY7_DATA_DIR", data_dir) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::null()) - .spawn() - .expect("could not start tty7-server --stdio"); - - let stdout = child.stdout.take().expect("piped"); - let stdin = child.stdin.take().expect("piped"); - let closer: Arc = Arc::new(ServerProcess { - child: Mutex::new(Some(child)), - }); - - let events: Arc>> = Arc::new(Mutex::new(Vec::new())); - let sink = Arc::clone(&events); - let control = ControlClient::connect_with( - stdout, - stdin, - Some(closer), - &ControlHello::host_rpc(token, "test-client"), - Box::new(move |event| sink.lock().unwrap_or_else(|e| e.into_inner()).push(event)), - ) - .expect("handshake with tty7-server --stdio"); - - let peer_features = control.hello().features.clone(); - Client { - control, - events, - peer_features, - } -} - -fn data_dir() -> tempfile::TempDir { - tempfile::TempDir::new().unwrap() -} - -fn store_file(dir: &tempfile::TempDir) -> PathBuf { - dir.path().join(STORE_FILE) -} - -fn record(id: &str, name: &str) -> serde_json::Value { - serde_json::json!({ - "id": id, - "name": name, - "session": {"active": 0, "tabs": [ - {"pane": {"Leaf": {"cwd": "/home/me/proj", "pane_id": 11}}, - "sidebar_group": "/home/me/proj"} - ]}, - "last_active": 1_753_600_000u64, - }) -} - -fn json(reply: ReplyOk) -> serde_json::Value { - match reply { - ReplyOk::Json(v) => v, - other => panic!("expected a Json reply, got {other:?}"), - } -} - -// --------------------------------------------------------------------------- - -/// The capability bit is the client's cue that asking is worth a round trip, so -/// it has to reflect what the shipped binary actually wired up. -#[test] -fn the_server_advertises_the_workspace_store() { - let dir = data_dir(); - let client = connect(dir.path(), "cap"); - assert!( - client - .peer_features - .iter() - .any(|f| f == feature::WORKSPACE_STORE), - "features were {:?}", - client.peer_features - ); -} - -/// **The milestone's proof for M5.** The four RPCs against a real server, and -/// the record ends up in a file that server owns — the storage split is not a -/// diagram, it is this file on that machine. -#[test] -fn records_survive_in_a_file_the_server_owns() { - let dir = data_dir(); - let client = connect(dir.path(), "rpc"); - - assert_eq!( - json(client.control.call(ControlRequest::WorkspaceList).unwrap()), - serde_json::json!([]) - ); - - for (id, name) in [("w-api", "api"), ("w-web", "web")] { - client - .control - .call(ControlRequest::WorkspacePut { - id: id.to_string(), - json: record(id, name), - }) - .expect("put"); - } - - // The file is on this machine only because the "remote" is this machine; - // the point is that the *test process* never wrote it. Reading it with - // plain `std::fs` is how we know the bytes went out through a pipe and came - // back as a syscall someone else made. - let text = std::fs::read_to_string(store_file(&dir)).expect("the server wrote its store"); - assert!(text.contains("w-api"), "{text}"); - assert!(text.contains("w-web"), "{text}"); - - // Get answers exactly what was put. - let got = json( - client - .control - .call(ControlRequest::WorkspaceGet { - id: "w-api".to_string(), - }) - .unwrap(), - ); - assert_eq!(got, record("w-api", "api")); - - // List answers both, in the order they were written. - let listed = json(client.control.call(ControlRequest::WorkspaceList).unwrap()); - let ids: Vec<&str> = listed - .as_array() - .unwrap() - .iter() - .map(|v| v["id"].as_str().unwrap()) - .collect(); - assert_eq!(ids, vec!["w-api", "w-web"]); - - // A missing id is an error the client can tell from an empty record. - let missing = client - .control - .call(ControlRequest::WorkspaceGet { - id: "not-a-workspace".to_string(), - }) - .unwrap_err(); - assert_eq!(missing.kind(), io::ErrorKind::NotFound); - - // Delete reaches the disk, and deleting again is still success. - for _ in 0..2 { - client - .control - .call(ControlRequest::WorkspaceDelete { - id: "w-api".to_string(), - }) - .expect("delete"); - } - let text = std::fs::read_to_string(store_file(&dir)).unwrap(); - assert!(!text.contains("w-api"), "{text}"); - assert!(text.contains("w-web"), "{text}"); -} - -/// A second connection to the same server sees the first one's records — that -/// is what "换台电脑连过来要看到同一份" means once the machine is fixed and the -/// client is not. -#[test] -fn a_later_client_sees_what_an_earlier_one_wrote() { - let dir = data_dir(); - { - let first = connect(dir.path(), "first"); - first - .control - .call(ControlRequest::WorkspacePut { - id: "w".to_string(), - json: record("w", "api"), - }) - .expect("put"); - first.control.close(); - } - - // A brand-new server process, reading the file the previous one left. - let second = connect(dir.path(), "second"); - let got = json( - second - .control - .call(ControlRequest::WorkspaceGet { - id: "w".to_string(), - }) - .unwrap(), - ); - assert_eq!(got["name"], "api"); - assert_eq!(got["session"]["tabs"][0]["pane"]["Leaf"]["pane_id"], 11); -} - -/// Two clients on one machine at once. A change by one has to reach the other, -/// and must not come back to its author. -/// -/// The store lives behind a listener, as it does under `--daemon`, and both -/// clients reach it as `--stdio --bridge` children — the same two-hop shape -/// `cli.rs` uses, and the configuration a user has when their laptop and their -/// desktop are both connected. A store per connection would pass every other -/// test in this file and fail this one. -#[test] -fn a_change_from_one_client_reaches_the_other() { - let dir = data_dir(); - let store = WorkspaceStore::open(store_file(&dir)); - let sock = dir.path().join("control.sock"); - let listener = server::bind_control_socket(&sock).unwrap(); - { - let store = Arc::clone(&store); - std::thread::spawn(move || { - server::serve_listener_with( - listener, - LocalHost::new(), - server::Services::with_workspaces(store), - ) - }); - } - - let writer = bridged(&sock, "writer"); - let watcher = bridged(&sock, "watcher"); - assert!( - writer - .peer_features - .iter() - .any(|f| f == feature::WORKSPACE_STORE) - ); - - writer - .control - .call(ControlRequest::WorkspacePut { - id: "shared".to_string(), - json: record("shared", "api"), - }) - .expect("put"); - - watcher.expect_changed("shared"); - assert_eq!( - writer.changed_count(), - 0, - "a client must not be pushed its own change" - ); - - // The watcher is looking at the same store, not at a copy. - let got = json( - watcher - .control - .call(ControlRequest::WorkspaceGet { - id: "shared".to_string(), - }) - .unwrap(), - ); - assert_eq!(got["name"], "api"); - - // A delete is a change too — and the watcher's own delete comes back to the - // writer, which is the same rule seen from the other side. - watcher - .control - .call(ControlRequest::WorkspaceDelete { - id: "shared".to_string(), - }) - .expect("delete"); - writer.expect_changed("shared"); - assert_eq!(watcher.changed_count(), 1, "still only the writer's put"); - assert_eq!(store.len(), 0); -} - -/// **The takeover, across two real processes.** -/// -/// The same two-client shape as the change-notification test, and for the same -/// reason: a takeover is by definition something one connection does to -/// *another*, so an in-process registry with two handles into it would prove -/// only that the data structure works. What has to hold is that the notice -/// crosses a pipe into a different program and that the displaced link actually -/// closes. -/// -/// D8 is the assertion in the middle: the newcomer holds the workspace -/// afterwards. Rejecting the second client would satisfy "only one at a time" -/// just as well and is the decision this test exists to rule out. -#[test] -fn a_later_client_takes_the_workspace_and_the_first_is_cut_off() { - let dir = data_dir(); - let store = WorkspaceStore::open(store_file(&dir)); - let sock = dir.path().join("control.sock"); - let listener = server::bind_control_socket(&sock).unwrap(); - { - let store = Arc::clone(&store); - std::thread::spawn(move || { - server::serve_listener_with( - listener, - LocalHost::new(), - server::Services::with_workspaces(store), - ) - }); - } - - let laptop = bridged_for(&sock, "tok-laptop", "laptop", Some("w")); - // The attach runs on the server thread after the handshake reply, so the - // record is what says it happened — not the fact that we got a `HELLO_OK`. - await_attachment(&store, "w", "laptop"); - assert!(laptop.control.call(ControlRequest::Ping).is_ok()); - - let desktop = bridged_for(&sock, "tok-desktop", "desktop", Some("w")); - - // The displaced client is told which workspace it lost and to whom. - laptop.expect_preempted("w", "desktop"); - // …and then its link is closed, because this connection existed for that - // workspace: the server closes its stream. - let deadline = Instant::now() + Duration::from_secs(10); - while laptop.control.is_connected() { - assert!( - Instant::now() < deadline, - "the displaced session's link stayed open" - ); - std::thread::sleep(Duration::from_millis(20)); - } - assert_eq!( - laptop - .control - .call(ControlRequest::Ping) - .unwrap_err() - .kind(), - io::ErrorKind::ConnectionReset - ); - - // D8: the newcomer is the one holding it, and it can still work. - await_attachment(&store, "w", "desktop"); - assert!(desktop.control.call(ControlRequest::Ping).is_ok()); - assert_eq!( - desktop.changed_count(), - 0, - "taking over is not a workspace change" - ); - - // Taking it back is the same operation in the other direction — that is all - // the [Take Back] button is. - let back = bridged_for(&sock, "tok-laptop-2", "laptop", None); - let reply = back - .control - .call(ControlRequest::WorkspaceAttach { id: "w".into() }) - .expect("attach"); - assert_eq!( - reply, - ReplyOk::Attached { - took_over_from: Some("desktop".to_string()) - } - ); - desktop.expect_preempted("w", "laptop"); - await_attachment(&store, "w", "laptop"); - - // The link that just did the taking was not opened *for* the workspace, so - // it is a plain machine connection and keeps working — that is the shape the - // GUI has, one link per machine. - assert!(back.control.is_connected()); - assert!(back.control.call(ControlRequest::Ping).is_ok()); -} - -/// Poll until `hostname` holds `workspace`, or fail saying who does. -fn await_attachment(store: &Arc, workspace: &str, hostname: &str) { - let deadline = Instant::now() + Duration::from_secs(10); - loop { - let who = store.attachment(workspace); - if who.as_ref().map(|a| a.hostname.as_str()) == Some(hostname) { - return; - } - assert!( - Instant::now() < deadline, - "{workspace} is held by {who:?}, not {hostname}" - ); - std::thread::sleep(Duration::from_millis(20)); - } -} - -/// A `--stdio --bridge` child connected to an already-listening control socket. -fn bridged(sock: &Path, token: &str) -> Client { - bridged_for(sock, token, "test-client", None) -} - -/// [`bridged`], naming the client machine and, optionally, the workspace this -/// connection is opened *for* — the hello field the takeover keys on. -fn bridged_for(sock: &Path, token: &str, hostname: &str, workspace: Option<&str>) -> Client { - let hello = ControlHello { - control_version: tty7_core::daemon::control::CONTROL_VERSION, - workspace: workspace.map(str::to_string), - client_token: token.to_string(), - client_hostname: hostname.to_string(), - }; - bridged_with(sock, hello) -} - -fn bridged_with(sock: &Path, hello: ControlHello) -> Client { - let mut child = Command::new(env!("CARGO_BIN_EXE_tty7-server")) - .args(["--stdio", "--bridge", "--control-sock"]) - .arg(sock) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::null()) - .spawn() - .expect("could not start the bridging client"); - let stdout = child.stdout.take().expect("piped"); - let stdin = child.stdin.take().expect("piped"); - let closer: Arc = Arc::new(ServerProcess { - child: Mutex::new(Some(child)), - }); - let events: Arc>> = Arc::new(Mutex::new(Vec::new())); - let sink = Arc::clone(&events); - let control = ControlClient::connect_with( - stdout, - stdin, - Some(closer), - &hello, - Box::new(move |e| sink.lock().unwrap_or_else(|e| e.into_inner()).push(e)), - ) - .expect("bridge handshake"); - let peer_features = control.hello().features.clone(); - Client { - control, - events, - peer_features, - } -} diff --git a/src/ui/remote_workspace.rs b/src/ui/remote_workspace.rs index f2489bd1..618494f3 100644 --- a/src/ui/remote_workspace.rs +++ b/src/ui/remote_workspace.rs @@ -1,7 +1,7 @@ //! The window's half of "Connect to Host". //! //! [`ui::remote_connect`](crate::ui::remote_connect) is the plumbing — SSH -//! specs, routed control connections, the remote workspace store. This is the +//! specs, routed control connections, the remote machine's tree. This is the //! part that lives on a window: the state the home page renders, the steps that //! move between those states, and the guards that keep a window on one machine. //! @@ -1440,10 +1440,6 @@ pub(crate) fn drain_events(cx: &mut gpui::App) { release_panes(cx, id); cx.refresh_windows(); } - // The retired record store's change notice. Nothing writes those - // records any more; the tree's Layout deltas below carry the same - // news with the change itself. - ControlEvent::WorkspaceChanged { .. } => {} // Another writer edited a workspace tree this client shows: apply // the delta to the mirror and the live window (or re-pull the // workspace when it will not apply cleanly). @@ -1545,7 +1541,7 @@ fn launch_attempt(cx: &mut gpui::App, host: HostId, target: RemoteTarget) { log::info!("took workspace {key} back from {who}"); } Ok(_) => {} - // A machine that has no workspace store (an older + // A machine that has no machine tree (an older // server) still serves files; the workspace is usable, // it simply cannot be claimed exclusively. Err(e) => log::warn!("could not attach to workspace {key}: {e}"), @@ -2153,14 +2149,8 @@ mod tests { assert_eq!(q.waiting(), 1); } - // ── `WorkspaceChanged` → re-read (B3's push, arriving) ─────────────────── + // ── The input gate ─────────────────────────────────────────────────────── - /// **A burst of changes costs one round trip per workspace.** - /// - /// B3's contract for this event is that it means only "read it again", so - /// losing one is safe and getting ten is safe. Collapsing them is the whole - /// of the logic that rule buys — and the thing that keeps a client with a - /// chatty peer from opening a `WorkspaceGet` per keystroke of theirs. #[test] fn every_state_says_what_it_means_for_the_keyboard() { let cases = [