Stop a project id from steering a Docker API DELETE, and stop trusting a store that lost its list

C-2 (critical). `destroy_ownerless_rollback_pin` validated its tag and not its
project id, then interpolated both into `triple-c-snapshot-{id}:{tag}` and handed
the result to bollard. bollard does not percent-encode: `Uri::parse` joins an
absolute path onto the base URL, which replaces the path outright and applies RFC
3986 dot-segment removal. An id of `a/../../v1.47/volumes/<name>?` turns a
"remove image tag" into `DELETE /v1.47/volumes/<name>`. That arm is reached
*because* `find_project` failed, so the id is unconstrained IPC input, and the
typed confirmation is no barrier — it compares the caller's own two strings.

Reproduced against the live daemon, and now a test: with the check removed the
volume is gone and the test fails; with it, the volume survives and a legitimate
ownerless pin still deletes. The reference that reaches `remove_image` is now the
daemon's own repo_tag, matched on the parsed pair, so nothing built from IPC
input addresses the API at all. The same id check now guards the owned arms of
`destroy` and `compact_snapshot`, which build volume names and image references
from a `projects.json` field.

H-1. The ownerless arm decided ownership from the in-memory list alone and then
called `sweep_orphaned_snapshots()`, which deletes the freshly dangling image on
the same pass — so a corrupt `projects.json` could reap a pin whose migration is
still awaiting confirmation, the one thing `pin_is_reapable` orders its
conditions to prevent. It now re-reads the store from disk, runs
`project_store_trust`, refuses an id the store knows, takes the project lock
before reading anything a decision rests on, and checks `has_record`.

H-3. The corrupt-store guard keyed on "empty list + file exists", and
`ProjectsStore::new()` swallows a corrupt file without rewriting it — so the
first `save()`, as little as starting a project, wrote `[{new}]` over it and the
guard passed with every other project's volumes unclaimed. A corrupt load is now
recorded in a sticky `projects.json.corrupt` marker beside the file, and the
existing `.bak` is no longer clobbered by a second corruption. A missing
`projects.json` is refused too: it cannot be told from a moved or partially
restored data directory, and the genuinely fresh case has nothing to find.

Also: the three migration commands surface the lock's real refusal instead of
substituting "a migration is already running"; `note_ownerless_since` re-checks
`has_record` after writing a tombstone, closing the window that could plant one
behind a valid record and reap the pin with zero grace; corrupt migration-record
copies are capped at four; `reconcile_migration` yields to any lock holder, not
only a migration; and a 22-space run in a refusal string is gone.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBq2rGum6GX7xXgsas1fDc

app/src-tauri/src/storage/migration_store.rs

