mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-10-06 02:48:25 +00:00
Close workflow goals over version dependencies
This commit is contained in:
Scott Werner
parent
a0845d8346
commit
63025bb748
4 changed files with 539 additions and 38 deletions
|
|
@ -116,7 +116,7 @@ mod tests {
|
|||
use axum::body::{Body, to_bytes};
|
||||
use axum::http::{Method, Request, StatusCode, header};
|
||||
use axum::response::IntoResponse;
|
||||
use fabro_types::WorkflowVersionId;
|
||||
use fabro_types::{BlobHash, WorkflowVersion, WorkflowVersionId};
|
||||
use serde_json::{Value, json};
|
||||
use tower::ServiceExt;
|
||||
|
||||
|
|
@ -233,6 +233,45 @@ mod tests {
|
|||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_rejects_workflow_config_with_missing_goal_file_before_storage() {
|
||||
let state = TestAppStateBuilder::new().build();
|
||||
let app = test_support::build_test_router(Arc::clone(&state));
|
||||
let payload = json!({
|
||||
"entrypoint": "workflow.fabro",
|
||||
"files": {
|
||||
"workflow.fabro": GRAPH,
|
||||
"workflow.toml": "_version = 1\n[run.goal]\nfile = \"prompts/goal.md\"\n"
|
||||
},
|
||||
"workflow_dependencies": {}
|
||||
});
|
||||
let version = serde_json::from_value::<WorkflowVersion>(payload.clone()).unwrap();
|
||||
let id = WorkflowVersionId::from(BlobHash::new(&version.canonical_bytes().unwrap()));
|
||||
|
||||
let response = app
|
||||
.oneshot(request(serde_json::to_vec(&payload).unwrap()))
|
||||
.await
|
||||
.unwrap();
|
||||
let body = fabro_test::expect_axum_json(
|
||||
response,
|
||||
StatusCode::UNPROCESSABLE_ENTITY,
|
||||
"POST /api/v1/workflow-versions with missing run goal file",
|
||||
)
|
||||
.await;
|
||||
|
||||
assert_eq!(error_code(&body), INVALID_VERSION_CODE);
|
||||
assert!(
|
||||
!state
|
||||
.store_ref()
|
||||
.blobs()
|
||||
.await
|
||||
.unwrap()
|
||||
.exists(&id.into())
|
||||
.await
|
||||
.unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unavailable_dependency_has_specific_code() {
|
||||
let state = TestAppStateBuilder::new().build();
|
||||
|
|
|
|||
|
|
@ -9,20 +9,23 @@
|
|||
use std::collections::{BTreeSet, HashMap, VecDeque};
|
||||
|
||||
use fabro_config::parse::{SettingsSource, validate_settings_source};
|
||||
use fabro_config::{EnvironmentDockerfileLayer, EnvironmentImageLayer, SettingsLayer};
|
||||
use fabro_config::{
|
||||
EnvironmentDockerfileLayer, EnvironmentImageLayer, RunGoalLayer, SettingsLayer,
|
||||
};
|
||||
use fabro_graphviz::parser;
|
||||
use fabro_template::{
|
||||
BundleTemplateStore, GraphReference, GraphReferenceError, StaticReferenceError,
|
||||
TemplateDiscoveryError, TemplateSource, discover_static_dependency_closure,
|
||||
TemplateDiscoveryError, TemplateLoadError, TemplateSource, discover_static_dependency_closure,
|
||||
validate_static_reference, visit_graph_references,
|
||||
};
|
||||
use fabro_types::graph::ReferenceKind;
|
||||
use fabro_types::settings::InterpString;
|
||||
use fabro_types::{ManifestPath, WorkflowPath, WorkflowPathParseError, WorkflowVersion};
|
||||
use thiserror::Error;
|
||||
|
||||
mod store;
|
||||
|
||||
pub use store::{WorkflowVersionStore, WorkflowVersionStoreError};
|
||||
pub use store::{LoadedWorkflowVersionClosure, WorkflowVersionStore, WorkflowVersionStoreError};
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum WorkflowVersionError {
|
||||
|
|
@ -88,8 +91,12 @@ pub struct ValidatedWorkflowVersion(WorkflowVersion);
|
|||
|
||||
impl ValidatedWorkflowVersion {
|
||||
pub fn new(version: WorkflowVersion) -> Result<Self, WorkflowVersionError> {
|
||||
validate_config(&version)?;
|
||||
validate_graph_closure(&version)?;
|
||||
let template_root = ManifestPath::from_wire(".")
|
||||
.expect("the template package root must be a valid manifest path");
|
||||
let mut template_roots = Vec::new();
|
||||
validate_config(&version, &template_root, &mut template_roots)?;
|
||||
validate_graph_closure(&version, &template_root, &mut template_roots)?;
|
||||
validate_template_closure(&version, template_roots)?;
|
||||
Ok(Self(version))
|
||||
}
|
||||
|
||||
|
|
@ -104,7 +111,11 @@ impl ValidatedWorkflowVersion {
|
|||
}
|
||||
}
|
||||
|
||||
fn validate_config(version: &WorkflowVersion) -> Result<(), WorkflowVersionError> {
|
||||
fn validate_config(
|
||||
version: &WorkflowVersion,
|
||||
template_root: &ManifestPath,
|
||||
template_roots: &mut Vec<TemplateSource>,
|
||||
) -> Result<(), WorkflowVersionError> {
|
||||
let config_path =
|
||||
WorkflowPath::new("workflow.toml").expect("the static workflow config path must be valid");
|
||||
let Some(source) = version.files().get(&config_path) else {
|
||||
|
|
@ -133,9 +144,47 @@ fn validate_config(version: &WorkflowVersion) -> Result<(), WorkflowVersionError
|
|||
for image in layer.environment_images() {
|
||||
validate_dockerfile(version, &config_path, image)?;
|
||||
}
|
||||
|
||||
match layer.run.as_ref().and_then(|run| run.goal.as_ref()) {
|
||||
Some(RunGoalLayer::Inline(goal)) => template_roots.push(TemplateSource::new(
|
||||
manifest_path(&config_path),
|
||||
template_root.clone(),
|
||||
unresolved_source(goal),
|
||||
)),
|
||||
Some(RunGoalLayer::File { file }) => {
|
||||
let reference = unresolved_source(file);
|
||||
validate_static_reference(&reference, ReferenceKind::RunGoalFile).map_err(
|
||||
|source| WorkflowVersionError::StaticReference {
|
||||
path: config_path.clone(),
|
||||
source,
|
||||
},
|
||||
)?;
|
||||
let target = resolve_reference(&config_path, ReferenceKind::RunGoalFile, &reference)?;
|
||||
let content = require_file(
|
||||
version,
|
||||
&config_path,
|
||||
ReferenceKind::RunGoalFile,
|
||||
target.clone(),
|
||||
)?;
|
||||
template_roots.push(TemplateSource::new(
|
||||
manifest_path(&target),
|
||||
template_root.clone(),
|
||||
content,
|
||||
));
|
||||
}
|
||||
None => {}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[expect(
|
||||
clippy::disallowed_methods,
|
||||
reason = "workflow-version validation preserves authored template source for dependency discovery"
|
||||
)]
|
||||
fn unresolved_source(value: &InterpString) -> String {
|
||||
value.as_source()
|
||||
}
|
||||
|
||||
fn validate_dockerfile(
|
||||
version: &WorkflowVersion,
|
||||
config_path: &WorkflowPath,
|
||||
|
|
@ -154,10 +203,11 @@ fn validate_dockerfile(
|
|||
require_file(version, config_path, ReferenceKind::Dockerfile, target).map(|_| ())
|
||||
}
|
||||
|
||||
fn validate_graph_closure(version: &WorkflowVersion) -> Result<(), WorkflowVersionError> {
|
||||
let template_store = template_store(version);
|
||||
let template_root = ManifestPath::from_wire(".")
|
||||
.expect("the template package root must be a valid manifest path");
|
||||
fn validate_graph_closure(
|
||||
version: &WorkflowVersion,
|
||||
template_root: &ManifestPath,
|
||||
template_roots: &mut Vec<TemplateSource>,
|
||||
) -> Result<(), WorkflowVersionError> {
|
||||
let mut queue = VecDeque::from([version.entrypoint().clone()]);
|
||||
let mut visited = BTreeSet::new();
|
||||
let mut child_workflows = BTreeSet::new();
|
||||
|
|
@ -185,10 +235,20 @@ fn validate_graph_closure(version: &WorkflowVersion) -> Result<(), WorkflowVersi
|
|||
let target = resolve_reference(&path, ReferenceKind::GraphGoalFile, reference)?;
|
||||
let content =
|
||||
require_file(version, &path, ReferenceKind::GraphGoalFile, target.clone())?;
|
||||
validate_template(&target, content, &template_store, &template_root)
|
||||
template_roots.push(TemplateSource::new(
|
||||
manifest_path(&target),
|
||||
template_root.clone(),
|
||||
content,
|
||||
));
|
||||
Ok(())
|
||||
}
|
||||
GraphReference::GoalInline { content } | GraphReference::InlinePrompt { content } => {
|
||||
validate_template(&path, content, &template_store, &template_root)
|
||||
template_roots.push(TemplateSource::new(
|
||||
manifest_path(&path),
|
||||
template_root.clone(),
|
||||
content,
|
||||
));
|
||||
Ok(())
|
||||
}
|
||||
GraphReference::Import { reference } => {
|
||||
let target = resolve_reference(&path, ReferenceKind::Import, reference)?;
|
||||
|
|
@ -206,7 +266,11 @@ fn validate_graph_closure(version: &WorkflowVersion) -> Result<(), WorkflowVersi
|
|||
let content =
|
||||
require_file(version, &path, ReferenceKind::FileInline, target.clone())?;
|
||||
if key == "prompt" {
|
||||
validate_template(&target, content, &template_store, &template_root)?;
|
||||
template_roots.push(TemplateSource::new(
|
||||
manifest_path(&target),
|
||||
template_root.clone(),
|
||||
content,
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -234,24 +298,39 @@ fn validate_graph_closure(version: &WorkflowVersion) -> Result<(), WorkflowVersi
|
|||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_template(
|
||||
path: &WorkflowPath,
|
||||
content: &str,
|
||||
store: &BundleTemplateStore,
|
||||
root: &ManifestPath,
|
||||
fn validate_template_closure(
|
||||
version: &WorkflowVersion,
|
||||
roots: Vec<TemplateSource>,
|
||||
) -> Result<(), WorkflowVersionError> {
|
||||
let manifest_path = manifest_path(path);
|
||||
discover_static_dependency_closure(
|
||||
[TemplateSource::new(manifest_path, root.clone(), content)],
|
||||
store,
|
||||
)
|
||||
.map_err(|source| WorkflowVersionError::Template {
|
||||
path: path.clone(),
|
||||
source: Box::new(source),
|
||||
discover_static_dependency_closure(roots, &template_store(version)).map_err(|source| {
|
||||
WorkflowVersionError::Template {
|
||||
path: template_discovery_path(&source),
|
||||
source: Box::new(source),
|
||||
}
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn template_discovery_path(error: &TemplateDiscoveryError) -> WorkflowPath {
|
||||
let path = match error {
|
||||
TemplateDiscoveryError::Parse(source) => source
|
||||
.source_name()
|
||||
.expect("dependency extraction must retain its source name")
|
||||
.to_owned(),
|
||||
TemplateDiscoveryError::Load(source) => match source {
|
||||
TemplateLoadError::UnsafeReference { parent, .. }
|
||||
| TemplateLoadError::EscapesRoot { parent, .. } => parent.to_string(),
|
||||
TemplateLoadError::DynamicDependency { path } => path.to_string(),
|
||||
TemplateLoadError::Io { .. } => {
|
||||
unreachable!("bundle template dependency discovery cannot perform filesystem I/O")
|
||||
}
|
||||
},
|
||||
TemplateDiscoveryError::Missing { parent, .. }
|
||||
| TemplateDiscoveryError::Dynamic { parent } => parent.to_string(),
|
||||
};
|
||||
WorkflowPath::new(path).expect("template paths sourced from a workflow version must be valid")
|
||||
}
|
||||
|
||||
fn template_store(version: &WorkflowVersion) -> BundleTemplateStore {
|
||||
BundleTemplateStore::new(
|
||||
version
|
||||
|
|
@ -300,6 +379,10 @@ fn manifest_path(path: &WorkflowPath) -> ManifestPath {
|
|||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use fabro_template::{TemplateDiscoveryError, TemplateLoadError};
|
||||
use fabro_types::graph::ReferenceKind;
|
||||
use fabro_types::{BlobHash, WorkflowPath, WorkflowVersion, WorkflowVersionId};
|
||||
|
||||
use super::{ValidatedWorkflowVersion, WorkflowVersionError};
|
||||
|
|
@ -332,6 +415,38 @@ mod tests {
|
|||
)
|
||||
}
|
||||
|
||||
fn version_with_config(
|
||||
config: String,
|
||||
extra_files: impl IntoIterator<Item = (&'static str, &'static str)>,
|
||||
) -> Result<ValidatedWorkflowVersion, WorkflowVersionError> {
|
||||
let mut files = extra_files
|
||||
.into_iter()
|
||||
.map(|(path_value, content)| (path(path_value), content.to_owned()))
|
||||
.collect::<BTreeMap<_, _>>();
|
||||
files.insert(path("workflow.fabro"), "digraph W {}".to_owned());
|
||||
files.insert(path("workflow.toml"), config);
|
||||
ValidatedWorkflowVersion::new(
|
||||
WorkflowVersion::new(path("workflow.fabro"), files, BTreeMap::default())
|
||||
.expect("test fixtures must be structurally valid"),
|
||||
)
|
||||
}
|
||||
|
||||
fn version_with_goal_file(
|
||||
reference: &str,
|
||||
) -> Result<ValidatedWorkflowVersion, WorkflowVersionError> {
|
||||
let reference = serde_json::to_string(reference).unwrap();
|
||||
version_with_config(format!("_version = 1\n[run.goal]\nfile = {reference}\n"), [
|
||||
])
|
||||
}
|
||||
|
||||
fn version_with_inline_goal(
|
||||
goal: &str,
|
||||
extra_files: impl IntoIterator<Item = (&'static str, &'static str)>,
|
||||
) -> Result<ValidatedWorkflowVersion, WorkflowVersionError> {
|
||||
let goal = serde_json::to_string(goal).unwrap();
|
||||
version_with_config(format!("_version = 1\n[run]\ngoal = {goal}\n"), extra_files)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validates_imports_templates_file_refs_and_dependencies() {
|
||||
let version = version_with(
|
||||
|
|
@ -430,6 +545,162 @@ mod tests {
|
|||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_missing_workflow_goal_file() {
|
||||
let error = version_with(
|
||||
[
|
||||
("workflow.fabro", "digraph W {}"),
|
||||
(
|
||||
"workflow.toml",
|
||||
"_version = 1\n[run.goal]\nfile = \"prompts/goal.md\"\n",
|
||||
),
|
||||
],
|
||||
[],
|
||||
)
|
||||
.unwrap_err();
|
||||
|
||||
assert!(matches!(
|
||||
error,
|
||||
WorkflowVersionError::MissingFile {
|
||||
path: source_path,
|
||||
kind,
|
||||
target,
|
||||
}
|
||||
if source_path == path("workflow.toml")
|
||||
&& kind == ReferenceKind::RunGoalFile
|
||||
&& target == path("prompts/goal.md")
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_inline_workflow_goal_with_static_template_closure() {
|
||||
let version = version_with_inline_goal(
|
||||
r#"Review {{ vars.target }} with {{ inputs.mode }} after {{ goal }}. {% include "prompts/shared.md" %}"#,
|
||||
[("prompts/shared.md", "Use {{ vars.detail }}")],
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(version.version().files().len(), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_file_workflow_goal_with_transitive_template_closure() {
|
||||
let version = version_with_config(
|
||||
"_version = 1\n[run.goal]\nfile = \"prompts/goal.md\"\n".to_owned(),
|
||||
[
|
||||
("prompts/goal.md", r#"{% include "partial.md" %}"#),
|
||||
("prompts/partial.md", r#"{% include "nested/detail.md" %}"#),
|
||||
("prompts/nested/detail.md", "Use {{ vars.detail }}"),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(version.version().files().len(), 5);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_non_static_or_nonportable_workflow_goal_file_references() {
|
||||
for reference in ["{{ vars.NAME }}", "{% include \"goal.md\" %}"] {
|
||||
let error = version_with_goal_file(reference).unwrap_err();
|
||||
let WorkflowVersionError::StaticReference {
|
||||
path: source_path,
|
||||
source,
|
||||
} = error
|
||||
else {
|
||||
panic!("expected static-reference error for {reference:?}");
|
||||
};
|
||||
assert_eq!(source_path, path("workflow.toml"));
|
||||
assert_eq!(source.kind(), ReferenceKind::RunGoalFile);
|
||||
}
|
||||
|
||||
for reference in [
|
||||
"",
|
||||
"/absolute.md",
|
||||
"../outside.md",
|
||||
"~/goal.md",
|
||||
"C:/goal.md",
|
||||
"prompts\\goal.md",
|
||||
"prompts//goal.md",
|
||||
"prompts/",
|
||||
"prompts/goal\n.md",
|
||||
] {
|
||||
let error = version_with_goal_file(reference).unwrap_err();
|
||||
assert!(
|
||||
matches!(
|
||||
&error,
|
||||
WorkflowVersionError::InvalidReference {
|
||||
path: source_path,
|
||||
kind: ReferenceKind::RunGoalFile,
|
||||
..
|
||||
} if *source_path == path("workflow.toml")
|
||||
),
|
||||
"expected invalid-reference error for {reference:?}, got {error:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_invalid_workflow_goal_template_closure() {
|
||||
let missing = version_with_inline_goal(r#"{% include "missing.md" %}"#, []).unwrap_err();
|
||||
let WorkflowVersionError::Template {
|
||||
path: source_path,
|
||||
source,
|
||||
} = missing
|
||||
else {
|
||||
panic!("expected missing template dependency");
|
||||
};
|
||||
assert_eq!(source_path, path("workflow.toml"));
|
||||
assert!(matches!(
|
||||
source.as_ref(),
|
||||
TemplateDiscoveryError::Missing { parent, reference }
|
||||
if parent.to_string() == "workflow.toml" && reference == "missing.md"
|
||||
));
|
||||
|
||||
let dynamic = version_with_inline_goal(r"{% include inputs.partial %}", []).unwrap_err();
|
||||
let WorkflowVersionError::Template { source, .. } = dynamic else {
|
||||
panic!("expected dynamic template dependency");
|
||||
};
|
||||
assert!(matches!(
|
||||
source.as_ref(),
|
||||
TemplateDiscoveryError::Dynamic { parent }
|
||||
if parent.to_string() == "workflow.toml"
|
||||
));
|
||||
|
||||
let escaping =
|
||||
version_with_inline_goal(r#"{% include "../outside.md" %}"#, []).unwrap_err();
|
||||
let WorkflowVersionError::Template { source, .. } = escaping else {
|
||||
panic!("expected escaping template dependency");
|
||||
};
|
||||
assert!(matches!(
|
||||
source.as_ref(),
|
||||
TemplateDiscoveryError::Load(TemplateLoadError::EscapesRoot { parent, .. })
|
||||
if parent.to_string() == "workflow.toml"
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validates_all_inline_graph_roots_that_share_the_graph_path() {
|
||||
let error = version_with(
|
||||
[(
|
||||
"workflow.fabro",
|
||||
r#"digraph W {
|
||||
graph [goal="valid"]
|
||||
step [prompt="{% include inputs.partial %}"]
|
||||
}"#,
|
||||
)],
|
||||
[],
|
||||
)
|
||||
.unwrap_err();
|
||||
|
||||
assert!(matches!(
|
||||
error,
|
||||
WorkflowVersionError::Template {
|
||||
source,
|
||||
..
|
||||
} if matches!(source.as_ref(), TemplateDiscoveryError::Dynamic { .. })
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_root_config_and_all_dockerfile_path_sources() {
|
||||
let version = version_with(
|
||||
|
|
|
|||
|
|
@ -40,6 +40,48 @@ pub enum WorkflowVersionStoreError {
|
|||
},
|
||||
}
|
||||
|
||||
/// A fully loaded and validated workflow-version dependency graph.
|
||||
///
|
||||
/// The requested root is always present exactly once alongside every unique
|
||||
/// transitive dependency, keyed by canonical content ID.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct LoadedWorkflowVersionClosure {
|
||||
root_id: WorkflowVersionId,
|
||||
versions: BTreeMap<WorkflowVersionId, ValidatedWorkflowVersion>,
|
||||
}
|
||||
|
||||
impl LoadedWorkflowVersionClosure {
|
||||
#[must_use]
|
||||
pub fn root_id(&self) -> WorkflowVersionId {
|
||||
self.root_id
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn root(&self) -> &WorkflowVersion {
|
||||
self.versions
|
||||
.get(&self.root_id)
|
||||
.expect("a loaded workflow-version closure must contain its root")
|
||||
.version()
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn get(&self, id: &WorkflowVersionId) -> Option<&WorkflowVersion> {
|
||||
self.versions.get(id).map(ValidatedWorkflowVersion::version)
|
||||
}
|
||||
|
||||
pub fn versions(&self) -> impl Iterator<Item = (WorkflowVersionId, &WorkflowVersion)> + '_ {
|
||||
self.versions
|
||||
.iter()
|
||||
.map(|(id, version)| (*id, version.version()))
|
||||
}
|
||||
|
||||
fn into_root(mut self) -> ValidatedWorkflowVersion {
|
||||
self.versions
|
||||
.remove(&self.root_id)
|
||||
.expect("a loaded workflow-version closure must contain its root")
|
||||
}
|
||||
}
|
||||
|
||||
/// Content-addressed storage for validated workflow versions.
|
||||
///
|
||||
/// `put` only accepts semantically validated versions; `get` re-validates
|
||||
|
|
@ -61,7 +103,7 @@ impl WorkflowVersionStore {
|
|||
version: &ValidatedWorkflowVersion,
|
||||
) -> Result<WorkflowVersionId, WorkflowVersionStoreError> {
|
||||
let canonical = version.version().canonical_bytes()?;
|
||||
self.validate_dependency_closure(version.version().workflow_dependencies())
|
||||
self.load_dependency_closure(version.version().workflow_dependencies(), HashSet::new())
|
||||
.await?;
|
||||
self.blobs
|
||||
.write(&canonical)
|
||||
|
|
@ -74,12 +116,30 @@ impl WorkflowVersionStore {
|
|||
&self,
|
||||
id: &WorkflowVersionId,
|
||||
) -> Result<Option<ValidatedWorkflowVersion>, WorkflowVersionStoreError> {
|
||||
let Some(version) = self.load_one(id).await? else {
|
||||
let Some(closure) = self.get_closure(id).await? else {
|
||||
return Ok(None);
|
||||
};
|
||||
self.validate_dependency_closure(version.version().workflow_dependencies())
|
||||
Ok(Some(closure.into_root()))
|
||||
}
|
||||
|
||||
pub async fn get_closure(
|
||||
&self,
|
||||
root_id: &WorkflowVersionId,
|
||||
) -> Result<Option<LoadedWorkflowVersionClosure>, WorkflowVersionStoreError> {
|
||||
let Some(root) = self.load_one(root_id).await? else {
|
||||
return Ok(None);
|
||||
};
|
||||
let mut versions = self
|
||||
.load_dependency_closure(
|
||||
root.version().workflow_dependencies(),
|
||||
HashSet::from([*root_id]),
|
||||
)
|
||||
.await?;
|
||||
Ok(Some(version))
|
||||
versions.insert(*root_id, root);
|
||||
Ok(Some(LoadedWorkflowVersionClosure {
|
||||
root_id: *root_id,
|
||||
versions,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn load_one(
|
||||
|
|
@ -105,15 +165,17 @@ impl WorkflowVersionStore {
|
|||
Ok(Some(validated))
|
||||
}
|
||||
|
||||
async fn validate_dependency_closure(
|
||||
async fn load_dependency_closure(
|
||||
&self,
|
||||
dependencies: &BTreeMap<WorkflowPath, WorkflowVersionId>,
|
||||
) -> Result<(), WorkflowVersionStoreError> {
|
||||
mut visited: HashSet<WorkflowVersionId>,
|
||||
) -> Result<BTreeMap<WorkflowVersionId, ValidatedWorkflowVersion>, WorkflowVersionStoreError>
|
||||
{
|
||||
let mut pending = dependencies
|
||||
.iter()
|
||||
.map(|(path, id)| (path.clone(), *id))
|
||||
.collect::<VecDeque<_>>();
|
||||
let mut visited = HashSet::new();
|
||||
let mut versions = BTreeMap::new();
|
||||
|
||||
while let Some((path, id)) = pending.pop_front() {
|
||||
if !visited.insert(id) {
|
||||
|
|
@ -128,6 +190,7 @@ impl WorkflowVersionStore {
|
|||
.iter()
|
||||
.map(|(path, id)| (path.clone(), *id)),
|
||||
);
|
||||
versions.insert(id, dependency);
|
||||
}
|
||||
Ok(None) => {
|
||||
return Err(WorkflowVersionStoreError::DependencyNotFound { path, id });
|
||||
|
|
@ -144,7 +207,7 @@ impl WorkflowVersionStore {
|
|||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
Ok(versions)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -180,6 +243,12 @@ mod tests {
|
|||
.unwrap()
|
||||
}
|
||||
|
||||
fn version_id(version: &ValidatedWorkflowVersion) -> WorkflowVersionId {
|
||||
WorkflowVersionId::from(fabro_types::BlobHash::new(
|
||||
&version.version().canonical_bytes().unwrap(),
|
||||
))
|
||||
}
|
||||
|
||||
async fn stores() -> (Arc<BlobStore>, WorkflowVersionStore) {
|
||||
let database = Database::new(
|
||||
Arc::new(InMemory::new()),
|
||||
|
|
@ -273,12 +342,132 @@ mod tests {
|
|||
));
|
||||
assert!(!blobs.exists(&root_id.into()).await.unwrap());
|
||||
assert!(matches!(
|
||||
store.get(&child_id).await.unwrap_err(),
|
||||
store.get_closure(&child_id).await.unwrap_err(),
|
||||
WorkflowVersionStoreError::DependencyNotFound { id, .. }
|
||||
if id == missing_grandchild_id
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_closure_returns_root_and_transitive_dependencies() {
|
||||
let (_, store) = stores().await;
|
||||
let grandchild = version("digraph Grandchild {}", BTreeMap::new());
|
||||
let grandchild_id = store.put(&grandchild).await.unwrap();
|
||||
let child = version(
|
||||
r#"digraph Child { grandchild [stack.child_workflow="grandchild.fabro"] }"#,
|
||||
BTreeMap::from([(path("grandchild.fabro"), grandchild_id)]),
|
||||
);
|
||||
let child_id = store.put(&child).await.unwrap();
|
||||
let root = version(
|
||||
r#"digraph Root { child [stack.child_workflow="child.fabro"] }"#,
|
||||
BTreeMap::from([(path("child.fabro"), child_id)]),
|
||||
);
|
||||
let root_id = store.put(&root).await.unwrap();
|
||||
|
||||
let closure = store.get_closure(&root_id).await.unwrap().unwrap();
|
||||
|
||||
assert_eq!(closure.root_id(), root_id);
|
||||
assert_eq!(closure.root(), root.version());
|
||||
assert_eq!(closure.get(&child_id), Some(child.version()));
|
||||
assert_eq!(closure.get(&grandchild_id), Some(grandchild.version()));
|
||||
assert_eq!(
|
||||
closure
|
||||
.versions()
|
||||
.map(|(id, version)| (id, version.clone()))
|
||||
.collect::<BTreeMap<_, _>>(),
|
||||
BTreeMap::from([
|
||||
(root_id, root.version().clone()),
|
||||
(child_id, child.version().clone()),
|
||||
(grandchild_id, grandchild.version().clone()),
|
||||
])
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_closure_deduplicates_a_diamond() {
|
||||
let (_, store) = stores().await;
|
||||
let leaf = version("digraph Leaf {}", BTreeMap::new());
|
||||
let leaf_id = store.put(&leaf).await.unwrap();
|
||||
let left = version(
|
||||
r#"digraph Left { leaf [stack.child_workflow="leaf.fabro"] }"#,
|
||||
BTreeMap::from([(path("leaf.fabro"), leaf_id)]),
|
||||
);
|
||||
let left_id = store.put(&left).await.unwrap();
|
||||
let right = version(
|
||||
r#"digraph Right { leaf [stack.child_workflow="leaf.fabro"] }"#,
|
||||
BTreeMap::from([(path("leaf.fabro"), leaf_id)]),
|
||||
);
|
||||
let right_id = store.put(&right).await.unwrap();
|
||||
let root = version(
|
||||
r#"digraph Root {
|
||||
left [stack.child_workflow="left.fabro"]
|
||||
right [stack.child_workflow="right.fabro"]
|
||||
}"#,
|
||||
BTreeMap::from([
|
||||
(path("left.fabro"), left_id),
|
||||
(path("right.fabro"), right_id),
|
||||
]),
|
||||
);
|
||||
let root_id = store.put(&root).await.unwrap();
|
||||
|
||||
let closure = store.get_closure(&root_id).await.unwrap().unwrap();
|
||||
let ids = closure.versions().map(|(id, _)| id).collect::<Vec<_>>();
|
||||
|
||||
assert_eq!(ids.len(), 4);
|
||||
assert_eq!(ids.iter().filter(|&&id| id == leaf_id).count(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_closure_preserves_noncanonical_dependency_errors() {
|
||||
let (blobs, store) = stores().await;
|
||||
let dependency = version("digraph Dependency {}", BTreeMap::new());
|
||||
let pretty = serde_json::to_vec_pretty(dependency.version()).unwrap();
|
||||
let dependency_id = WorkflowVersionId::from(blobs.write(&pretty).await.unwrap());
|
||||
let root = version(
|
||||
r#"digraph Root { dependency [stack.child_workflow="dependency.fabro"] }"#,
|
||||
BTreeMap::from([(path("dependency.fabro"), dependency_id)]),
|
||||
);
|
||||
let root_id = WorkflowVersionId::from(
|
||||
blobs
|
||||
.write(&root.version().canonical_bytes().unwrap())
|
||||
.await
|
||||
.unwrap(),
|
||||
);
|
||||
|
||||
let error = store.get_closure(&root_id).await.unwrap_err();
|
||||
let WorkflowVersionStoreError::DependencyInvalid { source, .. } = error else {
|
||||
panic!("expected invalid dependency error");
|
||||
};
|
||||
assert!(matches!(
|
||||
source.as_ref(),
|
||||
WorkflowVersionStoreError::NonCanonical { id } if *id == dependency_id
|
||||
));
|
||||
assert!(matches!(
|
||||
store.get_closure(&dependency_id).await.unwrap_err(),
|
||||
WorkflowVersionStoreError::NonCanonical { id } if id == dependency_id
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_projects_the_same_validated_root_as_get_closure() {
|
||||
let (_, store) = stores().await;
|
||||
let child = version("digraph Child {}", BTreeMap::new());
|
||||
let child_id = store.put(&child).await.unwrap();
|
||||
let root = version(
|
||||
r#"digraph Root { child [stack.child_workflow="child.fabro"] }"#,
|
||||
BTreeMap::from([(path("child.fabro"), child_id)]),
|
||||
);
|
||||
let root_id = store.put(&root).await.unwrap();
|
||||
|
||||
let closure = store.get_closure(&root_id).await.unwrap().unwrap();
|
||||
let projected = store.get(&root_id).await.unwrap().unwrap();
|
||||
|
||||
assert_eq!(projected.version(), closure.root());
|
||||
let absent = version_id(&version("digraph Absent {}", BTreeMap::new()));
|
||||
assert!(store.get_closure(&absent).await.unwrap().is_none());
|
||||
assert!(store.get(&absent).await.unwrap().is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_rejects_arbitrary_and_noncanonical_blobs() {
|
||||
let (blobs, store) = stores().await;
|
||||
|
|
|
|||
|
|
@ -611,12 +611,14 @@ pub enum ReferenceKind {
|
|||
Dockerfile,
|
||||
#[strum(to_string = "graph goal file reference")]
|
||||
GraphGoalFile,
|
||||
#[strum(to_string = "run goal file reference")]
|
||||
RunGoalFile,
|
||||
}
|
||||
|
||||
/// Kinds of static file references that graph attributes can carry: the
|
||||
/// subset of [`ReferenceKind`] that [`reference_kind_for_attribute`] can
|
||||
/// classify. Config-sourced kinds (Dockerfiles) are unrepresentable here by
|
||||
/// construction.
|
||||
/// classify. Config-sourced kinds (Dockerfiles, run goal files) are
|
||||
/// unrepresentable here by construction.
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
pub enum GraphReferenceKind {
|
||||
FileInline,
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue