Sausage Agent
49453fa9af
ci / check (push) Successful in 12m13s
Tapped copies {type:tapped, message:{id,from,to,reply_to,body}, ts} are
delivered directly (never tapped themselves), dropped when the tapper's
mailbox is full, and the tap is saved in actor.json like subscriptions.
299 lines
10 KiB
Rust
299 lines
10 KiB
Rust
//! On-disk layout of a Sausage world. The folder is the source of truth:
|
|
//! `sausage run` boots every actor that it finds in `actor/`.
|
|
//!
|
|
//! ```text
|
|
//! <world>/
|
|
//! actor/<id>/
|
|
//! actor.wasm the current version
|
|
//! actor.json version, running, subscriptions, tapped (broker)
|
|
//! source/ source of the current version
|
|
//! public/ written by the actor, read-only for others
|
|
//! data/ state, writable only by the actor
|
|
//! stderr.log stderr of the actor and notes of the broker
|
|
//! versions/<n>/ every version that was installed: actor.wasm, source/, meta.json
|
|
//! ```
|
|
|
|
use std::collections::BTreeSet;
|
|
use std::path::{Component, Path, PathBuf};
|
|
|
|
use anyhow::{Context, Result, bail};
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
/// Name prefix of the host's temporary files and folders. Copies skip them.
|
|
pub const TMP_PREFIX: &str = ".sausage-tmp-";
|
|
|
|
/// IDs of the actors that the host implements natively. `fs` sends the
|
|
/// requests for actor-served files (`vfs.rs`).
|
|
pub const NATIVE: &[&str] = &["broker", "console", "cargo", "fs"];
|
|
|
|
#[derive(Clone)]
|
|
pub struct World {
|
|
pub root: PathBuf,
|
|
}
|
|
|
|
/// `actor/<id>/actor.json`.
|
|
#[derive(Serialize, Deserialize, Clone, Debug, Default)]
|
|
pub struct ActorMeta {
|
|
/// The installed version. `None` for native actors and for a new
|
|
/// `actor.wasm` that the broker has not installed yet.
|
|
#[serde(default)]
|
|
pub version: Option<u64>,
|
|
#[serde(default = "yes")]
|
|
pub running: bool,
|
|
/// The actors whose published messages this actor receives.
|
|
#[serde(default)]
|
|
pub subscriptions: BTreeSet<String>,
|
|
/// Whether this actor gets a copy of every routed message (`tap`).
|
|
#[serde(default)]
|
|
pub tapped: bool,
|
|
}
|
|
|
|
fn yes() -> bool {
|
|
true
|
|
}
|
|
|
|
/// `versions/<n>/meta.json`.
|
|
#[derive(Serialize, Deserialize, Clone, Debug, Default)]
|
|
pub struct VersionMeta {
|
|
pub version: u64,
|
|
/// Path of the `.wasm` file, as the requester gave it.
|
|
pub wasm: String,
|
|
/// Path of the source folder, as the requester gave it.
|
|
pub source: Option<String>,
|
|
/// The actor that built the `.wasm` file.
|
|
pub built_by: Option<String>,
|
|
/// The actor that requested the version ("scan" for a file found on disk).
|
|
pub requested_by: String,
|
|
pub created_at: u64,
|
|
}
|
|
|
|
pub fn valid_id(id: &str) -> bool {
|
|
!id.is_empty()
|
|
&& id.len() <= 64
|
|
&& id.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-' || c == '_')
|
|
}
|
|
|
|
pub fn now() -> u64 {
|
|
std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map(|d| d.as_secs())
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
impl World {
|
|
pub fn new(root: impl Into<PathBuf>) -> Self {
|
|
Self { root: root.into() }
|
|
}
|
|
|
|
pub fn actors_root(&self) -> PathBuf {
|
|
self.root.join("actor")
|
|
}
|
|
pub fn actor_dir(&self, id: &str) -> PathBuf {
|
|
self.actors_root().join(id)
|
|
}
|
|
pub fn wasm_path(&self, id: &str) -> PathBuf {
|
|
self.actor_dir(id).join("actor.wasm")
|
|
}
|
|
pub fn public_dir(&self, id: &str) -> PathBuf {
|
|
self.actor_dir(id).join("public")
|
|
}
|
|
pub fn source_dir(&self, id: &str) -> PathBuf {
|
|
self.actor_dir(id).join("source")
|
|
}
|
|
pub fn data_dir(&self, id: &str) -> PathBuf {
|
|
self.actor_dir(id).join("data")
|
|
}
|
|
pub fn versions_dir(&self, id: &str) -> PathBuf {
|
|
self.actor_dir(id).join("versions")
|
|
}
|
|
pub fn version_dir(&self, id: &str, v: u64) -> PathBuf {
|
|
self.versions_dir(id).join(v.to_string())
|
|
}
|
|
/// Private scratch space of the broker (candidate copies, seeds).
|
|
pub fn scratch_dir(&self) -> PathBuf {
|
|
self.data_dir("broker").join("scratch")
|
|
}
|
|
|
|
pub fn exists(&self) -> bool {
|
|
self.actors_root().is_dir()
|
|
}
|
|
|
|
/// Create an empty world with the folders of the native actors.
|
|
pub fn create(root: &Path) -> Result<World> {
|
|
let world = World::new(std::path::absolute(root)?);
|
|
if world.exists() {
|
|
bail!("{} is already a Sausage world", root.display());
|
|
}
|
|
for id in NATIVE {
|
|
world.ensure_actor_dirs(id)?;
|
|
}
|
|
Ok(world)
|
|
}
|
|
|
|
pub fn ensure_actor_dirs(&self, id: &str) -> Result<()> {
|
|
std::fs::create_dir_all(self.public_dir(id))?;
|
|
std::fs::create_dir_all(self.data_dir(id))?;
|
|
Ok(())
|
|
}
|
|
|
|
/// IDs of every folder in `actor/`, sorted.
|
|
pub fn actor_ids(&self) -> Result<Vec<String>> {
|
|
let mut ids: Vec<String> = std::fs::read_dir(self.actors_root())?
|
|
.filter_map(|e| e.ok())
|
|
.filter(|e| e.path().is_dir())
|
|
.filter_map(|e| e.file_name().into_string().ok())
|
|
.filter(|id| valid_id(id))
|
|
.collect();
|
|
ids.sort();
|
|
Ok(ids)
|
|
}
|
|
|
|
pub fn read_meta(&self, id: &str) -> ActorMeta {
|
|
std::fs::read_to_string(self.actor_dir(id).join("actor.json"))
|
|
.ok()
|
|
.and_then(|s| serde_json::from_str(&s).ok())
|
|
.unwrap_or(ActorMeta { running: true, ..Default::default() })
|
|
}
|
|
|
|
pub fn write_meta(&self, id: &str, meta: &ActorMeta) -> Result<()> {
|
|
write_atomic(&self.actor_dir(id).join("actor.json"), serde_json::to_string_pretty(meta)?.as_bytes())
|
|
}
|
|
|
|
/// Make `actor.wasm` and `source/` a copy of `versions/<version>/`,
|
|
/// unless `actor.wasm` already is that version (it came from there).
|
|
pub fn install_files(&self, id: &str, version: u64) -> Result<()> {
|
|
let dir = self.version_dir(id, version);
|
|
self.ensure_actor_dirs(id)?;
|
|
let wasm = std::fs::read(dir.join("actor.wasm"))?;
|
|
if std::fs::read(self.wasm_path(id)).ok().as_ref() == Some(&wasm) {
|
|
return Ok(());
|
|
}
|
|
write_atomic(&self.wasm_path(id), &wasm)?;
|
|
replace_tree(&dir.join("source"), &self.source_dir(id))
|
|
}
|
|
|
|
/// Version numbers in `versions/`, ascending.
|
|
pub fn versions(&self, id: &str) -> Vec<u64> {
|
|
let mut v: Vec<u64> = std::fs::read_dir(self.versions_dir(id))
|
|
.map(|rd| rd.filter_map(|e| e.ok()?.file_name().to_str()?.parse().ok()).collect())
|
|
.unwrap_or_default();
|
|
v.sort();
|
|
v
|
|
}
|
|
|
|
pub fn next_version(&self, id: &str) -> u64 {
|
|
self.versions(id).last().map_or(1, |v| v + 1)
|
|
}
|
|
|
|
/// Map a path in an actor's namespace to a host path.
|
|
///
|
|
/// - `/actor/<id>/...` is the global filesystem.
|
|
/// - `/public/...` and `/data/...` belong to `sender`.
|
|
/// - Native senders (the console) can also give absolute host paths.
|
|
pub fn resolve(&self, sender: &str, path: &str) -> Result<PathBuf> {
|
|
let under = |base: PathBuf, rest: &str| -> Result<PathBuf> {
|
|
let mut out = base;
|
|
for c in Path::new(rest).components() {
|
|
match c {
|
|
Component::Normal(n) => out.push(n),
|
|
Component::CurDir | Component::RootDir => {}
|
|
_ => bail!("path must not contain '..': {path}"),
|
|
}
|
|
}
|
|
Ok(out)
|
|
};
|
|
let prefixed = |prefix: &str| {
|
|
path.strip_prefix(prefix).filter(|rest| rest.is_empty() || rest.starts_with('/'))
|
|
};
|
|
if let Some(rest) = prefixed("/actor") {
|
|
return under(self.actors_root(), rest);
|
|
}
|
|
if let Some(rest) = prefixed("/public") {
|
|
return under(self.public_dir(sender), rest);
|
|
}
|
|
if let Some(rest) = prefixed("/data") {
|
|
return under(self.data_dir(sender), rest);
|
|
}
|
|
if NATIVE.contains(&sender) && Path::new(path).is_absolute() {
|
|
return Ok(PathBuf::from(path));
|
|
}
|
|
bail!("cannot resolve path {path} for {sender}: use /actor/<id>/..., /public/... or /data/...")
|
|
}
|
|
|
|
/// The path of a host file in the global filesystem, if it has one.
|
|
pub fn global_path(&self, host: &Path) -> Option<String> {
|
|
let rel = host.strip_prefix(self.actors_root()).ok()?;
|
|
Some(format!("/actor/{}", rel.display()))
|
|
}
|
|
}
|
|
|
|
/// Write a file so that readers never see a partial version.
|
|
pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
|
|
let dir = path.parent().context("path has no parent")?;
|
|
std::fs::create_dir_all(dir)?;
|
|
let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("file");
|
|
let tmp = dir.join(format!("{TMP_PREFIX}host-{}-{name}", std::process::id()));
|
|
std::fs::write(&tmp, bytes)?;
|
|
std::fs::rename(&tmp, path)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Folders that a copy of a source folder leaves out.
|
|
const BUILD_DIRS: &[&str] = &["target", ".git"];
|
|
|
|
/// Copy a directory tree, except uncommitted files.
|
|
pub fn copy_tree(from: &Path, to: &Path) -> Result<()> {
|
|
copy_filtered(from, to, &[])
|
|
}
|
|
|
|
fn copy_filtered(from: &Path, to: &Path, skip: &[&str]) -> Result<()> {
|
|
std::fs::create_dir_all(to)?;
|
|
for entry in std::fs::read_dir(from).with_context(|| format!("read {}", from.display()))? {
|
|
let entry = entry?;
|
|
let name = entry.file_name();
|
|
let name = name.to_string_lossy();
|
|
if name.starts_with(TMP_PREFIX) {
|
|
continue;
|
|
}
|
|
let dst = to.join(&*name);
|
|
let ty = entry.file_type()?;
|
|
if ty.is_dir() && !skip.contains(&&*name) {
|
|
copy_filtered(&entry.path(), &dst, skip)?;
|
|
} else if ty.is_file() {
|
|
std::fs::copy(entry.path(), &dst)?;
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Replace the directory `dst` with a copy of `src` (an empty folder if `src`
|
|
/// does not exist): copy to a sibling, then swap with two renames.
|
|
pub fn replace_tree(src: &Path, dst: &Path) -> Result<()> {
|
|
replace_filtered(src, dst, &[])
|
|
}
|
|
|
|
/// Like `replace_tree`, for a source folder: leaves out `target/` and `.git/`.
|
|
pub fn replace_source(src: &Path, dst: &Path) -> Result<()> {
|
|
replace_filtered(src, dst, BUILD_DIRS)
|
|
}
|
|
|
|
fn replace_filtered(src: &Path, dst: &Path, skip: &[&str]) -> Result<()> {
|
|
let parent = dst.parent().context("path has no parent")?;
|
|
let name = dst.file_name().and_then(|n| n.to_str()).unwrap_or("dir");
|
|
let fresh = parent.join(format!("{TMP_PREFIX}new-{name}"));
|
|
let old = parent.join(format!("{TMP_PREFIX}old-{name}"));
|
|
let _ = std::fs::remove_dir_all(&fresh);
|
|
let _ = std::fs::remove_dir_all(&old);
|
|
if src.is_dir() {
|
|
copy_filtered(src, &fresh, skip)?;
|
|
} else {
|
|
std::fs::create_dir_all(&fresh)?;
|
|
}
|
|
if dst.exists() {
|
|
std::fs::rename(dst, &old)?;
|
|
}
|
|
std::fs::rename(&fresh, dst)?;
|
|
let _ = std::fs::remove_dir_all(&old);
|
|
Ok(())
|
|
}
|