diff --git a/Cargo.lock b/Cargo.lock index 0bcdbce..c11c2c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -226,6 +226,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d262e149917187838d5b42777c8253bcb64500067342904e7d429499a6f277e" dependencies = [ "find-msvc-tools", + "jobserver", + "libc", "shlex", ] @@ -408,10 +410,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" dependencies = [ "generic-array", - "rand_core", + "rand_core 0.6.4", "typenum", ] +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "derive_arbitrary" version = "1.4.2" @@ -589,6 +597,18 @@ dependencies = [ "wasi", ] +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi 5.3.0", + "wasip2", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -597,7 +617,7 @@ checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", "libc", - "r-efi", + "r-efi 6.0.0", ] [[package]] @@ -734,6 +754,16 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom 0.4.3", + "libc", +] + [[package]] name = "js-sys" version = "0.3.106" @@ -866,6 +896,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pkg-config" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" + [[package]] name = "poly1305" version = "0.8.0" @@ -877,6 +913,15 @@ dependencies = [ "universal-hash", ] +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "proc-macro2" version = "1.0.107" @@ -895,12 +940,38 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "r-efi" version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -910,6 +981,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "regex-automata" version = "0.4.18" @@ -1212,6 +1292,8 @@ version = "0.1.36" dependencies = [ "serde", "serde_json", + "tinystoragedrivers-core", + "tokio", ] [[package]] @@ -1263,8 +1345,10 @@ name = "tinybox-drivers" version = "0.1.36" dependencies = [ "log", + "serde", "serde_json", "tempfile", + "tinybox-bus", "tinybox-core", "tinybox-jail", "tinystoragedrivers-core", @@ -1423,8 +1507,8 @@ dependencies = [ [[package]] name = "tinystoragedrivers-bus" -version = "0.11.0" -source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v0.11.0#9eeef54aea403ac07461a612189bdf4982782823" +version = "1.0.0" +source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v1.0.0#141fd9ed0ad5d14a8fb5a1b3e837a719e81ac08d" dependencies = [ "serde", "serde_json", @@ -1433,8 +1517,8 @@ dependencies = [ [[package]] name = "tinystoragedrivers-busproxy" -version = "0.11.0" -source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v0.11.0#9eeef54aea403ac07461a612189bdf4982782823" +version = "1.0.0" +source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v1.0.0#141fd9ed0ad5d14a8fb5a1b3e837a719e81ac08d" dependencies = [ "async-trait", "serde", @@ -1448,8 +1532,8 @@ dependencies = [ [[package]] name = "tinystoragedrivers-client" -version = "0.11.0" -source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v0.11.0#9eeef54aea403ac07461a612189bdf4982782823" +version = "1.0.0" +source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v1.0.0#141fd9ed0ad5d14a8fb5a1b3e837a719e81ac08d" dependencies = [ "async-trait", "serde", @@ -1462,20 +1546,24 @@ dependencies = [ [[package]] name = "tinystoragedrivers-core" -version = "0.11.0" -source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v0.11.0#9eeef54aea403ac07461a612189bdf4982782823" +version = "1.0.0" +source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v1.0.0#141fd9ed0ad5d14a8fb5a1b3e837a719e81ac08d" dependencies = [ "async-trait", + "data-encoding", "serde", "serde_json", + "sha2", "thiserror", "tokio", + "ulid", + "zstd", ] [[package]] name = "tinystoragedrivers-secrets" -version = "0.11.0" -source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v0.11.0#9eeef54aea403ac07461a612189bdf4982782823" +version = "1.0.0" +source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v1.0.0#141fd9ed0ad5d14a8fb5a1b3e837a719e81ac08d" dependencies = [ "async-trait", "atomicwrites", @@ -1598,6 +1686,16 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "ulid" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "470dbf6591da1b39d43c14523b2b469c86879a53e8b758c8e090a470fe7b1fbe" +dependencies = [ + "rand", + "web-time", +] + [[package]] name = "unicode-ident" version = "1.0.24" @@ -1694,6 +1792,15 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasip2" +version = "1.0.4+wasi-0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" +dependencies = [ + "wit-bindgen", +] + [[package]] name = "wasm-bindgen" version = "0.2.129" @@ -1739,6 +1846,16 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-roots" version = "1.0.9" @@ -1899,6 +2016,12 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + [[package]] name = "xattr" version = "1.6.1" @@ -1909,6 +2032,26 @@ dependencies = [ "rustix 1.1.4", ] +[[package]] +name = "zerocopy" +version = "0.8.63" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5fe1f8f1b06191a00962174c61aa5005e0bb391a6d80d07e24d115c01a92ed8" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.63" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "863ad3ac83293fb4d740aedbfdc9240dd8d1a50c1099acd76ce80ce7c7230c7f" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "zeroize" version = "1.9.0" @@ -1949,3 +2092,31 @@ dependencies = [ "log", "simd-adler32", ] + +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64d80649ab6db9d9f6f9c80a40becd948eda4714a0a5ac8c4d157a32231c7882" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.1.2+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "481ada9c543b2323cbb7bbdbb5249177fbf0188ed959ed2c15e4b22421c84ed3" +dependencies = [ + "cc", + "pkg-config", +] diff --git a/Cargo.toml b/Cargo.toml index edd6285..4a0a423 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -69,12 +69,12 @@ tinybox-ssh = { path = "crates/tinybox-ssh", version = "0.1.25" } tinybox-sync = { path = "crates/tinybox-sync", version = "0.1.25" } # The storage ports only (no database client), by git tag so a host that vendors # tinystoragedrivers once can `[patch]` it. Used by `tinybox-drivers`. -tinystoragedrivers-core = { version = ">=0.8.0, <1", git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v0.11.0" } +tinystoragedrivers-core = { version = "1.0.0", git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v1.0.0" } # The module reaches the host's storage over the storage bus: the proxy yields # the same port traits a linked library gets. The client crate serves them in # tests. -tinystoragedrivers-busproxy = { version = ">=0.8.0, <1", git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v0.11.0" } -tinystoragedrivers-client = { version = ">=0.8.0, <1", git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v0.11.0" } +tinystoragedrivers-busproxy = { version = "1.0.0", git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v1.0.0" } +tinystoragedrivers-client = { version = "1.0.0", git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v1.0.0" } tinybox-drivers = { path = "crates/tinybox-drivers", version = "0.1.32" } # Lints apply to every crate in the workspace via `[lints] workspace = true`. diff --git a/crates/tinybox-bus/Cargo.toml b/crates/tinybox-bus/Cargo.toml index db647e0..002d584 100644 --- a/crates/tinybox-bus/Cargo.toml +++ b/crates/tinybox-bus/Cargo.toml @@ -10,10 +10,21 @@ repository.workspace = true [dependencies] # The bus boundary contains data only; serialization is its sole dependency. serde.workspace = true +# Only the `storage` feature: the schema the host composes. Ports only (no +# backend, no I/O), pure JSON steps. +serde_json = { workspace = true, optional = true } +tinystoragedrivers-core = { workspace = true, optional = true } + +[features] +default = [] +# `storage_schema()` and its pure v1 to v2 steps. Off by default so the +# contract's default dependency closure stays serde only; a host enables it. +storage = ["dep:serde_json", "dep:tinystoragedrivers-core"] [dev-dependencies] -# Round-trip tests pin the serialized public vocabulary. serde_json.workspace = true +# Migration tests run the steps over `MemoryStorage`. +tokio = { workspace = true } [lints] workspace = true diff --git a/crates/tinybox-bus/src/lib.rs b/crates/tinybox-bus/src/lib.rs index 7ee9614..ba696e0 100644 --- a/crates/tinybox-bus/src/lib.rs +++ b/crates/tinybox-bus/src/lib.rs @@ -27,6 +27,8 @@ pub use shell_execution::ShellExecRequest; pub use shell_selection::{SelectShellRequest, ShellFlavor, ShellInvocation}; mod stdio; +#[cfg(feature = "storage")] +pub mod storage; pub use stdio::{ MAX_STDIO_BUFFER_BYTES, MAX_STDIO_PROCESSES, MAX_STDIO_READ_BYTES, MAX_STDIO_WAIT_MS, MAX_STDIO_WRITE_BYTES, STDIO_CLOSED, StdioChunk, StdioExit, StdioReadRequest, diff --git a/crates/tinybox-bus/src/storage/box_v1.rs b/crates/tinybox-bus/src/storage/box_v1.rs new file mode 100644 index 0000000..5fffdab --- /dev/null +++ b/crates/tinybox-bus/src/storage/box_v1.rs @@ -0,0 +1,176 @@ +//! A v1 box row (a serialized `BoxInfo`) as the v2 `box.boxes` body. +//! +//! v1 stored the model's own serde form: `PascalCase` variants, externally +//! tagged enums, `Duration` as `{secs, nanos}` and `SystemTime` as +//! `{secs_since_epoch, nanos_since_epoch}`. The v2 body is `snake_case`, uses +//! internally tagged variants and one `*_ms` integer per instant. The +//! conversion is pure JSON to JSON. + +use std::collections::BTreeMap; +use std::path::PathBuf; + +use serde::Deserialize; +use serde_json::{Value, json}; + +#[derive(Deserialize)] +struct V1Box { + state: V1State, + spec: V1Spec, + #[serde(default)] + created_at: Option, +} + +#[derive(Deserialize)] +enum V1State { + Creating, + Ready, + Running, + Paused, + Stopped, + Archived, + Failed, +} + +#[derive(Deserialize)] +struct V1Spec { + runner: Value, + workspace: Value, + source: V1Source, + resources: Value, + lifecycle: V1Lifecycle, + network: V1Network, + #[serde(default)] + ports: Vec, + env: BTreeMap, +} + +#[derive(Deserialize)] +enum V1Source { + LocalDir(PathBuf), + OciImage(String), + Snapshot(String), + GitRepo { url: String, rev: String }, +} + +#[derive(Deserialize)] +struct V1Duration { + secs: u64, + nanos: u32, +} + +#[derive(Deserialize)] +struct V1Time { + secs_since_epoch: u64, + nanos_since_epoch: u32, +} + +#[derive(Deserialize)] +enum V1Lifecycle { + Ephemeral { + ttl: V1Duration, + }, + Persistent { + #[serde(default)] + autosnapshot: Option, + }, +} + +#[derive(Deserialize)] +enum V1Network { + Denied, + Egress, + Allowlist { hosts: Vec, ports: Vec }, + Open, +} + +fn duration_ms(duration: &V1Duration) -> u64 { + duration + .secs + .saturating_mul(1000) + .saturating_add(u64::from(duration.nanos / 1_000_000)) +} + +fn time_ms(time: &V1Time) -> i64 { + let ms = duration_ms(&V1Duration { + secs: time.secs_since_epoch, + nanos: time.nanos_since_epoch, + }); + i64::try_from(ms).unwrap_or(i64::MAX) +} + +const fn state_name(state: &V1State) -> &'static str { + match state { + V1State::Creating => "creating", + V1State::Ready => "ready", + V1State::Running => "running", + V1State::Paused => "paused", + V1State::Stopped => "stopped", + V1State::Archived => "archived", + V1State::Failed => "failed", + } +} + +fn source_body(source: V1Source) -> Value { + match source { + V1Source::LocalDir(path) => json!({"kind": "local_dir", "path": path}), + V1Source::OciImage(image) => json!({"kind": "oci_image", "image": image}), + V1Source::Snapshot(snapshot) => json!({"kind": "snapshot", "snapshot": snapshot}), + V1Source::GitRepo { url, rev } => json!({"kind": "git_repo", "url": url, "rev": rev}), + } +} + +fn lifecycle_body(lifecycle: &V1Lifecycle) -> Value { + match lifecycle { + V1Lifecycle::Ephemeral { ttl } => { + json!({"kind": "ephemeral", "ttl_ms": duration_ms(ttl)}) + } + V1Lifecycle::Persistent { autosnapshot } => json!({ + "kind": "persistent", + "autosnapshot_ms": autosnapshot.as_ref().map(duration_ms), + }), + } +} + +fn network_body(network: V1Network) -> Value { + match network { + V1Network::Denied => json!({"kind": "denied"}), + V1Network::Egress => json!({"kind": "egress"}), + V1Network::Allowlist { hosts, ports } => { + json!({"kind": "allowlist", "hosts": hosts, "ports": ports}) + } + V1Network::Open => json!({"kind": "open"}), + } +} + +/// The v2 body for a v1 box row. +/// +/// The box id is the document id and is not part of the body. A box recorded +/// before tinybox tracked time has no `created_at`; its `created_ms` stays +/// `null` so "unknown" is never mistaken for "now". +/// +/// # Errors +/// +/// A message when the row is not a v1 box this version understands. +pub fn to_body(source: Value) -> Result { + let old: V1Box = serde_json::from_value(source).map_err(|error| error.to_string())?; + let spec = old.spec; + Ok(json!({ + "state": state_name(&old.state), + "created_ms": old.created_at.as_ref().map(time_ms), + "spec": { + "runner": spec.runner, + "workspace": spec.workspace, + "source": source_body(spec.source), + "resources": spec.resources, + "lifecycle": lifecycle_body(&spec.lifecycle), + "network": network_body(spec.network), + "ports": spec.ports, + "env": spec.env, + }, + })) +} + +#[cfg(test)] +#[path = "box_v1_tests.rs"] +#[allow(clippy::expect_used, clippy::unwrap_used)] +mod tests; diff --git a/crates/tinybox-bus/src/storage/box_v1_tests.rs b/crates/tinybox-bus/src/storage/box_v1_tests.rs new file mode 100644 index 0000000..7a8bf5d --- /dev/null +++ b/crates/tinybox-bus/src/storage/box_v1_tests.rs @@ -0,0 +1,116 @@ +//! Tests for the v1 box row to v2 `box.boxes` body conversion. + +use super::*; + +fn v1() -> Value { + json!({ + "id": "b", + "state": "Paused", + "spec": { + "runner": {"host": "local", "sandbox": "passthrough"}, + "workspace": {"host": "local", "sandbox": "passthrough"}, + "source": {"GitRepo": {"url": "https://x.test/r", "rev": "main"}}, + "resources": {"cpu_millis": 1, "memory_bytes": 2, "pids_max": 3, "disk_bytes": 4}, + "lifecycle": {"Persistent": {"autosnapshot": {"secs": 5, "nanos": 7_000_000}}}, + "network": "Open", + "env": {}, + }, + "created_at": {"secs_since_epoch": 2, "nanos_since_epoch": 3_000_000}, + }) +} + +#[test] +fn every_variant_takes_its_v2_tagged_form() { + let body = to_body(v1()).expect("decodes"); + assert_eq!(body["state"], "paused"); + assert_eq!(body["created_ms"], 2003); + assert_eq!( + body["spec"]["source"], + json!({"kind": "git_repo", "url": "https://x.test/r", "rev": "main"}) + ); + assert_eq!( + body["spec"]["lifecycle"], + json!({"kind": "persistent", "autosnapshot_ms": 5007}) + ); + assert_eq!(body["spec"]["network"], json!({"kind": "open"})); + assert_eq!(body["spec"]["ports"], json!([])); + assert!(body.get("id").is_none(), "the id is the document id"); +} + +#[test] +fn a_persistent_box_without_autosnapshot_keeps_it_null() { + let mut old = v1(); + old["spec"]["lifecycle"] = json!({"Persistent": {}}); + let body = to_body(old).expect("decodes"); + assert_eq!(body["spec"]["lifecycle"]["autosnapshot_ms"], Value::Null); +} + +#[test] +fn an_unknown_state_is_refused() { + let mut old = v1(); + old["state"] = json!("Levitating"); + to_body(old).expect_err("unknown state"); +} + +#[test] +fn the_other_sources_lifecycles_and_networks_convert() { + let cases = [ + ( + "source", + json!({"LocalDir": "/w"}), + json!({"kind": "local_dir", "path": "/w"}), + ), + ( + "source", + json!({"OciImage": "img:1"}), + json!({"kind": "oci_image", "image": "img:1"}), + ), + ( + "source", + json!({"Snapshot": "s1"}), + json!({"kind": "snapshot", "snapshot": "s1"}), + ), + ( + "lifecycle", + json!({"Ephemeral": {"ttl": {"secs": 3, "nanos": 0}}}), + json!({"kind": "ephemeral", "ttl_ms": 3000}), + ), + ("network", json!("Denied"), json!({"kind": "denied"})), + ("network", json!("Egress"), json!({"kind": "egress"})), + ( + "network", + json!({"Allowlist": {"hosts": ["a.test"], "ports": [443]}}), + json!({"kind": "allowlist", "hosts": ["a.test"], "ports": [443]}), + ), + ]; + for (field, old, want) in cases { + let mut row = v1(); + row["spec"][field] = old; + let body = to_body(row).expect("decodes"); + assert_eq!(body["spec"][field], want); + } +} + +#[test] +fn every_state_has_its_snake_case_name() { + for (old, want) in [ + ("Creating", "creating"), + ("Ready", "ready"), + ("Running", "running"), + ("Paused", "paused"), + ("Stopped", "stopped"), + ("Archived", "archived"), + ("Failed", "failed"), + ] { + let mut row = v1(); + row["state"] = json!(old); + assert_eq!(to_body(row).expect("decodes")["state"], want); + } +} + +#[test] +fn a_box_without_a_creation_time_keeps_it_unknown() { + let mut row = v1(); + row.as_object_mut().expect("row").remove("created_at"); + assert_eq!(to_body(row).expect("decodes")["created_ms"], Value::Null); +} diff --git a/crates/tinybox-bus/src/storage/jail_index.rs b/crates/tinybox-bus/src/storage/jail_index.rs new file mode 100644 index 0000000..0ac166f --- /dev/null +++ b/crates/tinybox-bus/src/storage/jail_index.rs @@ -0,0 +1,88 @@ +//! Converting a jail registry's index between its own JSON and the v2 body. +//! +//! Both directions live here, in the contract, so the migration step and the +//! store that reads the migrated body share one definition. +//! +//! The registry (`tinybox-jail`) owns the index shape and keeps it in seconds +//! (`created_at_unix`, `updated_at_unix`) next to a `schema_version`, because +//! the same JSON is the legacy `index.json` file. The `box.jail_index` body +//! holds the same records with one millisecond field per instant +//! (`created_ms`, `updated_ms`) and no version field: the document's metadata +//! carries the body version. + +use std::io; + +use serde_json::Value; + +/// The index version the registry writes; restored on read. +const REGISTRY_SCHEMA_VERSION: u64 = 1; + +/// (registry field, body field) +const TIMES: [(&str, &str); 2] = [ + ("created_at_unix", "created_ms"), + ("updated_at_unix", "updated_ms"), +]; + +fn invalid(message: String) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, message) +} + +/// The v2 body for a registry's index JSON. +/// +/// # Errors +/// +/// [`io::ErrorKind::InvalidData`] when a time field is not a whole number of +/// seconds that fits in milliseconds. +pub fn to_body(mut index: Value) -> io::Result { + if let Some(object) = index.as_object_mut() { + object.remove("schema_version"); + } + for record in records_mut(&mut index) { + for (seconds_field, ms_field) in TIMES { + let Some(raw) = record.remove(seconds_field) else { + continue; + }; + let ms = raw + .as_u64() + .and_then(|seconds| seconds.checked_mul(1000)) + .and_then(|ms| i64::try_from(ms).ok()) + .ok_or_else(|| invalid(format!("{seconds_field} is not a time in seconds")))?; + record.insert(ms_field.to_owned(), Value::from(ms)); + } + } + Ok(index) +} + +/// The registry's index JSON for a stored v2 body. +#[must_use] +pub fn from_body(mut body: Value) -> Value { + for record in records_mut(&mut body) { + for (seconds_field, ms_field) in TIMES { + if let Some(ms) = record.remove(ms_field).and_then(|raw| raw.as_i64()) { + let seconds = u64::try_from(ms.div_euclid(1000)).unwrap_or(0); + record.insert(seconds_field.to_owned(), Value::from(seconds)); + } + } + } + if let Some(object) = body.as_object_mut() { + object.insert( + "schema_version".to_owned(), + Value::from(REGISTRY_SCHEMA_VERSION), + ); + } + body +} + +fn records_mut(index: &mut Value) -> impl Iterator> { + index + .get_mut("records") + .and_then(Value::as_object_mut) + .into_iter() + .flat_map(|records| records.values_mut()) + .filter_map(Value::as_object_mut) +} + +#[cfg(test)] +#[path = "jail_index_tests.rs"] +#[allow(clippy::expect_used, clippy::unwrap_used)] +mod tests; diff --git a/crates/tinybox-bus/src/storage/jail_index_tests.rs b/crates/tinybox-bus/src/storage/jail_index_tests.rs new file mode 100644 index 0000000..2f5b5f6 --- /dev/null +++ b/crates/tinybox-bus/src/storage/jail_index_tests.rs @@ -0,0 +1,81 @@ +//! Tests for the jail index conversion between registry seconds and body milliseconds. + +use super::*; +use serde_json::json; + +fn registry_json() -> Value { + json!({ + "schema_version": 1, + "records": { + "j1": { + "id": "j1", + "label": "build", + "dir": "/srv/j1", + "backend_at_create": "landlock", + "created_at_unix": 1_700_000_000_u64, + "updated_at_unix": 1_700_000_060_u64, + "notes": null, + }, + }, + }) +} + +fn body_json() -> Value { + json!({ + "records": { + "j1": { + "id": "j1", + "label": "build", + "dir": "/srv/j1", + "backend_at_create": "landlock", + "created_ms": 1_700_000_000_000_i64, + "updated_ms": 1_700_000_060_000_i64, + "notes": null, + }, + }, + }) +} + +#[test] +fn the_body_holds_millisecond_times_and_no_schema_version() { + assert_eq!(to_body(registry_json()).expect("body"), body_json()); +} + +#[test] +fn the_registry_form_comes_back_from_the_body() { + assert_eq!(from_body(body_json()), registry_json()); +} + +#[test] +fn an_index_without_records_converts_both_ways() { + assert_eq!( + to_body(json!({"records": {}})).expect("body"), + json!({"records": {}}) + ); + assert_eq!( + from_body(json!({"records": {}})), + json!({"records": {}, "schema_version": 1}) + ); +} + +#[test] +fn a_time_that_overflows_milliseconds_is_invalid_data() { + let mut index = registry_json(); + index["records"]["j1"]["created_at_unix"] = json!(u64::MAX); + let error = to_body(index).expect_err("overflow"); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); +} + +#[test] +fn a_time_that_is_not_a_number_is_invalid_data() { + let mut index = registry_json(); + index["records"]["j1"]["updated_at_unix"] = json!("yesterday"); + let error = to_body(index).expect_err("not a number"); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); +} + +#[test] +fn a_record_without_times_passes_through() { + let index = json!({"records": {"j1": {"id": "j1"}}}); + assert_eq!(to_body(index.clone()).expect("body"), index); +} diff --git a/crates/tinybox-bus/src/storage/mod.rs b/crates/tinybox-bus/src/storage/mod.rs new file mode 100644 index 0000000..f8f0b7f --- /dev/null +++ b/crates/tinybox-bus/src/storage/mod.rs @@ -0,0 +1,94 @@ +//! What tinybox stores, as a [`Schema`] the host composes. +//! +//! Owner `box`, both collections at body version 2 in the module scope: +//! +//! | v1 | v2 | +//! |---|---| +//! | `tinybox_boxes` | [`BOX_COLLECTION`] (`box.boxes`) | +//! | `tinybox_jail_index` | [`JAIL_INDEX_COLLECTION`] (`box.jail_index`) | +//! +//! A host grants the module exactly these items (`ScopeClass::Module`), so the +//! grant is derived from [`storage_schema`] rather than restated. The steps +//! bring v1 rows forward when the host runs `Registry::migrate`, and they are +//! pure: they transform JSON and touch no storage. +//! +//! The implementation crate stores and loads the v2 bodies; the v1 shapes are +//! decoded here by [`box_v1`] and [`jail_index`] so the contract owns both ends +//! of the migration. + +use tinystoragedrivers_core::schema::{ + Owner, Schema, ScopeClass, SourceRecord, Step, TargetRecord, +}; +use tinystoragedrivers_core::{Key, Result, StorageError}; + +pub mod box_v1; +pub mod jail_index; + +/// The collection that holds one document per box, keyed by box id. +pub const BOX_COLLECTION: &str = "box.boxes"; + +/// The collection that holds one index document per jail registry. +pub const JAIL_INDEX_COLLECTION: &str = "box.jail_index"; + +/// The v1 name of [`BOX_COLLECTION`]. +pub const V1_BOX_COLLECTION: &str = "tinybox_boxes"; + +/// The v1 name of [`JAIL_INDEX_COLLECTION`]. +pub const V1_JAIL_INDEX_COLLECTION: &str = "tinybox_jail_index"; + +/// The body version both collections are declared at. +pub const BODY_VERSION: u32 = 2; + +/// Everything tinybox stores, with the v1 to v2 migration steps. +/// +/// # Panics +/// +/// Never in practice: the owner name is a constant that satisfies the owner +/// grammar. +#[must_use] +pub fn storage_schema() -> Schema { + #[allow(clippy::expect_used, reason = "a constant that satisfies the grammar")] + let owner = Owner::new("box").expect("`box` is a valid owner"); + Schema::new(owner) + .collection(BOX_COLLECTION, BODY_VERSION, ScopeClass::Module) + .collection(JAIL_INDEX_COLLECTION, BODY_VERSION, ScopeClass::Module) + .step( + Step::new("rename_boxes", BOX_COLLECTION, 1, BODY_VERSION, |batch| { + batch.into_iter().map(box_to_v2).collect() + }) + .source(V1_BOX_COLLECTION), + ) + .step( + Step::new( + "rename_jail_index", + JAIL_INDEX_COLLECTION, + 1, + BODY_VERSION, + |batch| batch.into_iter().map(jail_index_to_v2).collect(), + ) + .source(V1_JAIL_INDEX_COLLECTION), + ) +} + +/// A v1 box row (a serialized `BoxInfo`) as a v2 record under the same id. +fn box_to_v2(source: SourceRecord) -> Result { + let id = Key::of([source.id.as_str()])?; + let body = box_v1::to_body(source.body).map_err(|error| { + StorageError::serialization(format!("v1 box `{}` does not decode: {error}", source.id)) + })?; + Ok(TargetRecord::new(BOX_COLLECTION, id.as_str(), body)) +} + +/// A v1 jail index row as a v2 record under the same id. +fn jail_index_to_v2(source: SourceRecord) -> Result { + let id = Key::of([source.id.as_str()])?; + let body = jail_index::to_body(source.body).map_err(|error| { + StorageError::serialization(format!("v1 jail index `{}`: {error}", source.id)) + })?; + Ok(TargetRecord::new(JAIL_INDEX_COLLECTION, id.as_str(), body)) +} + +#[cfg(test)] +#[path = "mod_tests.rs"] +#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)] +mod tests; diff --git a/crates/tinybox-bus/src/storage/mod_tests.rs b/crates/tinybox-bus/src/storage/mod_tests.rs new file mode 100644 index 0000000..86330a0 --- /dev/null +++ b/crates/tinybox-bus/src/storage/mod_tests.rs @@ -0,0 +1,280 @@ +//! Tests for `storage_schema` and the v1 to v2 migration steps. + +use super::*; +use serde_json::{Value, json}; +use tinystoragedrivers_core::CollectionSpec; +use tinystoragedrivers_core::schema::{Item, MigrateOptions, Registry}; +use tinystoragedrivers_core::{MemoryStorage, Precondition, Scope, ScopedStorage, StorageBackend}; + +/// A box as the v1 store wrote it: `BoxInfo` serialized by serde, keyed by id. +fn v1_box() -> Value { + json!({ + "id": "build-1", + "state": "Ready", + "spec": { + "runner": {"host": "local", "sandbox": "passthrough"}, + "workspace": {"host": "local", "sandbox": "passthrough"}, + "source": {"LocalDir": "/srv/work"}, + "resources": { + "cpu_millis": 2000, + "memory_bytes": 2_147_483_648_u64, + "pids_max": 512, + "disk_bytes": 8_589_934_592_u64, + }, + "lifecycle": {"Ephemeral": {"ttl": {"secs": 3600, "nanos": 0}}}, + "network": {"Allowlist": {"hosts": ["example.test"], "ports": [443]}}, + "ports": [{"guest": 80, "host": null}], + "env": {"KEY": "value"}, + }, + "created_at": {"secs_since_epoch": 1_700_000_000_u64, "nanos_since_epoch": 5_000_000_u32}, + }) +} + +fn v1_jail_index() -> Value { + json!({ + "schema_version": 1, + "records": { + "j1": { + "id": "j1", + "label": "build", + "dir": "/srv/j1", + "backend_at_create": "landlock", + "created_at_unix": 1_700_000_000_u64, + "updated_at_unix": 1_700_000_060_u64, + "notes": "keep", + }, + }, + }) +} + +async fn v1_storage(seed: &[(&str, &str, Value)]) -> (Registry, MemoryStorage, ScopedStorage) { + let registry = Registry::compose([storage_schema()]).expect("compose"); + let backend = MemoryStorage::new(); + registry.prepare(&backend).await.expect("prepare"); + let storage = backend.for_scope(&Scope::local()).expect("scope"); + for (collection, id, body) in seed { + let docs = storage.documents(); + docs.ensure_collection(&CollectionSpec::new(*collection)) + .await + .expect("legacy collection"); + docs.put(collection, id, body.clone(), Precondition::Absent) + .await + .expect("seed"); + } + (registry, backend, storage) +} + +#[test] +fn the_schema_declares_the_box_collections_for_the_module_scope() { + let schema = storage_schema(); + assert_eq!(schema.owner.as_str(), "box"); + let declared: Vec<_> = schema + .items + .iter() + .map(|item| match item { + Item::Collection { + name, + version, + scope, + .. + } => (name.as_str(), *version, *scope), + other => panic!("unexpected item {other:?}"), + }) + .collect(); + assert_eq!( + declared, + vec![ + ("box.boxes", 2, ScopeClass::Module), + ("box.jail_index", 2, ScopeClass::Module), + ] + ); +} + +#[test] +fn the_schema_composes() { + Registry::compose([storage_schema()]).expect("compose"); +} + +#[test] +fn the_collection_constants_are_the_declared_names() { + assert_eq!(BOX_COLLECTION, "box.boxes"); + assert_eq!(JAIL_INDEX_COLLECTION, "box.jail_index"); +} + +#[tokio::test] +async fn a_v1_box_migrates_to_the_v2_body_under_the_same_id() { + let (registry, _backend, storage) = v1_storage(&[("tinybox_boxes", "build-1", v1_box())]).await; + registry + .migrate(&storage, &MigrateOptions::new()) + .await + .expect("migrate"); + let docs = storage.documents(); + let found = docs + .get("box.boxes", "build-1") + .await + .expect("get") + .expect("migrated"); + assert_eq!( + found.doc, + json!({ + "state": "ready", + "created_ms": 1_700_000_000_005_i64, + "spec": { + "runner": {"host": "local", "sandbox": "passthrough"}, + "workspace": {"host": "local", "sandbox": "passthrough"}, + "source": {"kind": "local_dir", "path": "/srv/work"}, + "resources": { + "cpu_millis": 2000, + "memory_bytes": 2_147_483_648_u64, + "pids_max": 512, + "disk_bytes": 8_589_934_592_u64, + }, + "lifecycle": {"kind": "ephemeral", "ttl_ms": 3_600_000}, + "network": {"kind": "allowlist", "hosts": ["example.test"], "ports": [443]}, + "ports": [{"guest": 80, "host": null}], + "env": {"KEY": "value"}, + }, + }) + ); + assert_eq!(found.meta.schema, 2); + assert!( + docs.get("tinybox_boxes", "build-1") + .await + .expect("get") + .is_none(), + "the v1 row is removed once the step completes" + ); +} + +#[tokio::test] +async fn a_v1_box_from_before_ports_and_times_migrates() { + let mut old = v1_box(); + old["spec"].as_object_mut().expect("spec").remove("ports"); + old.as_object_mut().expect("box").remove("created_at"); + let (registry, _backend, storage) = v1_storage(&[("tinybox_boxes", "build-1", old)]).await; + registry + .migrate(&storage, &MigrateOptions::new()) + .await + .expect("migrate"); + let found = storage + .documents() + .get("box.boxes", "build-1") + .await + .expect("get") + .expect("migrated"); + assert_eq!(found.doc["created_ms"], Value::Null); + assert_eq!(found.doc["spec"]["ports"], json!([])); +} + +#[tokio::test] +async fn a_v1_box_that_does_not_decode_stops_the_step_and_is_kept() { + let (registry, _backend, storage) = v1_storage(&[( + "tinybox_boxes", + "bad", + json!({"id": "bad", "state": "Levitating"}), + )]) + .await; + registry + .migrate(&storage, &MigrateOptions::new()) + .await + .expect_err("undecodable box"); + assert!( + storage + .documents() + .get("tinybox_boxes", "bad") + .await + .expect("get") + .is_some() + ); +} + +#[tokio::test] +async fn a_v1_jail_index_migrates_to_millisecond_times() { + let (registry, _backend, storage) = + v1_storage(&[("tinybox_jail_index", "main", v1_jail_index())]).await; + registry + .migrate(&storage, &MigrateOptions::new()) + .await + .expect("migrate"); + let docs = storage.documents(); + let found = docs + .get("box.jail_index", "main") + .await + .expect("get") + .expect("migrated"); + assert_eq!( + found.doc, + json!({ + "records": { + "j1": { + "id": "j1", + "label": "build", + "dir": "/srv/j1", + "backend_at_create": "landlock", + "created_ms": 1_700_000_000_000_i64, + "updated_ms": 1_700_000_060_000_i64, + "notes": "keep", + }, + }, + }) + ); + assert_eq!(found.meta.schema, 2); + assert!( + docs.get("tinybox_jail_index", "main") + .await + .expect("get") + .is_none() + ); +} + +#[tokio::test] +async fn migrating_twice_changes_nothing() { + let (registry, _backend, storage) = v1_storage(&[("tinybox_boxes", "build-1", v1_box())]).await; + registry + .migrate(&storage, &MigrateOptions::new()) + .await + .expect("first"); + let report = registry + .migrate(&storage, &MigrateOptions::new()) + .await + .expect("second"); + assert_eq!(report.ran.len(), 0); +} + +#[tokio::test] +async fn a_v1_jail_index_with_a_bad_time_stops_the_step_and_is_kept() { + let mut bad = v1_jail_index(); + bad["records"]["j1"]["created_at_unix"] = json!("yesterday"); + let (registry, _backend, storage) = v1_storage(&[("tinybox_jail_index", "main", bad)]).await; + registry + .migrate(&storage, &MigrateOptions::new()) + .await + .expect_err("bad time"); + let docs = storage.documents(); + assert!( + docs.get("tinybox_jail_index", "main") + .await + .expect("get") + .is_some() + ); + assert!( + docs.get("box.jail_index", "main") + .await + .expect("get") + .is_none() + ); +} + +#[test] +fn a_box_the_model_cannot_name_is_refused_by_the_step() { + let error = box_to_v2(SourceRecord { + id: "x".to_owned(), + body: json!({"id": "x"}), + meta: tinystoragedrivers_core::Meta::unknown(), + }) + .expect_err("incomplete box"); + assert_eq!( + error.kind(), + tinystoragedrivers_core::ErrorKind::Serialization + ); +} diff --git a/crates/tinybox-drivers/Cargo.toml b/crates/tinybox-drivers/Cargo.toml index 14874c3..68e52b3 100644 --- a/crates/tinybox-drivers/Cargo.toml +++ b/crates/tinybox-drivers/Cargo.toml @@ -12,9 +12,11 @@ repository.workspace = true publish = false [dependencies] +tinybox-bus = { workspace = true, features = ["storage"] } tinybox-core.workspace = true tinybox-jail.workspace = true tinystoragedrivers-core = { workspace = true, features = ["blocking"] } +serde.workspace = true serde_json.workspace = true log = "0.4" diff --git a/crates/tinybox-drivers/src/box_store.rs b/crates/tinybox-drivers/src/box_store.rs index 910ea60..cf16447 100644 --- a/crates/tinybox-drivers/src/box_store.rs +++ b/crates/tinybox-drivers/src/box_store.rs @@ -2,29 +2,26 @@ use std::sync::Arc; -use serde_json::Value; use tinybox_core::{BoxId, BoxInfo, BoxState, Error, Result, Store}; use tinystoragedrivers_core::{ - Blocking, CollectionSpec, DocumentStore, DocumentStoreExt, ErrorKind, Precondition, Query, - StorageError, Versioned, + Blocking, DocumentStore, ErrorKind, Key, Query, Repo, StorageError, WriteOptions, }; -/// The collection that holds one document per box, keyed by box id. -pub const BOX_COLLECTION: &str = "tinybox_boxes"; +use crate::record::{BoxRecord, StateRecord}; +use tinybox_bus::storage::{BODY_VERSION, BOX_COLLECTION}; -/// How many times a read-modify-write retries when another writer wins. -const CAS_ATTEMPTS: usize = 8; - -/// Box records as documents, `tinybox_boxes/`. +/// Box records as documents, `box.boxes/`. /// /// Inserts use an absent-only precondition, so two processes racing for one /// name resolve to a [`Error::DuplicateBox`] for the loser (which /// [`insert_new`](tinybox_core::store::insert_new) retries). State changes are -/// compare-and-swap on the document version, so they never overwrite a -/// concurrent change. +/// one compare-and-swap loop on the document version ([`Repo::update`]), so +/// they never overwrite a concurrent change. +/// +/// The collection is declared by the host, from [`storage_schema`](crate::storage_schema). #[derive(Clone)] pub struct DocumentBoxStore { - documents: Arc, + boxes: Repo, bridge: Blocking, } @@ -35,22 +32,22 @@ impl std::fmt::Debug for DocumentBoxStore { } impl DocumentBoxStore { - /// A store on `documents`; the collection is declared here. + /// A store on `documents`. /// /// # Errors /// - /// Returns [`Error::Store`] when the bridge cannot start or the backend - /// refuses the collection. + /// Returns [`Error::Store`] when the bridge cannot start. + #[allow( + clippy::needless_pass_by_value, + reason = "the handle is taken by value, as it always was" + )] pub fn new(documents: Arc) -> Result { let bridge = Blocking::new().map_err(|error| store_error("open", &error))?; - let store = Self { documents, bridge }; - let docs = Arc::clone(&store.documents); - store.run("open", async move { - docs.ensure_collection(&CollectionSpec::new(BOX_COLLECTION)) - .await - })?; + let boxes = Repo::new(&documents, BOX_COLLECTION) + .map_err(|error| store_error("open", &error))? + .with_options(WriteOptions::new().schema(BODY_VERSION)); log::debug!("[tinybox_drivers] box store ready collection={BOX_COLLECTION}"); - Ok(store) + Ok(Self { boxes, bridge }) } fn run(&self, operation: &'static str, future: F) -> Result @@ -63,15 +60,6 @@ impl DocumentBoxStore { .and_then(|inner| inner) .map_err(|error| store_error(operation, &error)) } - - fn fetch(&self, id: &BoxId) -> Result> { - let docs = Arc::clone(&self.documents); - let key = id.as_str().to_owned(); - self.run("read", async move { docs.get(BOX_COLLECTION, &key).await })? - .ok_or_else(|| Error::UnknownBox { - id: id.as_str().to_owned(), - }) - } } fn store_error(operation: &'static str, error: &StorageError) -> Error { @@ -81,25 +69,45 @@ fn store_error(operation: &'static str, error: &StorageError) -> Error { } } -fn decode(found: Value) -> Result { - serde_json::from_value(found).map_err(|error| Error::Store { +fn key(id: &BoxId) -> Result { + Key::of([id.as_str()]) + .map(|key| key.as_str().to_owned()) + .map_err(|error| store_error("key", &error)) +} + +fn unknown(id: &BoxId) -> Error { + Error::UnknownBox { + id: id.as_str().to_owned(), + } +} + +/// The record stored under document id `key`, as a model record. +fn decode(key: &str, record: BoxRecord) -> Result { + let id = BoxId::new(key).map_err(|error| Error::Store { operation: "decode", message: error.to_string(), - }) + })?; + Ok(record.into_info(id)) +} + +/// Document ids and bodies are read through serde, so a body that is not a +/// [`BoxRecord`] arrives as a serialization error; report it as a decode. +fn read_error(operation: &'static str, error: &StorageError) -> Error { + if error.kind() == ErrorKind::Serialization { + store_error("decode", error) + } else { + store_error(operation, error) + } } impl Store for DocumentBoxStore { fn insert(&self, info: &BoxInfo) -> Result<()> { - let doc = serde_json::to_value(info).map_err(|error| Error::Store { - operation: "encode", - message: error.to_string(), - })?; - let docs = Arc::clone(&self.documents); - let key = info.id.as_str().to_owned(); - let outcome = self.bridge.run(async move { - docs.put(BOX_COLLECTION, &key, doc, Precondition::Absent) - .await - }); + let record = BoxRecord::try_from(info)?; + let key = key(&info.id)?; + let boxes = self.boxes.clone(); + let outcome = self + .bridge + .run(async move { boxes.insert(&key, &record).await }); match outcome.and_then(|inner| inner) { Ok(_) => Ok(()), Err(error) @@ -114,62 +122,70 @@ impl Store for DocumentBoxStore { } fn get(&self, id: &BoxId) -> Result { - decode(self.fetch(id)?.doc) + let key = key(id)?; + let boxes = self.boxes.clone(); + let lookup = key.clone(); + let found = self + .bridge + .run(async move { boxes.get(&lookup).await }) + .and_then(|inner| inner) + .map_err(|error| read_error("read", &error))?; + let found = found.ok_or_else(|| unknown(id))?; + decode(&key, found.doc) } fn list(&self) -> Result> { - let docs = Arc::clone(&self.documents); - let found = self.run("list", async move { - docs.query_all(BOX_COLLECTION, &Query::default()).await - })?; + let boxes = self.boxes.clone(); + let found = self + .bridge + .run(async move { + let mut query = Query::default(); + let mut all = Vec::new(); + loop { + let page = boxes.query(&query).await?; + all.extend(page.items); + match page.next { + Some(cursor) => query.cursor = Some(cursor), + None => return Ok(all), + } + } + }) + .and_then(|inner: std::result::Result, StorageError>| inner) + .map_err(|error| read_error("list", &error))?; // Documents come back ordered by id; keep that so output is stable. - found.into_iter().map(|entry| decode(entry.doc)).collect() + found + .into_iter() + .map(|entry| decode(&entry.id, entry.doc)) + .collect() } fn set_state(&self, id: &BoxId, state: BoxState) -> Result<()> { - for _ in 0..CAS_ATTEMPTS { - let found = self.fetch(id)?; - let mut info = decode(found.doc)?; - info.state = state; - let doc = serde_json::to_value(&info).map_err(|error| Error::Store { - operation: "encode", - message: error.to_string(), - })?; - let docs = Arc::clone(&self.documents); - let key = id.as_str().to_owned(); - let version = found.version; - let outcome = self.bridge.run(async move { - docs.put(BOX_COLLECTION, &key, doc, Precondition::Version(version)) + let state = StateRecord::try_from(state)?; + let key = key(id)?; + let boxes = self.boxes.clone(); + let updated = self + .bridge + .run(async move { + boxes + .update(&key, |current| { + Ok(current.map(|mut record| { + record.state = state; + record + })) + }) .await - }); - match outcome.and_then(|inner| inner) { - Ok(_) => return Ok(()), - // Another writer changed the record first: read it again. - Err(error) if matches!(error.kind(), ErrorKind::Conflict) => { - log::trace!("[tinybox_drivers] set_state conflict, retrying"); - } - Err(error) => return Err(store_error("write", &error)), - } - } - Err(Error::Store { - operation: "write", - message: format!("the record kept changing after {CAS_ATTEMPTS} attempts"), - }) + }) + .and_then(|inner| inner) + .map_err(|error| read_error("update", &error))?; + // The loop leaves a missing record alone. + updated.map(|_| ()).ok_or_else(|| unknown(id)) } fn remove(&self, id: &BoxId) -> Result<()> { - let docs = Arc::clone(&self.documents); - let key = id.as_str().to_owned(); - let removed = self.run("remove", async move { - docs.delete(BOX_COLLECTION, &key, Precondition::None).await - })?; - if removed { - Ok(()) - } else { - Err(Error::UnknownBox { - id: id.as_str().to_owned(), - }) - } + let key = key(id)?; + let boxes = self.boxes.clone(); + let removed = self.run("remove", async move { boxes.delete(&key).await })?; + if removed { Ok(()) } else { Err(unknown(id)) } } } diff --git a/crates/tinybox-drivers/src/box_store_tests.rs b/crates/tinybox-drivers/src/box_store_tests.rs index a46242b..2ec8c42 100644 --- a/crates/tinybox-drivers/src/box_store_tests.rs +++ b/crates/tinybox-drivers/src/box_store_tests.rs @@ -3,7 +3,7 @@ use crate::faulty::{Faults, Faulty}; use serde_json::json; use tinybox_core::store::insert_new; use tinybox_core::{BoxSpec, HostRef, Placement, SandboxRef, WorkspaceSource}; -use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; +use tinystoragedrivers_core::{DocumentStore, MemoryStorage, Precondition, Scope, StorageBackend}; fn documents() -> Arc { MemoryStorage::new() @@ -193,25 +193,6 @@ fn tokio_rt() -> tokio::runtime::Runtime { .expect("runtime") } -fn faulty(faults: Faults) -> Arc { - Faulty::new(documents(), faults) -} - -#[test] -fn a_backend_that_refuses_the_collection_fails_to_open() { - let result = DocumentBoxStore::new(faulty(Faults { - ensure: true, - ..Faults::default() - })); - assert!(matches!( - result, - Err(Error::Store { - operation: "open", - .. - }) - )); -} - #[test] fn backend_failures_surface_as_store_errors_per_operation() -> Result<()> { let seeded = documents(); @@ -235,7 +216,7 @@ fn backend_failures_surface_as_store_errors_per_operation() -> Result<()> { assert!(matches!( reading.set_state(&record.id, BoxState::Stopped), Err(Error::Store { - operation: "read", + operation: "update", .. }) )); @@ -266,7 +247,7 @@ fn backend_failures_surface_as_store_errors_per_operation() -> Result<()> { assert!(matches!( writing.set_state(&record.id, BoxState::Stopped), Err(Error::Store { - operation: "write", + operation: "update", .. }) )); @@ -305,7 +286,7 @@ fn set_state_gives_up_when_the_record_never_settles() -> Result<()> { assert!(matches!( store.set_state(&record.id, BoxState::Stopped), Err(Error::Store { - operation: "write", + operation: "update", .. }) )); @@ -318,3 +299,45 @@ fn the_debug_form_does_not_dump_the_backend() -> Result<()> { assert!(format!("{store:?}").starts_with("DocumentBoxStore")); Ok(()) } + +#[test] +fn a_stored_box_has_the_v2_body_and_schema_version() -> Result<()> { + let docs = documents(); + let store = DocumentBoxStore::new(Arc::clone(&docs))?; + store.insert(&info("build-1")?)?; + let bridge = Blocking::new().expect("bridge"); + let reader = Arc::clone(&docs); + let stored = bridge + .run(async move { reader.get(BOX_COLLECTION, "build-1").await }) + .expect("bridge") + .expect("get") + .expect("stored"); + assert_eq!(stored.meta.schema, 2); + assert_eq!(stored.doc["state"], json!("ready")); + assert_eq!(stored.doc["spec"]["lifecycle"]["kind"], json!("ephemeral")); + assert!(stored.doc.get("id").is_none(), "the id is the document id"); + Ok(()) +} + +#[test] +fn a_box_missing_from_a_set_state_is_left_missing() -> Result<()> { + let docs = documents(); + let store = DocumentBoxStore::new(Arc::clone(&docs))?; + let id = BoxId::new("ghost")?; + assert!(matches!( + store.set_state(&id, BoxState::Stopped), + Err(Error::UnknownBox { .. }) + )); + assert_eq!(store.list()?.len(), 0); + Ok(()) +} + +#[test] +fn more_boxes_than_one_page_all_list() -> Result<()> { + let store = DocumentBoxStore::new(documents())?; + for n in 0..120 { + store.insert(&info(&format!("b{n:03}"))?)?; + } + assert_eq!(store.list()?.len(), 120); + Ok(()) +} diff --git a/crates/tinybox-drivers/src/faulty_tests.rs b/crates/tinybox-drivers/src/faulty_tests.rs index 59343dd..bc2f2fe 100644 --- a/crates/tinybox-drivers/src/faulty_tests.rs +++ b/crates/tinybox-drivers/src/faulty_tests.rs @@ -8,7 +8,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use serde_json::Value; use tinystoragedrivers_core::{ Capabilities, CollectionSpec, DocumentStore, Filter, Page, Precondition, Query, Sort, - StorageError, Version, Versioned, async_trait, + StorageError, Version, Versioned, WriteOptions, async_trait, }; type Result = std::result::Result; @@ -83,6 +83,18 @@ impl DocumentStore for Faulty { id: &str, doc: Value, precondition: Precondition, + ) -> Result { + self.put_with(collection, id, doc, precondition, WriteOptions::default()) + .await + } + + async fn put_with( + &self, + collection: &str, + id: &str, + doc: Value, + precondition: Precondition, + options: WriteOptions, ) -> Result { if self.faults.put { return Err(boom("put")); @@ -96,7 +108,9 @@ impl DocumentStore for Faulty { return Err(StorageError::conflict("injected conflict")); } } - self.inner.put(collection, id, doc, precondition).await + self.inner + .put_with(collection, id, doc, precondition, options) + .await } async fn delete(&self, collection: &str, id: &str, precondition: Precondition) -> Result { diff --git a/crates/tinybox-drivers/src/index_store.rs b/crates/tinybox-drivers/src/index_store.rs index 52a08db..53e68b5 100644 --- a/crates/tinybox-drivers/src/index_store.rs +++ b/crates/tinybox-drivers/src/index_store.rs @@ -6,21 +6,25 @@ use std::sync::Arc; use serde_json::Value; use tinybox_jail::IndexStore; use tinystoragedrivers_core::{ - Blocking, CollectionSpec, DocumentStore, ErrorKind, Precondition, StorageError, + Blocking, DocumentStore, ErrorKind, Key, Repo, StorageError, WriteOptions, }; -/// The collection that holds one index document per registry. -pub const JAIL_INDEX_COLLECTION: &str = "tinybox_jail_index"; +use tinybox_bus::storage::jail_index as index_body; +use tinybox_bus::storage::{BODY_VERSION, JAIL_INDEX_COLLECTION}; -/// A jail registry's index as one document, `tinybox_jail_index/`. +/// A jail registry's index as one document, `box.jail_index/`. /// -/// The registry serializes the index; this keeps the JSON value, so the -/// document has the same shape as the legacy `index.json`. A document that is -/// not valid JSON reads as [`io::ErrorKind::InvalidData`]. +/// The registry serializes the index in its own form (the legacy `index.json` +/// shape); this keeps the equivalent v2 body: the same records with +/// millisecond times and no version field (see the `index_body` module). A +/// document that is not valid JSON reads as [`io::ErrorKind::InvalidData`]. +/// +/// The collection is declared by the host, from [`storage_schema`](crate::storage_schema). #[derive(Clone)] pub struct DocumentIndexStore { - documents: Arc, + index: Repo, bridge: Blocking, + key: String, registry: String, } @@ -33,27 +37,31 @@ impl std::fmt::Debug for DocumentIndexStore { } impl DocumentIndexStore { - /// An index store for the registry named `registry`; the collection is - /// declared here. + /// An index store for the registry named `registry`. /// /// # Errors /// - /// Returns an error when the bridge cannot start or the backend refuses - /// the collection. + /// Returns an error when the bridge cannot start or `registry` cannot be a + /// document id. + #[allow( + clippy::needless_pass_by_value, + reason = "the handle is taken by value, as it always was" + )] pub fn new(documents: Arc, registry: impl Into) -> io::Result { + let registry = registry.into(); let bridge = Blocking::new().map_err(|e| to_io(&e))?; - let docs = Arc::clone(&documents); - bridge - .run(async move { - docs.ensure_collection(&CollectionSpec::new(JAIL_INDEX_COLLECTION)) - .await - }) - .and_then(|inner| inner) - .map_err(|e| to_io(&e))?; + let key = Key::of([registry.as_str()]) + .map_err(|e| to_io(&e))? + .as_str() + .to_owned(); + let index = Repo::new(&documents, JAIL_INDEX_COLLECTION) + .map_err(|e| to_io(&e))? + .with_options(WriteOptions::new().schema(BODY_VERSION)); Ok(Self { - documents, + index, bridge, - registry: registry.into(), + key, + registry, }) } } @@ -69,31 +77,29 @@ fn to_io(error: &StorageError) -> io::Error { impl IndexStore for DocumentIndexStore { fn load(&self) -> io::Result>> { - let docs = Arc::clone(&self.documents); - let key = self.registry.clone(); + let index = self.index.clone(); + let key = self.key.clone(); let found = self .bridge - .run(async move { docs.get(JAIL_INDEX_COLLECTION, &key).await }) + .run(async move { index.get(&key).await }) .and_then(|inner| inner) .map_err(|e| to_io(&e))?; found .map(|entry| { - serde_json::to_vec(&entry.doc) + serde_json::to_vec(&index_body::from_body(entry.doc)) .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)) }) .transpose() } fn save(&self, bytes: &[u8]) -> io::Result<()> { - let doc: Value = serde_json::from_slice(bytes) + let registry_form: Value = serde_json::from_slice(bytes) .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; - let docs = Arc::clone(&self.documents); - let key = self.registry.clone(); + let body = index_body::to_body(registry_form)?; + let index = self.index.clone(); + let key = self.key.clone(); self.bridge - .run(async move { - docs.put(JAIL_INDEX_COLLECTION, &key, doc, Precondition::None) - .await - }) + .run(async move { index.put(&key, &body).await }) .and_then(|inner| inner) .map(|_| ()) .map_err(|e| to_io(&e)) diff --git a/crates/tinybox-drivers/src/index_store_tests.rs b/crates/tinybox-drivers/src/index_store_tests.rs index 29fe91f..64f23de 100644 --- a/crates/tinybox-drivers/src/index_store_tests.rs +++ b/crates/tinybox-drivers/src/index_store_tests.rs @@ -81,17 +81,7 @@ fn a_registry_keeps_its_jails_across_reopen_without_an_index_file() { #[test] fn backend_failures_surface_as_io_errors() { - let refusing = DocumentIndexStore::new( - Faulty::new( - documents(), - Faults { - ensure: true, - ..Faults::default() - }, - ), - "main", - ); - assert!(refusing.is_err()); + assert!(DocumentIndexStore::new(documents(), "").is_err()); let reading = DocumentIndexStore::new( Faulty::new( @@ -147,3 +137,26 @@ fn the_debug_form_names_the_registry() { let store = DocumentIndexStore::new(documents(), "main").expect("store"); assert!(format!("{store:?}").contains("main")); } + +#[test] +fn a_saved_index_is_stored_with_millisecond_times_and_the_v2_schema() { + let docs = documents(); + let store = DocumentIndexStore::new(Arc::clone(&docs), "main").expect("store"); + store + .save( + br#"{"records":{"j1":{"id":"j1","created_at_unix":2,"updated_at_unix":3}},"schema_version":1}"#, + ) + .expect("save"); + let bridge = tinystoragedrivers_core::Blocking::new().expect("bridge"); + let reader = Arc::clone(&docs); + let stored = bridge + .run(async move { reader.get(JAIL_INDEX_COLLECTION, "main").await }) + .expect("bridge") + .expect("get") + .expect("stored"); + assert_eq!(stored.meta.schema, 2); + assert_eq!( + stored.doc, + serde_json::json!({"records": {"j1": {"id": "j1", "created_ms": 2000, "updated_ms": 3000}}}) + ); +} diff --git a/crates/tinybox-drivers/src/lib.rs b/crates/tinybox-drivers/src/lib.rs index d9d60cb..8fec0bc 100644 --- a/crates/tinybox-drivers/src/lib.rs +++ b/crates/tinybox-drivers/src/lib.rs @@ -9,6 +9,12 @@ //! - [`DocumentIndexStore`]: the jail registry's index as one document, behind //! [`tinybox_jail::IndexStore`]. //! +//! - [`storage_schema`]: defined in `tinybox_bus::storage`; the `box` owner's collections (`box.boxes`, +//! `box.jail_index`) and the v1 to v2 steps. The host composes it into its +//! registry, declares the collections, runs the migration, and grants the +//! module exactly these items (`ScopeClass::Module`). The stores here never +//! declare a collection themselves. +//! //! Both traits are synchronous, so each store owns a [`Blocking`] bridge that //! runs the port's futures on a dedicated runtime thread. That works from inside //! or outside a tokio runtime. @@ -18,9 +24,11 @@ mod box_store; mod index_store; +mod record; -pub use box_store::{BOX_COLLECTION, DocumentBoxStore}; -pub use index_store::{DocumentIndexStore, JAIL_INDEX_COLLECTION}; +pub use box_store::DocumentBoxStore; +pub use index_store::DocumentIndexStore; +pub use tinybox_bus::storage::{BOX_COLLECTION, JAIL_INDEX_COLLECTION, storage_schema}; #[cfg(test)] #[path = "faulty_tests.rs"] diff --git a/crates/tinybox-drivers/src/record.rs b/crates/tinybox-drivers/src/record.rs new file mode 100644 index 0000000..fa944ef --- /dev/null +++ b/crates/tinybox-drivers/src/record.rs @@ -0,0 +1,287 @@ +//! The v2 storage body of a box record. +//! +//! [`BoxInfo`] and [`BoxSpec`] are model types: their serde form is shared with +//! templates and snapshot manifests, and it uses `Duration` and `SystemTime` +//! encodings and `PascalCase` variant names. The document body is its own type +//! so the stored shape follows the schema (`snake_case`, `*_ms` times, tagged +//! variants) without touching the model's wire form. +//! +//! The box id is the document id and is not repeated in the body. + +use std::collections::{BTreeMap, BTreeSet}; +use std::path::PathBuf; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; +use tinybox_core::{ + BoxId, BoxInfo, BoxSpec, BoxState, EgressAllowlist, Error, Lifecycle, NetworkPolicy, Placement, + PortMapping, Resources, Result, SnapshotId, WorkspaceSource, +}; + +/// One box as a `box.boxes` document body. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct BoxRecord { + pub(crate) state: StateRecord, + pub(crate) spec: SpecRecord, + /// When the box was created, if known (Unix epoch ms). + /// + /// This is not the document's `created_ms` metadata: the driver always + /// stamps that, but a box recorded before tinybox tracked time has no + /// creation time, and "unknown" must stay distinguishable from "now" or + /// the reaper would treat a migrated box as brand new (or ancient). + pub(crate) created_ms: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum StateRecord { + Creating, + Ready, + Running, + Paused, + Stopped, + Archived, + Failed, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct SpecRecord { + runner: Placement, + workspace: Placement, + source: SourceRecord, + resources: Resources, + lifecycle: LifecycleRecord, + network: NetworkRecord, + ports: BTreeSet, + env: BTreeMap, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum SourceRecord { + LocalDir { path: PathBuf }, + OciImage { image: String }, + Snapshot { snapshot: SnapshotId }, + GitRepo { url: String, rev: String }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum LifecycleRecord { + Ephemeral { ttl_ms: u64 }, + Persistent { autosnapshot_ms: Option }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum NetworkRecord { + Denied, + Egress, + Allowlist { hosts: Vec, ports: Vec }, + Open, +} + +fn unsupported(what: &str) -> Error { + Error::Store { + operation: "encode", + message: format!("this version cannot store a {what} it does not know"), + } +} + +fn millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + +fn system_time_ms(at: SystemTime) -> i64 { + match at.duration_since(UNIX_EPOCH) { + Ok(after) => i64::try_from(after.as_millis()).unwrap_or(i64::MAX), + Err(before) => i64::try_from(before.duration().as_millis()).map_or(i64::MIN, |ms| -ms), + } +} + +fn system_time_from_ms(ms: i64) -> SystemTime { + let magnitude = Duration::from_millis(ms.unsigned_abs()); + if ms >= 0 { + UNIX_EPOCH + magnitude + } else { + UNIX_EPOCH - magnitude + } +} + +impl TryFrom<&BoxInfo> for BoxRecord { + type Error = Error; + + fn try_from(info: &BoxInfo) -> Result { + Ok(Self { + state: StateRecord::try_from(info.state)?, + spec: SpecRecord::try_from(&info.spec)?, + created_ms: info.created_at.map(system_time_ms), + }) + } +} + +impl BoxRecord { + /// The model record for the box stored under `id`. + pub(crate) fn into_info(self, id: BoxId) -> BoxInfo { + let mut info = BoxInfo::new(id, self.state.into(), self.spec.into_spec()); + info.created_at = self.created_ms.map(system_time_from_ms); + info + } +} + +impl TryFrom for StateRecord { + type Error = Error; + + fn try_from(state: BoxState) -> Result { + Ok(match state { + BoxState::Creating => Self::Creating, + BoxState::Ready => Self::Ready, + BoxState::Running => Self::Running, + BoxState::Paused => Self::Paused, + BoxState::Stopped => Self::Stopped, + BoxState::Archived => Self::Archived, + BoxState::Failed => Self::Failed, + _ => return Err(unsupported("box state")), + }) + } +} + +impl From for BoxState { + fn from(state: StateRecord) -> Self { + match state { + StateRecord::Creating => Self::Creating, + StateRecord::Ready => Self::Ready, + StateRecord::Running => Self::Running, + StateRecord::Paused => Self::Paused, + StateRecord::Stopped => Self::Stopped, + StateRecord::Archived => Self::Archived, + StateRecord::Failed => Self::Failed, + } + } +} + +impl TryFrom<&BoxSpec> for SpecRecord { + type Error = Error; + + fn try_from(spec: &BoxSpec) -> Result { + Ok(Self { + runner: spec.runner.clone(), + workspace: spec.workspace.clone(), + source: SourceRecord::try_from(&spec.source)?, + resources: spec.resources, + lifecycle: LifecycleRecord::try_from(&spec.lifecycle)?, + network: NetworkRecord::try_from(&spec.network)?, + ports: spec.ports.clone(), + env: spec.env.clone(), + }) + } +} + +impl SpecRecord { + fn into_spec(self) -> BoxSpec { + let mut spec = BoxSpec::new(self.workspace, self.source.into()); + spec.runner = self.runner; + spec.resources = self.resources; + spec.lifecycle = self.lifecycle.into(); + spec.network = self.network.into(); + spec.ports = self.ports; + spec.env = self.env; + spec + } +} + +impl TryFrom<&WorkspaceSource> for SourceRecord { + type Error = Error; + + fn try_from(source: &WorkspaceSource) -> Result { + Ok(match source { + WorkspaceSource::LocalDir(path) => Self::LocalDir { path: path.clone() }, + WorkspaceSource::OciImage(image) => Self::OciImage { + image: image.clone(), + }, + WorkspaceSource::Snapshot(snapshot) => Self::Snapshot { + snapshot: snapshot.clone(), + }, + WorkspaceSource::GitRepo { url, rev } => Self::GitRepo { + url: url.clone(), + rev: rev.clone(), + }, + _ => return Err(unsupported("workspace source")), + }) + } +} + +impl From for WorkspaceSource { + fn from(source: SourceRecord) -> Self { + match source { + SourceRecord::LocalDir { path } => Self::LocalDir(path), + SourceRecord::OciImage { image } => Self::OciImage(image), + SourceRecord::Snapshot { snapshot } => Self::Snapshot(snapshot), + SourceRecord::GitRepo { url, rev } => Self::GitRepo { url, rev }, + } + } +} + +impl TryFrom<&Lifecycle> for LifecycleRecord { + type Error = Error; + + fn try_from(lifecycle: &Lifecycle) -> Result { + Ok(match lifecycle { + Lifecycle::Ephemeral { ttl } => Self::Ephemeral { + ttl_ms: millis(*ttl), + }, + Lifecycle::Persistent { autosnapshot } => Self::Persistent { + autosnapshot_ms: autosnapshot.map(millis), + }, + _ => return Err(unsupported("lifecycle")), + }) + } +} + +impl From for Lifecycle { + fn from(lifecycle: LifecycleRecord) -> Self { + match lifecycle { + LifecycleRecord::Ephemeral { ttl_ms } => Self::Ephemeral { + ttl: Duration::from_millis(ttl_ms), + }, + LifecycleRecord::Persistent { autosnapshot_ms } => Self::Persistent { + autosnapshot: autosnapshot_ms.map(Duration::from_millis), + }, + } + } +} + +impl TryFrom<&NetworkPolicy> for NetworkRecord { + type Error = Error; + + fn try_from(network: &NetworkPolicy) -> Result { + Ok(match network { + NetworkPolicy::Denied => Self::Denied, + NetworkPolicy::Egress => Self::Egress, + NetworkPolicy::Allowlist(list) => Self::Allowlist { + hosts: list.hosts.clone(), + ports: list.ports.clone(), + }, + NetworkPolicy::Open => Self::Open, + _ => return Err(unsupported("network policy")), + }) + } +} + +impl From for NetworkPolicy { + fn from(network: NetworkRecord) -> Self { + match network { + NetworkRecord::Denied => Self::Denied, + NetworkRecord::Egress => Self::Egress, + NetworkRecord::Allowlist { hosts, ports } => { + Self::Allowlist(EgressAllowlist::new(hosts, ports)) + } + NetworkRecord::Open => Self::Open, + } + } +} + +#[cfg(test)] +#[path = "record_tests.rs"] +#[allow(clippy::expect_used, clippy::unwrap_used)] +mod tests; diff --git a/crates/tinybox-drivers/src/record_tests.rs b/crates/tinybox-drivers/src/record_tests.rs new file mode 100644 index 0000000..6d334fc --- /dev/null +++ b/crates/tinybox-drivers/src/record_tests.rs @@ -0,0 +1,171 @@ +//! Tests for the `BoxRecord` v2 body encoding and its match with the v1 migration. + +use super::*; +use serde_json::json; +use std::time::{Duration, UNIX_EPOCH}; +use tinybox_core::{ + BoxId, EgressAllowlist, HostRef, Lifecycle, NetworkPolicy, Placement, PortMapping, SandboxRef, + SnapshotId, WorkspaceSource, +}; + +fn base_spec(source: WorkspaceSource) -> Result { + Ok(BoxSpec::new( + Placement::new(HostRef::new("local")?, SandboxRef::new("passthrough")?), + source, + )) +} + +fn info_with(spec: BoxSpec) -> Result { + Ok(BoxInfo::new(BoxId::new("b1")?, BoxState::Ready, spec)) +} + +fn round_trip(info: &BoxInfo) -> Result { + let record = BoxRecord::try_from(info)?; + let value = serde_json::to_value(&record).map_err(|e| Error::Store { + operation: "encode", + message: e.to_string(), + })?; + let back: BoxRecord = serde_json::from_value(value).map_err(|e| Error::Store { + operation: "decode", + message: e.to_string(), + })?; + Ok(back.into_info(info.id.clone())) +} + +#[test] +fn the_body_is_snake_case_with_millisecond_times_and_no_id() -> Result<()> { + let spec = base_spec(WorkspaceSource::LocalDir("/srv/work".into()))? + .with_env("KEY", "value") + .with_lifecycle(Lifecycle::Persistent { + autosnapshot: Some(Duration::from_secs(60)), + }); + let info = info_with(spec)?.created_at(UNIX_EPOCH + Duration::from_millis(1_700_000_000_123)); + let body = serde_json::to_value(BoxRecord::try_from(&info)?).expect("json"); + assert_eq!( + body, + json!({ + "state": "ready", + "created_ms": 1_700_000_000_123_i64, + "spec": { + "runner": {"host": "local", "sandbox": "passthrough"}, + "workspace": {"host": "local", "sandbox": "passthrough"}, + "source": {"kind": "local_dir", "path": "/srv/work"}, + "resources": { + "cpu_millis": 2000, + "memory_bytes": 2_147_483_648_u64, + "pids_max": 512, + "disk_bytes": 8_589_934_592_u64, + }, + "lifecycle": {"kind": "persistent", "autosnapshot_ms": 60_000}, + "network": {"kind": "denied"}, + "ports": [], + "env": {"KEY": "value"}, + }, + }) + ); + Ok(()) +} + +#[test] +fn every_source_round_trips() -> Result<()> { + for source in [ + WorkspaceSource::LocalDir("/w".into()), + WorkspaceSource::OciImage("alpine:3".into()), + WorkspaceSource::Snapshot(SnapshotId::new("snap-1")?), + WorkspaceSource::GitRepo { + url: "https://example.test/r.git".into(), + rev: "main".into(), + }, + ] { + let info = info_with(base_spec(source)?)?; + assert_eq!(round_trip(&info)?, info); + } + Ok(()) +} + +#[test] +fn every_network_policy_round_trips() -> Result<()> { + for network in [ + NetworkPolicy::Denied, + NetworkPolicy::Egress, + NetworkPolicy::Allowlist(EgressAllowlist::new(["example.test"], [443])), + NetworkPolicy::Open, + ] { + let info = + info_with(base_spec(WorkspaceSource::OciImage("a".into()))?.with_network(network))?; + assert_eq!(round_trip(&info)?, info); + } + Ok(()) +} + +#[test] +fn every_lifecycle_state_and_port_round_trips() -> Result<()> { + for lifecycle in [ + Lifecycle::Ephemeral { + ttl: Duration::from_secs(90), + }, + Lifecycle::Persistent { autosnapshot: None }, + Lifecycle::persistent(), + ] { + for state in [ + BoxState::Creating, + BoxState::Ready, + BoxState::Running, + BoxState::Paused, + BoxState::Stopped, + BoxState::Archived, + BoxState::Failed, + ] { + let spec = base_spec(WorkspaceSource::OciImage("a".into()))? + .with_lifecycle(lifecycle.clone()) + .with_port(PortMapping::fixed(80, 8080)) + .with_port(PortMapping::dynamic(22)); + let info = BoxInfo::new(BoxId::new("b1")?, state, spec); + assert_eq!(round_trip(&info)?, info); + } + } + Ok(()) +} + +#[test] +fn an_unknown_creation_time_stays_unknown() -> Result<()> { + let info = info_with(base_spec(WorkspaceSource::OciImage("a".into()))?)?; + let record = BoxRecord::try_from(&info)?; + assert_eq!(record.created_ms, None); + assert_eq!(round_trip(&info)?.created_at, None); + Ok(()) +} + +#[test] +fn a_creation_time_keeps_millisecond_precision() -> Result<()> { + let at = UNIX_EPOCH + Duration::from_millis(42_007); + let info = info_with(base_spec(WorkspaceSource::OciImage("a".into()))?)?.created_at(at); + assert_eq!(round_trip(&info)?.created_at, Some(at)); + Ok(()) +} + +#[test] +fn a_state_name_this_version_does_not_know_is_a_decode_error() { + let body = json!({"state": "levitating", "spec": {}, "created_ms": null}); + assert!(serde_json::from_value::(body).is_err()); +} + +#[test] +fn the_contracts_v1_migration_matches_the_record_encoding() { + let mut spec = base_spec(WorkspaceSource::LocalDir("/srv/w".into())).expect("spec"); + spec.lifecycle = Lifecycle::Ephemeral { + ttl: Duration::from_secs(60), + }; + spec.network = NetworkPolicy::Allowlist(EgressAllowlist::new(["a.test"], vec![443])); + spec.ports.insert(PortMapping { + guest: 80, + host: None, + }); + let info = info_with(spec) + .expect("info") + .created_at(UNIX_EPOCH + Duration::from_millis(1_700_000_000_005)); + let v1 = serde_json::to_value(&info).expect("v1 row"); + let migrated = tinybox_bus::storage::box_v1::to_body(v1).expect("migrates"); + let stored = serde_json::to_value(BoxRecord::try_from(&info).expect("record")).expect("json"); + assert_eq!(migrated, stored); +} diff --git a/crates/tinybox-module/src/tinybus_module/storage_tests.rs b/crates/tinybox-module/src/tinybus_module/storage_tests.rs index d252fe1..3228e7c 100644 --- a/crates/tinybox-module/src/tinybus_module/storage_tests.rs +++ b/crates/tinybox-module/src/tinybus_module/storage_tests.rs @@ -25,6 +25,10 @@ async fn rig(serve_storage: bool) -> Result> { let bus = MemoryBus::new(); Broker::new().spawn(bus.clone()); let backend = MemoryStorage::new(); + // The host declares what the module may store from the module's own schema. + tinystoragedrivers_core::schema::Registry::compose([tinybox_drivers::storage_schema()])? + .prepare(&backend) + .await?; let host = Connection::connect(bus.connect().await?).await?; let served = if serve_storage { let bound = backend.clone(); diff --git a/deny.toml b/deny.toml index e04239c..3fea948 100644 --- a/deny.toml +++ b/deny.toml @@ -7,7 +7,11 @@ all-features = true [advisories] # Fail on any crate with a security advisory or an unmaintained warning. # Add an entry here only with a comment explaining the exposure and the plan. -ignore = [] +# The three `zip` advisories below concern `ZipArchive::extract` and +# `ZipStreamReader::extract`. zip arrives only through tinybus (vendored), and +# nothing in this workspace or vendor/tinybus calls an extract API (checked by +# grep). Remove when tinybus moves to zip >= 9. +ignore = ["RUSTSEC-2026-0367", "RUSTSEC-2026-0368", "RUSTSEC-2026-0369"] [licenses] # Licenses accepted for this crate and its dependencies. Keep GPL-3.0-only for