use std::collections::HashSet;
use std::path::{Path, PathBuf};
use greentic_deploy_spec::{
BundleId, EnvId, LockedPack, PackId, PackListLock, Revision, RevisionId, SchemaVersion,
};
use sha2::{Digest, Sha256};
use crate::environment::LocalFsStore;
use crate::environment::atomic_write::atomic_write_json;
use super::OpError;
pub struct StagedBundle {
pub bundle_digest: String,
pub pack_list_lock_ref: PathBuf,
pub lock: PackListLock,
}
pub fn stage_local_bundle(
env_dir: &Path,
revision_id: RevisionId,
bundle_path: &Path,
) -> Result<StagedBundle, OpError> {
if !bundle_path.is_file() {
return Err(OpError::InvalidArgument(format!(
"bundle `{}` is not a file",
bundle_path.display()
)));
}
let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
stage_into(env_dir, &rev_dir, revision_id, bundle_path).inspect_err(|_| {
let _ = std::fs::remove_dir_all(&rev_dir);
})
}
pub fn materialize_revision_from_bundle(
store: &LocalFsStore,
env_id: &EnvId,
revision_id: RevisionId,
bundle_path: &Path,
) -> Result<(), OpError> {
store.transact(env_id, |locked| {
let env = locked.load()?;
let revision = env
.revisions
.iter()
.find(|r| r.revision_id == revision_id)
.ok_or_else(|| {
OpError::NotFound(format!(
"revision `{revision_id}` not found in env `{env_id}`"
))
})?;
let bundle_id = env
.bundles
.iter()
.find(|b| b.deployment_id == revision.deployment_id)
.map(|b| &b.bundle_id)
.ok_or_else(|| {
OpError::NotFound(format!(
"deployment `{}` for revision `{revision_id}` has no bundle in env `{env_id}`",
revision.deployment_id
))
})?;
let env_dir = store.env_dir(env_id)?;
let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
let backup_dir = env_dir.join("revisions").join(format!("{revision_id}.bak"));
let _ = std::fs::remove_dir_all(&backup_dir); let had_existing = rev_dir.exists();
if had_existing {
std::fs::rename(&rev_dir, &backup_dir).map_err(|source| OpError::Io {
path: rev_dir.clone(),
source,
})?;
}
let outcome =
materialize_into_rev_dir(&env_dir, &rev_dir, env_id, bundle_id, bundle_path, revision)
.and_then(|()| locked.refresh_runtime_config(&env).map_err(OpError::from));
match outcome {
Ok(()) => {
let _ = std::fs::remove_dir_all(&backup_dir);
Ok(())
}
Err(err) => {
let _ = std::fs::remove_dir_all(&rev_dir);
if had_existing {
let _ = std::fs::rename(&backup_dir, &rev_dir);
}
Err(err)
}
}
})
}
fn materialize_into_rev_dir(
env_dir: &Path,
rev_dir: &Path,
env_id: &EnvId,
bundle_id: &BundleId,
bundle_path: &Path,
revision: &Revision,
) -> Result<(), OpError> {
let revision_id = revision.revision_id;
let staged = stage_local_bundle(env_dir, revision_id, bundle_path)?;
if staged.bundle_digest != revision.bundle_digest {
return Err(OpError::Conflict(format!(
"pulled bundle digest `{}` does not match revision `{revision_id}`'s pinned `{}`; \
refusing to serve unpinned bytes",
staged.bundle_digest, revision.bundle_digest
)));
}
if staged.pack_list_lock_ref != revision.pack_list_lock_ref {
return Err(OpError::Conflict(format!(
"materialized pack-list lock ref `{}` diverges from revision `{revision_id}`'s `{}`",
staged.pack_list_lock_ref.display(),
revision.pack_list_lock_ref.display()
)));
}
let pinned_pack_ids: HashSet<String> = staged
.lock
.packs
.iter()
.map(|p| p.pack_id.as_str().to_string())
.collect();
super::pack_config_stage::materialize_pack_configs(
env_dir,
rev_dir,
revision_id,
env_id,
bundle_id,
&pinned_pack_ids,
)?;
Ok(())
}
fn stage_into(
env_dir: &Path,
rev_dir: &Path,
revision_id: RevisionId,
bundle_path: &Path,
) -> Result<StagedBundle, OpError> {
let extract_dir = rev_dir.join("bundle");
if extract_dir.exists() {
std::fs::remove_dir_all(&extract_dir).map_err(|source| OpError::Io {
path: extract_dir.clone(),
source,
})?;
}
std::fs::create_dir_all(&extract_dir).map_err(|source| OpError::Io {
path: extract_dir.clone(),
source,
})?;
let staged_bundle = rev_dir.join("bundle.gtbundle");
std::fs::copy(bundle_path, &staged_bundle).map_err(|source| OpError::Io {
path: staged_bundle.clone(),
source,
})?;
let bundle_digest = sha256_file(&staged_bundle).map_err(|source| OpError::Io {
path: staged_bundle.clone(),
source,
})?;
greentic_bundle::build::unbundle_artifact(&staged_bundle, &extract_dir).map_err(|err| {
OpError::InvalidArgument(format!(
"extract bundle `{}`: {err:#}",
bundle_path.display()
))
})?;
if !extract_dir.join("bundle-manifest.json").is_file() {
return Err(OpError::InvalidArgument(format!(
"`{}` is not a .gtbundle: extracted tree has no bundle-manifest.json",
bundle_path.display()
)));
}
let packs_dir = extract_dir.join("packs");
if !packs_dir.is_dir() {
return Err(OpError::InvalidArgument(format!(
"bundle `{}` has no packs/ directory",
bundle_path.display()
)));
}
let mut gtpacks = Vec::new();
collect_gtpacks(&packs_dir, &mut gtpacks).map_err(|source| OpError::Io {
path: packs_dir.clone(),
source,
})?;
if gtpacks.is_empty() {
return Err(OpError::InvalidArgument(format!(
"bundle `{}` contains no .gtpack artifacts under packs/",
bundle_path.display()
)));
}
let mut packs = Vec::with_capacity(gtpacks.len());
for path in gtpacks {
let digest = sha256_file(&path).map_err(|source| OpError::Io {
path: path.clone(),
source,
})?;
let rel = path
.strip_prefix(env_dir)
.map_err(|_| {
OpError::InvalidArgument(format!(
"extracted pack `{}` escaped the env directory",
path.display()
))
})?
.to_path_buf();
let pack_id = path
.file_stem()
.and_then(|s| s.to_str())
.map(PackId::new)
.ok_or_else(|| {
OpError::InvalidArgument(format!("pack `{}` has no file stem", path.display()))
})?;
packs.push(LockedPack {
pack_id,
path: rel,
digest,
});
}
packs.sort_by(|a, b| a.path.cmp(&b.path));
let lock = PackListLock {
schema: SchemaVersion::new(SchemaVersion::PACK_LIST_LOCK_V1),
revision_id,
packs,
};
let lock_path = rev_dir.join("pack-list.lock");
atomic_write_json(&lock_path, &lock)
.map_err(|e| OpError::Store(crate::environment::store::StoreError::from(e)))?;
let pack_list_lock_ref = lock_path
.strip_prefix(env_dir)
.map_err(|_| {
OpError::InvalidArgument(format!(
"pack-list.lock `{}` escaped the env directory",
lock_path.display()
))
})?
.to_path_buf();
Ok(StagedBundle {
bundle_digest,
pack_list_lock_ref,
lock,
})
}
pub(crate) fn sha256_file(path: &Path) -> std::io::Result<String> {
use std::io::Read;
let mut file = std::fs::File::open(path)?;
let mut hasher = Sha256::new();
let mut buf = [0u8; 64 * 1024];
loop {
let n = file.read(&mut buf)?;
if n == 0 {
break;
}
hasher.update(&buf[..n]);
}
Ok(format!("sha256:{}", hex::encode(hasher.finalize())))
}
fn collect_gtpacks(dir: &Path, out: &mut Vec<PathBuf>) -> std::io::Result<()> {
for entry in std::fs::read_dir(dir)? {
let entry = entry?;
let file_type = entry.file_type()?;
let path = entry.path();
if file_type.is_dir() {
collect_gtpacks(&path, out)?;
} else if file_type.is_file() && path.extension().and_then(|e| e.to_str()) == Some("gtpack")
{
out.push(path);
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
#[test]
fn collect_gtpacks_recurses_and_filters_by_extension() {
let dir = tempdir().unwrap();
let root = dir.path();
let dist = root.join("packs/alpha/dist");
std::fs::create_dir_all(&dist).unwrap();
std::fs::write(dist.join("alpha.gtpack"), b"PK\x03\x04").unwrap();
std::fs::write(dist.join("readme.txt"), b"not a pack").unwrap();
std::fs::write(root.join("stray.gtpack"), b"PK\x03\x04").unwrap();
let mut found = Vec::new();
collect_gtpacks(&root.join("packs"), &mut found).unwrap();
assert_eq!(found.len(), 1, "only the packs/ .gtpack, got {found:?}");
assert!(found[0].ends_with("alpha/dist/alpha.gtpack"));
}
}
#[cfg(test)]
mod materialize_tests {
use super::*;
use crate::cli::tests_common::{
make_bundle_deployment, make_env, make_revision, make_traffic_split,
};
use crate::environment::mint_revision_id;
use crate::environment::store::EnvironmentStore;
use greentic_deploy_spec::{PackListEntry, RevisionLifecycle};
use tempfile::tempdir;
fn fixture_bundle() -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("testdata/bundles/perf-smoke-bundle.gtbundle")
}
fn repin_to_unmatchable_digest(store: &LocalFsStore, env_id: &EnvId, revision_id: RevisionId) {
store
.transact(env_id, |locked| {
let mut env = locked.load()?;
env.revisions
.iter_mut()
.find(|r| r.revision_id == revision_id)
.unwrap()
.bundle_digest =
"sha256:0000000000000000000000000000000000000000000000000000000000000000"
.to_string();
locked.save(&env)
})
.unwrap();
}
fn seed_staged_env(store: &LocalFsStore, route: bool) -> (EnvId, RevisionId, PathBuf) {
let env_id = EnvId::try_from("local").unwrap();
let mut env = make_env("local");
let deployment = make_bundle_deployment("local", "fast2flow");
let did = deployment.deployment_id;
env.bundles.push(deployment);
let revision_id = mint_revision_id();
let env_dir = store.env_dir(&env_id).unwrap();
let staged = stage_local_bundle(&env_dir, revision_id, &fixture_bundle()).unwrap();
let mut revision = make_revision("local", "fast2flow", &did, 1, RevisionLifecycle::Ready);
revision.revision_id = revision_id;
revision.bundle_digest = staged.bundle_digest.clone();
revision.pack_list_lock_ref = staged.pack_list_lock_ref.clone();
revision.bundle_source_uri =
Some("oci://example.test/bundles/fast2flow@sha256:abc".to_string());
revision.pack_list = staged
.lock
.packs
.iter()
.map(|p| PackListEntry::from_lock_primitives(p.pack_id.clone(), p.digest.clone()))
.collect();
env.revisions.push(revision);
if route {
env.traffic_splits.push(make_traffic_split(
"local",
"fast2flow",
&did,
&revision_id,
"k1",
));
}
store.save(&env).unwrap();
(env_id, revision_id, env_dir)
}
#[test]
fn materializes_packs_lock_and_runtime_config_from_bundle() {
let dir = tempdir().unwrap();
let store = LocalFsStore::new(dir.path());
let (env_id, revision_id, env_dir) = seed_staged_env(&store, true);
let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
std::fs::remove_dir_all(&rev_dir).unwrap();
let runtime_config = env_dir.join("runtime-config.json");
let _ = std::fs::remove_file(&runtime_config);
assert!(!rev_dir.exists());
materialize_revision_from_bundle(&store, &env_id, revision_id, &fixture_bundle()).unwrap();
let lock_path = rev_dir.join("pack-list.lock");
assert!(lock_path.is_file(), "pack-list.lock must be restored");
let lock: PackListLock =
serde_json::from_slice(&std::fs::read(&lock_path).unwrap()).unwrap();
assert_eq!(lock.revision_id, revision_id);
assert!(!lock.packs.is_empty(), "fixture bundle has a .gtpack");
for pack in &lock.packs {
let pack_path = env_dir.join(&pack.path);
assert!(
pack_path.is_file(),
"extracted pack must exist: {}",
pack_path.display()
);
assert_eq!(
pack.digest,
sha256_file(&pack_path).unwrap(),
"pack digest must match the lock"
);
}
assert!(
runtime_config.is_file(),
"runtime-config.json must be written when a split routes the revision"
);
}
#[test]
fn rejects_bundle_whose_digest_does_not_match_the_pin() {
let dir = tempdir().unwrap();
let store = LocalFsStore::new(dir.path());
let (env_id, revision_id, env_dir) = seed_staged_env(&store, true);
let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
repin_to_unmatchable_digest(&store, &env_id, revision_id);
std::fs::remove_dir_all(&rev_dir).unwrap();
let err = materialize_revision_from_bundle(&store, &env_id, revision_id, &fixture_bundle())
.unwrap_err();
assert!(
matches!(err, OpError::Conflict(_)),
"digest mismatch must be a Conflict, got: {err}"
);
assert!(format!("{err}").contains("does not match"));
assert!(
!rev_dir.exists(),
"a rejected materialization must not leave a partial extraction"
);
}
#[test]
fn preserves_existing_revision_dir_when_re_materialization_fails() {
let dir = tempdir().unwrap();
let store = LocalFsStore::new(dir.path());
let (env_id, revision_id, env_dir) = seed_staged_env(&store, true);
let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
let lock_path = rev_dir.join("pack-list.lock");
assert!(
lock_path.is_file(),
"seed must leave a materialized rev dir"
);
let good_lock = std::fs::read(&lock_path).unwrap();
repin_to_unmatchable_digest(&store, &env_id, revision_id);
let err = materialize_revision_from_bundle(&store, &env_id, revision_id, &fixture_bundle())
.unwrap_err();
assert!(matches!(err, OpError::Conflict(_)), "got: {err}");
assert!(
lock_path.is_file(),
"existing pack-list.lock must survive a failed re-materialize"
);
assert_eq!(
std::fs::read(&lock_path).unwrap(),
good_lock,
"the restored lock must be byte-identical to the original"
);
assert!(
!env_dir
.join("revisions")
.join(format!("{revision_id}.bak"))
.exists(),
"the rollback backup must not leak"
);
}
#[test]
fn missing_revision_is_not_found() {
let dir = tempdir().unwrap();
let store = LocalFsStore::new(dir.path());
let (env_id, _rid, _env_dir) = seed_staged_env(&store, false);
let bogus = mint_revision_id();
let err = materialize_revision_from_bundle(&store, &env_id, bogus, &fixture_bundle())
.unwrap_err();
assert!(matches!(err, OpError::NotFound(_)), "got: {err}");
}
}