mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-08-28 05:27:41 +00:00
Serialize workflow-version canonical bytes once at construction
WorkflowVersion::new serialized the whole version just to enforce the size limit and threw the bytes away, the store re-serialized them to write the blob, and every read re-serialized a third time for the canonicality comparison. Cache the canonical bytes on the struct at construction (skipped during serde) and expose them as an infallible borrow; the now-unconstructable InvalidShape store error variant goes away with it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Scott Werner
parent
e688bd9876
commit
20c9fba0b1
2 changed files with 32 additions and 36 deletions
|
|
@ -2,7 +2,7 @@ use std::collections::{BTreeMap, HashSet, VecDeque};
|
|||
use std::sync::Arc;
|
||||
|
||||
use fabro_store::BlobStore;
|
||||
use fabro_types::{WorkflowPath, WorkflowVersion, WorkflowVersionId, WorkflowVersionShapeError};
|
||||
use fabro_types::{WorkflowPath, WorkflowVersion, WorkflowVersionId};
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::{ValidatedWorkflowVersion, WorkflowVersionError};
|
||||
|
|
@ -11,8 +11,6 @@ use crate::{ValidatedWorkflowVersion, WorkflowVersionError};
|
|||
pub enum WorkflowVersionStoreError {
|
||||
#[error(transparent)]
|
||||
InvalidVersion(#[from] WorkflowVersionError),
|
||||
#[error(transparent)]
|
||||
InvalidShape(#[from] WorkflowVersionShapeError),
|
||||
#[error("workflow-version dependency `{id}` at `{path}` is not stored")]
|
||||
DependencyNotFound {
|
||||
path: WorkflowPath,
|
||||
|
|
@ -60,11 +58,10 @@ impl WorkflowVersionStore {
|
|||
&self,
|
||||
version: &ValidatedWorkflowVersion,
|
||||
) -> Result<WorkflowVersionId, WorkflowVersionStoreError> {
|
||||
let canonical = version.version().canonical_bytes()?;
|
||||
self.validate_dependency_closure(version.version().workflow_dependencies())
|
||||
.await?;
|
||||
self.blobs
|
||||
.write(&canonical)
|
||||
.write(version.version().canonical_bytes())
|
||||
.await
|
||||
.map(WorkflowVersionId::from)
|
||||
.map_err(|source| WorkflowVersionStoreError::Storage { source })
|
||||
|
|
@ -98,8 +95,7 @@ impl WorkflowVersionStore {
|
|||
let version = serde_json::from_slice::<WorkflowVersion>(&bytes)
|
||||
.map_err(|source| WorkflowVersionStoreError::Decode { id: *id, source })?;
|
||||
let validated = ValidatedWorkflowVersion::new(version)?;
|
||||
let canonical = validated.version().canonical_bytes()?;
|
||||
if canonical.as_slice() != bytes.as_ref() {
|
||||
if validated.version().canonical_bytes() != bytes.as_ref() {
|
||||
return Err(WorkflowVersionStoreError::NonCanonical { id: *id });
|
||||
}
|
||||
Ok(Some(validated))
|
||||
|
|
@ -196,7 +192,7 @@ mod tests {
|
|||
async fn put_get_reuses_exact_blob_digest() {
|
||||
let (blobs, store) = stores().await;
|
||||
let version = version("digraph W {}", BTreeMap::new());
|
||||
let expected_bytes = version.version().canonical_bytes().unwrap();
|
||||
let expected_bytes = version.version().canonical_bytes().to_vec();
|
||||
let expected_id = WorkflowVersionId::from(fabro_types::BlobHash::new(&expected_bytes));
|
||||
|
||||
let id = store.put(&version).await.unwrap();
|
||||
|
|
@ -228,15 +224,14 @@ mod tests {
|
|||
let (blobs, store) = stores().await;
|
||||
let child = version("digraph Child {}", BTreeMap::new());
|
||||
let child_id = WorkflowVersionId::from(fabro_types::BlobHash::new(
|
||||
&child.version().canonical_bytes().unwrap(),
|
||||
child.version().canonical_bytes(),
|
||||
));
|
||||
let root = version(
|
||||
r#"digraph Root { child [stack.child_workflow="child.fabro"] }"#,
|
||||
BTreeMap::from([(path("child.fabro"), child_id)]),
|
||||
);
|
||||
let root_id = WorkflowVersionId::from(fabro_types::BlobHash::new(
|
||||
&root.version().canonical_bytes().unwrap(),
|
||||
));
|
||||
let root_id =
|
||||
WorkflowVersionId::from(fabro_types::BlobHash::new(root.version().canonical_bytes()));
|
||||
|
||||
let error = store.put(&root).await.unwrap_err();
|
||||
assert!(matches!(
|
||||
|
|
@ -256,15 +251,14 @@ mod tests {
|
|||
r#"digraph Child { grandchild [stack.child_workflow="grandchild.fabro"] }"#,
|
||||
BTreeMap::from([(path("grandchild.fabro"), missing_grandchild_id)]),
|
||||
);
|
||||
let child_bytes = child.version().canonical_bytes().unwrap();
|
||||
let child_id = WorkflowVersionId::from(blobs.write(&child_bytes).await.unwrap());
|
||||
let child_bytes = child.version().canonical_bytes();
|
||||
let child_id = WorkflowVersionId::from(blobs.write(child_bytes).await.unwrap());
|
||||
let root = version(
|
||||
r#"digraph Root { child [stack.child_workflow="child.fabro"] }"#,
|
||||
BTreeMap::from([(path("child.fabro"), child_id)]),
|
||||
);
|
||||
let root_id = WorkflowVersionId::from(fabro_types::BlobHash::new(
|
||||
&root.version().canonical_bytes().unwrap(),
|
||||
));
|
||||
let root_id =
|
||||
WorkflowVersionId::from(fabro_types::BlobHash::new(root.version().canonical_bytes()));
|
||||
|
||||
assert!(matches!(
|
||||
store.put(&root).await.unwrap_err(),
|
||||
|
|
|
|||
|
|
@ -53,6 +53,8 @@ pub struct WorkflowVersion {
|
|||
entrypoint: WorkflowPath,
|
||||
files: BTreeMap<WorkflowPath, String>,
|
||||
workflow_dependencies: BTreeMap<WorkflowPath, WorkflowVersionId>,
|
||||
#[serde(skip)]
|
||||
canonical: Vec<u8>,
|
||||
}
|
||||
|
||||
impl WorkflowVersion {
|
||||
|
|
@ -61,13 +63,22 @@ impl WorkflowVersion {
|
|||
files: BTreeMap<WorkflowPath, String>,
|
||||
workflow_dependencies: BTreeMap<WorkflowPath, WorkflowVersionId>,
|
||||
) -> Result<Self, WorkflowVersionShapeError> {
|
||||
let version = Self {
|
||||
let mut version = Self {
|
||||
entrypoint,
|
||||
files,
|
||||
workflow_dependencies,
|
||||
canonical: Vec::new(),
|
||||
};
|
||||
version.validate_shape()?;
|
||||
version.canonical_bytes()?;
|
||||
let canonical = serde_json::to_vec(&version)
|
||||
.map_err(|source| WorkflowVersionShapeError::Serialization { source })?;
|
||||
if canonical.len() > MAX_WORKFLOW_VERSION_BYTES {
|
||||
return Err(WorkflowVersionShapeError::VersionTooLarge {
|
||||
actual: canonical.len(),
|
||||
maximum: MAX_WORKFLOW_VERSION_BYTES,
|
||||
});
|
||||
}
|
||||
version.canonical = canonical;
|
||||
Ok(version)
|
||||
}
|
||||
|
||||
|
|
@ -86,20 +97,11 @@ impl WorkflowVersion {
|
|||
&self.workflow_dependencies
|
||||
}
|
||||
|
||||
/// Serialize to the canonical wire form.
|
||||
///
|
||||
/// Structural validity is guaranteed by construction, so this only
|
||||
/// serializes and enforces the canonical size limit.
|
||||
pub fn canonical_bytes(&self) -> Result<Vec<u8>, WorkflowVersionShapeError> {
|
||||
let bytes = serde_json::to_vec(self)
|
||||
.map_err(|source| WorkflowVersionShapeError::Serialization { source })?;
|
||||
if bytes.len() > MAX_WORKFLOW_VERSION_BYTES {
|
||||
return Err(WorkflowVersionShapeError::VersionTooLarge {
|
||||
actual: bytes.len(),
|
||||
maximum: MAX_WORKFLOW_VERSION_BYTES,
|
||||
});
|
||||
}
|
||||
Ok(bytes)
|
||||
/// Canonical wire bytes, serialized and size-checked once at
|
||||
/// construction.
|
||||
#[must_use]
|
||||
pub fn canonical_bytes(&self) -> &[u8] {
|
||||
&self.canonical
|
||||
}
|
||||
|
||||
fn validate_shape(&self) -> Result<(), WorkflowVersionShapeError> {
|
||||
|
|
@ -250,7 +252,7 @@ mod tests {
|
|||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
String::from_utf8(version.canonical_bytes().unwrap()).unwrap(),
|
||||
String::from_utf8(version.canonical_bytes().to_vec()).unwrap(),
|
||||
r#"{"entrypoint":"workflow.fabro","files":{"a.txt":"A","workflow.fabro":"digraph W {}","z.txt":"Z"},"workflow_dependencies":{}}"#
|
||||
);
|
||||
}
|
||||
|
|
@ -419,7 +421,7 @@ mod tests {
|
|||
BTreeMap::new(),
|
||||
)
|
||||
.unwrap();
|
||||
let remaining = MAX_WORKFLOW_VERSION_BYTES - empty.canonical_bytes().unwrap().len();
|
||||
let remaining = MAX_WORKFLOW_VERSION_BYTES - empty.canonical_bytes().len();
|
||||
let per_file = remaining / 4;
|
||||
let remainder = remaining % 4;
|
||||
for index in 0..4 {
|
||||
|
|
@ -434,7 +436,7 @@ mod tests {
|
|||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
exact_version.canonical_bytes().unwrap().len(),
|
||||
exact_version.canonical_bytes().len(),
|
||||
MAX_WORKFLOW_VERSION_BYTES
|
||||
);
|
||||
exact_version_files
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue