Skip to main content

objects/store/fs/
pack_install_journal.rs

1// SPDX-License-Identifier: Apache-2.0
2//! L8 A+ pack install journal: durable staging + intent (crash-safe install).
3//!
4//! # Layout under `.heddle/packs/`
5//! ```text
6//! packs/
7//!   <blake3-hex>.pack
8//!   <blake3-hex>.idx
9//!   .staging/<install_id>/{pack,idx}
10//!   .install-intent/<install_id>.json   # identifiers only (v2)
11//!   .install-intent/quarantine/         # malformed / unknown-version intents
12//!   .pack-install.lock
13//! ```
14//!
15//! # Intent (v2)
16//! Persists only `install_id`, `pack_name`, `phase`, `created_unix`.
17//! All paths are reconstructed from a trusted `packs_dir` — never executed
18//! from JSON (Codex review: path containment).
19//!
20//! # Protocol
21//! 1. Stage pack+idx under `.staging/<id>/` (outside the per-pack lock).
22//! 2. Take per-`pack_name` exclusive lock; write one durable **prepared** intent.
23//! 3. Publish pack → update intent to **pack_published**.
24//! 4. Publish index → **remove** intent (no Completed rewrite).
25//! 5. Best-effort remove staging; fsync intent dir after intent unlink.
26//!
27//! Recovery lists intents under a short global listing lock, then recovers each
28//! pack under `try_lock` on that pack (skip if a live install holds it).
29//! Paths are reconstructed, IDs validated; garbage intents are quarantined.
30//! See `docs/program/L8_PACK_INSTALL_JOURNAL.md`.
31
32use std::{
33    fs::{self, File},
34    io::{self, Read},
35    path::{Component, Path, PathBuf},
36    sync::atomic::{AtomicU64, Ordering},
37    time::{SystemTime, UNIX_EPOCH},
38};
39
40use serde::{Deserialize, Serialize};
41
42use crate::{
43    fault_inject,
44    fs_atomic::{create_dir_all_durable, publish_file_durable, sync_directory, write_file_atomic},
45    lock::RepoLock,
46};
47
48/// Intent schema version (v2 = identifiers only; paths reconstructed).
49pub const PACK_INSTALL_INTENT_VERSION: u32 = 2;
50
51/// Default TTL for abandoned install intents / orphan staging (24 hours).
52pub const DEFAULT_PACK_INSTALL_INTENT_TTL_SECS: i64 = 86_400;
53
54/// Tolerate clocks slightly ahead of wall time when computing TTL expiry.
55/// Far-future `created_unix` is clamped to `now` so intents cannot dodge expiry forever.
56pub const INTENT_CLOCK_SKEW_TOLERANCE_SECS: i64 = 300;
57
58const STAGING_DIR_NAME: &str = ".staging";
59const INTENT_DIR_NAME: &str = ".install-intent";
60const QUARANTINE_DIR_NAME: &str = "quarantine";
61const STAGED_PACK_NAME: &str = "pack";
62const STAGED_IDX_NAME: &str = "idx";
63const PACK_LOCKS_DIR_NAME: &str = ".pack-locks";
64/// Legacy global lock name (kept for recover directory scan serialization).
65const PACK_INSTALL_LOCK_NAME: &str = ".pack-install.lock";
66
67// ---------------------------------------------------------------------------
68// Process-local metrics (hosted/adapters can scrape; not a full product pipeline)
69// ---------------------------------------------------------------------------
70
71static METRIC_INSTALLS_OK: AtomicU64 = AtomicU64::new(0);
72static METRIC_INSTALLS_ERR: AtomicU64 = AtomicU64::new(0);
73static METRIC_RECOVER_COMPLETED: AtomicU64 = AtomicU64::new(0);
74static METRIC_RECOVER_ABORTED: AtomicU64 = AtomicU64::new(0);
75static METRIC_RECOVER_SKIPPED: AtomicU64 = AtomicU64::new(0);
76static METRIC_RECOVER_QUARANTINED: AtomicU64 = AtomicU64::new(0);
77
78/// Snapshot of process-local pack-install counters (resettable in tests).
79///
80/// Hosted / maintenance adapters scrape this; it is not a full product metrics
81/// pipeline, but it is the stable hook surface for recover/install observability.
82#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
83pub struct PackInstallMetricsSnapshot {
84    pub installs_ok: u64,
85    pub installs_err: u64,
86    pub recover_completed: u64,
87    pub recover_aborted: u64,
88    pub recover_skipped_in_progress: u64,
89    pub recover_quarantined: u64,
90}
91
92/// Read process-local pack-install metrics.
93pub fn pack_install_metrics_snapshot() -> PackInstallMetricsSnapshot {
94    PackInstallMetricsSnapshot {
95        installs_ok: METRIC_INSTALLS_OK.load(Ordering::Relaxed),
96        installs_err: METRIC_INSTALLS_ERR.load(Ordering::Relaxed),
97        recover_completed: METRIC_RECOVER_COMPLETED.load(Ordering::Relaxed),
98        recover_aborted: METRIC_RECOVER_ABORTED.load(Ordering::Relaxed),
99        recover_skipped_in_progress: METRIC_RECOVER_SKIPPED.load(Ordering::Relaxed),
100        recover_quarantined: METRIC_RECOVER_QUARANTINED.load(Ordering::Relaxed),
101    }
102}
103
104/// Reset process-local metrics (tests / process start hooks).
105pub fn pack_install_metrics_reset() {
106    METRIC_INSTALLS_OK.store(0, Ordering::Relaxed);
107    METRIC_INSTALLS_ERR.store(0, Ordering::Relaxed);
108    METRIC_RECOVER_COMPLETED.store(0, Ordering::Relaxed);
109    METRIC_RECOVER_ABORTED.store(0, Ordering::Relaxed);
110    METRIC_RECOVER_SKIPPED.store(0, Ordering::Relaxed);
111    METRIC_RECOVER_QUARANTINED.store(0, Ordering::Relaxed);
112}
113
114fn metric_inc(counter: &AtomicU64) {
115    counter.fetch_add(1, Ordering::Relaxed);
116}
117
118/// Install lifecycle phase recorded in the durable intent.
119#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
120#[serde(rename_all = "snake_case")]
121pub enum PackInstallPhase {
122    Prepared,
123    PackPublished,
124    /// Legacy; never written by v2 install. Recovery treats as cleanup.
125    #[serde(other)]
126    Completed,
127}
128
129/// Durable intent for a single pack+index install (identifiers only).
130#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
131pub struct PackInstallIntent {
132    pub version: u32,
133    pub install_id: String,
134    /// Content-addressed pack stem (blake3 hex of pack bytes).
135    pub pack_name: String,
136    pub phase: PackInstallPhase,
137    pub created_unix: i64,
138}
139
140/// Summary of recovery work performed.
141#[derive(Debug, Clone, Default, PartialEq, Eq)]
142pub struct PackInstallRecoverReport {
143    pub intents_seen: u64,
144    pub completed: u64,
145    pub aborted: u64,
146    pub cleaned_stale_completed: u64,
147    /// Non-expired intents left alone (likely a concurrent live install).
148    pub skipped_in_progress: u64,
149    /// Orphan `.staging/<id>` directories removed (no matching intent, past TTL).
150    pub orphan_staging_swept: u64,
151    /// Malformed / unknown-version intents moved to quarantine.
152    pub quarantined: u64,
153    pub errors: u64,
154}
155
156/// Per-`pack_name` exclusive lock (cross-thread + cross-process).
157pub(crate) fn acquire_pack_name_lock(
158    packs_dir: &Path,
159    pack_name: &str,
160) -> io::Result<crate::lock::WriteLockGuard> {
161    validate_pack_name(pack_name)?;
162    create_dir_all_durable(packs_dir)?;
163    let locks = packs_dir.join(PACK_LOCKS_DIR_NAME);
164    create_dir_all_durable(&locks)?;
165    let lock = RepoLock::at(locks.join(format!("{pack_name}.lock")));
166    lock.write().map_err(|e| io::Error::other(e.to_string()))
167}
168
169/// Non-blocking per-pack lock. `None` = another install holds this pack.
170pub(crate) fn try_acquire_pack_name_lock(
171    packs_dir: &Path,
172    pack_name: &str,
173) -> io::Result<Option<crate::lock::WriteLockGuard>> {
174    validate_pack_name(pack_name)?;
175    create_dir_all_durable(packs_dir)?;
176    let locks = packs_dir.join(PACK_LOCKS_DIR_NAME);
177    create_dir_all_durable(&locks)?;
178    let lock = RepoLock::at(locks.join(format!("{pack_name}.lock")));
179    lock.try_write().map_err(|e| io::Error::other(e.to_string()))
180}
181
182/// Short global listing lock (intent dir scan only).
183pub(crate) fn acquire_pack_install_lock(
184    packs_dir: &Path,
185) -> io::Result<crate::lock::WriteLockGuard> {
186    create_dir_all_durable(packs_dir)?;
187    let lock = RepoLock::at(packs_dir.join(PACK_INSTALL_LOCK_NAME));
188    lock.write().map_err(|e| io::Error::other(e.to_string()))
189}
190
191impl PackInstallIntent {
192    pub fn new(install_id: String, pack_name: String) -> Self {
193        Self {
194            version: PACK_INSTALL_INTENT_VERSION,
195            install_id,
196            pack_name,
197            phase: PackInstallPhase::Prepared,
198            created_unix: unix_now(),
199        }
200    }
201}
202
203// ---------------------------------------------------------------------------
204// Path reconstruction (trusted packs_dir only)
205// ---------------------------------------------------------------------------
206
207fn unix_now() -> i64 {
208    SystemTime::now()
209        .duration_since(UNIX_EPOCH)
210        .map(|d| d.as_secs() as i64)
211        .unwrap_or(0)
212}
213
214/// Single path component: no separators, no `..`, no empty, limited charset.
215pub(crate) fn validate_install_id(id: &str) -> io::Result<()> {
216    if id.is_empty() || id.len() > 128 {
217        return Err(io::Error::new(
218            io::ErrorKind::InvalidInput,
219            "invalid install_id length",
220        ));
221    }
222    if !id
223        .chars()
224        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
225    {
226        return Err(io::Error::new(
227            io::ErrorKind::InvalidInput,
228            "install_id contains illegal characters",
229        ));
230    }
231    if id == "." || id == ".." {
232        return Err(io::Error::new(
233            io::ErrorKind::InvalidInput,
234            "install_id must not be . or ..",
235        ));
236    }
237    Ok(())
238}
239
240/// Content-addressed pack stem: exactly 64 lowercase hex digits (BLAKE3).
241pub(crate) fn validate_pack_name(name: &str) -> io::Result<()> {
242    pack_name_to_digest(name).map(|_| ())
243}
244
245/// Decode a validated `pack_name` to the native 32-byte BLAKE3 digest.
246///
247/// Prefer comparing digests over hex strings once past the FS/JSON boundary.
248pub(crate) fn pack_name_to_digest(name: &str) -> io::Result<[u8; 32]> {
249    if name.len() != 64 {
250        return Err(io::Error::new(
251            io::ErrorKind::InvalidInput,
252            "pack_name must be exactly 64 lowercase hex digits (BLAKE3)",
253        ));
254    }
255    if !name
256        .bytes()
257        .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
258    {
259        return Err(io::Error::new(
260            io::ErrorKind::InvalidInput,
261            "pack_name must be lowercase hexadecimal",
262        ));
263    }
264    let mut digest = [0u8; 32];
265    // decode_to_slice avoids an intermediate Vec; name is already length-checked.
266    hex::decode_to_slice(name.as_bytes(), &mut digest).map_err(|e| {
267        io::Error::new(
268            io::ErrorKind::InvalidInput,
269            format!("pack_name hex decode failed: {e}"),
270        )
271    })?;
272    Ok(digest)
273}
274
275/// Hex form of a BLAKE3 digest for filenames / intent JSON (one allocation).
276fn digest_to_pack_name(digest: &[u8; 32]) -> String {
277    // blake3::Hash::to_hex is stack; to_string once at the FS boundary.
278    blake3::Hash::from_bytes(*digest).to_hex().to_string()
279}
280
281fn validate_intent_ids(intent: &PackInstallIntent) -> io::Result<()> {
282    validate_install_id(&intent.install_id)?;
283    pack_name_to_digest(&intent.pack_name)?;
284    Ok(())
285}
286
287pub(crate) fn staging_root(packs_dir: &Path) -> PathBuf {
288    packs_dir.join(STAGING_DIR_NAME)
289}
290
291pub(crate) fn intent_root(packs_dir: &Path) -> PathBuf {
292    packs_dir.join(INTENT_DIR_NAME)
293}
294
295fn quarantine_root(packs_dir: &Path) -> PathBuf {
296    intent_root(packs_dir).join(QUARANTINE_DIR_NAME)
297}
298
299pub(crate) fn intent_path(packs_dir: &Path, install_id: &str) -> PathBuf {
300    intent_root(packs_dir).join(format!("{install_id}.json"))
301}
302
303pub(crate) fn staging_dir(packs_dir: &Path, install_id: &str) -> PathBuf {
304    staging_root(packs_dir).join(install_id)
305}
306
307fn staging_pack_path(packs_dir: &Path, install_id: &str) -> PathBuf {
308    staging_dir(packs_dir, install_id).join(STAGED_PACK_NAME)
309}
310
311fn staging_idx_path(packs_dir: &Path, install_id: &str) -> PathBuf {
312    staging_dir(packs_dir, install_id).join(STAGED_IDX_NAME)
313}
314
315fn dst_pack_path(packs_dir: &Path, pack_name: &str) -> PathBuf {
316    packs_dir.join(format!("{pack_name}.pack"))
317}
318
319fn dst_idx_path(packs_dir: &Path, pack_name: &str) -> PathBuf {
320    packs_dir.join(format!("{pack_name}.idx"))
321}
322
323/// True when `path` is `root` or a descendant (component-wise).
324fn path_is_within(path: &Path, root: &Path) -> bool {
325    path == root || path.starts_with(root)
326}
327
328/// Ensure `candidate` cannot resolve outside `packs_dir`.
329///
330/// **Canonical containment is authoritative.** A lexical
331/// `candidate.starts_with(packs_dir)` must never override a canonical
332/// escape (classic case: `.staging` → symlink to `/tmp/evil`).
333///
334/// Walks every existing path component (including intermediate symlinks)
335/// and rejects any prefix whose `canonicalize` leaves `packs_dir`.
336/// Nonexistent trailing components are allowed only under a trusted base.
337fn assert_under_packs(packs_dir: &Path, candidate: &Path) -> io::Result<()> {
338    if !packs_dir.exists() {
339        return Err(io::Error::new(
340            io::ErrorKind::NotFound,
341            "packs_dir does not exist for path containment check",
342        ));
343    }
344    let packs_canon = fs::canonicalize(packs_dir)?;
345
346    let rel = candidate.strip_prefix(packs_dir).or_else(|_| {
347        candidate.strip_prefix(&packs_canon).map_err(|_| {
348            io::Error::new(
349                io::ErrorKind::InvalidInput,
350                "path is not under packs_dir",
351            )
352        })
353    })?;
354
355    for c in rel.components() {
356        match c {
357            Component::Normal(_) | Component::CurDir => {}
358            Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
359                return Err(io::Error::new(
360                    io::ErrorKind::InvalidInput,
361                    "path escapes packs_dir",
362                ));
363            }
364        }
365    }
366
367    let mut cur = packs_canon.clone();
368    for c in rel.components() {
369        let Component::Normal(name) = c else {
370            continue;
371        };
372        cur.push(name);
373        // symlink_metadata succeeds for files, dirs, and (possibly broken) symlinks.
374        match cur.symlink_metadata() {
375            Ok(_) => {
376                let canon = fs::canonicalize(&cur).map_err(|e| {
377                    io::Error::new(
378                        e.kind(),
379                        format!(
380                            "cannot resolve path under packs_dir ({}): {e}",
381                            cur.display()
382                        ),
383                    )
384                })?;
385                if !path_is_within(&canon, &packs_canon) {
386                    return Err(io::Error::new(
387                        io::ErrorKind::InvalidInput,
388                        "reconstructed path escapes packs_dir via symlink or mount",
389                    ));
390                }
391                // Continue from resolved location so nested escapes are still caught.
392                cur = canon;
393            }
394            Err(e) if e.kind() == io::ErrorKind::NotFound => {
395                // Remaining names are pure (already validated as Normal components).
396                break;
397            }
398            Err(e) => return Err(e),
399        }
400    }
401    Ok(())
402}
403
404/// Reject hostile journal roots (`.staging`, `.install-intent`, `.pack-locks`
405/// as symlinks/mounts that escape `packs_dir`).
406fn ensure_journal_layout_safe(packs_dir: &Path) -> io::Result<()> {
407    create_dir_all_durable(packs_dir)?;
408    for name in [STAGING_DIR_NAME, INTENT_DIR_NAME, PACK_LOCKS_DIR_NAME] {
409        let p = packs_dir.join(name);
410        if p.symlink_metadata().is_ok() {
411            assert_under_packs(packs_dir, &p)?;
412        }
413    }
414    Ok(())
415}
416
417// ---------------------------------------------------------------------------
418// Intent I/O
419// ---------------------------------------------------------------------------
420
421pub(crate) fn write_intent(packs_dir: &Path, intent: &PackInstallIntent) -> io::Result<()> {
422    validate_intent_ids(intent)?;
423    create_dir_all_durable(&intent_root(packs_dir))?;
424    let path = intent_path(packs_dir, &intent.install_id);
425    let bytes = serde_json::to_vec_pretty(intent)
426        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
427    write_file_atomic(&path, &bytes)
428}
429
430pub(crate) fn load_intent(path: &Path) -> io::Result<PackInstallIntent> {
431    let bytes = fs::read(path)?;
432    serde_json::from_slice(&bytes).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
433}
434
435pub(crate) fn remove_intent(packs_dir: &Path, install_id: &str) -> io::Result<()> {
436    validate_install_id(install_id)?;
437    let path = intent_path(packs_dir, install_id);
438    match fs::remove_file(&path) {
439        Ok(()) => {}
440        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
441        Err(e) => return Err(e),
442    }
443    // Make intent-dirent removal more durable (Codex #6).
444    let _ = sync_directory(&intent_root(packs_dir));
445    Ok(())
446}
447
448fn quarantine_intent_file(packs_dir: &Path, path: &Path) -> io::Result<()> {
449    let qroot = quarantine_root(packs_dir);
450    create_dir_all_durable(&qroot)?;
451    let name = path
452        .file_name()
453        .and_then(|n| n.to_str())
454        .unwrap_or("intent.json");
455    let dest = qroot.join(format!("{name}.{}.bad", unix_now()));
456    match fs::rename(path, &dest) {
457        Ok(()) => {
458            let _ = sync_directory(&qroot);
459            tracing::warn!(
460                from = %path.display(),
461                to = %dest.display(),
462                "quarantined unreadable pack install intent"
463            );
464            Ok(())
465        }
466        Err(e) => {
467            // If rename fails, leave original in place (do not delete).
468            Err(e)
469        }
470    }
471}
472
473fn remove_path_best_effort(path: &Path) {
474    if path.is_dir() {
475        let _ = fs::remove_dir_all(path);
476    } else {
477        let _ = fs::remove_file(path);
478    }
479}
480
481pub(crate) fn remove_staging(packs_dir: &Path, install_id: &str) {
482    if validate_install_id(install_id).is_err() {
483        return;
484    }
485    remove_path_best_effort(&staging_dir(packs_dir, install_id));
486}
487
488// ---------------------------------------------------------------------------
489// Abort / complete (paths always from packs_dir + ids)
490// ---------------------------------------------------------------------------
491
492pub(crate) fn abort_install(packs_dir: &Path, intent: &PackInstallIntent) -> io::Result<()> {
493    validate_intent_ids(intent)?;
494    let dst_pack = dst_pack_path(packs_dir, &intent.pack_name);
495    let dst_idx = dst_idx_path(packs_dir, &intent.pack_name);
496    assert_under_packs(packs_dir, &dst_pack)?;
497    assert_under_packs(packs_dir, &dst_idx)?;
498
499    // Only remove final pack if index is missing (partial publish).
500    if dst_pack.exists() && !dst_idx.exists() {
501        let _ = fs::remove_file(&dst_pack);
502        if let Some(parent) = dst_pack.parent() {
503            let _ = sync_directory(parent);
504        }
505    }
506    remove_staging(packs_dir, &intent.install_id);
507    remove_intent(packs_dir, &intent.install_id)?;
508    Ok(())
509}
510
511pub(crate) fn complete_from_staging(
512    packs_dir: &Path,
513    intent: &PackInstallIntent,
514) -> io::Result<()> {
515    validate_intent_ids(intent)?;
516    let staging_idx = staging_idx_path(packs_dir, &intent.install_id);
517    let dst_idx = dst_idx_path(packs_dir, &intent.pack_name);
518    let dst_pack = dst_pack_path(packs_dir, &intent.pack_name);
519    assert_under_packs(packs_dir, &staging_idx)?;
520    assert_under_packs(packs_dir, &dst_idx)?;
521    assert_under_packs(packs_dir, &dst_pack)?;
522
523    if dst_pack.exists() && dst_idx.exists() {
524        remove_staging(packs_dir, &intent.install_id);
525        remove_intent(packs_dir, &intent.install_id)?;
526        return Ok(());
527    }
528
529    if !dst_pack.exists() || !staging_idx.exists() {
530        return abort_install(packs_dir, intent);
531    }
532
533    publish_file_durable(&staging_idx, &dst_idx)?;
534    remove_staging(packs_dir, &intent.install_id);
535    remove_intent(packs_dir, &intent.install_id)?;
536    Ok(())
537}
538
539fn can_complete_quickly(packs_dir: &Path, intent: &PackInstallIntent) -> bool {
540    let dst_pack = dst_pack_path(packs_dir, &intent.pack_name);
541    let dst_idx = dst_idx_path(packs_dir, &intent.pack_name);
542    let staging_idx = staging_idx_path(packs_dir, &intent.install_id);
543    if dst_pack.exists() && dst_idx.exists() {
544        return true;
545    }
546    dst_pack.exists() && !dst_idx.exists() && staging_idx.exists()
547}
548
549/// Effective creation time for TTL when `created_unix` is only slightly ahead
550/// of wall time (within [`INTENT_CLOCK_SKEW_TOLERANCE_SECS`]).
551fn effective_created_unix(created_unix: i64, now: i64) -> i64 {
552    if created_unix > now {
553        now
554    } else {
555        created_unix
556    }
557}
558
559fn intent_expired(intent: &PackInstallIntent, ttl_secs: Option<i64>, now: i64) -> bool {
560    match ttl_secs {
561        Some(ttl) if ttl >= 0 => {
562            // Far-future / large clock rollback: expire immediately so a forged
563            // `created_unix = i64::MAX` (or a multi-minute clock jump) cannot
564            // keep an intent alive forever. Mild skew is clamped to `now`.
565            if intent.created_unix > now.saturating_add(INTENT_CLOCK_SKEW_TOLERANCE_SECS) {
566                return true;
567            }
568            let created = effective_created_unix(intent.created_unix, now);
569            created.saturating_add(ttl) < now
570        }
571        _ => false,
572    }
573}
574
575// ---------------------------------------------------------------------------
576// Existing pair identity validation
577// ---------------------------------------------------------------------------
578
579/// Stream-hash a pack file to a native 32-byte BLAKE3 digest (no hex).
580fn hash_file_blake3(path: &Path) -> io::Result<[u8; 32]> {
581    let mut file = File::open(path)?;
582    let mut hasher = blake3::Hasher::new();
583    let mut buf = vec![0u8; 64 * 1024];
584    loop {
585        let n = file.read(&mut buf)?;
586        if n == 0 {
587            break;
588        }
589        hasher.update(&buf[..n]);
590    }
591    Ok(*hasher.finalize().as_bytes())
592}
593
594/// True when a final pack+idx pair is safe to treat as already installed:
595/// pack content BLAKE3 equals `pack_name`, and the index **parses** as a
596/// [`crate::store::pack::PackIndex`].
597///
598/// Identity is checked as **native digests** (`[u8; 32]`), not hex strings.
599/// Hex is only the durable/public form of `pack_name` on disk.
600///
601/// This does **not** prove every index offset points at a live object in the
602/// pack (that is the pack reader's job on first use). It rejects empty,
603/// garbage, and structurally invalid indexes so install idempotency cannot
604/// accept a corrupt pair.
605/// String-name entry for tests and external call sites that only have hex.
606#[cfg(test)]
607fn existing_pair_matches_pack_name(packs_dir: &Path, pack_name: &str) -> io::Result<bool> {
608    let expected = pack_name_to_digest(pack_name)?;
609    existing_pair_matches_digest(packs_dir, pack_name, &expected)
610}
611
612/// Like [`existing_pair_matches_pack_name`], but reuses an already-decoded digest
613/// (install hot path: hash once in memory, compare file digest to those bytes).
614fn existing_pair_matches_digest(
615    packs_dir: &Path,
616    pack_name: &str,
617    expected: &[u8; 32],
618) -> io::Result<bool> {
619    let pack = dst_pack_path(packs_dir, pack_name);
620    let idx = dst_idx_path(packs_dir, pack_name);
621    assert_under_packs(packs_dir, &pack)?;
622    assert_under_packs(packs_dir, &idx)?;
623    if !pack.exists() || !idx.exists() {
624        return Ok(false);
625    }
626    // Symlink destinations must stay under packs (assert_under_packs) and the
627    // open path must be a regular file for install identity.
628    if !pack.is_file() || !idx.is_file() {
629        return Ok(false);
630    }
631    if idx.metadata()?.len() == 0 {
632        return Ok(false);
633    }
634    let actual = hash_file_blake3(&pack)?;
635    if actual != *expected {
636        return Ok(false);
637    }
638    let idx_bytes = fs::read(&idx)?;
639    match crate::store::pack::PackIndex::from_bytes(&idx_bytes) {
640        Ok(_) => Ok(true),
641        Err(_) => Ok(false),
642    }
643}
644
645// ---------------------------------------------------------------------------
646// Recovery
647// ---------------------------------------------------------------------------
648
649pub fn recover_pack_install_intents(packs_dir: &Path) -> io::Result<PackInstallRecoverReport> {
650    recover_pack_install_intents_with_ttl(packs_dir, Some(DEFAULT_PACK_INSTALL_INTENT_TTL_SECS))
651}
652
653pub fn recover_pack_install_intents_with_ttl(
654    packs_dir: &Path,
655    ttl_secs: Option<i64>,
656) -> io::Result<PackInstallRecoverReport> {
657    // Reject hostile journal roots (symlink escape) before any mutation.
658    ensure_journal_layout_safe(packs_dir)?;
659
660    // Snapshot intent paths under a short global listing lock, then recover
661    // each pack under its per-pack lock (try_lock → skip if install holds it).
662    let intent_paths: Vec<PathBuf> = {
663        let _list_guard = acquire_pack_install_lock(packs_dir)?;
664        let intent_dir = intent_root(packs_dir);
665        if !intent_dir.exists() {
666            let mut report = PackInstallRecoverReport::default();
667            let now = unix_now();
668            sweep_orphan_staging(packs_dir, ttl_secs, now, &mut report);
669            return Ok(report);
670        }
671        fs::read_dir(&intent_dir)?
672            .filter_map(|e| e.ok())
673            .map(|e| e.path())
674            .filter(|p| !p.is_dir() && p.extension().and_then(|x| x.to_str()) == Some("json"))
675            .collect()
676    };
677
678    let mut report = PackInstallRecoverReport::default();
679    let now = unix_now();
680
681    for path in intent_paths {
682        report.intents_seen += 1;
683
684        let intent = match load_intent(&path) {
685            Ok(i)
686                if i.version == PACK_INSTALL_INTENT_VERSION
687                    || i.version == 1 /* v1 ids still usable; paths ignored */ =>
688            {
689                match validate_intent_ids(&i) {
690                    Ok(()) => i,
691                    Err(_) => {
692                        let _ = quarantine_intent_file(packs_dir, &path);
693                        report.quarantined += 1;
694                        metric_inc(&METRIC_RECOVER_QUARANTINED);
695                        continue;
696                    }
697                }
698            }
699            Ok(_) | Err(_) => {
700                match quarantine_intent_file(packs_dir, &path) {
701                    Ok(()) => {
702                        report.quarantined += 1;
703                        metric_inc(&METRIC_RECOVER_QUARANTINED);
704                    }
705                    Err(_) => report.errors += 1,
706                }
707                continue;
708            }
709        };
710
711        // Per-pack try-lock: if a live install holds it, skip (in progress).
712        let pack_guard = match try_acquire_pack_name_lock(packs_dir, &intent.pack_name)? {
713            Some(g) => g,
714            None => {
715                report.skipped_in_progress += 1;
716                metric_inc(&METRIC_RECOVER_SKIPPED);
717                continue;
718            }
719        };
720
721        let expired = intent_expired(&intent, ttl_secs, now);
722        let before_aborted = report.aborted;
723        let before_completed = report.completed;
724        let before_skipped = report.skipped_in_progress;
725        if recover_one_intent(packs_dir, &intent, expired, &mut report).is_err() {
726            report.errors += 1;
727        }
728        if report.completed > before_completed {
729            metric_inc(&METRIC_RECOVER_COMPLETED);
730        }
731        if report.aborted > before_aborted {
732            metric_inc(&METRIC_RECOVER_ABORTED);
733        }
734        if report.skipped_in_progress > before_skipped {
735            metric_inc(&METRIC_RECOVER_SKIPPED);
736        }
737        drop(pack_guard);
738    }
739
740    sweep_orphan_staging(packs_dir, ttl_secs, now, &mut report);
741
742    if report.intents_seen > 0
743        || report.orphan_staging_swept > 0
744        || report.errors > 0
745        || report.completed > 0
746        || report.aborted > 0
747        || report.skipped_in_progress > 0
748        || report.quarantined > 0
749    {
750        tracing::info!(
751            ?packs_dir,
752            intents_seen = report.intents_seen,
753            completed = report.completed,
754            aborted = report.aborted,
755            skipped_in_progress = report.skipped_in_progress,
756            cleaned_stale_completed = report.cleaned_stale_completed,
757            orphan_staging_swept = report.orphan_staging_swept,
758            quarantined = report.quarantined,
759            errors = report.errors,
760            metrics = ?pack_install_metrics_snapshot(),
761            "pack install journal recovery"
762        );
763    } else {
764        tracing::debug!(?packs_dir, "pack install journal recovery: nothing to do");
765    }
766
767    Ok(report)
768}
769
770fn recover_one_intent(
771    packs_dir: &Path,
772    intent: &PackInstallIntent,
773    expired: bool,
774    report: &mut PackInstallRecoverReport,
775) -> io::Result<()> {
776    if can_complete_quickly(packs_dir, intent) {
777        return match intent.phase {
778            PackInstallPhase::Prepared | PackInstallPhase::PackPublished => {
779                let dst_pack = dst_pack_path(packs_dir, &intent.pack_name);
780                let dst_idx = dst_idx_path(packs_dir, &intent.pack_name);
781                if dst_pack.exists() && dst_idx.exists() {
782                    remove_staging(packs_dir, &intent.install_id);
783                    remove_intent(packs_dir, &intent.install_id)?;
784                    report.cleaned_stale_completed += 1;
785                    Ok(())
786                } else {
787                    complete_from_staging(packs_dir, intent)?;
788                    if dst_pack_path(packs_dir, &intent.pack_name).exists()
789                        && dst_idx_path(packs_dir, &intent.pack_name).exists()
790                    {
791                        report.completed += 1;
792                    } else {
793                        report.aborted += 1;
794                    }
795                    Ok(())
796                }
797            }
798            PackInstallPhase::Completed => {
799                remove_staging(packs_dir, &intent.install_id);
800                remove_intent(packs_dir, &intent.install_id)?;
801                report.cleaned_stale_completed += 1;
802                Ok(())
803            }
804        };
805    }
806
807    if expired {
808        tracing::debug!(
809            install_id = %intent.install_id,
810            pack_name = %intent.pack_name,
811            phase = ?intent.phase,
812            "aborting expired pack install intent"
813        );
814        abort_install(packs_dir, intent)?;
815        report.aborted += 1;
816        return Ok(());
817    }
818
819    match intent.phase {
820        PackInstallPhase::Prepared | PackInstallPhase::PackPublished => {
821            report.skipped_in_progress += 1;
822            Ok(())
823        }
824        PackInstallPhase::Completed => {
825            remove_staging(packs_dir, &intent.install_id);
826            remove_intent(packs_dir, &intent.install_id)?;
827            report.cleaned_stale_completed += 1;
828            Ok(())
829        }
830    }
831}
832
833fn sweep_orphan_staging(
834    packs_dir: &Path,
835    ttl_secs: Option<i64>,
836    now: i64,
837    report: &mut PackInstallRecoverReport,
838) {
839    let Some(ttl) = ttl_secs.filter(|t| *t >= 0) else {
840        return;
841    };
842    let staging = staging_root(packs_dir);
843    let entries = match fs::read_dir(&staging) {
844        Ok(e) => e,
845        Err(_) => return,
846    };
847    for entry in entries.flatten() {
848        let path = entry.path();
849        if !path.is_dir() {
850            continue;
851        }
852        let Some(id) = path.file_name().and_then(|n| n.to_str()) else {
853            continue;
854        };
855        if validate_install_id(id).is_err() {
856            continue;
857        }
858        if intent_path(packs_dir, id).exists() {
859            continue;
860        }
861        let age_ok_to_sweep = path_mtime_unix(&path)
862            .map(|mtime| mtime.saturating_add(ttl) < now)
863            .unwrap_or(true);
864        if !age_ok_to_sweep {
865            continue;
866        }
867        remove_path_best_effort(&path);
868        report.orphan_staging_swept += 1;
869    }
870}
871
872fn path_mtime_unix(path: &Path) -> Option<i64> {
873    let meta = fs::metadata(path).ok()?;
874    let modified = meta.modified().ok()?;
875    Some(
876        modified
877            .duration_since(UNIX_EPOCH)
878            .map(|d| d.as_secs() as i64)
879            .unwrap_or(0),
880    )
881}
882
883// ---------------------------------------------------------------------------
884// Install
885// ---------------------------------------------------------------------------
886
887fn new_install_id() -> String {
888    let t = unix_now() as u64;
889    let r: u64 = rand::random();
890    format!("{t:016x}-{r:016x}")
891}
892
893/// Journaled in-memory install. Returns content-addressed `pack_name`.
894pub fn install_pack_bytes_journaled(
895    packs_dir: &Path,
896    pack_data: &[u8],
897    index_data: &[u8],
898) -> io::Result<String> {
899    match install_pack_bytes_journaled_inner(packs_dir, pack_data, index_data) {
900        Ok(name) => {
901            metric_inc(&METRIC_INSTALLS_OK);
902            Ok(name)
903        }
904        Err(e) => {
905            metric_inc(&METRIC_INSTALLS_ERR);
906            Err(e)
907        }
908    }
909}
910
911fn install_pack_bytes_journaled_inner(
912    packs_dir: &Path,
913    pack_data: &[u8],
914    index_data: &[u8],
915) -> io::Result<String> {
916    ensure_journal_layout_safe(packs_dir)?;
917    // Hash once as native bytes; hex only for the FS/JSON name boundary.
918    let digest = *blake3::hash(pack_data).as_bytes();
919    let pack_name = digest_to_pack_name(&digest);
920
921    if existing_pair_matches_digest(packs_dir, &pack_name, &digest)? {
922        return Ok(pack_name);
923    }
924
925    // Stage outside per-pack lock (unique install_id).
926    let install_id = new_install_id();
927    validate_install_id(&install_id)?;
928    let stage = staging_dir(packs_dir, &install_id);
929    assert_under_packs(packs_dir, &stage)?;
930    create_dir_all_durable(&stage)?;
931    let staging_pack = staging_pack_path(packs_dir, &install_id);
932    let staging_idx = staging_idx_path(packs_dir, &install_id);
933    assert_under_packs(packs_dir, &staging_pack)?;
934    assert_under_packs(packs_dir, &staging_idx)?;
935    write_file_atomic(&staging_pack, pack_data)?;
936    fault_inject::maybe_fail_at("pack_install_after_stage_pack")?;
937    write_file_atomic(&staging_idx, index_data)?;
938    fault_inject::maybe_fail_at("pack_install_after_stage_idx")?;
939
940    // Per-pack lock for intent + publish (other pack names stay parallel).
941    let _guard = acquire_pack_name_lock(packs_dir, &pack_name)?;
942    fault_inject::maybe_fail_at("pack_install_after_pack_lock")?;
943
944    if existing_pair_matches_digest(packs_dir, &pack_name, &digest)? {
945        remove_staging(packs_dir, &install_id);
946        return Ok(pack_name);
947    }
948    let dst_pack = dst_pack_path(packs_dir, &pack_name);
949    let dst_idx = dst_idx_path(packs_dir, &pack_name);
950    assert_under_packs(packs_dir, &dst_pack)?;
951    assert_under_packs(packs_dir, &dst_idx)?;
952    if dst_pack.exists() && !dst_idx.exists() {
953        let _ = fs::remove_file(&dst_pack);
954    }
955    if dst_pack.exists() && dst_idx.exists() {
956        if !existing_pair_matches_digest(packs_dir, &pack_name, &digest)? {
957            let _ = fs::remove_file(&dst_pack);
958            let _ = fs::remove_file(&dst_idx);
959        } else {
960            remove_staging(packs_dir, &install_id);
961            return Ok(pack_name);
962        }
963    }
964
965    let mut intent = PackInstallIntent::new(install_id.clone(), pack_name.clone());
966    write_intent(packs_dir, &intent)?;
967    fault_inject::maybe_fail_at("pack_install_after_intent_prepared")?;
968
969    publish_file_durable(&staging_pack, &dst_pack)?;
970    fault_inject::maybe_fail_at("pack_install_after_publish_pack")?;
971    intent.phase = PackInstallPhase::PackPublished;
972    write_intent(packs_dir, &intent)?;
973    fault_inject::maybe_fail_at("pack_install_after_intent_pack_published")?;
974
975    publish_file_durable(&staging_idx, &dst_idx)?;
976    fault_inject::maybe_fail_at("pack_install_after_publish_idx")?;
977    remove_staging(packs_dir, &install_id);
978    remove_intent(packs_dir, &install_id)?;
979    fault_inject::maybe_fail_at("pack_install_after_intent_removed")?;
980    Ok(pack_name)
981}
982
983/// Journaled streaming install (consumes source pack/index paths).
984pub fn install_pack_files_journaled(
985    packs_dir: &Path,
986    src_pack_path: &Path,
987    src_index_path: &Path,
988    pack_name: &str,
989) -> io::Result<()> {
990    match install_pack_files_journaled_inner(packs_dir, src_pack_path, src_index_path, pack_name)
991    {
992        Ok(()) => {
993            metric_inc(&METRIC_INSTALLS_OK);
994            Ok(())
995        }
996        Err(e) => {
997            metric_inc(&METRIC_INSTALLS_ERR);
998            Err(e)
999        }
1000    }
1001}
1002
1003fn install_pack_files_journaled_inner(
1004    packs_dir: &Path,
1005    src_pack_path: &Path,
1006    src_index_path: &Path,
1007    pack_name: &str,
1008) -> io::Result<()> {
1009    // Decode once; identity checks stay on native digests.
1010    let expected = pack_name_to_digest(pack_name)?;
1011    ensure_journal_layout_safe(packs_dir)?;
1012
1013    if existing_pair_matches_digest(packs_dir, pack_name, &expected)? {
1014        let _ = fs::remove_file(src_pack_path);
1015        let _ = fs::remove_file(src_index_path);
1016        return Ok(());
1017    }
1018
1019    let install_id = new_install_id();
1020    validate_install_id(&install_id)?;
1021    let stage = staging_dir(packs_dir, &install_id);
1022    assert_under_packs(packs_dir, &stage)?;
1023    create_dir_all_durable(&stage)?;
1024    let staging_pack = staging_pack_path(packs_dir, &install_id);
1025    let staging_idx = staging_idx_path(packs_dir, &install_id);
1026    assert_under_packs(packs_dir, &staging_pack)?;
1027    assert_under_packs(packs_dir, &staging_idx)?;
1028    publish_file_durable(src_pack_path, &staging_pack)?;
1029    fault_inject::maybe_fail_at("pack_install_stream_after_stage_pack")?;
1030    publish_file_durable(src_index_path, &staging_idx)?;
1031    fault_inject::maybe_fail_at("pack_install_stream_after_stage_idx")?;
1032
1033    let _guard = acquire_pack_name_lock(packs_dir, pack_name)?;
1034    if existing_pair_matches_digest(packs_dir, pack_name, &expected)? {
1035        remove_staging(packs_dir, &install_id);
1036        return Ok(());
1037    }
1038    let dst_pack = dst_pack_path(packs_dir, pack_name);
1039    let dst_idx = dst_idx_path(packs_dir, pack_name);
1040    assert_under_packs(packs_dir, &dst_pack)?;
1041    assert_under_packs(packs_dir, &dst_idx)?;
1042    if dst_pack.exists() && !dst_idx.exists() {
1043        let _ = fs::remove_file(&dst_pack);
1044    }
1045    if dst_pack.exists() && dst_idx.exists() {
1046        if existing_pair_matches_digest(packs_dir, pack_name, &expected)? {
1047            remove_staging(packs_dir, &install_id);
1048            return Ok(());
1049        }
1050        let _ = fs::remove_file(&dst_pack);
1051        let _ = fs::remove_file(&dst_idx);
1052    }
1053
1054    let mut intent = PackInstallIntent::new(install_id.clone(), pack_name.to_string());
1055    write_intent(packs_dir, &intent)?;
1056    fault_inject::maybe_fail_at("pack_install_stream_after_intent_prepared")?;
1057
1058    publish_file_durable(&staging_pack, &dst_pack)?;
1059    fault_inject::maybe_fail_at("pack_install_stream_after_publish_pack")?;
1060    intent.phase = PackInstallPhase::PackPublished;
1061    write_intent(packs_dir, &intent)?;
1062    fault_inject::maybe_fail_at("pack_install_stream_after_intent_pack_published")?;
1063
1064    publish_file_durable(&staging_idx, &dst_idx)?;
1065    fault_inject::maybe_fail_at("pack_install_stream_after_publish_idx")?;
1066    remove_staging(packs_dir, &install_id);
1067    remove_intent(packs_dir, &install_id)?;
1068    Ok(())
1069}
1070
1071// ---------------------------------------------------------------------------
1072// Tests
1073// ---------------------------------------------------------------------------
1074
1075#[cfg(test)]
1076mod tests {
1077    use super::*;
1078    use crate::fs_atomic::write_file_atomic;
1079
1080    fn write_src(dir: &Path, name: &str, bytes: &[u8]) -> PathBuf {
1081        let p = dir.join(name);
1082        write_file_atomic(&p, bytes).unwrap();
1083        p
1084    }
1085
1086    /// Deterministic 64-char lowercase hex pack name from a seed.
1087    fn pack_id(seed: &str) -> String {
1088        digest_to_pack_name(blake3::hash(seed.as_bytes()).as_bytes())
1089    }
1090
1091    /// Minimal structurally valid pack index bytes.
1092    fn empty_idx_bytes() -> Vec<u8> {
1093        crate::store::pack::PackIndex::new().to_bytes()
1094    }
1095
1096    #[test]
1097    fn validate_ids_reject_path_traversal() {
1098        assert!(validate_install_id("../evil").is_err());
1099        assert!(validate_install_id("a/b").is_err());
1100        assert!(validate_install_id("").is_err());
1101        assert!(validate_pack_name("../x").is_err());
1102        assert!(validate_pack_name("not-hex!").is_err());
1103        assert!(validate_pack_name("deadbeef").is_err()); // too short
1104        assert!(validate_pack_name(&"A".repeat(64)).is_err()); // uppercase
1105        assert!(validate_install_id("abc-123_OK").is_ok());
1106        assert!(validate_pack_name(&pack_id("ok")).is_ok());
1107        assert_eq!(pack_id("ok").len(), 64);
1108    }
1109
1110    #[test]
1111    fn pack_name_digest_roundtrip_and_native_equality() {
1112        let body = b"digest-native-eq";
1113        let digest = *blake3::hash(body).as_bytes();
1114        let name = digest_to_pack_name(&digest);
1115        let parsed = pack_name_to_digest(&name).unwrap();
1116        assert_eq!(parsed, digest);
1117        // File identity uses bytes, not hex strings.
1118        let root = tempfile::tempdir().unwrap();
1119        let packs = root.path().join("packs");
1120        create_dir_all_durable(&packs).unwrap();
1121        write_file_atomic(&dst_pack_path(&packs, &name), body).unwrap();
1122        write_file_atomic(&dst_idx_path(&packs, &name), &empty_idx_bytes()).unwrap();
1123        assert!(existing_pair_matches_digest(&packs, &name, &digest).unwrap());
1124        let mut wrong = digest;
1125        wrong[0] ^= 0xff;
1126        assert!(!existing_pair_matches_digest(&packs, &name, &wrong).unwrap());
1127    }
1128
1129    #[test]
1130    fn malicious_intent_paths_ignored_reconstructed_from_packs_dir() {
1131        // Even if someone forged absolute paths in JSON, serde ignores unknown
1132        // fields and we only use install_id + pack_name.
1133        let root = tempfile::tempdir().unwrap();
1134        let packs = root.path().join("packs");
1135        create_dir_all_durable(&packs).unwrap();
1136        let outside = root.path().join("outside.txt");
1137        fs::write(&outside, b"secret").unwrap();
1138
1139        let install_id = "malicious1";
1140        let pack_name = pack_id("malicious-pack");
1141        let intent_dir = intent_root(&packs);
1142        create_dir_all_durable(&intent_dir).unwrap();
1143        let forged = serde_json::json!({
1144            "version": 2,
1145            "install_id": install_id,
1146            "pack_name": pack_name,
1147            "phase": "prepared",
1148            "created_unix": 1,
1149            "staging_pack": outside.display().to_string(),
1150            "dst_pack": outside.display().to_string(),
1151            "dst_idx": outside.display().to_string(),
1152        });
1153        fs::write(
1154            intent_path(&packs, install_id),
1155            serde_json::to_vec_pretty(&forged).unwrap(),
1156        )
1157        .unwrap();
1158
1159        // Expired → abort uses reconstructed paths only.
1160        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1161        assert_eq!(report.aborted, 1);
1162        // Outside file must survive.
1163        assert_eq!(fs::read(&outside).unwrap(), b"secret");
1164    }
1165
1166    #[test]
1167    fn quarantine_unknown_version_preserves_file() {
1168        let root = tempfile::tempdir().unwrap();
1169        let packs = root.path().join("packs");
1170        create_dir_all_durable(&packs).unwrap();
1171        let intent_dir = intent_root(&packs);
1172        create_dir_all_durable(&intent_dir).unwrap();
1173        let path = intent_dir.join("weird.json");
1174        let pn = pack_id("weird");
1175        fs::write(
1176            &path,
1177            format!(
1178                r#"{{"version":99,"install_id":"x","pack_name":"{pn}","phase":"prepared","created_unix":1}}"#
1179            ),
1180        )
1181        .unwrap();
1182
1183        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1184        assert_eq!(report.quarantined, 1);
1185        assert!(!path.exists());
1186        let q = quarantine_root(&packs);
1187        assert!(q.exists());
1188        assert!(fs::read_dir(&q).unwrap().count() >= 1);
1189    }
1190
1191    #[test]
1192    fn short_pack_name_in_intent_is_quarantined() {
1193        let root = tempfile::tempdir().unwrap();
1194        let packs = root.path().join("packs");
1195        create_dir_all_durable(&packs).unwrap();
1196        create_dir_all_durable(&intent_root(&packs)).unwrap();
1197        let path = intent_path(&packs, "shortname");
1198        fs::write(
1199            &path,
1200            br#"{"version":2,"install_id":"shortname","pack_name":"aa","phase":"prepared","created_unix":1}"#,
1201        )
1202        .unwrap();
1203        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1204        assert_eq!(report.quarantined, 1);
1205        assert!(!path.exists());
1206    }
1207
1208    #[test]
1209    fn quarantine_malformed_json() {
1210        let root = tempfile::tempdir().unwrap();
1211        let packs = root.path().join("packs");
1212        create_dir_all_durable(&packs).unwrap();
1213        let intent_dir = intent_root(&packs);
1214        create_dir_all_durable(&intent_dir).unwrap();
1215        let path = intent_dir.join("bad.json");
1216        fs::write(&path, b"not-json{{{{").unwrap();
1217
1218        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1219        assert_eq!(report.quarantined, 1);
1220        assert!(!path.exists());
1221    }
1222
1223    #[test]
1224    fn existing_pair_requires_hash_match_and_valid_index() {
1225        let root = tempfile::tempdir().unwrap();
1226        let packs = root.path().join("packs");
1227        create_dir_all_durable(&packs).unwrap();
1228        let body = b"pack-body-xyz";
1229        let name = format!("{}", blake3::hash(body).to_hex());
1230        write_file_atomic(&dst_pack_path(&packs, &name), body).unwrap();
1231        write_file_atomic(&dst_idx_path(&packs, &name), &empty_idx_bytes()).unwrap();
1232        assert!(existing_pair_matches_pack_name(&packs, &name).unwrap());
1233
1234        // Structurally invalid index → not a match.
1235        write_file_atomic(&dst_idx_path(&packs, &name), b"not-an-index").unwrap();
1236        assert!(!existing_pair_matches_pack_name(&packs, &name).unwrap());
1237
1238        // Wrong name for content (valid hex, wrong digest).
1239        let wrong = pack_id("wrong-name-for-body");
1240        write_file_atomic(&dst_pack_path(&packs, &wrong), body).unwrap();
1241        write_file_atomic(&dst_idx_path(&packs, &wrong), &empty_idx_bytes()).unwrap();
1242        assert!(!existing_pair_matches_pack_name(&packs, &wrong).unwrap());
1243    }
1244
1245    #[test]
1246    fn install_rejects_idempotent_false_pair() {
1247        let root = tempfile::tempdir().unwrap();
1248        let packs = root.path().join("packs");
1249        create_dir_all_durable(&packs).unwrap();
1250        let pack_bytes = b"real-pack-content-111";
1251        let idx_bytes = empty_idx_bytes();
1252        let name = format!("{}", blake3::hash(pack_bytes).to_hex());
1253
1254        // Corrupt pair with correct name but wrong content.
1255        write_file_atomic(&dst_pack_path(&packs, &name), b"wrong-content").unwrap();
1256        write_file_atomic(&dst_idx_path(&packs, &name), b"x").unwrap();
1257        assert!(!existing_pair_matches_pack_name(&packs, &name).unwrap());
1258
1259        let out = install_pack_bytes_journaled(&packs, pack_bytes, &idx_bytes).unwrap();
1260        assert_eq!(out, name);
1261        assert_eq!(fs::read(dst_pack_path(&packs, &name)).unwrap(), pack_bytes);
1262    }
1263
1264    #[test]
1265    fn journaled_install_produces_pair_and_cleans_intent() {
1266        let root = tempfile::tempdir().unwrap();
1267        let packs = root.path().join("packs");
1268        create_dir_all_durable(&packs).unwrap();
1269        let src_dir = root.path().join("src");
1270        create_dir_all_durable(&src_dir).unwrap();
1271
1272        let pack_bytes = b"fake-pack-bytes-aaa";
1273        let idx_bytes = b"fake-idx-bytes-aaa";
1274        let src_pack = write_src(&src_dir, "p.pack", pack_bytes);
1275        let src_idx = write_src(&src_dir, "p.idx", idx_bytes);
1276        let name = format!("{}", blake3::hash(pack_bytes).to_hex());
1277
1278        install_pack_files_journaled(&packs, &src_pack, &src_idx, &name).unwrap();
1279
1280        assert!(dst_pack_path(&packs, &name).exists());
1281        assert!(dst_idx_path(&packs, &name).exists());
1282        assert_eq!(fs::read(dst_pack_path(&packs, &name)).unwrap(), pack_bytes);
1283        assert!(
1284            !intent_root(&packs).exists()
1285                || fs::read_dir(intent_root(&packs))
1286                    .unwrap()
1287                    .filter(|e| {
1288                        e.as_ref()
1289                            .map(|e| e.path().extension().and_then(|x| x.to_str()) == Some("json"))
1290                            .unwrap_or(false)
1291                    })
1292                    .count()
1293                    == 0
1294        );
1295    }
1296
1297    #[test]
1298    fn recover_pack_published_completes_from_staging() {
1299        let root = tempfile::tempdir().unwrap();
1300        let packs = root.path().join("packs");
1301        create_dir_all_durable(&packs).unwrap();
1302
1303        let name = pack_id("deadbeef-seed");
1304        let install_id = "test-install-1";
1305        validate_install_id(install_id).unwrap();
1306        validate_pack_name(&name).unwrap();
1307        let stage = staging_dir(&packs, install_id);
1308        create_dir_all_durable(&stage).unwrap();
1309        write_file_atomic(&staging_pack_path(&packs, install_id), b"pack-body").unwrap();
1310        write_file_atomic(&staging_idx_path(&packs, install_id), b"idx-body").unwrap();
1311
1312        let dst_pack = dst_pack_path(&packs, &name);
1313        publish_file_durable(&staging_pack_path(&packs, install_id), &dst_pack).unwrap();
1314
1315        let intent = PackInstallIntent {
1316            version: PACK_INSTALL_INTENT_VERSION,
1317            install_id: install_id.to_string(),
1318            pack_name: name.clone(),
1319            phase: PackInstallPhase::PackPublished,
1320            created_unix: 1,
1321        };
1322        write_intent(&packs, &intent).unwrap();
1323
1324        let report = recover_pack_install_intents(&packs).unwrap();
1325        assert_eq!(report.intents_seen, 1);
1326        assert_eq!(report.completed, 1);
1327        assert!(dst_pack.exists());
1328        assert!(dst_idx_path(&packs, &name).exists());
1329        assert_eq!(fs::read(dst_idx_path(&packs, &name)).unwrap(), b"idx-body");
1330        assert!(!intent_path(&packs, install_id).exists());
1331    }
1332
1333    #[test]
1334    fn recover_prepared_aborts_without_finals_when_expired() {
1335        let root = tempfile::tempdir().unwrap();
1336        let packs = root.path().join("packs");
1337        create_dir_all_durable(&packs).unwrap();
1338        let install_id = "prep-abort";
1339        let stage = staging_dir(&packs, install_id);
1340        create_dir_all_durable(&stage).unwrap();
1341        write_file_atomic(&staging_pack_path(&packs, install_id), b"p").unwrap();
1342        write_file_atomic(&staging_idx_path(&packs, install_id), b"i").unwrap();
1343
1344        let mut intent = PackInstallIntent::new(install_id.into(), pack_id("aa"));
1345        intent.created_unix = 1;
1346        write_intent(&packs, &intent).unwrap();
1347
1348        let report = recover_pack_install_intents_with_ttl(&packs, Some(60)).unwrap();
1349        assert_eq!(report.aborted, 1);
1350        assert!(!dst_pack_path(&packs, &pack_id("aa")).exists());
1351        assert!(!intent_path(&packs, install_id).exists());
1352        assert!(!stage.exists());
1353    }
1354
1355    #[test]
1356    fn recover_prepared_fresh_skips_in_progress() {
1357        let root = tempfile::tempdir().unwrap();
1358        let packs = root.path().join("packs");
1359        create_dir_all_durable(&packs).unwrap();
1360        let install_id = "live-prep";
1361        let stage = staging_dir(&packs, install_id);
1362        create_dir_all_durable(&stage).unwrap();
1363        let staging_pack = staging_pack_path(&packs, install_id);
1364        let staging_idx = staging_idx_path(&packs, install_id);
1365        write_file_atomic(&staging_pack, b"live-p").unwrap();
1366        write_file_atomic(&staging_idx, b"live-i").unwrap();
1367
1368        let intent = PackInstallIntent::new(install_id.into(), pack_id("bb"));
1369        write_intent(&packs, &intent).unwrap();
1370
1371        let report = recover_pack_install_intents_with_ttl(&packs, Some(86_400)).unwrap();
1372        assert_eq!(report.skipped_in_progress, 1);
1373        assert_eq!(report.aborted, 0);
1374        assert!(staging_pack.exists());
1375        assert!(intent_path(&packs, install_id).exists());
1376    }
1377
1378    #[test]
1379    fn recover_pack_published_without_staging_idx_aborts_orphan_pack() {
1380        let root = tempfile::tempdir().unwrap();
1381        let packs = root.path().join("packs");
1382        create_dir_all_durable(&packs).unwrap();
1383        let name = pack_id("cc");
1384        let install_id = "orph-1";
1385        let dst_pack = dst_pack_path(&packs, &name);
1386        write_file_atomic(&dst_pack, b"only-pack").unwrap();
1387
1388        let intent = PackInstallIntent {
1389            version: PACK_INSTALL_INTENT_VERSION,
1390            install_id: install_id.into(),
1391            pack_name: name,
1392            phase: PackInstallPhase::PackPublished,
1393            created_unix: 1,
1394        };
1395        write_intent(&packs, &intent).unwrap();
1396
1397        let report = recover_pack_install_intents(&packs).unwrap();
1398        assert_eq!(report.aborted, 1);
1399        assert!(!dst_pack.exists());
1400        assert!(!intent_path(&packs, install_id).exists());
1401    }
1402
1403    #[test]
1404    fn pack_lock_prevents_recover_aborting_live_expired_looking_install() {
1405        use std::{
1406            sync::{Arc, Barrier},
1407            thread,
1408            time::Duration,
1409        };
1410
1411        // Per-pack try_lock: recover must skip (not abort) while install holds
1412        // the pack lock — even if created_unix looks TTL-expired.
1413        let root = tempfile::tempdir().unwrap();
1414        let packs = Arc::new(root.path().join("packs"));
1415        create_dir_all_durable(&packs).unwrap();
1416
1417        let planted_under_lock = Arc::new(Barrier::new(2));
1418        let both_done = Arc::new(Barrier::new(2));
1419
1420        let packs_a = Arc::clone(&packs);
1421        let planted_a = Arc::clone(&planted_under_lock);
1422        let done_a = Arc::clone(&both_done);
1423
1424        let installer = thread::spawn(move || {
1425            let packs = packs_a.as_path();
1426            let guard = acquire_pack_name_lock(packs, &pack_id("dd")).expect("pack lock");
1427
1428            let install_id = "flock-live";
1429            let stage = staging_dir(packs, install_id);
1430            create_dir_all_durable(&stage).unwrap();
1431            let staging_pack = staging_pack_path(packs, install_id);
1432            let staging_idx = staging_idx_path(packs, install_id);
1433            write_file_atomic(&staging_pack, b"flock-pack").unwrap();
1434            write_file_atomic(&staging_idx, b"flock-idx").unwrap();
1435            let dst_pack = dst_pack_path(packs, &pack_id("dd"));
1436            let dst_idx = dst_idx_path(packs, &pack_id("dd"));
1437
1438            let mut intent = PackInstallIntent::new(install_id.into(), pack_id("dd"));
1439            intent.created_unix = 1; // looks expired under any short TTL
1440            write_intent(packs, &intent).unwrap();
1441
1442            planted_a.wait();
1443            thread::sleep(Duration::from_millis(60));
1444
1445            assert!(staging_pack.exists() && staging_idx.exists());
1446            assert!(intent_path(packs, install_id).exists());
1447
1448            publish_file_durable(&staging_pack, &dst_pack).unwrap();
1449            intent.phase = PackInstallPhase::PackPublished;
1450            write_intent(packs, &intent).unwrap();
1451            publish_file_durable(&staging_idx, &dst_idx).unwrap();
1452            remove_staging(packs, install_id);
1453            remove_intent(packs, install_id).unwrap();
1454
1455            drop(guard);
1456            assert!(dst_pack.exists() && dst_idx.exists());
1457            done_a.wait();
1458        });
1459
1460        let packs_b = Arc::clone(&packs);
1461        let planted_b = Arc::clone(&planted_under_lock);
1462        let done_b = Arc::clone(&both_done);
1463
1464        let recoverer = thread::spawn(move || {
1465            let packs = packs_b.as_path();
1466            planted_b.wait();
1467            // While pack lock is held, recover try_locks and skips — must not abort.
1468            let mid =
1469                recover_pack_install_intents_with_ttl(packs, Some(1)).expect("recover under pack lock");
1470            assert_eq!(mid.aborted, 0, "must not abort live install: {mid:?}");
1471            assert!(
1472                mid.skipped_in_progress >= 1 || dst_pack_path(packs, &pack_id("dd")).exists(),
1473                "either skip in-progress or install already finished: {mid:?}"
1474            );
1475            done_b.wait();
1476            // After installer finishes, finals exist and no intent remains.
1477            assert!(dst_pack_path(packs, &pack_id("dd")).exists());
1478            assert!(dst_idx_path(packs, &pack_id("dd")).exists());
1479            assert!(!intent_path(packs, "flock-live").exists());
1480        });
1481
1482        installer.join().expect("installer");
1483        recoverer.join().expect("recoverer");
1484    }
1485
1486    #[test]
1487    fn recover_prepared_with_pack_and_staging_idx_completes() {
1488        let root = tempfile::tempdir().unwrap();
1489        let packs = root.path().join("packs");
1490        create_dir_all_durable(&packs).unwrap();
1491        let install_id = "prep-complete";
1492        let stage = staging_dir(&packs, install_id);
1493        create_dir_all_durable(&stage).unwrap();
1494        write_file_atomic(&staging_pack_path(&packs, install_id), b"pack-x").unwrap();
1495        write_file_atomic(&staging_idx_path(&packs, install_id), b"idx-x").unwrap();
1496        let dst_pack = dst_pack_path(&packs, &pack_id("ee"));
1497        publish_file_durable(&staging_pack_path(&packs, install_id), &dst_pack).unwrap();
1498
1499        let intent = PackInstallIntent::new(install_id.into(), pack_id("ee"));
1500        write_intent(&packs, &intent).unwrap();
1501
1502        let report = recover_pack_install_intents(&packs).unwrap();
1503        assert_eq!(report.completed, 1);
1504        assert!(dst_pack.exists());
1505        assert!(dst_idx_path(&packs, &pack_id("ee")).exists());
1506        assert_eq!(fs::read(dst_idx_path(&packs, &pack_id("ee"))).unwrap(), b"idx-x");
1507    }
1508
1509    #[test]
1510    fn journaled_install_idempotent_when_pair_exists() {
1511        let root = tempfile::tempdir().unwrap();
1512        let packs = root.path().join("packs");
1513        create_dir_all_durable(&packs).unwrap();
1514        let pack_bytes = b"idemp-pack-bytes";
1515        let name = format!("{}", blake3::hash(pack_bytes).to_hex());
1516        write_file_atomic(&dst_pack_path(&packs, &name), pack_bytes).unwrap();
1517        write_file_atomic(&dst_idx_path(&packs, &name), &empty_idx_bytes()).unwrap();
1518        assert!(existing_pair_matches_pack_name(&packs, &name).unwrap());
1519
1520        let src_dir = root.path().join("src");
1521        create_dir_all_durable(&src_dir).unwrap();
1522        let src_pack = write_src(&src_dir, "a", b"other");
1523        let src_idx = write_src(&src_dir, "b", b"other-i");
1524
1525        install_pack_files_journaled(&packs, &src_pack, &src_idx, &name).unwrap();
1526        assert_eq!(fs::read(dst_pack_path(&packs, &name)).unwrap(), pack_bytes);
1527    }
1528
1529    #[test]
1530    fn install_pack_bytes_journaled_happy_path() {
1531        let root = tempfile::tempdir().unwrap();
1532        let packs = root.path().join("packs");
1533        create_dir_all_durable(&packs).unwrap();
1534
1535        let pack_bytes = b"in-memory-pack-body-zzz";
1536        let idx_bytes = empty_idx_bytes();
1537        let expected_name = format!("{}", blake3::hash(pack_bytes).to_hex());
1538
1539        let name = install_pack_bytes_journaled(&packs, pack_bytes, &idx_bytes).unwrap();
1540        assert_eq!(name, expected_name);
1541        assert!(existing_pair_matches_pack_name(&packs, &name).unwrap());
1542
1543        let name2 = install_pack_bytes_journaled(&packs, pack_bytes, &idx_bytes).unwrap();
1544        assert_eq!(name2, expected_name);
1545    }
1546
1547    #[test]
1548    fn ttl_aborts_old_prepared_intent() {
1549        let root = tempfile::tempdir().unwrap();
1550        let packs = root.path().join("packs");
1551        create_dir_all_durable(&packs).unwrap();
1552        let install_id = "ttl-prep";
1553        let stage = staging_dir(&packs, install_id);
1554        create_dir_all_durable(&stage).unwrap();
1555        write_file_atomic(&staging_pack_path(&packs, install_id), b"stale-p").unwrap();
1556        write_file_atomic(&staging_idx_path(&packs, install_id), b"stale-i").unwrap();
1557
1558        let mut intent = PackInstallIntent::new(install_id.into(), pack_id("ff"));
1559        intent.created_unix = 1;
1560        write_intent(&packs, &intent).unwrap();
1561
1562        let report = recover_pack_install_intents_with_ttl(&packs, Some(60)).unwrap();
1563        assert_eq!(report.aborted, 1);
1564        assert!(!intent_path(&packs, install_id).exists());
1565        assert!(!stage.exists());
1566    }
1567
1568    #[test]
1569    fn complete_preferred_over_ttl_when_staging_idx_present() {
1570        let root = tempfile::tempdir().unwrap();
1571        let packs = root.path().join("packs");
1572        create_dir_all_durable(&packs).unwrap();
1573
1574        let name = pack_id("11");
1575        let install_id = "ttl-complete-1";
1576        let stage = staging_dir(&packs, install_id);
1577        create_dir_all_durable(&stage).unwrap();
1578        write_file_atomic(&staging_pack_path(&packs, install_id), b"pack-ttl").unwrap();
1579        write_file_atomic(&staging_idx_path(&packs, install_id), b"idx-ttl").unwrap();
1580
1581        let dst_pack = dst_pack_path(&packs, &name);
1582        publish_file_durable(&staging_pack_path(&packs, install_id), &dst_pack).unwrap();
1583
1584        let intent = PackInstallIntent {
1585            version: PACK_INSTALL_INTENT_VERSION,
1586            install_id: install_id.into(),
1587            pack_name: name.clone(),
1588            phase: PackInstallPhase::PackPublished,
1589            created_unix: 1,
1590        };
1591        write_intent(&packs, &intent).unwrap();
1592
1593        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1594        assert_eq!(report.completed, 1);
1595        assert!(dst_idx_path(&packs, &name).exists());
1596    }
1597
1598    #[test]
1599    fn relocated_repo_recovery_uses_new_packs_dir() {
1600        // Intent has only ids; moving the packs tree still recovers via new root.
1601        let root = tempfile::tempdir().unwrap();
1602        let packs = root.path().join("old").join("packs");
1603        create_dir_all_durable(&packs).unwrap();
1604        let install_id = "reloc1";
1605        let name = pack_id("22");
1606        create_dir_all_durable(&staging_dir(&packs, install_id)).unwrap();
1607        write_file_atomic(&staging_pack_path(&packs, install_id), b"p").unwrap();
1608        write_file_atomic(&staging_idx_path(&packs, install_id), b"i").unwrap();
1609        publish_file_durable(
1610            &staging_pack_path(&packs, install_id),
1611            &dst_pack_path(&packs, &name),
1612        )
1613        .unwrap();
1614        // Restage idx after pack publish consumed staging pack
1615        write_file_atomic(&staging_idx_path(&packs, install_id), b"i").unwrap();
1616        let intent = PackInstallIntent {
1617            version: 2,
1618            install_id: install_id.into(),
1619            pack_name: name.clone(),
1620            phase: PackInstallPhase::PackPublished,
1621            created_unix: 1,
1622        };
1623        write_intent(&packs, &intent).unwrap();
1624
1625        // "Move" repo: rename packs directory
1626        let new_packs = root.path().join("new").join("packs");
1627        create_dir_all_durable(new_packs.parent().unwrap()).unwrap();
1628        fs::rename(&packs, &new_packs).unwrap();
1629
1630        let report = recover_pack_install_intents(&new_packs).unwrap();
1631        assert_eq!(report.completed, 1);
1632        assert!(dst_idx_path(&new_packs, &name).exists());
1633    }
1634
1635    #[test]
1636    fn concurrent_same_pack_installs_converge() {
1637        use std::thread;
1638
1639        let root = tempfile::tempdir().unwrap();
1640        let packs = root.path().join("packs");
1641        create_dir_all_durable(&packs).unwrap();
1642        let pack_bytes = b"same-pack-concurrent-body";
1643        let idx_bytes = empty_idx_bytes();
1644        let expected = format!("{}", blake3::hash(pack_bytes).to_hex());
1645
1646        let packs1 = packs.clone();
1647        let packs2 = packs.clone();
1648        let idx1 = idx_bytes.clone();
1649        let idx2 = idx_bytes.clone();
1650        let t1 = thread::spawn(move || {
1651            install_pack_bytes_journaled(&packs1, pack_bytes, &idx1).unwrap()
1652        });
1653        let t2 = thread::spawn(move || {
1654            install_pack_bytes_journaled(&packs2, pack_bytes, &idx2).unwrap()
1655        });
1656        let n1 = t1.join().unwrap();
1657        let n2 = t2.join().unwrap();
1658        assert_eq!(n1, expected);
1659        assert_eq!(n2, expected);
1660        assert!(existing_pair_matches_pack_name(&packs, &expected).unwrap());
1661    }
1662
1663    #[test]
1664    fn concurrent_many_distinct_pack_installs() {
1665        use std::thread;
1666
1667        // Distinct pack_names take distinct locks — installs progress in parallel.
1668        let root = tempfile::tempdir().unwrap();
1669        let packs = root.path().join("packs");
1670        create_dir_all_durable(&packs).unwrap();
1671        let idx_bytes = empty_idx_bytes();
1672
1673        let mut handles = Vec::new();
1674        for i in 0..8u8 {
1675            let packs = packs.clone();
1676            let idx_bytes = idx_bytes.clone();
1677            handles.push(thread::spawn(move || {
1678                let pack_bytes = format!("many-pack-body-{i}").into_bytes();
1679                install_pack_bytes_journaled(&packs, &pack_bytes, &idx_bytes).unwrap()
1680            }));
1681        }
1682        let mut names = Vec::new();
1683        for h in handles {
1684            names.push(h.join().unwrap());
1685        }
1686        names.sort();
1687        names.dedup();
1688        assert_eq!(names.len(), 8, "expected 8 distinct pack names");
1689        for name in &names {
1690            assert!(existing_pair_matches_pack_name(&packs, name).unwrap());
1691        }
1692    }
1693
1694    #[test]
1695    fn far_future_created_unix_expires_immediately_under_ttl() {
1696        let root = tempfile::tempdir().unwrap();
1697        let packs = root.path().join("packs");
1698        create_dir_all_durable(&packs).unwrap();
1699        let install_id = "far-future";
1700        create_dir_all_durable(&staging_dir(&packs, install_id)).unwrap();
1701        write_file_atomic(&staging_pack_path(&packs, install_id), b"p").unwrap();
1702        write_file_atomic(&staging_idx_path(&packs, install_id), b"i").unwrap();
1703
1704        let mut intent = PackInstallIntent::new(install_id.into(), pack_id("aa"));
1705        // Beyond skew tolerance — must not dodge expiry.
1706        intent.created_unix = unix_now().saturating_add(INTENT_CLOCK_SKEW_TOLERANCE_SECS + 10_000);
1707        write_intent(&packs, &intent).unwrap();
1708
1709        let report = recover_pack_install_intents_with_ttl(&packs, Some(86_400)).unwrap();
1710        assert_eq!(report.aborted, 1, "far-future must expire: {report:?}");
1711        assert!(!intent_path(&packs, install_id).exists());
1712    }
1713
1714    #[test]
1715    fn mild_clock_skew_does_not_expire_fresh_intent() {
1716        let root = tempfile::tempdir().unwrap();
1717        let packs = root.path().join("packs");
1718        create_dir_all_durable(&packs).unwrap();
1719        let install_id = "mild-skew";
1720        create_dir_all_durable(&staging_dir(&packs, install_id)).unwrap();
1721        write_file_atomic(&staging_pack_path(&packs, install_id), b"p").unwrap();
1722        write_file_atomic(&staging_idx_path(&packs, install_id), b"i").unwrap();
1723
1724        let mut intent = PackInstallIntent::new(install_id.into(), pack_id("bb"));
1725        // Slightly ahead of wall clock (within tolerance) — still "in progress".
1726        intent.created_unix = unix_now().saturating_add(INTENT_CLOCK_SKEW_TOLERANCE_SECS / 2);
1727        write_intent(&packs, &intent).unwrap();
1728
1729        let report = recover_pack_install_intents_with_ttl(&packs, Some(86_400)).unwrap();
1730        assert_eq!(report.skipped_in_progress, 1, "mild skew: {report:?}");
1731        assert_eq!(report.aborted, 0);
1732        assert!(intent_path(&packs, install_id).exists());
1733    }
1734
1735    #[test]
1736    fn fault_after_intent_prepared_is_recoverable() {
1737        let root = tempfile::tempdir().unwrap();
1738        let packs = root.path().join("packs");
1739        create_dir_all_durable(&packs).unwrap();
1740
1741        let pack_bytes = b"fault-inject-pack-body";
1742        let idx_bytes = empty_idx_bytes();
1743        let expected = format!("{}", blake3::hash(pack_bytes).to_hex());
1744
1745        let before_err = pack_install_metrics_snapshot().installs_err;
1746        let err = fault_inject::with_fault_points(&["pack_install_after_intent_prepared"], || {
1747            install_pack_bytes_journaled(&packs, pack_bytes, &idx_bytes)
1748        })
1749        .expect_err("fault should fire");
1750        assert!(
1751            err.to_string().contains("pack_install_after_intent_prepared"),
1752            "err={err}"
1753        );
1754        // Process-global counters can race under parallel tests; assert non-decreasing delta.
1755        assert!(pack_install_metrics_snapshot().installs_err >= before_err);
1756
1757        // Staging + prepared intent should remain for recovery.
1758        assert_eq!(intent_count_json(&packs), 1);
1759        assert!(!dst_pack_path(&packs, &expected).exists());
1760
1761        // Force-expire prepared intent, abort, then reinstall succeeds.
1762        for entry in fs::read_dir(intent_root(&packs)).unwrap().flatten() {
1763            let p = entry.path();
1764            if p.extension().and_then(|x| x.to_str()) != Some("json") {
1765                continue;
1766            }
1767            let mut intent: PackInstallIntent =
1768                serde_json::from_slice(&fs::read(&p).unwrap()).unwrap();
1769            intent.created_unix = 1;
1770            write_intent(&packs, &intent).unwrap();
1771        }
1772        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1773        assert_eq!(report.aborted, 1, "report={report:?}");
1774        assert_eq!(intent_count_json(&packs), 0);
1775
1776        let name = install_pack_bytes_journaled(&packs, pack_bytes, &idx_bytes).unwrap();
1777        assert_eq!(name, expected);
1778        assert!(existing_pair_matches_pack_name(&packs, &expected).unwrap());
1779    }
1780
1781    #[test]
1782    fn fault_after_publish_pack_recovers_to_complete() {
1783        let root = tempfile::tempdir().unwrap();
1784        let packs = root.path().join("packs");
1785        create_dir_all_durable(&packs).unwrap();
1786
1787        let pack_bytes = b"fault-after-pack-publish-body";
1788        let idx_bytes = empty_idx_bytes();
1789        let expected = format!("{}", blake3::hash(pack_bytes).to_hex());
1790
1791        let err = fault_inject::with_fault_points(&["pack_install_after_publish_pack"], || {
1792            install_pack_bytes_journaled(&packs, pack_bytes, &idx_bytes)
1793        })
1794        .expect_err("fault should fire after pack publish");
1795        assert!(err.to_string().contains("pack_install_after_publish_pack"));
1796
1797        // Pack published, intent may still be prepared (fault is after publish, before
1798        // phase rewrite) or pack_published depending on checkpoint placement.
1799        assert!(dst_pack_path(&packs, &expected).exists());
1800        assert!(!dst_idx_path(&packs, &expected).exists());
1801
1802        let report = recover_pack_install_intents(&packs).unwrap();
1803        assert_eq!(report.completed, 1, "report={report:?}");
1804        assert!(dst_idx_path(&packs, &expected).exists());
1805        assert!(existing_pair_matches_pack_name(&packs, &expected).unwrap());
1806    }
1807
1808    fn intent_count_json(packs: &Path) -> usize {
1809        let dir = intent_root(packs);
1810        if !dir.exists() {
1811            return 0;
1812        }
1813        fs::read_dir(dir)
1814            .unwrap()
1815            .filter(|e| {
1816                e.as_ref()
1817                    .map(|e| e.path().extension().and_then(|x| x.to_str()) == Some("json"))
1818                    .unwrap_or(false)
1819            })
1820            .count()
1821    }
1822
1823    #[test]
1824    fn metrics_snapshot_tracks_install_and_recover() {
1825        // Process-global atomics: measure deltas so parallel tests don't flake.
1826        let before = pack_install_metrics_snapshot();
1827        let root = tempfile::tempdir().unwrap();
1828        let packs = root.path().join("packs");
1829        create_dir_all_durable(&packs).unwrap();
1830        let _ = install_pack_bytes_journaled(&packs, b"metrics-pack", &empty_idx_bytes()).unwrap();
1831        let after_install = pack_install_metrics_snapshot();
1832        assert!(after_install.installs_ok > before.installs_ok);
1833
1834        // Plant expired prepared → recover aborts.
1835        let install_id = "metrics-abort";
1836        create_dir_all_durable(&staging_dir(&packs, install_id)).unwrap();
1837        write_file_atomic(&staging_pack_path(&packs, install_id), b"p").unwrap();
1838        write_file_atomic(&staging_idx_path(&packs, install_id), b"i").unwrap();
1839        let mut intent = PackInstallIntent::new(install_id.into(), pack_id("cc"));
1840        intent.created_unix = 1;
1841        write_intent(&packs, &intent).unwrap();
1842        let before_abort = pack_install_metrics_snapshot();
1843        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1844        assert_eq!(report.aborted, 1);
1845        let after_abort = pack_install_metrics_snapshot();
1846        assert!(after_abort.recover_aborted > before_abort.recover_aborted);
1847    }
1848
1849    #[cfg(unix)]
1850    #[test]
1851    fn assert_under_packs_rejects_staging_symlink_escape() {
1852        use std::os::unix::fs::symlink;
1853
1854        let root = tempfile::tempdir().unwrap();
1855        let packs = root.path().join("packs");
1856        create_dir_all_durable(&packs).unwrap();
1857        let outside = root.path().join("outside");
1858        create_dir_all_durable(&outside).unwrap();
1859
1860        // Lexically under packs, but .staging is a symlink out.
1861        symlink(&outside, packs.join(STAGING_DIR_NAME)).unwrap();
1862
1863        let stage = staging_dir(&packs, "id1");
1864        let err = assert_under_packs(&packs, &stage).unwrap_err();
1865        assert!(
1866            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1867            "err={err}"
1868        );
1869
1870        // Journal layout guard must fail before install.
1871        let err = ensure_journal_layout_safe(&packs).unwrap_err();
1872        assert!(
1873            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1874            "err={err}"
1875        );
1876        let err = install_pack_bytes_journaled(&packs, b"x", &empty_idx_bytes()).unwrap_err();
1877        assert!(
1878            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1879            "err={err}"
1880        );
1881    }
1882
1883    #[cfg(unix)]
1884    #[test]
1885    fn assert_under_packs_rejects_intent_root_symlink_escape() {
1886        use std::os::unix::fs::symlink;
1887
1888        let root = tempfile::tempdir().unwrap();
1889        let packs = root.path().join("packs");
1890        create_dir_all_durable(&packs).unwrap();
1891        let outside = root.path().join("outside-intent");
1892        create_dir_all_durable(&outside).unwrap();
1893        symlink(&outside, packs.join(INTENT_DIR_NAME)).unwrap();
1894
1895        let err = ensure_journal_layout_safe(&packs).unwrap_err();
1896        assert!(
1897            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1898            "err={err}"
1899        );
1900    }
1901
1902    #[cfg(unix)]
1903    #[test]
1904    fn assert_under_packs_rejects_pack_locks_symlink_escape() {
1905        use std::os::unix::fs::symlink;
1906
1907        let root = tempfile::tempdir().unwrap();
1908        let packs = root.path().join("packs");
1909        create_dir_all_durable(&packs).unwrap();
1910        let outside = root.path().join("outside-locks");
1911        create_dir_all_durable(&outside).unwrap();
1912        symlink(&outside, packs.join(PACK_LOCKS_DIR_NAME)).unwrap();
1913
1914        let err = ensure_journal_layout_safe(&packs).unwrap_err();
1915        assert!(
1916            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1917            "err={err}"
1918        );
1919    }
1920
1921    #[cfg(unix)]
1922    #[test]
1923    fn assert_under_packs_rejects_destination_file_symlink_escape() {
1924        use std::os::unix::fs::symlink;
1925
1926        let root = tempfile::tempdir().unwrap();
1927        let packs = root.path().join("packs");
1928        create_dir_all_durable(&packs).unwrap();
1929        let outside = root.path().join("evil.pack");
1930        fs::write(&outside, b"evil").unwrap();
1931
1932        let name = pack_id("symlink-dst");
1933        let dst = dst_pack_path(&packs, &name);
1934        symlink(&outside, &dst).unwrap();
1935
1936        let err = assert_under_packs(&packs, &dst).unwrap_err();
1937        assert!(
1938            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1939            "err={err}"
1940        );
1941        // existing_pair must refuse a symlink-out destination (error or false).
1942        let pair = existing_pair_matches_pack_name(&packs, &name);
1943        assert!(
1944            pair.as_ref().map(|v| !v).unwrap_or(true),
1945            "pair={pair:?}"
1946        );
1947    }
1948
1949    #[cfg(unix)]
1950    #[test]
1951    fn assert_under_packs_rejects_install_id_staging_symlink() {
1952        use std::os::unix::fs::symlink;
1953
1954        let root = tempfile::tempdir().unwrap();
1955        let packs = root.path().join("packs");
1956        create_dir_all_durable(&packs).unwrap();
1957        create_dir_all_durable(&staging_root(&packs)).unwrap();
1958        let outside = root.path().join("outside-stage-id");
1959        create_dir_all_durable(&outside).unwrap();
1960        let install_id = "symlink-stage-id";
1961        symlink(&outside, staging_dir(&packs, install_id)).unwrap();
1962
1963        let pack_path = staging_pack_path(&packs, install_id);
1964        let err = assert_under_packs(&packs, &pack_path).unwrap_err();
1965        assert!(
1966            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1967            "err={err}"
1968        );
1969    }
1970
1971    #[test]
1972    fn assert_under_packs_accepts_normal_reconstructed_paths() {
1973        let root = tempfile::tempdir().unwrap();
1974        let packs = root.path().join("packs");
1975        create_dir_all_durable(&packs).unwrap();
1976        let name = pack_id("normal");
1977        let install_id = "normal-id";
1978        assert_under_packs(&packs, &dst_pack_path(&packs, &name)).unwrap();
1979        assert_under_packs(&packs, &staging_pack_path(&packs, install_id)).unwrap();
1980        assert_under_packs(&packs, &intent_path(&packs, install_id)).unwrap();
1981    }
1982}