Live documents

AGENTS.md

@@ -29,14 +29,15 @@ crates/
   proto     craft-cloud-proto     wire types: names, IDs, hashes, manifests, commit ops, API bodies, limits, join codes and join strings (`join.rs`)
   chunker   craft-cloud-chunker   FastCDC `fastcdc-v1` + BLAKE3, frozen by golden vectors
   store     craft-cloud-store     blob store: disk, S3, memory; checks every blob in and out
-  meta      craft-cloud-meta      metadata on SQLite or Postgres; the batch commit; the change feed; invites and pairings (`join.rs`)
+  meta      craft-cloud-meta      metadata on SQLite or Postgres; the batch commit; the change feed; invites and pairings (`join.rs`); live documents' logs and snapshots (`live.rs`)
   sync      craft-cloud-sync      the sync engine over `Fs`/`Transport` traits; `MemFs`; scenario tests and the simulation
-  app       craft-cloud-app       what apps link: a blocking client of the agent's local API (status, sign-in, joining, computers, pause, autostart, recents)
+  app       craft-cloud-app       what apps link: a blocking client of the agent's local API (status, sign-in, joining, computers, pause, autostart, recents, live documents at the level of bytes: `live.rs`)
 apps/
-  craft-cloud-server              HTTP API (`/api/v1`), the live service (`/api/v1/live`, `live.rs`), join codes (`join.rs`: pairings, join, the `/join/{code}` page), auth, admin CLI (`init`, `serve`, `user` with `invite`, `device`, `reset-epoch`, `live`)
-  craft-cloud-agent               the sync agent: real disk `Fs`, HTTP `Transport` over a `Connection`, saved account and state, root marker, CLI (`login`, `join`, `run`, `sync`, `status`, `logout`, `recents`, `add-computer`, `devices`, `pause`, `resume`, `quit`, `autostart`), the `run` loop, the local API (`local_api.rs`), the live hub (`live/`: recents document, replica, store, link, hub), start at login (`autostart/`: Windows, macOS, Linux), the background build (`src/bin/bg.rs`, logging to `runlog.rs`)
-    tests/network/                the simulation over a simulated network (turmoil): `main.rs` histories and scenarios, `harness.rs` server and agent hosts
+  craft-cloud-server              HTTP API (`/api/v1`), the live service (`/api/v1/live`, `live.rs`: logs, documents by import, snapshots and compaction, presence), join codes (`join.rs`: pairings, join, the `/join/{code}` page), auth, admin CLI (`init`, `serve`, `user` with `invite`, `device`, `reset-epoch`, `live`)
+  craft-cloud-agent               the sync agent: real disk `Fs`, HTTP `Transport` over a `Connection`, saved account and state, root marker, CLI (`login`, `join`, `run`, `sync`, `status`, `logout`, `recents`, `doc` (`doc.rs`), `add-computer`, `devices`, `pause`, `resume`, `quit`, `autostart`), the `run` loop, the local API (`local_api.rs`), the live hub (`live/`: a document's Loro copy, recents document, replica (recents and documents of files), store (with the documents' journals), link, hub, the local API's docs socket), start at login (`autostart/`: Windows, macOS, Linux), the background build (`src/bin/bg.rs`, logging to `runlog.rs`)
+    tests/network/                the simulation over a simulated network (turmoil): `main.rs` histories and scenarios, `harness.rs` server and agent hosts, `live.rs` live documents over WebSockets on it
     tests/recents_sim.rs          the recents convergence simulation: replicas against the server's live sessions
+    tests/documents_sim.rs        the documents convergence simulation (shares `tests/docs_common/` with `documents.rs` and `live_speed.rs`)
   craft-cloud-desktop             the Craft Cloud app (`craft-cloud`, eframe/egui): the pure model (`model.rs`), the agent behind a trait and calls on threads (`agent.rs`), starting the agent (`start_agent.rs`), one app per user (`single_instance.rs`), the window (`ui/`, `app.rs`), the tray (`tray/`: tray-icon in the window's process on Windows and macOS; on Linux a StatusNotifierItem over zbus in a process of its own, `craft-cloud --tray`, `tray/process.rs`, since Wayland can't hide a window), the app's own start at login (`autostart.rs`), its icons (`assets/`, drawn by `cargo xtask icons`)
     tests/snapshots/              every screen as kittest draws it (`UPDATE_SNAPSHOTS=1` redraws them)
   craft-cloud-setup               the Windows setup (`craft-cloud-setup.exe`): install, upgrade, move from a hand-made install, `--uninstall`; the exes built in by `cargo xtask package`

