Marketplace Finish: repair sets whose marker says complete but files are missing

"2 chosen maps are only partly downloaded (Population density, Yandex
satellite imagery). Finish" - and Finish did nothing, silently.

Root cause: the banner and the button asked two different questions.
market::unfinished counts a set as unfinished when check_installed's file
count found it short (launcher.log: "popdark says it is installed but holds
293056 of 400268 files", "yandex ... 2371 of 28213"). Finish went through
apply_market, whose market::to_add only reads the markers - .blocks-done.json
listing every block - so it found nothing to add, nothing to remove, and
returned Ok. And even a forced install could not have repaired it: the
installer skips every block the marker lists (install.rs `pending`).

- odian-packs: InstallOptions::recheck_tile_sets. For those sets install_tiles
  counts the tiles on disk per block (tiles_on_disk_per_block: z/x/y.ext names
  only, so .blocks-done.json and the tile server's .missing negatives never
  count), takes off `done` every block holding fewer than the index names
  (forget_short_blocks - more is still whole: a PC bakes its own tiles in),
  rewrites the marker at once so a stopped run reads as partial, and fetches
  just those blocks; finish_block's rename replaces what is there. The
  drive-copy path skips its "recorded complete" shortcut for them too.
- launcher: Finish passes market::short_tile_sets(market_checked) through
  Inner::market_recheck (taken by install_market_items), apply_market runs when
  a re-check is all there is to do, and the log says which sets are repaired.

Counted, not hashed: the index names a block's tiles only by digest, so a
block whose lost tiles were replaced by as many baked ones reads whole.

Spot-check: odian-packs install::tests (a short block is dropped, a whole one
and one with extra baked tiles kept, markers not counted); launcher
market::tests::finish_repairs_the_sets_a_count_found_short with the counts
from the real log.

crates/odian-packs/src/install.rs

@@ -36,11 +36,25 @@ pub struct InstallOptions {
     pub tile_sets: Option<BTreeSet<String>>,
     /// The source files to install, by manifest path; `None` is all of them.
     pub src_paths: Option<BTreeSet<String>>,
+    /// Tile sets (`style/dataset`) whose `.blocks-done.json` is NOT to be taken at its word:
+    /// each block it lists is counted on disk first, and one holding fewer tiles than the
+    /// index names is fetched again. For a set a count has already found short - its marker
+    /// says every block landed, the folder says otherwise (tiles deleted by hand, a disk
+    /// cleaner, a copy that lost files) - which without this no install could ever repair,
+    /// since the marker is exactly what an install skips on.
+    pub recheck_tile_sets: BTreeSet<String>,
 }
 
 impl Default for InstallOptions {
     fn default() -> Self {
-        InstallOptions { workers: 6, only: vec![Stage::App, Stage::Db, Stage::Src, Stage::Tiles], db_paths: None, tile_sets: None, src_paths: None }
+        InstallOptions {
+            workers: 6,
+            only: vec![Stage::App, Stage::Db, Stage::Src, Stage::Tiles],
+            db_paths: None,
+            tile_sets: None,
+            src_paths: None,
+            recheck_tile_sets: BTreeSet::new(),
+        }
     }
 }
 
@@ -179,7 +193,7 @@ where
         stage_index += 1;
         let keep = |s: &&TileSet| opts.tile_sets.as_ref().is_none_or(|sets| sets.contains(&format!("{}/{}", s.style, s.dataset)));
         let tiles = TilesSection { sets: tiles.sets.iter().filter(keep).cloned().collect(), ..tiles.clone() };
-        summary.tiles = Some(install_tiles(&ctx, stage_index, &version, &tiles, &data_dir).await?);
+        summary.tiles = Some(install_tiles(&ctx, stage_index, &version, &tiles, &data_dir, &opts.recheck_tile_sets).await?);
     }
 
     check_cancel()?;
