1use std::collections::HashSet;
18use std::path::{Path, PathBuf};
19
20use greentic_deploy_spec::{
21 BundleId, EnvId, LockedPack, PackId, PackListLock, Revision, RevisionId, SchemaVersion,
22};
23use sha2::{Digest, Sha256};
24
25use crate::environment::LocalFsStore;
26use crate::environment::atomic_write::atomic_write_json;
27
28use super::OpError;
29
30pub struct StagedBundle {
34 pub bundle_digest: String,
36 pub pack_list_lock_ref: PathBuf,
38 pub lock: PackListLock,
40}
41
42pub fn stage_local_bundle(
49 env_dir: &Path,
50 revision_id: RevisionId,
51 bundle_path: &Path,
52) -> Result<StagedBundle, OpError> {
53 if !bundle_path.is_file() {
54 return Err(OpError::InvalidArgument(format!(
55 "bundle `{}` is not a file",
56 bundle_path.display()
57 )));
58 }
59 let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
60 stage_into(env_dir, &rev_dir, revision_id, bundle_path).inspect_err(|_| {
64 let _ = std::fs::remove_dir_all(&rev_dir);
65 })
66}
67
68pub fn materialize_revision_from_bundle(
98 store: &LocalFsStore,
99 env_id: &EnvId,
100 revision_id: RevisionId,
101 bundle_path: &Path,
102) -> Result<(), OpError> {
103 store.transact(env_id, |locked| {
104 let env = locked.load()?;
105 let revision = env
106 .revisions
107 .iter()
108 .find(|r| r.revision_id == revision_id)
109 .ok_or_else(|| {
110 OpError::NotFound(format!(
111 "revision `{revision_id}` not found in env `{env_id}`"
112 ))
113 })?;
114 let bundle_id = env
115 .bundles
116 .iter()
117 .find(|b| b.deployment_id == revision.deployment_id)
118 .map(|b| &b.bundle_id)
119 .ok_or_else(|| {
120 OpError::NotFound(format!(
121 "deployment `{}` for revision `{revision_id}` has no bundle in env `{env_id}`",
122 revision.deployment_id
123 ))
124 })?;
125
126 let env_dir = store.env_dir(env_id)?;
127 let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
128
129 let backup_dir = env_dir.join("revisions").join(format!("{revision_id}.bak"));
137 let _ = std::fs::remove_dir_all(&backup_dir); let had_existing = rev_dir.exists();
139 if had_existing {
140 std::fs::rename(&rev_dir, &backup_dir).map_err(|source| OpError::Io {
141 path: rev_dir.clone(),
142 source,
143 })?;
144 }
145
146 let outcome =
149 materialize_into_rev_dir(&env_dir, &rev_dir, env_id, bundle_id, bundle_path, revision)
150 .and_then(|()| locked.refresh_runtime_config(&env).map_err(OpError::from));
151
152 match outcome {
153 Ok(()) => {
154 let _ = std::fs::remove_dir_all(&backup_dir);
157 Ok(())
158 }
159 Err(err) => {
160 let _ = std::fs::remove_dir_all(&rev_dir);
163 if had_existing {
164 let _ = std::fs::rename(&backup_dir, &rev_dir);
165 }
166 Err(err)
167 }
168 }
169 })
170}
171
172fn materialize_into_rev_dir(
178 env_dir: &Path,
179 rev_dir: &Path,
180 env_id: &EnvId,
181 bundle_id: &BundleId,
182 bundle_path: &Path,
183 revision: &Revision,
184) -> Result<(), OpError> {
185 let revision_id = revision.revision_id;
186
187 let staged = stage_local_bundle(env_dir, revision_id, bundle_path)?;
189
190 if staged.bundle_digest != revision.bundle_digest {
195 return Err(OpError::Conflict(format!(
196 "pulled bundle digest `{}` does not match revision `{revision_id}`'s pinned `{}`; \
197 refusing to serve unpinned bytes",
198 staged.bundle_digest, revision.bundle_digest
199 )));
200 }
201 if staged.pack_list_lock_ref != revision.pack_list_lock_ref {
205 return Err(OpError::Conflict(format!(
206 "materialized pack-list lock ref `{}` diverges from revision `{revision_id}`'s `{}`",
207 staged.pack_list_lock_ref.display(),
208 revision.pack_list_lock_ref.display()
209 )));
210 }
211
212 let pinned_pack_ids: HashSet<String> = staged
217 .lock
218 .packs
219 .iter()
220 .map(|p| p.pack_id.as_str().to_string())
221 .collect();
222 super::pack_config_stage::materialize_pack_configs(
223 env_dir,
224 rev_dir,
225 revision_id,
226 env_id,
227 bundle_id,
228 &pinned_pack_ids,
229 )?;
230 Ok(())
231}
232
233fn stage_into(
234 env_dir: &Path,
235 rev_dir: &Path,
236 revision_id: RevisionId,
237 bundle_path: &Path,
238) -> Result<StagedBundle, OpError> {
239 let extract_dir = rev_dir.join("bundle");
240
241 if extract_dir.exists() {
243 std::fs::remove_dir_all(&extract_dir).map_err(|source| OpError::Io {
244 path: extract_dir.clone(),
245 source,
246 })?;
247 }
248 std::fs::create_dir_all(&extract_dir).map_err(|source| OpError::Io {
249 path: extract_dir.clone(),
250 source,
251 })?;
252
253 let staged_bundle = rev_dir.join("bundle.gtbundle");
261 std::fs::copy(bundle_path, &staged_bundle).map_err(|source| OpError::Io {
262 path: staged_bundle.clone(),
263 source,
264 })?;
265 let bundle_digest = sha256_file(&staged_bundle).map_err(|source| OpError::Io {
266 path: staged_bundle.clone(),
267 source,
268 })?;
269
270 greentic_bundle::build::unbundle_artifact(&staged_bundle, &extract_dir).map_err(|err| {
273 OpError::InvalidArgument(format!(
274 "extract bundle `{}`: {err:#}",
275 bundle_path.display()
276 ))
277 })?;
278
279 if !extract_dir.join("bundle-manifest.json").is_file() {
282 return Err(OpError::InvalidArgument(format!(
283 "`{}` is not a .gtbundle: extracted tree has no bundle-manifest.json",
284 bundle_path.display()
285 )));
286 }
287
288 let packs_dir = extract_dir.join("packs");
296 if !packs_dir.is_dir() {
297 return Err(OpError::InvalidArgument(format!(
298 "bundle `{}` has no packs/ directory",
299 bundle_path.display()
300 )));
301 }
302 let mut gtpacks = Vec::new();
303 collect_gtpacks(&packs_dir, &mut gtpacks).map_err(|source| OpError::Io {
304 path: packs_dir.clone(),
305 source,
306 })?;
307 if gtpacks.is_empty() {
308 return Err(OpError::InvalidArgument(format!(
309 "bundle `{}` contains no .gtpack artifacts under packs/",
310 bundle_path.display()
311 )));
312 }
313
314 let mut packs = Vec::with_capacity(gtpacks.len());
315 for path in gtpacks {
316 let digest = sha256_file(&path).map_err(|source| OpError::Io {
317 path: path.clone(),
318 source,
319 })?;
320 let rel = path
321 .strip_prefix(env_dir)
322 .map_err(|_| {
323 OpError::InvalidArgument(format!(
324 "extracted pack `{}` escaped the env directory",
325 path.display()
326 ))
327 })?
328 .to_path_buf();
329 let pack_id = path
330 .file_stem()
331 .and_then(|s| s.to_str())
332 .map(PackId::new)
333 .ok_or_else(|| {
334 OpError::InvalidArgument(format!("pack `{}` has no file stem", path.display()))
335 })?;
336 packs.push(LockedPack {
337 pack_id,
338 path: rel,
339 digest,
340 });
341 }
342 packs.sort_by(|a, b| a.path.cmp(&b.path));
344
345 let lock = PackListLock {
346 schema: SchemaVersion::new(SchemaVersion::PACK_LIST_LOCK_V1),
347 revision_id,
348 packs,
349 };
350 let lock_path = rev_dir.join("pack-list.lock");
351 atomic_write_json(&lock_path, &lock)
352 .map_err(|e| OpError::Store(crate::environment::store::StoreError::from(e)))?;
353 let pack_list_lock_ref = lock_path
354 .strip_prefix(env_dir)
355 .map_err(|_| {
356 OpError::InvalidArgument(format!(
357 "pack-list.lock `{}` escaped the env directory",
358 lock_path.display()
359 ))
360 })?
361 .to_path_buf();
362
363 Ok(StagedBundle {
364 bundle_digest,
365 pack_list_lock_ref,
366 lock,
367 })
368}
369
370pub(crate) fn sha256_file(path: &Path) -> std::io::Result<String> {
375 use std::io::Read;
376 let mut file = std::fs::File::open(path)?;
377 let mut hasher = Sha256::new();
378 let mut buf = [0u8; 64 * 1024];
379 loop {
380 let n = file.read(&mut buf)?;
381 if n == 0 {
382 break;
383 }
384 hasher.update(&buf[..n]);
385 }
386 Ok(format!("sha256:{}", hex::encode(hasher.finalize())))
387}
388
389fn collect_gtpacks(dir: &Path, out: &mut Vec<PathBuf>) -> std::io::Result<()> {
394 for entry in std::fs::read_dir(dir)? {
395 let entry = entry?;
396 let file_type = entry.file_type()?;
397 let path = entry.path();
398 if file_type.is_dir() {
399 collect_gtpacks(&path, out)?;
400 } else if file_type.is_file() && path.extension().and_then(|e| e.to_str()) == Some("gtpack")
401 {
402 out.push(path);
403 }
404 }
405 Ok(())
406}
407
408#[cfg(test)]
409mod tests {
410 use super::*;
411 use tempfile::tempdir;
412
413 #[test]
417 fn collect_gtpacks_recurses_and_filters_by_extension() {
418 let dir = tempdir().unwrap();
419 let root = dir.path();
420
421 let dist = root.join("packs/alpha/dist");
423 std::fs::create_dir_all(&dist).unwrap();
424 std::fs::write(dist.join("alpha.gtpack"), b"PK\x03\x04").unwrap();
425 std::fs::write(dist.join("readme.txt"), b"not a pack").unwrap();
426
427 std::fs::write(root.join("stray.gtpack"), b"PK\x03\x04").unwrap();
429
430 let mut found = Vec::new();
431 collect_gtpacks(&root.join("packs"), &mut found).unwrap();
432
433 assert_eq!(found.len(), 1, "only the packs/ .gtpack, got {found:?}");
434 assert!(found[0].ends_with("alpha/dist/alpha.gtpack"));
435 }
436}
437
438#[cfg(test)]
439mod materialize_tests {
440 use super::*;
441 use crate::cli::tests_common::{
442 make_bundle_deployment, make_env, make_revision, make_traffic_split,
443 };
444 use crate::environment::mint_revision_id;
445 use crate::environment::store::EnvironmentStore;
446 use greentic_deploy_spec::{PackListEntry, RevisionLifecycle};
447 use tempfile::tempdir;
448
449 fn fixture_bundle() -> PathBuf {
450 PathBuf::from(env!("CARGO_MANIFEST_DIR"))
451 .join("testdata/bundles/perf-smoke-bundle.gtbundle")
452 }
453
454 fn repin_to_unmatchable_digest(store: &LocalFsStore, env_id: &EnvId, revision_id: RevisionId) {
458 store
459 .transact(env_id, |locked| {
460 let mut env = locked.load()?;
461 env.revisions
462 .iter_mut()
463 .find(|r| r.revision_id == revision_id)
464 .unwrap()
465 .bundle_digest =
466 "sha256:0000000000000000000000000000000000000000000000000000000000000000"
467 .to_string();
468 locked.save(&env)
469 })
470 .unwrap();
471 }
472
473 fn seed_staged_env(store: &LocalFsStore, route: bool) -> (EnvId, RevisionId, PathBuf) {
479 let env_id = EnvId::try_from("local").unwrap();
480 let mut env = make_env("local");
481 let deployment = make_bundle_deployment("local", "fast2flow");
482 let did = deployment.deployment_id;
483 env.bundles.push(deployment);
484
485 let revision_id = mint_revision_id();
486 let env_dir = store.env_dir(&env_id).unwrap();
487 let staged = stage_local_bundle(&env_dir, revision_id, &fixture_bundle()).unwrap();
488
489 let mut revision = make_revision("local", "fast2flow", &did, 1, RevisionLifecycle::Ready);
490 revision.revision_id = revision_id;
491 revision.bundle_digest = staged.bundle_digest.clone();
492 revision.pack_list_lock_ref = staged.pack_list_lock_ref.clone();
493 revision.bundle_source_uri =
494 Some("oci://example.test/bundles/fast2flow@sha256:abc".to_string());
495 revision.pack_list = staged
496 .lock
497 .packs
498 .iter()
499 .map(|p| PackListEntry::from_lock_primitives(p.pack_id.clone(), p.digest.clone()))
500 .collect();
501 env.revisions.push(revision);
502 if route {
503 env.traffic_splits.push(make_traffic_split(
504 "local",
505 "fast2flow",
506 &did,
507 &revision_id,
508 "k1",
509 ));
510 }
511 store.save(&env).unwrap();
512 (env_id, revision_id, env_dir)
513 }
514
515 #[test]
520 fn materializes_packs_lock_and_runtime_config_from_bundle() {
521 let dir = tempdir().unwrap();
522 let store = LocalFsStore::new(dir.path());
523 let (env_id, revision_id, env_dir) = seed_staged_env(&store, true);
524 let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
525
526 std::fs::remove_dir_all(&rev_dir).unwrap();
529 let runtime_config = env_dir.join("runtime-config.json");
530 let _ = std::fs::remove_file(&runtime_config);
531 assert!(!rev_dir.exists());
532
533 materialize_revision_from_bundle(&store, &env_id, revision_id, &fixture_bundle()).unwrap();
534
535 let lock_path = rev_dir.join("pack-list.lock");
536 assert!(lock_path.is_file(), "pack-list.lock must be restored");
537 let lock: PackListLock =
538 serde_json::from_slice(&std::fs::read(&lock_path).unwrap()).unwrap();
539 assert_eq!(lock.revision_id, revision_id);
540 assert!(!lock.packs.is_empty(), "fixture bundle has a .gtpack");
541 for pack in &lock.packs {
542 let pack_path = env_dir.join(&pack.path);
543 assert!(
544 pack_path.is_file(),
545 "extracted pack must exist: {}",
546 pack_path.display()
547 );
548 assert_eq!(
549 pack.digest,
550 sha256_file(&pack_path).unwrap(),
551 "pack digest must match the lock"
552 );
553 }
554 assert!(
555 runtime_config.is_file(),
556 "runtime-config.json must be written when a split routes the revision"
557 );
558 }
559
560 #[test]
564 fn rejects_bundle_whose_digest_does_not_match_the_pin() {
565 let dir = tempdir().unwrap();
566 let store = LocalFsStore::new(dir.path());
567 let (env_id, revision_id, env_dir) = seed_staged_env(&store, true);
568 let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
569
570 repin_to_unmatchable_digest(&store, &env_id, revision_id);
572 std::fs::remove_dir_all(&rev_dir).unwrap();
573
574 let err = materialize_revision_from_bundle(&store, &env_id, revision_id, &fixture_bundle())
575 .unwrap_err();
576 assert!(
577 matches!(err, OpError::Conflict(_)),
578 "digest mismatch must be a Conflict, got: {err}"
579 );
580 assert!(format!("{err}").contains("does not match"));
581 assert!(
582 !rev_dir.exists(),
583 "a rejected materialization must not leave a partial extraction"
584 );
585 }
586
587 #[test]
592 fn preserves_existing_revision_dir_when_re_materialization_fails() {
593 let dir = tempdir().unwrap();
594 let store = LocalFsStore::new(dir.path());
595 let (env_id, revision_id, env_dir) = seed_staged_env(&store, true);
596 let rev_dir = env_dir.join("revisions").join(revision_id.to_string());
597 let lock_path = rev_dir.join("pack-list.lock");
598
599 assert!(
602 lock_path.is_file(),
603 "seed must leave a materialized rev dir"
604 );
605 let good_lock = std::fs::read(&lock_path).unwrap();
606
607 repin_to_unmatchable_digest(&store, &env_id, revision_id);
611
612 let err = materialize_revision_from_bundle(&store, &env_id, revision_id, &fixture_bundle())
613 .unwrap_err();
614 assert!(matches!(err, OpError::Conflict(_)), "got: {err}");
615
616 assert!(
618 lock_path.is_file(),
619 "existing pack-list.lock must survive a failed re-materialize"
620 );
621 assert_eq!(
622 std::fs::read(&lock_path).unwrap(),
623 good_lock,
624 "the restored lock must be byte-identical to the original"
625 );
626 assert!(
627 !env_dir
628 .join("revisions")
629 .join(format!("{revision_id}.bak"))
630 .exists(),
631 "the rollback backup must not leak"
632 );
633 }
634
635 #[test]
637 fn missing_revision_is_not_found() {
638 let dir = tempdir().unwrap();
639 let store = LocalFsStore::new(dir.path());
640 let (env_id, _rid, _env_dir) = seed_staged_env(&store, false);
641 let bogus = mint_revision_id();
642 let err = materialize_revision_from_bundle(&store, &env_id, bogus, &fixture_bundle())
643 .unwrap_err();
644 assert!(matches!(err, OpError::NotFound(_)), "got: {err}");
645 }
646}