README.md

@@ -7,7 +7,8 @@ all your machines, and the folder `~/Craft Cloud` stays in sync, with version hi
 There is no hosted service and no account with anyone but you. Nothing phones home.
 
 > **Status: early development.** File sync, shared recent files and the Craft Cloud app work. Live
-> documents (an edit shows on your other machines as you type) are next.
+> documents (an edit shows on your other machines as you type) work in the agent and the server; the apps take
+> them up next.
 
 - **Big files sync quickly.** Files are split into content-defined chunks, so a change to a 20 GB project
   uploads only what changed, and an interrupted upload resumes.

apps/craft-cloud-agent/Cargo.toml

@@ -16,6 +16,8 @@ bytes.workspace = true
 clap.workspace = true
 # `recents`, and `login`/`logout` while `run` runs, are clients of the running agent like an app.
 craft-cloud-app.workspace = true
+# Live documents' snapshots go up as files' content does: cut into chunks.
+craft-cloud-chunker.workspace = true
 craft-cloud-proto.workspace = true
 craft-cloud-sync.workspace = true
 file-id.workspace = true
@@ -57,7 +59,6 @@ path = "src/main.rs"
 winresource.workspace = true
 
 [dev-dependencies]
-craft-cloud-chunker.workspace = true
 craft-cloud-meta = { workspace = true, features = ["test-support"] }
 craft-cloud-server = { workspace = true, features = ["test-support"] }
 craft-cloud-store.workspace = true

apps/craft-cloud-agent/src/cli.rs

@@ -1,5 +1,5 @@
 //! The `craft-cloud-agent` command line: `login` or `join`, `run`, `sync`, `status`, `logout`, and through the running
-//! agent `recents`, `add-computer`, `devices`, `pause`, `resume`, `quit` and `autostart`. Each command is a library
+//! agent `recents`, `doc` ([`crate::doc`]), `add-computer`, `devices`, `pause`, `resume`, `quit` and `autostart`. Each command is a library
 //! function too, so tests drive exactly what a user runs.
 //!
 //! The password is read from standard input (one line), so it never appears in the process list or shell
@@ -94,6 +94,9 @@ pub enum Command {
     },
     /// Show an app's recent files (shared by every machine of the account), or add one, through the running agent.
     Recents(RecentsArgs),
+    /// Open a file of the synced folder as a live plain-text document, through the running agent, like an app: print
+    /// its text, append to it, or print it again after every change. A test tool: it never writes the file.
+    Doc(DocArgs),
     /// Join a Craft Cloud with a join string: an invite from the server's admin, or a code from a signed-in computer
     /// (`add-computer` there). An invite for a new account asks for its password, twice, on standard input.
     Join(JoinArgs),
@@ -142,6 +145,18 @@ pub struct RecentsArgs {
     pub action: Option<RecentsAction>,
 }
 
+#[derive(Debug, Clone, clap::Args)]
+pub struct DocArgs {
+    /// The file: its path in the synced folder (absolute, or relative to the folder). Plain text (UTF-8).
+    pub path: PathBuf,
+    /// Append this text to the document (on every machine that has it open).
+    #[arg(long)]
+    pub append: Option<String>,
+    /// Keep printing the text, again after every change, until stopped (Ctrl-C).
+    #[arg(long)]
+    pub watch: bool,
+}
+
 #[derive(Debug, Clone, Subcommand)]
 pub enum RecentsAction {
     /// Add a file: its path in the synced folder (absolute, or relative to the folder).
@@ -419,6 +434,7 @@ pub async fn run(cli: Cli, input: &mut impl BufRead, out: &mut impl Write) -> Re
             Ok(Exit::Done)
         }
         Command::Recents(args) => recents(&data, args, out).await,