@@ -83,9 +83,9 @@ pub fn load(project_id: &str) -> Result<Option<MigrationState>, String> {
         Ok(state) => Ok(Some(state)),
         Err(e) => {
             let backup = corrupt_backup_path(&path, &chrono::Utc::now());
-            let copied = if backup.exists() {
-                // Already kept a copy of this exact corruption this second;
-                // nothing to add.
+            let copied = if backup.exists() || corrupt_backups_full(&path) {
+                // Already kept a copy of this exact corruption this second, or
+                // kept as many as are worth keeping. Either way nothing to add.
                 Ok(())
             } else {
                 fs::copy(&path, &backup).map(|_| ())
@@ -114,6 +114,50 @@ fn corrupt_backup_path(path: &std::path::Path, now: &chrono::DateTime<chrono::Ut
     path.with_extension(format!("json.corrupt-{}.bak", now.format("%Y%m%d-%H%M%S")))
 }
 
+/// How many timestamped copies of one project's corrupt record are kept.
+///
+/// Timestamping fixed the "second corruption overwrote the first" bug and
+/// introduced its opposite: [`load`] runs on every reconcile, every survey and
+/// every reaper pass, so a record that is *persistently* unparseable — the
+/// normal case, since nothing repairs it — mints a new copy every time the
+/// clock's second changes. Nothing ever reads them back and nothing ever
+/// removed them.
+///
+/// Four is enough for the only use there is: a human looking at what the file
+/// held. See [`corrupt_backups_full`] for why the cap is applied before the
+/// copy rather than by pruning after it.
+const MAX_CORRUPT_BACKUPS: usize = 4;
+
+/// Whether [`MAX_CORRUPT_BACKUPS`] copies of this record already exist.
+///
+/// Asked *before* the copy rather than pruning after it, so the cap is not
+/// implemented by writing a file and deleting it again on every pass — and so
+/// the copies that survive are the oldest, which are the ones taken closest to
+/// whatever produced the corruption.
+///
+/// A directory that cannot be listed answers "not full": failing open here
+/// costs at most one extra file, and failing closed would drop the very first
+/// copy of a record nothing else has kept.
+fn corrupt_backups_full(path: &std::path::Path) -> bool {
+    let (Some(dir), Some(stem)) = (path.parent(), path.file_stem()) else {
+        return false;
+    };
+    // `{stem}.json.corrupt-` — the same shape `corrupt_backup_path` builds, so
+    // this can never match another project's copies or an unrelated `.bak`.
+    let prefix = format!("{}.json.corrupt-", stem.to_string_lossy());
+    let Ok(entries) = fs::read_dir(dir) else {
+        return false;
+    };
+    entries
+        .flatten()
+        .filter(|e| {
+            let name = e.file_name().to_string_lossy().to_string();
+            name.starts_with(&prefix) && name.ends_with(".bak")
+        })
+        .count()
+        >= MAX_CORRUPT_BACKUPS
+}
+
 /// Whether a project has a migration record on disk *at all*, without parsing
 /// it.
 ///
@@ -251,6 +295,22 @@ pub fn peek_ownerless_since(
 /// the period below what has actually elapsed on the *marker's* terms, because
 /// there is nothing to compare against but wall time; what it cannot do any
 /// more is make every pin instantly reapable, which dating from the tag did.
+///
+/// ## Why the write re-checks `has_record`
+///
+/// Both reapers ask [`has_record`] and only call this when the answer is no,
+/// which leaves a window: a [`save`] landing between the two runs its
+/// `clear_ownerless_for_project` against a marker that does not exist yet, and
+/// this then plants one — dated *now* — behind a perfectly valid record. The
+/// marker is invisible while the record stands, so nothing notices. It only
+/// matters later, if that record is legitimately lost: the pin is then already
+/// fourteen days ownerless on its very first check and is reaped with **zero**
+/// grace, which is the exact failure the tombstone exists to prevent.
+///
+/// So the write is followed by a second `has_record`, and a marker that turns
+/// out to sit behind a record is removed again. The two orderings that remain
+/// are both safe: a `save` completing *after* this re-check clears the marker
+/// itself, and one completing before it is what the re-check sees.
 pub fn note_ownerless_since(
     project_id: &str,
     tag: &str,
@@ -274,6 +334,19 @@ pub fn note_ownerless_since(
                     tag,
                     e
                 );
+                return None;
+            }
+            // A record that appeared while this was being written owns the pin,
+            // and a tombstone behind an owned pin is a fourteen-day head start
+            // on reaping it the moment that record is next lost.
+            if has_record(project_id).unwrap_or(false) {
+                log::debug!(
+                    "A migration record for {} appeared while marking {} ownerless; \
+                     the marker was dropped again",
+                    project_id,
+                    tag
+                );
+                clear_ownerless(project_id, tag);
             }
             None
         }
@@ -339,6 +412,39 @@ pub fn clear_staging(project_id: &str) -> Result<(), String> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn corrupt_copies_of_one_record_are_capped() {
+        // `load` runs on every reconcile, every survey and every reaper pass,
+        // and nothing repairs an unparseable record — so a persistently corrupt
+        // one minted a new timestamped copy every time the clock's second
+        // changed, and nothing ever removed them.
+        let dir = std::env::temp_dir().join(format!(
+            "triple-c-corrupt-cap-{}",
+            uuid::Uuid::new_v4().simple()
+        ));
+        fs::create_dir_all(&dir).expect("temp dir");
+        let record = dir.join("some-project.json");
+
+        assert!(!corrupt_backups_full(&record), "an empty directory is not full");
+        for n in 0..MAX_CORRUPT_BACKUPS {
+            fs::write(
+                dir.join(format!("some-project.json.corrupt-2026010{}-000000.bak", n)),
+                "x",
+            )
+            .unwrap();
+        }
+        assert!(corrupt_backups_full(&record));
+
+        // Another project's copies, and an unrelated `.bak`, are not this
+        // record's — the prefix is the whole point of the naming.
+        let other = dir.join("other-project.json");
+        assert!(!corrupt_backups_full(&other));
+        fs::write(dir.join("some-project.json.bak"), "x").unwrap();
+        assert!(!corrupt_backups_full(&other));
+
+        fs::remove_dir_all(&dir).ok();
+    }
+
     #[test]
     fn project_ids_cannot_escape_the_migrations_directory() {
         assert_eq!(sanitize("../../etc/passwd"), "______etc_passwd");