diff --git a/Cargo.lock b/Cargo.lock index 1d27c383e..397988c6d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2041,6 +2041,7 @@ dependencies = [ "hex", "md5", "mime_guess", + "object_store", "predicates", "rand 0.8.5", "regex", diff --git a/lib/crates/fabro-workflow/Cargo.toml b/lib/crates/fabro-workflow/Cargo.toml index 25ea58832..3bef794c8 100644 --- a/lib/crates/fabro-workflow/Cargo.toml +++ b/lib/crates/fabro-workflow/Cargo.toml @@ -40,6 +40,7 @@ thiserror.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true +object_store.workspace = true ulid.workspace = true uuid.workspace = true rand.workspace = true @@ -64,6 +65,7 @@ base64.workspace = true toml.workspace = true fabro-mcp = { path = "../fabro-mcp" } tokio = { workspace = true, features = ["test-util", "macros"] } +object_store.workspace = true tempfile = "3" assert_cmd = "2" predicates = "3" diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 686050ab8..35677d59b 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -1,13 +1,17 @@ -use std::collections::HashMap; use std::collections::BTreeMap; +use std::collections::HashMap; use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; use chrono::{Local, Utc}; use fabro_config::{FabroSettings, FabroSettingsExt}; use fabro_graphviz::graph::{AttrValue, Graph}; use fabro_model::{Catalog, Provider}; use fabro_sandbox::SandboxProvider; +use fabro_store::SlateStore; use fabro_types::RunId; +use object_store::local::LocalFileSystem; use crate::error::FabroError; use crate::pipeline::types::PersistOptions; @@ -17,9 +21,13 @@ use crate::run_lookup::default_runs_base; use crate::run_status::{RunStatus, write_run_status}; use crate::transforms::{Transform, expand_vars}; use fabro_sandbox::daytona::detect_repo_info; +use tokio::runtime::Builder; -use crate::event::{WorkflowRunEvent, append_progress_event, canonicalize_event_at}; +use super::hydrate::open_or_hydrate_run; use super::source::{ResolveWorkflowInput, WorkflowInput, resolve_workflow}; +use crate::event::{ + WorkflowRunEvent, append_progress_event, canonicalize_event_at, normalize_json_value, +}; const RUN_CONFIG_FILE: &str = "workflow.toml"; @@ -123,6 +131,7 @@ pub fn create(request: CreateRunInput) -> Result { &resolved.raw_source, resolved.workflow_toml_path.as_deref(), )?; + hydrate_created_run(&persisted)?; Ok(CreatedRun { persisted, @@ -139,10 +148,11 @@ fn emit_run_created_event( ) -> Result<(), FabroError> { let record = persisted.run_record(); let workflow_config = workflow_toml_path.and_then(|path| std::fs::read_to_string(path).ok()); - let settings = sort_json_value( - serde_json::to_value(&record.settings).map_err(|err| FabroError::engine(err.to_string()))?, + let settings = normalize_json_value( + serde_json::to_value(&record.settings) + .map_err(|err| FabroError::engine(err.to_string()))?, ); - let graph = sort_json_value( + let graph = normalize_json_value( serde_json::to_value(&record.graph).map_err(|err| FabroError::engine(err.to_string()))?, ); let event = WorkflowRunEvent::RunCreated { @@ -151,7 +161,11 @@ fn emit_run_created_event( graph, workflow_source: (!workflow_source.is_empty()).then(|| workflow_source.to_string()), workflow_config, - labels: record.labels.clone().into_iter().collect::>(), + labels: record + .labels + .clone() + .into_iter() + .collect::>(), run_dir: persisted.run_dir().display().to_string(), working_directory: record.working_directory.display().to_string(), host_repo_path: record.host_repo_path.clone(), @@ -164,20 +178,44 @@ fn emit_run_created_event( .map_err(|err| FabroError::engine(err.to_string())) } -fn sort_json_value(value: serde_json::Value) -> serde_json::Value { - match value { - serde_json::Value::Object(map) => serde_json::Value::Object( - map.into_iter() - .map(|(key, value)| (key, sort_json_value(value))) - .collect(), - ), - serde_json::Value::Array(values) => { - serde_json::Value::Array(values.into_iter().map(sort_json_value).collect()) - } - other => other, +fn hydrate_created_run(persisted: &Persisted) -> Result<(), FabroError> { + let storage_dir = persisted.run_record().settings.storage_dir(); + let run_dir = persisted.run_dir().to_path_buf(); + let join = std::thread::spawn(move || -> Result<(), FabroError> { + let runtime = Builder::new_current_thread() + .enable_all() + .build() + .map_err(|err| FabroError::engine(err.to_string()))?; + runtime.block_on(async move { + let store = build_durable_store(&storage_dir)?; + open_or_hydrate_run(store.as_ref(), &run_dir) + .await + .map(|_| ()) + }) + }); + + match join.join() { + Ok(result) => result, + Err(_) => Err(FabroError::engine( + "store hydration thread panicked".to_string(), + )), } } +fn build_durable_store(storage_dir: &Path) -> Result, FabroError> { + let store_path = storage_dir.join("store"); + std::fs::create_dir_all(&store_path)?; + let object_store = Arc::new( + LocalFileSystem::new_with_prefix(&store_path) + .map_err(|err| FabroError::engine(err.to_string()))?, + ); + Ok(Arc::new(SlateStore::new( + object_store, + "", + Duration::from_millis(5), + ))) +} + fn validate_sandbox_provider(settings: &FabroSettings) -> Result<(), FabroError> { if let Some(provider) = settings .sandbox_settings() @@ -374,7 +412,11 @@ pub(crate) fn make_run_dir(runs_base: &Path, run_id: &str, dry_run: bool) -> Pat mod tests { use super::*; use fabro_graphviz::graph::AttrValue; + use fabro_store::{SlateStore, Store}; use fabro_types::fixtures; + use object_store::local::LocalFileSystem; + use std::sync::Arc; + use std::time::Duration; use crate::operations::{ValidateInput, validate}; use crate::run_status::RunStatusRecordExt; @@ -745,4 +787,45 @@ mod tests { ) ); } + + #[tokio::test] + async fn create_hydrates_run_created_event_into_store() { + let dir = tempfile::tempdir().unwrap(); + let storage_dir = dir.path().join("storage"); + let run_dir = dir.path().join("run"); + let created = create(CreateRunInput { + workflow: WorkflowInput::DotSource { + source: MINIMAL_DOT.to_string(), + base_dir: None, + }, + settings: FabroSettings { + storage_dir: Some(storage_dir.clone()), + dry_run: Some(true), + ..Default::default() + }, + cwd: dir.path().to_path_buf(), + workflow_slug: Some("slug".to_string()), + run_dir: Some(run_dir.clone()), + run_id: Some(fixtures::RUN_3), + host_repo_path: None, + base_branch: None, + }) + .unwrap(); + + std::fs::create_dir_all(storage_dir.join("store")).unwrap(); + let object_store = + Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).unwrap()); + let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(5))); + let run_store = store + .open_run_reader(&created.run_id) + .await + .unwrap() + .unwrap(); + let events = run_store.list_events().await.unwrap(); + + assert_eq!( + events.first().unwrap().payload.as_value()["event"], + "run.created" + ); + } }