ODIAN/launcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
Kelorastus
2026-09-23 16:05:19 +02:00
parent 4a92cc0b63
commit 4d620fc5ad
27 changed files with 3571 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/odian-packs/src/install.rs
+275 -19
View File
@@ -1,6 +1,6 @@
//! The installer (fetch.py's `fetch`): verified manifest in, verified install out.
use std::collections::BTreeSet;
use std::collections::{BTreeSet, HashMap};
use std::future::Future;
use std::path::{Path, PathBuf};
use std::sync::Arc;
@@ -15,7 +15,7 @@ use tokio_util::sync::CancellationToken;
use crate::copy::{atomic_write, copy_body, with_suffix};
use crate::error::{Error, IntegrityKind, Result};
use crate::hashing::{chunk_range, hash_file_prefix, hex, sha256_file};
use crate::manifest::{safe_relpath, AppFile, DbFile, SrcFile, SrcSection, TilesSection, VerifiedManifest};
use crate::manifest::{safe_relpath, AppFile, DbFile, SrcFile, SrcSection, TileSet, TilesSection, VerifiedManifest};
use crate::progress::{Progress, Reporter, Stage};
use crate::source::{blocking, Backend, Pack, Source};
use crate::tileblocks::{copy_drive_block, fetch_site_block, DriveBlock};
@@ -28,11 +28,19 @@ pub struct InstallOptions {
pub workers: usize,
/// Stages to run; `App`, `Db`, `Src` and `Tiles` are meaningful, others are ignored.
pub only: Vec<Stage>,
/// The database files this PC needs, by manifest path; `None` is all of them. A PC that
/// keeps its own database needs only the dumps of tables it lacks, not the whole snapshot.
pub db_paths: Option<BTreeSet<String>>,
/// The tile sets to install, as `style/dataset`; `None` is all of them. The launcher
/// leaves out the marketplace sets the operator did not choose.
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>>,
}
impl Default for InstallOptions {
fn default() -> Self {
InstallOptions { workers: 6, only: vec![Stage::App, Stage::Db, Stage::Src, Stage::Tiles] }
InstallOptions { workers: 6, only: vec![Stage::App, Stage::Db, Stage::Src, Stage::Tiles], db_paths: None, tile_sets: None, src_paths: None }
}
}
@@ -41,6 +49,8 @@ pub struct AppSummary {
pub files: usize,
pub downloaded: usize,
pub skipped: usize,
/// Identical files taken from another installed version instead of downloaded.
pub reused: usize,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
@@ -114,6 +124,7 @@ where
let backend = Backend::new(source)?;
let m = manifest.manifest();
let version = m.version.clone();
backend.use_version(&version);
let wants = |stage: Stage| opts.only.contains(&stage);
let do_app = wants(Stage::App);
@@ -146,23 +157,29 @@ where
if do_app {
check_cancel()?;
stage_index += 1;
summary.app = Some(install_app(&ctx, stage_index, &m.app.files, &version_dir).await?);
let (dest, current) = (dest.to_path_buf(), version.clone());
let reuse = blocking(move || Ok(installed_app_files(&dest, &current))).await?;
summary.app = Some(install_app(&ctx, stage_index, &m.app.files, &reuse, &version_dir).await?);
}
if let (true, Some(db)) = (do_db, &m.db) {
check_cancel()?;
stage_index += 1;
let (files, downloaded, skipped) = install_db(&ctx, stage_index, &db.files, &data_dir).await?;
let (files, downloaded, skipped) = install_db(&ctx, stage_index, &db.files, opts.db_paths.as_ref(), &data_dir).await?;
summary.db = Some(DbSummary { snapshot: db.snapshot.clone(), files, downloaded, skipped });
}
if let (true, Some(src)) = (do_src, &m.src) {
check_cancel()?;
stage_index += 1;
summary.src = Some(install_src(&ctx, stage_index, src, &data_dir).await?);
let keep = |f: &&SrcFile| opts.src_paths.as_ref().is_none_or(|paths| paths.contains(&f.path));
let src = SrcSection { files: src.files.iter().filter(keep).cloned().collect(), ..src.clone() };
summary.src = Some(install_src(&ctx, stage_index, &src, &data_dir).await?);
}
if let (true, Some(tiles)) = (do_tiles, &m.tiles) {
check_cancel()?;
stage_index += 1;
summary.tiles = Some(install_tiles(&ctx, stage_index, &version, tiles, &data_dir).await?);
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?);
}
check_cancel()?;
@@ -312,37 +329,103 @@ async fn download_file(ctx: &Ctx<'_>, pack: Pack<'_>, dest: &Path, size: u64, sh
))
}
async fn install_app(ctx: &Ctx<'_>, stage_index: u32, files: &[AppFile], version_dir: &Path) -> Result<AppSummary> {
/// The app files other fully installed versions hold (their manifest.json is written last),
/// by content. An update takes an unchanged file from there instead of downloading it again.
fn installed_app_files(dest: &Path, except_version: &str) -> HashMap<String, PathBuf> {
let mut found = HashMap::new();
let Ok(entries) = std::fs::read_dir(dest.join("versions")) else { return found };
for entry in entries.flatten() {
if entry.file_name().to_string_lossy() == except_version {
continue;
}
let dir = entry.path();
let Ok(raw) = std::fs::read(dir.join("manifest.json")) else { continue };
let Ok(doc) = serde_json::from_slice::<Value>(&raw) else { continue };
for file in doc["app"]["files"].as_array().into_iter().flatten() {
let (Some(path), Some(sha256)) = (file["path"].as_str(), file["sha256"].as_str()) else { continue };
if let Ok(full) = join_rel(&dir, path) {
found.entry(sha256.to_string()).or_insert(full);
}
}
}
found
}
/// Put a copy of `from` at `target`: a hard link where the volume allows one (no bytes
/// written, and nothing left behind when the old version is removed), else a real copy.
/// Best effort: the caller verifies `target` like any file on disk, and downloads over it
/// when it is missing or wrong -- that write goes to `.part` and replaces the link, so it
/// never touches `from`.
fn link_or_copy(from: &Path, target: &Path) -> bool {
if target.exists() || !from.is_file() {
return false;
}
if let Some(parent) = target.parent() {
if std::fs::create_dir_all(parent).is_err() {
return false;
}
}
std::fs::hard_link(from, target).is_ok() || std::fs::copy(from, target).is_ok()
}
async fn install_app(
ctx: &Ctx<'_>,
stage_index: u32,
files: &[AppFile],
reuse: &HashMap<String, PathBuf>,
version_dir: &Path,
) -> Result<AppSummary> {
let total: u64 = files.iter().map(|f| f.size).sum();
ctx.reporter.begin_stage(Stage::App, stage_index, total, files.len() as u64);
ctx.reporter.begin_sub("app".into(), "files", files.len() as u64);
let (mut downloaded, mut skipped) = (0, 0);
let (mut downloaded, mut skipped, mut reused) = (0, 0, 0);
run_pool(
files.iter().collect(),
ctx.workers,
&ctx.cancel,
|entry: &AppFile, stop| async move {
let target = join_rel(version_dir, &entry.path)?;
download_file(ctx, Pack::App(&entry.sha256), &target, entry.size, &entry.sha256, &stop).await
let linked = match reuse.get(&entry.sha256) {
Some(from) => {
let (from, to) = (from.clone(), target.clone());
blocking(move || Ok(link_or_copy(&from, &to))).await?
}
None => false,
};
let pack = Pack::App { sha: &entry.sha256, path: &entry.path };
let outcome = download_file(ctx, pack, &target, entry.size, &entry.sha256, &stop).await?;
Ok((outcome, linked))
},
|outcome| {
if outcome == Outcome::Downloaded { downloaded += 1 } else { skipped += 1 }
|(outcome, linked)| {
match (outcome, linked) {
(Outcome::Downloaded, _) => downloaded += 1,
(Outcome::Skipped, true) => reused += 1,
(Outcome::Skipped, false) => skipped += 1,
}
ctx.reporter.items_done(1);
Ok(())
},
)
.await?;
ctx.reporter.finish_stage();
Ok(AppSummary { files: files.len(), downloaded, skipped })
Ok(AppSummary { files: files.len(), downloaded, skipped, reused })
}
async fn install_db(ctx: &Ctx<'_>, stage_index: u32, files: &[DbFile], data_dir: &Path) -> Result<(usize, usize, usize)> {
async fn install_db(
ctx: &Ctx<'_>,
stage_index: u32,
files: &[DbFile],
only: Option<&BTreeSet<String>>,
data_dir: &Path,
) -> Result<(usize, usize, usize)> {
let files: Vec<&DbFile> = files.iter().filter(|f| only.is_none_or(|keep| keep.contains(&f.path))).collect();
let count = files.len();
let total: u64 = files.iter().map(|f| f.size).sum();
ctx.reporter.begin_stage(Stage::Db, stage_index, total, files.len() as u64);
ctx.reporter.begin_sub("db".into(), "files", files.len() as u64);
ctx.reporter.begin_stage(Stage::Db, stage_index, total, count as u64);
ctx.reporter.begin_sub("db".into(), "files", count as u64);
let (mut downloaded, mut skipped) = (0, 0);
run_pool(
files.iter().collect(),
files,
ctx.workers.min(4),
&ctx.cancel,
|entry: &DbFile, stop| async move {
@@ -357,7 +440,7 @@ async fn install_db(ctx: &Ctx<'_>, stage_index: u32, files: &[DbFile], data_dir:
)
.await?;
ctx.reporter.finish_stage();
Ok((files.len(), downloaded, skipped))
Ok((count, downloaded, skipped))
}
// ---- source files -----------------------------------------------------------------
@@ -495,13 +578,186 @@ async fn fetch_chunk(ctx: &Ctx<'_>, entry: &SrcFile, chunk_bytes: u64, dest: &Pa
// ---- tiles --------------------------------------------------------------------------
/// The installer's record of which blocks of a set a PC holds, beside the set's tiles.
const DONE_MARKER: &str = ".blocks-done.json";
/// `{"index_sha256": "...", "blocks": ["z/bx/by:digest", ...]}`, as Python's `json.dumps` writes it.
fn blocks_done_json(index_sha256: &str, done: &BTreeSet<String>) -> String {
let list: Vec<String> = done.iter().map(|k| format!("\"{k}\"")).collect();
format!("{{\"index_sha256\": \"{index_sha256}\", \"blocks\": [{}]}}", list.join(", "))
}
/// Tile sets from another install, copied as they lie.
///
/// An install keeps no per-block index (only a site store has one), so the block-by-block
/// verification of a download has nothing to check against here. What makes this copy
/// trustworthy instead is every single file: its bytes are hashed at the source and read back
/// from the disk they landed on, and a mismatch fails the install rather than being installed.
/// The source's own record of which blocks it holds (`.blocks-done.json`) travels with the
/// tiles, so the PC knows what it has and an update fills in only the rest. Tiles the source
/// drew itself, beyond the blocks, come along: they are valid tiles of the same set.
async fn copy_install_tiles(
ctx: &Ctx<'_>,
stage_index: u32,
tiles: &TilesSection,
root: &Path,
data_dir: &Path,
) -> 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();
ctx.reporter.begin_stage(Stage::Tiles, stage_index, total, files);
let mut out = Vec::with_capacity(tiles.sets.len());
for set in &tiles.sets {
let (style, dataset) = (set.style.as_str(), set.dataset.as_str());
let from = root.join("data").join("tiles").join(style).join(dataset);
let to = data_dir.join("tiles").join(style).join(dataset);
// 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) {
ctx.reporter.begin_sub(format!("tiles {style}/{dataset}"), "files", set.files);
ctx.reporter.items_done(set.files);
ctx.reporter.add_done_bytes(set.bytes);
out.push(TileSetSummary {
set: format!("{style}/{dataset}"),
blocks: set.blocks as usize,
downloaded: 0,
files: set.files,
bytes: set.bytes,
});
continue;
}
if !from.is_dir() {
return Err(Error::NotFound { what: format!("{}: tiles {style}/{dataset}", from.display()) });
}
ctx.reporter.begin_sub(format!("tiles {style}/{dataset}"), "files", set.files);
let (mut copied, mut seen, mut bytes) = (0usize, 0usize, 0u64);
// The marker is the claim that the whole set is here, so it is written last, not
// copied somewhere in the middle of the walk: a copy stopped halfway must leave a
// set that still reads as partial, and resume rather than be trusted whole.
let marker = from.join(DONE_MARKER);
let mut folders = vec![from.clone()];
while let Some(folder) = folders.pop() {
if ctx.cancel.is_cancelled() {
return Err(Error::Cancelled);
}
let (batch, mut deeper) = {
let (folder, from, to) = (folder.clone(), from.clone(), to.clone());
blocking(move || list_folder(&folder, &from, &to)).await?
};
folders.append(&mut deeper);
let batch: Vec<(PathBuf, PathBuf)> = batch.into_iter().filter(|(file, _)| *file != marker).collect();
seen += batch.len();
let counted = ctx.reporter.clone();
run_pool(
batch,
ctx.workers,
&ctx.cancel,
|pair: (PathBuf, PathBuf), stop| {
let reporter = ctx.reporter.clone();
async move { blocking(move || copy_checked(&pair.0, &pair.1, &reporter, &stop)).await }
},
|(size, fresh)| {
bytes += size;
copied += usize::from(fresh);
counted.items_done(1);
Ok(())
},
)
.await?;
}
// Every file of the set landed and was read back: what the source recorded about it
// is now true here too.
if marker.is_file() {
let raw = blocking({
let marker = marker.clone();
move || std::fs::read(&marker).map_err(Error::io(&marker))
})
.await?;
atomic_write(&to.join(DONE_MARKER), &raw)?;
}
out.push(TileSetSummary {
set: format!("{style}/{dataset}"),
blocks: set.blocks as usize,
downloaded: copied,
files: seen as u64,
bytes,
});
}
ctx.reporter.finish_stage();
Ok(out)
}
/// Whether `dir` holds the whole of `set` by the installer's own record: the marker names
/// this index and lists every block of it. Written by a download (each block verified against
/// the site's index) or by a copy (every file hashed in and read back), so either way the set
/// it describes is this set.
fn set_recorded_complete(dir: &Path, set: &TileSet) -> bool {
let Ok(raw) = std::fs::read(dir.join(DONE_MARKER)) else {
return false;
};
let Ok(saved) = serde_json::from_slice::<Value>(&raw) else {
return false;
};
saved.get("index_sha256").and_then(Value::as_str) == Some(set.index_sha256.as_str())
&& saved.get("blocks").and_then(Value::as_array).is_some_and(|blocks| blocks.len() as u64 >= set.blocks)
}
/// One folder of a tile set: the files in it (source and destination), and the folders under it.
fn list_folder(folder: &Path, from: &Path, to: &Path) -> Result<(Vec<(PathBuf, PathBuf)>, Vec<PathBuf>)> {
let mut files = Vec::new();
let mut folders = Vec::new();
for entry in std::fs::read_dir(folder).map_err(Error::io(folder))? {
let entry = entry.map_err(Error::io(folder))?;
let path = entry.path();
let kind = entry.file_type().map_err(Error::io(&path))?;
if kind.is_dir() {
folders.push(path);
} else if kind.is_file() {
let rel = path.strip_prefix(from).map_err(|_| Error::io(&path)(std::io::Error::other("outside the set")))?;
files.push((path.clone(), to.join(rel)));
}
}
Ok((files, folders))
}
/// Copy one file and prove it: the source's bytes are hashed, and so are the bytes read back
/// from where they landed. Returns its size and whether this run wrote it (a file already
/// there and identical is left alone, so a copy that stopped resumes without redoing it).
fn copy_checked(from: &Path, to: &Path, reporter: &Reporter, cancel: &CancellationToken) -> Result<(u64, bool)> {
let (size, want) = sha256_file(from, cancel)?;
let there = std::fs::metadata(to).is_ok_and(|md| md.is_file() && md.len() == size)
&& sha256_file(to, cancel).is_ok_and(|(_, got)| got == want);
if there {
reporter.add_done_bytes(size);
return Ok((size, false));
}
if let Some(parent) = to.parent() {
std::fs::create_dir_all(parent).map_err(Error::io(parent))?;
}
let part = with_suffix(to, ".part");
std::fs::copy(from, &part).map_err(Error::io(&part))?;
let (_, got) = sha256_file(&part, cancel)?;
if got != want {
let _ = std::fs::remove_file(&part);
return Err(Error::integrity(
IntegrityKind::FileHash,
from.display().to_string(),
format!("copied as {got}, expected {want}"),
));
}
std::fs::rename(&part, to).map_err(Error::io(to))?;
reporter.add_transferred(size);
Ok((size, true))
}
async fn install_tiles(ctx: &Ctx<'_>, stage_index: u32, version: &str, tiles: &TilesSection, data_dir: &Path) -> 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;
}
let total: u64 = tiles.sets.iter().map(|s| s.bytes).sum();
let blocks: u64 = tiles.sets.iter().map(|s| s.blocks).sum();
ctx.reporter.begin_stage(Stage::Tiles, stage_index, total, blocks);
@@ -523,7 +779,7 @@ async fn install_tiles(ctx: &Ctx<'_>, stage_index: u32, version: &str, tiles: &T
let index = verify_index(&raw, &set.index_sha256, &what)?;
let set_dir = data_dir.join("tiles").join(style).join(dataset);
let done_path = set_dir.join(".blocks-done.json");
let done_path = set_dir.join(DONE_MARKER);
let mut done: BTreeSet<String> = BTreeSet::new();
if let Some(saved) = std::fs::read(&done_path).ok().and_then(|raw| serde_json::from_slice::<Value>(&raw).ok()) {
if saved.get("index_sha256").and_then(Value::as_str) == Some(set.index_sha256.as_str()) {
crates/odian-packs/src/manifest.rs
+78
View File
@@ -194,6 +194,24 @@ fn object<'a>(value: &'a Value, message: &str) -> Result<&'a Map<String, Value>>
/// Error unless `manifest` is a well-formed schema-1 manifest. Structure only; the
/// signature is checked by [`crate::signing::verify`].
/// The marketplace fields a tile set or source file may carry (manifest.py `_check_catalog`).
/// An item without `"optional": true` is part of every install.
const CATALOG_TEXT_LIMITS: [(&str, usize); 2] = [("title", 80), ("description", 1000)];
fn check_catalog(entry: &Map<String, Value>, place: &str) -> Result<()> {
if let Some(optional) = entry.get("optional") {
require(optional.is_boolean(), || format!("{place}: optional must be true or false"))?;
}
for (key, limit) in CATALOG_TEXT_LIMITS {
if let Some(text) = entry.get(key) {
// Characters, as Python's len() counts them.
let ok = text.as_str().is_some_and(|s| !s.trim().is_empty() && s.chars().count() <= limit);
require(ok, || format!("{place}: {key} must be text of 1 to {limit} characters"))?;
}
}
Ok(())
}
pub fn validate(manifest: &Value) -> Result<()> {
let m = object(manifest, "manifest must be a JSON object")?;
require(m.get("schema").and_then(Value::as_u64) == Some(SCHEMA), || format!("schema must be {SCHEMA}"))?;
@@ -280,6 +298,7 @@ pub fn validate(manifest: &Value) -> Result<()> {
require(entry.get("index").and_then(Value::as_str) == Some(expected.as_str()), || {
format!("{place}: index must be {expected:?}")
})?;
check_catalog(entry, &place)?;
sets_seen.insert((style, dataset));
}
}
@@ -309,6 +328,7 @@ pub fn validate(manifest: &Value) -> Result<()> {
require(entry.get("chunks_sha256").and_then(Value::as_str) == Some(chunks_digest(&hashes).as_str()), || {
format!("{place}: chunks_sha256 does not match chunks")
})?;
check_catalog(entry, &place)?;
}
}
@@ -377,6 +397,13 @@ pub struct TileSet {
pub bytes: u64,
pub index: String,
pub index_sha256: String,
/// Offered in the launcher's marketplace instead of installed with everything else.
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub optional: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub title: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -391,6 +418,13 @@ pub struct SrcFile {
pub size: u64,
pub chunks: Vec<String>,
pub chunks_sha256: String,
/// Offered in the launcher's marketplace instead of installed with everything else.
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub optional: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub title: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -481,6 +515,50 @@ mod tests {
}
}
#[test]
fn marketplace_fields_are_checked_like_python_and_default_to_part_of_the_install() {
let chunks = [hex(&Sha256::digest(b"c0"))];
let manifest = |set_extra: Value, src_extra: Value| {
let mut set = serde_json::json!({
"style": "dark_v2", "dataset": "ds", "journal_seq": 1, "blocks": 1, "files": 1, "bytes": 9,
"index": tile_index_name("dark_v2", "ds"), "index_sha256": "a".repeat(64),
});
let mut src = serde_json::json!({
"path": "x.pmtiles", "size": 5, "chunks": chunks, "chunks_sha256": chunks_digest(&[chunks[0].as_str()]),
});
set.as_object_mut().unwrap().extend(set_extra.as_object().unwrap().clone());
src.as_object_mut().unwrap().extend(src_extra.as_object().unwrap().clone());
serde_json::json!({
"schema": 1, "product": "odian", "version": "0.61.0", "released_at": "2026-09-22T12:00:00Z",
"app": {"files": []},
"tiles": {"block_shift": BLOCK_SHIFT, "sets": [set]},
"src": {"enabled": true, "chunk_bytes": 10, "files": [src]},
})
};
let good = manifest(
serde_json::json!({"optional": true, "title": "Dark V2 map", "description": "Buildings and forests."}),
serde_json::json!({"optional": true, "title": "Map source"}),
);
validate(&good).unwrap();
let typed: Manifest = serde_json::from_value(good).unwrap();
let (set, src) = (&typed.tiles.as_ref().unwrap().sets[0], &typed.src.as_ref().unwrap().files[0]);
assert!(set.optional && src.optional);
assert_eq!((set.title.as_deref(), src.description.as_deref()), (Some("Dark V2 map"), None));
let plain: Manifest = serde_json::from_value(manifest(serde_json::json!({}), serde_json::json!({}))).unwrap();
assert!(!plain.tiles.unwrap().sets[0].optional, "no field: part of every install");
// A 1000-character description passes however many bytes its letters take.
validate(&manifest(serde_json::json!({"description": "é".repeat(1000)}), serde_json::json!({}))).unwrap();
for (set_extra, src_extra) in [
(serde_json::json!({"optional": "yes"}), serde_json::json!({})),
(serde_json::json!({"title": " "}), serde_json::json!({})),
(serde_json::json!({}), serde_json::json!({"description": "x".repeat(1001)})),
(serde_json::json!({}), serde_json::json!({"title": 7})),
] {
assert!(validate(&manifest(set_extra.clone(), src_extra.clone())).is_err(), "{set_extra} {src_extra}");
}
}
#[test]
fn versions() {
assert_eq!(parse_version("0.51.0").unwrap(), (0, 51, 0));
crates/odian-packs/src/source.rs
+125 -24
View File
@@ -1,5 +1,6 @@
//! Where a version comes from: a site's update service over HTTP, or an ODIAN drive that
//! holds the site store layout (`packs/`, `tiles/`, `src/`).
//! Where a version comes from: a site's update service over HTTP, an ODIAN drive that holds
//! the site store layout (`packs/`, `tiles/`, `src/`), or another ODIAN install, whose files
//! are copied out of it (a drive an operator carries holds one of those, not a store).
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
@@ -21,6 +22,9 @@ pub enum Source {
Site { base_url: String, token: String },
/// An ODIAN drive (SSD) whose root holds `packs/`, `tiles/` (with `tiles.db`) and `src/`.
Drive { root: PathBuf },
/// Another ODIAN install (`versions\<v>\`, `data\`), e.g. the one a drive carries. Its
/// files are copied, and checked against the signed manifest of the version it holds.
Install { root: PathBuf },
}
const RETRIES: u32 = 4;
@@ -97,7 +101,9 @@ pub(crate) enum Body {
#[derive(Clone, Copy)]
pub(crate) enum Pack<'a> {
/// `app/<sha256>`.
App(&'a str),
/// By content where the store is content-addressed (a site, a drive), by path inside
/// the version folder of another install.
App { sha: &'a str, path: &'a str },
/// A manifest `db.files[].path` (`db/<snapshot>/...`).
Db(&'a str),
}
@@ -112,9 +118,16 @@ pub(crate) struct Drive {
root: PathBuf,
}
pub(crate) struct Install {
root: PathBuf,
/// The version whose manifest was read: where its app files are. Set when the manifest is.
version: Mutex<String>,
}
pub(crate) enum Backend {
Site(Site),
Drive(Drive),
Install(Install),
}
impl Backend {
@@ -130,6 +143,9 @@ impl Backend {
Ok(Backend::Site(Site { base: base_url.trim_end_matches('/').to_string(), token: token.clone(), client }))
}
Source::Drive { root } => Ok(Backend::Drive(Drive { root: root.clone() })),
Source::Install { root } => {
Ok(Backend::Install(Install { root: root.clone(), version: Mutex::new(String::new()) }))
}
}
}
@@ -151,6 +167,12 @@ impl Backend {
.map_err(|_| Error::manifest(format!("{} does not hold a valid version", pointer.display())))?;
drive.read(join_rel(&drive.root, &format!("packs/versions/{text}/manifest.json"))).await
}
Backend::Install(install) => {
let root = install.root.clone();
let (version, raw) = blocking(move || newest_installed_version(&root)).await?;
install.set_version(&version);
Ok(raw)
}
}
}
@@ -159,6 +181,11 @@ impl Backend {
match self {
Backend::Site(site) => site.get_bytes(&format!("/v1/versions/{version}/manifest"), cancel).await,
Backend::Drive(drive) => drive.read(join_rel(&drive.root, &format!("packs/versions/{version}/manifest.json"))).await,
Backend::Install(install) => {
let raw = read_file(install.root.join("versions").join(version).join("manifest.json")).await?;
install.set_version(version);
Ok(raw)
}
}
}
@@ -167,6 +194,8 @@ impl Backend {
match self {
Backend::Site(site) => site.get_bytes(&format!("/v1/versions/{version}/notes"), cancel).await,
Backend::Drive(drive) => drive.read(join_rel(&drive.root, &format!("packs/versions/{version}/notes.md"))).await,
// An install keeps its notes with its own state, not with the version.
Backend::Install(_) => Err(Error::NotFound { what: format!("notes for {version}") }),
}
}
@@ -180,13 +209,18 @@ impl Backend {
let rel = format!("packs/versions/{version}/{}", tile_index_name(style, dataset));
drive.read(join_rel(&drive.root, &rel)).await
}
// An install keeps no index: its tile sets are copied as they lie (see install_tiles).
Backend::Install(_) => {
Err(Error::NotFound { what: format!("tile index {style}/{dataset} of {version}") })
}
}
}
pub fn pack_label(&self, pack: Pack<'_>) -> String {
match pack {
Pack::App(sha) => format!("/packs/app/{sha}"),
Pack::Db(path) => format!("/packs/{path}"),
match (self, pack) {
(Backend::Install(_), Pack::App { path, .. }) => path.to_string(),
(_, Pack::App { sha, .. }) => format!("/packs/app/{sha}"),
(_, Pack::Db(path)) => format!("/packs/{path}"),
}
}
@@ -196,7 +230,7 @@ impl Backend {
match self {
Backend::Site(site) => {
let path = match pack {
Pack::App(sha) => format!("/packs/app/{}", encode_path(sha)),
Pack::App { sha, .. } => format!("/packs/app/{}", encode_path(sha)),
Pack::Db(rel) => format!("/packs/{}", encode_path(rel)),
};
let range = (offset > 0).then(|| format!("bytes={offset}-"));
@@ -206,10 +240,18 @@ impl Backend {
}
Backend::Drive(drive) => {
let path = match pack {
Pack::App(sha) => drive.root.join("packs").join("app").join(sha),
Pack::App { sha, .. } => drive.root.join("packs").join("app").join(sha),
Pack::Db(rel) => join_rel(&drive.root.join("packs"), rel),
};
let body = drive.open_range(path, offset, u64::MAX).await?;
let body = open_range(path, offset, u64::MAX).await?;
Ok((body, true))
}
Backend::Install(install) => {
let path = match pack {
Pack::App { path, .. } => join_rel(&install.version_dir(), path),
Pack::Db(rel) => join_rel(&install.root.join("data"), rel),
};
let body = open_range(path, offset, u64::MAX).await?;
Ok((body, true))
}
}
@@ -230,25 +272,95 @@ impl Backend {
}
Ok(Body::Http { response, url })
}
Backend::Drive(drive) => drive.open_range(join_rel(&drive.root.join("src"), rel), start, end - start + 1).await,
Backend::Drive(drive) => open_range(join_rel(&drive.root.join("src"), rel), start, end - start + 1).await,
Backend::Install(install) => {
open_range(join_rel(&install.root.join("data").join("src"), rel), start, end - start + 1).await
}
}
}
/// Which version's files to copy, when the source is another install: the manifest is
/// read through a backend of its own, so the one doing the work is told separately.
pub fn use_version(&self, version: &str) {
if let Backend::Install(install) = self {
install.set_version(version);
}
}
/// The root of another install this one is being copied from.
pub fn install_root(&self) -> Option<&Path> {
match self {
Backend::Install(install) => Some(&install.root),
_ => None,
}
}
pub fn drive_root(&self) -> Option<&Path> {
match self {
Backend::Drive(drive) => Some(&drive.root),
Backend::Site(_) => None,
_ => None,
}
}
pub fn site(&self) -> Option<&Site> {
match self {
Backend::Site(site) => Some(site),
Backend::Drive(_) => None,
_ => None,
}
}
}
impl Install {
fn set_version(&self, version: &str) {
*self.version.lock().unwrap_or_else(|e| e.into_inner()) = version.to_string();
}
fn version_dir(&self) -> PathBuf {
let version = self.version.lock().unwrap_or_else(|e| e.into_inner()).clone();
self.root.join("versions").join(version)
}
}
/// The newest whole version another install carries. `versions\<v>\manifest.json` is written
/// last, so a folder without one is a download that stopped part way and is not offered.
fn newest_installed_version(root: &Path) -> Result<(String, Vec<u8>)> {
let versions = root.join("versions");
let entries = std::fs::read_dir(&versions).map_err(open_error(&versions))?;
let mut best: Option<((u64, u64, u64), String)> = None;
for entry in entries.flatten() {
let name = entry.file_name().to_string_lossy().into_owned();
let Ok(parsed) = parse_version(&name) else { continue };
if !entry.path().join("manifest.json").is_file() {
continue;
}
if best.as_ref().is_none_or(|(best, _)| *best < parsed) {
best = Some((parsed, name));
}
}
let version = best
.ok_or_else(|| Error::NotFound { what: format!("{}: no ODIAN version installed here", versions.display()) })?
.1;
let path = versions.join(&version).join("manifest.json");
let raw = std::fs::read(&path).map_err(open_error(&path))?;
Ok((version, raw))
}
pub(crate) async fn read_file(path: PathBuf) -> Result<Vec<u8>> {
blocking(move || std::fs::read(&path).map_err(open_error(&path))).await
}
/// `len` bytes from `start` of a local file.
pub(crate) async fn open_range(path: PathBuf, start: u64, len: u64) -> Result<Body> {
blocking(move || {
let mut file = std::fs::File::open(&path).map_err(open_error(&path))?;
if start > 0 {
file.seek(SeekFrom::Start(start)).map_err(Error::io(&path))?;
}
Ok(Body::File { reader: file.take(len), path })
})
.await
}
impl Site {
/// GET with the Bearer token. Retries 502/503/504 (honouring Retry-After) and
/// connection errors with a doubling delay, like fetch.py's `Site.open`.
@@ -326,18 +438,7 @@ impl Site {
impl Drive {
async fn read(&self, path: PathBuf) -> Result<Vec<u8>> {
blocking(move || std::fs::read(&path).map_err(open_error(&path))).await
}
async fn open_range(&self, path: PathBuf, start: u64, len: u64) -> Result<Body> {
blocking(move || {
let mut file = std::fs::File::open(&path).map_err(open_error(&path))?;
if start > 0 {
file.seek(SeekFrom::Start(start)).map_err(Error::io(&path))?;
}
Ok(Body::File { reader: file.take(len), path })
})
.await
read_file(path).await
}
}
crates/odian-packs/tests/e2e_site.rs
+180 -1
View File
@@ -13,7 +13,7 @@
//!
//! `ODIAN_PACKS_PYTHON` picks the interpreter (default `python`).
use std::collections::BTreeMap;
use std::collections::{BTreeMap, BTreeSet};
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
@@ -290,6 +290,141 @@ async fn site_install_verifies_everything_then_rerun_downloads_nothing() {
assert!(matches!(version_manifest(&site, "9.9.9", &key).await, Err(Error::NotFound { .. })));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "spawns Python: set ODIAN_PACKS_PYTHONPATH and run with --ignored (see the file header)"]
async fn an_update_reuses_the_installed_files_and_fetches_only_the_db_files_asked_for() {
let stub = Stub::start(&[]);
let (site, key, dest) = (stub.site(), stub.key(), stub.dest("install"));
install_from(&site, &dest, &key, only(&[Stage::App])).await.unwrap();
// What is installed becomes an older version: installing the current one is an update.
let old = dest.join("versions").join("0.50.0");
std::fs::rename(dest.join("versions").join("0.51.0"), &old).unwrap();
// One old file went bad on disk: taken from there anyway, it must be caught.
let m = stub.store_manifest();
let bad = m["app"]["files"].as_array().unwrap().iter().find(|f| f["size"].as_u64().unwrap() > 0).unwrap();
let bad_path = bad["path"].as_str().unwrap().to_string();
flip(&old.join(&bad_path), 0);
let bad_old = std::fs::read(old.join(&bad_path)).unwrap();
stub.ctl("/_ctl/reset").await;
let first_db = m["db"]["files"][0]["path"].as_str().unwrap().to_string();
let opts = InstallOptions {
only: vec![Stage::App, Stage::Db],
db_paths: Some([first_db.clone()].into_iter().collect()),
..InstallOptions::default()
};
let summary = install_from(&site, &dest, &key, opts).await.unwrap();
let app = summary.app.unwrap();
assert_eq!((app.files, app.downloaded, app.reused, app.skipped), (5, 1, 4, 0), "{app:?}");
let app_requests: Vec<String> =
stub.requests().await.into_iter().map(|(p, _)| p).filter(|p| p.starts_with("/packs/app/")).collect();
assert_eq!(app_requests, vec![format!("/packs/app/{}", stub.app_sha(&bad_path))], "only the bad file is downloaded");
let new = dest.join("versions").join("0.51.0");
for f in m["app"]["files"].as_array().unwrap() {
let path = new.join(f["path"].as_str().unwrap());
assert_eq!(sha256_hex(&std::fs::read(&path).unwrap()), f["sha256"].as_str().unwrap(), "{}", path.display());
}
// The download replaced the link; it never wrote through it into the old version.
assert_eq!(std::fs::read(old.join(&bad_path)).unwrap(), bad_old);
let db = summary.db.unwrap();
assert_eq!((db.files, db.downloaded), (1, 1));
for f in m["db"]["files"].as_array().unwrap() {
let path = f["path"].as_str().unwrap();
assert_eq!(dest.join("data").join(path).is_file(), path == first_db, "{path}");
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "spawns Python: set ODIAN_PACKS_PYTHONPATH and run with --ignored (see the file header)"]
async fn only_the_chosen_tile_sets_and_source_files_are_installed() {
let stub = Stub::start(&[]);
let (site, key, dest) = (stub.site(), stub.key(), stub.dest("install"));
let opts = InstallOptions {
only: vec![Stage::Src, Stage::Tiles],
tile_sets: Some(BTreeSet::new()),
src_paths: Some(["osm_source/small.bin".to_string()].into_iter().collect()),
..InstallOptions::default()
};
let summary = install_from(&site, &dest, &key, opts).await.unwrap();
let src: Vec<String> = summary.src.unwrap().into_iter().map(|s| s.path).collect();
assert_eq!(src, ["osm_source/small.bin"]);
assert!(summary.tiles.unwrap().is_empty());
assert!(dest.join("data/src/osm_source/small.bin").is_file());
assert!(!dest.join("data/src/osm_source/world_aeroway_v1.sqlite").exists());
assert!(!dest.join("data/tiles/dark/test_ds").exists(), "no tile set was chosen");
let log = stub.requests().await;
assert!(!log.iter().any(|(p, _)| p.contains("/block/")), "no tile block fetched: {log:?}");
// Choosing the tile set later adds it; what is installed stays as it is.
let opts = InstallOptions {
only: vec![Stage::Src, Stage::Tiles],
tile_sets: Some(["dark/test_ds".to_string()].into_iter().collect()),
src_paths: Some(["osm_source/small.bin".to_string()].into_iter().collect()),
..InstallOptions::default()
};
let summary = install_from(&site, &dest, &key, opts).await.unwrap();
assert_eq!(summary.tiles.unwrap()[0].downloaded, 4);
assert_eq!(summary.src.unwrap()[0].chunks, 0, "the source file was already there");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "spawns Python: set ODIAN_PACKS_PYTHONPATH and run with --ignored (see the file header)"]
async fn a_pc_installs_by_copying_another_install_and_checks_every_file() {
let stub = Stub::start(&[]);
let (site, key) = (stub.site(), stub.key());
let first = stub.dest("first");
install_from(&site, &first, &key, InstallOptions::default()).await.unwrap();
// The drive an operator carries holds that install, not a store: the second PC copies it.
stub.ctl("/_ctl/reset").await;
let drive = Source::Install { root: first.clone() };
let second = stub.dest("second");
let summary = install_from(&drive, &second, &key, InstallOptions::default()).await.unwrap();
assert!(stub.requests().await.is_empty(), "nothing was asked of the site");
assert_eq!(summary.version, "0.51.0");
assert_eq!(files(&second.join("versions").join("0.51.0")), files(&first.join("versions").join("0.51.0")));
assert_eq!(files(&second.join("data")), files(&first.join("data")), "same data, tile by tile");
let done = second.join("data/tiles/dark/test_ds/.blocks-done.json");
assert!(done.is_file(), "the blocks the source holds travel with the tiles");
// Running it again copies nothing: what is already there is recognised.
let again = install_from(&drive, &second, &key, InstallOptions::default()).await.unwrap();
assert_eq!(again.app.unwrap().downloaded, 0);
assert_eq!(again.tiles.unwrap().iter().map(|t| t.downloaded).sum::<usize>(), 0);
// A set this PC already recorded as complete is not read again: the marker names this
// index and every block of it, so what lies here IS the set. Hundreds of thousands of
// files are not hashed twice to conclude "identical", and a tile that changes on the
// drive afterwards stays there, as a site update leaves a complete set alone too.
let on_drive = first.join("data/tiles/dark/test_ds/0/0/0.png");
let here = second.join("data/tiles/dark/test_ds/0/0/0.png");
let kept = std::fs::read(&here).unwrap();
flip(&on_drive, 3);
install_from(&drive, &second, &key, only(&[Stage::Tiles])).await.unwrap();
assert_eq!(std::fs::read(&here).unwrap(), kept, "the set was recorded complete: it was left alone");
flip(&on_drive, 3);
// An app file that went bad on the drive is caught: the manifest says what it should be.
let third = stub.dest("third");
flip(&first.join("versions").join("0.51.0").join("Odian").join("Odian.exe"), 3);
let what = integrity(install_from(&drive, &third, &key, InstallOptions::default()).await, IntegrityKind::FileHash);
assert!(what.contains("Odian.exe"), "{what}");
// A TILE cannot be checked that way: only a site's index carries per-tile hashes, and an
// install keeps none. What is checked is the copy - the bytes that land are the bytes read.
let fourth = stub.dest("fourth");
let tile = first.join("data/tiles/dark/test_ds/0/0/0.png");
flip(&tile, 3);
install_from(&drive, &fourth, &key, only(&[Stage::Tiles])).await.unwrap();
assert_eq!(
std::fs::read(fourth.join("data/tiles/dark/test_ds/0/0/0.png")).unwrap(),
std::fs::read(&tile).unwrap(),
"faithful to the drive, whatever the drive holds"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "spawns Python: set ODIAN_PACKS_PYTHONPATH and run with --ignored (see the file header)"]
async fn wrong_token_is_unauthorized_and_edited_manifest_is_a_signature_error() {
@@ -480,6 +615,50 @@ async fn cancel_stops_and_keeps_verified_work() {
assert!(dest.join("versions/0.51.0/manifest.json").exists());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "spawns Python: set ODIAN_PACKS_PYTHONPATH and run with --ignored (see the file header)"]
async fn a_copy_stopped_halfway_leaves_the_set_partial() {
let stub = Stub::start(&[]);
let (site, key) = (stub.site(), stub.key());
let drive_install = stub.dest("drive");
install_from(&site, &drive_install, &key, InstallOptions::default()).await.unwrap();
let marker = ".blocks-done.json";
let carried = drive_install.join("data/tiles/dark/test_ds");
assert!(carried.join(marker).is_file(), "the drive holds the whole set");
// A set of a few tiles copies faster than the first progress report: give the drive enough
// of them (tiles drawn beyond the index, which travel too) to be stopped in the middle.
for x in 0..40 {
let dir = carried.join("9").join(x.to_string());
std::fs::create_dir_all(&dir).unwrap();
for y in 0..10 {
std::fs::write(dir.join(format!("{y}.png")), vec![b't'; 4096]).unwrap();
}
}
// Stop the copy in the middle of the tiles.
let drive = Source::Install { root: drive_install.clone() };
let manifest = current_manifest(&drive, &key).await.unwrap();
let dest = stub.dest("half");
let cancel = CancellationToken::new();
let trigger = cancel.clone();
let result = install(&drive, &dest, &manifest, only(&[Stage::Tiles]), move |p| {
if p.stage == Stage::Tiles && p.done_items >= 1 {
trigger.cancel();
}
}, cancel)
.await;
assert!(matches!(result, Err(Error::Cancelled)), "{result:?}");
let set_dir = dest.join("data/tiles/dark/test_ds");
// Tiles landed, but nothing here claims the set is whole: the next run copies the rest
// instead of trusting a record written before the files it speaks for.
assert!(!set_dir.join(marker).exists(), "a half-copied set must not read as complete");
let summary = install(&drive, &dest, &manifest, only(&[Stage::Tiles]), |_| {}, CancellationToken::new()).await.unwrap();
assert!(set_dir.join(marker).is_file(), "finished, so it is recorded");
assert!(summary.tiles.unwrap().iter().map(|t| t.downloaded).sum::<usize>() > 0, "the rest was copied");
assert_eq!(files(&set_dir), files(&carried), "same set, tile by tile");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "spawns Python: set ODIAN_PACKS_PYTHONPATH and run with --ignored (see the file header)"]
async fn drive_install_equals_site_install() {