From c372960a32469952eb4e74024ec824cd8520c141 Mon Sep 17 00:00:00 2001 From: Tom Dupuis <60640908+tomdps@users.noreply.github.com> Date: Thu, 16 Jul 2026 18:05:48 -0700 Subject: [PATCH] feat: implement #699 - feat(zeroshot-rust): add injected ArtifactStore and local CAS --- AGENTS.md | 10 +- Cargo.lock | 34 ++ Cargo.toml | 3 +- zeroshot-rust/Cargo.toml | 4 +- zeroshot-rust/src/artifact_store.rs | 387 ++++++++++++++ zeroshot-rust/src/artifact_store/fake.rs | 345 +++++++++++++ zeroshot-rust/src/artifact_store/local_cas.rs | 272 ++++++++++ .../artifact_store/local_cas/filesystem.rs | 442 ++++++++++++++++ .../artifact_store/local_cas/operations.rs | 488 ++++++++++++++++++ zeroshot-rust/src/lib.rs | 1 + zeroshot-rust/tests/architecture.rs | 89 +++- zeroshot-rust/tests/artifact_store.rs | 334 ++++++++++++ zeroshot-rust/tests/local_cas.rs | 453 ++++++++++++++++ zeroshot-rust/tests/local_cas/recovery.rs | 137 +++++ zeroshot-rust/tests/support/mod.rs | 39 ++ 15 files changed, 3032 insertions(+), 6 deletions(-) create mode 100644 zeroshot-rust/src/artifact_store.rs create mode 100644 zeroshot-rust/src/artifact_store/fake.rs create mode 100644 zeroshot-rust/src/artifact_store/local_cas.rs create mode 100644 zeroshot-rust/src/artifact_store/local_cas/filesystem.rs create mode 100644 zeroshot-rust/src/artifact_store/local_cas/operations.rs create mode 100644 zeroshot-rust/tests/artifact_store.rs create mode 100644 zeroshot-rust/tests/local_cas.rs create mode 100644 zeroshot-rust/tests/local_cas/recovery.rs create mode 100644 zeroshot-rust/tests/support/mod.rs diff --git a/AGENTS.md b/AGENTS.md index 22e772c2..767c057c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -54,6 +54,8 @@ Destructive commands (need permission): `zeroshot kill`, `zeroshot clear`, `zero | Shared wire-value bounds | `crates/openengine-cluster-protocol/src/value.rs` | | Cluster dispatch/stdio | `crates/openengine-cluster-server/` | | Native product construction | `zeroshot-rust/` | +| Artifact store port/fake | `zeroshot-rust/src/artifact_store.rs`, `artifact_store/fake.rs` | +| Product-local artifact CAS | `zeroshot-rust/src/artifact_store/local_cas.rs`, `local_cas/` | | Issue provider contracts | `zeroshot-rust/src/issue_provider.rs`, `issue_provider/` | | Source provider contracts | `zeroshot-rust/src/source_code_provider.rs`, `source_code_provider/` | | Provider value bounds | `zeroshot-rust/src/provider_value.rs`, `provider_value/` | @@ -86,8 +88,12 @@ verify byte-for-byte drift with `npm run protocol:check`. These generator-format are excluded from Prettier; never format them independently. The protocol and server crates own wire contracts, backend traits, the dispatcher, and transports. `zeroshot-rust/` owns the concrete `NativeBackend`, product-local `NativeBackendFactory` -construction root, and product-private, secret-free issue/source provider contracts. Issue and -source registries and identifiers remain independent; neither is a worker/model provider. Keep +construction root, product-private artifact byte-store port/local CAS, and product-private, +secret-free issue/source provider contracts. Artifact stages, bytes, roots, filesystem paths, +locks, and manifests remain product-private; only verified protocol `ArtifactRef` receipts cross +the engine boundary. `LocalCasArtifactStore` takes an explicit root, is a single-writer local +filesystem store, and must preserve ref-first release plus synchronized blob-then-ref publication. +Issue and source registries and identifiers remain independent; neither is a worker/model provider. Keep protocol, transport, daemon, compatibility, adapter, credential resolution, ledger, and workspace behavior outside it. Native engine faults must be constructed only by `FaultFactory` from closed `ModuleEvidence`. diff --git a/Cargo.lock b/Cargo.lock index 83998189..af51cb5c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -205,6 +205,16 @@ dependencies = [ "num", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "futures-core" version = "0.3.32" @@ -1029,6 +1039,28 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-link" version = "0.2.1" @@ -1125,10 +1157,12 @@ name = "zeroshot-rust" version = "0.1.0" dependencies = [ "async-trait", + "fs2", "openengine-cluster-protocol", "openengine-cluster-server", "serde", "serde_json", + "sha2", "thiserror", "tokio", ] diff --git a/Cargo.toml b/Cargo.toml index 6ec6c7e0..1c590b0b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,12 +16,13 @@ repository = "https://github.com/the-open-engine/zeroshot" [workspace.dependencies] async-trait = "0.1.89" +fs2 = "0.4.3" schemars = "1.1.0" serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.145" sha2 = "0.10.9" thiserror = "2.0.17" -tokio = { version = "1.48.0", features = ["io-std", "io-util", "macros", "process", "rt-multi-thread", "sync"] } +tokio = { version = "1.48.0", features = ["fs", "io-std", "io-util", "macros", "process", "rt-multi-thread", "sync"] } jsonschema = { version = "0.29.1", default-features = false } openengine-cluster-protocol = { path = "crates/openengine-cluster-protocol" } diff --git a/zeroshot-rust/Cargo.toml b/zeroshot-rust/Cargo.toml index 7f34310d..3766f730 100644 --- a/zeroshot-rust/Cargo.toml +++ b/zeroshot-rust/Cargo.toml @@ -16,11 +16,13 @@ path = "src/main.rs" [dependencies] async-trait.workspace = true +fs2.workspace = true openengine-cluster-protocol.workspace = true openengine-cluster-server.workspace = true serde.workspace = true serde_json.workspace = true +sha2.workspace = true thiserror.workspace = true +tokio.workspace = true [dev-dependencies] -tokio.workspace = true diff --git a/zeroshot-rust/src/artifact_store.rs b/zeroshot-rust/src/artifact_store.rs new file mode 100644 index 00000000..dacbb9d8 --- /dev/null +++ b/zeroshot-rust/src/artifact_store.rs @@ -0,0 +1,387 @@ +//! Product-private artifact byte storage behind protocol-owned receipts. + +use std::error::Error; +use std::fmt; + +use async_trait::async_trait; +use openengine_cluster_protocol::{ + ArtifactId, ArtifactLineage, ArtifactProducer, ArtifactRef, ByteLength, MediaType, + RedactionClass, Sha256Digest, TypeId, +}; +use sha2::{Digest, Sha256}; +use tokio::io::{AsyncRead, AsyncReadExt}; + +use crate::fault::{EvidenceClass, FaultContext, FaultModule, ModuleEvidence}; + +pub mod fake; +pub mod local_cas; + +pub const MAX_ARTIFACT_BYTES: u64 = 64 * 1024 * 1024; +const ARTIFACT_ID_DOMAIN: &str = "zeroshot-rust.artifact-ref/v1"; + +pub type ArtifactByteStream = Box; +pub type VerifiedArtifactStream = Box; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ArtifactIntent { + pub expected_sha256: Sha256Digest, + pub expected_byte_length: ByteLength, + pub media_type: MediaType, + pub type_id: TypeId, + pub producer: ArtifactProducer, + pub lineage: ArtifactLineage, + pub redaction: RedactionClass, +} + +impl ArtifactIntent { + #[must_use] + pub fn artifact_ref(&self) -> ArtifactRef { + ArtifactRef { + artifact_id: derive_artifact_id(self), + sha256: self.expected_sha256.clone(), + byte_length: self.expected_byte_length, + media_type: self.media_type.clone(), + type_id: self.type_id.clone(), + producer: self.producer.clone(), + lineage: self.lineage.clone(), + redaction: self.redaction, + } + } +} + +/// Opaque, store-bound handle. Its debug form never reveals storage details. +pub struct StagedArtifact { + store_key: usize, + stage_key: u64, + artifact_ref: ArtifactRef, +} + +impl StagedArtifact { + pub(crate) const fn new(store_key: usize, stage_key: u64, artifact_ref: ArtifactRef) -> Self { + Self { + store_key, + stage_key, + artifact_ref, + } + } + + pub(crate) const fn store_key(&self) -> usize { + self.store_key + } + + pub(crate) const fn stage_key(&self) -> u64 { + self.stage_key + } + + #[must_use] + pub fn artifact_id(&self) -> &ArtifactId { + &self.artifact_ref.artifact_id + } + + pub(crate) fn artifact_ref(&self) -> &ArtifactRef { + &self.artifact_ref + } +} + +impl fmt::Debug for StagedArtifact { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("StagedArtifact") + .field("artifact_id", &self.artifact_ref.artifact_id) + .field("storage", &"[redacted]") + .finish() + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DiscardResult { + Discarded, + AlreadyDiscarded, + AlreadyPublished, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ReleaseResult { + Released, + NotFound, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ArtifactStoreOperation { + Configuration, + Stage, + Publish, + Inspect, + Open, + Discard, + Release, +} + +impl ArtifactStoreOperation { + const fn fault_context(self) -> FaultContext { + match self { + Self::Configuration => FaultContext::Configuration, + Self::Stage | Self::Publish => FaultContext::Settlement, + Self::Inspect | Self::Open => FaultContext::Recovery, + Self::Discard | Self::Release => FaultContext::Cleanup, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ArtifactStoreFailureKind { + Oversize, + LengthMismatch, + HashMismatch, + RootUnavailable, + LockUnavailable, + Io(ArtifactStoreOperation), + PermissionDenied(ArtifactStoreOperation), + CorruptContent, + MissingCommittedContent, + IdentityConflict, + InvalidStage, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ArtifactStoreFailure { + kind: ArtifactStoreFailureKind, +} + +impl ArtifactStoreFailure { + #[must_use] + pub const fn new(kind: ArtifactStoreFailureKind) -> Self { + Self { kind } + } + + #[must_use] + pub const fn kind(self) -> ArtifactStoreFailureKind { + self.kind + } + + #[must_use] + pub const fn module_evidence(self) -> ModuleEvidence { + match self.kind { + ArtifactStoreFailureKind::Oversize => ModuleEvidence::new( + FaultModule::Worker, + FaultContext::Settlement, + EvidenceClass::ResourceExhausted, + ), + ArtifactStoreFailureKind::LengthMismatch | ArtifactStoreFailureKind::HashMismatch => { + ModuleEvidence::new( + FaultModule::Worker, + FaultContext::Settlement, + EvidenceClass::MalformedExternalData, + ) + } + ArtifactStoreFailureKind::RootUnavailable + | ArtifactStoreFailureKind::LockUnavailable => ModuleEvidence::new( + FaultModule::Storage, + FaultContext::Configuration, + EvidenceClass::Unavailable, + ), + ArtifactStoreFailureKind::Io(operation) => ModuleEvidence::new( + FaultModule::Storage, + operation.fault_context(), + EvidenceClass::Unavailable, + ), + ArtifactStoreFailureKind::PermissionDenied(operation) => ModuleEvidence::new( + FaultModule::Storage, + operation.fault_context(), + EvidenceClass::PermissionDenied, + ), + ArtifactStoreFailureKind::CorruptContent + | ArtifactStoreFailureKind::MissingCommittedContent + | ArtifactStoreFailureKind::IdentityConflict + | ArtifactStoreFailureKind::InvalidStage => ModuleEvidence::new( + FaultModule::Storage, + FaultContext::Recovery, + EvidenceClass::IntegrityFailure, + ), + } + } +} + +impl fmt::Display for ArtifactStoreFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + let message = match self.kind { + ArtifactStoreFailureKind::Io(_) => "artifact storage operation is unavailable", + ArtifactStoreFailureKind::PermissionDenied(_) => { + "artifact storage operation lacks permission" + } + kind => simple_failure_message(kind), + }; + formatter.write_str(message) + } +} + +const fn simple_failure_message(kind: ArtifactStoreFailureKind) -> &'static str { + match kind { + ArtifactStoreFailureKind::Oversize => "artifact exceeds the byte limit", + ArtifactStoreFailureKind::LengthMismatch => "artifact length does not match intent", + ArtifactStoreFailureKind::HashMismatch => "artifact digest does not match intent", + ArtifactStoreFailureKind::RootUnavailable => "artifact store root is unavailable", + ArtifactStoreFailureKind::LockUnavailable => "artifact store lock is unavailable", + kind => integrity_failure_message(kind), + } +} + +const fn integrity_failure_message(kind: ArtifactStoreFailureKind) -> &'static str { + match kind { + ArtifactStoreFailureKind::CorruptContent => "committed artifact content is corrupt", + ArtifactStoreFailureKind::MissingCommittedContent => { + "committed artifact content is missing" + } + ArtifactStoreFailureKind::IdentityConflict => "artifact identity conflicts", + ArtifactStoreFailureKind::InvalidStage => "artifact stage is invalid", + _ => "artifact storage operation failed", + } +} + +impl Error for ArtifactStoreFailure {} + +#[async_trait] +pub trait ArtifactStore: Send + Sync { + async fn stage( + &self, + intent: ArtifactIntent, + bytes: ArtifactByteStream, + ) -> Result; + + async fn publish(&self, staged: &StagedArtifact) -> Result; + + async fn inspect( + &self, + artifact_id: &ArtifactId, + ) -> Result, ArtifactStoreFailure>; + + async fn open( + &self, + artifact_id: &ArtifactId, + ) -> Result; + + async fn discard(&self, staged: &StagedArtifact) + -> Result; + + async fn release( + &self, + artifact_id: &ArtifactId, + ) -> Result; +} + +#[must_use] +pub fn derive_artifact_id(intent: &ArtifactIntent) -> ArtifactId { + let mut preimage = Vec::new(); + append_identity_field(&mut preimage, ARTIFACT_ID_DOMAIN.as_bytes()); + append_identity_field(&mut preimage, intent.expected_sha256.as_str().as_bytes()); + append_identity_field( + &mut preimage, + intent.expected_byte_length.get().to_string().as_bytes(), + ); + append_identity_field(&mut preimage, intent.media_type.as_str().as_bytes()); + append_identity_field(&mut preimage, intent.type_id.as_str().as_bytes()); + append_identity_field(&mut preimage, intent.producer.node.as_str().as_bytes()); + append_identity_field(&mut preimage, intent.producer.worker.as_str().as_bytes()); + append_identity_field( + &mut preimage, + intent.lineage.generation.get().to_string().as_bytes(), + ); + append_identity_field(&mut preimage, intent.lineage.run_id.as_str().as_bytes()); + append_identity_field( + &mut preimage, + intent.lineage.attempt.get().to_string().as_bytes(), + ); + append_identity_field( + &mut preimage, + match intent.redaction { + RedactionClass::Public => b"public", + RedactionClass::Internal => b"internal", + RedactionClass::Confidential => b"confidential", + RedactionClass::Restricted => b"restricted", + }, + ); + ArtifactId::new(format!("cas-v1-{}", sha256_hex(&preimage))) + .expect("derived artifact identity is protocol-valid") +} + +fn append_identity_field(preimage: &mut Vec, value: &[u8]) { + preimage.extend_from_slice( + &u64::try_from(value.len()) + .expect("protocol field length fits u64") + .to_be_bytes(), + ); + preimage.extend_from_slice(value); +} + +pub(crate) async fn read_verified_bytes( + intent: &ArtifactIntent, + mut bytes: ArtifactByteStream, +) -> Result, ArtifactStoreFailure> { + let declared = intent.expected_byte_length.get(); + if declared > MAX_ARTIFACT_BYTES { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::Oversize, + )); + } + + let capacity = usize::try_from(declared).expect("artifact limit fits usize"); + let mut output = Vec::with_capacity(capacity); + let mut limited = (&mut bytes).take(declared + 1); + limited + .read_to_end(&mut output) + .await + .map_err(|error| failure_from_io(error, ArtifactStoreOperation::Stage))?; + if output.len() as u64 > declared { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::Oversize, + )); + } + if output.len() as u64 != declared { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::LengthMismatch, + )); + } + if sha256_hex(&output) != intent.expected_sha256.as_str() { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::HashMismatch, + )); + } + Ok(output) +} + +pub(crate) fn verify_bytes( + artifact_ref: &ArtifactRef, + bytes: &[u8], +) -> Result<(), ArtifactStoreFailure> { + if bytes.len() as u64 != artifact_ref.byte_length.get() + || sha256_hex(bytes) != artifact_ref.sha256.as_str() + { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::CorruptContent, + )); + } + Ok(()) +} + +pub(crate) fn sha256_hex(bytes: &[u8]) -> String { + let digest = Sha256::digest(bytes); + let mut output = String::with_capacity(64); + const HEX: &[u8; 16] = b"0123456789abcdef"; + for byte in digest { + output.push(char::from(HEX[usize::from(byte >> 4)])); + output.push(char::from(HEX[usize::from(byte & 0x0f)])); + } + output +} + +pub(crate) fn failure_from_io( + error: std::io::Error, + operation: ArtifactStoreOperation, +) -> ArtifactStoreFailure { + let kind = if error.kind() == std::io::ErrorKind::PermissionDenied { + ArtifactStoreFailureKind::PermissionDenied(operation) + } else { + ArtifactStoreFailureKind::Io(operation) + }; + ArtifactStoreFailure::new(kind) +} diff --git a/zeroshot-rust/src/artifact_store/fake.rs b/zeroshot-rust/src/artifact_store/fake.rs new file mode 100644 index 00000000..fca830ea --- /dev/null +++ b/zeroshot-rust/src/artifact_store/fake.rs @@ -0,0 +1,345 @@ +//! Deterministic in-memory artifact store for conformance and recovery tests. + +use std::collections::{BTreeMap, VecDeque}; +use std::io::Cursor; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; + +use async_trait::async_trait; +use openengine_cluster_protocol::{ArtifactId, ArtifactRef}; + +use super::{ + ArtifactByteStream, ArtifactIntent, ArtifactStore, ArtifactStoreFailure, + ArtifactStoreFailureKind, DiscardResult, ReleaseResult, StagedArtifact, VerifiedArtifactStream, + read_verified_bytes, verify_bytes, +}; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum FakeFailurePoint { + BeforeStageCommit, + BeforePublishCommit, + AfterPublishCommit, + Inspect, + Open, + Discard, + Release, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ScriptedFailure { + pub point: FakeFailurePoint, + pub kind: ArtifactStoreFailureKind, +} + +#[derive(Clone)] +pub struct FakeArtifactStore { + inner: Arc, +} + +struct Inner { + state: Mutex, + next_stage: AtomicU64, +} + +#[derive(Default)] +struct State { + stages: BTreeMap, + blobs: BTreeMap>, + refs: BTreeMap, + failures: VecDeque, +} + +#[derive(Clone)] +struct FakeStage { + artifact_ref: ArtifactRef, + bytes: Vec, + status: FakeStageStatus, +} + +#[derive(Clone, Copy, Eq, PartialEq)] +enum FakeStageStatus { + Pending, + Published, + Discarded, +} + +impl FakeArtifactStore { + #[must_use] + pub fn new() -> Self { + Self { + inner: Arc::new(Inner { + state: Mutex::new(State::default()), + next_stage: AtomicU64::new(1), + }), + } + } + + pub fn script_failure(&self, point: FakeFailurePoint, kind: ArtifactStoreFailureKind) { + self.state() + .failures + .push_back(ScriptedFailure { point, kind }); + } + + pub fn script_failures(&self, failures: impl IntoIterator) { + self.state().failures.extend(failures); + } + + /// Simulates process recovery: committed data remains and every stage is abandoned. + pub fn restart(&self) { + let mut state = self.state(); + state.stages.clear(); + state.failures.clear(); + } + + #[must_use] + pub fn blob_count(&self) -> usize { + self.state().blobs.len() + } + + #[must_use] + pub fn committed_ref_count(&self) -> usize { + self.state().refs.len() + } + + #[must_use] + pub fn staged_count(&self) -> usize { + self.state().stages.len() + } + + fn state(&self) -> MutexGuard<'_, State> { + self.inner + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + fn store_key(&self) -> usize { + Arc::as_ptr(&self.inner) as usize + } + + fn check_stage(&self, staged: &StagedArtifact) -> Result<(), ArtifactStoreFailure> { + if staged.store_key() != self.store_key() { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::InvalidStage, + )); + } + Ok(()) + } +} + +impl Default for FakeArtifactStore { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl ArtifactStore for FakeArtifactStore { + async fn stage( + &self, + intent: ArtifactIntent, + bytes: ArtifactByteStream, + ) -> Result { + let bytes = read_verified_bytes(&intent, bytes).await?; + let artifact_ref = intent.artifact_ref(); + let stage_key = self.inner.next_stage.fetch_add(1, Ordering::Relaxed); + let mut state = self.state(); + maybe_fail(&mut state, FakeFailurePoint::BeforeStageCommit)?; + state.stages.insert( + stage_key, + FakeStage { + artifact_ref: artifact_ref.clone(), + bytes, + status: FakeStageStatus::Pending, + }, + ); + Ok(StagedArtifact::new( + self.store_key(), + stage_key, + artifact_ref, + )) + } + + async fn publish(&self, staged: &StagedArtifact) -> Result { + self.check_stage(staged)?; + let mut state = self.state(); + maybe_fail(&mut state, FakeFailurePoint::BeforePublishCommit)?; + let stage = checked_fake_stage(&state, staged)?; + if let Some(published) = published_retry(&stage)? { + return Ok(published); + } + commit_fake_blob(&mut state, &stage)?; + commit_fake_ref(&mut state, &stage.artifact_ref)?; + state + .stages + .get_mut(&staged.stage_key()) + .expect("stage was checked while holding store lock") + .status = FakeStageStatus::Published; + maybe_fail(&mut state, FakeFailurePoint::AfterPublishCommit)?; + Ok(stage.artifact_ref) + } + + async fn inspect( + &self, + artifact_id: &ArtifactId, + ) -> Result, ArtifactStoreFailure> { + let mut state = self.state(); + maybe_fail(&mut state, FakeFailurePoint::Inspect)?; + let Some(artifact_ref) = state.refs.get(artifact_id.as_str()).cloned() else { + return Ok(None); + }; + let bytes = state + .blobs + .get(artifact_ref.sha256.as_str()) + .ok_or_else(|| { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::MissingCommittedContent) + })?; + verify_bytes(&artifact_ref, bytes)?; + Ok(Some(artifact_ref)) + } + + async fn open( + &self, + artifact_id: &ArtifactId, + ) -> Result { + let mut state = self.state(); + maybe_fail(&mut state, FakeFailurePoint::Open)?; + let artifact_ref = state.refs.get(artifact_id.as_str()).ok_or_else(|| { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::MissingCommittedContent) + })?; + let bytes = state + .blobs + .get(artifact_ref.sha256.as_str()) + .ok_or_else(|| { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::MissingCommittedContent) + })?; + verify_bytes(artifact_ref, bytes)?; + Ok(Box::new(Cursor::new(bytes.clone()))) + } + + async fn discard( + &self, + staged: &StagedArtifact, + ) -> Result { + self.check_stage(staged)?; + let mut state = self.state(); + maybe_fail(&mut state, FakeFailurePoint::Discard)?; + let stage = state + .stages + .get_mut(&staged.stage_key()) + .ok_or_else(|| ArtifactStoreFailure::new(ArtifactStoreFailureKind::InvalidStage))?; + let result = match stage.status { + FakeStageStatus::Pending => { + stage.bytes.clear(); + stage.status = FakeStageStatus::Discarded; + DiscardResult::Discarded + } + FakeStageStatus::Discarded => DiscardResult::AlreadyDiscarded, + FakeStageStatus::Published => DiscardResult::AlreadyPublished, + }; + Ok(result) + } + + async fn release( + &self, + artifact_id: &ArtifactId, + ) -> Result { + let mut state = self.state(); + maybe_fail(&mut state, FakeFailurePoint::Release)?; + let Some(artifact_ref) = state.refs.get(artifact_id.as_str()).cloned() else { + return Ok(ReleaseResult::NotFound); + }; + let bytes = state + .blobs + .get(artifact_ref.sha256.as_str()) + .ok_or_else(|| { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::MissingCommittedContent) + })?; + verify_bytes(&artifact_ref, bytes)?; + state.refs.remove(artifact_id.as_str()); + let digest_is_referenced = state + .refs + .values() + .any(|candidate| candidate.sha256 == artifact_ref.sha256); + if !digest_is_referenced { + state.blobs.remove(artifact_ref.sha256.as_str()); + } + Ok(ReleaseResult::Released) + } +} + +fn maybe_fail(state: &mut State, point: FakeFailurePoint) -> Result<(), ArtifactStoreFailure> { + if state + .failures + .front() + .is_some_and(|failure| failure.point == point) + { + let failure = state + .failures + .pop_front() + .expect("front failure was present"); + return Err(ArtifactStoreFailure::new(failure.kind)); + } + Ok(()) +} + +fn checked_fake_stage( + state: &State, + staged: &StagedArtifact, +) -> Result { + let stage = state + .stages + .get(&staged.stage_key()) + .cloned() + .ok_or_else(invalid_stage)?; + if stage.artifact_ref != *staged.artifact_ref() { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::IdentityConflict, + )); + } + Ok(stage) +} + +fn published_retry(stage: &FakeStage) -> Result, ArtifactStoreFailure> { + match stage.status { + FakeStageStatus::Pending => Ok(None), + FakeStageStatus::Published => Ok(Some(stage.artifact_ref.clone())), + FakeStageStatus::Discarded => Err(invalid_stage()), + } +} + +fn commit_fake_blob(state: &mut State, stage: &FakeStage) -> Result<(), ArtifactStoreFailure> { + let digest = stage.artifact_ref.sha256.as_str().to_owned(); + let Some(existing) = state.blobs.get(&digest) else { + state.blobs.insert(digest, stage.bytes.clone()); + return Ok(()); + }; + verify_bytes(&stage.artifact_ref, existing)?; + if existing != &stage.bytes { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::IdentityConflict, + )); + } + Ok(()) +} + +fn commit_fake_ref( + state: &mut State, + artifact_ref: &ArtifactRef, +) -> Result<(), ArtifactStoreFailure> { + let artifact_id = artifact_ref.artifact_id.as_str().to_owned(); + let Some(existing) = state.refs.get(&artifact_id) else { + state.refs.insert(artifact_id, artifact_ref.clone()); + return Ok(()); + }; + if existing != artifact_ref { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::IdentityConflict, + )); + } + Ok(()) +} + +fn invalid_stage() -> ArtifactStoreFailure { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::InvalidStage) +} diff --git a/zeroshot-rust/src/artifact_store/local_cas.rs b/zeroshot-rust/src/artifact_store/local_cas.rs new file mode 100644 index 00000000..890b6c11 --- /dev/null +++ b/zeroshot-rust/src/artifact_store/local_cas.rs @@ -0,0 +1,272 @@ +//! Single-writer, product-local filesystem content-addressed artifact store. + +use std::collections::BTreeMap; +use std::io::Cursor; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; + +use async_trait::async_trait; +use openengine_cluster_protocol::{ArtifactId, ArtifactRef}; +use tokio::sync::Mutex as AsyncMutex; + +use super::{ + ArtifactByteStream, ArtifactIntent, ArtifactStore, ArtifactStoreFailure, + ArtifactStoreFailureKind, ArtifactStoreOperation, DiscardResult, MAX_ARTIFACT_BYTES, + ReleaseResult, StagedArtifact, VerifiedArtifactStream, +}; + +mod filesystem; +mod operations; + +use filesystem::{ + acquire_root_lock, cleanup_abandoned_stages, prepare_owned_directory, prepare_root, + reject_symlink_or_non_file_if_present, sync_directory, validate_artifact_id, +}; + +const MAX_MANIFEST_BYTES: u64 = 16 * 1024; + +#[derive(Clone)] +pub struct LocalCasArtifactStore { + pub(super) inner: Arc, +} + +pub(super) struct Inner { + pub(super) root: PathBuf, + _root_lock: std::fs::File, + pub(super) operation: AsyncMutex<()>, + stages: Mutex>, + next_stage: AtomicU64, +} + +#[derive(Clone)] +pub(super) struct LocalStage { + pub(super) artifact_ref: ArtifactRef, + pub(super) path: PathBuf, + pub(super) status: LocalStageStatus, +} + +#[derive(Clone, Copy, Eq, PartialEq)] +pub(super) enum LocalStageStatus { + Pending, + Published, + Discarded, +} + +impl LocalCasArtifactStore { + pub fn new(root: impl AsRef) -> Result { + let root = root.as_ref().to_path_buf(); + prepare_root(&root)?; + let lock_path = root.join("store.lock"); + reject_symlink_or_non_file_if_present(&root, &lock_path)?; + let lock = acquire_root_lock(&root, &lock_path)?; + prepare_store_directories(&root)?; + cleanup_abandoned_stages(&root, &root.join("staging"))?; + Ok(Self::from_locked_root(root, lock)) + } + + fn from_locked_root(root: PathBuf, lock: std::fs::File) -> Self { + Self { + inner: Arc::new(Inner { + root, + _root_lock: lock, + operation: AsyncMutex::new(()), + stages: Mutex::new(BTreeMap::new()), + next_stage: AtomicU64::new(1), + }), + } + } + + fn store_key(&self) -> usize { + Arc::as_ptr(&self.inner) as usize + } + + pub(super) fn stages(&self) -> MutexGuard<'_, BTreeMap> { + self.inner + .stages + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + fn checked_stage(&self, staged: &StagedArtifact) -> Result { + if staged.store_key() != self.store_key() { + return Err(invalid_stage()); + } + let stage = self + .stages() + .get(&staged.stage_key()) + .cloned() + .ok_or_else(invalid_stage)?; + if stage.artifact_ref != *staged.artifact_ref() { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::IdentityConflict, + )); + } + Ok(stage) + } + + pub(super) fn staging_directory(&self) -> PathBuf { + self.inner.root.join("staging") + } + + pub(super) fn refs_directory(&self) -> PathBuf { + self.inner.root.join("refs") + } + + pub(super) fn ref_path( + &self, + artifact_id: &ArtifactId, + ) -> Result { + validate_artifact_id(artifact_id)?; + Ok(self + .refs_directory() + .join(format!("{}.json", artifact_id.as_str()))) + } + + pub(super) fn blob_path(&self, artifact_ref: &ArtifactRef) -> PathBuf { + let digest = artifact_ref.sha256.as_str(); + self.inner + .root + .join("blobs/sha256") + .join(&digest[..2]) + .join(digest) + } + + fn stage_path(&self, stage_key: u64) -> PathBuf { + self.staging_directory() + .join(format!("stage-{}-{stage_key}.tmp", std::process::id())) + } + + fn manifest_stage_path(&self, stage_key: u64, artifact_id: &ArtifactId) -> PathBuf { + self.staging_directory() + .join(format!("ref-{}-{stage_key}.tmp", artifact_id.as_str())) + } + + fn register_stage( + &self, + stage_key: u64, + path: PathBuf, + artifact_ref: ArtifactRef, + ) -> StagedArtifact { + self.stages().insert( + stage_key, + LocalStage { + artifact_ref: artifact_ref.clone(), + path, + status: LocalStageStatus::Pending, + }, + ); + StagedArtifact::new(self.store_key(), stage_key, artifact_ref) + } +} + +#[async_trait] +impl ArtifactStore for LocalCasArtifactStore { + async fn stage( + &self, + intent: ArtifactIntent, + bytes: ArtifactByteStream, + ) -> Result { + let _operation = self.inner.operation.lock().await; + validate_declared_length(&intent)?; + let stage_key = self.inner.next_stage.fetch_add(1, Ordering::Relaxed); + let path = self.stage_path(stage_key); + operations::write_verified_stage(&self.inner.root, &path, &intent, bytes).await?; + sync_directory( + &self.inner.root, + &self.staging_directory(), + ArtifactStoreOperation::Stage, + )?; + Ok(self.register_stage(stage_key, path, intent.artifact_ref())) + } + + async fn publish(&self, staged: &StagedArtifact) -> Result { + let _operation = self.inner.operation.lock().await; + let stage = self.checked_stage(staged)?; + match stage.status { + LocalStageStatus::Discarded => Err(invalid_stage()), + LocalStageStatus::Published => self.load_published_ref(&stage.artifact_ref).await, + LocalStageStatus::Pending => self.publish_pending(staged.stage_key(), stage).await, + } + } + + async fn inspect( + &self, + artifact_id: &ArtifactId, + ) -> Result, ArtifactStoreFailure> { + let _operation = self.inner.operation.lock().await; + Ok(self + .load_committed(artifact_id) + .await? + .map(|(artifact_ref, _)| artifact_ref)) + } + + async fn open( + &self, + artifact_id: &ArtifactId, + ) -> Result { + let _operation = self.inner.operation.lock().await; + let (_, bytes) = self + .load_committed(artifact_id) + .await? + .ok_or_else(missing_content)?; + Ok(Box::new(Cursor::new(bytes))) + } + + async fn discard( + &self, + staged: &StagedArtifact, + ) -> Result { + let _operation = self.inner.operation.lock().await; + let stage = self.checked_stage(staged)?; + match stage.status { + LocalStageStatus::Pending => self.discard_pending(staged.stage_key(), &stage).await, + LocalStageStatus::Discarded => Ok(DiscardResult::AlreadyDiscarded), + LocalStageStatus::Published => Ok(DiscardResult::AlreadyPublished), + } + } + + async fn release( + &self, + artifact_id: &ArtifactId, + ) -> Result { + let _operation = self.inner.operation.lock().await; + self.release_committed(artifact_id).await + } +} + +fn prepare_store_directories(root: &Path) -> Result<(), ArtifactStoreFailure> { + for directory in [ + root.join("staging"), + root.join("blobs"), + root.join("blobs/sha256"), + root.join("refs"), + ] { + prepare_owned_directory(root, &directory, ArtifactStoreOperation::Configuration)?; + sync_directory( + root, + directory + .parent() + .expect("store directory always has an owned parent"), + ArtifactStoreOperation::Configuration, + )?; + } + Ok(()) +} + +fn validate_declared_length(intent: &ArtifactIntent) -> Result<(), ArtifactStoreFailure> { + if intent.expected_byte_length.get() > MAX_ARTIFACT_BYTES { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::Oversize, + )); + } + Ok(()) +} + +fn invalid_stage() -> ArtifactStoreFailure { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::InvalidStage) +} + +fn missing_content() -> ArtifactStoreFailure { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::MissingCommittedContent) +} diff --git a/zeroshot-rust/src/artifact_store/local_cas/filesystem.rs b/zeroshot-rust/src/artifact_store/local_cas/filesystem.rs new file mode 100644 index 00000000..b5a5ea40 --- /dev/null +++ b/zeroshot-rust/src/artifact_store/local_cas/filesystem.rs @@ -0,0 +1,442 @@ +use std::path::Path; + +use fs2::FileExt; +use openengine_cluster_protocol::ArtifactId; +use tokio::io::AsyncReadExt; + +use super::super::{ + ArtifactStoreFailure, ArtifactStoreFailureKind, ArtifactStoreOperation, failure_from_io, +}; + +#[derive(Clone, Copy)] +pub(super) struct BoundedReadOptions { + operation: ArtifactStoreOperation, + missing_is_integrity_failure: bool, +} + +impl BoundedReadOptions { + pub(super) const fn optional(operation: ArtifactStoreOperation) -> Self { + Self { + operation, + missing_is_integrity_failure: false, + } + } + + pub(super) const fn committed(operation: ArtifactStoreOperation) -> Self { + Self { + operation, + missing_is_integrity_failure: true, + } + } +} + +pub(super) fn prepare_root(root: &Path) -> Result<(), ArtifactStoreFailure> { + match std::fs::symlink_metadata(root) { + Ok(metadata) => validate_directory_metadata(&metadata)?, + Err(error) => prepare_missing_root(root, error)?, + } + set_owner_directory_permissions(root) +} + +fn prepare_missing_root(root: &Path, error: std::io::Error) -> Result<(), ArtifactStoreFailure> { + match error.kind() { + std::io::ErrorKind::NotFound => { + create_owner_directory(root, ArtifactStoreOperation::Configuration) + } + std::io::ErrorKind::PermissionDenied => Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::PermissionDenied(ArtifactStoreOperation::Configuration), + )), + _ => Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::RootUnavailable, + )), + } +} + +pub(super) fn prepare_owned_directory( + root: &Path, + path: &Path, + operation: ArtifactStoreOperation, +) -> Result<(), ArtifactStoreFailure> { + validate_parent_directory(root, path)?; + match std::fs::symlink_metadata(path) { + Ok(metadata) => { + validate_directory_metadata(&metadata)?; + set_owner_directory_permissions(path)?; + Ok(()) + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + create_owner_directory(path, operation)?; + Ok(()) + } + Err(error) => Err(failure_from_io(error, operation)), + } +} + +fn validate_parent_directory(root: &Path, path: &Path) -> Result<(), ArtifactStoreFailure> { + let parent = path.parent().ok_or_else(corrupt_content)?; + validate_directory_path(root, parent) +} + +pub(super) fn validate_directory_path( + root: &Path, + directory: &Path, +) -> Result<(), ArtifactStoreFailure> { + let relative = directory + .strip_prefix(root) + .map_err(|_| corrupt_content())?; + let root_metadata = std::fs::symlink_metadata(root).map_err(configuration_io)?; + validate_directory_metadata(&root_metadata)?; + + let mut current = root.to_path_buf(); + for component in relative.components() { + let std::path::Component::Normal(component) = component else { + return Err(corrupt_content()); + }; + current.push(component); + let metadata = std::fs::symlink_metadata(¤t).map_err(configuration_io)?; + validate_directory_metadata(&metadata)?; + } + Ok(()) +} + +fn validate_directory_metadata(metadata: &std::fs::Metadata) -> Result<(), ArtifactStoreFailure> { + if metadata.file_type().is_symlink() || !metadata.is_dir() { + return Err(corrupt_content()); + } + Ok(()) +} + +fn create_owner_directory( + path: &Path, + operation: ArtifactStoreOperation, +) -> Result<(), ArtifactStoreFailure> { + let mut builder = std::fs::DirBuilder::new(); + #[cfg(unix)] + { + use std::os::unix::fs::DirBuilderExt; + builder.mode(0o700); + } + builder + .create(path) + .map_err(|error| failure_from_io(error, operation))?; + set_owner_directory_permissions(path) +} + +pub(super) fn acquire_root_lock( + root: &Path, + path: &Path, +) -> Result { + validate_parent_directory(root, path)?; + let file = open_owner_lock_file(path)?; + file.try_lock_exclusive().map_err(lock_failure)?; + Ok(file) +} + +fn lock_failure(error: std::io::Error) -> ArtifactStoreFailure { + if error.kind() == std::io::ErrorKind::PermissionDenied { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::PermissionDenied( + ArtifactStoreOperation::Configuration, + )) + } else { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::LockUnavailable) + } +} + +fn open_owner_lock_file(path: &Path) -> Result { + let mut options = std::fs::OpenOptions::new(); + options.create(true).read(true).write(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let file = options.open(path).map_err(configuration_io)?; + set_owner_file_permissions(path)?; + Ok(file) +} + +pub(super) async fn open_new_owner_file( + root: &Path, + path: &Path, + operation: ArtifactStoreOperation, +) -> Result { + validate_parent_directory_for_operation(root, path, operation)?; + let mut options = tokio::fs::OpenOptions::new(); + options.create_new(true).read(true).write(true); + #[cfg(unix)] + options.mode(0o600); + let file = options + .open(path) + .await + .map_err(|error| failure_from_io(error, operation))?; + set_owner_file_permissions(path)?; + Ok(file) +} + +pub(super) fn cleanup_abandoned_stages( + root: &Path, + directory: &Path, +) -> Result<(), ArtifactStoreFailure> { + validate_directory_path(root, directory)?; + let mut removed = false; + for entry in std::fs::read_dir(directory).map_err(configuration_io)? { + let entry = entry.map_err(configuration_io)?; + let metadata = std::fs::symlink_metadata(entry.path()).map_err(configuration_io)?; + validate_regular_metadata(&metadata, u64::MAX)?; + std::fs::remove_file(entry.path()).map_err(configuration_io)?; + removed = true; + } + if removed { + sync_directory(root, directory, ArtifactStoreOperation::Configuration)?; + } + Ok(()) +} + +pub(super) fn reject_symlink_or_non_file_if_present( + root: &Path, + path: &Path, +) -> Result<(), ArtifactStoreFailure> { + validate_parent_directory(root, path)?; + match std::fs::symlink_metadata(path) { + Ok(metadata) => validate_regular_metadata(&metadata, u64::MAX), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(configuration_io(error)), + } +} + +pub(super) async fn read_regular_bounded( + root: &Path, + path: &Path, + limit: u64, + options: BoundedReadOptions, +) -> Result>, ArtifactStoreFailure> { + validate_read_parent_directory( + root, + path, + options.operation, + options.missing_is_integrity_failure, + )?; + let Some(metadata) = regular_metadata( + path, + options.operation, + options.missing_is_integrity_failure, + ) + .await? + else { + return Ok(None); + }; + validate_regular_metadata(&metadata, limit)?; + read_bounded_file(path, metadata.len(), limit, options.operation) + .await + .map(Some) +} + +fn validate_read_parent_directory( + root: &Path, + path: &Path, + operation: ArtifactStoreOperation, + missing_is_integrity_failure: bool, +) -> Result<(), ArtifactStoreFailure> { + let parent = path.parent().ok_or_else(corrupt_content)?; + let relative = parent.strip_prefix(root).map_err(|_| corrupt_content())?; + validate_read_directory_component(root, operation, missing_is_integrity_failure)?; + + let mut current = root.to_path_buf(); + for component in relative.components() { + let std::path::Component::Normal(component) = component else { + return Err(corrupt_content()); + }; + current.push(component); + validate_read_directory_component(¤t, operation, missing_is_integrity_failure)?; + } + Ok(()) +} + +fn validate_read_directory_component( + path: &Path, + operation: ArtifactStoreOperation, + missing_is_integrity_failure: bool, +) -> Result<(), ArtifactStoreFailure> { + let metadata = match std::fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) + if error.kind() == std::io::ErrorKind::NotFound && missing_is_integrity_failure => + { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::MissingCommittedContent, + )); + } + Err(error) => return Err(failure_from_io(error, operation)), + }; + validate_directory_metadata(&metadata) +} + +async fn regular_metadata( + path: &Path, + operation: ArtifactStoreOperation, + missing_is_integrity_failure: bool, +) -> Result, ArtifactStoreFailure> { + match tokio::fs::symlink_metadata(path).await { + Ok(metadata) => Ok(Some(metadata)), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + missing_metadata(missing_is_integrity_failure) + } + Err(error) => Err(failure_from_io(error, operation)), + } +} + +fn missing_metadata( + missing_is_integrity_failure: bool, +) -> Result, ArtifactStoreFailure> { + if missing_is_integrity_failure { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::MissingCommittedContent, + )); + } + Ok(None) +} + +fn validate_regular_metadata( + metadata: &std::fs::Metadata, + limit: u64, +) -> Result<(), ArtifactStoreFailure> { + if metadata.file_type().is_symlink() || !metadata.is_file() || metadata.len() > limit { + return Err(corrupt_content()); + } + Ok(()) +} + +async fn read_bounded_file( + path: &Path, + capacity: u64, + limit: u64, + operation: ArtifactStoreOperation, +) -> Result, ArtifactStoreFailure> { + let file = tokio::fs::OpenOptions::new() + .read(true) + .open(path) + .await + .map_err(|error| failure_from_io(error, operation))?; + let mut bytes = Vec::with_capacity( + usize::try_from(capacity).expect("artifact and manifest limits fit usize"), + ); + file.take(limit + 1) + .read_to_end(&mut bytes) + .await + .map_err(|error| failure_from_io(error, operation))?; + if bytes.len() as u64 > limit { + return Err(corrupt_content()); + } + Ok(bytes) +} + +pub(super) async fn remove_regular_if_present( + root: &Path, + path: &Path, + operation: ArtifactStoreOperation, +) -> Result { + validate_parent_directory_for_operation(root, path, operation)?; + match tokio::fs::symlink_metadata(path).await { + Ok(metadata) => validate_regular_metadata(&metadata, u64::MAX)?, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(failure_from_io(error, operation)), + } + tokio::fs::remove_file(path) + .await + .map_err(|error| failure_from_io(error, operation))?; + Ok(true) +} + +fn validate_parent_directory_for_operation( + root: &Path, + path: &Path, + operation: ArtifactStoreOperation, +) -> Result<(), ArtifactStoreFailure> { + let parent = path.parent().ok_or_else(corrupt_content)?; + let relative = parent.strip_prefix(root).map_err(|_| corrupt_content())?; + validate_operation_directory_component(root, operation)?; + + let mut current = root.to_path_buf(); + for component in relative.components() { + let std::path::Component::Normal(component) = component else { + return Err(corrupt_content()); + }; + current.push(component); + validate_operation_directory_component(¤t, operation)?; + } + Ok(()) +} + +fn validate_operation_directory_component( + path: &Path, + operation: ArtifactStoreOperation, +) -> Result<(), ArtifactStoreFailure> { + let metadata = + std::fs::symlink_metadata(path).map_err(|error| failure_from_io(error, operation))?; + if metadata.file_type().is_symlink() { + return Err(corrupt_content()); + } + if !metadata.is_dir() { + return Err(ArtifactStoreFailure::new(ArtifactStoreFailureKind::Io( + operation, + ))); + } + Ok(()) +} + +pub(super) fn sync_directory( + root: &Path, + directory: &Path, + operation: ArtifactStoreOperation, +) -> Result<(), ArtifactStoreFailure> { + validate_directory_path(root, directory)?; + let file = std::fs::File::open(directory).map_err(|error| failure_from_io(error, operation))?; + file.sync_all() + .map_err(|error| failure_from_io(error, operation)) +} + +pub(super) fn set_owner_file_permissions(path: &Path) -> Result<(), ArtifactStoreFailure> { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)) + .map_err(configuration_io)?; + } + Ok(()) +} + +fn set_owner_directory_permissions(path: &Path) -> Result<(), ArtifactStoreFailure> { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)) + .map_err(configuration_io)?; + } + Ok(()) +} + +pub(super) fn validate_artifact_id(artifact_id: &ArtifactId) -> Result<(), ArtifactStoreFailure> { + let Some(digest) = artifact_id.as_str().strip_prefix("cas-v1-") else { + return Err(identity_conflict()); + }; + if digest.len() != 64 || !digest.bytes().all(is_lower_hex) { + return Err(identity_conflict()); + } + Ok(()) +} + +fn is_lower_hex(byte: u8) -> bool { + byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte) +} + +fn configuration_io(error: std::io::Error) -> ArtifactStoreFailure { + failure_from_io(error, ArtifactStoreOperation::Configuration) +} + +fn corrupt_content() -> ArtifactStoreFailure { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::CorruptContent) +} + +fn identity_conflict() -> ArtifactStoreFailure { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::IdentityConflict) +} diff --git a/zeroshot-rust/src/artifact_store/local_cas/operations.rs b/zeroshot-rust/src/artifact_store/local_cas/operations.rs new file mode 100644 index 00000000..4071c7be --- /dev/null +++ b/zeroshot-rust/src/artifact_store/local_cas/operations.rs @@ -0,0 +1,488 @@ +use std::path::Path; + +use openengine_cluster_protocol::{ArtifactId, ArtifactRef}; +use sha2::{Digest, Sha256}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use super::filesystem::{ + BoundedReadOptions, open_new_owner_file, prepare_owned_directory, read_regular_bounded, + remove_regular_if_present, set_owner_file_permissions, sync_directory, +}; +use super::{LocalCasArtifactStore, LocalStage, LocalStageStatus, MAX_MANIFEST_BYTES, missing_content}; +use crate::artifact_store::{ + ArtifactByteStream, ArtifactIntent, ArtifactStoreFailure, ArtifactStoreFailureKind, + ArtifactStoreOperation, DiscardResult, MAX_ARTIFACT_BYTES, ReleaseResult, derive_artifact_id, + failure_from_io, verify_bytes, +}; + +impl LocalCasArtifactStore { + pub(super) async fn load_committed( + &self, + artifact_id: &ArtifactId, + ) -> Result)>, ArtifactStoreFailure> { + self.load_committed_for(artifact_id, ArtifactStoreOperation::Inspect) + .await + } + + async fn load_committed_for( + &self, + artifact_id: &ArtifactId, + operation: ArtifactStoreOperation, + ) -> Result)>, ArtifactStoreFailure> { + let Some(artifact_ref) = self.read_manifest(artifact_id, operation).await? else { + return Ok(None); + }; + let blob_path = self.blob_path(&artifact_ref); + let bytes = read_regular_bounded( + &self.inner.root, + &blob_path, + artifact_ref.byte_length.get() + 1, + BoundedReadOptions::committed(operation), + ) + .await? + .ok_or_else(missing_content)?; + verify_bytes(&artifact_ref, &bytes)?; + Ok(Some((artifact_ref, bytes))) + } + + async fn read_manifest( + &self, + artifact_id: &ArtifactId, + operation: ArtifactStoreOperation, + ) -> Result, ArtifactStoreFailure> { + let path = self.ref_path(artifact_id)?; + let Some(manifest) = read_regular_bounded( + &self.inner.root, + &path, + MAX_MANIFEST_BYTES, + BoundedReadOptions::optional(operation), + ) + .await? + else { + return Ok(None); + }; + let artifact_ref = decode_manifest(&manifest)?; + validate_manifest_identity(&artifact_ref, artifact_id)?; + Ok(Some(artifact_ref)) + } + + pub(super) async fn publish_pending( + &self, + stage_key: u64, + stage: LocalStage, + ) -> Result { + self.ensure_blob(&stage).await?; + self.commit_manifest(stage_key, &stage.artifact_ref).await?; + self.stages() + .get_mut(&stage_key) + .expect("stage remains registered during serialized publish") + .status = LocalStageStatus::Published; + Ok(stage.artifact_ref) + } + + pub(super) async fn load_published_ref( + &self, + expected: &ArtifactRef, + ) -> Result { + let (committed, _) = self + .load_committed_for(&expected.artifact_id, ArtifactStoreOperation::Publish) + .await? + .ok_or_else(missing_content)?; + require_matching_ref(&committed, expected)?; + Ok(committed) + } + + async fn ensure_blob(&self, stage: &LocalStage) -> Result<(), ArtifactStoreFailure> { + let blob_path = self.blob_path(&stage.artifact_ref); + let blob_directory = blob_path + .parent() + .expect("content-addressed blob always has a parent"); + prepare_owned_directory( + &self.inner.root, + blob_directory, + ArtifactStoreOperation::Publish, + )?; + sync_directory( + &self.inner.root, + blob_directory + .parent() + .expect("blob prefix always has an owned parent"), + ArtifactStoreOperation::Publish, + )?; + let staged = self.read_staged_bytes(stage).await?; + let existing = read_regular_bounded( + &self.inner.root, + &blob_path, + stage.artifact_ref.byte_length.get() + 1, + BoundedReadOptions::optional(ArtifactStoreOperation::Publish), + ) + .await?; + match existing { + Some(existing) => self.reuse_blob(stage, staged, existing).await, + None => self.publish_new_blob(stage, staged.is_some(), &blob_path), + }?; + sync_directory( + &self.inner.root, + blob_directory, + ArtifactStoreOperation::Publish, + ) + } + + async fn read_staged_bytes( + &self, + stage: &LocalStage, + ) -> Result>, ArtifactStoreFailure> { + let staged = read_regular_bounded( + &self.inner.root, + &stage.path, + stage.artifact_ref.byte_length.get() + 1, + BoundedReadOptions::optional(ArtifactStoreOperation::Publish), + ) + .await?; + if let Some(bytes) = staged.as_ref() { + verify_bytes(&stage.artifact_ref, bytes)?; + } + Ok(staged) + } + + async fn reuse_blob( + &self, + stage: &LocalStage, + staged: Option>, + existing: Vec, + ) -> Result<(), ArtifactStoreFailure> { + verify_bytes(&stage.artifact_ref, &existing)?; + let Some(staged) = staged else { + return Ok(()); + }; + if staged != existing { + return Err(identity_conflict()); + } + remove_regular_if_present( + &self.inner.root, + &stage.path, + ArtifactStoreOperation::Publish, + ) + .await?; + sync_directory( + &self.inner.root, + &self.staging_directory(), + ArtifactStoreOperation::Publish, + ) + } + + fn publish_new_blob( + &self, + stage: &LocalStage, + stage_exists: bool, + blob_path: &Path, + ) -> Result<(), ArtifactStoreFailure> { + if !stage_exists { + return Err(missing_content()); + } + std::fs::rename(&stage.path, blob_path).map_err(publish_io)?; + set_owner_file_permissions(blob_path)?; + Ok(()) + } + + async fn commit_manifest( + &self, + stage_key: u64, + artifact_ref: &ArtifactRef, + ) -> Result<(), ArtifactStoreFailure> { + let existing = self + .load_committed_for(&artifact_ref.artifact_id, ArtifactStoreOperation::Publish) + .await?; + if let Some((existing, _)) = existing { + return require_matching_ref(&existing, artifact_ref); + } + self.write_new_manifest(stage_key, artifact_ref).await + } + + async fn write_new_manifest( + &self, + stage_key: u64, + artifact_ref: &ArtifactRef, + ) -> Result<(), ArtifactStoreFailure> { + let destination = self.ref_path(&artifact_ref.artifact_id)?; + let temporary = self.manifest_stage_path(stage_key, &artifact_ref.artifact_id); + remove_regular_if_present( + &self.inner.root, + &temporary, + ArtifactStoreOperation::Publish, + ) + .await?; + let encoded = serde_json::to_vec(artifact_ref).map_err(manifest_encoding_failure)?; + write_synced_file( + &self.inner.root, + &temporary, + &encoded, + ArtifactStoreOperation::Publish, + ) + .await?; + tokio::fs::rename(&temporary, &destination) + .await + .map_err(publish_io)?; + sync_directory( + &self.inner.root, + &self.refs_directory(), + ArtifactStoreOperation::Publish, + )?; + sync_directory( + &self.inner.root, + &self.staging_directory(), + ArtifactStoreOperation::Publish, + ) + } + + pub(super) async fn discard_pending( + &self, + stage_key: u64, + stage: &LocalStage, + ) -> Result { + remove_regular_if_present( + &self.inner.root, + &stage.path, + ArtifactStoreOperation::Discard, + ) + .await?; + sync_directory( + &self.inner.root, + &self.staging_directory(), + ArtifactStoreOperation::Discard, + )?; + self.stages() + .get_mut(&stage_key) + .expect("stage remains registered during serialized discard") + .status = LocalStageStatus::Discarded; + Ok(DiscardResult::Discarded) + } + + pub(super) async fn release_committed( + &self, + artifact_id: &ArtifactId, + ) -> Result { + let target = self + .load_committed_for(artifact_id, ArtifactStoreOperation::Release) + .await?; + let Some((target, _)) = target else { + return Ok(ReleaseResult::NotFound); + }; + let shared_count = self.count_digest_refs(&target).await?; + self.remove_ref(artifact_id).await?; + if shared_count == 1 { + self.remove_blob(&target).await?; + } + Ok(ReleaseResult::Released) + } + + async fn count_digest_refs(&self, target: &ArtifactRef) -> Result { + let mut count = 0_usize; + let entries = std::fs::read_dir(self.refs_directory()).map_err(release_io)?; + for entry in entries { + let id = artifact_id_from_entry(entry.map_err(release_io)?)?; + let (candidate, _) = self + .load_committed_for(&id, ArtifactStoreOperation::Release) + .await? + .ok_or_else(corrupt_content)?; + if candidate.sha256 == target.sha256 { + count += 1; + } + } + Ok(count) + } + + async fn remove_ref(&self, artifact_id: &ArtifactId) -> Result<(), ArtifactStoreFailure> { + let path = self.ref_path(artifact_id)?; + remove_regular_if_present(&self.inner.root, &path, ArtifactStoreOperation::Release).await?; + sync_directory( + &self.inner.root, + &self.refs_directory(), + ArtifactStoreOperation::Release, + ) + } + + async fn remove_blob(&self, artifact_ref: &ArtifactRef) -> Result<(), ArtifactStoreFailure> { + let path = self.blob_path(artifact_ref); + remove_regular_if_present(&self.inner.root, &path, ArtifactStoreOperation::Release).await?; + sync_directory( + &self.inner.root, + path.parent() + .expect("content-addressed blob always has a parent"), + ArtifactStoreOperation::Release, + ) + } +} + +pub(super) async fn write_verified_stage( + root: &Path, + path: &Path, + intent: &ArtifactIntent, + bytes: ArtifactByteStream, +) -> Result<(), ArtifactStoreFailure> { + let mut file = open_new_owner_file(root, path, ArtifactStoreOperation::Stage).await?; + let result = copy_and_verify(&mut file, intent, bytes).await; + drop(file); + if let Err(failure) = result { + remove_regular_if_present(root, path, ArtifactStoreOperation::Stage).await?; + sync_directory( + root, + path.parent() + .expect("artifact stage always has a parent directory"), + ArtifactStoreOperation::Stage, + )?; + return Err(failure); + } + Ok(()) +} + +async fn copy_and_verify( + file: &mut tokio::fs::File, + intent: &ArtifactIntent, + mut bytes: ArtifactByteStream, +) -> Result<(), ArtifactStoreFailure> { + let declared = intent.expected_byte_length.get(); + let mut hasher = Sha256::new(); + let mut total = 0_u64; + let mut buffer = [0_u8; 16 * 1024]; + loop { + let remaining = declared.saturating_add(1).saturating_sub(total); + if remaining == 0 { + break; + } + let limit = usize::try_from(remaining.min(buffer.len() as u64)) + .expect("bounded read length fits usize"); + let read = bytes.read(&mut buffer[..limit]).await.map_err(stage_io)?; + if read == 0 { + break; + } + total += read as u64; + if total > declared { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::Oversize, + )); + } + hasher.update(&buffer[..read]); + file.write_all(&buffer[..read]).await.map_err(stage_io)?; + } + verify_staged_digest(intent, total, hasher.finalize())?; + file.flush().await.map_err(stage_io)?; + file.sync_all().await.map_err(stage_io) +} + +fn verify_staged_digest( + intent: &ArtifactIntent, + total: u64, + digest: impl AsRef<[u8]>, +) -> Result<(), ArtifactStoreFailure> { + if total != intent.expected_byte_length.get() { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::LengthMismatch, + )); + } + if digest_to_hex(digest) != intent.expected_sha256.as_str() { + return Err(ArtifactStoreFailure::new( + ArtifactStoreFailureKind::HashMismatch, + )); + } + Ok(()) +} + +async fn write_synced_file( + root: &Path, + path: &Path, + bytes: &[u8], + operation: ArtifactStoreOperation, +) -> Result<(), ArtifactStoreFailure> { + let mut file = open_new_owner_file(root, path, operation).await?; + file.write_all(bytes).await.map_err(publish_io)?; + file.flush().await.map_err(publish_io)?; + file.sync_all().await.map_err(publish_io) +} + +fn decode_manifest(bytes: &[u8]) -> Result { + serde_json::from_slice(bytes).map_err(|_| corrupt_content()) +} + +fn validate_manifest_identity( + artifact_ref: &ArtifactRef, + expected_id: &ArtifactId, +) -> Result<(), ArtifactStoreFailure> { + if artifact_ref.artifact_id != *expected_id + || derive_artifact_id(&intent_from_ref(artifact_ref)) != *expected_id + { + return Err(identity_conflict()); + } + if artifact_ref.byte_length.get() > MAX_ARTIFACT_BYTES { + return Err(corrupt_content()); + } + Ok(()) +} + +fn intent_from_ref(artifact_ref: &ArtifactRef) -> ArtifactIntent { + ArtifactIntent { + expected_sha256: artifact_ref.sha256.clone(), + expected_byte_length: artifact_ref.byte_length, + media_type: artifact_ref.media_type.clone(), + type_id: artifact_ref.type_id.clone(), + producer: artifact_ref.producer.clone(), + lineage: artifact_ref.lineage.clone(), + redaction: artifact_ref.redaction, + } +} + +fn require_matching_ref( + existing: &ArtifactRef, + expected: &ArtifactRef, +) -> Result<(), ArtifactStoreFailure> { + if existing != expected { + return Err(identity_conflict()); + } + Ok(()) +} + +fn artifact_id_from_entry(entry: std::fs::DirEntry) -> Result { + let file_name = entry.file_name(); + let file_name = file_name.to_str().ok_or_else(corrupt_content)?; + let id = file_name + .strip_suffix(".json") + .ok_or_else(corrupt_content)?; + ArtifactId::new(id.to_owned()).map_err(|_| corrupt_content()) +} + +fn digest_to_hex(digest: impl AsRef<[u8]>) -> String { + let bytes = digest.as_ref(); + let mut output = String::with_capacity(bytes.len() * 2); + const HEX: &[u8; 16] = b"0123456789abcdef"; + for &byte in bytes { + output.push(char::from(HEX[usize::from(byte >> 4)])); + output.push(char::from(HEX[usize::from(byte & 0x0f)])); + } + output +} + +fn stage_io(error: std::io::Error) -> ArtifactStoreFailure { + failure_from_io(error, ArtifactStoreOperation::Stage) +} + +fn publish_io(error: std::io::Error) -> ArtifactStoreFailure { + failure_from_io(error, ArtifactStoreOperation::Publish) +} + +fn release_io(error: std::io::Error) -> ArtifactStoreFailure { + failure_from_io(error, ArtifactStoreOperation::Release) +} + +fn manifest_encoding_failure(_: serde_json::Error) -> ArtifactStoreFailure { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::Io( + ArtifactStoreOperation::Publish, + )) +} + +fn corrupt_content() -> ArtifactStoreFailure { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::CorruptContent) +} + +fn identity_conflict() -> ArtifactStoreFailure { + ArtifactStoreFailure::new(ArtifactStoreFailureKind::IdentityConflict) +} diff --git a/zeroshot-rust/src/lib.rs b/zeroshot-rust/src/lib.rs index 05d12285..b295590c 100644 --- a/zeroshot-rust/src/lib.rs +++ b/zeroshot-rust/src/lib.rs @@ -1,3 +1,4 @@ +pub mod artifact_store; pub mod issue_provider; mod provider_value; pub mod source_code_provider; diff --git a/zeroshot-rust/tests/architecture.rs b/zeroshot-rust/tests/architecture.rs index ff39ab02..5409caa0 100644 --- a/zeroshot-rust/tests/architecture.rs +++ b/zeroshot-rust/tests/architecture.rs @@ -104,6 +104,11 @@ fn product_uses_the_root_workspace_and_a_rust_only_layout() { relative_files(&product, &product, &mut files); for required in [ "Cargo.toml", + "src/artifact_store.rs", + "src/artifact_store/fake.rs", + "src/artifact_store/local_cas.rs", + "src/artifact_store/local_cas/filesystem.rs", + "src/artifact_store/local_cas/operations.rs", "src/fault.rs", "src/fault/redaction.rs", "src/fault/taxonomy.rs", @@ -114,10 +119,13 @@ fn product_uses_the_root_workspace_and_a_rust_only_layout() { "src/issue_provider.rs", "src/source_code_provider.rs", "tests/architecture.rs", + "tests/artifact_store.rs", "tests/backend_boundary.rs", "tests/fault_contract.rs", + "tests/local_cas.rs", "tests/observability_contract.rs", "tests/provider_contracts.rs", + "tests/support/mod.rs", "tests/provider_bounds.rs", ] { assert!(files.contains(required), "missing product file: {required}"); @@ -152,8 +160,10 @@ fn workspace_metadata_preserves_package_lib_and_bin_identity() { targets, BTreeSet::from([ ("architecture".to_owned(), "test".to_owned()), + ("artifact_store".to_owned(), "test".to_owned()), ("backend_boundary".to_owned(), "test".to_owned()), ("fault_contract".to_owned(), "test".to_owned()), + ("local_cas".to_owned(), "test".to_owned()), ("observability_contract".to_owned(), "test".to_owned()), ("provider_bounds".to_owned(), "test".to_owned()), ("provider_contracts".to_owned(), "test".to_owned()), @@ -182,6 +192,7 @@ fn product_dependencies_stay_inside_native_contract_and_backend_boundaries() { .collect::>(); let allowed = BTreeSet::from([ ("async-trait".to_owned(), "normal".to_owned()), + ("fs2".to_owned(), "normal".to_owned()), ( "openengine-cluster-protocol".to_owned(), "normal".to_owned(), @@ -189,8 +200,9 @@ fn product_dependencies_stay_inside_native_contract_and_backend_boundaries() { ("openengine-cluster-server".to_owned(), "normal".to_owned()), ("serde".to_owned(), "normal".to_owned()), ("serde_json".to_owned(), "normal".to_owned()), + ("sha2".to_owned(), "normal".to_owned()), ("thiserror".to_owned(), "normal".to_owned()), - ("tokio".to_owned(), "dev".to_owned()), + ("tokio".to_owned(), "normal".to_owned()), ]); assert_eq!(dependencies, allowed); } @@ -278,7 +290,6 @@ fn runtime_has_no_future_product_concerns() { "scheduler", "worker", "workspace", - "artifact", ] { assert!( !words.contains(forbidden_word), @@ -287,6 +298,80 @@ fn runtime_has_no_future_product_concerns() { } } +#[test] +fn artifact_storage_stays_product_private_and_receipts_stay_byte_free() { + let product = product_root(); + let repository = repository_root(); + let artifact_contract = + read(&repository.join("crates/openengine-cluster-protocol/src/artifact.rs")); + for forbidden in [ + "Vec", + "AsyncRead", + "PathBuf", + "StagedArtifact", + "ArtifactStore", + "signed_url", + "download_url", + "storage_root", + "manifest_path", + ] { + assert!( + !artifact_contract.contains(forbidden), + "protocol artifact receipt exposed storage detail: {forbidden}" + ); + } + + for relative in [ + "protocol/openengine-cluster/v1/schema.json", + "protocol/openengine-cluster/v1/worker.schema.json", + "protocol/openengine-cluster/v1/fixtures/graph/positive/artifact-ref.json", + ] { + let projection = read(&repository.join(relative)); + for forbidden in [ + "localPath", + "signedUrl", + "downloadUrl", + "storageRoot", + "stagePath", + "manifestPath", + ] { + assert!( + !projection.contains(forbidden), + "generated artifact projection exposed storage detail: {relative}: {forbidden}" + ); + } + } + + let lib = read(&product.join("src/lib.rs")); + assert!( + lib.contains("pub struct NativeBackend;"), + "NativeBackend must remain uninjected until composition issue #693" + ); + assert!(!lib.contains("ArtifactStore>")); + assert!(!lib.contains("artifact_store:")); + + let lifecycle_and_backend = format!( + "{}\n{}\n{}", + read(&repository.join("crates/openengine-cluster-protocol/src/lifecycle.rs")), + read(&repository.join("crates/openengine-cluster-server/src/lifecycle.rs")), + read(&repository.join("crates/openengine-cluster-server/src/lib.rs")) + ); + for forbidden in [ + "StagedArtifact", + "ArtifactByteStream", + "LocalCasArtifactStore", + "manifest_path", + "storage_root", + "signed_url", + "download_url", + ] { + assert!( + !lifecycle_and_backend.contains(forbidden), + "lifecycle/backend parameter exposed artifact storage detail: {forbidden}" + ); + } +} + #[test] fn provider_contracts_add_no_ledger_workspace_worker_protocol_adapter_or_fault_behavior() { let product = product_root(); diff --git a/zeroshot-rust/tests/artifact_store.rs b/zeroshot-rust/tests/artifact_store.rs new file mode 100644 index 00000000..0d92637f --- /dev/null +++ b/zeroshot-rust/tests/artifact_store.rs @@ -0,0 +1,334 @@ +use std::sync::Arc; + +use openengine_cluster_protocol::{ByteLength, RunId}; +use tokio::io::AsyncReadExt; +use zeroshot_engine::artifact_store::fake::{FakeArtifactStore, FakeFailurePoint}; +use zeroshot_engine::artifact_store::{ + ArtifactStore, ArtifactStoreFailureKind, ArtifactStoreOperation, DiscardResult, + MAX_ARTIFACT_BYTES, ReleaseResult, derive_artifact_id, +}; +use zeroshot_engine::fault::{EvidenceClass, FaultContext, FaultModule}; + +#[path = "support/mod.rs"] +mod support; + +use support::{byte_stream as stream, test_intent as intent}; + +#[tokio::test] +async fn store_is_object_safe_and_accepts_the_exact_limit() { + let store: Arc = Arc::new(FakeArtifactStore::new()); + let bytes = vec![0x5a; MAX_ARTIFACT_BYTES as usize]; + let staged = store + .stage(intent(&bytes, "exact-limit"), stream(bytes)) + .await + .expect("the exact artifact limit must stage"); + let artifact_ref = store + .publish(&staged) + .await + .expect("the exact artifact limit must publish"); + assert_eq!(artifact_ref.byte_length.get(), MAX_ARTIFACT_BYTES); +} + +#[tokio::test] +async fn stage_rejects_declared_and_streamed_overflow_short_input_and_hash_mismatch() { + let store = FakeArtifactStore::new(); + let one = vec![1]; + let mut oversized = intent(&one, "declared-overflow"); + oversized.expected_byte_length = + ByteLength::new(MAX_ARTIFACT_BYTES + 1).expect("protocol length permits store overflow"); + assert_eq!( + store + .stage(oversized, stream(Vec::new())) + .await + .expect_err("oversized declaration must fail") + .kind(), + ArtifactStoreFailureKind::Oversize + ); + + let expected = vec![1, 2]; + assert_eq!( + store + .stage(intent(&expected, "stream-overflow"), stream(vec![1, 2, 3])) + .await + .expect_err("sentinel byte must detect streamed overflow") + .kind(), + ArtifactStoreFailureKind::Oversize + ); + assert_eq!( + store + .stage(intent(&expected, "short"), stream(vec![1])) + .await + .expect_err("short input must fail") + .kind(), + ArtifactStoreFailureKind::LengthMismatch + ); + assert_eq!( + store + .stage(intent(&expected, "hash"), stream(vec![2, 1])) + .await + .expect_err("digest mismatch must fail") + .kind(), + ArtifactStoreFailureKind::HashMismatch + ); +} + +#[test] +fn artifact_identity_has_a_golden_domain_separated_projection() { + let artifact_intent = intent(b"golden artifact", "run-golden"); + assert_eq!( + derive_artifact_id(&artifact_intent).as_str(), + "cas-v1-ffa3f25c49fda68dbd65543d55d83861ff6b28e8e8f647e57662af90b0ab653b" + ); + let mut other_lineage = artifact_intent.clone(); + other_lineage.lineage.run_id = RunId::new("run-other"); + assert_ne!( + derive_artifact_id(&artifact_intent), + derive_artifact_id(&other_lineage) + ); +} + +#[tokio::test] +async fn publish_is_idempotent_and_lineage_refs_share_content() { + let store = FakeArtifactStore::new(); + let bytes = b"shared bytes".to_vec(); + let first = store + .stage(intent(&bytes, "run-one"), stream(bytes.clone())) + .await + .expect("first stage succeeds"); + let first_ref = store.publish(&first).await.expect("first publish succeeds"); + assert_eq!( + store.publish(&first).await.expect("retry is idempotent"), + first_ref + ); + + let duplicate = store + .stage(intent(&bytes, "run-one"), stream(bytes.clone())) + .await + .expect("duplicate stage succeeds"); + assert_eq!( + store + .publish(&duplicate) + .await + .expect("duplicate intent publishes"), + first_ref + ); + + let second = store + .stage(intent(&bytes, "run-two"), stream(bytes)) + .await + .expect("second lineage stages"); + let second_ref = store + .publish(&second) + .await + .expect("second lineage publishes"); + assert_ne!(first_ref.artifact_id, second_ref.artifact_id); + assert_eq!(store.blob_count(), 1); + assert_eq!(store.committed_ref_count(), 2); + + assert_eq!( + store + .release(&first_ref.artifact_id) + .await + .expect("first release succeeds"), + ReleaseResult::Released + ); + assert_eq!(store.blob_count(), 1); + assert!( + store + .inspect(&second_ref.artifact_id) + .await + .expect("inspect succeeds") + .is_some() + ); + assert_eq!( + store + .release(&second_ref.artifact_id) + .await + .expect("last release succeeds"), + ReleaseResult::Released + ); + assert_eq!(store.blob_count(), 0); + assert_eq!( + store + .release(&second_ref.artifact_id) + .await + .expect("release retry succeeds"), + ReleaseResult::NotFound + ); +} + +#[tokio::test] +async fn response_loss_recovers_through_inspect_and_open_yields_verified_bytes() { + let store = FakeArtifactStore::new(); + let bytes = b"durable artifact".to_vec(); + let staged = store + .stage(intent(&bytes, "response-loss"), stream(bytes.clone())) + .await + .expect("stage succeeds"); + store.script_failure( + FakeFailurePoint::AfterPublishCommit, + ArtifactStoreFailureKind::Io(ArtifactStoreOperation::Publish), + ); + assert_eq!( + store + .publish(&staged) + .await + .expect_err("script loses publish response") + .kind(), + ArtifactStoreFailureKind::Io(ArtifactStoreOperation::Publish) + ); + let recovered = store + .inspect(staged.artifact_id()) + .await + .expect("inspect succeeds") + .expect("commit is authoritative"); + assert_eq!(&recovered.artifact_id, staged.artifact_id()); + let mut opened = store + .open(staged.artifact_id()) + .await + .expect("open succeeds"); + let mut actual = Vec::new(); + opened + .read_to_end(&mut actual) + .await + .expect("verified stream reads"); + assert_eq!(actual, bytes); +} + +#[tokio::test] +async fn fake_scripts_every_boundary_in_fifo_order() { + let failure = ArtifactStoreFailureKind::Io(ArtifactStoreOperation::Stage); + let stage_store = FakeArtifactStore::new(); + stage_store.script_failure(FakeFailurePoint::BeforeStageCommit, failure); + assert_eq!( + stage_store + .stage(intent(b"x", "stage-fail"), stream(b"x".to_vec())) + .await + .expect_err("stage boundary fails") + .kind(), + failure + ); + + let store = FakeArtifactStore::new(); + let staged = store + .stage(intent(b"x", "boundaries"), stream(b"x".to_vec())) + .await + .expect("stage succeeds"); + store.script_failure( + FakeFailurePoint::BeforePublishCommit, + ArtifactStoreFailureKind::Io(ArtifactStoreOperation::Publish), + ); + assert!(store.publish(&staged).await.is_err()); + let artifact_ref = store + .publish(&staged) + .await + .expect("publish retry succeeds"); + + for (point, operation) in [ + (FakeFailurePoint::Inspect, ArtifactStoreOperation::Inspect), + (FakeFailurePoint::Open, ArtifactStoreOperation::Open), + (FakeFailurePoint::Release, ArtifactStoreOperation::Release), + ] { + store.script_failure(point, ArtifactStoreFailureKind::Io(operation)); + let result = match point { + FakeFailurePoint::Inspect => store.inspect(&artifact_ref.artifact_id).await.map(|_| ()), + FakeFailurePoint::Open => store.open(&artifact_ref.artifact_id).await.map(|_| ()), + FakeFailurePoint::Release => store.release(&artifact_ref.artifact_id).await.map(|_| ()), + _ => unreachable!("only direct operation boundaries are listed"), + }; + assert_eq!( + result.expect_err("scripted boundary fails").kind(), + ArtifactStoreFailureKind::Io(operation) + ); + } + + let pending = store + .stage(intent(b"y", "discard"), stream(b"y".to_vec())) + .await + .expect("pending stage succeeds"); + store.script_failure( + FakeFailurePoint::Discard, + ArtifactStoreFailureKind::Io(ArtifactStoreOperation::Discard), + ); + assert!(store.discard(&pending).await.is_err()); + assert_eq!( + store + .discard(&pending) + .await + .expect("discard retry succeeds"), + DiscardResult::Discarded + ); + assert_eq!( + store + .discard(&pending) + .await + .expect("discard is idempotent"), + DiscardResult::AlreadyDiscarded + ); +} + +#[tokio::test] +async fn restart_discards_only_uncommitted_stages() { + let store = FakeArtifactStore::new(); + let committed = store + .stage( + intent(b"committed", "restart"), + stream(b"committed".to_vec()), + ) + .await + .expect("committed stage succeeds"); + let committed_ref = store.publish(&committed).await.expect("publish succeeds"); + let _pending = store + .stage(intent(b"pending", "restart"), stream(b"pending".to_vec())) + .await + .expect("pending stage succeeds"); + assert_eq!(store.staged_count(), 2); + store.restart(); + assert_eq!(store.staged_count(), 0); + assert!( + store + .inspect(&committed_ref.artifact_id) + .await + .expect("inspect succeeds after restart") + .is_some() + ); +} + +#[test] +fn failure_mapping_is_closed_and_contains_no_raw_text() { + let cases = [ + ( + ArtifactStoreFailureKind::Oversize, + FaultModule::Worker, + FaultContext::Settlement, + EvidenceClass::ResourceExhausted, + ), + ( + ArtifactStoreFailureKind::HashMismatch, + FaultModule::Worker, + FaultContext::Settlement, + EvidenceClass::MalformedExternalData, + ), + ( + ArtifactStoreFailureKind::LockUnavailable, + FaultModule::Storage, + FaultContext::Configuration, + EvidenceClass::Unavailable, + ), + ( + ArtifactStoreFailureKind::CorruptContent, + FaultModule::Storage, + FaultContext::Recovery, + EvidenceClass::IntegrityFailure, + ), + ]; + for (kind, module, context, class) in cases { + let failure = zeroshot_engine::artifact_store::ArtifactStoreFailure::new(kind); + let evidence = failure.module_evidence(); + assert_eq!(evidence.module(), module); + assert_eq!(evidence.context(), context); + assert_eq!(evidence.class(), class); + assert!(!failure.to_string().contains('/')); + assert!(!format!("{failure:?}").contains("path")); + } +} diff --git a/zeroshot-rust/tests/local_cas.rs b/zeroshot-rust/tests/local_cas.rs new file mode 100644 index 00000000..9245e2e5 --- /dev/null +++ b/zeroshot-rust/tests/local_cas.rs @@ -0,0 +1,453 @@ +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use openengine_cluster_protocol::{ArtifactRef, MediaType}; +use tokio::io::AsyncReadExt; +use zeroshot_engine::artifact_store::local_cas::LocalCasArtifactStore; +use zeroshot_engine::artifact_store::{ArtifactStore, ArtifactStoreFailureKind, ReleaseResult}; + +#[path = "local_cas/recovery.rs"] +mod recovery; +#[path = "support/mod.rs"] +mod support; + +use support::{byte_stream as stream, test_intent as intent}; + +static NEXT_ROOT: AtomicU64 = AtomicU64::new(1); + +struct TestRoot(PathBuf); + +impl TestRoot { + fn new(label: &str) -> Self { + let sequence = NEXT_ROOT.fetch_add(1, Ordering::Relaxed); + Self(std::env::temp_dir().join(format!( + "zeroshot-local-cas-{label}-{}-{sequence}", + std::process::id() + ))) + } + + fn path(&self) -> &Path { + &self.0 + } +} + +impl Drop for TestRoot { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn blob_path(root: &Path, artifact_ref: &ArtifactRef) -> PathBuf { + let digest = artifact_ref.sha256.as_str(); + root.join("blobs/sha256").join(&digest[..2]).join(digest) +} + +fn ref_path(root: &Path, artifact_ref: &ArtifactRef) -> PathBuf { + root.join("refs") + .join(format!("{}.json", artifact_ref.artifact_id.as_str())) +} + +#[tokio::test] +async fn one_writer_lock_is_shared_only_by_clones() { + let root = TestRoot::new("lock"); + let store = LocalCasArtifactStore::new(root.path()).expect("first writer locks root"); + let clone = store.clone(); + let failure = match LocalCasArtifactStore::new(root.path()) { + Ok(_) => panic!("independent writer must fail"), + Err(failure) => failure, + }; + assert_eq!(failure.kind(), ArtifactStoreFailureKind::LockUnavailable); + + let staged = clone + .stage(intent(b"clone", "clone"), stream(b"clone".to_vec())) + .await + .expect("clone shares writer"); + clone.publish(&staged).await.expect("clone publishes"); + drop(store); + let failure = match LocalCasArtifactStore::new(root.path()) { + Ok(_) => panic!("clone must retain the lock"), + Err(failure) => failure, + }; + assert_eq!(failure.kind(), ArtifactStoreFailureKind::LockUnavailable); + drop(clone); + LocalCasArtifactStore::new(root.path()).expect("lock releases with final clone"); +} + +#[test] +fn startup_removes_only_regular_abandoned_stages() { + let root = TestRoot::new("cleanup"); + std::fs::create_dir_all(root.path().join("staging")).expect("create staging directory"); + std::fs::write(root.path().join("staging/abandoned.tmp"), b"partial") + .expect("write abandoned stage"); + let _store = LocalCasArtifactStore::new(root.path()).expect("startup cleanup succeeds"); + assert_eq!( + std::fs::read_dir(root.path().join("staging")) + .expect("read staging directory") + .count(), + 0 + ); +} + +#[tokio::test] +async fn publish_is_synchronized_atomic_and_independent_of_source_directory() { + let root = TestRoot::new("atomic"); + let source = root.path().with_extension("source"); + std::fs::create_dir(&source).expect("create source directory"); + let source_file = source.join("artifact.bin"); + let bytes = b"workspace-independent artifact".to_vec(); + std::fs::write(&source_file, &bytes).expect("write source artifact"); + let input = tokio::fs::File::open(&source_file) + .await + .expect("open source artifact"); + + let store = LocalCasArtifactStore::new(root.path()).expect("construct local CAS"); + let staged = store + .stage(intent(&bytes, "source-removal"), Box::new(input)) + .await + .expect("stage source artifact"); + std::fs::remove_dir_all(&source).expect("remove producing workspace"); + let artifact_ref = store.publish(&staged).await.expect("publish staged bytes"); + assert_eq!( + store + .publish(&staged) + .await + .expect("publish retry is idempotent"), + artifact_ref + ); + + let mut opened = store + .open(&artifact_ref.artifact_id) + .await + .expect("open committed artifact"); + let mut actual = Vec::new(); + opened + .read_to_end(&mut actual) + .await + .expect("read verified stream"); + assert_eq!(actual, bytes); + assert!(blob_path(root.path(), &artifact_ref).is_file()); + assert!(ref_path(root.path(), &artifact_ref).is_file()); + assert_eq!( + std::fs::read_dir(root.path().join("staging")) + .expect("read staging") + .count(), + 0 + ); + assert!( + std::fs::read_dir(root.path().join("refs")) + .expect("read refs") + .all(|entry| entry + .expect("read ref entry") + .file_name() + .to_string_lossy() + .ends_with(".json")) + ); +} + +#[tokio::test] +async fn open_and_inspect_reject_truncated_modified_missing_and_conflicting_content() { + let root = TestRoot::new("corruption"); + let store = LocalCasArtifactStore::new(root.path()).expect("construct local CAS"); + let bytes = b"verified bytes".to_vec(); + let staged = store + .stage(intent(&bytes, "corruption"), stream(bytes.clone())) + .await + .expect("stage succeeds"); + let artifact_ref = store.publish(&staged).await.expect("publish succeeds"); + let blob = blob_path(root.path(), &artifact_ref); + let manifest = ref_path(root.path(), &artifact_ref); + + std::fs::write(&blob, &bytes[..3]).expect("truncate blob"); + let failure = match store.open(&artifact_ref.artifact_id).await { + Ok(_) => panic!("truncated blob must fail"), + Err(failure) => failure, + }; + assert_eq!(failure.kind(), ArtifactStoreFailureKind::CorruptContent); + let mut modified = bytes.clone(); + modified[0] ^= 0xff; + std::fs::write(&blob, &modified).expect("modify blob"); + assert_eq!( + store + .inspect(&artifact_ref.artifact_id) + .await + .expect_err("modified blob must fail") + .kind(), + ArtifactStoreFailureKind::CorruptContent + ); + std::fs::remove_file(&blob).expect("remove blob"); + let failure = match store.open(&artifact_ref.artifact_id).await { + Ok(_) => panic!("missing blob must fail"), + Err(failure) => failure, + }; + assert_eq!( + failure.kind(), + ArtifactStoreFailureKind::MissingCommittedContent + ); + + std::fs::remove_dir(blob.parent().expect("blob prefix exists")) + .expect("remove empty blob prefix"); + let failure = match store.open(&artifact_ref.artifact_id).await { + Ok(_) => panic!("missing blob prefix must fail"), + Err(failure) => failure, + }; + assert_eq!( + failure.kind(), + ArtifactStoreFailureKind::MissingCommittedContent + ); + + std::fs::create_dir(blob.parent().expect("blob prefix exists")).expect("restore blob prefix"); + std::fs::write(&blob, &bytes).expect("restore blob"); + let mut conflicting = artifact_ref.clone(); + conflicting.media_type = MediaType::new("application/conflict").expect("media type is valid"); + std::fs::write( + &manifest, + serde_json::to_vec(&conflicting).expect("encode conflicting manifest"), + ) + .expect("replace manifest"); + assert_eq!( + store + .inspect(&artifact_ref.artifact_id) + .await + .expect_err("identity-conflicting manifest must fail") + .kind(), + ArtifactStoreFailureKind::IdentityConflict + ); +} + +#[tokio::test] +async fn conflicting_blob_is_rejected_before_manifest_publication() { + let root = TestRoot::new("blob-conflict"); + let store = LocalCasArtifactStore::new(root.path()).expect("construct local CAS"); + let bytes = b"expected bytes".to_vec(); + let artifact_intent = intent(&bytes, "blob-conflict"); + let projected = artifact_intent.artifact_ref(); + let blob = blob_path(root.path(), &projected); + std::fs::create_dir_all(blob.parent().expect("blob parent exists")) + .expect("create blob prefix"); + std::fs::write(&blob, b"wrong content!").expect("write conflicting blob"); + let staged = store + .stage(artifact_intent, stream(bytes)) + .await + .expect("stage succeeds"); + assert_eq!( + store + .publish(&staged) + .await + .expect_err("conflicting blob must fail") + .kind(), + ArtifactStoreFailureKind::CorruptContent + ); + assert!(!ref_path(root.path(), &projected).exists()); +} + +#[tokio::test] +async fn lineage_refs_share_bytes_until_the_last_release() { + let root = TestRoot::new("release"); + let store = LocalCasArtifactStore::new(root.path()).expect("construct local CAS"); + let bytes = b"shared local bytes".to_vec(); + let first = store + .stage(intent(&bytes, "one"), stream(bytes.clone())) + .await + .expect("first stage succeeds"); + let first_ref = store.publish(&first).await.expect("first publish succeeds"); + let second = store + .stage(intent(&bytes, "two"), stream(bytes)) + .await + .expect("second stage succeeds"); + let second_ref = store + .publish(&second) + .await + .expect("second publish succeeds"); + let blob = blob_path(root.path(), &first_ref); + assert_eq!(blob, blob_path(root.path(), &second_ref)); + + assert_eq!( + store + .release(&first_ref.artifact_id) + .await + .expect("first release succeeds"), + ReleaseResult::Released + ); + assert!(blob.is_file()); + assert!( + store + .inspect(&second_ref.artifact_id) + .await + .expect("remaining ref inspects") + .is_some() + ); + assert_eq!( + store + .release(&second_ref.artifact_id) + .await + .expect("last release succeeds"), + ReleaseResult::Released + ); + assert!(!blob.exists()); + assert_eq!( + store + .release(&second_ref.artifact_id) + .await + .expect("release retry succeeds"), + ReleaseResult::NotFound + ); +} + +#[cfg(unix)] +#[test] +fn startup_rejects_symlink_and_non_regular_entries() { + use std::os::unix::fs::symlink; + + let root = TestRoot::new("symlink-startup"); + let target = root.path().with_extension("target"); + std::fs::create_dir(&target).expect("create symlink target"); + symlink(&target, root.path()).expect("create root symlink"); + let failure = match LocalCasArtifactStore::new(root.path()) { + Ok(_) => panic!("symlink root must fail"), + Err(failure) => failure, + }; + assert_eq!(failure.kind(), ArtifactStoreFailureKind::CorruptContent); + std::fs::remove_file(root.path()).expect("remove root symlink"); + std::fs::remove_dir(&target).expect("remove symlink target"); + + std::fs::create_dir_all(root.path().join("staging/non-regular")) + .expect("create non-regular stage entry"); + let failure = match LocalCasArtifactStore::new(root.path()) { + Ok(_) => panic!("non-regular stage entry must fail"), + Err(failure) => failure, + }; + assert_eq!(failure.kind(), ArtifactStoreFailureKind::CorruptContent); +} + +#[cfg(unix)] +#[tokio::test] +async fn publish_and_inspect_reject_symlink_entries() { + use std::os::unix::fs::symlink; + + let root = TestRoot::new("symlink-runtime"); + let store = LocalCasArtifactStore::new(root.path()).expect("construct local CAS"); + let bytes = b"symlink protected".to_vec(); + let artifact_intent = intent(&bytes, "symlink"); + let projected = artifact_intent.artifact_ref(); + let staged = store + .stage(artifact_intent, stream(bytes.clone())) + .await + .expect("stage succeeds"); + let blob = blob_path(root.path(), &projected); + std::fs::create_dir_all(blob.parent().expect("blob parent exists")) + .expect("create blob parent"); + let target = root.path().join("target.bin"); + std::fs::write(&target, &bytes).expect("write target"); + symlink(&target, &blob).expect("create blob symlink"); + assert_eq!( + store + .publish(&staged) + .await + .expect_err("blob symlink must fail") + .kind(), + ArtifactStoreFailureKind::CorruptContent + ); + std::fs::remove_file(&blob).expect("remove blob symlink"); + let artifact_ref = store.publish(&staged).await.expect("publish after repair"); + let manifest = ref_path(root.path(), &artifact_ref); + std::fs::remove_file(&manifest).expect("remove manifest"); + symlink(&target, &manifest).expect("create manifest symlink"); + assert_eq!( + store + .inspect(&artifact_ref.artifact_id) + .await + .expect_err("manifest symlink must fail") + .kind(), + ArtifactStoreFailureKind::CorruptContent + ); +} + +#[cfg(unix)] +#[tokio::test] +async fn inspect_and_open_reject_symlinked_blob_prefix_directory() { + use std::os::unix::fs::symlink; + + let root = TestRoot::new("symlink-prefix"); + let store = LocalCasArtifactStore::new(root.path()).expect("construct local CAS"); + let bytes = b"prefix protected".to_vec(); + let staged = store + .stage(intent(&bytes, "symlink-prefix"), stream(bytes.clone())) + .await + .expect("stage succeeds"); + let artifact_ref = store.publish(&staged).await.expect("publish succeeds"); + let prefix = blob_path(root.path(), &artifact_ref) + .parent() + .expect("blob prefix exists") + .to_path_buf(); + drop(store); + + let displaced = root.path().join("displaced-prefix"); + std::fs::rename(&prefix, &displaced).expect("displace blob prefix"); + symlink(&displaced, &prefix).expect("replace blob prefix with symlink"); + let reopened = LocalCasArtifactStore::new(root.path()).expect("reopen local CAS"); + + assert_eq!( + reopened + .inspect(&artifact_ref.artifact_id) + .await + .expect_err("inspect must reject a symlinked blob prefix") + .kind(), + ArtifactStoreFailureKind::CorruptContent + ); + let failure = match reopened.open(&artifact_ref.artifact_id).await { + Ok(_) => panic!("open must reject a symlinked blob prefix"), + Err(failure) => failure, + }; + assert_eq!(failure.kind(), ArtifactStoreFailureKind::CorruptContent); +} + +#[cfg(unix)] +#[tokio::test] +async fn local_layout_is_owner_only() { + use std::os::unix::fs::PermissionsExt; + + let root = TestRoot::new("permissions"); + let store = LocalCasArtifactStore::new(root.path()).expect("construct local CAS"); + let staged = store + .stage( + intent(b"private", "permissions"), + stream(b"private".to_vec()), + ) + .await + .expect("stage succeeds"); + let artifact_ref = store.publish(&staged).await.expect("publish succeeds"); + for directory in [ + root.path().to_path_buf(), + root.path().join("staging"), + root.path().join("blobs"), + root.path().join("blobs/sha256"), + blob_path(root.path(), &artifact_ref) + .parent() + .expect("blob parent exists") + .to_path_buf(), + root.path().join("refs"), + ] { + assert_eq!( + std::fs::metadata(directory) + .expect("directory metadata") + .permissions() + .mode() + & 0o777, + 0o700 + ); + } + for file in [ + root.path().join("store.lock"), + blob_path(root.path(), &artifact_ref), + ref_path(root.path(), &artifact_ref), + ] { + assert_eq!( + std::fs::metadata(file) + .expect("file metadata") + .permissions() + .mode() + & 0o777, + 0o600 + ); + } +} diff --git a/zeroshot-rust/tests/local_cas/recovery.rs b/zeroshot-rust/tests/local_cas/recovery.rs new file mode 100644 index 00000000..96903c91 --- /dev/null +++ b/zeroshot-rust/tests/local_cas/recovery.rs @@ -0,0 +1,137 @@ +#[cfg(unix)] +use std::pin::Pin; +#[cfg(unix)] +use std::sync::Arc; +#[cfg(unix)] +use std::sync::atomic::{AtomicBool, Ordering}; +#[cfg(unix)] +use std::task::{Context, Poll}; + +#[cfg(unix)] +use tokio::io::{AsyncRead, AsyncWriteExt, ReadBuf}; +#[cfg(unix)] +use zeroshot_engine::artifact_store::{ArtifactStoreFailureKind, ArtifactStoreOperation}; + +use super::*; + +#[cfg(unix)] +struct SignalingReader { + inner: R, + polled: Arc, +} + +#[cfg(unix)] +impl AsyncRead for SignalingReader { + fn poll_read( + self: Pin<&mut Self>, + context: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + this.polled.store(true, Ordering::SeqCst); + Pin::new(&mut this.inner).poll_read(context, buffer) + } +} + +#[tokio::test] +async fn abandoned_manifest_stage_cannot_block_release_after_restart() { + let root = TestRoot::new("manifest-recovery"); + let store = LocalCasArtifactStore::new(root.path()).expect("construct local CAS"); + let bytes = b"durable committed bytes".to_vec(); + let staged = store + .stage(intent(&bytes, "manifest-recovery"), stream(bytes)) + .await + .expect("stage succeeds"); + let artifact_ref = store.publish(&staged).await.expect("publish succeeds"); + drop(store); + + let abandoned = root + .path() + .join("staging") + .join(format!("ref-{}-999.tmp", artifact_ref.artifact_id.as_str())); + std::fs::write(&abandoned, b"synchronized but uncommitted manifest") + .expect("simulate crash before manifest rename"); + + let restarted = LocalCasArtifactStore::new(root.path()).expect("restart cleans staging"); + assert!(!abandoned.exists()); + assert_eq!( + restarted + .release(&artifact_ref.artifact_id) + .await + .expect("abandoned manifest stage cannot block release"), + ReleaseResult::Released + ); +} + +#[cfg(unix)] +#[tokio::test] +async fn failed_stage_cleanup_errors_are_reported_and_recovered_on_restart() { + let root = TestRoot::new("failed-stage-cleanup"); + let store = LocalCasArtifactStore::new(root.path()).expect("construct local CAS"); + let staging = root.path().join("staging"); + let displaced_staging = root.path().join("displaced-staging"); + let (reader, mut writer) = tokio::io::duplex(16); + let reader_polled = Arc::new(AtomicBool::new(false)); + let signaling_reader = SignalingReader { + inner: reader, + polled: Arc::clone(&reader_polled), + }; + let staging_store = store.clone(); + let stage_task = tokio::spawn(async move { + staging_store + .stage( + intent(b"expected", "failed-stage-cleanup"), + Box::new(signaling_reader), + ) + .await + }); + + let mut stage_is_reading = false; + for _ in 0..1_000 { + if reader_polled.load(Ordering::SeqCst) { + stage_is_reading = true; + break; + } + tokio::task::yield_now().await; + } + assert!( + stage_is_reading, + "stage reader must be blocked before injecting cleanup failure" + ); + std::fs::rename(&staging, &displaced_staging).expect("displace staging directory"); + std::fs::write(&staging, b"not a directory").expect("block failed-stage cleanup"); + writer + .write_all(b"different") + .await + .expect("send oversized stage bytes"); + drop(writer); + + let failure = stage_task + .await + .expect("stage task joins") + .expect_err("cleanup failure must replace the content-validation failure"); + assert_eq!( + failure.kind(), + ArtifactStoreFailureKind::Io(ArtifactStoreOperation::Stage) + ); + + std::fs::remove_file(&staging).expect("remove cleanup blocker"); + std::fs::rename(&displaced_staging, &staging).expect("restore staging directory"); + assert_eq!( + std::fs::read_dir(&staging) + .expect("read restored staging directory") + .count(), + 1, + "failed cleanup leaves one recoverable stage" + ); + drop(store); + + let _restarted = + LocalCasArtifactStore::new(root.path()).expect("restart cleans failed stage residue"); + assert_eq!( + std::fs::read_dir(&staging) + .expect("read cleaned staging directory") + .count(), + 0 + ); +} diff --git a/zeroshot-rust/tests/support/mod.rs b/zeroshot-rust/tests/support/mod.rs new file mode 100644 index 00000000..3761addf --- /dev/null +++ b/zeroshot-rust/tests/support/mod.rs @@ -0,0 +1,39 @@ +use std::io::Cursor; + +use openengine_cluster_protocol::{ + ArtifactLineage, ArtifactProducer, ByteLength, Generation, MediaType, NodeName, + PositiveInteger, RedactionClass, RunId, Sha256Digest, TypeId, WorkerRef, +}; +use zeroshot_engine::artifact_store::{ArtifactByteStream, ArtifactIntent}; + +pub fn digest(bytes: &[u8]) -> Sha256Digest { + use sha2::{Digest, Sha256}; + let value = Sha256::digest(bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + Sha256Digest::new(value).expect("test digest is valid") +} + +pub fn test_intent(bytes: &[u8], run_id: &str) -> ArtifactIntent { + ArtifactIntent { + expected_sha256: digest(bytes), + expected_byte_length: ByteLength::new(bytes.len() as u64).expect("test length is valid"), + media_type: MediaType::new("application/test").expect("test media type is valid"), + type_id: TypeId::new("test.artifact@1").expect("test type is valid"), + producer: ArtifactProducer { + node: NodeName::new("produce").expect("test node is valid"), + worker: WorkerRef::new("test.worker@1").expect("test worker is valid"), + }, + lineage: ArtifactLineage { + generation: Generation::new(7).expect("test generation is valid"), + run_id: RunId::new(run_id), + attempt: PositiveInteger::new(2).expect("test attempt is valid"), + }, + redaction: RedactionClass::Internal, + } +} + +pub fn byte_stream(bytes: Vec) -> ArtifactByteStream { + Box::new(Cursor::new(bytes)) +}