scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor: replace artifact values with run blobs

Browse source 
Rename durable artifact values to raw byte blobs keyed by RunBlobId,
add the blob type in fabro-types, switch SlateRunStore to write/read/list
blob APIs, and export blobs from store dumps by UUID.
This commit is contained in:
Bryan Helmkamp 2026-04-03 23:59:27 -07:00
parent 98e93d22fc
commit 988f267268
11 changed files with 178 additions and 101 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -2010,7 +2010,9 @@ dependencies = [
"fabro-macros",
"serde",
"serde_json",
"sha2",
"ulid",
"uuid",
]
[[package]]

2
Cargo.toml
View file

@ -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"

22
lib/crates/fabro-cli/src/commands/store/dump.rs
View file

@ -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(

27
lib/crates/fabro-store/src/keys.rs
View file

@ -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<u32> {
parse_seq(key, EVENTS_PREFIX)
}
pub(crate) fn parse_artifact_value_id(key: &str) -> Option<String> {
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<RunBlobId> {
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

2
lib/crates/fabro-store/src/lib.rs
View file

@ -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};

39
lib/crates/fabro-store/src/slate/mod.rs
View file

@ -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(),

55
lib/crates/fabro-store/src/slate/run_store.rs
View file

@ -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<RunBlobId> {
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<Option<serde_json::Value>> {
self.inner
.db
.get_json(&keys::artifact_value(artifact_id))
.await
pub async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
self.inner.db.get_bytes(&keys::blob_key(id)).await
}
pub async fn list_artifact_values(&self) -> Result<Vec<String>> {
self.inner.db.list_artifact_values().await
pub async fn list_blobs(&self) -> Result<Vec<RunBlobId>> {
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<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>> {
match self {
Self::Writer(db) => get_json(db, key).await,
Self::Reader(db) => get_json(db.as_ref(), key).await,
}
}
async fn put_json<T: Serialize>(&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<Vec<String>> {
async fn list_blobs(&self) -> Result<Vec<RunBlobId>> {
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<R>(db: &R) -> Result<Vec<String>>
async fn list_blobs<R>(db: &R) -> Result<Vec<RunBlobId>>
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<R>(db: &R) -> Result<Vec<NodeAsset>>

2
lib/crates/fabro-types/Cargo.toml
View file

@ -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

2
lib/crates/fabro-types/src/lib.rs
View file

@ -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;

108
lib/crates/fabro-types/src/run_blob_id.rs Normal file
View file

@ -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<Self, Self::Err> {
Ok(Self(Uuid::parse_str(s)?))
}
}
impl Serialize for RunBlobId {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(&self.to_string())
}
}
impl<'de> Deserialize<'de> for RunBlobId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
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);
}
}

18
lib/crates/fabro-workflow/src/run_dump.rs
View file

@ -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(),
));
}