diff --git a/Cargo.lock b/Cargo.lock index 4c79e562d..5daa6d372 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2010,7 +2010,9 @@ dependencies = [ "fabro-macros", "serde", "serde_json", + "sha2", "ulid", + "uuid", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 0403b3ee0..e7961d863 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,7 +19,7 @@ serde_json = { version = "1", features = ["preserve_order"] } tokio = { version = "1", features = ["full"] } reqwest = { version = "0.13", default-features = false, features = ["json", "stream", "rustls", "query", "form"] } ulid = "1" -uuid = { version = "1", features = ["v4", "v7"] } +uuid = { version = "1", features = ["v4", "v7", "v8"] } rand = "0.8" dotenvy = "0.15" futures = "0.3" diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index f9ce6e11d..f7dc741b6 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -521,12 +521,8 @@ mod tests { ) .await .unwrap(); - run.put_artifact_value("summary", &serde_json::json!({"done": true})) - .await - .unwrap(); - run.put_artifact_value("plan", &serde_json::json!({"steps": 3})) - .await - .unwrap(); + let summary_blob = run.write_blob(br#"{"done":true}"#).await.unwrap(); + let plan_blob = run.write_blob(br#"{"steps":3}"#).await.unwrap(); run.put_asset(&node, "src/lib.rs", b"fn main() {}") .await .unwrap(); @@ -599,12 +595,14 @@ mod tests { assert_eq!(first_checkpoint.current_node, "plan"); assert_eq!(second_checkpoint.current_node, "code"); - let exported_plan: serde_json::Value = - read_json(&output.path().join("artifacts/values/plan.json")); - let exported_summary: serde_json::Value = - read_json(&output.path().join("artifacts/values/summary.json")); - assert_eq!(exported_plan, serde_json::json!({"steps": 3})); - assert_eq!(exported_summary, serde_json::json!({"done": true})); + assert_eq!( + std::fs::read(output.path().join("blobs").join(plan_blob.to_string())).unwrap(), + br#"{"steps":3}"# + ); + assert_eq!( + std::fs::read(output.path().join("blobs").join(summary_blob.to_string())).unwrap(), + br#"{"done":true}"# + ); assert_eq!( std::fs::read( diff --git a/lib/crates/fabro-store/src/keys.rs b/lib/crates/fabro-store/src/keys.rs index 25b81ff13..2669a0bc8 100644 --- a/lib/crates/fabro-store/src/keys.rs +++ b/lib/crates/fabro-store/src/keys.rs @@ -1,8 +1,9 @@ use crate::StageId; +use fabro_types::RunBlobId; pub(crate) const INIT_KEY: &str = "_init.json"; pub(crate) const EVENTS_PREFIX: &str = "events#"; -pub(crate) const ARTIFACT_VALUES_PREFIX: &str = "artifacts#values#"; +pub(crate) const BLOBS_PREFIX: &str = "blobs#"; pub(crate) const ARTIFACT_NODES_PREFIX: &str = "artifacts#nodes#"; pub(crate) fn init() -> &'static str { @@ -13,8 +14,8 @@ pub(crate) fn event_key(seq: u32, epoch_ms: i64) -> String { format!("{EVENTS_PREFIX}{seq:06}-{epoch_ms}.json") } -pub(crate) fn artifact_value(artifact_id: &str) -> String { - format!("{ARTIFACT_VALUES_PREFIX}{artifact_id}.json") +pub(crate) fn blob_key(id: &RunBlobId) -> String { + format!("{BLOBS_PREFIX}{id}") } pub(crate) fn node_asset_prefix(node: &StageId) -> String { @@ -33,10 +34,8 @@ pub(crate) fn parse_event_seq(key: &str) -> Option { parse_seq(key, EVENTS_PREFIX) } -pub(crate) fn parse_artifact_value_id(key: &str) -> Option { - key.strip_prefix(ARTIFACT_VALUES_PREFIX) - .and_then(|s| s.strip_suffix(".json")) - .map(ToString::to_string) +pub(crate) fn parse_blob_id(key: &str) -> Option { + key.strip_prefix(BLOBS_PREFIX)?.parse().ok() } pub(crate) fn parse_node_asset_key(key: &str) -> Option<(StageId, String)> { @@ -72,7 +71,8 @@ mod tests { #[test] fn artifact_keys_match_spec() { let node = StageId::new("code", 2); - assert_eq!(artifact_value("summary"), "artifacts#values#summary.json"); + let blob_id = RunBlobId::new(&"01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(), b"summary"); + assert_eq!(blob_key(&blob_id), format!("blobs#{blob_id}")); assert_eq!( node_asset(&node, "src/main.rs"), "artifacts#nodes#code#visit-2#src/main.rs" @@ -82,10 +82,8 @@ mod tests { #[test] fn parse_helpers_extract_sequences_and_node_visits() { assert_eq!(parse_event_seq("events#000007-123.json"), Some(7)); - assert_eq!( - parse_artifact_value_id("artifacts#values#summary.json"), - Some("summary".to_string()) - ); + let blob_id = RunBlobId::new(&"01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(), b"summary"); + assert_eq!(parse_blob_id(&format!("blobs#{blob_id}")), Some(blob_id)); assert_eq!( parse_node_asset_key("artifacts#nodes#code#visit-2#src/main.rs"), Some((StageId::new("code", 2), "src/main.rs".to_string())) @@ -95,10 +93,7 @@ mod tests { #[test] fn parse_helpers_reject_invalid_keys() { assert_eq!(parse_event_seq("events#not-a-seq.json"), None); - assert_eq!( - parse_artifact_value_id("artifacts#values#summary.txt"), - None - ); + assert_eq!(parse_blob_id("blobs#not-a-uuid"), None); assert_eq!( parse_node_asset_key("artifacts#nodes#code#status.json"), None diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index f655df7cc..7b47df0da 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -10,7 +10,7 @@ mod slate; mod types; pub use error::{Result, StoreError}; -pub use fabro_types::StageId; +pub use fabro_types::{RunBlobId, StageId}; pub use run_state::{NodeState, RunProjection}; pub use runtime::RuntimeState; pub use slate::{NodeAsset, SlateRunStore, SlateStore}; diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index d6b06b693..4c8eb0c49 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -955,16 +955,15 @@ mod tests { let (object_store, store) = make_store(); let _created_at = dt("2026-03-27T12:00:00Z"); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); - run.put_artifact_value("summary", &serde_json::json!({"done": true})) - .await - .unwrap(); + let blob_id = run.write_blob(br#"{"done":true}"#).await.unwrap(); + assert_eq!( + run.read_blob(&blob_id).await.unwrap(), + Some(Bytes::from_static(br#"{"done":true}"#)) + ); store.delete_run(&test_run_id("run-1")).await.unwrap(); - let err = run - .put_artifact_value("summary", &serde_json::json!({"done": false})) - .await - .unwrap_err(); + let err = run.write_blob(br#"{"done":false}"#).await.unwrap_err(); assert!(matches!( err, StoreError::Slate(err) if matches!(err.kind(), ErrorKind::Closed(CloseReason::Clean)) @@ -979,9 +978,7 @@ mod tests { let _created_at = dt("2026-03-27T12:00:00Z"); let _wrong_time = dt("2026-03-27T11:00:00Z"); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); - run.put_artifact_value("summary", &serde_json::json!({"done": true})) - .await - .unwrap(); + run.write_blob(br#"{"done":true}"#).await.unwrap(); let _locator = catalog::read_locator(object_store.clone(), "runs/", &test_run_id("run-1")) .await @@ -1085,16 +1082,12 @@ mod tests { } #[tokio::test] - async fn slate_run_store_lists_artifact_values_and_assets() { + async fn slate_run_store_lists_blobs_and_assets() { let (_object_store, store) = make_store(); let _created_at = dt("2026-03-27T12:00:00Z"); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); - run.put_artifact_value("summary", &serde_json::json!({"done": true})) - .await - .unwrap(); - run.put_artifact_value("plan", &serde_json::json!({"steps": 3})) - .await - .unwrap(); + let summary_blob = run.write_blob(br#"{"done":true}"#).await.unwrap(); + let plan_blob = run.write_blob(br#"{"steps":3}"#).await.unwrap(); let snapshot_node = StageId::new("code", 2); run.put_asset(&snapshot_node, "src/lib.rs", b"fn main() {}") @@ -1107,8 +1100,8 @@ mod tests { .unwrap(); assert_eq!( - run.list_artifact_values().await.unwrap(), - vec!["plan".to_string(), "summary".to_string()] + run.list_blobs().await.unwrap(), + vec![plan_blob, summary_blob] ); assert_eq!( run.list_all_assets().await.unwrap(), @@ -1363,9 +1356,7 @@ mod tests { )) .await .unwrap(); - run.put_artifact_value("summary", &serde_json::json!({"done": true})) - .await - .unwrap(); + let summary_blob = run.write_blob(br#"{"done":true}"#).await.unwrap(); run.put_asset(&node, "src/lib.rs", b"fn main() {}") .await .unwrap(); @@ -1407,8 +1398,8 @@ mod tests { assert_eq!(state.retro_prompt.as_deref(), Some("How did it go?")); assert_eq!(state.retro_response.as_deref(), Some("Smooth enough")); assert_eq!( - run.get_artifact_value("summary").await.unwrap(), - Some(serde_json::json!({"done": true})) + run.read_blob(&summary_blob).await.unwrap(), + Some(Bytes::from_static(br#"{"done":true}"#)) ); assert_eq!( run.get_asset(&node, "src/lib.rs").await.unwrap(), diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 74df34b16..4474fda2f 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -15,7 +15,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use crate::keys; use crate::run_state::EventProjectionCache; use crate::{EventEnvelope, EventPayload, Result, RunProjection, RunSummary, StageId, StoreError}; -use fabro_types::RunId; +use fabro_types::{RunBlobId, RunId}; #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] pub struct NodeAsset { @@ -213,26 +213,18 @@ impl SlateRunStore { Ok(Box::pin(UnboundedReceiverStream::new(receiver))) } - pub async fn put_artifact_value( - &self, - artifact_id: &str, - value: &serde_json::Value, - ) -> Result<()> { - self.inner - .db - .put_json(&keys::artifact_value(artifact_id), value) - .await + pub async fn write_blob(&self, data: &[u8]) -> Result { + let id = RunBlobId::new(&self.inner.run_id, data); + self.inner.db.put_bytes(&keys::blob_key(&id), data).await?; + Ok(id) } - pub async fn get_artifact_value(&self, artifact_id: &str) -> Result> { - self.inner - .db - .get_json(&keys::artifact_value(artifact_id)) - .await + pub async fn read_blob(&self, id: &RunBlobId) -> Result> { + self.inner.db.get_bytes(&keys::blob_key(id)).await } - pub async fn list_artifact_values(&self) -> Result> { - self.inner.db.list_artifact_values().await + pub async fn list_blobs(&self) -> Result> { + self.inner.db.list_blobs().await } pub async fn put_asset(&self, node: &StageId, filename: &str, data: &[u8]) -> Result<()> { @@ -273,13 +265,6 @@ impl SlateRunDb { } } - async fn get_json(&self, key: &str) -> Result> { - match self { - Self::Writer(db) => get_json(db, key).await, - Self::Reader(db) => get_json(db.as_ref(), key).await, - } - } - async fn put_json(&self, key: &str, value: &T) -> Result<()> { put_json(self.writer()?, key, value).await } @@ -302,10 +287,10 @@ impl SlateRunDb { } } - async fn list_artifact_values(&self) -> Result> { + async fn list_blobs(&self) -> Result> { match self { - Self::Writer(db) => list_artifact_values(db).await, - Self::Reader(db) => list_artifact_values(db.as_ref()).await, + Self::Writer(db) => list_blobs(db).await, + Self::Reader(db) => list_blobs(db.as_ref()).await, } } @@ -381,23 +366,21 @@ where Ok(events) } -async fn list_artifact_values(db: &R) -> Result> +async fn list_blobs(db: &R) -> Result> where R: DbRead + Sync, { - let mut iter = db - .scan_prefix(keys::ARTIFACT_VALUES_PREFIX.as_bytes()) - .await?; - let mut artifact_ids = Vec::new(); + let mut iter = db.scan_prefix(keys::BLOBS_PREFIX.as_bytes()).await?; + let mut blob_ids = Vec::new(); while let Some(entry) = iter.next().await? { let key = key_to_string(&entry.key)?; - let Some(artifact_id) = keys::parse_artifact_value_id(&key) else { + let Some(blob_id) = keys::parse_blob_id(&key) else { continue; }; - artifact_ids.push(artifact_id); + blob_ids.push(blob_id); } - artifact_ids.sort(); - Ok(artifact_ids) + blob_ids.sort(); + Ok(blob_ids) } async fn list_all_assets(db: &R) -> Result> diff --git a/lib/crates/fabro-types/Cargo.toml b/lib/crates/fabro-types/Cargo.toml index c66ac28c0..af2baa954 100644 --- a/lib/crates/fabro-types/Cargo.toml +++ b/lib/crates/fabro-types/Cargo.toml @@ -23,4 +23,6 @@ dirs.workspace = true fabro-macros = { path = "../fabro-macros" } serde.workspace = true serde_json.workspace = true +sha2.workspace = true ulid.workspace = true +uuid.workspace = true diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 7741df0cd..6170db70e 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -10,6 +10,7 @@ pub mod outcome; pub mod pull_request; pub mod retro; pub mod run; +pub mod run_blob_id; pub mod run_id; pub mod sandbox_record; pub mod settings; @@ -30,6 +31,7 @@ pub use retro::{ OpenItemKind, Retro, RetroNarrative, SmoothnessRating, StageRetro, }; pub use run::RunRecord; +pub use run_blob_id::RunBlobId; pub use run_id::RunId; pub use run_id::fixtures; pub use sandbox_record::SandboxRecord; diff --git a/lib/crates/fabro-types/src/run_blob_id.rs b/lib/crates/fabro-types/src/run_blob_id.rs new file mode 100644 index 000000000..732110b99 --- /dev/null +++ b/lib/crates/fabro-types/src/run_blob_id.rs @@ -0,0 +1,108 @@ +use std::fmt; +use std::str::FromStr; + +use serde::de::Error as _; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use sha2::{Digest, Sha256}; +use ulid::Ulid; +use uuid::Uuid; + +use crate::RunId; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct RunBlobId(Uuid); + +impl RunBlobId { + pub fn new(run_id: &RunId, content: &[u8]) -> Self { + let ulid: Ulid = (*run_id).into(); + let ulid_bytes = ulid.to_bytes(); + let hash = Sha256::digest(content); + let mut buf = [0_u8; 16]; + buf[..8].copy_from_slice(&ulid_bytes[..8]); + buf[8..].copy_from_slice(&hash[..8]); + Self(Uuid::new_v8(buf)) + } + + pub fn uuid(&self) -> &Uuid { + &self.0 + } +} + +impl fmt::Display for RunBlobId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.hyphenated().fmt(f) + } +} + +impl FromStr for RunBlobId { + type Err = uuid::Error; + + fn from_str(s: &str) -> Result { + Ok(Self(Uuid::parse_str(s)?)) + } +} + +impl Serialize for RunBlobId { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(&self.to_string()) + } +} + +impl<'de> Deserialize<'de> for RunBlobId { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = String::deserialize(deserializer)?; + value.parse().map_err(D::Error::custom) + } +} + +#[cfg(test)] +mod tests { + use crate::{RunBlobId, RunId}; + + #[test] + fn same_content_and_run_id_produce_same_blob_id() { + let run_id = RunId::new(); + assert_eq!( + RunBlobId::new(&run_id, b"hello"), + RunBlobId::new(&run_id, b"hello") + ); + } + + #[test] + fn different_run_ids_produce_different_blob_ids() { + assert_ne!( + RunBlobId::new(&RunId::new(), b"hello"), + RunBlobId::new(&RunId::new(), b"hello") + ); + } + + #[test] + fn different_content_produces_different_blob_ids() { + let run_id = RunId::new(); + assert_ne!( + RunBlobId::new(&run_id, b"hello"), + RunBlobId::new(&run_id, b"world") + ); + } + + #[test] + fn display_and_parse_round_trip() { + let blob_id = RunBlobId::new(&RunId::new(), b"hello"); + let parsed: RunBlobId = blob_id.to_string().parse().unwrap(); + assert_eq!(parsed, blob_id); + } + + #[test] + fn serde_round_trip() { + let blob_id = RunBlobId::new(&RunId::new(), b"hello"); + let value = serde_json::to_value(blob_id).unwrap(); + let parsed: RunBlobId = serde_json::from_value(value).unwrap(); + assert_eq!(parsed, blob_id); + } +} diff --git a/lib/crates/fabro-workflow/src/run_dump.rs b/lib/crates/fabro-workflow/src/run_dump.rs index 115623795..c3dee22d7 100644 --- a/lib/crates/fabro-workflow/src/run_dump.rs +++ b/lib/crates/fabro-workflow/src/run_dump.rs @@ -206,19 +206,15 @@ impl RunDump { ); } - for artifact_id in run_store.list_artifact_values().await? { - let artifact_id_segment = validate_single_path_segment("artifact id", &artifact_id)?; + for blob_id in run_store.list_blobs().await? { + let blob_name = validate_single_path_segment("blob id", &blob_id.to_string())?; let value = run_store - .get_artifact_value(&artifact_id) + .read_blob(&blob_id) .await? - .with_context(|| { - format!("artifact value {artifact_id:?} is missing from the store") - })?; - entries.push(RunDumpEntry::json_path( - &PathBuf::from("artifacts") - .join("values") - .join(format!("{}.json", artifact_id_segment.display())), - value, + .with_context(|| format!("blob {blob_id:?} is missing from the store"))?; + entries.push(RunDumpEntry::bytes_path( + &PathBuf::from("blobs").join(blob_name), + value.to_vec(), )); }