+        Command::Doc(args) => crate::doc::doc(&data, args, out).await,
         Command::Join(args) => {
             if let Some(account) = data.load_account()? {
                 return Err(already_signed_in(&account));
@@ -687,15 +703,15 @@ pub async fn main(args: Vec<std::ffi::OsString>, input: &mut impl BufRead, out:
 }
 
 /// Runs a blocking call of `craft-cloud-app` off the runtime.
-async fn blocking<T: Send + 'static>(f: impl FnOnce() -> Result<T, craft_cloud_app::Error> + Send + 'static) -> Result<T, AgentError> {
+pub(crate) async fn blocking<T: Send + 'static>(f: impl FnOnce() -> Result<T, craft_cloud_app::Error> + Send + 'static) -> Result<T, AgentError> {
     tokio::task::spawn_blocking(f)
         .await
         .map_err(|e| AgentError::Usage(format!("the call to the running agent failed: {e}")))?
         .map_err(AgentError::Local)
 }
 
 /// The local API's refusals as the errors the CLI has for them.
-fn local_error(e: AgentError) -> AgentError {
+pub(crate) fn local_error(e: AgentError) -> AgentError {
     use craft_cloud_proto::local::codes;
     match e {
         AgentError::Local(craft_cloud_app::Error::Refused { code, message }) => match code.as_str() {
@@ -805,7 +821,7 @@ async fn recents(data: &DataDir, args: RecentsArgs, out: &mut impl Write) -> Res
 
 /// A path in the synced folder as the local API wants it: relative, `/`-separated. An absolute one (or, on Windows, one
 /// with a root but no drive) must lie in the folder.
-fn relative_to_folder(path: &Path, folder: Option<&Path>) -> Result<String, AgentError> {
+pub(crate) fn relative_to_folder(path: &Path, folder: Option<&Path>) -> Result<String, AgentError> {
     let relative = if path.is_absolute() || path.has_root() {
         let folder = folder.ok_or(AgentError::NotSignedIn)?;
         let (path, folder) = (resolved(path), resolved(folder));

apps/craft-cloud-agent/src/daemon.rs

@@ -612,6 +612,7 @@ pub async fn run_loop(data: &DataDir, options: RunOptions, mut stop: mpsc::Unbou
             server: account.server.clone(),
             token: account.token.clone(),
             machine: account.device_name.clone(),
+            folder: account.folder.clone(),
         };
         let hub = match Hub::start(hub_config, control.clone(), user_paused) {
             Ok(hub) => Some(hub),
@@ -907,6 +908,12 @@ pub async fn run_loop(data: &DataDir, options: RunOptions, mut stop: mpsc::Unbou
                         watching.watcher = None;
                         let _ = reply.send(Ok(()));
                     }
+                    // An app opened a file that isn't uploaded yet: a pass soon (not while paused).
+                    Some(Command::SyncSoon) => {
+                        if !user_paused {
+                            scheduler.urgent_change();
+                        }
+                    }
                     None => commands_open = false,
                 },
                 () = sleep => {}
@@ -1029,6 +1036,7 @@ async fn wait_needing_you(
                     let _ = reply.send(Ok(()));
                     return Stuck::Stop;
                 }
+                Some(Command::SyncSoon) => {}
                 None => commands_open = false,
             },
             signal = stop.recv(), if *stop_open => match signal {
@@ -1091,6 +1099,7 @@ async fn wait_for_sign_in(
                     let _ = reply.send(Ok(()));
                     return SignedOut::Stop;
                 }
+                Some(Command::SyncSoon) => {}
                 None => commands_open = false,
             },
             signal = stop.recv(), if *stop_open => match signal {

apps/craft-cloud-agent/src/doc.rs

@@ -0,0 +1,176 @@
+//! `craft-cloud-agent doc <path> [--append <text>] [--watch]`: a file of the synced folder as a live plain-text
+//! document, through the running agent's local API, a client like an app (`craft-cloud-app`'s live documents). The
+//! mapping is the plainest there is: one Loro text under `text`. A file that has never been live is imported from its
+//! content as UTF-8 (one that isn't is refused). It prints the text, appends to it, or prints it again after every
+//! change. A test tool: it never writes the file.
+
+use std::io::Write;
+use std::path::{Path, PathBuf};
+use std::time::{Duration, Instant};
+
+use craft_cloud_app::live::{Event, Opened};
+use loro::{ExportMode, LoroDoc};
+
+use crate::account::DataDir;
+use crate::cli::{AgentError, DocArgs, Exit, blocking, local_error, relative_to_folder};
+
+/// The app this tool opens documents as.
+pub const APP: &str = "plaintext";
+/// The text's container in the document.
+const TEXT: &str = "text";
+/// How long an appended change may take to be on the agent's disk.
+const ACK_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// A document of one text holding `text`, as a Loro snapshot (the import of a file that has never been live).
+pub fn plain_text_snapshot(text: &str) -> Result<Vec<u8>, String> {
+    let doc = LoroDoc::new();
+    doc.get_text(TEXT).insert(0, text).map_err(|e| e.to_string())?;
+    doc.commit();
+    doc.export(ExportMode::Snapshot).map_err(|e| e.to_string())
+}
+
+fn usage(message: impl Into<String>) -> AgentError {
+    AgentError::Usage(message.into())
+}
+
+/// What the tool's thread tells the command to print.
+enum Line {
+    Out(String),
+    Done(Exit),
+}
+
+pub async fn doc(data: &DataDir, args: DocArgs, out: &mut impl Write) -> Result<Exit, AgentError> {
+    let data_path = data.path().to_owned();
+    let client = blocking(move || craft_cloud_app::Client::connect_in(&data_path)).await.map_err(|e| match e {
+        AgentError::Local(craft_cloud_app::Error::NotRunning(why)) => {
+            usage(format!("the agent isn't running ({why}); start `craft-cloud-agent run` first"))
+        }
+        other => other,
+    })?;
+    let status = {
+        let c = client.clone();
+        blocking(move || c.status()).await?
+    };
+    let folder = status.folder.map(PathBuf::from).ok_or(AgentError::NotSignedIn)?;
+    let path = relative_to_folder(&args.path, Some(&folder))?;
+    // The crate blocks: it runs on a thread of its own, and what it has to say comes back here.
+    let (lines_tx, mut lines) = tokio::sync::mpsc::unbounded_channel::<Line>();
+    let task = tokio::task::spawn_blocking(move || {
+        let said = lines_tx.clone();
+        let say = move |line: String| {
+            let _ = said.send(Line::Out(line));
+        };
+        let exit = session(&client, &folder, &path, args.append.as_deref(), args.watch, &say);
+        if let Ok(exit) = &exit {
+            let _ = lines_tx.send(Line::Done(*exit));
+        }
+        exit
+    });
+    let mut exit = Exit::Done;
+    while let Some(line) = lines.recv().await {
+        match line {
+            Line::Out(text) => {
+                writeln!(out, "{text}").map_err(|e| AgentError::Io { path: Path::new("standard output").to_owned(), source: e })?;
+                let _ = out.flush();
+            }
+            Line::Done(done) => exit = done,
+        }
+    }
+    task.await.map_err(|e| usage(format!("the document's session failed: {e}")))?.map_err(local_error)?;
+    Ok(exit)
+}
+
+fn session(
+    client: &craft_cloud_app::Client,
+    folder: &Path,
+    path: &str,
+    append: Option<&str>,
+    watch: bool,
+    say: &dyn Fn(String),
+) -> Result<Exit, AgentError> {
+    let docs = client.documents().map_err(AgentError::Local)?;
+    let document = match docs.open(path, APP).map_err(AgentError::Local)? {
+        Opened::Live(document) => document,
+        Opened::NeedsImport(importer) => {
+            let file = folder.join(path);
+            let bytes = std::fs::read(&file).map_err(|source| AgentError::Io { path: file.clone(), source })?;
+            let text = String::from_utf8(bytes).map_err(|_| usage(format!("{path} isn't UTF-8 text; `doc` reads plain text only")))?;
+            let snapshot = plain_text_snapshot(&text).map_err(usage)?;
+            importer.import(&snapshot).map_err(AgentError::Local)?
+        }
+        Opened::NotLive(reason) => {
+            say(format!("{path} isn't live ({reason:?}); apps open it as a plain file for now"));
+            return Ok(Exit::Failed);
+        }
+    };
+    let mirror = LoroDoc::new();
+    mirror.import(document.state()).map_err(|e| usage(format!("the document's state doesn't read: {e}")))?;
+    let text = |mirror: &LoroDoc| mirror.get_text(TEXT).to_string();
+    // What came right after the state (updates still waiting for their history), before anything else.
+    let take = |mirror: &LoroDoc, event: &Event| {
+        if let Event::Update(bytes) = event {
+            // (An update this copy can't take is the other side's; the agent has it, and so will a fresh open.)
+            let _ = mirror.import(bytes);
+        }
+    };
+    while let Some(event) = document.try_recv() {
+        take(&mirror, &event);
+    }
+    if let Some(addition) = append {
+        let before = mirror.oplog_vv();
+        let t = mirror.get_text(TEXT);
+        t.insert(t.len_unicode(), addition).map_err(|e| usage(e.to_string()))?;
+        mirror.commit();
+        let update = mirror.export(ExportMode::updates(&before)).map_err(|e| usage(e.to_string()))?;
+        let seq = document.send(&update).map_err(AgentError::Local)?;
+        let until = Instant::now() + ACK_TIMEOUT;
+        loop {
+            let left = until.saturating_duration_since(Instant::now());
+            match document.recv_timeout(left) {
+                Some(Event::Acked(acked)) if acked == seq => break,
+                Some(Event::Refused { seq: refused, message }) if refused == seq => {
+                    return Err(usage(format!("the agent refused the change: {message}")));
+                }
+                Some(Event::Closed(why)) => return Err(usage(format!("the connection to the agent ended: {why}"))),
+                Some(event) => take(&mirror, &event),
+                None => return Err(usage("the agent didn't take the change in time")),
+            }
+        }
+    }
+    say(text(&mirror));
+    if !watch {
+        return Ok(Exit::Done);
+    }
+    loop {
+        match document.recv_timeout(Duration::from_secs(1)) {
+            Some(Event::Update(bytes)) => {
+                let _ = mirror.import(&bytes);
+                say("--".into());
+                say(text(&mirror));
+            }
+            Some(Event::Status { online, waiting, deleted, stopped }) => {
+                let deleted = if deleted { "; its file was deleted" } else { "" };
+                let stopped = if stopped { "; the agent stopped following it until it restarts" } else { "" };
+                say(format!("({}, {waiting} change(s) waiting{deleted}{stopped})", if online { "online" } else { "offline" }));
+            }
+            Some(Event::Closed(why)) => {
+                say(format!("the connection to the agent ended: {why}"));
+                return Ok(Exit::Failed);
+            }
+            Some(Event::Presence(_) | Event::Acked(_) | Event::Refused { .. }) | None => {}
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_plain_text_document_holds_its_text() {
+        let snapshot = plain_text_snapshot("héllo\nwörld").unwrap();
+        let doc = LoroDoc::new();
+        doc.import(&snapshot).unwrap();
+        assert_eq!(doc.get_text(TEXT).to_string(), "héllo\nwörld");
+    }
+}

apps/craft-cloud-agent/src/lib.rs

@@ -5,14 +5,17 @@
 //! ([`account`]), guards the folder with a root marker ([`root`]) and runs it from the command line ([`cli`]): one
 //! pass at a time (`sync`), or on its own (`run`: [`daemon`], scheduled by [`schedule`], woken by [`watch`], with
 //! the stability window kept by [`settle`]). [`autostart`] registers it to start at login; the background build
-//! (`craft-cloud-agent-bg`, no console on Windows) logs to [`runlog`].
+//! (`craft-cloud-agent-bg`, no console on Windows) logs to [`runlog`]. While signed in, the live hub ([`live`]) keeps
+//! the account's live documents (apps' recents, and files opened live) in step with the server, and serves them to
+//! apps through the local API ([`local_api`]); [`doc`] is a plain-text client of it, as a test tool.
 #![deny(clippy::unwrap_used, clippy::expect_used, clippy::panic, clippy::unimplemented, clippy::todo, clippy::unreachable)]
 
 pub mod account;
 pub mod autostart;
 pub mod cli;
 pub mod daemon;
 pub mod disk;
+pub mod doc;
 pub mod http;
 pub mod live;
 pub mod local_api;