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()
180        .map_err(|e| io::Error::other(e.to_string()))
181}
182
183/// Short global listing lock (intent dir scan only).
184pub(crate) fn acquire_pack_install_lock(
185    packs_dir: &Path,
186) -> io::Result<crate::lock::WriteLockGuard> {
187    create_dir_all_durable(packs_dir)?;
188    let lock = RepoLock::at(packs_dir.join(PACK_INSTALL_LOCK_NAME));
189    lock.write().map_err(|e| io::Error::other(e.to_string()))
190}
191
192impl PackInstallIntent {
193    pub fn new(install_id: String, pack_name: String) -> Self {
194        Self {
195            version: PACK_INSTALL_INTENT_VERSION,
196            install_id,
197            pack_name,
198            phase: PackInstallPhase::Prepared,
199            created_unix: unix_now(),
200        }
201    }
202}
203
204// ---------------------------------------------------------------------------
205// Path reconstruction (trusted packs_dir only)
206// ---------------------------------------------------------------------------
207
208fn unix_now() -> i64 {
209    SystemTime::now()
210        .duration_since(UNIX_EPOCH)
211        .map(|d| d.as_secs() as i64)
212        .unwrap_or(0)
213}
214
215/// Single path component: no separators, no `..`, no empty, limited charset.
216pub(crate) fn validate_install_id(id: &str) -> io::Result<()> {
217    if id.is_empty() || id.len() > 128 {
218        return Err(io::Error::new(
219            io::ErrorKind::InvalidInput,
220            "invalid install_id length",
221        ));
222    }
223    if !id
224        .chars()
225        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
226    {
227        return Err(io::Error::new(
228            io::ErrorKind::InvalidInput,
229            "install_id contains illegal characters",
230        ));
231    }
232    if id == "." || id == ".." {
233        return Err(io::Error::new(
234            io::ErrorKind::InvalidInput,
235            "install_id must not be . or ..",
236        ));
237    }
238    Ok(())
239}
240
241/// Content-addressed pack stem: exactly 64 lowercase hex digits (BLAKE3).
242pub(crate) fn validate_pack_name(name: &str) -> io::Result<()> {
243    pack_name_to_digest(name).map(|_| ())
244}
245
246/// Decode a validated `pack_name` to the native 32-byte BLAKE3 digest.
247///
248/// Prefer comparing digests over hex strings once past the FS/JSON boundary.
249pub(crate) fn pack_name_to_digest(name: &str) -> io::Result<[u8; 32]> {
250    if name.len() != 64 {
251        return Err(io::Error::new(
252            io::ErrorKind::InvalidInput,
253            "pack_name must be exactly 64 lowercase hex digits (BLAKE3)",
254        ));
255    }
256    if !name
257        .bytes()
258        .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
259    {
260        return Err(io::Error::new(
261            io::ErrorKind::InvalidInput,
262            "pack_name must be lowercase hexadecimal",
263        ));
264    }
265    let mut digest = [0u8; 32];
266    // decode_to_slice avoids an intermediate Vec; name is already length-checked.
267    hex::decode_to_slice(name.as_bytes(), &mut digest).map_err(|e| {
268        io::Error::new(
269            io::ErrorKind::InvalidInput,
270            format!("pack_name hex decode failed: {e}"),
271        )
272    })?;
273    Ok(digest)
274}
275
276/// Hex form of a BLAKE3 digest for filenames / intent JSON (one allocation).
277fn digest_to_pack_name(digest: &[u8; 32]) -> String {
278    // blake3::Hash::to_hex is stack; to_string once at the FS boundary.
279    blake3::Hash::from_bytes(*digest).to_hex().to_string()
280}
281
282fn validate_intent_ids(intent: &PackInstallIntent) -> io::Result<()> {
283    validate_install_id(&intent.install_id)?;
284    pack_name_to_digest(&intent.pack_name)?;
285    Ok(())
286}
287
288pub(crate) fn staging_root(packs_dir: &Path) -> PathBuf {
289    packs_dir.join(STAGING_DIR_NAME)
290}
291
292pub(crate) fn intent_root(packs_dir: &Path) -> PathBuf {
293    packs_dir.join(INTENT_DIR_NAME)
294}
295
296fn quarantine_root(packs_dir: &Path) -> PathBuf {
297    intent_root(packs_dir).join(QUARANTINE_DIR_NAME)
298}
299
300pub(crate) fn intent_path(packs_dir: &Path, install_id: &str) -> PathBuf {
301    intent_root(packs_dir).join(format!("{install_id}.json"))
302}
303
304pub(crate) fn staging_dir(packs_dir: &Path, install_id: &str) -> PathBuf {
305    staging_root(packs_dir).join(install_id)
306}
307
308fn staging_pack_path(packs_dir: &Path, install_id: &str) -> PathBuf {
309    staging_dir(packs_dir, install_id).join(STAGED_PACK_NAME)
310}
311
312fn staging_idx_path(packs_dir: &Path, install_id: &str) -> PathBuf {
313    staging_dir(packs_dir, install_id).join(STAGED_IDX_NAME)
314}
315
316fn dst_pack_path(packs_dir: &Path, pack_name: &str) -> PathBuf {
317    packs_dir.join(format!("{pack_name}.pack"))
318}
319
320fn dst_idx_path(packs_dir: &Path, pack_name: &str) -> PathBuf {
321    packs_dir.join(format!("{pack_name}.idx"))
322}
323
324/// True when `path` is `root` or a descendant (component-wise).
325fn path_is_within(path: &Path, root: &Path) -> bool {
326    path == root || path.starts_with(root)
327}
328
329/// Ensure `candidate` cannot resolve outside `packs_dir`.
330///
331/// **Canonical containment is authoritative.** A lexical
332/// `candidate.starts_with(packs_dir)` must never override a canonical
333/// escape (classic case: `.staging` → symlink to `/tmp/evil`).
334///
335/// Walks every existing path component (including intermediate symlinks)
336/// and rejects any prefix whose `canonicalize` leaves `packs_dir`.
337/// Nonexistent trailing components are allowed only under a trusted base.
338fn assert_under_packs(packs_dir: &Path, candidate: &Path) -> io::Result<()> {
339    if !packs_dir.exists() {
340        return Err(io::Error::new(
341            io::ErrorKind::NotFound,
342            "packs_dir does not exist for path containment check",
343        ));
344    }
345    let packs_canon = fs::canonicalize(packs_dir)?;
346
347    let rel = candidate.strip_prefix(packs_dir).or_else(|_| {
348        candidate
349            .strip_prefix(&packs_canon)
350            .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "path is not under packs_dir"))
351    })?;
352
353    for c in rel.components() {
354        match c {
355            Component::Normal(_) | Component::CurDir => {}
356            Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
357                return Err(io::Error::new(
358                    io::ErrorKind::InvalidInput,
359                    "path escapes packs_dir",
360                ));
361            }
362        }
363    }
364
365    let mut cur = packs_canon.clone();
366    for c in rel.components() {
367        let Component::Normal(name) = c else {
368            continue;
369        };
370        cur.push(name);
371        // symlink_metadata succeeds for files, dirs, and (possibly broken) symlinks.
372        match cur.symlink_metadata() {
373            Ok(_) => {
374                let canon = fs::canonicalize(&cur).map_err(|e| {
375                    io::Error::new(
376                        e.kind(),
377                        format!(
378                            "cannot resolve path under packs_dir ({}): {e}",
379                            cur.display()
380                        ),
381                    )
382                })?;
383                if !path_is_within(&canon, &packs_canon) {
384                    return Err(io::Error::new(
385                        io::ErrorKind::InvalidInput,
386                        "reconstructed path escapes packs_dir via symlink or mount",
387                    ));
388                }
389                // Continue from resolved location so nested escapes are still caught.
390                cur = canon;
391            }
392            Err(e) if e.kind() == io::ErrorKind::NotFound => {
393                // Remaining names are pure (already validated as Normal components).
394                break;
395            }
396            Err(e) => return Err(e),
397        }
398    }
399    Ok(())
400}
401
402/// Reject hostile journal roots (`.staging`, `.install-intent`, `.pack-locks`
403/// as symlinks/mounts that escape `packs_dir`).
404fn ensure_journal_layout_safe(packs_dir: &Path) -> io::Result<()> {
405    create_dir_all_durable(packs_dir)?;
406    for name in [STAGING_DIR_NAME, INTENT_DIR_NAME, PACK_LOCKS_DIR_NAME] {
407        let p = packs_dir.join(name);
408        if p.symlink_metadata().is_ok() {
409            assert_under_packs(packs_dir, &p)?;
410        }
411    }
412    Ok(())
413}
414
415// ---------------------------------------------------------------------------
416// Intent I/O
417// ---------------------------------------------------------------------------
418
419pub(crate) fn write_intent(packs_dir: &Path, intent: &PackInstallIntent) -> io::Result<()> {
420    validate_intent_ids(intent)?;
421    create_dir_all_durable(&intent_root(packs_dir))?;
422    let path = intent_path(packs_dir, &intent.install_id);
423    let bytes = serde_json::to_vec_pretty(intent)
424        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
425    write_file_atomic(&path, &bytes)
426}
427
428pub(crate) fn load_intent(path: &Path) -> io::Result<PackInstallIntent> {
429    let bytes = fs::read(path)?;
430    serde_json::from_slice(&bytes).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
431}
432
433pub(crate) fn remove_intent(packs_dir: &Path, install_id: &str) -> io::Result<()> {
434    validate_install_id(install_id)?;
435    let path = intent_path(packs_dir, install_id);
436    match fs::remove_file(&path) {
437        Ok(()) => {}
438        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
439        Err(e) => return Err(e),
440    }
441    // Make intent-dirent removal more durable (Codex #6).
442    let _ = sync_directory(&intent_root(packs_dir));
443    Ok(())
444}
445
446fn quarantine_intent_file(packs_dir: &Path, path: &Path) -> io::Result<()> {
447    let qroot = quarantine_root(packs_dir);
448    create_dir_all_durable(&qroot)?;
449    let name = path
450        .file_name()
451        .and_then(|n| n.to_str())
452        .unwrap_or("intent.json");
453    let dest = qroot.join(format!("{name}.{}.bad", unix_now()));
454    match fs::rename(path, &dest) {
455        Ok(()) => {
456            let _ = sync_directory(&qroot);
457            tracing::warn!(
458                from = %path.display(),
459                to = %dest.display(),
460                "quarantined unreadable pack install intent"
461            );
462            Ok(())
463        }
464        Err(e) => {
465            // If rename fails, leave original in place (do not delete).
466            Err(e)
467        }
468    }
469}
470
471fn remove_path_best_effort(path: &Path) {
472    if path.is_dir() {
473        let _ = fs::remove_dir_all(path);
474    } else {
475        let _ = fs::remove_file(path);
476    }
477}
478
479pub(crate) fn remove_staging(packs_dir: &Path, install_id: &str) {
480    if validate_install_id(install_id).is_err() {
481        return;
482    }
483    remove_path_best_effort(&staging_dir(packs_dir, install_id));
484}
485
486// ---------------------------------------------------------------------------
487// Abort / complete (paths always from packs_dir + ids)
488// ---------------------------------------------------------------------------
489
490pub(crate) fn abort_install(packs_dir: &Path, intent: &PackInstallIntent) -> io::Result<()> {
491    validate_intent_ids(intent)?;
492    let dst_pack = dst_pack_path(packs_dir, &intent.pack_name);
493    let dst_idx = dst_idx_path(packs_dir, &intent.pack_name);
494    assert_under_packs(packs_dir, &dst_pack)?;
495    assert_under_packs(packs_dir, &dst_idx)?;
496
497    // Only remove final pack if index is missing (partial publish).
498    if dst_pack.exists() && !dst_idx.exists() {
499        let _ = fs::remove_file(&dst_pack);
500        if let Some(parent) = dst_pack.parent() {
501            let _ = sync_directory(parent);
502        }
503    }
504    remove_staging(packs_dir, &intent.install_id);
505    remove_intent(packs_dir, &intent.install_id)?;
506    Ok(())
507}
508
509pub(crate) fn complete_from_staging(
510    packs_dir: &Path,
511    intent: &PackInstallIntent,
512) -> io::Result<()> {
513    validate_intent_ids(intent)?;
514    let staging_idx = staging_idx_path(packs_dir, &intent.install_id);
515    let dst_idx = dst_idx_path(packs_dir, &intent.pack_name);
516    let dst_pack = dst_pack_path(packs_dir, &intent.pack_name);
517    assert_under_packs(packs_dir, &staging_idx)?;
518    assert_under_packs(packs_dir, &dst_idx)?;
519    assert_under_packs(packs_dir, &dst_pack)?;
520
521    if dst_pack.exists() && dst_idx.exists() {
522        remove_staging(packs_dir, &intent.install_id);
523        remove_intent(packs_dir, &intent.install_id)?;
524        return Ok(());
525    }
526
527    if !dst_pack.exists() || !staging_idx.exists() {
528        return abort_install(packs_dir, intent);
529    }
530
531    publish_file_durable(&staging_idx, &dst_idx)?;
532    remove_staging(packs_dir, &intent.install_id);
533    remove_intent(packs_dir, &intent.install_id)?;
534    Ok(())
535}
536
537fn can_complete_quickly(packs_dir: &Path, intent: &PackInstallIntent) -> bool {
538    let dst_pack = dst_pack_path(packs_dir, &intent.pack_name);
539    let dst_idx = dst_idx_path(packs_dir, &intent.pack_name);
540    let staging_idx = staging_idx_path(packs_dir, &intent.install_id);
541    if dst_pack.exists() && dst_idx.exists() {
542        return true;
543    }
544    dst_pack.exists() && !dst_idx.exists() && staging_idx.exists()
545}
546
547/// Effective creation time for TTL when `created_unix` is only slightly ahead
548/// of wall time (within [`INTENT_CLOCK_SKEW_TOLERANCE_SECS`]).
549fn effective_created_unix(created_unix: i64, now: i64) -> i64 {
550    if created_unix > now {
551        now
552    } else {
553        created_unix
554    }
555}
556
557fn intent_expired(intent: &PackInstallIntent, ttl_secs: Option<i64>, now: i64) -> bool {
558    match ttl_secs {
559        Some(ttl) if ttl >= 0 => {
560            // Far-future / large clock rollback: expire immediately so a forged
561            // `created_unix = i64::MAX` (or a multi-minute clock jump) cannot
562            // keep an intent alive forever. Mild skew is clamped to `now`.
563            if intent.created_unix > now.saturating_add(INTENT_CLOCK_SKEW_TOLERANCE_SECS) {
564                return true;
565            }
566            let created = effective_created_unix(intent.created_unix, now);
567            created.saturating_add(ttl) < now
568        }
569        _ => false,
570    }
571}
572
573// ---------------------------------------------------------------------------
574// Existing pair identity validation
575// ---------------------------------------------------------------------------
576
577/// Stream-hash a pack file to a native 32-byte BLAKE3 digest (no hex).
578fn hash_file_blake3(path: &Path) -> io::Result<[u8; 32]> {
579    let mut file = File::open(path)?;
580    let mut hasher = blake3::Hasher::new();
581    let mut buf = vec![0u8; 64 * 1024];
582    loop {
583        let n = file.read(&mut buf)?;
584        if n == 0 {
585            break;
586        }
587        hasher.update(&buf[..n]);
588    }
589    Ok(*hasher.finalize().as_bytes())
590}
591
592/// True when a final pack+idx pair is safe to treat as already installed:
593/// pack content BLAKE3 equals `pack_name`, and the index **parses** as a
594/// [`crate::store::pack::PackIndex`].
595///
596/// Identity is checked as **native digests** (`[u8; 32]`), not hex strings.
597/// Hex is only the durable/public form of `pack_name` on disk.
598///
599/// This does **not** prove every index offset points at a live object in the
600/// pack (that is the pack reader's job on first use). It rejects empty,
601/// garbage, and structurally invalid indexes so install idempotency cannot
602/// accept a corrupt pair.
603/// String-name entry for tests and external call sites that only have hex.
604#[cfg(test)]
605fn existing_pair_matches_pack_name(packs_dir: &Path, pack_name: &str) -> io::Result<bool> {
606    let expected = pack_name_to_digest(pack_name)?;
607    existing_pair_matches_digest(packs_dir, pack_name, &expected)
608}
609
610/// Like [`existing_pair_matches_pack_name`], but reuses an already-decoded digest
611/// (install hot path: hash once in memory, compare file digest to those bytes).
612fn existing_pair_matches_digest(
613    packs_dir: &Path,
614    pack_name: &str,
615    expected: &[u8; 32],
616) -> io::Result<bool> {
617    let pack = dst_pack_path(packs_dir, pack_name);
618    let idx = dst_idx_path(packs_dir, pack_name);
619    assert_under_packs(packs_dir, &pack)?;
620    assert_under_packs(packs_dir, &idx)?;
621    if !pack.exists() || !idx.exists() {
622        return Ok(false);
623    }
624    // Symlink destinations must stay under packs (assert_under_packs) and the
625    // open path must be a regular file for install identity.
626    if !pack.is_file() || !idx.is_file() {
627        return Ok(false);
628    }
629    if idx.metadata()?.len() == 0 {
630        return Ok(false);
631    }
632    let actual = hash_file_blake3(&pack)?;
633    if actual != *expected {
634        return Ok(false);
635    }
636    let idx_bytes = fs::read(&idx)?;
637    match crate::store::pack::PackIndex::from_bytes(&idx_bytes) {
638        Ok(_) => Ok(true),
639        Err(_) => Ok(false),
640    }
641}
642
643// ---------------------------------------------------------------------------
644// Recovery
645// ---------------------------------------------------------------------------
646
647pub fn recover_pack_install_intents(packs_dir: &Path) -> io::Result<PackInstallRecoverReport> {
648    recover_pack_install_intents_with_ttl(packs_dir, Some(DEFAULT_PACK_INSTALL_INTENT_TTL_SECS))
649}
650
651pub fn recover_pack_install_intents_with_ttl(
652    packs_dir: &Path,
653    ttl_secs: Option<i64>,
654) -> io::Result<PackInstallRecoverReport> {
655    // Reject hostile journal roots (symlink escape) before any mutation.
656    ensure_journal_layout_safe(packs_dir)?;
657
658    // Snapshot intent paths under a short global listing lock, then recover
659    // each pack under its per-pack lock (try_lock → skip if install holds it).
660    let intent_paths: Vec<PathBuf> = {
661        let _list_guard = acquire_pack_install_lock(packs_dir)?;
662        let intent_dir = intent_root(packs_dir);
663        if !intent_dir.exists() {
664            let mut report = PackInstallRecoverReport::default();
665            let now = unix_now();
666            sweep_orphan_staging(packs_dir, ttl_secs, now, &mut report);
667            return Ok(report);
668        }
669        fs::read_dir(&intent_dir)?
670            .filter_map(|e| e.ok())
671            .map(|e| e.path())
672            .filter(|p| !p.is_dir() && p.extension().and_then(|x| x.to_str()) == Some("json"))
673            .collect()
674    };
675
676    let mut report = PackInstallRecoverReport::default();
677    let now = unix_now();
678
679    for path in intent_paths {
680        report.intents_seen += 1;
681
682        let intent = match load_intent(&path) {
683            Ok(i)
684                if i.version == PACK_INSTALL_INTENT_VERSION
685                    || i.version == 1 /* v1 ids still usable; paths ignored */ =>
686            {
687                match validate_intent_ids(&i) {
688                    Ok(()) => i,
689                    Err(_) => {
690                        let _ = quarantine_intent_file(packs_dir, &path);
691                        report.quarantined += 1;
692                        metric_inc(&METRIC_RECOVER_QUARANTINED);
693                        continue;
694                    }
695                }
696            }
697            Ok(_) | Err(_) => {
698                match quarantine_intent_file(packs_dir, &path) {
699                    Ok(()) => {
700                        report.quarantined += 1;
701                        metric_inc(&METRIC_RECOVER_QUARANTINED);
702                    }
703                    Err(_) => report.errors += 1,
704                }
705                continue;
706            }
707        };
708
709        // Per-pack try-lock: if a live install holds it, skip (in progress).
710        let pack_guard = match try_acquire_pack_name_lock(packs_dir, &intent.pack_name)? {
711            Some(g) => g,
712            None => {
713                report.skipped_in_progress += 1;
714                metric_inc(&METRIC_RECOVER_SKIPPED);
715                continue;
716            }
717        };
718
719        let expired = intent_expired(&intent, ttl_secs, now);
720        let before_aborted = report.aborted;
721        let before_completed = report.completed;
722        let before_skipped = report.skipped_in_progress;
723        if recover_one_intent(packs_dir, &intent, expired, &mut report).is_err() {
724            report.errors += 1;
725        }
726        if report.completed > before_completed {
727            metric_inc(&METRIC_RECOVER_COMPLETED);
728        }
729        if report.aborted > before_aborted {
730            metric_inc(&METRIC_RECOVER_ABORTED);
731        }
732        if report.skipped_in_progress > before_skipped {
733            metric_inc(&METRIC_RECOVER_SKIPPED);
734        }
735        drop(pack_guard);
736    }
737
738    sweep_orphan_staging(packs_dir, ttl_secs, now, &mut report);
739
740    if report.intents_seen > 0
741        || report.orphan_staging_swept > 0
742        || report.errors > 0
743        || report.completed > 0
744        || report.aborted > 0
745        || report.skipped_in_progress > 0
746        || report.quarantined > 0
747    {
748        tracing::info!(
749            ?packs_dir,
750            intents_seen = report.intents_seen,
751            completed = report.completed,
752            aborted = report.aborted,
753            skipped_in_progress = report.skipped_in_progress,
754            cleaned_stale_completed = report.cleaned_stale_completed,
755            orphan_staging_swept = report.orphan_staging_swept,
756            quarantined = report.quarantined,
757            errors = report.errors,
758            metrics = ?pack_install_metrics_snapshot(),
759            "pack install journal recovery"
760        );
761    } else {
762        tracing::debug!(?packs_dir, "pack install journal recovery: nothing to do");
763    }
764
765    Ok(report)
766}
767
768fn recover_one_intent(
769    packs_dir: &Path,
770    intent: &PackInstallIntent,
771    expired: bool,
772    report: &mut PackInstallRecoverReport,
773) -> io::Result<()> {
774    if can_complete_quickly(packs_dir, intent) {
775        return match intent.phase {
776            PackInstallPhase::Prepared | PackInstallPhase::PackPublished => {
777                let dst_pack = dst_pack_path(packs_dir, &intent.pack_name);
778                let dst_idx = dst_idx_path(packs_dir, &intent.pack_name);
779                if dst_pack.exists() && dst_idx.exists() {
780                    remove_staging(packs_dir, &intent.install_id);
781                    remove_intent(packs_dir, &intent.install_id)?;
782                    report.cleaned_stale_completed += 1;
783                    Ok(())
784                } else {
785                    complete_from_staging(packs_dir, intent)?;
786                    if dst_pack_path(packs_dir, &intent.pack_name).exists()
787                        && dst_idx_path(packs_dir, &intent.pack_name).exists()
788                    {
789                        report.completed += 1;
790                    } else {
791                        report.aborted += 1;
792                    }
793                    Ok(())
794                }
795            }
796            PackInstallPhase::Completed => {
797                remove_staging(packs_dir, &intent.install_id);
798                remove_intent(packs_dir, &intent.install_id)?;
799                report.cleaned_stale_completed += 1;
800                Ok(())
801            }
802        };
803    }
804
805    if expired {
806        tracing::debug!(
807            install_id = %intent.install_id,
808            pack_name = %intent.pack_name,
809            phase = ?intent.phase,
810            "aborting expired pack install intent"
811        );
812        abort_install(packs_dir, intent)?;
813        report.aborted += 1;
814        return Ok(());
815    }
816
817    match intent.phase {
818        PackInstallPhase::Prepared | PackInstallPhase::PackPublished => {
819            report.skipped_in_progress += 1;
820            Ok(())
821        }
822        PackInstallPhase::Completed => {
823            remove_staging(packs_dir, &intent.install_id);
824            remove_intent(packs_dir, &intent.install_id)?;
825            report.cleaned_stale_completed += 1;
826            Ok(())
827        }
828    }
829}
830
831fn sweep_orphan_staging(
832    packs_dir: &Path,
833    ttl_secs: Option<i64>,
834    now: i64,
835    report: &mut PackInstallRecoverReport,
836) {
837    let Some(ttl) = ttl_secs.filter(|t| *t >= 0) else {
838        return;
839    };
840    let staging = staging_root(packs_dir);
841    let entries = match fs::read_dir(&staging) {
842        Ok(e) => e,
843        Err(_) => return,
844    };
845    for entry in entries.flatten() {
846        let path = entry.path();
847        if !path.is_dir() {
848            continue;
849        }
850        let Some(id) = path.file_name().and_then(|n| n.to_str()) else {
851            continue;
852        };
853        if validate_install_id(id).is_err() {
854            continue;
855        }
856        if intent_path(packs_dir, id).exists() {
857            continue;
858        }
859        let age_ok_to_sweep = path_mtime_unix(&path)
860            .map(|mtime| mtime.saturating_add(ttl) < now)
861            .unwrap_or(true);
862        if !age_ok_to_sweep {
863            continue;
864        }
865        remove_path_best_effort(&path);
866        report.orphan_staging_swept += 1;
867    }
868}
869
870fn path_mtime_unix(path: &Path) -> Option<i64> {
871    let meta = fs::metadata(path).ok()?;
872    let modified = meta.modified().ok()?;
873    Some(
874        modified
875            .duration_since(UNIX_EPOCH)
876            .map(|d| d.as_secs() as i64)
877            .unwrap_or(0),
878    )
879}
880
881// ---------------------------------------------------------------------------
882// Install
883// ---------------------------------------------------------------------------
884
885fn new_install_id() -> String {
886    let t = unix_now() as u64;
887    let r: u64 = rand::random();
888    format!("{t:016x}-{r:016x}")
889}
890
891/// Journaled in-memory install. Returns content-addressed `pack_name`.
892pub fn install_pack_bytes_journaled(
893    packs_dir: &Path,
894    pack_data: &[u8],
895    index_data: &[u8],
896) -> io::Result<String> {
897    match install_pack_bytes_journaled_inner(packs_dir, pack_data, index_data) {
898        Ok(name) => {
899            metric_inc(&METRIC_INSTALLS_OK);
900            Ok(name)
901        }
902        Err(e) => {
903            metric_inc(&METRIC_INSTALLS_ERR);
904            Err(e)
905        }
906    }
907}
908
909fn install_pack_bytes_journaled_inner(
910    packs_dir: &Path,
911    pack_data: &[u8],
912    index_data: &[u8],
913) -> io::Result<String> {
914    ensure_journal_layout_safe(packs_dir)?;
915    // Hash once as native bytes; hex only for the FS/JSON name boundary.
916    let digest = *blake3::hash(pack_data).as_bytes();
917    let pack_name = digest_to_pack_name(&digest);
918
919    if existing_pair_matches_digest(packs_dir, &pack_name, &digest)? {
920        return Ok(pack_name);
921    }
922
923    // Stage outside per-pack lock (unique install_id).
924    let install_id = new_install_id();
925    validate_install_id(&install_id)?;
926    let stage = staging_dir(packs_dir, &install_id);
927    assert_under_packs(packs_dir, &stage)?;
928    create_dir_all_durable(&stage)?;
929    let staging_pack = staging_pack_path(packs_dir, &install_id);
930    let staging_idx = staging_idx_path(packs_dir, &install_id);
931    assert_under_packs(packs_dir, &staging_pack)?;
932    assert_under_packs(packs_dir, &staging_idx)?;
933    write_file_atomic(&staging_pack, pack_data)?;
934    fault_inject::maybe_fail_at("pack_install_after_stage_pack")?;
935    write_file_atomic(&staging_idx, index_data)?;
936    fault_inject::maybe_fail_at("pack_install_after_stage_idx")?;
937
938    // Per-pack lock for intent + publish (other pack names stay parallel).
939    let _guard = acquire_pack_name_lock(packs_dir, &pack_name)?;
940    fault_inject::maybe_fail_at("pack_install_after_pack_lock")?;
941
942    if existing_pair_matches_digest(packs_dir, &pack_name, &digest)? {
943        remove_staging(packs_dir, &install_id);
944        return Ok(pack_name);
945    }
946    let dst_pack = dst_pack_path(packs_dir, &pack_name);
947    let dst_idx = dst_idx_path(packs_dir, &pack_name);
948    assert_under_packs(packs_dir, &dst_pack)?;
949    assert_under_packs(packs_dir, &dst_idx)?;
950    if dst_pack.exists() && !dst_idx.exists() {
951        let _ = fs::remove_file(&dst_pack);
952    }
953    if dst_pack.exists() && dst_idx.exists() {
954        if !existing_pair_matches_digest(packs_dir, &pack_name, &digest)? {
955            let _ = fs::remove_file(&dst_pack);
956            let _ = fs::remove_file(&dst_idx);
957        } else {
958            remove_staging(packs_dir, &install_id);
959            return Ok(pack_name);
960        }
961    }
962
963    let mut intent = PackInstallIntent::new(install_id.clone(), pack_name.clone());
964    write_intent(packs_dir, &intent)?;
965    fault_inject::maybe_fail_at("pack_install_after_intent_prepared")?;
966
967    publish_file_durable(&staging_pack, &dst_pack)?;
968    fault_inject::maybe_fail_at("pack_install_after_publish_pack")?;
969    intent.phase = PackInstallPhase::PackPublished;
970    write_intent(packs_dir, &intent)?;
971    fault_inject::maybe_fail_at("pack_install_after_intent_pack_published")?;
972
973    publish_file_durable(&staging_idx, &dst_idx)?;
974    fault_inject::maybe_fail_at("pack_install_after_publish_idx")?;
975    remove_staging(packs_dir, &install_id);
976    remove_intent(packs_dir, &install_id)?;
977    fault_inject::maybe_fail_at("pack_install_after_intent_removed")?;
978    Ok(pack_name)
979}
980
981/// Journaled streaming install (consumes source pack/index paths).
982pub fn install_pack_files_journaled(
983    packs_dir: &Path,
984    src_pack_path: &Path,
985    src_index_path: &Path,
986    pack_name: &str,
987) -> io::Result<()> {
988    match install_pack_files_journaled_inner(packs_dir, src_pack_path, src_index_path, pack_name) {
989        Ok(()) => {
990            metric_inc(&METRIC_INSTALLS_OK);
991            Ok(())
992        }
993        Err(e) => {
994            metric_inc(&METRIC_INSTALLS_ERR);
995            Err(e)
996        }
997    }
998}
999
1000fn install_pack_files_journaled_inner(
1001    packs_dir: &Path,
1002    src_pack_path: &Path,
1003    src_index_path: &Path,
1004    pack_name: &str,
1005) -> io::Result<()> {
1006    // Decode once; identity checks stay on native digests.
1007    let expected = pack_name_to_digest(pack_name)?;
1008    ensure_journal_layout_safe(packs_dir)?;
1009
1010    if existing_pair_matches_digest(packs_dir, pack_name, &expected)? {
1011        let _ = fs::remove_file(src_pack_path);
1012        let _ = fs::remove_file(src_index_path);
1013        return Ok(());
1014    }
1015
1016    let install_id = new_install_id();
1017    validate_install_id(&install_id)?;
1018    let stage = staging_dir(packs_dir, &install_id);
1019    assert_under_packs(packs_dir, &stage)?;
1020    create_dir_all_durable(&stage)?;
1021    let staging_pack = staging_pack_path(packs_dir, &install_id);
1022    let staging_idx = staging_idx_path(packs_dir, &install_id);
1023    assert_under_packs(packs_dir, &staging_pack)?;
1024    assert_under_packs(packs_dir, &staging_idx)?;
1025    publish_file_durable(src_pack_path, &staging_pack)?;
1026    fault_inject::maybe_fail_at("pack_install_stream_after_stage_pack")?;
1027    publish_file_durable(src_index_path, &staging_idx)?;
1028    fault_inject::maybe_fail_at("pack_install_stream_after_stage_idx")?;
1029
1030    let _guard = acquire_pack_name_lock(packs_dir, pack_name)?;
1031    if existing_pair_matches_digest(packs_dir, pack_name, &expected)? {
1032        remove_staging(packs_dir, &install_id);
1033        return Ok(());
1034    }
1035    let dst_pack = dst_pack_path(packs_dir, pack_name);
1036    let dst_idx = dst_idx_path(packs_dir, pack_name);
1037    assert_under_packs(packs_dir, &dst_pack)?;
1038    assert_under_packs(packs_dir, &dst_idx)?;
1039    if dst_pack.exists() && !dst_idx.exists() {
1040        let _ = fs::remove_file(&dst_pack);
1041    }
1042    if dst_pack.exists() && dst_idx.exists() {
1043        if existing_pair_matches_digest(packs_dir, pack_name, &expected)? {
1044            remove_staging(packs_dir, &install_id);
1045            return Ok(());
1046        }
1047        let _ = fs::remove_file(&dst_pack);
1048        let _ = fs::remove_file(&dst_idx);
1049    }
1050
1051    let mut intent = PackInstallIntent::new(install_id.clone(), pack_name.to_string());
1052    write_intent(packs_dir, &intent)?;
1053    fault_inject::maybe_fail_at("pack_install_stream_after_intent_prepared")?;
1054
1055    publish_file_durable(&staging_pack, &dst_pack)?;
1056    fault_inject::maybe_fail_at("pack_install_stream_after_publish_pack")?;
1057    intent.phase = PackInstallPhase::PackPublished;
1058    write_intent(packs_dir, &intent)?;
1059    fault_inject::maybe_fail_at("pack_install_stream_after_intent_pack_published")?;
1060
1061    publish_file_durable(&staging_idx, &dst_idx)?;
1062    fault_inject::maybe_fail_at("pack_install_stream_after_publish_idx")?;
1063    remove_staging(packs_dir, &install_id);
1064    remove_intent(packs_dir, &install_id)?;
1065    Ok(())
1066}
1067
1068// ---------------------------------------------------------------------------
1069// Tests
1070// ---------------------------------------------------------------------------
1071
1072#[cfg(test)]
1073mod tests {
1074    use super::*;
1075    use crate::fs_atomic::write_file_atomic;
1076
1077    fn write_src(dir: &Path, name: &str, bytes: &[u8]) -> PathBuf {
1078        let p = dir.join(name);
1079        write_file_atomic(&p, bytes).unwrap();
1080        p
1081    }
1082
1083    /// Deterministic 64-char lowercase hex pack name from a seed.
1084    fn pack_id(seed: &str) -> String {
1085        digest_to_pack_name(blake3::hash(seed.as_bytes()).as_bytes())
1086    }
1087
1088    /// Minimal structurally valid pack index bytes.
1089    fn empty_idx_bytes() -> Vec<u8> {
1090        crate::store::pack::PackIndex::new().to_bytes()
1091    }
1092
1093    #[test]
1094    fn validate_ids_reject_path_traversal() {
1095        assert!(validate_install_id("../evil").is_err());
1096        assert!(validate_install_id("a/b").is_err());
1097        assert!(validate_install_id("").is_err());
1098        assert!(validate_pack_name("../x").is_err());
1099        assert!(validate_pack_name("not-hex!").is_err());
1100        assert!(validate_pack_name("deadbeef").is_err()); // too short
1101        assert!(validate_pack_name(&"A".repeat(64)).is_err()); // uppercase
1102        assert!(validate_install_id("abc-123_OK").is_ok());
1103        assert!(validate_pack_name(&pack_id("ok")).is_ok());
1104        assert_eq!(pack_id("ok").len(), 64);
1105    }
1106
1107    #[test]
1108    fn pack_name_digest_roundtrip_and_native_equality() {
1109        let body = b"digest-native-eq";
1110        let digest = *blake3::hash(body).as_bytes();
1111        let name = digest_to_pack_name(&digest);
1112        let parsed = pack_name_to_digest(&name).unwrap();
1113        assert_eq!(parsed, digest);
1114        // File identity uses bytes, not hex strings.
1115        let root = tempfile::tempdir().unwrap();
1116        let packs = root.path().join("packs");
1117        create_dir_all_durable(&packs).unwrap();
1118        write_file_atomic(&dst_pack_path(&packs, &name), body).unwrap();
1119        write_file_atomic(&dst_idx_path(&packs, &name), &empty_idx_bytes()).unwrap();
1120        assert!(existing_pair_matches_digest(&packs, &name, &digest).unwrap());
1121        let mut wrong = digest;
1122        wrong[0] ^= 0xff;
1123        assert!(!existing_pair_matches_digest(&packs, &name, &wrong).unwrap());
1124    }
1125
1126    #[test]
1127    fn malicious_intent_paths_ignored_reconstructed_from_packs_dir() {
1128        // Even if someone forged absolute paths in JSON, serde ignores unknown
1129        // fields and we only use install_id + pack_name.
1130        let root = tempfile::tempdir().unwrap();
1131        let packs = root.path().join("packs");
1132        create_dir_all_durable(&packs).unwrap();
1133        let outside = root.path().join("outside.txt");
1134        fs::write(&outside, b"secret").unwrap();
1135
1136        let install_id = "malicious1";
1137        let pack_name = pack_id("malicious-pack");
1138        let intent_dir = intent_root(&packs);
1139        create_dir_all_durable(&intent_dir).unwrap();
1140        let forged = serde_json::json!({
1141            "version": 2,
1142            "install_id": install_id,
1143            "pack_name": pack_name,
1144            "phase": "prepared",
1145            "created_unix": 1,
1146            "staging_pack": outside.display().to_string(),
1147            "dst_pack": outside.display().to_string(),
1148            "dst_idx": outside.display().to_string(),
1149        });
1150        fs::write(
1151            intent_path(&packs, install_id),
1152            serde_json::to_vec_pretty(&forged).unwrap(),
1153        )
1154        .unwrap();
1155
1156        // Expired → abort uses reconstructed paths only.
1157        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1158        assert_eq!(report.aborted, 1);
1159        // Outside file must survive.
1160        assert_eq!(fs::read(&outside).unwrap(), b"secret");
1161    }
1162
1163    #[test]
1164    fn quarantine_unknown_version_preserves_file() {
1165        let root = tempfile::tempdir().unwrap();
1166        let packs = root.path().join("packs");
1167        create_dir_all_durable(&packs).unwrap();
1168        let intent_dir = intent_root(&packs);
1169        create_dir_all_durable(&intent_dir).unwrap();
1170        let path = intent_dir.join("weird.json");
1171        let pn = pack_id("weird");
1172        fs::write(
1173            &path,
1174            format!(
1175                r#"{{"version":99,"install_id":"x","pack_name":"{pn}","phase":"prepared","created_unix":1}}"#
1176            ),
1177        )
1178        .unwrap();
1179
1180        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1181        assert_eq!(report.quarantined, 1);
1182        assert!(!path.exists());
1183        let q = quarantine_root(&packs);
1184        assert!(q.exists());
1185        assert!(fs::read_dir(&q).unwrap().count() >= 1);
1186    }
1187
1188    #[test]
1189    fn short_pack_name_in_intent_is_quarantined() {
1190        let root = tempfile::tempdir().unwrap();
1191        let packs = root.path().join("packs");
1192        create_dir_all_durable(&packs).unwrap();
1193        create_dir_all_durable(&intent_root(&packs)).unwrap();
1194        let path = intent_path(&packs, "shortname");
1195        fs::write(
1196            &path,
1197            br#"{"version":2,"install_id":"shortname","pack_name":"aa","phase":"prepared","created_unix":1}"#,
1198        )
1199        .unwrap();
1200        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1201        assert_eq!(report.quarantined, 1);
1202        assert!(!path.exists());
1203    }
1204
1205    #[test]
1206    fn quarantine_malformed_json() {
1207        let root = tempfile::tempdir().unwrap();
1208        let packs = root.path().join("packs");
1209        create_dir_all_durable(&packs).unwrap();
1210        let intent_dir = intent_root(&packs);
1211        create_dir_all_durable(&intent_dir).unwrap();
1212        let path = intent_dir.join("bad.json");
1213        fs::write(&path, b"not-json{{{{").unwrap();
1214
1215        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1216        assert_eq!(report.quarantined, 1);
1217        assert!(!path.exists());
1218    }
1219
1220    #[test]
1221    fn existing_pair_requires_hash_match_and_valid_index() {
1222        let root = tempfile::tempdir().unwrap();
1223        let packs = root.path().join("packs");
1224        create_dir_all_durable(&packs).unwrap();
1225        let body = b"pack-body-xyz";
1226        let name = format!("{}", blake3::hash(body).to_hex());
1227        write_file_atomic(&dst_pack_path(&packs, &name), body).unwrap();
1228        write_file_atomic(&dst_idx_path(&packs, &name), &empty_idx_bytes()).unwrap();
1229        assert!(existing_pair_matches_pack_name(&packs, &name).unwrap());
1230
1231        // Structurally invalid index → not a match.
1232        write_file_atomic(&dst_idx_path(&packs, &name), b"not-an-index").unwrap();
1233        assert!(!existing_pair_matches_pack_name(&packs, &name).unwrap());
1234
1235        // Wrong name for content (valid hex, wrong digest).
1236        let wrong = pack_id("wrong-name-for-body");
1237        write_file_atomic(&dst_pack_path(&packs, &wrong), body).unwrap();
1238        write_file_atomic(&dst_idx_path(&packs, &wrong), &empty_idx_bytes()).unwrap();
1239        assert!(!existing_pair_matches_pack_name(&packs, &wrong).unwrap());
1240    }
1241
1242    #[test]
1243    fn install_rejects_idempotent_false_pair() {
1244        let root = tempfile::tempdir().unwrap();
1245        let packs = root.path().join("packs");
1246        create_dir_all_durable(&packs).unwrap();
1247        let pack_bytes = b"real-pack-content-111";
1248        let idx_bytes = empty_idx_bytes();
1249        let name = format!("{}", blake3::hash(pack_bytes).to_hex());
1250
1251        // Corrupt pair with correct name but wrong content.
1252        write_file_atomic(&dst_pack_path(&packs, &name), b"wrong-content").unwrap();
1253        write_file_atomic(&dst_idx_path(&packs, &name), b"x").unwrap();
1254        assert!(!existing_pair_matches_pack_name(&packs, &name).unwrap());
1255
1256        let out = install_pack_bytes_journaled(&packs, pack_bytes, &idx_bytes).unwrap();
1257        assert_eq!(out, name);
1258        assert_eq!(fs::read(dst_pack_path(&packs, &name)).unwrap(), pack_bytes);
1259    }
1260
1261    #[test]
1262    fn journaled_install_produces_pair_and_cleans_intent() {
1263        let root = tempfile::tempdir().unwrap();
1264        let packs = root.path().join("packs");
1265        create_dir_all_durable(&packs).unwrap();
1266        let src_dir = root.path().join("src");
1267        create_dir_all_durable(&src_dir).unwrap();
1268
1269        let pack_bytes = b"fake-pack-bytes-aaa";
1270        let idx_bytes = b"fake-idx-bytes-aaa";
1271        let src_pack = write_src(&src_dir, "p.pack", pack_bytes);
1272        let src_idx = write_src(&src_dir, "p.idx", idx_bytes);
1273        let name = format!("{}", blake3::hash(pack_bytes).to_hex());
1274
1275        install_pack_files_journaled(&packs, &src_pack, &src_idx, &name).unwrap();
1276
1277        assert!(dst_pack_path(&packs, &name).exists());
1278        assert!(dst_idx_path(&packs, &name).exists());
1279        assert_eq!(fs::read(dst_pack_path(&packs, &name)).unwrap(), pack_bytes);
1280        assert!(
1281            !intent_root(&packs).exists()
1282                || fs::read_dir(intent_root(&packs))
1283                    .unwrap()
1284                    .filter(|e| {
1285                        e.as_ref()
1286                            .map(|e| e.path().extension().and_then(|x| x.to_str()) == Some("json"))
1287                            .unwrap_or(false)
1288                    })
1289                    .count()
1290                    == 0
1291        );
1292    }
1293
1294    #[test]
1295    fn recover_pack_published_completes_from_staging() {
1296        let root = tempfile::tempdir().unwrap();
1297        let packs = root.path().join("packs");
1298        create_dir_all_durable(&packs).unwrap();
1299
1300        let name = pack_id("deadbeef-seed");
1301        let install_id = "test-install-1";
1302        validate_install_id(install_id).unwrap();
1303        validate_pack_name(&name).unwrap();
1304        let stage = staging_dir(&packs, install_id);
1305        create_dir_all_durable(&stage).unwrap();
1306        write_file_atomic(&staging_pack_path(&packs, install_id), b"pack-body").unwrap();
1307        write_file_atomic(&staging_idx_path(&packs, install_id), b"idx-body").unwrap();
1308
1309        let dst_pack = dst_pack_path(&packs, &name);
1310        publish_file_durable(&staging_pack_path(&packs, install_id), &dst_pack).unwrap();
1311
1312        let intent = PackInstallIntent {
1313            version: PACK_INSTALL_INTENT_VERSION,
1314            install_id: install_id.to_string(),
1315            pack_name: name.clone(),
1316            phase: PackInstallPhase::PackPublished,
1317            created_unix: 1,
1318        };
1319        write_intent(&packs, &intent).unwrap();
1320
1321        let report = recover_pack_install_intents(&packs).unwrap();
1322        assert_eq!(report.intents_seen, 1);
1323        assert_eq!(report.completed, 1);
1324        assert!(dst_pack.exists());
1325        assert!(dst_idx_path(&packs, &name).exists());
1326        assert_eq!(fs::read(dst_idx_path(&packs, &name)).unwrap(), b"idx-body");
1327        assert!(!intent_path(&packs, install_id).exists());
1328    }
1329
1330    #[test]
1331    fn recover_prepared_aborts_without_finals_when_expired() {
1332        let root = tempfile::tempdir().unwrap();
1333        let packs = root.path().join("packs");
1334        create_dir_all_durable(&packs).unwrap();
1335        let install_id = "prep-abort";
1336        let stage = staging_dir(&packs, install_id);
1337        create_dir_all_durable(&stage).unwrap();
1338        write_file_atomic(&staging_pack_path(&packs, install_id), b"p").unwrap();
1339        write_file_atomic(&staging_idx_path(&packs, install_id), b"i").unwrap();
1340
1341        let mut intent = PackInstallIntent::new(install_id.into(), pack_id("aa"));
1342        intent.created_unix = 1;
1343        write_intent(&packs, &intent).unwrap();
1344
1345        let report = recover_pack_install_intents_with_ttl(&packs, Some(60)).unwrap();
1346        assert_eq!(report.aborted, 1);
1347        assert!(!dst_pack_path(&packs, &pack_id("aa")).exists());
1348        assert!(!intent_path(&packs, install_id).exists());
1349        assert!(!stage.exists());
1350    }
1351
1352    #[test]
1353    fn recover_prepared_fresh_skips_in_progress() {
1354        let root = tempfile::tempdir().unwrap();
1355        let packs = root.path().join("packs");
1356        create_dir_all_durable(&packs).unwrap();
1357        let install_id = "live-prep";
1358        let stage = staging_dir(&packs, install_id);
1359        create_dir_all_durable(&stage).unwrap();
1360        let staging_pack = staging_pack_path(&packs, install_id);
1361        let staging_idx = staging_idx_path(&packs, install_id);
1362        write_file_atomic(&staging_pack, b"live-p").unwrap();
1363        write_file_atomic(&staging_idx, b"live-i").unwrap();
1364
1365        let intent = PackInstallIntent::new(install_id.into(), pack_id("bb"));
1366        write_intent(&packs, &intent).unwrap();
1367
1368        let report = recover_pack_install_intents_with_ttl(&packs, Some(86_400)).unwrap();
1369        assert_eq!(report.skipped_in_progress, 1);
1370        assert_eq!(report.aborted, 0);
1371        assert!(staging_pack.exists());
1372        assert!(intent_path(&packs, install_id).exists());
1373    }
1374
1375    #[test]
1376    fn recover_pack_published_without_staging_idx_aborts_orphan_pack() {
1377        let root = tempfile::tempdir().unwrap();
1378        let packs = root.path().join("packs");
1379        create_dir_all_durable(&packs).unwrap();
1380        let name = pack_id("cc");
1381        let install_id = "orph-1";
1382        let dst_pack = dst_pack_path(&packs, &name);
1383        write_file_atomic(&dst_pack, b"only-pack").unwrap();
1384
1385        let intent = PackInstallIntent {
1386            version: PACK_INSTALL_INTENT_VERSION,
1387            install_id: install_id.into(),
1388            pack_name: name,
1389            phase: PackInstallPhase::PackPublished,
1390            created_unix: 1,
1391        };
1392        write_intent(&packs, &intent).unwrap();
1393
1394        let report = recover_pack_install_intents(&packs).unwrap();
1395        assert_eq!(report.aborted, 1);
1396        assert!(!dst_pack.exists());
1397        assert!(!intent_path(&packs, install_id).exists());
1398    }
1399
1400    #[test]
1401    fn pack_lock_prevents_recover_aborting_live_expired_looking_install() {
1402        use std::{
1403            sync::{Arc, Barrier},
1404            thread,
1405            time::Duration,
1406        };
1407
1408        // Per-pack try_lock: recover must skip (not abort) while install holds
1409        // the pack lock — even if created_unix looks TTL-expired.
1410        let root = tempfile::tempdir().unwrap();
1411        let packs = Arc::new(root.path().join("packs"));
1412        create_dir_all_durable(&packs).unwrap();
1413
1414        let planted_under_lock = Arc::new(Barrier::new(2));
1415        let both_done = Arc::new(Barrier::new(2));
1416
1417        let packs_a = Arc::clone(&packs);
1418        let planted_a = Arc::clone(&planted_under_lock);
1419        let done_a = Arc::clone(&both_done);
1420
1421        let installer = thread::spawn(move || {
1422            let packs = packs_a.as_path();
1423            let guard = acquire_pack_name_lock(packs, &pack_id("dd")).expect("pack lock");
1424
1425            let install_id = "flock-live";
1426            let stage = staging_dir(packs, install_id);
1427            create_dir_all_durable(&stage).unwrap();
1428            let staging_pack = staging_pack_path(packs, install_id);
1429            let staging_idx = staging_idx_path(packs, install_id);
1430            write_file_atomic(&staging_pack, b"flock-pack").unwrap();
1431            write_file_atomic(&staging_idx, b"flock-idx").unwrap();
1432            let dst_pack = dst_pack_path(packs, &pack_id("dd"));
1433            let dst_idx = dst_idx_path(packs, &pack_id("dd"));
1434
1435            let mut intent = PackInstallIntent::new(install_id.into(), pack_id("dd"));
1436            intent.created_unix = 1; // looks expired under any short TTL
1437            write_intent(packs, &intent).unwrap();
1438
1439            planted_a.wait();
1440            thread::sleep(Duration::from_millis(60));
1441
1442            assert!(staging_pack.exists() && staging_idx.exists());
1443            assert!(intent_path(packs, install_id).exists());
1444
1445            publish_file_durable(&staging_pack, &dst_pack).unwrap();
1446            intent.phase = PackInstallPhase::PackPublished;
1447            write_intent(packs, &intent).unwrap();
1448            publish_file_durable(&staging_idx, &dst_idx).unwrap();
1449            remove_staging(packs, install_id);
1450            remove_intent(packs, install_id).unwrap();
1451
1452            drop(guard);
1453            assert!(dst_pack.exists() && dst_idx.exists());
1454            done_a.wait();
1455        });
1456
1457        let packs_b = Arc::clone(&packs);
1458        let planted_b = Arc::clone(&planted_under_lock);
1459        let done_b = Arc::clone(&both_done);
1460
1461        let recoverer = thread::spawn(move || {
1462            let packs = packs_b.as_path();
1463            planted_b.wait();
1464            // While pack lock is held, recover try_locks and skips — must not abort.
1465            let mid = recover_pack_install_intents_with_ttl(packs, Some(1))
1466                .expect("recover under pack lock");
1467            assert_eq!(mid.aborted, 0, "must not abort live install: {mid:?}");
1468            assert!(
1469                mid.skipped_in_progress >= 1 || dst_pack_path(packs, &pack_id("dd")).exists(),
1470                "either skip in-progress or install already finished: {mid:?}"
1471            );
1472            done_b.wait();
1473            // After installer finishes, finals exist and no intent remains.
1474            assert!(dst_pack_path(packs, &pack_id("dd")).exists());
1475            assert!(dst_idx_path(packs, &pack_id("dd")).exists());
1476            assert!(!intent_path(packs, "flock-live").exists());
1477        });
1478
1479        installer.join().expect("installer");
1480        recoverer.join().expect("recoverer");
1481    }
1482
1483    #[test]
1484    fn recover_prepared_with_pack_and_staging_idx_completes() {
1485        let root = tempfile::tempdir().unwrap();
1486        let packs = root.path().join("packs");
1487        create_dir_all_durable(&packs).unwrap();
1488        let install_id = "prep-complete";
1489        let stage = staging_dir(&packs, install_id);
1490        create_dir_all_durable(&stage).unwrap();
1491        write_file_atomic(&staging_pack_path(&packs, install_id), b"pack-x").unwrap();
1492        write_file_atomic(&staging_idx_path(&packs, install_id), b"idx-x").unwrap();
1493        let dst_pack = dst_pack_path(&packs, &pack_id("ee"));
1494        publish_file_durable(&staging_pack_path(&packs, install_id), &dst_pack).unwrap();
1495
1496        let intent = PackInstallIntent::new(install_id.into(), pack_id("ee"));
1497        write_intent(&packs, &intent).unwrap();
1498
1499        let report = recover_pack_install_intents(&packs).unwrap();
1500        assert_eq!(report.completed, 1);
1501        assert!(dst_pack.exists());
1502        assert!(dst_idx_path(&packs, &pack_id("ee")).exists());
1503        assert_eq!(
1504            fs::read(dst_idx_path(&packs, &pack_id("ee"))).unwrap(),
1505            b"idx-x"
1506        );
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()
1752                .contains("pack_install_after_intent_prepared"),
1753            "err={err}"
1754        );
1755        // Process-global counters can race under parallel tests; assert non-decreasing delta.
1756        assert!(pack_install_metrics_snapshot().installs_err >= before_err);
1757
1758        // Staging + prepared intent should remain for recovery.
1759        assert_eq!(intent_count_json(&packs), 1);
1760        assert!(!dst_pack_path(&packs, &expected).exists());
1761
1762        // Force-expire prepared intent, abort, then reinstall succeeds.
1763        for entry in fs::read_dir(intent_root(&packs)).unwrap().flatten() {
1764            let p = entry.path();
1765            if p.extension().and_then(|x| x.to_str()) != Some("json") {
1766                continue;
1767            }
1768            let mut intent: PackInstallIntent =
1769                serde_json::from_slice(&fs::read(&p).unwrap()).unwrap();
1770            intent.created_unix = 1;
1771            write_intent(&packs, &intent).unwrap();
1772        }
1773        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1774        assert_eq!(report.aborted, 1, "report={report:?}");
1775        assert_eq!(intent_count_json(&packs), 0);
1776
1777        let name = install_pack_bytes_journaled(&packs, pack_bytes, &idx_bytes).unwrap();
1778        assert_eq!(name, expected);
1779        assert!(existing_pair_matches_pack_name(&packs, &expected).unwrap());
1780    }
1781
1782    #[test]
1783    fn fault_after_publish_pack_recovers_to_complete() {
1784        let root = tempfile::tempdir().unwrap();
1785        let packs = root.path().join("packs");
1786        create_dir_all_durable(&packs).unwrap();
1787
1788        let pack_bytes = b"fault-after-pack-publish-body";
1789        let idx_bytes = empty_idx_bytes();
1790        let expected = format!("{}", blake3::hash(pack_bytes).to_hex());
1791
1792        let err = fault_inject::with_fault_points(&["pack_install_after_publish_pack"], || {
1793            install_pack_bytes_journaled(&packs, pack_bytes, &idx_bytes)
1794        })
1795        .expect_err("fault should fire after pack publish");
1796        assert!(err.to_string().contains("pack_install_after_publish_pack"));
1797
1798        // Pack published, intent may still be prepared (fault is after publish, before
1799        // phase rewrite) or pack_published depending on checkpoint placement.
1800        assert!(dst_pack_path(&packs, &expected).exists());
1801        assert!(!dst_idx_path(&packs, &expected).exists());
1802
1803        let report = recover_pack_install_intents(&packs).unwrap();
1804        assert_eq!(report.completed, 1, "report={report:?}");
1805        assert!(dst_idx_path(&packs, &expected).exists());
1806        assert!(existing_pair_matches_pack_name(&packs, &expected).unwrap());
1807    }
1808
1809    fn intent_count_json(packs: &Path) -> usize {
1810        let dir = intent_root(packs);
1811        if !dir.exists() {
1812            return 0;
1813        }
1814        fs::read_dir(dir)
1815            .unwrap()
1816            .filter(|e| {
1817                e.as_ref()
1818                    .map(|e| e.path().extension().and_then(|x| x.to_str()) == Some("json"))
1819                    .unwrap_or(false)
1820            })
1821            .count()
1822    }
1823
1824    #[test]
1825    fn metrics_snapshot_tracks_install_and_recover() {
1826        // Process-global atomics: measure deltas so parallel tests don't flake.
1827        let before = pack_install_metrics_snapshot();
1828        let root = tempfile::tempdir().unwrap();
1829        let packs = root.path().join("packs");
1830        create_dir_all_durable(&packs).unwrap();
1831        let _ = install_pack_bytes_journaled(&packs, b"metrics-pack", &empty_idx_bytes()).unwrap();
1832        let after_install = pack_install_metrics_snapshot();
1833        assert!(after_install.installs_ok > before.installs_ok);
1834
1835        // Plant expired prepared → recover aborts.
1836        let install_id = "metrics-abort";
1837        create_dir_all_durable(&staging_dir(&packs, install_id)).unwrap();
1838        write_file_atomic(&staging_pack_path(&packs, install_id), b"p").unwrap();
1839        write_file_atomic(&staging_idx_path(&packs, install_id), b"i").unwrap();
1840        let mut intent = PackInstallIntent::new(install_id.into(), pack_id("cc"));
1841        intent.created_unix = 1;
1842        write_intent(&packs, &intent).unwrap();
1843        let before_abort = pack_install_metrics_snapshot();
1844        let report = recover_pack_install_intents_with_ttl(&packs, Some(1)).unwrap();
1845        assert_eq!(report.aborted, 1);
1846        let after_abort = pack_install_metrics_snapshot();
1847        assert!(after_abort.recover_aborted > before_abort.recover_aborted);
1848    }
1849
1850    #[cfg(unix)]
1851    #[test]
1852    fn assert_under_packs_rejects_staging_symlink_escape() {
1853        use std::os::unix::fs::symlink;
1854
1855        let root = tempfile::tempdir().unwrap();
1856        let packs = root.path().join("packs");
1857        create_dir_all_durable(&packs).unwrap();
1858        let outside = root.path().join("outside");
1859        create_dir_all_durable(&outside).unwrap();
1860
1861        // Lexically under packs, but .staging is a symlink out.
1862        symlink(&outside, packs.join(STAGING_DIR_NAME)).unwrap();
1863
1864        let stage = staging_dir(&packs, "id1");
1865        let err = assert_under_packs(&packs, &stage).unwrap_err();
1866        assert!(
1867            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1868            "err={err}"
1869        );
1870
1871        // Journal layout guard must fail before install.
1872        let err = ensure_journal_layout_safe(&packs).unwrap_err();
1873        assert!(
1874            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1875            "err={err}"
1876        );
1877        let err = install_pack_bytes_journaled(&packs, b"x", &empty_idx_bytes()).unwrap_err();
1878        assert!(
1879            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1880            "err={err}"
1881        );
1882    }
1883
1884    #[cfg(unix)]
1885    #[test]
1886    fn assert_under_packs_rejects_intent_root_symlink_escape() {
1887        use std::os::unix::fs::symlink;
1888
1889        let root = tempfile::tempdir().unwrap();
1890        let packs = root.path().join("packs");
1891        create_dir_all_durable(&packs).unwrap();
1892        let outside = root.path().join("outside-intent");
1893        create_dir_all_durable(&outside).unwrap();
1894        symlink(&outside, packs.join(INTENT_DIR_NAME)).unwrap();
1895
1896        let err = ensure_journal_layout_safe(&packs).unwrap_err();
1897        assert!(
1898            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1899            "err={err}"
1900        );
1901    }
1902
1903    #[cfg(unix)]
1904    #[test]
1905    fn assert_under_packs_rejects_pack_locks_symlink_escape() {
1906        use std::os::unix::fs::symlink;
1907
1908        let root = tempfile::tempdir().unwrap();
1909        let packs = root.path().join("packs");
1910        create_dir_all_durable(&packs).unwrap();
1911        let outside = root.path().join("outside-locks");
1912        create_dir_all_durable(&outside).unwrap();
1913        symlink(&outside, packs.join(PACK_LOCKS_DIR_NAME)).unwrap();
1914
1915        let err = ensure_journal_layout_safe(&packs).unwrap_err();
1916        assert!(
1917            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1918            "err={err}"
1919        );
1920    }
1921
1922    #[cfg(unix)]
1923    #[test]
1924    fn assert_under_packs_rejects_destination_file_symlink_escape() {
1925        use std::os::unix::fs::symlink;
1926
1927        let root = tempfile::tempdir().unwrap();
1928        let packs = root.path().join("packs");
1929        create_dir_all_durable(&packs).unwrap();
1930        let outside = root.path().join("evil.pack");
1931        fs::write(&outside, b"evil").unwrap();
1932
1933        let name = pack_id("symlink-dst");
1934        let dst = dst_pack_path(&packs, &name);
1935        symlink(&outside, &dst).unwrap();
1936
1937        let err = assert_under_packs(&packs, &dst).unwrap_err();
1938        assert!(
1939            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1940            "err={err}"
1941        );
1942        // existing_pair must refuse a symlink-out destination (error or false).
1943        let pair = existing_pair_matches_pack_name(&packs, &name);
1944        assert!(pair.as_ref().map(|v| !v).unwrap_or(true), "pair={pair:?}");
1945    }
1946
1947    #[cfg(unix)]
1948    #[test]
1949    fn assert_under_packs_rejects_install_id_staging_symlink() {
1950        use std::os::unix::fs::symlink;
1951
1952        let root = tempfile::tempdir().unwrap();
1953        let packs = root.path().join("packs");
1954        create_dir_all_durable(&packs).unwrap();
1955        create_dir_all_durable(&staging_root(&packs)).unwrap();
1956        let outside = root.path().join("outside-stage-id");
1957        create_dir_all_durable(&outside).unwrap();
1958        let install_id = "symlink-stage-id";
1959        symlink(&outside, staging_dir(&packs, install_id)).unwrap();
1960
1961        let pack_path = staging_pack_path(&packs, install_id);
1962        let err = assert_under_packs(&packs, &pack_path).unwrap_err();
1963        assert!(
1964            err.to_string().contains("escapes") || err.to_string().contains("symlink"),
1965            "err={err}"
1966        );
1967    }
1968
1969    #[test]
1970    fn assert_under_packs_accepts_normal_reconstructed_paths() {
1971        let root = tempfile::tempdir().unwrap();
1972        let packs = root.path().join("packs");
1973        create_dir_all_durable(&packs).unwrap();
1974        let name = pack_id("normal");
1975        let install_id = "normal-id";
1976        assert_under_packs(&packs, &dst_pack_path(&packs, &name)).unwrap();
1977        assert_under_packs(&packs, &staging_pack_path(&packs, install_id)).unwrap();
1978        assert_under_packs(&packs, &intent_path(&packs, install_id)).unwrap();
1979    }
1980}