mirror of
https://github.com/storytold/figcraft.git
synced 2026-10-28 20:36:55 -06:00
Merge remote-tracking branch 'origin/main' into feat/spell-check
# Conflicts: # ROADMAP.md # docs/gaps.md # docs/roadmap.md # docs/target-app-parity.md # docs/ui-parity.md
This commit is contained in:
bflatastic
@@ -158,6 +158,7 @@ fn measure(table: &mut Table, col: usize, layers: usize) -> CliResult<()> {
|
||||
Point::new(bounds.x0 + bounds.width() * fx, bounds.y0 + bounds.height() * fy)
|
||||
};
|
||||
table.put(col, "hit test (µs each)", mean(hits, |i| _ = doc.hit_test(page, pt(i), 1.0)) * 1000.0);
|
||||
measure_boolean_drag(table, col);
|
||||
|
||||
let mut s = Session::with_document(doc);
|
||||
let doc = s.doc().map_err(failed)?.clone();
|
||||
@@ -240,6 +241,41 @@ fn measure(table: &mut Table, col: usize, layers: usize) -> CliResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Live boolean editing: one drag frame of an operand of a boolean group of 24 overlapping
|
||||
/// circles, i.e. the group's result recomputed with the operand somewhere new (each frame a new
|
||||
/// position, as in a real drag). The top operand recombines one step; the bottom one, all 23.
|
||||
fn measure_boolean_drag(table: &mut Table, col: usize) {
|
||||
let mut d = Document::new("booleans");
|
||||
let Some(page) = d.first_page() else { return };
|
||||
let Ok(b) = d.create(page, NodeKind::BooleanOp { op: figcraft_doc::BoolOp::Union }, "Union", 0.0, 0.0, 0.0, 0.0) else { return };
|
||||
let kids: Vec<NodeId> = (0..24)
|
||||
.filter_map(|i| {
|
||||
let a = f64::from(i) * std::f64::consts::TAU / 24.0;
|
||||
d.create(
|
||||
b,
|
||||
NodeKind::Ellipse { start: 0.0, sweep: std::f64::consts::TAU, inner: 0.0 },
|
||||
"Circle",
|
||||
200.0 + 120.0 * a.cos(),
|
||||
200.0 + 120.0 * a.sin(),
|
||||
90.0,
|
||||
90.0,
|
||||
)
|
||||
.ok()
|
||||
})
|
||||
.collect();
|
||||
let _ = d.fit_to_children(b);
|
||||
let frame = |which: usize, k: usize| {
|
||||
let mut moved = d.clone();
|
||||
if let Some(n) = kids.get(which).and_then(|id| moved.node_mut(*id)) {
|
||||
// Distinct offsets per run and column, so no frame is served from the cache.
|
||||
n.transform = figcraft_doc::geom::Affine::translate((0.37 * (k + 1) as f64 + 1000.0 * col as f64, 0.11 * k as f64)) * n.transform;
|
||||
}
|
||||
moved.node(b).and_then(|n| figcraft_render::node_outline(&moved, n)).is_some()
|
||||
};
|
||||
table.put(col, "boolean of 24: drag frame, top operand", mean(20, |k| _ = frame(23, k)));
|
||||
table.put(col, "boolean of 24: drag frame, bottom operand", mean(5, |k| _ = frame(0, 100 + k)));
|
||||
}
|
||||
|
||||
/// Time the hot paths for each document size in `sizes` (layers) and, with the synthetic design
|
||||
/// system, for each instance count in `instances`; print the tables.
|
||||
pub fn run(sizes: &[usize], instances: &[usize]) -> CliResult<ExitCode> {
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
[package]
|
||||
name = "figcraft-server"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
publish = false
|
||||
description = "FigCraft's standalone collaboration server: keeps many files with share links (edit or view), persists them as snapshots plus a log, and serves live sessions (headless, native only)"
|
||||
|
||||
[lib]
|
||||
path = "src/lib.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "figcraft-server"
|
||||
path = "src/main.rs"
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
[dependencies]
|
||||
figcraft-collab = { workspace = true }
|
||||
figcraft-collab-net = { workspace = true }
|
||||
log = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
sha2 = { workspace = true }
|
||||
crc32fast = { workspace = true }
|
||||
toml = "1"
|
||||
env_logger = { version = "0.11", default-features = false, features = ["humantime"] }
|
||||
|
||||
[dev-dependencies]
|
||||
figcraft-engine = { workspace = true }
|
||||
figcraft-doc = { workspace = true }
|
||||
tungstenite = { version = "0.30", default-features = false, features = ["handshake"] }
|
||||
@@ -0,0 +1,294 @@
|
||||
//! The command line: `figcraft-server` serves; `key`, `files` and `links` manage the data
|
||||
//! directory. When a server is running on the directory they ask it (so revoking a link
|
||||
//! disconnects whoever joined with it at once); otherwise they change the directory themselves.
|
||||
|
||||
use std::fmt::Write as _;
|
||||
use std::net::{IpAddr, SocketAddr};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::Duration;
|
||||
|
||||
use figcraft_collab::message::{Access, Admin, AdminReply, FileInfo, LinkInfo, NewLink};
|
||||
use figcraft_collab_net::hub::Backend;
|
||||
use figcraft_collab_net::{Endpoint, Invite, manage};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::disk;
|
||||
use crate::store::{self, KEY, Store};
|
||||
|
||||
/// Where a running server says it listens.
|
||||
pub const SERVER_JSON: &str = "server.json";
|
||||
|
||||
pub const USAGE: &str = "\
|
||||
figcraft-server: FigCraft's collaboration server (files, share links, live sessions)
|
||||
|
||||
Usage:
|
||||
figcraft-server [serve] [--listen ADDR] [--data DIR] [--public-url URL] [--config FILE]
|
||||
figcraft-server key [--rotate] the address to publish to, with the server key
|
||||
figcraft-server files [list] the files kept
|
||||
figcraft-server files remove FILE remove a file and its links for good
|
||||
figcraft-server links list FILE a file's share links
|
||||
figcraft-server links create FILE --access edit|view [--label TEXT]
|
||||
figcraft-server links revoke FILE LINK revoke a link (whoever joined with it is disconnected)
|
||||
|
||||
Options (also keys of the --config TOML file):
|
||||
--listen ADDR where to listen (default 127.0.0.1:47900; 0.0.0.0:47900 for every network)
|
||||
--data DIR the data directory (default ./figcraft-data)
|
||||
--public-url URL how links name the server behind a reverse proxy (wss://host/path/)
|
||||
--config FILE settings: listen, data, public_url, max_files, max_file_mb, max_links,
|
||||
max_connections, max_people, compact_after, compact_mb
|
||||
-h, --help this text
|
||||
-V, --version the version
|
||||
|
||||
Logs go to standard output (RUST_LOG=debug for more); they never contain keys or tokens.
|
||||
";
|
||||
|
||||
/// What to do.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub enum Command {
|
||||
Serve,
|
||||
Key { rotate: bool },
|
||||
Files,
|
||||
RemoveFile { file: String },
|
||||
Links { file: String },
|
||||
CreateLink { file: String, access: Access, label: String },
|
||||
Revoke { file: String, link: String },
|
||||
Help,
|
||||
Version,
|
||||
}
|
||||
|
||||
/// A parsed command line.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct Cli {
|
||||
pub config: Config,
|
||||
pub command: Command,
|
||||
}
|
||||
|
||||
/// Read the arguments (after the program name).
|
||||
pub fn parse(args: &[String]) -> Result<Cli, String> {
|
||||
// The config file first: flags override it.
|
||||
let mut config = match args.iter().position(|a| a == "--config") {
|
||||
Some(i) => Config::load(Path::new(args.get(i + 1).ok_or("--config needs a file")?))?,
|
||||
None => Config::default(),
|
||||
};
|
||||
let mut words = Vec::new();
|
||||
let mut access = None;
|
||||
let mut label = String::new();
|
||||
let mut rotate = false;
|
||||
let mut it = args.iter();
|
||||
while let Some(a) = it.next() {
|
||||
let mut value = |name: &str| it.next().cloned().ok_or_else(|| format!("{name} needs a value"));
|
||||
match a.as_str() {
|
||||
"-h" | "--help" => return Ok(Cli { config, command: Command::Help }),
|
||||
"-V" | "--version" => return Ok(Cli { config, command: Command::Version }),
|
||||
"--config" => {
|
||||
value("--config")?;
|
||||
}
|
||||
"--listen" => config.listen = value("--listen")?,
|
||||
"--data" => config.data = PathBuf::from(value("--data")?),
|
||||
"--public-url" => config.public_url = Some(value("--public-url")?),
|
||||
"--access" => access = Some(Access::parse(&value("--access")?).ok_or("--access must be edit or view")?),
|
||||
"--label" => label = value("--label")?,
|
||||
"--rotate" => rotate = true,
|
||||
s if s.starts_with('-') => return Err(format!("unknown option {s}")),
|
||||
s => words.push(s.to_string()),
|
||||
}
|
||||
}
|
||||
let w: Vec<&str> = words.iter().map(String::as_str).collect();
|
||||
let command = match w.as_slice() {
|
||||
[] | ["serve"] => Command::Serve,
|
||||
["key"] => Command::Key { rotate },
|
||||
["files"] | ["files", "list"] => Command::Files,
|
||||
["files", "remove", file] => Command::RemoveFile { file: file.to_string() },
|
||||
["links", "list", file] => Command::Links { file: file.to_string() },
|
||||
["links", "create", file] => {
|
||||
Command::CreateLink { file: file.to_string(), access: access.ok_or("links create needs --access edit or --access view")?, label }
|
||||
}
|
||||
["links", "revoke", file, link] => Command::Revoke { file: file.to_string(), link: link.to_string() },
|
||||
_ => return Err(format!("unknown command `{}`", words.join(" "))),
|
||||
};
|
||||
config.check()?;
|
||||
Ok(Cli { config, command })
|
||||
}
|
||||
|
||||
/// What a running server writes into its data directory.
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct Running {
|
||||
pub listen: SocketAddr,
|
||||
pub pid: u32,
|
||||
}
|
||||
|
||||
/// Note where the server listens (for the command line).
|
||||
pub fn write_running(data: &Path, listen: SocketAddr) -> Result<(), String> {
|
||||
let text = serde_json::to_vec_pretty(&Running { listen, pid: std::process::id() }).map_err(|e| e.to_string())?;
|
||||
disk::write_atomic(&data.join(SERVER_JSON), &text).map_err(|e| format!("can't write {SERVER_JSON}: {e}"))
|
||||
}
|
||||
|
||||
/// `addr` as something this machine can connect to (the loopback address for "every interface").
|
||||
fn reachable(addr: SocketAddr) -> SocketAddr {
|
||||
match addr.ip() {
|
||||
IpAddr::V4(ip) if ip.is_unspecified() => SocketAddr::from(([127, 0, 0, 1], addr.port())),
|
||||
IpAddr::V6(ip) if ip.is_unspecified() => SocketAddr::from((std::net::Ipv6Addr::LOCALHOST, addr.port())),
|
||||
_ => addr,
|
||||
}
|
||||
}
|
||||
|
||||
/// How share links and the publish address name the server: `public_url`, else the listen
|
||||
/// address (this machine's network address for "every interface").
|
||||
pub fn public_endpoint(cfg: &Config, listen: SocketAddr) -> Result<Endpoint, String> {
|
||||
if let Some(e) = cfg.public()? {
|
||||
return Ok(e);
|
||||
}
|
||||
let ip = if listen.ip().is_unspecified() { figcraft_collab_net::lan_ip().unwrap_or(reachable(listen).ip()) } else { listen.ip() };
|
||||
Ok(Endpoint::plain(SocketAddr::new(ip, listen.port()).to_string()))
|
||||
}
|
||||
|
||||
/// Run a command other than serving; what to print.
|
||||
pub fn run(cli: &Cli) -> Result<String, String> {
|
||||
let cfg = &cli.config;
|
||||
match &cli.command {
|
||||
Command::Help => Ok(USAGE.to_string()),
|
||||
Command::Version => Ok(format!("figcraft-server {}\n", env!("CARGO_PKG_VERSION"))),
|
||||
Command::Serve => Err("serving runs in the foreground: run figcraft-server without a command".into()),
|
||||
Command::Key { rotate } => key(cfg, *rotate),
|
||||
Command::Files => match admin(cfg, Admin::Files)? {
|
||||
AdminReply::Files(files) => Ok(files_table(&files)),
|
||||
other => unexpected(other),
|
||||
},
|
||||
Command::RemoveFile { file } => match admin(cfg, Admin::Remove { file: file.clone() })? {
|
||||
AdminReply::Done => Ok(format!("Removed file {file} and its links.\n")),
|
||||
other => unexpected(other),
|
||||
},
|
||||
Command::Links { file } => match admin(cfg, Admin::Links { file: file.clone() })? {
|
||||
AdminReply::Links(links) => Ok(links_table(&links)),
|
||||
other => unexpected(other),
|
||||
},
|
||||
Command::CreateLink { file, access, label } => {
|
||||
match admin(cfg, Admin::CreateLink { file: file.clone(), access: *access, label: label.clone() })? {
|
||||
AdminReply::Link(l) => Ok(new_link(cfg, &l)?),
|
||||
other => unexpected(other),
|
||||
}
|
||||
}
|
||||
Command::Revoke { file, link } => match admin(cfg, Admin::Revoke { file: file.clone(), link: link.clone() })? {
|
||||
AdminReply::Done => Ok(format!("Revoked link {link}: it no longer joins, and whoever was in the file with it was disconnected.\n")),
|
||||
other => unexpected(other),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn unexpected(reply: AdminReply) -> Result<String, String> {
|
||||
match reply {
|
||||
AdminReply::Failed { reason } => Err(reason),
|
||||
other => Err(format!("the server answered {other:?}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// The running server's address, when one holds the data directory.
|
||||
fn running(data: &Path) -> Result<SocketAddr, String> {
|
||||
let bytes = disk::read_capped(&data.join(SERVER_JSON), 1 << 16).map_err(|e| format!("a server is using {} but {e}", data.display()))?;
|
||||
let r: Running = serde_json::from_slice(&bytes).map_err(|e| format!("{SERVER_JSON} is damaged ({e})"))?;
|
||||
Ok(reachable(r.listen))
|
||||
}
|
||||
|
||||
/// Do one request: through the running server, or on the data directory.
|
||||
fn admin(cfg: &Config, req: Admin) -> Result<AdminReply, String> {
|
||||
match Store::try_open(&cfg.data, cfg.store())? {
|
||||
Some(mut store) => Ok(store.admin(req, &|_| None).reply),
|
||||
None => {
|
||||
let addr = running(&cfg.data)?;
|
||||
let key = std::fs::read_to_string(cfg.data.join(KEY)).map_err(|e| format!("can't read the server key: {e}"))?;
|
||||
let replies = manage::request(&Endpoint::plain(addr.to_string()), key.trim(), vec![req], &cfg.net(), Duration::from_secs(5))?;
|
||||
replies.into_iter().next().ok_or_else(|| "the server didn't answer".to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The address to publish to, with the key.
|
||||
fn key(cfg: &Config, rotate: bool) -> Result<String, String> {
|
||||
let mut store = Store::try_open(&cfg.data, cfg.store())?;
|
||||
let listen = match &store {
|
||||
Some(_) => cfg.addr()?,
|
||||
None => running(&cfg.data)?,
|
||||
};
|
||||
let key = match (store.as_mut(), rotate) {
|
||||
(Some(s), true) => s.rotate_key()?,
|
||||
(Some(s), false) => s.key().to_string(),
|
||||
(None, true) => return Err("stop the server to replace its key".into()),
|
||||
(None, false) => std::fs::read_to_string(cfg.data.join(KEY)).map_err(|e| format!("can't read the server key: {e}"))?.trim().to_string(),
|
||||
};
|
||||
let public = public_endpoint(cfg, listen)?;
|
||||
let mut out = String::new();
|
||||
if rotate {
|
||||
out.push_str("A new server key: the old one no longer works.\n");
|
||||
}
|
||||
let _ = writeln!(out, "Publish to this address (in FigCraft: Share > Publish to server…):");
|
||||
let _ = writeln!(out, " {}?key={key}", public.url());
|
||||
let _ = writeln!(out, "Whoever has it can publish files and manage their links: share it only with people you trust.");
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
fn new_link(cfg: &Config, l: &NewLink) -> Result<String, String> {
|
||||
let listen = match running(&cfg.data) {
|
||||
Ok(a) if store::lock(&cfg.data)?.is_none() => a,
|
||||
_ => cfg.addr()?,
|
||||
};
|
||||
let invite = Invite { server: public_endpoint(cfg, listen)?, file: l.link.file.clone(), token: l.token.clone() };
|
||||
let mut out = String::new();
|
||||
let what = if l.link.access == Access::View { "view" } else { "edit" };
|
||||
let _ = writeln!(out, "New {what} link {} for file {}:", l.link.id, l.link.file);
|
||||
let _ = writeln!(out, " {}", invite.link());
|
||||
let _ = writeln!(out, " {}", invite.url());
|
||||
let _ = writeln!(out, "This is the only time the link is shown (the server keeps only a hash of its token).");
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
fn files_table(files: &[FileInfo]) -> String {
|
||||
if files.is_empty() {
|
||||
return "No files yet (publish one from FigCraft: Share > Publish to server…).\n".into();
|
||||
}
|
||||
let mut out = String::from("ID NAME PEOPLE UPDATED SIZE\n");
|
||||
for f in files {
|
||||
let _ = writeln!(out, "{:<16} {:<30} {:>6} {:<20} {}", f.id, elide(&f.name, 30), f.people, date(f.updated), size(f.bytes));
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn links_table(links: &[LinkInfo]) -> String {
|
||||
if links.is_empty() {
|
||||
return "No links (make one with `figcraft-server links create FILE --access view`).\n".into();
|
||||
}
|
||||
let mut out = String::from("LINK ACCESS CREATED LABEL\n");
|
||||
for l in links {
|
||||
let _ = writeln!(out, "{:<16} {:<6} {:<20} {}", l.id, l.access.id(), date(l.created), l.label);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn elide(s: &str, n: usize) -> String {
|
||||
if s.chars().count() <= n { s.to_string() } else { format!("{}…", s.chars().take(n - 1).collect::<String>()) }
|
||||
}
|
||||
|
||||
fn size(bytes: u64) -> String {
|
||||
match bytes {
|
||||
b if b >= 1 << 20 => format!("{:.1} MiB", b as f64 / f64::from(1 << 20)),
|
||||
b if b >= 1 << 10 => format!("{:.1} KiB", b as f64 / 1024.0),
|
||||
b => format!("{b} B"),
|
||||
}
|
||||
}
|
||||
|
||||
/// `YYYY-MM-DD HH:MM UTC` for seconds since 1970.
|
||||
pub fn date(secs: u64) -> String {
|
||||
let days = i64::try_from(secs / 86_400).unwrap_or(0);
|
||||
let (h, m) = ((secs % 86_400) / 3600, (secs % 3600) / 60);
|
||||
// Days to a civil date (proleptic Gregorian).
|
||||
let z = days + 719_468;
|
||||
let era = z.div_euclid(146_097);
|
||||
let doe = z.rem_euclid(146_097);
|
||||
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
|
||||
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
|
||||
let mp = (5 * doy + 2) / 153;
|
||||
let d = doy - (153 * mp + 2) / 5 + 1;
|
||||
let mo = if mp < 10 { mp + 3 } else { mp - 9 };
|
||||
let y = yoe + era * 400 + i64::from(mo <= 2);
|
||||
format!("{y:04}-{mo:02}-{d:02} {h:02}:{m:02} UTC")
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
//! The command line: parsing (flags over the config file), and managing a data directory both
|
||||
//! on its own and through the server running on it.
|
||||
|
||||
use figcraft_collab::Replica;
|
||||
use figcraft_collab::message::Access;
|
||||
use figcraft_doc::Document;
|
||||
|
||||
use crate::cli::{self, Cli, Command, date, parse};
|
||||
use crate::config::Config;
|
||||
use crate::store::Store;
|
||||
use crate::test_util::{TempDir, config};
|
||||
|
||||
fn args(s: &[&str]) -> Vec<String> {
|
||||
s.iter().map(|a| a.to_string()).collect()
|
||||
}
|
||||
|
||||
fn run(dir: &std::path::Path, command: Command) -> Result<String, String> {
|
||||
cli::run(&Cli { config: config(dir), command })
|
||||
}
|
||||
|
||||
/// A data directory with one published file: its id.
|
||||
fn with_file(dir: &std::path::Path) -> String {
|
||||
let mut store = Store::open(dir, config(dir).store()).unwrap();
|
||||
let snapshot = Replica::new(Document::new("Board")).snapshot().unwrap();
|
||||
store.publish("Board", &snapshot).unwrap().0.id
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parsing() {
|
||||
let c = parse(&[]).unwrap();
|
||||
assert_eq!(c.command, Command::Serve);
|
||||
assert_eq!(c.config, Config::default());
|
||||
let c = parse(&args(&["serve", "--listen", "0.0.0.0:47900", "--data", "d", "--public-url", "wss://f.example.com/live/"])).unwrap();
|
||||
assert_eq!((c.config.listen.as_str(), c.config.data.to_str()), ("0.0.0.0:47900", Some("d")));
|
||||
assert_eq!(
|
||||
parse(&args(&["links", "create", "f1", "--access", "view", "--label", "Client"])).unwrap().command,
|
||||
Command::CreateLink { file: "f1".into(), access: Access::View, label: "Client".into() }
|
||||
);
|
||||
assert_eq!(parse(&args(&["files"])).unwrap().command, Command::Files);
|
||||
assert_eq!(parse(&args(&["key", "--rotate"])).unwrap().command, Command::Key { rotate: true });
|
||||
assert_eq!(parse(&args(&["-h"])).unwrap().command, Command::Help);
|
||||
for bad in
|
||||
[&["--nope"][..], &["links", "create", "f1"], &["--listen", "nowhere"], &["--listen"], &["frobnicate"], &["--public-url", "ws://h/?key=k"]]
|
||||
{
|
||||
assert!(parse(&args(bad)).is_err(), "{bad:?}");
|
||||
}
|
||||
// A config file, with flags over it; unknown keys are errors.
|
||||
let dir = TempDir::new("cli-config");
|
||||
let file = dir.path().join("server.toml");
|
||||
std::fs::write(&file, "listen = \"0.0.0.0:47901\"\nmax_files = 7\ncompact_after = 50\n").unwrap();
|
||||
let c = parse(&args(&["--config", file.to_str().unwrap(), "--listen", "127.0.0.1:1"])).unwrap();
|
||||
assert_eq!((c.config.listen.as_str(), c.config.max_files, c.config.compact_after), ("127.0.0.1:1", 7, 50));
|
||||
std::fs::write(&file, "max_files = 0\n").unwrap();
|
||||
assert!(parse(&args(&["--config", file.to_str().unwrap()])).unwrap_err().contains("max_files"));
|
||||
std::fs::write(&file, "colour = \"blue\"\n").unwrap();
|
||||
assert!(parse(&args(&["--config", file.to_str().unwrap()])).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn managing_a_directory_on_its_own() {
|
||||
let dir = TempDir::new("cli-offline");
|
||||
let file = with_file(dir.path());
|
||||
let list = run(dir.path(), Command::Files).unwrap();
|
||||
assert!(list.contains(&file) && list.contains("Board"), "{list}");
|
||||
let made = run(dir.path(), Command::CreateLink { file: file.clone(), access: Access::View, label: "Client".into() }).unwrap();
|
||||
assert!(made.contains("New view link") && made.contains(&format!("file={file}&token=")), "{made}");
|
||||
let links = run(dir.path(), Command::Links { file: file.clone() }).unwrap();
|
||||
assert_eq!(links.lines().count(), 4, "a header and three links: {links}");
|
||||
let id = links.lines().find(|l| l.contains("Client")).and_then(|l| l.split_whitespace().next()).unwrap().to_string();
|
||||
run(dir.path(), Command::Revoke { file: file.clone(), link: id.clone() }).unwrap();
|
||||
assert!(run(dir.path(), Command::Revoke { file: file.clone(), link: id }).unwrap_err().contains("no such link"));
|
||||
// The key: printed as the address to publish to; replaced on request.
|
||||
let key = run(dir.path(), Command::Key { rotate: false }).unwrap();
|
||||
assert!(key.contains("ws://127.0.0.1:") && key.contains("?key="), "{key}");
|
||||
let rotated = run(dir.path(), Command::Key { rotate: true }).unwrap();
|
||||
assert_ne!(key.lines().nth(1), rotated.lines().nth(2));
|
||||
run(dir.path(), Command::RemoveFile { file: file.clone() }).unwrap();
|
||||
assert!(run(dir.path(), Command::Files).unwrap().contains("No files yet"));
|
||||
assert!(run(dir.path(), Command::Links { file }).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn managing_through_the_running_server() {
|
||||
let dir = TempDir::new("cli-online");
|
||||
let file = with_file(dir.path());
|
||||
let mut server = crate::start(&config(dir.path())).unwrap();
|
||||
// The directory is the server's now: commands ask it.
|
||||
let made = run(dir.path(), Command::CreateLink { file: file.clone(), access: Access::Edit, label: "Team".into() }).unwrap();
|
||||
assert!(made.contains(&format!("host={}", server.addr())), "links name the running server: {made}");
|
||||
let links = run(dir.path(), Command::Links { file: file.clone() }).unwrap();
|
||||
assert!(links.contains("Team"), "{links}");
|
||||
let id = links.lines().find(|l| l.contains("Team")).and_then(|l| l.split_whitespace().next()).unwrap().to_string();
|
||||
assert!(run(dir.path(), Command::Revoke { file: file.clone(), link: id }).unwrap().contains("Revoked"));
|
||||
assert!(run(dir.path(), Command::Files).unwrap().contains("Board"));
|
||||
assert!(run(dir.path(), Command::Key { rotate: true }).unwrap_err().contains("stop the server"));
|
||||
assert!(run(dir.path(), Command::Key { rotate: false }).unwrap().contains(&format!("ws://{}/?key=", server.addr())));
|
||||
server.stop();
|
||||
// Stopped: the directory is free again.
|
||||
assert!(run(dir.path(), Command::Files).unwrap().contains("Board"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dates() {
|
||||
assert_eq!(date(0), "1970-01-01 00:00 UTC");
|
||||
assert_eq!(date(951_782_400), "2000-02-29 00:00 UTC");
|
||||
assert_eq!(date(1_791_986_580), "2026-10-14 14:03 UTC");
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
//! The server's settings: defaults, then a TOML file (`--config server.toml`), then flags.
|
||||
|
||||
use std::net::SocketAddr;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use figcraft_collab_net::{Endpoint, NetLimits};
|
||||
use serde::Deserialize;
|
||||
|
||||
use crate::store::StoreConfig;
|
||||
|
||||
/// The port the server listens on unless told otherwise (the app's sessions try it first too).
|
||||
pub const DEFAULT_PORT: u16 = figcraft_collab_net::DEFAULT_PORT;
|
||||
|
||||
/// Everything the server can be told (the TOML file's keys are these names).
|
||||
#[derive(Clone, Debug, PartialEq, Deserialize)]
|
||||
#[serde(default, deny_unknown_fields)]
|
||||
pub struct Config {
|
||||
/// Where to listen: `127.0.0.1:47900` (this machine only) unless told otherwise.
|
||||
pub listen: String,
|
||||
/// The data directory (files, links, the server key).
|
||||
pub data: PathBuf,
|
||||
/// How share links name this server when it isn't reached at `listen` (a reverse proxy:
|
||||
/// `wss://figcraft.example.com/`).
|
||||
pub public_url: Option<String>,
|
||||
/// Most files kept.
|
||||
pub max_files: usize,
|
||||
/// Largest file (its snapshot), in MiB (a welcome carries the whole file, and messages are at
|
||||
/// most 64 MiB).
|
||||
pub max_file_mb: u64,
|
||||
/// Most share links per file.
|
||||
pub max_links: usize,
|
||||
/// Most open connections.
|
||||
pub max_connections: usize,
|
||||
/// Most people in one file at once.
|
||||
pub max_people: usize,
|
||||
/// Compact a file's log into a new snapshot after this many entries…
|
||||
pub compact_after: usize,
|
||||
/// …or once the log is this large (MiB).
|
||||
pub compact_mb: u64,
|
||||
}
|
||||
|
||||
impl Default for Config {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
listen: format!("127.0.0.1:{DEFAULT_PORT}"),
|
||||
data: PathBuf::from("figcraft-data"),
|
||||
public_url: None,
|
||||
max_files: 1000,
|
||||
max_file_mb: 64,
|
||||
max_links: 100,
|
||||
max_connections: 512,
|
||||
max_people: 256,
|
||||
compact_after: 1000,
|
||||
compact_mb: 16,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Config {
|
||||
/// Read a TOML file's settings (unknown keys are errors).
|
||||
pub fn load(path: &Path) -> Result<Config, String> {
|
||||
let text = std::fs::read_to_string(path).map_err(|e| format!("can't read {}: {e}", path.display()))?;
|
||||
toml::from_str(&text).map_err(|e| format!("{}: {e}", path.display()))
|
||||
}
|
||||
|
||||
/// The listen address.
|
||||
pub fn addr(&self) -> Result<SocketAddr, String> {
|
||||
self.listen.trim().parse().map_err(|_| format!("`listen` must be an address with a port, like 0.0.0.0:{DEFAULT_PORT}"))
|
||||
}
|
||||
|
||||
/// The public URL, when set: a `ws://` or `wss://` address without a query.
|
||||
pub fn public(&self) -> Result<Option<Endpoint>, String> {
|
||||
let Some(url) = self.public_url.as_deref().map(str::trim).filter(|u| !u.is_empty()) else { return Ok(None) };
|
||||
let (e, query) = Endpoint::parse(url).map_err(|e| format!("`public_url`: {e}"))?;
|
||||
if !query.is_empty() {
|
||||
return Err("`public_url` can't have a query (no keys or tokens in it)".into());
|
||||
}
|
||||
Ok(Some(e))
|
||||
}
|
||||
|
||||
/// Check everything (before starting).
|
||||
pub fn check(&self) -> Result<(), String> {
|
||||
self.addr()?;
|
||||
self.public()?;
|
||||
for (name, v) in [
|
||||
("max_files", self.max_files as u64),
|
||||
("max_file_mb", self.max_file_mb),
|
||||
("max_links", self.max_links as u64),
|
||||
("max_connections", self.max_connections as u64),
|
||||
("max_people", self.max_people as u64),
|
||||
("compact_after", self.compact_after as u64),
|
||||
("compact_mb", self.compact_mb),
|
||||
] {
|
||||
if v == 0 {
|
||||
return Err(format!("`{name}` must be at least 1"));
|
||||
}
|
||||
}
|
||||
if self.max_file_mb > 4096 || self.compact_mb > 4096 {
|
||||
return Err("sizes are in MiB and at most 4096".into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The store's part.
|
||||
pub fn store(&self) -> StoreConfig {
|
||||
let mib = |n: u64| n.saturating_mul(1 << 20);
|
||||
StoreConfig {
|
||||
max_files: self.max_files,
|
||||
max_file_bytes: mib(self.max_file_mb),
|
||||
max_links: self.max_links,
|
||||
compact_after: self.compact_after,
|
||||
compact_bytes: mib(self.compact_mb),
|
||||
core: figcraft_collab::Limits { max_peers: self.max_people, ..figcraft_collab::Limits::default() },
|
||||
}
|
||||
}
|
||||
|
||||
/// The transport's part.
|
||||
pub fn net(&self) -> NetLimits {
|
||||
NetLimits { max_connections: self.max_connections, ..NetLimits::default() }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,200 @@
|
||||
//! The server's files on disk and reading them back without trusting them: atomic writes (a
|
||||
//! temporary file beside the target, flushed, then renamed over it), the snapshot format and the
|
||||
//! append-only log format.
|
||||
//!
|
||||
//! - `snapshot.bin`: `FCSN`, a format byte, the generation (u64 LE), a CRC-32 of the rest (u32 LE),
|
||||
//! then the session state ([`SessionState`] in the wire encoding).
|
||||
//! - `log.bin`: `FCLG`, a format byte, the generation of the snapshot it continues, then records:
|
||||
//! length (u32 LE), CRC-32 (u32 LE), one [`Journal`] entry in the wire encoding.
|
||||
//!
|
||||
//! Compaction writes the next generation's snapshot, then an empty log of that generation. A log
|
||||
//! whose generation isn't the snapshot's is from before a compaction that was cut short: what it
|
||||
//! holds is in the snapshot already. A log ends at its first damaged record (a write cut short).
|
||||
|
||||
use std::fs::{self, File};
|
||||
use std::io::{self, Read, Write};
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use figcraft_collab::message;
|
||||
use figcraft_collab::{Journal, SessionState};
|
||||
|
||||
pub const SNAPSHOT: &str = "snapshot.bin";
|
||||
pub const LOG: &str = "log.bin";
|
||||
pub const META: &str = "meta.json";
|
||||
pub const LINKS: &str = "links.json";
|
||||
|
||||
const SNAP_MAGIC: &[u8; 4] = b"FCSN";
|
||||
const LOG_MAGIC: &[u8; 4] = b"FCLG";
|
||||
const FORMAT: u8 = 1;
|
||||
const SNAP_HEADER: usize = 4 + 1 + 8 + 4;
|
||||
pub const LOG_HEADER: usize = 4 + 1 + 8;
|
||||
const RECORD_HEADER: usize = 8;
|
||||
|
||||
/// `path` with `.tmp` added: where its next version is written first.
|
||||
pub fn temp_of(path: &Path) -> PathBuf {
|
||||
let mut name = path.file_name().map(|n| n.to_os_string()).unwrap_or_default();
|
||||
name.push(".tmp");
|
||||
path.with_file_name(name)
|
||||
}
|
||||
|
||||
/// Replace `path` with `bytes` atomically: a reader sees the old or the new content, never a mix.
|
||||
pub fn write_atomic(path: &Path, bytes: &[u8]) -> io::Result<()> {
|
||||
let tmp = temp_of(path);
|
||||
{
|
||||
let mut f = File::create(&tmp)?;
|
||||
f.write_all(bytes)?;
|
||||
f.sync_all()?;
|
||||
}
|
||||
fs::rename(&tmp, path)?;
|
||||
sync_dir(path.parent());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Make a rename in `dir` durable (directories can't be opened for that on Windows).
|
||||
pub fn sync_dir(dir: Option<&Path>) {
|
||||
#[cfg(unix)]
|
||||
if let Some(d) = dir
|
||||
&& let Ok(f) = File::open(d)
|
||||
{
|
||||
let _ = f.sync_all();
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
let _ = dir;
|
||||
}
|
||||
|
||||
/// Read a file of at most `max` bytes (a longer one is refused without being read).
|
||||
pub fn read_capped(path: &Path, max: u64) -> Result<Vec<u8>, String> {
|
||||
let f = File::open(path).map_err(|e| format!("can't open {}: {e}", name(path)))?;
|
||||
let len = f.metadata().map_err(|e| e.to_string())?.len();
|
||||
if len > max {
|
||||
return Err(format!("{} is {} bytes, over the {max}-byte cap", name(path), len));
|
||||
}
|
||||
let mut bytes = Vec::with_capacity(usize::try_from(len).unwrap_or(0));
|
||||
f.take(max).read_to_end(&mut bytes).map_err(|e| format!("can't read {}: {e}", name(path)))?;
|
||||
Ok(bytes)
|
||||
}
|
||||
|
||||
fn name(path: &Path) -> String {
|
||||
path.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_default()
|
||||
}
|
||||
|
||||
fn crc(bytes: &[u8]) -> u32 {
|
||||
crc32fast::hash(bytes)
|
||||
}
|
||||
|
||||
fn u32_at(b: &[u8], at: usize) -> Option<u32> {
|
||||
b.get(at..at + 4).and_then(|s| s.try_into().ok()).map(u32::from_le_bytes)
|
||||
}
|
||||
|
||||
fn u64_at(b: &[u8], at: usize) -> Option<u64> {
|
||||
b.get(at..at + 8).and_then(|s| s.try_into().ok()).map(u64::from_le_bytes)
|
||||
}
|
||||
|
||||
/// A snapshot file's bytes.
|
||||
pub fn encode_snapshot(generation: u64, state: &SessionState) -> Result<Vec<u8>, String> {
|
||||
let body = message::encode(state).map_err(|e| e.to_string())?;
|
||||
let mut out = Vec::with_capacity(SNAP_HEADER + body.len());
|
||||
out.extend_from_slice(SNAP_MAGIC);
|
||||
out.push(FORMAT);
|
||||
out.extend_from_slice(&generation.to_le_bytes());
|
||||
out.extend_from_slice(&crc(&body).to_le_bytes());
|
||||
out.extend_from_slice(&body);
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// A snapshot's generation and body, when its header and checksum are right.
|
||||
fn snapshot_body(bytes: &[u8]) -> Result<(u64, &[u8]), String> {
|
||||
if bytes.get(..4) != Some(SNAP_MAGIC.as_slice()) {
|
||||
return Err("the snapshot isn't one (wrong header)".into());
|
||||
}
|
||||
if bytes.get(4) != Some(&FORMAT) {
|
||||
return Err("the snapshot is in a format this version doesn't read".into());
|
||||
}
|
||||
let (Some(generation), Some(sum), Some(body)) = (u64_at(bytes, 5), u32_at(bytes, 13), bytes.get(SNAP_HEADER..)) else {
|
||||
return Err("the snapshot is cut short".into());
|
||||
};
|
||||
if crc(body) != sum {
|
||||
return Err("the snapshot is damaged (checksum mismatch)".into());
|
||||
}
|
||||
Ok((generation, body))
|
||||
}
|
||||
|
||||
/// A snapshot's generation, after checking its header and checksum (no decoding).
|
||||
pub fn check_snapshot(bytes: &[u8]) -> Result<u64, String> {
|
||||
snapshot_body(bytes).map(|(g, _)| g)
|
||||
}
|
||||
|
||||
/// A snapshot's generation and session state.
|
||||
pub fn decode_snapshot(bytes: &[u8]) -> Result<(u64, SessionState), String> {
|
||||
let (generation, body) = snapshot_body(bytes)?;
|
||||
let state = message::decode::<SessionState>(body, body.len()).map_err(|e| format!("the snapshot can't be read: {e}"))?;
|
||||
Ok((generation, state))
|
||||
}
|
||||
|
||||
/// An empty log continuing snapshot `generation`.
|
||||
pub fn log_header(generation: u64) -> Vec<u8> {
|
||||
let mut out = Vec::with_capacity(LOG_HEADER);
|
||||
out.extend_from_slice(LOG_MAGIC);
|
||||
out.push(FORMAT);
|
||||
out.extend_from_slice(&generation.to_le_bytes());
|
||||
out
|
||||
}
|
||||
|
||||
/// One log record.
|
||||
pub fn record(entry: &Journal) -> Result<Vec<u8>, String> {
|
||||
let body = message::encode(entry).map_err(|e| e.to_string())?;
|
||||
let len = u32::try_from(body.len()).map_err(|_| "a journal entry over 4 GiB".to_string())?;
|
||||
let mut out = Vec::with_capacity(RECORD_HEADER + body.len());
|
||||
out.extend_from_slice(&len.to_le_bytes());
|
||||
out.extend_from_slice(&crc(&body).to_le_bytes());
|
||||
out.extend_from_slice(&body);
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// What a log holds.
|
||||
#[derive(Debug)]
|
||||
pub struct LogRead {
|
||||
pub generation: u64,
|
||||
pub entries: Vec<Journal>,
|
||||
/// Why it ends before the end of the file (a damaged or cut-short record), if it does.
|
||||
pub damaged: Option<String>,
|
||||
}
|
||||
|
||||
/// Read a log: its generation and the entries up to the first damaged record. Only a header that
|
||||
/// isn't a log's is an error.
|
||||
pub fn read_log(bytes: &[u8]) -> Result<LogRead, String> {
|
||||
if bytes.get(..4) != Some(LOG_MAGIC.as_slice()) {
|
||||
return Err("the log isn't one (wrong header)".into());
|
||||
}
|
||||
if bytes.get(4) != Some(&FORMAT) {
|
||||
return Err("the log is in a format this version doesn't read".into());
|
||||
}
|
||||
let generation = u64_at(bytes, 5).ok_or("the log is cut short")?;
|
||||
let mut at = LOG_HEADER;
|
||||
let mut entries = Vec::new();
|
||||
let mut damaged = None;
|
||||
while at < bytes.len() {
|
||||
let (Some(len), Some(sum)) = (u32_at(bytes, at), u32_at(bytes, at + 4)) else {
|
||||
damaged = Some(format!("a record cut short at byte {at}"));
|
||||
break;
|
||||
};
|
||||
let start = at + RECORD_HEADER;
|
||||
let Some(body) = usize::try_from(len).ok().and_then(|n| bytes.get(start..start.checked_add(n)?)) else {
|
||||
damaged = Some(format!("a record cut short at byte {at}"));
|
||||
break;
|
||||
};
|
||||
if crc(body) != sum {
|
||||
damaged = Some(format!("a damaged record at byte {at}"));
|
||||
break;
|
||||
}
|
||||
match message::decode::<Journal>(body, body.len()) {
|
||||
Ok(e) => entries.push(e),
|
||||
Err(e) => {
|
||||
damaged = Some(format!("an unreadable record at byte {at} ({e})"));
|
||||
break;
|
||||
}
|
||||
}
|
||||
at = start + body.len();
|
||||
}
|
||||
Ok(LogRead { generation, entries, damaged })
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
//! `figcraft-server`: FigCraft's standalone collaboration server (milestone M9, step 5 of
|
||||
//! `docs/collab-protocol.md`; running it: `docs/server.md`).
|
||||
//!
|
||||
//! It keeps many files in a data directory ([`store`]), each with share links that let people
|
||||
//! edit or only view, and serves their live sessions through the same hub the desktop app hosts
|
||||
//! with ([`figcraft_collab_net::hub`]). Files persist as a snapshot plus an append-only log of what
|
||||
//! changed ([`disk`]), compacted into a new snapshot from time to time, so a restart loses nothing
|
||||
//! that was confirmed and clients resume where they were. There are no user accounts yet: share
|
||||
//! links are the access, and the server key manages files.
|
||||
|
||||
pub mod cli;
|
||||
pub mod config;
|
||||
pub mod disk;
|
||||
pub mod store;
|
||||
|
||||
use std::net::{SocketAddr, TcpListener};
|
||||
|
||||
use figcraft_collab_net::hub::Hosting;
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::store::Store;
|
||||
|
||||
/// A running server. Dropping it stops it: open files are saved, everyone is told the server is
|
||||
/// restarting (their apps reconnect once it is back).
|
||||
pub struct Running {
|
||||
hosting: Hosting,
|
||||
}
|
||||
|
||||
impl Running {
|
||||
/// Where it listens.
|
||||
pub fn addr(&self) -> SocketAddr {
|
||||
self.hosting.addr()
|
||||
}
|
||||
|
||||
/// Open connections.
|
||||
pub fn connections(&self) -> usize {
|
||||
self.hosting.connections()
|
||||
}
|
||||
|
||||
/// Stop (see the type's docs).
|
||||
pub fn stop(&mut self) {
|
||||
self.hosting.stop();
|
||||
}
|
||||
}
|
||||
|
||||
/// Start serving `cfg.data` on `cfg.listen`. Files that can't be read are skipped and logged.
|
||||
pub fn start(cfg: &Config) -> Result<Running, String> {
|
||||
cfg.check()?;
|
||||
let store = Store::open(&cfg.data, cfg.store())?;
|
||||
for p in store.problems() {
|
||||
log::warn!("{p}");
|
||||
}
|
||||
let files = store.len();
|
||||
let addr = cfg.addr()?;
|
||||
let listener = TcpListener::bind(addr).map_err(|e| format!("can't listen on {addr}: {e}"))?;
|
||||
let addr = listener.local_addr().map_err(|e| e.to_string())?;
|
||||
cli::write_running(&cfg.data, addr)?;
|
||||
let public = cli::public_endpoint(cfg, addr)?;
|
||||
let hosting = Hosting::serve(listener, store, cfg.net()).map_err(|e| format!("can't start serving: {e}"))?;
|
||||
log::info!(
|
||||
"figcraft-server {}: {files} file{} in {}; listening on {addr}; links name {}",
|
||||
env!("CARGO_PKG_VERSION"),
|
||||
if files == 1 { "" } else { "s" },
|
||||
cfg.data.display(),
|
||||
public.url()
|
||||
);
|
||||
if addr.ip().is_loopback() {
|
||||
log::info!("only this machine can connect: listen on 0.0.0.0:{} (or behind a reverse proxy) to share", addr.port());
|
||||
}
|
||||
Ok(Running { hosting })
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod cli_tests;
|
||||
#[cfg(test)]
|
||||
mod server_tests;
|
||||
#[cfg(test)]
|
||||
mod store_tests;
|
||||
#[cfg(test)]
|
||||
mod test_util;
|
||||
@@ -0,0 +1,40 @@
|
||||
//! `figcraft-server`: see `docs/server.md`, or run it with `--help`.
|
||||
|
||||
use std::process::ExitCode;
|
||||
|
||||
use figcraft_server::cli::{self, Command};
|
||||
|
||||
fn main() -> ExitCode {
|
||||
let args: Vec<String> = std::env::args().skip(1).collect();
|
||||
let cli = match cli::parse(&args) {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
eprintln!("figcraft-server: {e}\n(figcraft-server --help lists what it takes)");
|
||||
return ExitCode::from(2);
|
||||
}
|
||||
};
|
||||
if cli.command != Command::Serve {
|
||||
return match cli::run(&cli) {
|
||||
Ok(out) => {
|
||||
print!("{out}");
|
||||
ExitCode::SUCCESS
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("figcraft-server: {e}");
|
||||
ExitCode::FAILURE
|
||||
}
|
||||
};
|
||||
}
|
||||
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).target(env_logger::Target::Stdout).init();
|
||||
match figcraft_server::start(&cli.config) {
|
||||
// Serves until the process is stopped: what was confirmed is on disk already, and clients
|
||||
// reconnect once it runs again.
|
||||
Ok(_running) => loop {
|
||||
std::thread::park();
|
||||
},
|
||||
Err(e) => {
|
||||
log::error!("{e}");
|
||||
ExitCode::FAILURE
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,280 @@
|
||||
//! The server end to end over loopback, in-process with a temporary data directory: an app
|
||||
//! publishes its file and syncs through the server, others join with the edit link and the view
|
||||
//! link, edits converge and the viewer's are refused (by its app, and by the server for a client
|
||||
//! that skips the check), the server restarts mid-session and everyone resumes without losing a
|
||||
//! confirmed edit, revoked links stop working at once, and fuzzed frames don't hurt it.
|
||||
|
||||
use std::net::{SocketAddr, TcpStream};
|
||||
use std::path::Path;
|
||||
use std::time::Duration;
|
||||
|
||||
use figcraft_collab::message::{self, Access, Admin, AdminReply, ClientMsg, ServerMsg};
|
||||
use figcraft_collab::sim::Rng;
|
||||
use figcraft_collab::{Client, VIEW_ONLY};
|
||||
use figcraft_collab_net::{Endpoint, Invite, NetLimits, manage};
|
||||
use figcraft_doc::NodeId;
|
||||
use figcraft_engine::Session;
|
||||
use figcraft_engine::live::{Published, Status};
|
||||
use serde_json::{Value, json};
|
||||
use tungstenite::{Bytes, Message, WebSocket};
|
||||
|
||||
use crate::test_util::{TempDir, config, pump, same, session};
|
||||
|
||||
fn key(dir: &Path) -> String {
|
||||
std::fs::read_to_string(dir.join(crate::store::KEY)).unwrap().trim().to_string()
|
||||
}
|
||||
|
||||
fn id_of(v: &Value) -> NodeId {
|
||||
NodeId(v["id"].as_u64().unwrap())
|
||||
}
|
||||
|
||||
fn live(s: &Session) -> bool {
|
||||
s.live.status == Status::Live && s.live.syncing()
|
||||
}
|
||||
|
||||
/// `a` publishes its open file to the server at `addr`; the links.
|
||||
fn publish(a: &mut Session, dir: &Path, addr: SocketAddr) -> Published {
|
||||
let r = a.execute("collab.publish", &json!({"server": format!("ws://{addr}/?key={}", key(dir)), "name": "Team file"})).unwrap();
|
||||
assert_eq!(r["status"], "connecting");
|
||||
assert_eq!(r["readOnly"], true, "read-only while it is on its way");
|
||||
pump(&mut [a], "the publish", |s| live(s[0]) && s[0].live.published.is_some());
|
||||
a.live.published.clone().unwrap()
|
||||
}
|
||||
|
||||
fn join(link: &str, name: &str) -> Session {
|
||||
let mut s = session(name);
|
||||
s.execute("collab.join", &json!({"link": link})).unwrap();
|
||||
s
|
||||
}
|
||||
|
||||
/// A bare client: hello, then what it's told.
|
||||
struct Raw(WebSocket<TcpStream>);
|
||||
|
||||
impl Raw {
|
||||
fn connect(addr: SocketAddr) -> Raw {
|
||||
let s = TcpStream::connect(addr).unwrap();
|
||||
s.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
|
||||
Raw(tungstenite::client::client(format!("ws://{addr}/"), s).unwrap().0)
|
||||
}
|
||||
|
||||
fn send(&mut self, m: &ClientMsg) {
|
||||
self.0.send(Message::Binary(Bytes::from(message::encode(m).unwrap()))).unwrap();
|
||||
}
|
||||
|
||||
fn next(&mut self) -> ServerMsg {
|
||||
loop {
|
||||
if let Message::Binary(b) = self.0.read().unwrap() {
|
||||
return message::decode(&b, usize::MAX).unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn publish_then_edit_and_view_together() {
|
||||
let dir = TempDir::new("share");
|
||||
let mut server = crate::start(&config(dir.path())).unwrap();
|
||||
let mut a = session("Ada");
|
||||
a.execute("shape.rectangle", &json!({"x": 0, "y": 0, "width": 80, "height": 80})).unwrap();
|
||||
let addr = server.addr();
|
||||
let r = a.execute("collab.publish", &json!({"server": format!("ws://{addr}/?key={}", key(dir.path()))})).unwrap();
|
||||
assert_eq!(r["status"], "connecting");
|
||||
// Nothing changes it until it is on the server.
|
||||
let e = a.execute("shape.ellipse", &json!({"x": 0, "y": 0, "width": 10, "height": 10})).unwrap_err().to_string();
|
||||
assert!(e.contains("being published"), "{e}");
|
||||
pump(&mut [&mut a], "the publish", |s| live(s[0]) && s[0].live.published.is_some());
|
||||
let p = a.live.published.clone().unwrap();
|
||||
assert_eq!(a.docs.len(), 1, "the open tab syncs; no new tab");
|
||||
assert_eq!(a.live_summary()["readOnly"], false);
|
||||
assert!(p.edit.starts_with("figcraft://join?host=127.0.0.1:") && p.view.starts_with("figcraft://join?host=127.0.0.1:"));
|
||||
assert_ne!(p.edit, p.view);
|
||||
// Bo joins with the edit link, Cy with the view link.
|
||||
let mut b = join(&p.edit, "Bo");
|
||||
let mut c = join(&p.view, "Cy");
|
||||
pump(&mut [&mut a, &mut b, &mut c], "everyone in", |s| s.iter().all(|x| live(x)) && s[0].live.peers.len() == 2);
|
||||
assert_eq!(b.live_summary()["access"], "edit");
|
||||
assert_eq!(c.live_summary()["access"], "view");
|
||||
assert_eq!(c.live_summary()["readOnly"], true);
|
||||
assert!(same(a.doc().unwrap(), b.doc().unwrap()) && same(a.doc().unwrap(), c.doc().unwrap()));
|
||||
// Ada and Bo edit; everyone converges, Cy included.
|
||||
let rect = id_of(&a.execute("shape.rectangle", &json!({"x": 200, "y": 0, "width": 40, "height": 40})).unwrap());
|
||||
pump(&mut [&mut a, &mut b, &mut c], "Ada's rectangle at Bo", |s| s[1].doc().unwrap().contains(rect));
|
||||
b.execute("node.set", &json!({"ids": [rect.0], "props": {"name": "Card"}})).unwrap();
|
||||
pump(&mut [&mut a, &mut b, &mut c], "everyone agreeing", |s| {
|
||||
let (ad, bd, cd) = (s[0].doc().unwrap(), s[1].doc().unwrap(), s[2].doc().unwrap());
|
||||
same(ad, bd) && same(ad, cd) && cd.node(rect).is_some_and(|n| n.name == "Card")
|
||||
});
|
||||
// Cy can't change anything: commands and gestures are refused before anything is sent.
|
||||
let before = c.doc().unwrap().clone();
|
||||
let e = c.execute("shape.rectangle", &json!({"x": 0, "y": 0, "width": 5, "height": 5})).unwrap_err().to_string();
|
||||
assert!(e.contains("View only"), "{e}");
|
||||
assert!(c.execute("node.set", &json!({"ids": [rect.0], "props": {"name": "Mine"}})).is_err());
|
||||
assert!(c.begin_interaction("Move").is_err());
|
||||
assert!(same(&before, c.doc().unwrap()));
|
||||
// Cy is still seen: presence goes both ways.
|
||||
let cy = c.live.info.as_ref().unwrap().me;
|
||||
c.execute("collab.cursor", &json!({"x": 12, "y": 34})).unwrap();
|
||||
pump(&mut [&mut a, &mut b, &mut c], "Cy's cursor at Ada", |s| s[0].live.presence.get(&cy).and_then(|p| p.cursor).is_some_and(|c| c.1 == 12.0));
|
||||
// A client that skips the app's check is refused by the server.
|
||||
let view = Invite::parse(&p.view).unwrap();
|
||||
let mut raw = Raw::connect(addr);
|
||||
raw.send(&Client::hello(&view.file, &view.token, "Raw", ""));
|
||||
let welcome = loop {
|
||||
if let w @ ServerMsg::Welcome { .. } = raw.next() {
|
||||
break w;
|
||||
}
|
||||
};
|
||||
let mut client = Client::join(welcome).unwrap();
|
||||
assert_eq!(client.access(), Access::View);
|
||||
let mut d = client.document().clone();
|
||||
d.node_mut(rect).unwrap().name = "Sneaky".into();
|
||||
raw.send(&client.edit(&d).unwrap().unwrap());
|
||||
loop {
|
||||
if let ServerMsg::Reject { reason, .. } = raw.next() {
|
||||
assert_eq!(reason, VIEW_ONLY);
|
||||
break;
|
||||
}
|
||||
}
|
||||
pump(&mut [&mut a], "nothing changed", |s| s[0].doc().unwrap().node(rect).is_some_and(|n| n.name == "Card"));
|
||||
// Leaving keeps a local copy.
|
||||
c.execute("collab.leave", &json!({})).unwrap();
|
||||
assert_eq!(c.live.status, Status::Off);
|
||||
assert!(c.doc().unwrap().contains(rect));
|
||||
assert_eq!(c.live_summary()["readOnly"], false, "the copy is yours to edit");
|
||||
c.execute("shape.rectangle", &json!({"x": 0, "y": 0, "width": 5, "height": 5})).unwrap();
|
||||
server.stop();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_restart_mid_session_loses_no_confirmed_edit() {
|
||||
let dir = TempDir::new("restart");
|
||||
let mut cfg = config(dir.path());
|
||||
let mut server = crate::start(&cfg).unwrap();
|
||||
let addr = server.addr();
|
||||
let mut a = session("Ada");
|
||||
let p = publish(&mut a, dir.path(), addr);
|
||||
let mut b = join(&p.edit, "Bo");
|
||||
pump(&mut [&mut a, &mut b], "Bo in", |s| live(s[1]) && s[0].live.peers.len() == 1);
|
||||
let rect = id_of(&a.execute("shape.rectangle", &json!({"x": 10, "y": 10, "width": 40, "height": 40})).unwrap());
|
||||
pump(&mut [&mut a, &mut b], "the rectangle confirmed", |s| s[1].doc().unwrap().contains(rect));
|
||||
let (ada, bo) = (a.live.info.as_ref().unwrap().me, b.live.info.as_ref().unwrap().me);
|
||||
// The server stops: both reconnect, and what Ada makes meanwhile waits.
|
||||
server.stop();
|
||||
pump(&mut [&mut a, &mut b], "the drop", |s| s.iter().all(|x| matches!(x.live.status, Status::Reconnecting { .. })));
|
||||
let offline = id_of(&a.execute("shape.ellipse", &json!({"x": 100, "y": 10, "width": 30, "height": 30})).unwrap());
|
||||
cfg.listen = addr.to_string();
|
||||
let mut server = start_again(&cfg);
|
||||
pump(&mut [&mut a, &mut b], "everyone back with everything", |s| {
|
||||
s.iter().all(|x| live(x)) && s[1].doc().unwrap().contains(offline) && s[1].doc().unwrap().contains(rect)
|
||||
});
|
||||
assert_eq!(a.live.info.as_ref().unwrap().me, ada, "Ada resumed as herself");
|
||||
assert_eq!(b.live.info.as_ref().unwrap().me, bo, "so did Bo");
|
||||
assert!(same(a.doc().unwrap(), b.doc().unwrap()));
|
||||
// Someone new gets it all from the restarted server.
|
||||
let mut c = join(&p.view, "Cy");
|
||||
pump(&mut [&mut a, &mut b, &mut c], "Cy in", |s| live(s[2]));
|
||||
assert!(same(a.doc().unwrap(), c.doc().unwrap()));
|
||||
server.stop();
|
||||
}
|
||||
|
||||
/// Start a server on `cfg.listen` again (the port may take a moment to come free).
|
||||
fn start_again(cfg: &crate::config::Config) -> crate::Running {
|
||||
let start = std::time::Instant::now();
|
||||
loop {
|
||||
match crate::start(cfg) {
|
||||
Ok(s) => return s,
|
||||
Err(e) if start.elapsed() < Duration::from_secs(10) && e.contains("can't listen") => std::thread::sleep(Duration::from_millis(50)),
|
||||
Err(e) => panic!("{e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn revoked_links_stop_working_at_once() {
|
||||
let dir = TempDir::new("revoke");
|
||||
let mut server = crate::start(&config(dir.path())).unwrap();
|
||||
let addr = server.addr();
|
||||
let mut a = session("Ada");
|
||||
let p = publish(&mut a, dir.path(), addr);
|
||||
let file = Invite::parse(&p.edit).unwrap().file;
|
||||
let endpoint = Endpoint::plain(addr.to_string());
|
||||
let ask = |req: Admin| manage::request(&endpoint, &key(dir.path()), vec![req], &NetLimits::default(), Duration::from_secs(5)).unwrap().remove(0);
|
||||
let AdminReply::Link(review) = ask(Admin::CreateLink { file: file.clone(), access: Access::View, label: "Review".into() }) else { panic!() };
|
||||
let link = Invite { server: endpoint.clone(), file: file.clone(), token: review.token.clone() }.link();
|
||||
let mut b = join(&p.edit, "Bo");
|
||||
let mut c = join(&link, "Cy");
|
||||
pump(&mut [&mut a, &mut b, &mut c], "everyone in", |s| s.iter().all(|x| live(x)));
|
||||
let AdminReply::Files(files) = ask(Admin::Files) else { panic!() };
|
||||
assert_eq!((files.len(), files[0].people), (1, 3));
|
||||
// Revoked: Cy is out at once, the others stay.
|
||||
assert_eq!(ask(Admin::Revoke { file: file.clone(), link: review.link.id.clone() }), AdminReply::Done);
|
||||
pump(&mut [&mut a, &mut b, &mut c], "Cy out", |s| {
|
||||
matches!(&s[2].live.status, Status::Failed { reason } if reason.contains("revoked")) && s[0].live.peers.len() == 1
|
||||
});
|
||||
assert!(live(&b));
|
||||
// The link no longer joins.
|
||||
c.execute("collab.join", &json!({"link": link})).unwrap();
|
||||
pump(&mut [&mut c], "the refusal", |s| matches!(&s[0].live.status, Status::Failed { reason } if reason.contains("isn't valid")));
|
||||
// Managing needs the key.
|
||||
let e = manage::request(&endpoint, "WrongWrongWrongWrong00", vec![Admin::Files], &NetLimits::default(), Duration::from_secs(5)).unwrap_err();
|
||||
assert!(e.contains("wrong server key"), "{e}");
|
||||
// Removing the file ends it for everyone in it.
|
||||
assert_eq!(ask(Admin::Remove { file: file.clone() }), AdminReply::Done);
|
||||
pump(&mut [&mut a, &mut b], "everyone out", |s| {
|
||||
s.iter().all(|x| matches!(&x.live.status, Status::Failed { reason } if reason.contains("removed")))
|
||||
});
|
||||
server.stop();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fuzzed_frames_dont_hurt_the_server() {
|
||||
let dir = TempDir::new("fuzz");
|
||||
let mut server = crate::start(&config(dir.path())).unwrap();
|
||||
let addr = server.addr();
|
||||
let mut a = session("Ada");
|
||||
let p = publish(&mut a, dir.path(), addr);
|
||||
let edit = Invite::parse(&p.edit).unwrap();
|
||||
let valid: Vec<Vec<u8>> = [
|
||||
Client::hello(&edit.file, &edit.token, "Fuzz", ""),
|
||||
ClientMsg::Manage(message::Manage { protocol: message::PROTOCOL, key: key(dir.path()), client_version: String::new() }),
|
||||
ClientMsg::Admin(Admin::Links { file: "0000000000000000".into() }),
|
||||
ClientMsg::Ping,
|
||||
]
|
||||
.iter()
|
||||
.map(|m| message::encode(m).unwrap())
|
||||
.collect();
|
||||
let mut rng = Rng::new(11);
|
||||
for round in 0..80 {
|
||||
let Ok(s) = TcpStream::connect(addr) else { continue };
|
||||
let _ = s.set_read_timeout(Some(Duration::from_millis(200)));
|
||||
let Ok((mut ws, _)) = tungstenite::client::client(format!("ws://{addr}/"), s) else { continue };
|
||||
for _ in 0..1 + rng.below(3) {
|
||||
let mut bytes = rng.pick(&valid).cloned().unwrap_or_default();
|
||||
if round % 5 == 0 {
|
||||
bytes = vec![0x5A; 100_000];
|
||||
}
|
||||
for _ in 0..1 + rng.below(4) {
|
||||
if bytes.is_empty() {
|
||||
break;
|
||||
}
|
||||
let i = rng.below(bytes.len());
|
||||
match rng.below(3) {
|
||||
0 => bytes[i] = rng.next_u64() as u8,
|
||||
1 => bytes.insert(i, rng.next_u64() as u8),
|
||||
_ => bytes.truncate(i.max(1)),
|
||||
}
|
||||
}
|
||||
let frame =
|
||||
if round % 7 == 0 { Message::Text(String::from_utf8_lossy(&bytes).into_owned().into()) } else { Message::Binary(Bytes::from(bytes)) };
|
||||
if ws.send(frame).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
let _ = ws.read();
|
||||
}
|
||||
// Still serving, the file intact.
|
||||
let mut b = join(&p.edit, "Bo");
|
||||
pump(&mut [&mut a, &mut b], "Bo in after the noise", |s| live(s[1]) && live(s[0]));
|
||||
assert!(same(a.doc().unwrap(), b.doc().unwrap()));
|
||||
server.stop();
|
||||
}
|
||||
@@ -0,0 +1,615 @@
|
||||
//! The data directory: the files a server keeps, their share links and the server key, and the
|
||||
//! [`Backend`] the hub serves them through.
|
||||
//!
|
||||
//! ```text
|
||||
//! <data>/LOCK held while a server (or a command changing the directory) uses it
|
||||
//! <data>/server.json where the running server listens (for the command line)
|
||||
//! <data>/server.key the server key: managing files (publish, links, removal)
|
||||
//! <data>/files/<id>/meta.json name, created, updated, sequence number
|
||||
//! <data>/files/<id>/links.json share links: id, access, label, created, SHA-256 of the token
|
||||
//! <data>/files/<id>/snapshot.bin the session state at the last compaction
|
||||
//! <data>/files/<id>/log.bin what changed since (see `disk`)
|
||||
//! ```
|
||||
//!
|
||||
//! An open file appends its server's journal to its log; the hub has the store make the appends
|
||||
//! durable before anyone hears of them. After [`StoreConfig::compact_after`] entries or
|
||||
//! [`StoreConfig::compact_bytes`] of log, when everyone has left, and when the server stops, the
|
||||
//! state is written as the next snapshot and the log starts over. Link tokens are stored only as
|
||||
//! hashes and compared in constant time. Whatever is damaged at startup is skipped and reported,
|
||||
//! never deleted.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::fs::{self, File, OpenOptions, TryLockError};
|
||||
use std::io::Write;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use figcraft_collab::message::{Access, Admin, AdminReply, FileInfo, LinkInfo, NewLink};
|
||||
use figcraft_collab::{Change, Replica, Server, same_secret};
|
||||
use figcraft_collab_net::hub::{Admission, Backend, Closing, Handled, OpenFile};
|
||||
use figcraft_collab_net::token;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use crate::disk::{self, LINKS, LOG, LOG_HEADER, META, SNAPSHOT};
|
||||
|
||||
pub const LOCK: &str = "LOCK";
|
||||
pub const KEY: &str = "server.key";
|
||||
pub const FILES: &str = "files";
|
||||
|
||||
/// What the store enforces.
|
||||
#[derive(Clone, Copy, Debug, PartialEq)]
|
||||
pub struct StoreConfig {
|
||||
pub max_files: usize,
|
||||
pub max_file_bytes: u64,
|
||||
pub max_links: usize,
|
||||
pub compact_after: usize,
|
||||
pub compact_bytes: u64,
|
||||
pub core: figcraft_collab::Limits,
|
||||
}
|
||||
|
||||
/// A file's description (`meta.json`).
|
||||
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
|
||||
struct Meta {
|
||||
name: String,
|
||||
created: u64,
|
||||
updated: u64,
|
||||
#[serde(default)]
|
||||
seq: u64,
|
||||
}
|
||||
|
||||
/// A share link (`links.json`): the token itself is never stored.
|
||||
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
|
||||
struct Link {
|
||||
id: String,
|
||||
access: Access,
|
||||
label: String,
|
||||
created: u64,
|
||||
/// SHA-256 of the token, hex.
|
||||
hash: String,
|
||||
}
|
||||
|
||||
impl Link {
|
||||
fn info(&self, file: &str) -> LinkInfo {
|
||||
LinkInfo { id: self.id.clone(), file: file.to_string(), access: self.access, label: self.label.clone(), created: self.created }
|
||||
}
|
||||
}
|
||||
|
||||
/// An open file's log.
|
||||
struct Log {
|
||||
generation: u64,
|
||||
file: File,
|
||||
/// Records not written yet (written and synced at the next flush).
|
||||
pending: Vec<u8>,
|
||||
/// Entries and bytes since the last compaction.
|
||||
entries: usize,
|
||||
bytes: u64,
|
||||
/// Changed since the last compaction.
|
||||
dirty: bool,
|
||||
}
|
||||
|
||||
struct Entry {
|
||||
meta: Meta,
|
||||
links: Vec<Link>,
|
||||
/// Open: its log.
|
||||
log: Option<Log>,
|
||||
}
|
||||
|
||||
/// The data directory in use (it holds the directory's lock until dropped).
|
||||
pub struct Store {
|
||||
root: PathBuf,
|
||||
cfg: StoreConfig,
|
||||
key: String,
|
||||
files: BTreeMap<String, Entry>,
|
||||
/// What was skipped at startup or found damaged since, for the operator.
|
||||
problems: Vec<String>,
|
||||
_lock: File,
|
||||
}
|
||||
|
||||
/// Seconds since 1970.
|
||||
pub fn now() -> u64 {
|
||||
SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| d.as_secs())
|
||||
}
|
||||
|
||||
/// SHA-256 of a token, hex.
|
||||
pub fn hash(token: &str) -> String {
|
||||
Sha256::digest(token.as_bytes()).iter().map(|b| format!("{b:02x}")).collect()
|
||||
}
|
||||
|
||||
/// A file or link id as this store makes them (and nothing a path could misuse).
|
||||
fn valid_id(s: &str) -> bool {
|
||||
(1..=64).contains(&s.len()) && s.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
|
||||
}
|
||||
|
||||
/// A name others may see: trimmed, no control characters, at most 256 characters.
|
||||
fn clean_name(s: &str) -> String {
|
||||
s.trim().chars().filter(|c| !c.is_control()).take(256).collect()
|
||||
}
|
||||
|
||||
/// Take the data directory's lock: `None` when someone else holds it.
|
||||
pub fn lock(root: &Path) -> Result<Option<File>, String> {
|
||||
fs::create_dir_all(root).map_err(|e| format!("can't create the data directory {}: {e}", root.display()))?;
|
||||
let f = OpenOptions::new()
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.write(true)
|
||||
.open(root.join(LOCK))
|
||||
.map_err(|e| format!("can't open {}: {e}", root.join(LOCK).display()))?;
|
||||
match f.try_lock() {
|
||||
Ok(()) => Ok(Some(f)),
|
||||
Err(TryLockError::WouldBlock) => Ok(None),
|
||||
Err(TryLockError::Error(e)) => Err(format!("can't lock the data directory: {e}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// Read the server key, or make one.
|
||||
fn load_key(root: &Path) -> Result<String, String> {
|
||||
let path = root.join(KEY);
|
||||
match fs::read_to_string(&path) {
|
||||
Ok(k) if valid_id(k.trim()) && k.trim().len() >= 16 => Ok(k.trim().to_string()),
|
||||
Ok(_) => Err(format!("{} isn't a server key (delete it to make a new one)", path.display())),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => write_key(root),
|
||||
Err(e) => Err(format!("can't read {}: {e}", path.display())),
|
||||
}
|
||||
}
|
||||
|
||||
/// Make a new server key (only the operator's account can read it on Unix).
|
||||
fn write_key(root: &Path) -> Result<String, String> {
|
||||
let key = token::secret()?;
|
||||
let path = root.join(KEY);
|
||||
disk::write_atomic(&path, format!("{key}\n").as_bytes()).map_err(|e| format!("can't write {}: {e}", path.display()))?;
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
let _ = fs::set_permissions(&path, fs::Permissions::from_mode(0o600));
|
||||
}
|
||||
Ok(key)
|
||||
}
|
||||
|
||||
impl Store {
|
||||
/// Use the data directory `root` (made if missing). Fails when another server or command holds
|
||||
/// it. Files that can't be read are skipped and listed in [`problems`](Self::problems).
|
||||
pub fn open(root: &Path, cfg: StoreConfig) -> Result<Store, String> {
|
||||
Self::try_open(root, cfg)?.ok_or_else(|| format!("another figcraft-server (or a command) is using {}", root.display()))
|
||||
}
|
||||
|
||||
/// [`open`](Self::open), or `None` when someone else holds the directory.
|
||||
pub fn try_open(root: &Path, cfg: StoreConfig) -> Result<Option<Store>, String> {
|
||||
let Some(lock) = lock(root)? else { return Ok(None) };
|
||||
let key = load_key(root)?;
|
||||
let dir = root.join(FILES);
|
||||
fs::create_dir_all(&dir).map_err(|e| format!("can't create {}: {e}", dir.display()))?;
|
||||
let mut store = Store { root: root.to_path_buf(), cfg, key, files: BTreeMap::new(), problems: Vec::new(), _lock: lock };
|
||||
store.scan(&dir);
|
||||
Ok(Some(store))
|
||||
}
|
||||
|
||||
fn scan(&mut self, dir: &Path) {
|
||||
let entries = match fs::read_dir(dir) {
|
||||
Ok(e) => e,
|
||||
Err(e) => return self.problems.push(format!("can't list {}: {e}", dir.display())),
|
||||
};
|
||||
let mut ids: Vec<(String, PathBuf)> = entries.flatten().map(|e| (e.file_name().to_string_lossy().into_owned(), e.path())).collect();
|
||||
ids.sort();
|
||||
for (id, path) in ids {
|
||||
if !path.is_dir() {
|
||||
continue;
|
||||
}
|
||||
if !valid_id(&id) {
|
||||
self.problems.push(format!("skipped {}: not a file folder of this server", path.display()));
|
||||
continue;
|
||||
}
|
||||
if self.files.len() >= self.cfg.max_files {
|
||||
self.problems.push(format!("skipped file {id}: over the {}-file cap", self.cfg.max_files));
|
||||
continue;
|
||||
}
|
||||
match self.check(&path) {
|
||||
Ok(entry) => {
|
||||
self.files.insert(id, entry);
|
||||
}
|
||||
Err(why) => self.problems.push(format!("skipped file {id}: {why}")),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A file folder's description and links, after checking its snapshot and log. Leftover
|
||||
/// temporary files (writes cut short) are removed.
|
||||
fn check(&self, path: &Path) -> Result<Entry, String> {
|
||||
for name in [SNAPSHOT, LOG, META, LINKS] {
|
||||
let tmp = disk::temp_of(&path.join(name));
|
||||
if tmp.exists() {
|
||||
let _ = fs::remove_file(&tmp);
|
||||
}
|
||||
}
|
||||
let meta: Meta = serde_json::from_slice(&disk::read_capped(&path.join(META), 1 << 20)?).map_err(|e| format!("{META} is damaged ({e})"))?;
|
||||
let links: Vec<Link> =
|
||||
serde_json::from_slice(&disk::read_capped(&path.join(LINKS), 16 << 20)?).map_err(|e| format!("{LINKS} is damaged ({e})"))?;
|
||||
let generation = disk::check_snapshot(&disk::read_capped(&path.join(SNAPSHOT), self.cfg.max_file_bytes)?)?;
|
||||
// A damaged log only loses its end; it is reported when the file opens.
|
||||
if let Ok(bytes) = disk::read_capped(&path.join(LOG), self.cfg.max_file_bytes)
|
||||
&& let Ok(log) = disk::read_log(&bytes)
|
||||
&& log.generation != generation
|
||||
{
|
||||
log::info!("file at {}: its log predates the last compaction; it is ignored", path.display());
|
||||
}
|
||||
Ok(Entry { meta, links, log: None })
|
||||
}
|
||||
|
||||
/// The server key (for managing files).
|
||||
pub fn key(&self) -> &str {
|
||||
&self.key
|
||||
}
|
||||
|
||||
/// Replace the server key; the new one.
|
||||
pub fn rotate_key(&mut self) -> Result<String, String> {
|
||||
self.key = write_key(&self.root)?;
|
||||
Ok(self.key.clone())
|
||||
}
|
||||
|
||||
/// What was skipped or found damaged.
|
||||
pub fn problems(&self) -> &[String] {
|
||||
&self.problems
|
||||
}
|
||||
|
||||
/// How many files are kept.
|
||||
pub fn len(&self) -> usize {
|
||||
self.files.len()
|
||||
}
|
||||
|
||||
/// No files are kept.
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.files.is_empty()
|
||||
}
|
||||
|
||||
fn dir(&self, id: &str) -> PathBuf {
|
||||
self.root.join(FILES).join(id)
|
||||
}
|
||||
|
||||
fn entry(&mut self, id: &str) -> Result<&mut Entry, String> {
|
||||
self.files.get_mut(id).ok_or_else(|| "this file isn't on the server (it may have been removed)".to_string())
|
||||
}
|
||||
|
||||
fn info(&self, id: &str, e: &Entry, open: Option<OpenFile>) -> FileInfo {
|
||||
let dir = self.dir(id);
|
||||
let bytes = [SNAPSHOT, LOG].iter().filter_map(|n| fs::metadata(dir.join(n)).ok()).map(|m| m.len()).sum();
|
||||
FileInfo {
|
||||
id: id.to_string(),
|
||||
name: e.meta.name.clone(),
|
||||
created: e.meta.created,
|
||||
updated: e.meta.updated,
|
||||
seq: open.map_or(e.meta.seq, |o| o.seq),
|
||||
bytes,
|
||||
people: open.map_or(0, |o| o.people),
|
||||
}
|
||||
}
|
||||
|
||||
/// Keep a document (its snapshot) as a new file with an edit link and a view link.
|
||||
pub fn publish(&mut self, name: &str, snapshot: &[Change]) -> Result<(FileInfo, NewLink, NewLink), String> {
|
||||
if self.files.len() >= self.cfg.max_files {
|
||||
return Err(format!("the server keeps at most {} files", self.cfg.max_files));
|
||||
}
|
||||
let replica = Replica::from_snapshot(snapshot).map_err(|e| format!("the file couldn't be read: {e}"))?;
|
||||
if replica.document().node_count() > self.cfg.core.max_nodes {
|
||||
return Err(format!("the file has more than {} layers", self.cfg.core.max_nodes));
|
||||
}
|
||||
let mut doc = replica.document().clone();
|
||||
let name = match clean_name(name) {
|
||||
n if n.is_empty() => doc.name.clone(),
|
||||
n => n,
|
||||
};
|
||||
doc.name = name.clone();
|
||||
let server = Server::new(doc, "", self.cfg.core);
|
||||
let state = server.state().map_err(|e| e.to_string())?;
|
||||
let bytes = disk::encode_snapshot(1, &state)?;
|
||||
if bytes.len() as u64 > self.cfg.max_file_bytes {
|
||||
return Err(format!("the file is {} MiB; this server keeps files up to {} MiB", bytes.len() >> 20, self.cfg.max_file_bytes >> 20));
|
||||
}
|
||||
let id = loop {
|
||||
let id = token::file_id()?;
|
||||
if !self.files.contains_key(&id) && !self.dir(&id).exists() {
|
||||
break id;
|
||||
}
|
||||
};
|
||||
let t = now();
|
||||
let (edit, edit_token) = new_link(Access::Edit, "Published (edit)", t)?;
|
||||
let (view, view_token) = new_link(Access::View, "Published (view)", t)?;
|
||||
let meta = Meta { name, created: t, updated: t, seq: 0 };
|
||||
let dir = self.dir(&id);
|
||||
let written = fs::create_dir_all(&dir)
|
||||
.map_err(|e| e.to_string())
|
||||
.and_then(|()| disk::write_atomic(&dir.join(SNAPSHOT), &bytes).map_err(|e| e.to_string()))
|
||||
.and_then(|()| disk::write_atomic(&dir.join(LOG), &disk::log_header(1)).map_err(|e| e.to_string()))
|
||||
.and_then(|()| write_json(&dir.join(LINKS), &[edit.clone(), view.clone()]))
|
||||
// Last: a folder without its description is an upload cut short.
|
||||
.and_then(|()| write_json(&dir.join(META), &meta));
|
||||
if let Err(e) = written {
|
||||
let _ = fs::remove_dir_all(&dir);
|
||||
return Err(format!("couldn't keep the file: {e}"));
|
||||
}
|
||||
let entry = Entry { meta, links: vec![edit.clone(), view.clone()], log: None };
|
||||
let info = self.info(&id, &entry, None);
|
||||
self.files.insert(id.clone(), entry);
|
||||
log::info!("published file {id} ({} bytes)", bytes.len());
|
||||
Ok((info, NewLink { link: edit.info(&id), token: edit_token }, NewLink { link: view.info(&id), token: view_token }))
|
||||
}
|
||||
|
||||
/// The files kept.
|
||||
pub fn list(&self, open: &dyn Fn(&str) -> Option<OpenFile>) -> Vec<FileInfo> {
|
||||
self.files.iter().map(|(id, e)| self.info(id, e, open(id))).collect()
|
||||
}
|
||||
|
||||
/// A file's links.
|
||||
pub fn links(&mut self, file: &str) -> Result<Vec<LinkInfo>, String> {
|
||||
Ok(self.entry(file)?.links.iter().map(|l| l.info(file)).collect())
|
||||
}
|
||||
|
||||
/// A new link to a file.
|
||||
pub fn create_link(&mut self, file: &str, access: Access, label: &str) -> Result<NewLink, String> {
|
||||
let max = self.cfg.max_links;
|
||||
let path = self.dir(file).join(LINKS);
|
||||
let entry = self.entry(file)?;
|
||||
if entry.links.len() >= max {
|
||||
return Err(format!("a file has at most {max} links: revoke one first"));
|
||||
}
|
||||
let (link, token) = new_link(access, &clean_name(label), now())?;
|
||||
entry.links.push(link.clone());
|
||||
if let Err(e) = write_json(&path, &entry.links) {
|
||||
entry.links.pop();
|
||||
return Err(format!("couldn't keep the link: {e}"));
|
||||
}
|
||||
Ok(NewLink { link: link.info(file), token })
|
||||
}
|
||||
|
||||
/// Revoke a link.
|
||||
pub fn revoke(&mut self, file: &str, link: &str) -> Result<(), String> {
|
||||
let path = self.dir(file).join(LINKS);
|
||||
let entry = self.entry(file)?;
|
||||
let at = entry.links.iter().position(|l| l.id == link).ok_or("that file has no such link")?;
|
||||
let gone = entry.links.remove(at);
|
||||
if let Err(e) = write_json(&path, &entry.links) {
|
||||
entry.links.insert(at, gone);
|
||||
return Err(format!("couldn't revoke the link: {e}"));
|
||||
}
|
||||
log::info!("revoked link {link} of file {file}");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Remove a file and its links for good. `Err` when its folder couldn't be deleted (it is no
|
||||
/// longer served either way).
|
||||
pub fn remove(&mut self, file: &str) -> Result<(), String> {
|
||||
// Closed first: an open log would keep the folder from being deleted on Windows.
|
||||
let entry = self.files.remove(file).ok_or("this file isn't on the server (it may have been removed)")?;
|
||||
drop(entry);
|
||||
let dir = self.dir(file);
|
||||
fs::remove_dir_all(&dir)
|
||||
.map_err(|e| format!("it is no longer served, but its folder {} couldn't be deleted ({e}); delete it by hand", dir.display()))?;
|
||||
log::info!("removed file {file}");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Load a file's session: its snapshot, then its log (a damaged end is dropped and reported).
|
||||
fn load(&mut self, file: &str) -> Result<Server, String> {
|
||||
let dir = self.dir(file);
|
||||
let (generation, state) = disk::decode_snapshot(&disk::read_capped(&dir.join(SNAPSHOT), self.cfg.max_file_bytes)?)?;
|
||||
let mut server = Server::restore(state, "", self.cfg.core).map_err(|e| format!("the snapshot can't be used: {e}"))?;
|
||||
let mut issues = Vec::new();
|
||||
let mut fresh_log = true;
|
||||
let mut kept = 0;
|
||||
match disk::read_capped(&dir.join(LOG), self.cfg.max_file_bytes).and_then(|b| disk::read_log(&b)) {
|
||||
Ok(log) if log.generation == generation => {
|
||||
fresh_log = false;
|
||||
for entry in log.entries {
|
||||
if let Err(e) = server.replay(entry) {
|
||||
issues.push(format!("file {file}: its log stops making sense after {kept} entries ({e}); the rest was dropped"));
|
||||
fresh_log = true;
|
||||
break;
|
||||
}
|
||||
kept += 1;
|
||||
}
|
||||
if let Some(why) = log.damaged {
|
||||
issues.push(format!("file {file}: its log ended with {why}; that part was dropped"));
|
||||
fresh_log = true;
|
||||
}
|
||||
}
|
||||
// From before a compaction that was cut short: the snapshot has it all.
|
||||
Ok(_) => {}
|
||||
Err(why) => issues.push(format!("file {file}: {why}; starting a new log")),
|
||||
}
|
||||
for i in issues {
|
||||
log::warn!("{i}");
|
||||
self.problems.push(i);
|
||||
}
|
||||
server.journaling(true);
|
||||
let file_handle = OpenOptions::new().append(true).open(dir.join(LOG));
|
||||
let log = match (fresh_log, file_handle) {
|
||||
(false, Ok(f)) => {
|
||||
let bytes = f.metadata().map_or(LOG_HEADER as u64, |m| m.len());
|
||||
// Replayed entries make it worth compacting once everyone has left.
|
||||
Log { generation, file: f, pending: Vec::new(), entries: kept, bytes, dirty: kept > 0 }
|
||||
}
|
||||
// A new snapshot and log of the next generation, so appends start clean.
|
||||
_ => self.rewrite(file, generation + 1, &server)?,
|
||||
};
|
||||
self.entry(file)?.log = Some(log);
|
||||
Ok(server)
|
||||
}
|
||||
|
||||
/// Write `server`'s state as snapshot `generation` and start an empty log of it.
|
||||
fn rewrite(&mut self, file: &str, generation: u64, server: &Server) -> Result<Log, String> {
|
||||
let dir = self.dir(file);
|
||||
let state = server.state().map_err(|e| e.to_string())?;
|
||||
let bytes = disk::encode_snapshot(generation, &state)?;
|
||||
if bytes.len() as u64 > self.cfg.max_file_bytes {
|
||||
log::warn!("file {file} is {} MiB, over the {} MiB cap (kept anyway)", bytes.len() >> 20, self.cfg.max_file_bytes >> 20);
|
||||
}
|
||||
disk::write_atomic(&dir.join(SNAPSHOT), &bytes).map_err(|e| format!("couldn't write the snapshot: {e}"))?;
|
||||
disk::write_atomic(&dir.join(LOG), &disk::log_header(generation)).map_err(|e| format!("couldn't start the log: {e}"))?;
|
||||
let f = OpenOptions::new().append(true).open(dir.join(LOG)).map_err(|e| format!("couldn't open the log: {e}"))?;
|
||||
if let Ok(e) = self.entry(file) {
|
||||
e.meta.seq = server.seq();
|
||||
let meta = e.meta.clone();
|
||||
if let Err(err) = write_json(&dir.join(META), &meta) {
|
||||
log::warn!("file {file}: couldn't update {META}: {err}");
|
||||
}
|
||||
}
|
||||
Ok(Log { generation, file: f, pending: Vec::new(), entries: 0, bytes: LOG_HEADER as u64, dirty: false })
|
||||
}
|
||||
|
||||
/// Compact an open file: its state as the next snapshot, an empty log.
|
||||
fn compact(&mut self, file: &str, server: &Server) {
|
||||
let Some(generation) = self.files.get(file).and_then(|e| e.log.as_ref()).map(|l| l.generation) else { return };
|
||||
match self.rewrite(file, generation + 1, server) {
|
||||
Ok(log) => {
|
||||
if let Ok(e) = self.entry(file) {
|
||||
e.log = Some(log);
|
||||
}
|
||||
log::debug!("compacted file {file}");
|
||||
}
|
||||
// The old snapshot and log still hold everything: carry on appending.
|
||||
Err(e) => log::error!("file {file}: compaction failed: {e}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Write and sync what's pending in every open log.
|
||||
fn flush_all(&mut self) {
|
||||
for (id, e) in &mut self.files {
|
||||
if let Some(log) = e.log.as_mut() {
|
||||
flush_log(id, log);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Write and sync a log's pending records (kept for the next try when that fails).
|
||||
fn flush_log(id: &str, log: &mut Log) {
|
||||
if log.pending.is_empty() {
|
||||
return;
|
||||
}
|
||||
match log.file.write_all(&log.pending).and_then(|()| log.file.sync_data()) {
|
||||
Ok(()) => {
|
||||
log.bytes += log.pending.len() as u64;
|
||||
log.pending.clear();
|
||||
}
|
||||
Err(e) => log::error!("file {id}: couldn't write its log: {e}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn write_json<T: Serialize + ?Sized>(path: &Path, value: &T) -> Result<(), String> {
|
||||
let text = serde_json::to_vec_pretty(value).map_err(|e| e.to_string())?;
|
||||
disk::write_atomic(path, &text).map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// A new link and its token.
|
||||
fn new_link(access: Access, label: &str, created: u64) -> Result<(Link, String), String> {
|
||||
let token = token::secret()?;
|
||||
Ok((Link { id: token::file_id()?, access, label: label.to_string(), created, hash: hash(&token) }, token))
|
||||
}
|
||||
|
||||
impl Backend for Store {
|
||||
fn admit(&mut self, file: &str, token: &str) -> Result<Admission, String> {
|
||||
let entry = self.entry(file)?;
|
||||
let presented = hash(token);
|
||||
// Every link is compared (in constant time), so timing doesn't tell which one matched.
|
||||
let mut found = None;
|
||||
for l in &entry.links {
|
||||
if same_secret(&presented, &l.hash) {
|
||||
found = Some(Admission { access: l.access, link: Some(l.id.clone()) });
|
||||
}
|
||||
}
|
||||
found.ok_or_else(|| "this link isn't valid (it may have been revoked)".to_string())
|
||||
}
|
||||
|
||||
fn open(&mut self, file: &str) -> Result<Server, String> {
|
||||
self.load(file).map_err(|e| {
|
||||
let why = format!("file {file} can't be opened: {e}");
|
||||
log::error!("{why}");
|
||||
self.problems.push(why);
|
||||
"this file can't be opened on the server (its data is damaged); tell the server's owner".to_string()
|
||||
})
|
||||
}
|
||||
|
||||
fn changed(&mut self, file: &str, server: &mut Server) {
|
||||
let journal = server.take_journal();
|
||||
if journal.is_empty() {
|
||||
return;
|
||||
}
|
||||
let (after, max_bytes) = (self.cfg.compact_after, self.cfg.compact_bytes);
|
||||
let Some(e) = self.files.get_mut(file) else { return };
|
||||
let Some(log) = e.log.as_mut() else { return };
|
||||
for j in &journal {
|
||||
match disk::record(j) {
|
||||
Ok(r) => log.pending.extend(r),
|
||||
Err(err) => log::error!("file {file}: a change couldn't be logged: {err}"),
|
||||
}
|
||||
}
|
||||
log.entries += journal.len();
|
||||
log.dirty = true;
|
||||
e.meta.updated = now();
|
||||
if log.entries >= after || log.bytes + log.pending.len() as u64 >= max_bytes {
|
||||
// The snapshot holds what's pending too.
|
||||
self.compact(file, server);
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(&mut self) {
|
||||
self.flush_all();
|
||||
}
|
||||
|
||||
fn idle(&mut self, file: &str, server: &mut Server) -> bool {
|
||||
let dirty = self.files.get_mut(file).and_then(|e| e.log.as_mut()).is_some_and(|log| {
|
||||
flush_log(file, log);
|
||||
log.dirty
|
||||
});
|
||||
if dirty {
|
||||
self.compact(file, server);
|
||||
}
|
||||
if let Some(e) = self.files.get_mut(file) {
|
||||
e.log = None;
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
fn stop(&mut self, open: Vec<(&str, &mut Server)>) {
|
||||
self.flush_all();
|
||||
for (file, server) in open {
|
||||
if self.files.get(file).and_then(|e| e.log.as_ref()).is_some_and(|l| l.dirty) {
|
||||
self.compact(file, server);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn manage(&mut self, key: &str) -> Result<(), String> {
|
||||
if same_secret(key, &self.key) { Ok(()) } else { Err("wrong server key".into()) }
|
||||
}
|
||||
|
||||
fn admin(&mut self, req: Admin, open: &dyn Fn(&str) -> Option<OpenFile>) -> Handled {
|
||||
let failed = |reason: String| Handled::from(AdminReply::Failed { reason });
|
||||
match req {
|
||||
Admin::Publish { name, snapshot } => match self.publish(&name, &snapshot) {
|
||||
Ok((file, edit, view)) => AdminReply::Published { file, edit, view }.into(),
|
||||
Err(e) => failed(e),
|
||||
},
|
||||
Admin::Files => AdminReply::Files(self.list(open)).into(),
|
||||
Admin::Links { file } => self.links(&file).map_or_else(failed, |l| AdminReply::Links(l).into()),
|
||||
Admin::CreateLink { file, access, label } => self.create_link(&file, access, &label).map_or_else(failed, |l| AdminReply::Link(l).into()),
|
||||
Admin::Revoke { file, link } => match self.revoke(&file, &link) {
|
||||
Ok(()) => Handled { reply: AdminReply::Done, revoked: Some((file, link)), removed: None },
|
||||
Err(e) => failed(e),
|
||||
},
|
||||
Admin::Remove { file } => {
|
||||
if !self.files.contains_key(&file) {
|
||||
return failed("this file isn't on the server (it may have been removed)".into());
|
||||
}
|
||||
let reply = match self.remove(&file) {
|
||||
Ok(()) => AdminReply::Done,
|
||||
Err(reason) => AdminReply::Failed { reason },
|
||||
};
|
||||
Handled { reply, revoked: None, removed: Some(file) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn closing(&self) -> Closing {
|
||||
Closing::Restarting
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,271 @@
|
||||
//! The data directory on its own: publishing, links and their hashes, persistence across a crash
|
||||
//! (snapshot plus log) and compaction, damaged files skipped and reported, caps, the lock, and
|
||||
//! temporary directories cleaned up.
|
||||
|
||||
use std::fs;
|
||||
|
||||
use figcraft_collab::message::{Access, ClientMsg};
|
||||
use figcraft_collab::sim::Rng;
|
||||
use figcraft_collab::{Client, Outgoing, Replica, Server};
|
||||
use figcraft_collab_net::hub::Backend;
|
||||
use figcraft_doc::{Document, NodeKind};
|
||||
|
||||
use crate::disk::{self, LINKS, LOG, META, SNAPSHOT};
|
||||
use crate::store::{FILES, Store, StoreConfig, hash};
|
||||
use crate::test_util::{TempDir, config};
|
||||
|
||||
fn cfg(dir: &std::path::Path) -> StoreConfig {
|
||||
config(dir).store()
|
||||
}
|
||||
|
||||
fn doc(name: &str) -> Document {
|
||||
let mut d = Document::new(name);
|
||||
let page = d.first_page().unwrap();
|
||||
d.create(page, NodeKind::Rectangle, "Card", 0.0, 0.0, 100.0, 60.0).unwrap();
|
||||
d.repair();
|
||||
d
|
||||
}
|
||||
|
||||
fn snapshot(d: Document) -> Vec<figcraft_collab::Change> {
|
||||
Replica::new(d).snapshot().unwrap()
|
||||
}
|
||||
|
||||
/// Publish `name`; its id and the edit and view tokens.
|
||||
fn publish(store: &mut Store, name: &str) -> (String, String, String) {
|
||||
let (info, edit, view) = store.publish(name, &snapshot(doc(name))).unwrap();
|
||||
assert_eq!(edit.link.access, Access::Edit);
|
||||
assert_eq!(view.link.access, Access::View);
|
||||
(info.id, edit.token, view.token)
|
||||
}
|
||||
|
||||
/// A client in `server` (as the hub would let it in) and what it was told.
|
||||
fn join(server: &mut Server, access: Access) -> Client {
|
||||
let ClientMsg::Hello(h) = Client::hello("f", "", "Ada", "#FF8800") else { panic!() };
|
||||
let (_, out) = server.join_as(h, access).unwrap();
|
||||
let Some(Outgoing::To(_, w)) = out.into_iter().next() else { panic!() };
|
||||
Client::join(w).unwrap()
|
||||
}
|
||||
|
||||
/// Rename the first page through `c`; the server takes the batch.
|
||||
fn rename(server: &mut Server, c: &mut Client, name: &str) {
|
||||
let mut d = c.document().clone();
|
||||
let page = d.first_page().unwrap();
|
||||
d.node_mut(page).unwrap().name = name.into();
|
||||
let m = c.edit(&d).unwrap().unwrap();
|
||||
for o in server.receive(c.id(), m) {
|
||||
if let Outgoing::All(m) | Outgoing::To(_, m) = o {
|
||||
c.receive(m, &figcraft_collab::NoRecompute).unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn files_links_and_access() {
|
||||
let dir = TempDir::new("links");
|
||||
let mut store = Store::open(dir.path(), cfg(dir.path())).unwrap();
|
||||
assert!(store.is_empty());
|
||||
let (id, edit, view) = publish(&mut store, "Café ☕");
|
||||
let files = store.list(&|_| None);
|
||||
assert_eq!(files.len(), 1);
|
||||
assert_eq!(files[0].name, "Café ☕");
|
||||
assert!(files[0].bytes > 0);
|
||||
// Each link lets in with its access; anything else doesn't.
|
||||
assert_eq!(store.admit(&id, &edit).unwrap().access, Access::Edit);
|
||||
assert_eq!(store.admit(&id, &view).unwrap().access, Access::View);
|
||||
assert!(store.admit(&id, "WrongWrongWrongWrong00").unwrap_err().contains("isn't valid"));
|
||||
assert!(store.admit("0000000000000000", &edit).unwrap_err().contains("isn't on the server"));
|
||||
// Tokens are kept only as hashes.
|
||||
let on_disk = fs::read_to_string(dir.path().join(FILES).join(&id).join(LINKS)).unwrap();
|
||||
assert!(!on_disk.contains(&edit) && !on_disk.contains(&view));
|
||||
assert!(on_disk.contains(&hash(&edit)) && on_disk.contains(&hash(&view)));
|
||||
// More links; revoking one stops it, the others keep working.
|
||||
let review = store.create_link(&id, Access::View, " Review\n").unwrap();
|
||||
assert_eq!(review.link.label, "Review");
|
||||
assert_eq!(store.links(&id).unwrap().len(), 3);
|
||||
assert_eq!(store.admit(&id, &review.token).unwrap().link.as_deref(), Some(review.link.id.as_str()));
|
||||
store.revoke(&id, &review.link.id).unwrap();
|
||||
assert!(store.admit(&id, &review.token).is_err());
|
||||
assert!(store.admit(&id, &view).is_ok());
|
||||
assert!(store.revoke(&id, &review.link.id).is_err(), "already gone");
|
||||
// Removed for good.
|
||||
store.remove(&id).unwrap();
|
||||
assert!(store.is_empty());
|
||||
assert!(!dir.path().join(FILES).join(&id).exists());
|
||||
assert!(store.admit(&id, &edit).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_file_survives_a_crash_and_compacts() {
|
||||
let dir = TempDir::new("crash");
|
||||
let c = StoreConfig { compact_after: 1000, ..cfg(dir.path()) };
|
||||
let (id, before, seq) = {
|
||||
let mut store = Store::open(dir.path(), c).unwrap();
|
||||
let (id, _, _) = publish(&mut store, "Crash");
|
||||
let mut server = store.open(&id).unwrap();
|
||||
let mut a = join(&mut server, Access::Edit);
|
||||
store.changed(&id, &mut server);
|
||||
for i in 0..5 {
|
||||
rename(&mut server, &mut a, &format!("Page {i}"));
|
||||
store.changed(&id, &mut server);
|
||||
}
|
||||
store.flush();
|
||||
// A crash: no compaction, nothing else written.
|
||||
(id, server.replica().clone(), server.seq())
|
||||
};
|
||||
let mut store = Store::open(dir.path(), c).unwrap();
|
||||
assert!(store.problems().is_empty(), "{:?}", store.problems());
|
||||
let mut server = store.open(&id).unwrap();
|
||||
assert_eq!(server.replica(), &before, "the log brought it back");
|
||||
assert_eq!(server.seq(), seq);
|
||||
// Compaction: everyone left; the snapshot has it all and the log starts over.
|
||||
assert!(store.idle(&id, &mut server));
|
||||
let snap = disk::decode_snapshot(&fs::read(dir.path().join(FILES).join(&id).join(SNAPSHOT)).unwrap()).unwrap();
|
||||
assert_eq!(snap.1.seq, seq);
|
||||
let log = disk::read_log(&fs::read(dir.path().join(FILES).join(&id).join(LOG)).unwrap()).unwrap();
|
||||
assert_eq!(log.generation, snap.0);
|
||||
assert!(log.entries.is_empty());
|
||||
assert_eq!(store.list(&|_| None)[0].seq, seq, "the description follows");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_long_log_is_compacted_while_open() {
|
||||
let dir = TempDir::new("compact");
|
||||
let c = StoreConfig { compact_after: 3, ..cfg(dir.path()) };
|
||||
let mut store = Store::open(dir.path(), c).unwrap();
|
||||
let (id, _, _) = publish(&mut store, "Busy");
|
||||
let mut server = store.open(&id).unwrap();
|
||||
let mut a = join(&mut server, Access::Edit);
|
||||
store.changed(&id, &mut server);
|
||||
let generation = || disk::check_snapshot(&fs::read(dir.path().join(FILES).join(&id).join(SNAPSHOT)).unwrap()).unwrap();
|
||||
assert_eq!(generation(), 1);
|
||||
for i in 0..3 {
|
||||
rename(&mut server, &mut a, &format!("P{i}"));
|
||||
store.changed(&id, &mut server);
|
||||
store.flush();
|
||||
}
|
||||
assert_eq!(generation(), 2, "compacted after three entries");
|
||||
rename(&mut server, &mut a, "After");
|
||||
store.changed(&id, &mut server);
|
||||
store.flush();
|
||||
drop(store);
|
||||
let mut store = Store::open(dir.path(), c).unwrap();
|
||||
let restored = store.open(&id).unwrap();
|
||||
assert_eq!(restored.replica(), server.replica());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn damaged_files_are_skipped_and_reported() {
|
||||
let dir = TempDir::new("damaged");
|
||||
let (bad_snapshot, bad_meta, cut_log, good) = {
|
||||
let mut store = Store::open(dir.path(), cfg(dir.path())).unwrap();
|
||||
let ids: Vec<String> = ["A", "B", "C", "D"].iter().map(|n| publish(&mut store, n).0).collect();
|
||||
// C gets a log with entries.
|
||||
let mut server = store.open(&ids[2]).unwrap();
|
||||
let mut c = join(&mut server, Access::Edit);
|
||||
store.changed(&ids[2], &mut server);
|
||||
rename(&mut server, &mut c, "Kept");
|
||||
store.changed(&ids[2], &mut server);
|
||||
rename(&mut server, &mut c, "Lost");
|
||||
store.changed(&ids[2], &mut server);
|
||||
store.flush();
|
||||
(ids[0].clone(), ids[1].clone(), ids[2].clone(), ids[3].clone())
|
||||
};
|
||||
let folder = |id: &str| dir.path().join(FILES).join(id);
|
||||
// A: a flipped byte in its snapshot; B: garbage description; C: its log cut mid-record;
|
||||
// D: a write that was cut short left a temporary file.
|
||||
let mut snap = fs::read(folder(&bad_snapshot).join(SNAPSHOT)).unwrap();
|
||||
let last = snap.len() - 1;
|
||||
snap[last] ^= 0xFF;
|
||||
fs::write(folder(&bad_snapshot).join(SNAPSHOT), snap).unwrap();
|
||||
fs::write(folder(&bad_meta).join(META), b"{not json").unwrap();
|
||||
let log = fs::read(folder(&cut_log).join(LOG)).unwrap();
|
||||
fs::write(folder(&cut_log).join(LOG), &log[..log.len() - 3]).unwrap();
|
||||
fs::write(disk::temp_of(&folder(&good).join(SNAPSHOT)), b"half").unwrap();
|
||||
fs::create_dir_all(dir.path().join(FILES).join("not a file id")).unwrap();
|
||||
let mut store = Store::open(dir.path(), cfg(dir.path())).unwrap();
|
||||
let problems = store.problems().join("\n");
|
||||
assert!(problems.contains(&bad_snapshot) && problems.contains("checksum"), "{problems}");
|
||||
assert!(problems.contains(&bad_meta) && problems.contains(META), "{problems}");
|
||||
assert!(problems.contains("not a file id"), "{problems}");
|
||||
assert_eq!(store.len(), 2, "C and D are served");
|
||||
// Nothing is deleted.
|
||||
assert!(folder(&bad_snapshot).exists() && folder(&bad_meta).exists());
|
||||
assert!(!disk::temp_of(&folder(&good).join(SNAPSHOT)).exists(), "the leftover is gone");
|
||||
// C opens without the cut-short record, and says so.
|
||||
let server = store.open(&cut_log).unwrap();
|
||||
let page = server.document().first_page().unwrap();
|
||||
assert_eq!(server.document().node(page).unwrap().name, "Kept");
|
||||
assert!(store.problems().iter().any(|p| p.contains(&cut_log) && p.contains("cut short")));
|
||||
// Damage after startup: the file can't be opened, with a reason for the person joining.
|
||||
fs::write(folder(&good).join(SNAPSHOT), b"FCSN garbage").unwrap();
|
||||
assert!(store.open(&good).unwrap_err().contains("can't be opened"));
|
||||
assert!(store.problems().iter().any(|p| p.contains(&good)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn caps_hold() {
|
||||
let dir = TempDir::new("caps");
|
||||
let c = StoreConfig { max_files: 1, max_links: 3, ..cfg(dir.path()) };
|
||||
let mut store = Store::open(dir.path(), c).unwrap();
|
||||
let (id, _, _) = publish(&mut store, "One");
|
||||
assert!(store.publish("Two", &snapshot(doc("Two"))).unwrap_err().contains("at most 1"));
|
||||
store.create_link(&id, Access::View, "").unwrap();
|
||||
assert!(store.create_link(&id, Access::View, "").unwrap_err().contains("at most 3"));
|
||||
let dir2 = TempDir::new("caps-size");
|
||||
let mut tiny = Store::open(dir2.path(), StoreConfig { max_file_bytes: 64, ..cfg(dir2.path()) }).unwrap();
|
||||
assert!(tiny.publish("Big", &snapshot(doc("Big"))).unwrap_err().contains("MiB"));
|
||||
assert!(tiny.publish("Junk", &[figcraft_collab::Change::Delete { id: figcraft_doc::NodeId(1) }]).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn one_server_per_data_directory() {
|
||||
let dir = TempDir::new("lock");
|
||||
let first = Store::open(dir.path(), cfg(dir.path())).unwrap();
|
||||
assert!(Store::open(dir.path(), cfg(dir.path())).err().unwrap().contains("is using"));
|
||||
assert!(Store::try_open(dir.path(), cfg(dir.path())).unwrap().is_none());
|
||||
let key = first.key().to_string();
|
||||
drop(first);
|
||||
let again = Store::open(dir.path(), cfg(dir.path())).unwrap();
|
||||
assert_eq!(again.key(), key, "the key is kept");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn temporary_directories_are_removed() {
|
||||
let path = {
|
||||
let dir = TempDir::new("cleanup");
|
||||
let mut store = Store::open(dir.path(), cfg(dir.path())).unwrap();
|
||||
publish(&mut store, "Gone");
|
||||
drop(store);
|
||||
dir.path().to_path_buf()
|
||||
};
|
||||
assert!(!path.exists());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn random_bytes_never_panic_the_readers() {
|
||||
let mut rng = Rng::new(7);
|
||||
let state = Server::new(doc("Fuzz"), "", figcraft_collab::Limits::default()).state().unwrap();
|
||||
let snap = disk::encode_snapshot(3, &state).unwrap();
|
||||
let mut log = disk::log_header(3);
|
||||
log.extend(disk::record(&figcraft_collab::Journal::Issued { client: 9 }).unwrap());
|
||||
assert_eq!(disk::decode_snapshot(&snap).unwrap().1, state);
|
||||
assert_eq!(disk::read_log(&log).unwrap().entries.len(), 1);
|
||||
for round in 0..5_000 {
|
||||
let base = if round % 2 == 0 { &snap } else { &log };
|
||||
let mut bytes = base.clone();
|
||||
for _ in 0..1 + rng.below(4) {
|
||||
if bytes.is_empty() {
|
||||
break;
|
||||
}
|
||||
let i = rng.below(bytes.len());
|
||||
match rng.below(3) {
|
||||
0 => bytes[i] = rng.next_u64() as u8,
|
||||
1 => bytes.insert(i, rng.next_u64() as u8),
|
||||
_ => bytes.truncate(i),
|
||||
}
|
||||
}
|
||||
let _ = disk::decode_snapshot(&bytes);
|
||||
let _ = disk::check_snapshot(&bytes);
|
||||
let _ = disk::read_log(&bytes);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
//! What the tests share: temporary data directories (removed when dropped), settings, and engine
|
||||
//! sessions linked over loopback.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use figcraft_collab_net::NetLink;
|
||||
use figcraft_collab_net::guest::GuestConfig;
|
||||
use figcraft_engine::Session;
|
||||
use serde_json::json;
|
||||
|
||||
use crate::config::Config;
|
||||
|
||||
/// A directory of its own under the system's temporary directory, removed with everything in it
|
||||
/// when dropped.
|
||||
pub struct TempDir(PathBuf);
|
||||
|
||||
impl TempDir {
|
||||
pub fn new(tag: &str) -> TempDir {
|
||||
static NEXT: AtomicU32 = AtomicU32::new(0);
|
||||
let nanos = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.subsec_nanos());
|
||||
let n = NEXT.fetch_add(1, Ordering::Relaxed);
|
||||
let dir = std::env::temp_dir().join("figcraft-server-tests").join(format!("{tag}-{}-{nanos}-{n}", std::process::id()));
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
TempDir(dir)
|
||||
}
|
||||
|
||||
pub fn path(&self) -> &Path {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TempDir {
|
||||
fn drop(&mut self) {
|
||||
let _ = std::fs::remove_dir_all(&self.0);
|
||||
}
|
||||
}
|
||||
|
||||
/// Settings for a server on `dir`, listening on a free loopback port.
|
||||
pub fn config(dir: &Path) -> Config {
|
||||
Config { listen: "127.0.0.1:0".into(), data: dir.to_path_buf(), ..Config::default() }
|
||||
}
|
||||
|
||||
/// Joins quickly, and keeps trying long enough to outlast a server restart.
|
||||
pub fn quick(mut c: GuestConfig) -> GuestConfig {
|
||||
c.backoff = Duration::from_millis(20);
|
||||
c.max_backoff = Duration::from_millis(200);
|
||||
c.attempts = 100;
|
||||
c.limits.hello_timeout = Duration::from_secs(5);
|
||||
c
|
||||
}
|
||||
|
||||
/// An engine session with the network link, named `name`.
|
||||
pub fn session(name: &str) -> Session {
|
||||
let mut s = Session::new();
|
||||
s.live.install(Box::new(NetLink::new(Arc::new(|| {})).with_guest_config(quick)));
|
||||
s.execute("pref.userName", &json!({"name": name})).unwrap();
|
||||
s
|
||||
}
|
||||
|
||||
/// Pump every session (as their windows would each frame) until `f` holds.
|
||||
pub fn pump(sessions: &mut [&mut Session], what: &str, f: impl Fn(&[&mut Session]) -> bool) {
|
||||
let start = Instant::now();
|
||||
loop {
|
||||
for s in sessions.iter_mut() {
|
||||
s.live_poll();
|
||||
s.live_sync();
|
||||
}
|
||||
if f(sessions) {
|
||||
return;
|
||||
}
|
||||
let states: Vec<String> = sessions.iter().map(|s| s.live_summary().to_string()).collect();
|
||||
assert!(start.elapsed() < Duration::from_secs(20), "timed out waiting for {what}: {states:#?}");
|
||||
std::thread::sleep(Duration::from_millis(5));
|
||||
}
|
||||
}
|
||||
|
||||
/// The synced properties of two documents agree.
|
||||
pub fn same(a: &figcraft_doc::Document, b: &figcraft_doc::Document) -> bool {
|
||||
figcraft_collab::diff(&figcraft_collab::Replica::new(a.clone()), b).is_ok_and(|d| d.changes.is_empty())
|
||||
}
|
||||
Reference in New Issue
Block a user