@@ -603,6 +617,7 @@ async fn copy_install_tiles(
     tiles: &TilesSection,
     root: &Path,
     data_dir: &Path,
+    recheck: &BTreeSet<String>,
 ) -> Result<Vec<TileSetSummary>> {
     let total: u64 = tiles.sets.iter().map(|s| s.bytes).sum();
     let files: u64 = tiles.sets.iter().map(|s| s.files).sum();
@@ -615,8 +630,10 @@ async fn copy_install_tiles(
         // This PC already recorded the very same set as complete: `.blocks-done.json` names
         // this index and all of its blocks, the record the marketplace itself reads to say
         // "Installed". Hashing its hundreds of thousands of files to learn what the marker
-        // already states costs minutes a set and can only ever conclude "identical".
-        if set_recorded_complete(&to, set) {
+        // already states costs minutes a set and can only ever conclude "identical" - unless
+        // the caller has counted the set short, when the copy below (which leaves identical
+        // files alone) is the repair.
+        if set_recorded_complete(&to, set) && !recheck.contains(&format!("{style}/{dataset}")) {
             ctx.reporter.begin_sub(format!("tiles {style}/{dataset}"), "files", set.files);
             ctx.reporter.items_done(set.files);
             ctx.reporter.add_done_bytes(set.bytes);
@@ -753,10 +770,59 @@ fn copy_checked(from: &Path, to: &Path, reporter: &Reporter, cancel: &Cancellati
     Ok((size, true))
 }
 
-async fn install_tiles(ctx: &Ctx<'_>, stage_index: u32, version: &str, tiles: &TilesSection, data_dir: &Path) -> Result<Vec<TileSetSummary>> {
+/// How many tiles `set_dir` holds, per block. Only names the installer itself would write
+/// count (`z/x/y.ext`, `parse_tile_name`'s rule), so the markers beside them - its own
+/// `.blocks-done.json`, the tile server's `.missing` negatives - never pass for tiles.
+fn tiles_on_disk_per_block(set_dir: &Path) -> HashMap<(u64, u64, u64), u64> {
+    let mut counts = HashMap::new();
+    let Ok(zooms) = std::fs::read_dir(set_dir) else { return counts };
+    for zoom in zooms.flatten() {
+        let Ok(columns) = std::fs::read_dir(zoom.path()) else { continue };
+        let z = zoom.file_name().to_string_lossy().into_owned();
+        for column in columns.flatten() {
+            let Ok(rows) = std::fs::read_dir(column.path()) else { continue };
+            let x = column.file_name().to_string_lossy().into_owned();
+            for row in rows.flatten() {
+                if !row.file_type().is_ok_and(|kind| kind.is_file()) {
+                    continue;
+                }
+                let name = format!("{z}/{x}/{}", row.file_name().to_string_lossy());
+                if let Some((z, x, y, _)) = crate::tiles::parse_tile_name(&name) {
+                    *counts.entry(crate::tiles::block_of(z, x, y)).or_default() += 1;
+                }
+            }
+        }
+    }
+    counts
+}
+
+/// Take off `done` every block the disk holds fewer tiles of than the index names, and say
+/// how many. A block holding MORE is whole: a PC bakes tiles of its own into the same folders.
+/// Counted, not hashed - the index names a block's tiles only through their digest - so a
+/// block whose missing tiles were replaced by as many baked ones reads whole; what a count
+/// cannot see, a hash of every tile of the set could, at minutes a set.
+fn forget_short_blocks(done: &mut BTreeSet<String>, blocks: &[IndexBlock], on_disk: &HashMap<(u64, u64, u64), u64>) -> usize {
+    let mut dropped = 0;
+    for block in blocks {
+        let held = on_disk.get(&(block.z, block.bx, block.by)).copied().unwrap_or(0);
+        if held < block.files && done.remove(&block.done_key()) {
+            dropped += 1;
+        }
+    }
+    dropped
+}
+
+async fn install_tiles(
+    ctx: &Ctx<'_>,
+    stage_index: u32,
+    version: &str,
+    tiles: &TilesSection,
+    data_dir: &Path,
+    recheck: &BTreeSet<String>,
+) -> Result<Vec<TileSetSummary>> {
     if let Some(root) = ctx.backend.install_root() {
         let root = root.to_path_buf();
-        return copy_install_tiles(ctx, stage_index, tiles, &root, data_dir).await;
+        return copy_install_tiles(ctx, stage_index, tiles, &root, data_dir, recheck).await;
     }
     let total: u64 = tiles.sets.iter().map(|s| s.bytes).sum();
     let blocks: u64 = tiles.sets.iter().map(|s| s.blocks).sum();
@@ -787,6 +853,16 @@ async fn install_tiles(ctx: &Ctx<'_>, stage_index: u32, version: &str, tiles: &T
                 done.extend(list.iter().filter_map(Value::as_str).map(str::to_string));
             }
         }
+        if !done.is_empty() && recheck.contains(&format!("{style}/{dataset}")) {
+            ctx.reporter.begin_sub(format!("tiles {style}/{dataset}: counting what is on disk"), "", 0);
+            let dir = set_dir.clone();
+            let on_disk = blocking(move || Ok(tiles_on_disk_per_block(&dir))).await?;
+            if forget_short_blocks(&mut done, &index.blocks, &on_disk) > 0 {
+                // Written now, not when the blocks land: a run stopped before then must leave
+                // a set that reads as partial, so the next one resumes instead of trusting it.
+                atomic_write(&done_path, blocks_done_json(&set.index_sha256, &done).as_bytes())?;
+            }
+        }
         let pending: Vec<IndexBlock> = index.blocks.iter().filter(|b| !done.contains(&b.done_key())).cloned().collect();
         ctx.reporter.begin_sub(format!("tiles {style}/{dataset}"), "blocks", index.blocks.len() as u64);
         let skipped_bytes: u64 = index.blocks.iter().filter(|b| done.contains(&b.done_key())).map(|b| b.bytes).sum();
@@ -854,6 +930,42 @@ async fn install_tiles(ctx: &Ctx<'_>, stage_index: u32, version: &str, tiles: &T
 mod tests {
     use super::*;
 
+    fn block(z: u64, bx: u64, by: u64, files: u64, digest: &str) -> IndexBlock {
+        IndexBlock { z, bx, by, files, bytes: files * 10, digest: digest.to_string() }
+    }
+
+    #[test]
+    fn a_block_the_disk_holds_short_of_is_fetched_again() {
+        let dir = std::env::temp_dir().join(format!("odian-recheck-{}", crate::copy::unique_hex()));
+        // Block (8, 2, 1) is x 128..=191, y 64..=127: two of its three tiles are here, plus a
+        // tile server negative and the installer's marker, neither of which is a tile.
+        // Block (8, 0, 0) has its one tile and one this PC baked itself: more is still whole.
+        for rel in ["8/128/64.png", "8/129/64.png", "8/129/65.png.missing", "8/1/1.png", "8/2/2.png"] {
+            let path = dir.join(rel);
+            std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+            std::fs::write(&path, b"t").unwrap();
+        }
+        std::fs::write(dir.join(DONE_MARKER), b"{}").unwrap();
+
+        let on_disk = tiles_on_disk_per_block(&dir);
+        assert_eq!(on_disk.get(&(8, 2, 1)), Some(&2));
+        assert_eq!(on_disk.get(&(8, 0, 0)), Some(&2));
+
+        let blocks = [block(8, 2, 1, 3, "aa"), block(8, 0, 0, 1, "bb"), block(9, 0, 0, 4, "cc")];
+        let mut done: BTreeSet<String> = blocks.iter().map(IndexBlock::done_key).collect();
+        // The short block and the one with nothing on disk go; the whole one stays.
+        assert_eq!(forget_short_blocks(&mut done, &blocks, &on_disk), 2);
+        assert_eq!(done, ["8/0/0:bb".to_string()].into_iter().collect::<BTreeSet<_>>());
+        // A block that was never marked done is not "forgotten": it was already pending.
+        assert_eq!(forget_short_blocks(&mut done, &blocks, &on_disk), 0);
+        std::fs::remove_dir_all(&dir).unwrap();
+    }
+
+    #[test]
+    fn an_empty_or_missing_folder_counts_nothing() {
+        assert!(tiles_on_disk_per_block(Path::new(r"Z:\no\such\set")).is_empty());
+    }
+
     #[test]
     fn progress_files_match_python_json_dumps() {
         let verified: BTreeSet<u64> = [2, 0, 1].into_iter().collect();