ODIAN Launcher: hub, installer, local stack and remembered sign-in (M2)
- Tauri 2 + Svelte 5 hub in ODIAN dark: patch notes and roadmap, install and update progress, local stack status, settings with DHP and MCP addresses. - crates/odian-packs: Rust installer for pack format v1 (signed manifests, resumable verified downloads from a DHP or an ODIAN drive), byte-compatible with ODIAN_MCP_Server's Python odian_packs. - Local stack on 127.0.0.1: Postgres -> Engine API -> tile server -> Odian.exe, one Job Object, clean shutdown; database snapshot restore on a fresh PC only. - Keycloak sign-in through the DHP's MCP, remembered until Sign out with a launcher session that renews the Engine token silently. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kelorastus
@@ -0,0 +1,116 @@
|
||||
//! Canonical JSON: the one serialisation that gets hashed or signed.
|
||||
//!
|
||||
//! Byte-for-byte equal to Python's
|
||||
//! `json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")`.
|
||||
//!
|
||||
//! Object keys are sorted HERE, by code point (UTF-8 byte order is the same order).
|
||||
//! Nothing relies on `serde_json::Map` iteration order, which changes when any crate in
|
||||
//! the build enables serde_json's `preserve_order` feature.
|
||||
//!
|
||||
//! Numbers must be integers in the i64/u64 range. Floats (`1.0`, `1e3`, `-0`) and integers
|
||||
//! outside that range are refused: Python would serialise them, but no pack document holds
|
||||
//! one, and refusing them keeps float formatting out of the signed bytes.
|
||||
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
|
||||
/// Canonical bytes of `value`, or [`Error::Manifest`] if it holds a non-integer number.
|
||||
pub fn canonical_bytes(value: &Value) -> Result<Vec<u8>> {
|
||||
let mut out = Vec::with_capacity(1024);
|
||||
write_value(value, &mut out)?;
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
fn write_value(value: &Value, out: &mut Vec<u8>) -> Result<()> {
|
||||
match value {
|
||||
Value::Null => out.extend_from_slice(b"null"),
|
||||
Value::Bool(true) => out.extend_from_slice(b"true"),
|
||||
Value::Bool(false) => out.extend_from_slice(b"false"),
|
||||
Value::Number(n) => {
|
||||
if let Some(u) = n.as_u64() {
|
||||
out.extend_from_slice(u.to_string().as_bytes());
|
||||
} else if let Some(i) = n.as_i64() {
|
||||
out.extend_from_slice(i.to_string().as_bytes());
|
||||
} else {
|
||||
return Err(Error::manifest(format!(
|
||||
"number {n} is not an integer in the 64-bit range; pack documents hold integers only"
|
||||
)));
|
||||
}
|
||||
}
|
||||
Value::String(s) => write_str(s, out),
|
||||
Value::Array(items) => {
|
||||
out.push(b'[');
|
||||
for (i, item) in items.iter().enumerate() {
|
||||
if i > 0 {
|
||||
out.push(b',');
|
||||
}
|
||||
write_value(item, out)?;
|
||||
}
|
||||
out.push(b']');
|
||||
}
|
||||
Value::Object(map) => {
|
||||
let mut entries: Vec<(&String, &Value)> = map.iter().collect();
|
||||
entries.sort_unstable_by(|a, b| a.0.as_bytes().cmp(b.0.as_bytes()));
|
||||
out.push(b'{');
|
||||
for (i, (key, item)) in entries.into_iter().enumerate() {
|
||||
if i > 0 {
|
||||
out.push(b',');
|
||||
}
|
||||
write_str(key, out);
|
||||
out.push(b':');
|
||||
write_value(item, out)?;
|
||||
}
|
||||
out.push(b'}');
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Python's `encode_basestring` (the `ensure_ascii=False` escaper).
|
||||
fn write_str(s: &str, out: &mut Vec<u8>) {
|
||||
const HEX: &[u8; 16] = b"0123456789abcdef";
|
||||
out.push(b'"');
|
||||
// Escapes only ever replace ASCII bytes, and bytes of multi-byte UTF-8 sequences are
|
||||
// all >= 0x80, so a byte loop copies every other character through untouched.
|
||||
for &b in s.as_bytes() {
|
||||
match b {
|
||||
b'"' => out.extend_from_slice(b"\\\""),
|
||||
b'\\' => out.extend_from_slice(b"\\\\"),
|
||||
b'\n' => out.extend_from_slice(b"\\n"),
|
||||
b'\r' => out.extend_from_slice(b"\\r"),
|
||||
b'\t' => out.extend_from_slice(b"\\t"),
|
||||
0x08 => out.extend_from_slice(b"\\b"),
|
||||
0x0c => out.extend_from_slice(b"\\f"),
|
||||
0x00..=0x1f => {
|
||||
out.extend_from_slice(b"\\u00");
|
||||
out.push(HEX[(b >> 4) as usize]);
|
||||
out.push(HEX[(b & 0x0f) as usize]);
|
||||
}
|
||||
_ => out.push(b),
|
||||
}
|
||||
}
|
||||
out.push(b'"');
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn sorts_and_escapes_like_python() {
|
||||
let v = json!({"b": 1, "a": [1, {"d": 2, "c": 3}]});
|
||||
assert_eq!(canonical_bytes(&v).unwrap(), br#"{"a":[1,{"c":3,"d":2}],"b":1}"#);
|
||||
let v = json!({"s": "q\"b\\n\n\u{1}\u{7f}é/"});
|
||||
assert_eq!(canonical_bytes(&v).unwrap(), "{\"s\":\"q\\\"b\\\\n\\n\\u0001\u{7f}é/\"}".as_bytes());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn refuses_floats() {
|
||||
for text in ["1.0", "1e3", "-0", "[0.5]", "18446744073709551616"] {
|
||||
let v: Value = serde_json::from_str(text).unwrap();
|
||||
assert!(canonical_bytes(&v).is_err(), "{text}");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,198 @@
|
||||
//! Copy a body into a file while hashing it, off the async threads, with cancellation.
|
||||
|
||||
use std::io::{BufWriter, Read, Write};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use bytes::Bytes;
|
||||
use futures_util::StreamExt;
|
||||
use sha2::{Digest, Sha256};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
use crate::hashing::{hex, BUF};
|
||||
use crate::progress::Reporter;
|
||||
use crate::source::{blocking, Body};
|
||||
|
||||
pub(crate) struct Copied {
|
||||
/// Bytes written by this copy.
|
||||
pub written: u64,
|
||||
pub hasher: Sha256,
|
||||
/// The body held more than `limit` bytes; the excess was not written.
|
||||
pub overflow: bool,
|
||||
}
|
||||
|
||||
/// Copy `body` into `file` (opened and positioned by the caller), feeding `hasher`.
|
||||
/// At most `limit` bytes are written. On a cut-short body or cancellation, whatever was
|
||||
/// written stays on disk (a `.part` file resumes from it).
|
||||
pub(crate) async fn copy_body(
|
||||
body: Body,
|
||||
file: std::fs::File,
|
||||
file_path: PathBuf,
|
||||
hasher: Sha256,
|
||||
limit: u64,
|
||||
reporter: Arc<Reporter>,
|
||||
cancel: CancellationToken,
|
||||
) -> Result<Copied> {
|
||||
match body {
|
||||
Body::File { mut reader, path } => {
|
||||
blocking(move || {
|
||||
let mut out = BufWriter::with_capacity(BUF, file);
|
||||
let mut hasher = hasher;
|
||||
let mut buf = vec![0u8; BUF];
|
||||
let (mut written, mut overflow) = (0u64, false);
|
||||
let result = (|| {
|
||||
loop {
|
||||
if cancel.is_cancelled() {
|
||||
return Err(Error::Cancelled);
|
||||
}
|
||||
let n = reader.read(&mut buf).map_err(Error::io(&path))?;
|
||||
if n == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
let room = (limit - written).min(n as u64) as usize;
|
||||
overflow = room < n;
|
||||
out.write_all(&buf[..room]).map_err(Error::io(&file_path))?;
|
||||
hasher.update(&buf[..room]);
|
||||
written += room as u64;
|
||||
reporter.add_transferred(room as u64);
|
||||
if overflow {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
})();
|
||||
let flushed = out.flush().map_err(Error::io(&file_path));
|
||||
result?;
|
||||
flushed?;
|
||||
Ok(Copied { written, hasher, overflow })
|
||||
})
|
||||
.await
|
||||
}
|
||||
Body::Http { response, url } => {
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<Bytes>(16);
|
||||
let writer = tokio::task::spawn_blocking(move || -> Result<(u64, Sha256)> {
|
||||
let mut out = BufWriter::with_capacity(BUF, file);
|
||||
let mut hasher = hasher;
|
||||
let mut written = 0u64;
|
||||
while let Some(chunk) = rx.blocking_recv() {
|
||||
out.write_all(&chunk).map_err(Error::io(&file_path))?;
|
||||
hasher.update(&chunk);
|
||||
written += chunk.len() as u64;
|
||||
}
|
||||
out.flush().map_err(Error::io(&file_path))?;
|
||||
Ok((written, hasher))
|
||||
});
|
||||
let mut stream = response.bytes_stream();
|
||||
let (mut sent, mut overflow) = (0u64, false);
|
||||
let mut outcome: Result<()> = Ok(());
|
||||
loop {
|
||||
let next = tokio::select! {
|
||||
_ = cancel.cancelled() => { outcome = Err(Error::Cancelled); break; }
|
||||
next = stream.next() => next,
|
||||
};
|
||||
match next {
|
||||
Some(Ok(mut chunk)) => {
|
||||
let room = limit - sent;
|
||||
if chunk.len() as u64 > room {
|
||||
chunk.truncate(room as usize);
|
||||
overflow = true;
|
||||
}
|
||||
let len = chunk.len() as u64;
|
||||
if len > 0 {
|
||||
if tx.send(chunk).await.is_err() {
|
||||
break; // the writer failed; its error is reported below
|
||||
}
|
||||
sent += len;
|
||||
reporter.add_transferred(len);
|
||||
}
|
||||
if overflow {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Some(Err(e)) => {
|
||||
outcome = Err(Error::Http { url: url.clone(), status: None, message: format!("body cut short: {e}") });
|
||||
break;
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
drop(tx);
|
||||
let (written, hasher) = match writer.await {
|
||||
Ok(result) => result?,
|
||||
Err(e) if e.is_panic() => std::panic::resume_unwind(e.into_panic()),
|
||||
Err(_) => return Err(Error::Cancelled),
|
||||
};
|
||||
outcome?;
|
||||
Ok(Copied { written, hasher, overflow })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Blocking: copy a reader into a new file, hashing. Returns (bytes, sha256 hex).
|
||||
pub(crate) fn copy_hashing<R: Read>(
|
||||
reader: &mut R,
|
||||
target: &Path,
|
||||
buf: &mut [u8],
|
||||
reporter: &Reporter,
|
||||
cancel: &CancellationToken,
|
||||
read_error: &dyn Fn(std::io::Error) -> Error,
|
||||
) -> Result<(u64, String)> {
|
||||
let file = std::fs::File::create(target).map_err(Error::io(target))?;
|
||||
let mut out = BufWriter::with_capacity(256 * 1024, file);
|
||||
let mut hasher = Sha256::new();
|
||||
let mut written = 0u64;
|
||||
loop {
|
||||
if cancel.is_cancelled() {
|
||||
return Err(Error::Cancelled);
|
||||
}
|
||||
let n = reader.read(buf).map_err(read_error)?;
|
||||
if n == 0 {
|
||||
break;
|
||||
}
|
||||
out.write_all(&buf[..n]).map_err(Error::io(target))?;
|
||||
hasher.update(&buf[..n]);
|
||||
written += n as u64;
|
||||
reporter.add_transferred(n as u64);
|
||||
}
|
||||
out.flush().map_err(Error::io(target))?;
|
||||
Ok((written, hex(&hasher.finalize())))
|
||||
}
|
||||
|
||||
/// 32 hex characters, unique within this process and very likely across processes
|
||||
/// (the role `uuid.uuid4().hex` plays in fetch.py).
|
||||
pub(crate) fn unique_hex() -> String {
|
||||
static COUNTER: AtomicU64 = AtomicU64::new(0);
|
||||
let n = COUNTER.fetch_add(1, Ordering::Relaxed);
|
||||
let nanos = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_nanos()).unwrap_or(0);
|
||||
let mut hasher = Sha256::new();
|
||||
hasher.update(nanos.to_le_bytes());
|
||||
hasher.update(std::process::id().to_le_bytes());
|
||||
hasher.update(n.to_le_bytes());
|
||||
hex(&hasher.finalize()[..16])
|
||||
}
|
||||
|
||||
/// Write `data` to `path` through a temp file and a rename (fetch.py's `_atomic_write`).
|
||||
pub(crate) fn atomic_write(path: &Path, data: &[u8]) -> Result<()> {
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(Error::io(parent))?;
|
||||
}
|
||||
let mut temp = path.as_os_str().to_owned();
|
||||
temp.push(format!(".{}.tmp", unique_hex()));
|
||||
let temp = PathBuf::from(temp);
|
||||
let result = std::fs::write(&temp, data).map_err(Error::io(&temp)).and_then(|()| {
|
||||
std::fs::rename(&temp, path).map_err(Error::io(path))
|
||||
});
|
||||
if result.is_err() {
|
||||
let _ = std::fs::remove_file(&temp);
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
/// `path` + `suffix` (e.g. `Odian.exe` -> `Odian.exe.part`).
|
||||
pub(crate) fn with_suffix(path: &Path, suffix: &str) -> PathBuf {
|
||||
let mut s = path.as_os_str().to_owned();
|
||||
s.push(suffix);
|
||||
PathBuf::from(s)
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
//! One error type for the whole crate, with the variants the launcher UI branches on.
|
||||
|
||||
use std::fmt;
|
||||
use std::path::PathBuf;
|
||||
|
||||
pub type Result<T, E = Error> = std::result::Result<T, E>;
|
||||
|
||||
/// Which check failed when [`Error::Integrity`] is returned.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum IntegrityKind {
|
||||
/// The manifest is unsigned, altered after signing, or signed by another key.
|
||||
Signature,
|
||||
/// The site's signed manifest names another version than the one asked for.
|
||||
Version,
|
||||
/// A whole file (app or database dump) does not match its size and sha256.
|
||||
FileHash,
|
||||
/// A source-file chunk does not match its sha256.
|
||||
ChunkHash,
|
||||
/// A tile index does not match the manifest's `index_sha256`.
|
||||
IndexHash,
|
||||
/// A tile block's tiles do not add up to the signed block digest.
|
||||
BlockDigest,
|
||||
/// A tile block holds an entry that is not a tile of that block.
|
||||
TarEntry,
|
||||
}
|
||||
|
||||
impl fmt::Display for IntegrityKind {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.write_str(match self {
|
||||
IntegrityKind::Signature => "signature",
|
||||
IntegrityKind::Version => "version",
|
||||
IntegrityKind::FileHash => "file hash",
|
||||
IntegrityKind::ChunkHash => "chunk hash",
|
||||
IntegrityKind::IndexHash => "tile index hash",
|
||||
IntegrityKind::BlockDigest => "tile block digest",
|
||||
IntegrityKind::TarEntry => "tile block entry",
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum Error {
|
||||
/// The site refused the token (HTTP 401 or 403).
|
||||
#[error("the DHP refused the sign-in (HTTP {status}) for {url}")]
|
||||
Unauthorized { url: String, status: u16 },
|
||||
|
||||
/// A version, file or tile set does not exist on the site (404) or the drive.
|
||||
#[error("not found: {what}")]
|
||||
NotFound { what: String },
|
||||
|
||||
/// The site kept answering 503 (stream limit, or its Engine is briefly unreachable).
|
||||
#[error("the DHP is busy: {url} answered 503 {attempts} times")]
|
||||
SiteBusy { url: String, attempts: u32, retry_after_secs: Option<f64> },
|
||||
|
||||
/// Bytes do not match what the signed manifest (or the signed tile index) pins.
|
||||
#[error("integrity check failed ({kind}): {what}: {detail}")]
|
||||
Integrity { kind: IntegrityKind, what: String, detail: String },
|
||||
|
||||
/// Any other HTTP failure: connection error, unexpected status, body cut short.
|
||||
#[error("HTTP error for {url}: {message}")]
|
||||
Http { url: String, status: Option<u16>, message: String },
|
||||
|
||||
#[error("I/O error on {}: {source}", path.display())]
|
||||
Io { path: PathBuf, #[source] source: std::io::Error },
|
||||
|
||||
/// The manifest (or a document it pins) is malformed.
|
||||
#[error("manifest error: {0}")]
|
||||
Manifest(String),
|
||||
|
||||
#[error("cancelled")]
|
||||
Cancelled,
|
||||
}
|
||||
|
||||
impl Error {
|
||||
pub(crate) fn io(path: impl Into<PathBuf>) -> impl FnOnce(std::io::Error) -> Error {
|
||||
let path = path.into();
|
||||
move |source| Error::Io { path, source }
|
||||
}
|
||||
|
||||
pub(crate) fn manifest(message: impl Into<String>) -> Error {
|
||||
Error::Manifest(message.into())
|
||||
}
|
||||
|
||||
pub(crate) fn integrity(kind: IntegrityKind, what: impl Into<String>, detail: impl Into<String>) -> Error {
|
||||
Error::Integrity { kind, what: what.into(), detail: detail.into() }
|
||||
}
|
||||
|
||||
pub fn is_cancelled(&self) -> bool {
|
||||
matches!(self, Error::Cancelled)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
//! sha256 helpers (the Python module `odian_packs.hashing`).
|
||||
|
||||
use std::io::Read;
|
||||
use std::path::Path;
|
||||
|
||||
use sha2::{Digest, Sha256};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
|
||||
pub(crate) const BUF: usize = 1024 * 1024;
|
||||
|
||||
/// Lowercase hex of any byte slice.
|
||||
pub fn hex(bytes: &[u8]) -> String {
|
||||
const HEX: &[u8; 16] = b"0123456789abcdef";
|
||||
let mut s = String::with_capacity(bytes.len() * 2);
|
||||
for &b in bytes {
|
||||
s.push(HEX[(b >> 4) as usize] as char);
|
||||
s.push(HEX[(b & 0x0f) as usize] as char);
|
||||
}
|
||||
s
|
||||
}
|
||||
|
||||
/// sha256 hex of a byte slice.
|
||||
pub fn sha256_hex(data: &[u8]) -> String {
|
||||
hex(&Sha256::digest(data))
|
||||
}
|
||||
|
||||
/// Inclusive byte range of chunk `index` of a `size`-byte file: the form an HTTP Range takes.
|
||||
pub fn chunk_range(index: u64, size: u64, chunk_bytes: u64) -> Option<(u64, u64)> {
|
||||
let start = index.checked_mul(chunk_bytes)?;
|
||||
if start >= size {
|
||||
return None;
|
||||
}
|
||||
Some((start, size.min(start.saturating_add(chunk_bytes)) - 1))
|
||||
}
|
||||
|
||||
/// Feed the first `limit` bytes (or all, if `None`) of a file into `hasher`. Blocking.
|
||||
/// Returns how many bytes were read.
|
||||
pub(crate) fn hash_file_prefix(
|
||||
path: &Path,
|
||||
hasher: &mut Sha256,
|
||||
limit: Option<u64>,
|
||||
cancel: &CancellationToken,
|
||||
) -> Result<u64> {
|
||||
let file = std::fs::File::open(path).map_err(Error::io(path))?;
|
||||
let mut reader: Box<dyn Read> = match limit {
|
||||
Some(n) => Box::new(file.take(n)),
|
||||
None => Box::new(file),
|
||||
};
|
||||
let mut buf = vec![0u8; BUF];
|
||||
let mut total = 0u64;
|
||||
loop {
|
||||
if cancel.is_cancelled() {
|
||||
return Err(Error::Cancelled);
|
||||
}
|
||||
let n = reader.read(&mut buf).map_err(Error::io(path))?;
|
||||
if n == 0 {
|
||||
return Ok(total);
|
||||
}
|
||||
hasher.update(&buf[..n]);
|
||||
total += n as u64;
|
||||
}
|
||||
}
|
||||
|
||||
/// (size, sha256 hex) of a whole file. Blocking.
|
||||
pub(crate) fn sha256_file(path: &Path, cancel: &CancellationToken) -> Result<(u64, String)> {
|
||||
let mut hasher = Sha256::new();
|
||||
let size = hash_file_prefix(path, &mut hasher, None, cancel)?;
|
||||
Ok((size, hex(&hasher.finalize())))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn ranges_match_python() {
|
||||
assert_eq!(chunk_range(2, 25, 10), Some((20, 24)));
|
||||
assert_eq!(chunk_range(0, 25, 10), Some((0, 9)));
|
||||
assert_eq!(chunk_range(3, 25, 10), None);
|
||||
assert_eq!(chunk_range(0, 0, 10), None);
|
||||
assert_eq!(sha256_hex(b""), "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,620 @@
|
||||
//! The installer (fetch.py's `fetch`): verified manifest in, verified install out.
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
use std::future::Future;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
use futures_util::StreamExt;
|
||||
use serde::Serialize;
|
||||
use serde_json::Value;
|
||||
use sha2::{Digest, Sha256};
|
||||
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::progress::{Progress, Reporter, Stage};
|
||||
use crate::source::{blocking, Backend, Pack, Source};
|
||||
use crate::tileblocks::{copy_drive_block, fetch_site_block, DriveBlock};
|
||||
use crate::tiles::{verify_index, IndexBlock};
|
||||
|
||||
/// How to install.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct InstallOptions {
|
||||
/// Parallel downloads (db uses at most 4, source chunks at most 3, like fetch.py).
|
||||
pub workers: usize,
|
||||
/// Stages to run; `App`, `Db`, `Src` and `Tiles` are meaningful, others are ignored.
|
||||
pub only: Vec<Stage>,
|
||||
}
|
||||
|
||||
impl Default for InstallOptions {
|
||||
fn default() -> Self {
|
||||
InstallOptions { workers: 6, only: vec![Stage::App, Stage::Db, Stage::Src, Stage::Tiles] }
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct AppSummary {
|
||||
pub files: usize,
|
||||
pub downloaded: usize,
|
||||
pub skipped: usize,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct DbSummary {
|
||||
pub snapshot: String,
|
||||
pub files: usize,
|
||||
pub downloaded: usize,
|
||||
pub skipped: usize,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct SrcFileSummary {
|
||||
pub path: String,
|
||||
/// Chunks fetched by this run.
|
||||
pub chunks: usize,
|
||||
pub skipped: usize,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct TileSetSummary {
|
||||
pub set: String,
|
||||
pub blocks: usize,
|
||||
/// Blocks fetched by this run.
|
||||
pub downloaded: usize,
|
||||
pub files: u64,
|
||||
pub bytes: u64,
|
||||
}
|
||||
|
||||
/// What an install did (fetch.py's summary dict, plus the bytes that were transferred).
|
||||
#[derive(Debug, Clone, PartialEq, Serialize)]
|
||||
pub struct Summary {
|
||||
pub version: String,
|
||||
pub app: Option<AppSummary>,
|
||||
pub db: Option<DbSummary>,
|
||||
pub src: Option<Vec<SrcFileSummary>>,
|
||||
pub tiles: Option<Vec<TileSetSummary>>,
|
||||
/// Bytes received from the site (or copied from the drive) by this run.
|
||||
pub transferred_bytes: u64,
|
||||
pub seconds: f64,
|
||||
}
|
||||
|
||||
struct Ctx<'a> {
|
||||
backend: &'a Backend,
|
||||
reporter: Arc<Reporter>,
|
||||
cancel: CancellationToken,
|
||||
workers: usize,
|
||||
}
|
||||
|
||||
fn join_rel(root: &Path, rel: &str) -> Result<PathBuf> {
|
||||
let mut path = root.to_path_buf();
|
||||
for part in safe_relpath(rel)?.split('/') {
|
||||
path.push(part);
|
||||
}
|
||||
Ok(path)
|
||||
}
|
||||
|
||||
/// Install `manifest` from `source` into `dest`. Resumable: re-running skips everything
|
||||
/// already verified. `versions/<v>/manifest.json` is written last, only on success.
|
||||
pub async fn install<F>(
|
||||
source: &Source,
|
||||
dest: &Path,
|
||||
manifest: &VerifiedManifest,
|
||||
opts: InstallOptions,
|
||||
progress: F,
|
||||
cancel: CancellationToken,
|
||||
) -> Result<Summary>
|
||||
where
|
||||
F: Fn(Progress) + Send + Sync + 'static,
|
||||
{
|
||||
let started = Instant::now();
|
||||
let backend = Backend::new(source)?;
|
||||
let m = manifest.manifest();
|
||||
let version = m.version.clone();
|
||||
|
||||
let wants = |stage: Stage| opts.only.contains(&stage);
|
||||
let do_app = wants(Stage::App);
|
||||
let do_db = wants(Stage::Db) && m.db.is_some();
|
||||
let do_src = wants(Stage::Src) && m.src.as_ref().is_some_and(|s| s.enabled);
|
||||
let do_tiles = wants(Stage::Tiles) && m.tiles.is_some();
|
||||
let stage_count = 1 + [do_app, do_db, do_src, do_tiles].iter().filter(|b| **b).count() as u32;
|
||||
|
||||
let reporter = Arc::new(Reporter::new(Arc::new(progress), stage_count));
|
||||
reporter.begin_stage(Stage::Manifest, 1, 0, 1);
|
||||
reporter.begin_sub(format!("manifest {version}: signature OK"), "", 0);
|
||||
reporter.items_done(1);
|
||||
reporter.finish_stage();
|
||||
|
||||
let ctx = Ctx { backend: &backend, reporter: reporter.clone(), cancel: cancel.clone(), workers: opts.workers.max(1) };
|
||||
let version_dir = dest.join("versions").join(&version);
|
||||
let data_dir = dest.join("data");
|
||||
let mut summary = Summary {
|
||||
version: version.clone(),
|
||||
app: None,
|
||||
db: None,
|
||||
src: None,
|
||||
tiles: None,
|
||||
transferred_bytes: 0,
|
||||
seconds: 0.0,
|
||||
};
|
||||
let mut stage_index = 1;
|
||||
let check_cancel = || if cancel.is_cancelled() { Err(Error::Cancelled) } else { Ok(()) };
|
||||
|
||||
if do_app {
|
||||
check_cancel()?;
|
||||
stage_index += 1;
|
||||
summary.app = Some(install_app(&ctx, stage_index, &m.app.files, &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?;
|
||||
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?);
|
||||
}
|
||||
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?);
|
||||
}
|
||||
|
||||
check_cancel()?;
|
||||
// Written last: its presence means this version is fully installed.
|
||||
let manifest_path = version_dir.join("manifest.json");
|
||||
let raw = manifest.raw().to_vec();
|
||||
blocking(move || atomic_write(&manifest_path, &raw)).await?;
|
||||
|
||||
summary.transferred_bytes = reporter.transferred();
|
||||
summary.seconds = (started.elapsed().as_secs_f64() * 10.0).round() / 10.0;
|
||||
Ok(summary)
|
||||
}
|
||||
|
||||
/// Run `f` over `items` with at most `workers` in flight. On the first error nothing new
|
||||
/// starts, running items are told to stop (their verified work is kept) and drained, and
|
||||
/// that first error is returned (a real error wins over the `Cancelled` it caused).
|
||||
async fn run_pool<I, T, Fut, F, G>(items: Vec<I>, workers: usize, cancel: &CancellationToken, f: F, mut on_ok: G) -> Result<()>
|
||||
where
|
||||
F: Fn(I, CancellationToken) -> Fut,
|
||||
Fut: Future<Output = Result<T>>,
|
||||
G: FnMut(T) -> Result<()>,
|
||||
{
|
||||
let stop = cancel.child_token();
|
||||
let futures = items.into_iter().map(|item| {
|
||||
let token = stop.clone();
|
||||
let fut = f(item, token.clone());
|
||||
async move {
|
||||
if token.is_cancelled() {
|
||||
return Err(Error::Cancelled);
|
||||
}
|
||||
fut.await
|
||||
}
|
||||
});
|
||||
let mut stream = futures_util::stream::iter(futures).buffer_unordered(workers.max(1));
|
||||
let mut first: Option<Error> = None;
|
||||
while let Some(result) = stream.next().await {
|
||||
if let Err(e) = result.and_then(&mut on_ok) {
|
||||
stop.cancel();
|
||||
first = match first {
|
||||
None => Some(e),
|
||||
Some(Error::Cancelled) if !e.is_cancelled() => Some(e),
|
||||
keep => keep,
|
||||
};
|
||||
}
|
||||
}
|
||||
match first {
|
||||
Some(e) => Err(e),
|
||||
None if cancel.is_cancelled() => Err(Error::Cancelled),
|
||||
None => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
// ---- whole files (app, db) -------------------------------------------------------
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum Outcome {
|
||||
Downloaded,
|
||||
Skipped,
|
||||
}
|
||||
|
||||
/// One whole file, resuming `<dest>.part` with a Range request; verified before the rename.
|
||||
async fn download_file(ctx: &Ctx<'_>, pack: Pack<'_>, dest: &Path, size: u64, sha256: &str, cancel: &CancellationToken) -> Result<Outcome> {
|
||||
let existing = {
|
||||
let (dest, cancel) = (dest.to_path_buf(), cancel.clone());
|
||||
blocking(move || match std::fs::metadata(&dest) {
|
||||
Ok(md) if md.is_file() && md.len() == size => sha256_file(&dest, &cancel).map(|(_, sha)| Some(sha)),
|
||||
_ => Ok(None),
|
||||
})
|
||||
.await?
|
||||
};
|
||||
if existing.as_deref() == Some(sha256) {
|
||||
ctx.reporter.add_done_bytes(size);
|
||||
return Ok(Outcome::Skipped);
|
||||
}
|
||||
if let Some(parent) = dest.parent() {
|
||||
tokio::fs::create_dir_all(parent).await.map_err(Error::io(parent))?;
|
||||
}
|
||||
let part = with_suffix(dest, ".part");
|
||||
let mut got = (0u64, String::new());
|
||||
for _ in 0..2 {
|
||||
let mut counted = 0u64;
|
||||
let mut have = match tokio::fs::metadata(&part).await {
|
||||
Ok(md) if md.is_file() => md.len(),
|
||||
_ => 0,
|
||||
};
|
||||
if have > size {
|
||||
tokio::fs::remove_file(&part).await.map_err(Error::io(&part))?;
|
||||
have = 0;
|
||||
}
|
||||
let (got_size, got_sha) = if have < size {
|
||||
let mut hasher = Sha256::new();
|
||||
if have > 0 {
|
||||
let (path, cancel) = (part.clone(), cancel.clone());
|
||||
hasher = blocking(move || {
|
||||
let mut hasher = Sha256::new();
|
||||
hash_file_prefix(&path, &mut hasher, Some(have), &cancel)?;
|
||||
Ok(hasher)
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
let (body, starts_at_offset) = ctx.backend.open_pack(pack, have, cancel).await?;
|
||||
if have > 0 && !starts_at_offset {
|
||||
have = 0; // the server ignored the Range; start over
|
||||
hasher = Sha256::new();
|
||||
}
|
||||
ctx.reporter.add_done_bytes(have);
|
||||
counted += have;
|
||||
let file = {
|
||||
let path = part.clone();
|
||||
blocking(move || {
|
||||
let mut options = std::fs::OpenOptions::new();
|
||||
if have > 0 { options.append(true) } else { options.write(true).create(true).truncate(true) };
|
||||
options.open(&path).map_err(Error::io(&path))
|
||||
})
|
||||
.await?
|
||||
};
|
||||
let copied = copy_body(body, file, part.clone(), hasher, size - have, ctx.reporter.clone(), cancel.clone()).await?;
|
||||
counted += copied.written;
|
||||
(have + copied.written + u64::from(copied.overflow), hex(&copied.hasher.finalize()))
|
||||
} else {
|
||||
let (path, cancel) = (part.clone(), cancel.clone());
|
||||
let (s, h) = blocking(move || {
|
||||
if !path.is_file() {
|
||||
std::fs::File::create(&path).map_err(Error::io(&path))?; // an empty file: nothing to fetch
|
||||
}
|
||||
sha256_file(&path, &cancel)
|
||||
})
|
||||
.await?;
|
||||
ctx.reporter.add_done_bytes(s);
|
||||
counted += s;
|
||||
(s, h)
|
||||
};
|
||||
if got_size == size && got_sha == sha256 {
|
||||
let (from, to) = (part.clone(), dest.to_path_buf());
|
||||
blocking(move || std::fs::rename(&from, &to).map_err(Error::io(&to))).await?;
|
||||
return Ok(Outcome::Downloaded);
|
||||
}
|
||||
// Corrupt or stale partial: one clean retry, then give up.
|
||||
ctx.reporter.retract_bytes(counted);
|
||||
let _ = tokio::fs::remove_file(&part).await;
|
||||
got = (got_size, got_sha);
|
||||
}
|
||||
Err(Error::integrity(
|
||||
IntegrityKind::FileHash,
|
||||
ctx.backend.pack_label(pack),
|
||||
format!("expected {size} bytes sha256 {sha256}, got {} bytes sha256 {}", got.0, got.1),
|
||||
))
|
||||
}
|
||||
|
||||
async fn install_app(ctx: &Ctx<'_>, stage_index: u32, files: &[AppFile], 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);
|
||||
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
|
||||
},
|
||||
|outcome| {
|
||||
if outcome == Outcome::Downloaded { downloaded += 1 } else { skipped += 1 }
|
||||
ctx.reporter.items_done(1);
|
||||
Ok(())
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
ctx.reporter.finish_stage();
|
||||
Ok(AppSummary { files: files.len(), downloaded, skipped })
|
||||
}
|
||||
|
||||
async fn install_db(ctx: &Ctx<'_>, stage_index: u32, files: &[DbFile], data_dir: &Path) -> Result<(usize, usize, usize)> {
|
||||
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);
|
||||
let (mut downloaded, mut skipped) = (0, 0);
|
||||
run_pool(
|
||||
files.iter().collect(),
|
||||
ctx.workers.min(4),
|
||||
&ctx.cancel,
|
||||
|entry: &DbFile, stop| async move {
|
||||
let target = join_rel(data_dir, &entry.path)?;
|
||||
download_file(ctx, Pack::Db(&entry.path), &target, entry.size, &entry.sha256, &stop).await
|
||||
},
|
||||
|outcome| {
|
||||
if outcome == Outcome::Downloaded { downloaded += 1 } else { skipped += 1 }
|
||||
ctx.reporter.items_done(1);
|
||||
Ok(())
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
ctx.reporter.finish_stage();
|
||||
Ok((files.len(), downloaded, skipped))
|
||||
}
|
||||
|
||||
// ---- source files -----------------------------------------------------------------
|
||||
|
||||
async fn install_src(ctx: &Ctx<'_>, stage_index: u32, src: &SrcSection, data_dir: &Path) -> Result<Vec<SrcFileSummary>> {
|
||||
let total: u64 = src.files.iter().map(|f| f.size).sum();
|
||||
let chunks: u64 = src.files.iter().map(|f| f.chunks.len() as u64).sum();
|
||||
ctx.reporter.begin_stage(Stage::Src, stage_index, total, chunks);
|
||||
let mut out = Vec::with_capacity(src.files.len());
|
||||
for entry in &src.files {
|
||||
if ctx.cancel.is_cancelled() {
|
||||
return Err(Error::Cancelled);
|
||||
}
|
||||
let target = join_rel(&data_dir.join("src"), &entry.path)?;
|
||||
ctx.reporter.begin_sub(format!("src {}", entry.path), "chunks", entry.chunks.len() as u64);
|
||||
out.push(fetch_source_file(ctx, entry, src.chunk_bytes, &target).await?);
|
||||
}
|
||||
ctx.reporter.finish_stage();
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// `{"chunks_sha256": "...", "verified": [0, 1]}`, exactly as Python's `json.dumps` writes it.
|
||||
fn chunks_progress_json(chunks_sha256: &str, verified: &BTreeSet<u64>) -> String {
|
||||
let list: Vec<String> = verified.iter().map(u64::to_string).collect();
|
||||
format!("{{\"chunks_sha256\": \"{chunks_sha256}\", \"verified\": [{}]}}", list.join(", "))
|
||||
}
|
||||
|
||||
/// One source file, chunk by chunk from byte ranges, written in place.
|
||||
async fn fetch_source_file(ctx: &Ctx<'_>, entry: &SrcFile, chunk_bytes: u64, dest: &Path) -> Result<SrcFileSummary> {
|
||||
let n = entry.chunks.len() as u64;
|
||||
let marker = with_suffix(dest, ".ok");
|
||||
let progress_path = with_suffix(dest, ".chunks.json");
|
||||
let (done, verified) = {
|
||||
let (dest, marker, progress_path) = (dest.to_path_buf(), marker.clone(), progress_path.clone());
|
||||
let (size, digest) = (entry.size, entry.chunks_sha256.clone());
|
||||
blocking(move || -> Result<(bool, BTreeSet<u64>)> {
|
||||
let dest_ok = std::fs::metadata(&dest).is_ok_and(|md| md.is_file() && md.len() == size);
|
||||
if dest_ok && marker.is_file() {
|
||||
if let Ok(text) = std::fs::read_to_string(&marker) {
|
||||
if text.trim() == digest {
|
||||
return Ok((true, BTreeSet::new()));
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Some(parent) = dest.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(Error::io(parent))?;
|
||||
}
|
||||
let mut verified = BTreeSet::new();
|
||||
if dest_ok && progress_path.is_file() {
|
||||
let saved = std::fs::read(&progress_path).ok().and_then(|raw| serde_json::from_slice::<Value>(&raw).ok());
|
||||
if let Some(saved) = saved.filter(|s| s.get("chunks_sha256").and_then(Value::as_str) == Some(&digest)) {
|
||||
let list = saved.get("verified").and_then(Value::as_array).cloned().unwrap_or_default();
|
||||
verified.extend(list.iter().filter_map(Value::as_u64).filter(|i| *i < n));
|
||||
}
|
||||
}
|
||||
if !dest_ok {
|
||||
let file = std::fs::File::create(&dest).map_err(Error::io(&dest))?;
|
||||
file.set_len(size).map_err(Error::io(&dest))?;
|
||||
verified.clear();
|
||||
}
|
||||
Ok((false, verified))
|
||||
})
|
||||
.await?
|
||||
};
|
||||
if done {
|
||||
ctx.reporter.add_done_bytes(entry.size);
|
||||
ctx.reporter.items_done(n);
|
||||
return Ok(SrcFileSummary { path: entry.path.clone(), chunks: 0, skipped: n as usize });
|
||||
}
|
||||
for index in &verified {
|
||||
let (start, end) = chunk_range(*index, entry.size, chunk_bytes).expect("validated chunk index");
|
||||
ctx.reporter.add_done_bytes(end - start + 1);
|
||||
}
|
||||
ctx.reporter.items_done(verified.len() as u64);
|
||||
|
||||
let pending: Vec<u64> = (0..n).filter(|i| !verified.contains(i)).collect();
|
||||
let fetched = pending.len();
|
||||
let mut verified = verified;
|
||||
run_pool(
|
||||
pending,
|
||||
ctx.workers.min(3),
|
||||
&ctx.cancel,
|
||||
|index, stop| fetch_chunk(ctx, entry, chunk_bytes, dest, index, stop),
|
||||
|index| {
|
||||
verified.insert(index);
|
||||
atomic_write(&progress_path, chunks_progress_json(&entry.chunks_sha256, &verified).as_bytes())?;
|
||||
ctx.reporter.items_done(1);
|
||||
Ok(())
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
let (marker_path, progress_file, digest) = (marker, progress_path, entry.chunks_sha256.clone());
|
||||
blocking(move || {
|
||||
atomic_write(&marker_path, digest.as_bytes())?;
|
||||
match std::fs::remove_file(&progress_file) {
|
||||
Err(e) if e.kind() != std::io::ErrorKind::NotFound => Err(Error::Io { path: progress_file, source: e }),
|
||||
_ => Ok(()),
|
||||
}
|
||||
})
|
||||
.await?;
|
||||
Ok(SrcFileSummary { path: entry.path.clone(), chunks: fetched, skipped: n as usize - fetched })
|
||||
}
|
||||
|
||||
async fn fetch_chunk(ctx: &Ctx<'_>, entry: &SrcFile, chunk_bytes: u64, dest: &Path, index: u64, cancel: CancellationToken) -> Result<u64> {
|
||||
let (start, end) = chunk_range(index, entry.size, chunk_bytes).expect("validated chunk index");
|
||||
let len = end - start + 1;
|
||||
let expected = &entry.chunks[index as usize];
|
||||
let mut got = String::new();
|
||||
for _ in 0..2 {
|
||||
let body = ctx.backend.open_src_range(&entry.path, start, end, &cancel).await?;
|
||||
let file = {
|
||||
let path = dest.to_path_buf();
|
||||
blocking(move || {
|
||||
use std::io::{Seek, SeekFrom};
|
||||
let mut file = std::fs::OpenOptions::new().write(true).open(&path).map_err(Error::io(&path))?;
|
||||
file.seek(SeekFrom::Start(start)).map_err(Error::io(&path))?;
|
||||
Ok(file)
|
||||
})
|
||||
.await?
|
||||
};
|
||||
let copied = copy_body(body, file, dest.to_path_buf(), Sha256::new(), len, ctx.reporter.clone(), cancel.clone()).await?;
|
||||
got = hex(&copied.hasher.finalize());
|
||||
if copied.written == len && !copied.overflow && got == *expected {
|
||||
return Ok(index);
|
||||
}
|
||||
ctx.reporter.retract_bytes(copied.written);
|
||||
}
|
||||
Err(Error::integrity(
|
||||
IntegrityKind::ChunkHash,
|
||||
format!("/src/{} chunk {index}", entry.path),
|
||||
format!("sha256 mismatch after retry: expected {expected}, got {got}"),
|
||||
))
|
||||
}
|
||||
|
||||
// ---- tiles --------------------------------------------------------------------------
|
||||
|
||||
/// `{"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(", "))
|
||||
}
|
||||
|
||||
async fn install_tiles(ctx: &Ctx<'_>, stage_index: u32, version: &str, tiles: &TilesSection, data_dir: &Path) -> Result<Vec<TileSetSummary>> {
|
||||
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);
|
||||
let journal = match ctx.backend.drive_root() {
|
||||
Some(root) if !tiles.sets.is_empty() => {
|
||||
let path = root.join("tiles").join("tiles.db");
|
||||
Some(Arc::new(blocking(move || crate::source::Journal::open(&path)).await?))
|
||||
}
|
||||
_ => None,
|
||||
};
|
||||
let mut out = Vec::with_capacity(tiles.sets.len());
|
||||
for set in &tiles.sets {
|
||||
if ctx.cancel.is_cancelled() {
|
||||
return Err(Error::Cancelled);
|
||||
}
|
||||
let (style, dataset) = (set.style.as_str(), set.dataset.as_str());
|
||||
let what = format!("tile index {style}/{dataset}");
|
||||
let raw = ctx.backend.tile_index(version, style, dataset, &ctx.cancel).await?;
|
||||
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 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()) {
|
||||
let list = saved.get("blocks").and_then(Value::as_array).cloned().unwrap_or_default();
|
||||
done.extend(list.iter().filter_map(Value::as_str).map(str::to_string));
|
||||
}
|
||||
}
|
||||
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();
|
||||
ctx.reporter.add_done_bytes(skipped_bytes);
|
||||
ctx.reporter.items_done((index.blocks.len() - pending.len()) as u64);
|
||||
tokio::fs::create_dir_all(&set_dir).await.map_err(Error::io(&set_dir))?;
|
||||
|
||||
let fetched = pending.len();
|
||||
let mut count = 0usize;
|
||||
let set_root = ctx.backend.drive_root().map(|root| root.join("tiles").join(style).join(dataset));
|
||||
let result = run_pool(
|
||||
pending,
|
||||
ctx.workers,
|
||||
&ctx.cancel,
|
||||
|block: IndexBlock, stop| {
|
||||
let (set_dir, reporter, journal, set_root) = (set_dir.clone(), ctx.reporter.clone(), journal.clone(), set_root.clone());
|
||||
let (style, dataset, journal_seq) = (style.to_string(), dataset.to_string(), index.journal_seq);
|
||||
async move {
|
||||
let key = block.done_key();
|
||||
let url = format!("/v1/versions/{version}/tiles/{style}/{dataset}/block/{}/{}/{}.tar", block.z, block.bx, block.by);
|
||||
match (ctx.backend.site(), journal, set_root) {
|
||||
(Some(site), _, _) => fetch_site_block(site, url, block, set_dir, reporter, stop).await?,
|
||||
(None, Some(journal), Some(set_root)) => {
|
||||
blocking(move || {
|
||||
let drive = DriveBlock { journal: &journal, set_root: &set_root, style: &style, dataset: &dataset, journal_seq };
|
||||
let what = format!("tiles/{style}/{dataset} block {}/{}/{}", block.z, block.bx, block.by);
|
||||
copy_drive_block(&drive, &what, &block, &set_dir, &reporter, &stop)
|
||||
})
|
||||
.await?
|
||||
}
|
||||
_ => unreachable!("a drive install always opens the journal"),
|
||||
}
|
||||
Ok(key)
|
||||
}
|
||||
},
|
||||
|key| {
|
||||
done.insert(key);
|
||||
count += 1;
|
||||
ctx.reporter.items_done(1);
|
||||
if count % 50 == 0 || count == fetched {
|
||||
atomic_write(&done_path, blocks_done_json(&set.index_sha256, &done).as_bytes())?;
|
||||
}
|
||||
Ok(())
|
||||
},
|
||||
)
|
||||
.await;
|
||||
if result.is_err() && count % 50 != 0 {
|
||||
// Keep the blocks that did verify, so the next run does not fetch them again.
|
||||
let _ = atomic_write(&done_path, blocks_done_json(&set.index_sha256, &done).as_bytes());
|
||||
}
|
||||
result?;
|
||||
out.push(TileSetSummary {
|
||||
set: format!("{style}/{dataset}"),
|
||||
blocks: index.blocks.len(),
|
||||
downloaded: fetched,
|
||||
files: index.files,
|
||||
bytes: index.bytes,
|
||||
});
|
||||
}
|
||||
ctx.reporter.finish_stage();
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn progress_files_match_python_json_dumps() {
|
||||
let verified: BTreeSet<u64> = [2, 0, 1].into_iter().collect();
|
||||
assert_eq!(chunks_progress_json("ab", &verified), r#"{"chunks_sha256": "ab", "verified": [0, 1, 2]}"#);
|
||||
let done: BTreeSet<String> = ["7/1/0:bb".to_string(), "0/0/0:aa".to_string()].into_iter().collect();
|
||||
assert_eq!(blocks_done_json("cd", &done), r#"{"index_sha256": "cd", "blocks": ["0/0/0:aa", "7/1/0:bb"]}"#);
|
||||
assert_eq!(blocks_done_json("cd", &BTreeSet::new()), r#"{"index_sha256": "cd", "blocks": []}"#);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn install_future_is_send() {
|
||||
fn assert_send<T: Send>(_: &T) {}
|
||||
let source = Source::Drive { root: PathBuf::from(".") };
|
||||
let raw = b"{}".to_vec();
|
||||
if let Ok(manifest) = VerifiedManifest::from_bytes(raw, crate::bundled_public_key()) {
|
||||
let fut = install(&source, Path::new("."), &manifest, InstallOptions::default(), |_| {}, CancellationToken::new());
|
||||
assert_send(&fut);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
//! ODIAN packs, format v1: verify a signed version manifest and install that version from a
|
||||
//! site's update service or from an ODIAN drive, checking every byte.
|
||||
//!
|
||||
//! A byte-for-byte compatible port of the Python installer (`odian_packs.fetch` in
|
||||
//! ODIAN_MCP_Server; the format is its `src/odian_packs/README.md`). An install made by
|
||||
//! either one is resumed and re-verified by the other: same layout, same progress files.
|
||||
//!
|
||||
//! ```text
|
||||
//! dest/versions/<v>/manifest.json written LAST: its presence means "installed"
|
||||
//! dest/versions/<v>/<app paths>
|
||||
//! dest/data/db/<snapshot>/...
|
||||
//! dest/data/src/<paths> + <path>.chunks.json while partial, <path>.ok when done
|
||||
//! dest/data/tiles/<style>/<dataset>/<z>/<x>/<y><ext> + .blocks-done.json
|
||||
//! ```
|
||||
|
||||
pub mod canonical;
|
||||
pub mod error;
|
||||
pub mod hashing;
|
||||
pub mod manifest;
|
||||
pub mod signing;
|
||||
pub mod tiles;
|
||||
|
||||
mod copy;
|
||||
mod install;
|
||||
mod progress;
|
||||
mod source;
|
||||
mod tileblocks;
|
||||
|
||||
pub use canonical::canonical_bytes;
|
||||
pub use error::{Error, IntegrityKind, Result};
|
||||
pub use install::{
|
||||
install, AppSummary, DbSummary, InstallOptions, SrcFileSummary, Summary, TileSetSummary,
|
||||
};
|
||||
pub use manifest::{safe_relpath, validate, Manifest, VerifiedManifest};
|
||||
pub use progress::{Progress, Stage};
|
||||
pub use source::Source;
|
||||
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
/// The ODIAN release public key that ships with the launcher (`ed25519:<base64>`).
|
||||
pub fn bundled_public_key() -> &'static str {
|
||||
include_str!("../keys/odian-release.pub").trim()
|
||||
}
|
||||
|
||||
/// The source's CURRENT version manifest: fetched, validated and signature-checked.
|
||||
/// The raw bytes are kept ([`VerifiedManifest::raw`]) because the install writes them verbatim.
|
||||
pub async fn current_manifest(source: &Source, public_key: &str) -> Result<VerifiedManifest> {
|
||||
let backend = source::Backend::new(source)?;
|
||||
let raw = backend.current_manifest(&CancellationToken::new()).await?;
|
||||
VerifiedManifest::from_bytes(raw, public_key)
|
||||
}
|
||||
|
||||
/// A specific version's manifest, verified, and checked to really be that version.
|
||||
pub async fn version_manifest(source: &Source, version: &str, public_key: &str) -> Result<VerifiedManifest> {
|
||||
let backend = source::Backend::new(source)?;
|
||||
let raw = backend.version_manifest(version, &CancellationToken::new()).await?;
|
||||
let verified = VerifiedManifest::from_bytes(raw, public_key)?;
|
||||
if verified.version() != version {
|
||||
return Err(Error::integrity(
|
||||
IntegrityKind::Version,
|
||||
format!("manifest for {version}"),
|
||||
format!("asked for {version}, the signed manifest says {}", verified.version()),
|
||||
));
|
||||
}
|
||||
Ok(verified)
|
||||
}
|
||||
|
||||
/// A version's patch notes (Markdown). **Untrusted text**: not covered by the signature.
|
||||
pub async fn fetch_notes(source: &Source, version: &str) -> Result<String> {
|
||||
let backend = source::Backend::new(source)?;
|
||||
let raw = backend.notes(version, &CancellationToken::new()).await?;
|
||||
Ok(String::from_utf8_lossy(&raw).into_owned())
|
||||
}
|
||||
@@ -0,0 +1,454 @@
|
||||
//! Version manifest: schema v1, validation, and the verified form the installer uses
|
||||
//! (the Python module `odian_packs.manifest`).
|
||||
|
||||
use std::collections::HashSet;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{Map, Value};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use crate::canonical::canonical_bytes;
|
||||
use crate::error::{Error, Result};
|
||||
use crate::hashing::hex;
|
||||
|
||||
pub const SCHEMA: u64 = 1;
|
||||
pub const CHUNK_BYTES: u64 = 256 * 1024 * 1024;
|
||||
pub const BLOCK_SHIFT: u32 = 6;
|
||||
pub const DB_SECTIONS: [&str; 4] = ["pre-data", "data", "sequences", "post-data"];
|
||||
|
||||
const WINDOWS_FORBIDDEN: &[char] = &['<', '>', ':', '"', '|', '?', '*'];
|
||||
|
||||
/// `MAJOR.MINOR.PATCH` with no leading zeros (ASCII digits only).
|
||||
pub fn parse_version(text: &str) -> Result<(u64, u64, u64)> {
|
||||
let bad = || Error::manifest(format!("version must be MAJOR.MINOR.PATCH, got {text:?}"));
|
||||
let parts: Vec<&str> = text.split('.').collect();
|
||||
if parts.len() != 3 {
|
||||
return Err(bad());
|
||||
}
|
||||
let mut out = [0u64; 3];
|
||||
for (slot, part) in out.iter_mut().zip(&parts) {
|
||||
let ok = !part.is_empty()
|
||||
&& part.bytes().all(|b| b.is_ascii_digit())
|
||||
&& (*part == "0" || !part.starts_with('0'));
|
||||
if !ok {
|
||||
return Err(bad());
|
||||
}
|
||||
*slot = part.parse().map_err(|_| bad())?;
|
||||
}
|
||||
Ok((out[0], out[1], out[2]))
|
||||
}
|
||||
|
||||
/// A path that stays inside the directory it is joined to, and that Windows can create as
|
||||
/// a plain file. Same rules and order of checks as Python's `safe_relpath`.
|
||||
pub fn safe_relpath(path: &str) -> Result<&str> {
|
||||
if path.is_empty() {
|
||||
return Err(Error::manifest("empty path"));
|
||||
}
|
||||
if path.contains('\\') || path.starts_with('/') {
|
||||
return Err(Error::manifest(format!("path must be relative with '/' separators: {path:?}")));
|
||||
}
|
||||
let parts: Vec<&str> = path.split('/').collect();
|
||||
if parts.iter().any(|p| matches!(*p, "" | "." | "..")) {
|
||||
return Err(Error::manifest(format!("path must not contain empty, '.' or '..' parts: {path:?}")));
|
||||
}
|
||||
for part in parts {
|
||||
if part.chars().any(|c| (c as u32) < 32 || WINDOWS_FORBIDDEN.contains(&c)) {
|
||||
return Err(Error::manifest(format!(
|
||||
"path part {part:?} holds a control character or one of <>:\"|?*: {path:?}"
|
||||
)));
|
||||
}
|
||||
if part.ends_with('.') || part.ends_with(' ') {
|
||||
return Err(Error::manifest(format!("path part {part:?} ends with a dot or space: {path:?}")));
|
||||
}
|
||||
if is_windows_reserved(part) {
|
||||
return Err(Error::manifest(format!("path part {part:?} is a reserved Windows device name: {path:?}")));
|
||||
}
|
||||
}
|
||||
Ok(path)
|
||||
}
|
||||
|
||||
/// `^(con|prn|aux|nul|com[0-9¹²³]|lpt[0-9¹²³])(\..*)?$`, case-insensitive.
|
||||
fn is_windows_reserved(part: &str) -> bool {
|
||||
let base = match part.find('.') {
|
||||
Some(i) => &part[..i],
|
||||
None => part,
|
||||
};
|
||||
let lower = base.to_ascii_lowercase();
|
||||
if matches!(lower.as_str(), "con" | "prn" | "aux" | "nul") {
|
||||
return true;
|
||||
}
|
||||
for prefix in ["com", "lpt"] {
|
||||
if let Some(rest) = lower.strip_prefix(prefix) {
|
||||
let mut chars = rest.chars();
|
||||
if let (Some(c), None) = (chars.next(), chars.next()) {
|
||||
if c.is_ascii_digit() || matches!(c, '¹' | '²' | '³') {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
pub fn chunk_count(size: u64, chunk_bytes: u64) -> u64 {
|
||||
size.div_ceil(chunk_bytes)
|
||||
}
|
||||
|
||||
/// sha256 over the concatenated chunk hashes (hex text): a source file's identity.
|
||||
pub fn chunks_digest<S: AsRef<str>>(chunks: &[S]) -> String {
|
||||
let mut digest = Sha256::new();
|
||||
for chunk in chunks {
|
||||
digest.update(chunk.as_ref().as_bytes());
|
||||
}
|
||||
hex(&digest.finalize())
|
||||
}
|
||||
|
||||
/// Where a version keeps a tile set's frozen index, relative to `versions/<v>/`.
|
||||
pub fn tile_index_name(style: &str, dataset: &str) -> String {
|
||||
format!("tiles/{style}.{dataset}.index.json")
|
||||
}
|
||||
|
||||
/// What the Ed25519 signature covers: canonical JSON of the manifest minus `signature`.
|
||||
pub fn signing_payload(manifest: &Value) -> Result<Vec<u8>> {
|
||||
let object = manifest.as_object().ok_or_else(|| Error::manifest("manifest must be a JSON object"))?;
|
||||
let mut copy = Map::new();
|
||||
for (k, v) in object {
|
||||
if k != "signature" {
|
||||
copy.insert(k.clone(), v.clone());
|
||||
}
|
||||
}
|
||||
canonical_bytes(&Value::Object(copy))
|
||||
}
|
||||
|
||||
// ---- validation -------------------------------------------------------------
|
||||
|
||||
fn require(cond: bool, message: impl FnOnce() -> String) -> Result<()> {
|
||||
if cond { Ok(()) } else { Err(Error::Manifest(message())) }
|
||||
}
|
||||
|
||||
fn is_count(value: Option<&Value>) -> bool {
|
||||
value.and_then(Value::as_u64).is_some()
|
||||
}
|
||||
|
||||
fn is_sha256(value: Option<&Value>) -> bool {
|
||||
matches!(value.and_then(Value::as_str), Some(s) if s.len() == 64 && s.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')))
|
||||
}
|
||||
|
||||
fn check_sha(value: Option<&Value>, place: &str) -> Result<()> {
|
||||
require(is_sha256(value), || format!("{place}: sha256 must be 64 lowercase hex chars"))
|
||||
}
|
||||
|
||||
fn is_name(s: &str) -> bool {
|
||||
(1..=128).contains(&s.len()) && s.bytes().all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'.' | b'-'))
|
||||
}
|
||||
|
||||
fn is_snapshot(s: &str) -> bool {
|
||||
let b = s.as_bytes();
|
||||
b.len() == 16
|
||||
&& b[..8].iter().all(u8::is_ascii_digit)
|
||||
&& b[8] == b'T'
|
||||
&& b[9..15].iter().all(u8::is_ascii_digit)
|
||||
&& b[15] == b'Z'
|
||||
}
|
||||
|
||||
fn relpath_value(value: Option<&Value>) -> Result<&str> {
|
||||
match value.and_then(Value::as_str) {
|
||||
Some(s) => safe_relpath(s),
|
||||
None => Err(Error::manifest("empty path")),
|
||||
}
|
||||
}
|
||||
|
||||
fn version_value(value: Option<&Value>) -> Result<()> {
|
||||
match value {
|
||||
None => parse_version("").map(|_| ()),
|
||||
Some(Value::String(s)) => parse_version(s).map(|_| ()),
|
||||
Some(other) => Err(Error::manifest(format!("version must be MAJOR.MINOR.PATCH, got {other}"))),
|
||||
}
|
||||
}
|
||||
|
||||
fn entries<'a>(value: Option<&'a Value>, place: &str) -> Result<Vec<&'a Map<String, Value>>> {
|
||||
let list = value.and_then(Value::as_array).ok_or_else(|| Error::manifest(format!("{place} must be a list")))?;
|
||||
list.iter()
|
||||
.enumerate()
|
||||
.map(|(i, entry)| entry.as_object().ok_or_else(|| Error::manifest(format!("{place}[{i}] must be an object"))))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn object<'a>(value: &'a Value, message: &str) -> Result<&'a Map<String, Value>> {
|
||||
value.as_object().ok_or_else(|| Error::manifest(message))
|
||||
}
|
||||
|
||||
/// Error unless `manifest` is a well-formed schema-1 manifest. Structure only; the
|
||||
/// signature is checked by [`crate::signing::verify`].
|
||||
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}"))?;
|
||||
require(m.get("product").and_then(Value::as_str) == Some("odian"), || "product must be 'odian'".into())?;
|
||||
version_value(m.get("version"))?;
|
||||
require(m.get("released_at").is_some_and(Value::is_string), || "released_at must be an ISO-8601 string".into())?;
|
||||
if let Some(v) = m.get("min_launcher") {
|
||||
version_value(Some(v))?;
|
||||
}
|
||||
if let Some(v) = m.get("notes") {
|
||||
relpath_value(Some(v))?;
|
||||
}
|
||||
|
||||
let app_files = match m.get("app") {
|
||||
None => Vec::new(),
|
||||
Some(app) => entries(object(app, "app must be an object")?.get("files"), "app.files")?,
|
||||
};
|
||||
let mut seen: HashSet<String> = HashSet::new();
|
||||
for (i, entry) in app_files.iter().enumerate() {
|
||||
let place = format!("app.files[{i}]");
|
||||
let path = relpath_value(entry.get("path"))?;
|
||||
require(seen.insert(path.to_lowercase()), || format!("{place}: duplicate path {path:?}"))?;
|
||||
require(is_count(entry.get("size")), || format!("{place}: size must be a non-negative integer"))?;
|
||||
check_sha(entry.get("sha256"), &place)?;
|
||||
}
|
||||
|
||||
if let Some(db) = m.get("db") {
|
||||
let db = object(db, "db must be an object")?;
|
||||
require(is_count(db.get("schema_version")), || "db.schema_version must be a non-negative integer".into())?;
|
||||
let snapshot = db
|
||||
.get("snapshot")
|
||||
.and_then(Value::as_str)
|
||||
.filter(|s| is_snapshot(s))
|
||||
.ok_or_else(|| Error::manifest("db.snapshot must look like 20260915T101500Z"))?;
|
||||
let files = entries(db.get("files"), "db.files")?;
|
||||
require(!files.is_empty(), || "db.files must not be empty".into())?;
|
||||
let prefix = format!("db/{snapshot}/");
|
||||
let mut seen: HashSet<String> = HashSet::new();
|
||||
for (i, entry) in files.iter().enumerate() {
|
||||
let place = format!("db.files[{i}]");
|
||||
let section = entry.get("section").and_then(Value::as_str).filter(|s| DB_SECTIONS.contains(s));
|
||||
let section = section.ok_or_else(|| Error::manifest(format!("{place}: section must be one of {DB_SECTIONS:?}")))?;
|
||||
if section == "data" {
|
||||
require(entry.get("table").and_then(Value::as_str).is_some_and(|t| !t.is_empty()), || {
|
||||
format!("{place}: data dumps name their table")
|
||||
})?;
|
||||
}
|
||||
let path = relpath_value(entry.get("path"))?;
|
||||
require(path.starts_with(&prefix), || format!("{place}: path must live under {prefix}"))?;
|
||||
require(seen.insert(path.to_lowercase()), || format!("{place}: duplicate path {path:?}"))?;
|
||||
require(is_count(entry.get("size")), || format!("{place}: size must be a non-negative integer"))?;
|
||||
check_sha(entry.get("sha256"), &place)?;
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(tiles) = m.get("tiles") {
|
||||
let tiles = object(tiles, "tiles must be an object")?;
|
||||
require(tiles.get("block_shift").and_then(Value::as_u64) == Some(BLOCK_SHIFT as u64), || {
|
||||
format!("tiles.block_shift must be {BLOCK_SHIFT}")
|
||||
})?;
|
||||
let mut sets_seen: HashSet<(String, String)> = HashSet::new();
|
||||
for (i, entry) in entries(tiles.get("sets"), "tiles.sets")?.iter().enumerate() {
|
||||
let place = format!("tiles.sets[{i}]");
|
||||
for key in ["style", "dataset"] {
|
||||
require(entry.get(key).and_then(Value::as_str).is_some_and(is_name), || format!("{place}: bad {key}"))?;
|
||||
}
|
||||
let style = entry["style"].as_str().unwrap_or_default().to_string();
|
||||
let dataset = entry["dataset"].as_str().unwrap_or_default().to_string();
|
||||
require(!sets_seen.contains(&(style.clone(), dataset.clone())), || {
|
||||
format!("{place}: duplicate tile set {style}/{dataset}")
|
||||
})?;
|
||||
for count in ["journal_seq", "blocks", "files", "bytes"] {
|
||||
require(is_count(entry.get(count)), || format!("{place}: {count} must be a non-negative integer"))?;
|
||||
}
|
||||
check_sha(entry.get("index_sha256"), &place)?;
|
||||
let expected = tile_index_name(&style, &dataset);
|
||||
require(entry.get("index").and_then(Value::as_str) == Some(expected.as_str()), || {
|
||||
format!("{place}: index must be {expected:?}")
|
||||
})?;
|
||||
sets_seen.insert((style, dataset));
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(src) = m.get("src") {
|
||||
let src = object(src, "src must be an object")?;
|
||||
require(src.get("enabled").is_some_and(Value::is_boolean), || "src.enabled must be true or false".into())?;
|
||||
let chunk_bytes = src.get("chunk_bytes").and_then(Value::as_u64).filter(|n| *n > 0);
|
||||
let chunk_bytes = chunk_bytes.ok_or_else(|| Error::manifest("src.chunk_bytes must be a positive integer"))?;
|
||||
let mut seen: HashSet<String> = HashSet::new();
|
||||
for (i, entry) in entries(src.get("files"), "src.files")?.iter().enumerate() {
|
||||
let place = format!("src.files[{i}]");
|
||||
let path = relpath_value(entry.get("path"))?;
|
||||
require(seen.insert(path.to_lowercase()), || format!("{place}: duplicate path {path:?}"))?;
|
||||
let size = entry.get("size").and_then(Value::as_u64);
|
||||
let size = size.ok_or_else(|| Error::manifest(format!("{place}: size must be a non-negative integer")))?;
|
||||
let chunks = entry.get("chunks").and_then(Value::as_array);
|
||||
let chunks = chunks.ok_or_else(|| Error::manifest(format!("{place}: chunks must be a list")))?;
|
||||
require(chunks.len() as u64 == chunk_count(size, chunk_bytes), || {
|
||||
format!("{place}: {} chunks for {size} bytes at {chunk_bytes} per chunk", chunks.len())
|
||||
})?;
|
||||
let mut hashes = Vec::with_capacity(chunks.len());
|
||||
for chunk in chunks {
|
||||
check_sha(Some(chunk), &place)?;
|
||||
hashes.push(chunk.as_str().unwrap_or_default());
|
||||
}
|
||||
require(entry.get("chunks_sha256").and_then(Value::as_str) == Some(chunks_digest(&hashes).as_str()), || {
|
||||
format!("{place}: chunks_sha256 does not match chunks")
|
||||
})?;
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(signature) = m.get("signature") {
|
||||
require(signature.as_str().is_some_and(|s| s.starts_with("ed25519:")), || {
|
||||
"signature must look like 'ed25519:<base64>'".into()
|
||||
})?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ---- typed view -------------------------------------------------------------
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct AppFile {
|
||||
pub path: String,
|
||||
pub size: u64,
|
||||
pub sha256: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
|
||||
pub struct AppSection {
|
||||
pub files: Vec<AppFile>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct DbFile {
|
||||
pub section: String,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub table: Option<String>,
|
||||
pub path: String,
|
||||
pub size: u64,
|
||||
pub sha256: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct DbSection {
|
||||
pub schema_version: u64,
|
||||
pub snapshot: String,
|
||||
pub files: Vec<DbFile>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct TileSet {
|
||||
pub style: String,
|
||||
pub dataset: String,
|
||||
pub journal_seq: u64,
|
||||
pub blocks: u64,
|
||||
pub files: u64,
|
||||
pub bytes: u64,
|
||||
pub index: String,
|
||||
pub index_sha256: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct TilesSection {
|
||||
pub block_shift: u64,
|
||||
pub sets: Vec<TileSet>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct SrcFile {
|
||||
pub path: String,
|
||||
pub size: u64,
|
||||
pub chunks: Vec<String>,
|
||||
pub chunks_sha256: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct SrcSection {
|
||||
pub enabled: bool,
|
||||
pub chunk_bytes: u64,
|
||||
pub files: Vec<SrcFile>,
|
||||
}
|
||||
|
||||
/// A validated manifest's fields (unknown keys are kept only in the raw bytes).
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct Manifest {
|
||||
pub schema: u64,
|
||||
pub product: String,
|
||||
pub version: String,
|
||||
pub released_at: String,
|
||||
#[serde(default)]
|
||||
pub min_launcher: Option<String>,
|
||||
#[serde(default)]
|
||||
pub notes: Option<String>,
|
||||
#[serde(default)]
|
||||
pub app: AppSection,
|
||||
#[serde(default)]
|
||||
pub db: Option<DbSection>,
|
||||
#[serde(default)]
|
||||
pub tiles: Option<TilesSection>,
|
||||
#[serde(default)]
|
||||
pub src: Option<SrcSection>,
|
||||
#[serde(default)]
|
||||
pub signature: Option<String>,
|
||||
}
|
||||
|
||||
/// A manifest that parsed, validated and carries a good signature by the given key.
|
||||
/// Only [`VerifiedManifest::from_bytes`] builds one.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct VerifiedManifest {
|
||||
raw: Vec<u8>,
|
||||
value: Value,
|
||||
manifest: Manifest,
|
||||
}
|
||||
|
||||
impl VerifiedManifest {
|
||||
/// Parse, validate and signature-check manifest bytes (e.g. an installed
|
||||
/// `versions/<v>/manifest.json`, re-checked offline).
|
||||
pub fn from_bytes(raw: Vec<u8>, public_key: &str) -> Result<Self> {
|
||||
let value: Value = serde_json::from_slice(&raw)
|
||||
.map_err(|e| Error::manifest(format!("manifest is not valid UTF-8 JSON: {e}")))?;
|
||||
// Refuse non-integer numbers anywhere, before anything is hashed.
|
||||
canonical_bytes(&value)?;
|
||||
validate(&value)?;
|
||||
crate::signing::verify(&value, public_key)?;
|
||||
let manifest: Manifest = serde_json::from_value(value.clone())
|
||||
.map_err(|e| Error::manifest(format!("validated manifest does not map to its fields: {e}")))?;
|
||||
Ok(VerifiedManifest { raw, value, manifest })
|
||||
}
|
||||
|
||||
/// The bytes exactly as served; written verbatim to `versions/<v>/manifest.json`.
|
||||
pub fn raw(&self) -> &[u8] {
|
||||
&self.raw
|
||||
}
|
||||
|
||||
pub fn value(&self) -> &Value {
|
||||
&self.value
|
||||
}
|
||||
|
||||
pub fn manifest(&self) -> &Manifest {
|
||||
&self.manifest
|
||||
}
|
||||
|
||||
pub fn version(&self) -> &str {
|
||||
&self.manifest.version
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn relpaths_like_python_tests() {
|
||||
assert!(safe_relpath("Odian/x.exe").is_ok());
|
||||
for bad in ["../a", "a/../b", "/a", "C:/a", "a\\b", "a//b", "", "./a", "Odian/Odian.exe:hidden", "a/b.", "a/b ",
|
||||
"CON", "a/aux.txt", "a/Lpt1.log", "a/b\u{0}c", "a/b\nc", "a/b?c", "a/b*c", "a/b\"c", "a/b|c", "a/com¹"] {
|
||||
assert!(safe_relpath(bad).is_err(), "{bad:?}");
|
||||
}
|
||||
for fine in ["a/console.txt", "a/auxiliary", "a/com10", "a/.hidden", "a/b c.txt", "a/comA"] {
|
||||
assert!(safe_relpath(fine).is_ok(), "{fine:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn versions() {
|
||||
assert_eq!(parse_version("0.51.0").unwrap(), (0, 51, 0));
|
||||
for bad in ["1.0", "01.2.3", "0.51.0-rc1", "0.51.0\n", "", "1..2", "a.b.c"] {
|
||||
assert!(parse_version(bad).is_err(), "{bad:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,207 @@
|
||||
//! Install progress, shaped like `InstallProgress` in the launcher's docs/CONTRACT.md,
|
||||
//! throttled to about ten reports per second.
|
||||
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// Install stages. This crate reports `manifest`, `app`, `db`, `src` and `tiles`;
|
||||
/// `restore` and `finalize` exist so the launcher can reuse the type for its own steps.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum Stage {
|
||||
Manifest,
|
||||
App,
|
||||
Db,
|
||||
Src,
|
||||
Tiles,
|
||||
Restore,
|
||||
Finalize,
|
||||
}
|
||||
|
||||
/// One progress report (`InstallProgress` on the wire).
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct Progress {
|
||||
pub stage: Stage,
|
||||
/// 1-based.
|
||||
pub stage_index: u32,
|
||||
pub stage_count: u32,
|
||||
pub done_bytes: u64,
|
||||
/// 0 when unknown.
|
||||
pub total_bytes: u64,
|
||||
pub done_items: u64,
|
||||
pub total_items: u64,
|
||||
pub bytes_per_sec: f64,
|
||||
pub message: Option<String>,
|
||||
}
|
||||
|
||||
const INTERVAL: Duration = Duration::from_millis(100);
|
||||
/// Time constant of the smoothed transfer rate.
|
||||
const RATE_WINDOW_SECS: f64 = 2.0;
|
||||
|
||||
pub(crate) type ProgressFn = Arc<dyn Fn(Progress) + Send + Sync>;
|
||||
|
||||
struct State {
|
||||
stage: Stage,
|
||||
stage_index: u32,
|
||||
stage_count: u32,
|
||||
total_bytes: u64,
|
||||
total_items: u64,
|
||||
label: String,
|
||||
unit: &'static str,
|
||||
sub_total: u64,
|
||||
last_emit: Option<Instant>,
|
||||
rate_bytes: u64,
|
||||
rate_at: Instant,
|
||||
rate: f64,
|
||||
}
|
||||
|
||||
/// Shared by every worker of one install. Counters are atomics; a report is built and
|
||||
/// sent at most every [`INTERVAL`], except stage starts and ends, which always go out.
|
||||
pub(crate) struct Reporter {
|
||||
callback: ProgressFn,
|
||||
done_bytes: AtomicU64,
|
||||
done_items: AtomicU64,
|
||||
sub_done: AtomicU64,
|
||||
transferred: AtomicU64,
|
||||
state: Mutex<State>,
|
||||
}
|
||||
|
||||
impl Reporter {
|
||||
pub fn new(callback: ProgressFn, stage_count: u32) -> Self {
|
||||
Reporter {
|
||||
callback,
|
||||
done_bytes: AtomicU64::new(0),
|
||||
done_items: AtomicU64::new(0),
|
||||
sub_done: AtomicU64::new(0),
|
||||
transferred: AtomicU64::new(0),
|
||||
state: Mutex::new(State {
|
||||
stage: Stage::Manifest,
|
||||
stage_index: 1,
|
||||
stage_count,
|
||||
total_bytes: 0,
|
||||
total_items: 0,
|
||||
label: String::new(),
|
||||
unit: "",
|
||||
sub_total: 0,
|
||||
last_emit: None,
|
||||
rate_bytes: 0,
|
||||
rate_at: Instant::now(),
|
||||
rate: 0.0,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn begin_stage(&self, stage: Stage, stage_index: u32, total_bytes: u64, total_items: u64) {
|
||||
{
|
||||
let mut st = self.state.lock().unwrap_or_else(|p| p.into_inner());
|
||||
st.stage = stage;
|
||||
st.stage_index = stage_index;
|
||||
st.total_bytes = total_bytes;
|
||||
st.total_items = total_items;
|
||||
st.label.clear();
|
||||
st.unit = "";
|
||||
st.sub_total = 0;
|
||||
self.done_bytes.store(0, Ordering::Relaxed);
|
||||
self.done_items.store(0, Ordering::Relaxed);
|
||||
self.sub_done.store(0, Ordering::Relaxed);
|
||||
}
|
||||
self.emit(true);
|
||||
}
|
||||
|
||||
/// The part of the stage the message describes ("tiles dark/ds", 1402 "blocks").
|
||||
pub fn begin_sub(&self, label: String, unit: &'static str, sub_total: u64) {
|
||||
{
|
||||
let mut st = self.state.lock().unwrap_or_else(|p| p.into_inner());
|
||||
st.label = label;
|
||||
st.unit = unit;
|
||||
st.sub_total = sub_total;
|
||||
self.sub_done.store(0, Ordering::Relaxed);
|
||||
}
|
||||
self.emit(true);
|
||||
}
|
||||
|
||||
/// Bytes that just crossed the network (or were copied from a drive).
|
||||
pub fn add_transferred(&self, n: u64) {
|
||||
self.transferred.fetch_add(n, Ordering::Relaxed);
|
||||
self.done_bytes.fetch_add(n, Ordering::Relaxed);
|
||||
self.emit(false);
|
||||
}
|
||||
|
||||
/// Bytes found already in place (skipped files, resumed prefixes, verified chunks).
|
||||
pub fn add_done_bytes(&self, n: u64) {
|
||||
self.done_bytes.fetch_add(n, Ordering::Relaxed);
|
||||
self.emit(false);
|
||||
}
|
||||
|
||||
/// Take back bytes counted for a download that failed verification and restarts.
|
||||
pub fn retract_bytes(&self, n: u64) {
|
||||
let _ = self.done_bytes.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| Some(v.saturating_sub(n)));
|
||||
}
|
||||
|
||||
pub fn items_done(&self, n: u64) {
|
||||
self.done_items.fetch_add(n, Ordering::Relaxed);
|
||||
self.sub_done.fetch_add(n, Ordering::Relaxed);
|
||||
self.emit(false);
|
||||
}
|
||||
|
||||
pub fn finish_stage(&self) {
|
||||
self.emit(true);
|
||||
}
|
||||
|
||||
pub fn transferred(&self) -> u64 {
|
||||
self.transferred.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
fn emit(&self, force: bool) {
|
||||
let now = Instant::now();
|
||||
let progress = {
|
||||
let mut st = if force {
|
||||
self.state.lock().unwrap_or_else(|p| p.into_inner())
|
||||
} else {
|
||||
match self.state.try_lock() {
|
||||
Ok(guard) => guard,
|
||||
Err(_) => return, // another worker is reporting right now
|
||||
}
|
||||
};
|
||||
if !force && st.last_emit.is_some_and(|last| now.duration_since(last) < INTERVAL) {
|
||||
return;
|
||||
}
|
||||
st.last_emit = Some(now);
|
||||
let transferred = self.transferred.load(Ordering::Relaxed);
|
||||
let dt = now.duration_since(st.rate_at).as_secs_f64();
|
||||
if dt >= 0.05 {
|
||||
let instant = transferred.saturating_sub(st.rate_bytes) as f64 / dt;
|
||||
let alpha = (dt / RATE_WINDOW_SECS).min(1.0);
|
||||
st.rate += alpha * (instant - st.rate);
|
||||
st.rate_bytes = transferred;
|
||||
st.rate_at = now;
|
||||
}
|
||||
let mut done_bytes = self.done_bytes.load(Ordering::Relaxed);
|
||||
if st.total_bytes > 0 {
|
||||
done_bytes = done_bytes.min(st.total_bytes);
|
||||
}
|
||||
let message = (!st.label.is_empty()).then(|| {
|
||||
if st.sub_total == 0 {
|
||||
st.label.clone()
|
||||
} else {
|
||||
format!("{}: {}/{} {}", st.label, self.sub_done.load(Ordering::Relaxed), st.sub_total, st.unit)
|
||||
}
|
||||
});
|
||||
Progress {
|
||||
stage: st.stage,
|
||||
stage_index: st.stage_index,
|
||||
stage_count: st.stage_count,
|
||||
done_bytes,
|
||||
total_bytes: st.total_bytes,
|
||||
done_items: self.done_items.load(Ordering::Relaxed),
|
||||
total_items: st.total_items,
|
||||
bytes_per_sec: st.rate.max(0.0).round(),
|
||||
message,
|
||||
}
|
||||
};
|
||||
(self.callback)(progress);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
//! Ed25519 release signatures (verification only; the Python module `odian_packs.signing`).
|
||||
//!
|
||||
//! Key text: `ed25519:<base64 of the raw 32 bytes>`. Signature: Ed25519 over
|
||||
//! [`crate::manifest::signing_payload`], written as `ed25519:<base64 of 64 bytes>`.
|
||||
|
||||
use base64::alphabet;
|
||||
use base64::engine::{DecodePaddingMode, GeneralPurpose, GeneralPurposeConfig};
|
||||
use base64::Engine;
|
||||
use ed25519_dalek::{Signature, VerifyingKey};
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::error::{Error, IntegrityKind, Result};
|
||||
use crate::manifest::signing_payload;
|
||||
|
||||
/// Python's `base64.b64decode(text, validate=True)`: standard alphabet, correct padding,
|
||||
/// no whitespace, and (like CPython) non-zero trailing bits tolerated.
|
||||
const B64: GeneralPurpose = GeneralPurpose::new(
|
||||
&alphabet::STANDARD,
|
||||
GeneralPurposeConfig::new()
|
||||
.with_decode_allow_trailing_bits(true)
|
||||
.with_decode_padding_mode(DecodePaddingMode::RequireCanonical),
|
||||
);
|
||||
|
||||
fn signature_error(detail: &str) -> Error {
|
||||
Error::integrity(IntegrityKind::Signature, "manifest", detail)
|
||||
}
|
||||
|
||||
/// Parse `ed25519:<base64>` (surrounding whitespace ignored).
|
||||
pub fn load_public_key(text: &str) -> Result<VerifyingKey> {
|
||||
let body = text
|
||||
.trim()
|
||||
.strip_prefix("ed25519:")
|
||||
.ok_or_else(|| signature_error("public key must look like 'ed25519:<base64>'"))?;
|
||||
let raw = B64.decode(body).map_err(|_| signature_error("public key is not valid base64"))?;
|
||||
let bytes: [u8; 32] = raw.try_into().map_err(|_| signature_error("an Ed25519 public key is 32 bytes"))?;
|
||||
VerifyingKey::from_bytes(&bytes).map_err(|_| signature_error("public key is not a valid Ed25519 point"))
|
||||
}
|
||||
|
||||
/// Error ([`IntegrityKind::Signature`]) unless `manifest` carries a valid signature by that key.
|
||||
pub fn verify(manifest: &Value, public_key: &str) -> Result<()> {
|
||||
let body = manifest
|
||||
.get("signature")
|
||||
.and_then(Value::as_str)
|
||||
.and_then(|s| s.strip_prefix("ed25519:"))
|
||||
.ok_or_else(|| signature_error("manifest is not signed"))?;
|
||||
let raw = B64.decode(body).map_err(|_| signature_error("signature is not valid base64"))?;
|
||||
let key = load_public_key(public_key)?;
|
||||
let payload = signing_payload(manifest)?;
|
||||
let mismatch = || signature_error("signature does not match: manifest altered, or signed by another key");
|
||||
let signature = Signature::from_slice(&raw).map_err(|_| mismatch())?;
|
||||
key.verify_strict(&payload, &signature).map_err(|_| mismatch())
|
||||
}
|
||||
@@ -0,0 +1,425 @@
|
||||
//! 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/`).
|
||||
|
||||
use std::io::{Read, Seek, SeekFrom};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Mutex;
|
||||
use std::time::Duration;
|
||||
|
||||
use futures_util::StreamExt;
|
||||
use reqwest::header::{HeaderValue, RANGE, RETRY_AFTER};
|
||||
use rusqlite::{Connection, OpenFlags};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
use crate::manifest::{parse_version, tile_index_name};
|
||||
|
||||
/// Where to install from.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Source {
|
||||
/// A site's update service, e.g. `http://192.168.1.41:30097`, with an Engine token.
|
||||
Site { base_url: String, token: String },
|
||||
/// An ODIAN drive (SSD) whose root holds `packs/`, `tiles/` (with `tiles.db`) and `src/`.
|
||||
Drive { root: PathBuf },
|
||||
}
|
||||
|
||||
const RETRIES: u32 = 4;
|
||||
const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
const READ_TIMEOUT: Duration = Duration::from_secs(120);
|
||||
/// A Retry-After beyond this is treated as this (Python sleeps whatever the site says).
|
||||
const MAX_RETRY_AFTER_SECS: f64 = 120.0;
|
||||
const ERROR_BODY_BYTES: usize = 300;
|
||||
|
||||
/// Run blocking work off the async threads.
|
||||
pub(crate) async fn blocking<T, F>(f: F) -> Result<T>
|
||||
where
|
||||
T: Send + 'static,
|
||||
F: FnOnce() -> Result<T> + Send + 'static,
|
||||
{
|
||||
match tokio::task::spawn_blocking(f).await {
|
||||
Ok(result) => result,
|
||||
Err(e) if e.is_panic() => std::panic::resume_unwind(e.into_panic()),
|
||||
Err(_) => Err(Error::Cancelled),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn sleep_or_cancel(secs: f64, cancel: &CancellationToken) -> Result<()> {
|
||||
tokio::select! {
|
||||
_ = cancel.cancelled() => Err(Error::Cancelled),
|
||||
_ = tokio::time::sleep(Duration::from_secs_f64(secs.max(0.0))) => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Percent-encode each segment of a relative path (manifest paths may hold spaces,
|
||||
/// `%`, `#` or non-ASCII; nginx decodes them back before looking the file up).
|
||||
pub(crate) fn encode_path(rel: &str) -> String {
|
||||
let mut out = String::with_capacity(rel.len());
|
||||
for (i, segment) in rel.split('/').enumerate() {
|
||||
if i > 0 {
|
||||
out.push('/');
|
||||
}
|
||||
for b in segment.bytes() {
|
||||
if b.is_ascii_alphanumeric() || matches!(b, b'-' | b'.' | b'_' | b'~') {
|
||||
out.push(b as char);
|
||||
} else {
|
||||
out.push_str(&format!("%{b:02X}"));
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn join_rel(root: &Path, rel: &str) -> PathBuf {
|
||||
let mut path = root.to_path_buf();
|
||||
for part in rel.split('/') {
|
||||
path.push(part);
|
||||
}
|
||||
path
|
||||
}
|
||||
|
||||
fn open_error(path: &Path) -> impl FnOnce(std::io::Error) -> Error + '_ {
|
||||
move |e| {
|
||||
if e.kind() == std::io::ErrorKind::NotFound {
|
||||
Error::NotFound { what: path.display().to_string() }
|
||||
} else {
|
||||
Error::Io { path: path.to_path_buf(), source: e }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A body to copy: an HTTP response, or a drive file already positioned and limited.
|
||||
pub(crate) enum Body {
|
||||
Http { response: reqwest::Response, url: String },
|
||||
File { reader: std::io::Take<std::fs::File>, path: PathBuf },
|
||||
}
|
||||
|
||||
/// A pack-store object.
|
||||
#[derive(Clone, Copy)]
|
||||
pub(crate) enum Pack<'a> {
|
||||
/// `app/<sha256>`.
|
||||
App(&'a str),
|
||||
/// A manifest `db.files[].path` (`db/<snapshot>/...`).
|
||||
Db(&'a str),
|
||||
}
|
||||
|
||||
pub(crate) struct Site {
|
||||
base: String,
|
||||
token: String,
|
||||
client: reqwest::Client,
|
||||
}
|
||||
|
||||
pub(crate) struct Drive {
|
||||
root: PathBuf,
|
||||
}
|
||||
|
||||
pub(crate) enum Backend {
|
||||
Site(Site),
|
||||
Drive(Drive),
|
||||
}
|
||||
|
||||
impl Backend {
|
||||
pub fn new(source: &Source) -> Result<Self> {
|
||||
match source {
|
||||
Source::Site { base_url, token } => {
|
||||
let client = reqwest::Client::builder()
|
||||
.connect_timeout(CONNECT_TIMEOUT)
|
||||
.read_timeout(READ_TIMEOUT)
|
||||
.user_agent(concat!("odian-packs/", env!("CARGO_PKG_VERSION")))
|
||||
.build()
|
||||
.map_err(|e| Error::Http { url: base_url.clone(), status: None, message: e.to_string() })?;
|
||||
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() })),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn current_manifest(&self, cancel: &CancellationToken) -> Result<Vec<u8>> {
|
||||
match self {
|
||||
Backend::Site(site) => site.get_bytes("/v1/versions/current", cancel).await,
|
||||
Backend::Drive(drive) => {
|
||||
let pointer = drive.root.join("packs").join("versions").join("CURRENT");
|
||||
let raw = blocking({
|
||||
let pointer = pointer.clone();
|
||||
move || std::fs::read(&pointer).map_err(open_error(&pointer))
|
||||
})
|
||||
.await?;
|
||||
let text = String::from_utf8_lossy(&raw[..raw.len().min(256)]).trim().to_string();
|
||||
if text.is_empty() {
|
||||
return Err(Error::NotFound { what: format!("{}: no current version", pointer.display()) });
|
||||
}
|
||||
parse_version(&text)
|
||||
.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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn version_manifest(&self, version: &str, cancel: &CancellationToken) -> Result<Vec<u8>> {
|
||||
parse_version(version)?;
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn notes(&self, version: &str, cancel: &CancellationToken) -> Result<Vec<u8>> {
|
||||
parse_version(version)?;
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
/// `style` and `dataset` come from a validated manifest (name rule already applied).
|
||||
pub async fn tile_index(&self, version: &str, style: &str, dataset: &str, cancel: &CancellationToken) -> Result<Vec<u8>> {
|
||||
match self {
|
||||
Backend::Site(site) => {
|
||||
site.get_bytes(&format!("/v1/versions/{version}/tiles/{style}/{dataset}/index"), cancel).await
|
||||
}
|
||||
Backend::Drive(drive) => {
|
||||
let rel = format!("packs/versions/{version}/{}", tile_index_name(style, dataset));
|
||||
drive.read(join_rel(&drive.root, &rel)).await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn pack_label(&self, pack: Pack<'_>) -> String {
|
||||
match pack {
|
||||
Pack::App(sha) => format!("/packs/app/{sha}"),
|
||||
Pack::Db(path) => format!("/packs/{path}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Open a pack object from `offset`. The flag says whether the body really starts at
|
||||
/// `offset` (a site may ignore the Range and send the whole file).
|
||||
pub async fn open_pack(&self, pack: Pack<'_>, offset: u64, cancel: &CancellationToken) -> Result<(Body, bool)> {
|
||||
match self {
|
||||
Backend::Site(site) => {
|
||||
let path = match pack {
|
||||
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}-"));
|
||||
let (response, url) = site.get(&path, range, cancel).await?;
|
||||
let partial = offset > 0 && response.status() == reqwest::StatusCode::PARTIAL_CONTENT;
|
||||
Ok((Body::Http { response, url }, offset == 0 || partial))
|
||||
}
|
||||
Backend::Drive(drive) => {
|
||||
let path = match pack {
|
||||
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?;
|
||||
Ok((body, true))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Inclusive byte range `[start, end]` of a source file.
|
||||
pub async fn open_src_range(&self, rel: &str, start: u64, end: u64, cancel: &CancellationToken) -> Result<Body> {
|
||||
match self {
|
||||
Backend::Site(site) => {
|
||||
let path = format!("/src/{}", encode_path(rel));
|
||||
let (response, url) = site.get(&path, Some(format!("bytes={start}-{end}")), cancel).await?;
|
||||
if response.status() != reqwest::StatusCode::PARTIAL_CONTENT {
|
||||
return Err(Error::Http {
|
||||
url,
|
||||
status: Some(response.status().as_u16()),
|
||||
message: "the DHP did not honour the byte range".into(),
|
||||
});
|
||||
}
|
||||
Ok(Body::Http { response, url })
|
||||
}
|
||||
Backend::Drive(drive) => drive.open_range(join_rel(&drive.root.join("src"), rel), start, end - start + 1).await,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn drive_root(&self) -> Option<&Path> {
|
||||
match self {
|
||||
Backend::Drive(drive) => Some(&drive.root),
|
||||
Backend::Site(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn site(&self) -> Option<&Site> {
|
||||
match self {
|
||||
Backend::Site(site) => Some(site),
|
||||
Backend::Drive(_) => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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`.
|
||||
pub async fn get(&self, path: &str, range: Option<String>, cancel: &CancellationToken) -> Result<(reqwest::Response, String)> {
|
||||
let url = format!("{}{}", self.base, path);
|
||||
let mut delay = 1.0f64;
|
||||
for attempt in 1..=RETRIES {
|
||||
if cancel.is_cancelled() {
|
||||
return Err(Error::Cancelled);
|
||||
}
|
||||
let mut request = self.client.get(&url).bearer_auth(&self.token);
|
||||
if let Some(range) = &range {
|
||||
request = request.header(RANGE, HeaderValue::from_str(range).expect("ascii range"));
|
||||
}
|
||||
let sent = tokio::select! {
|
||||
_ = cancel.cancelled() => return Err(Error::Cancelled),
|
||||
sent = request.send() => sent,
|
||||
};
|
||||
let last = attempt == RETRIES;
|
||||
match sent {
|
||||
Ok(response) if response.status().is_success() => return Ok((response, url)),
|
||||
Ok(response) => {
|
||||
let status = response.status().as_u16();
|
||||
let retry_after = response
|
||||
.headers()
|
||||
.get(RETRY_AFTER)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|v| v.trim().parse::<f64>().ok())
|
||||
.filter(|v| v.is_finite() && *v >= 0.0)
|
||||
.map(|v| v.min(MAX_RETRY_AFTER_SECS));
|
||||
if matches!(status, 502..=504) && !last {
|
||||
sleep_or_cancel(retry_after.unwrap_or(delay), cancel).await?;
|
||||
delay *= 2.0;
|
||||
continue;
|
||||
}
|
||||
let body = response.bytes().await.map(|b| b.to_vec()).unwrap_or_default();
|
||||
let body = String::from_utf8_lossy(&body[..body.len().min(ERROR_BODY_BYTES)]).into_owned();
|
||||
return Err(match status {
|
||||
401 | 403 => Error::Unauthorized { url, status },
|
||||
404 => Error::NotFound { what: format!("{url}: {body}") },
|
||||
503 => Error::SiteBusy { url, attempts: attempt, retry_after_secs: retry_after },
|
||||
_ => Error::Http { url, status: Some(status), message: format!("HTTP {status}: {body}") },
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
if !last {
|
||||
sleep_or_cancel(delay, cancel).await?;
|
||||
delay *= 2.0;
|
||||
continue;
|
||||
}
|
||||
return Err(Error::Http { url, status: None, message: format!("request failed after {RETRIES} attempts: {e}") });
|
||||
}
|
||||
}
|
||||
}
|
||||
unreachable!("the loop returns on its last attempt")
|
||||
}
|
||||
|
||||
pub async fn get_bytes(&self, path: &str, cancel: &CancellationToken) -> Result<Vec<u8>> {
|
||||
let (response, url) = self.get(path, None, cancel).await?;
|
||||
let mut stream = response.bytes_stream();
|
||||
let mut out = Vec::new();
|
||||
loop {
|
||||
let next = tokio::select! {
|
||||
_ = cancel.cancelled() => return Err(Error::Cancelled),
|
||||
next = stream.next() => next,
|
||||
};
|
||||
match next {
|
||||
Some(Ok(chunk)) => out.extend_from_slice(&chunk),
|
||||
Some(Err(e)) => return Err(Error::Http { url, status: None, message: format!("body cut short: {e}") }),
|
||||
None => return Ok(out),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
/// One tile row straight from the journal: `(z, x, y, ext, size, sha256)`.
|
||||
pub(crate) type JournalRow = (i64, i64, i64, String, i64, String);
|
||||
|
||||
/// A drive's tile journal (`tiles/tiles.db`), opened read-only.
|
||||
pub(crate) struct Journal {
|
||||
conn: Mutex<Connection>,
|
||||
path: PathBuf,
|
||||
}
|
||||
|
||||
fn sqlite_uri(path: &Path, immutable: bool) -> String {
|
||||
let absolute = std::path::absolute(path).unwrap_or_else(|_| path.to_path_buf());
|
||||
let text = absolute.to_string_lossy().replace('\\', "/");
|
||||
let mut uri = String::from(if text.starts_with('/') { "file://" } else { "file:///" });
|
||||
for b in text.bytes() {
|
||||
if b.is_ascii_alphanumeric() || matches!(b, b'-' | b'.' | b'_' | b'~' | b'/' | b':') {
|
||||
uri.push(b as char);
|
||||
} else {
|
||||
uri.push_str(&format!("%{b:02X}"));
|
||||
}
|
||||
}
|
||||
uri.push_str("?mode=ro");
|
||||
if immutable {
|
||||
uri.push_str("&immutable=1");
|
||||
}
|
||||
uri
|
||||
}
|
||||
|
||||
impl Journal {
|
||||
/// Blocking. `mode=ro` first; on "unable to open"/"readonly" (a WAL journal whose
|
||||
/// -wal/-shm cannot be created) retry once with `immutable=1`, like the update service.
|
||||
pub fn open(path: &Path) -> Result<Journal> {
|
||||
if !path.is_file() {
|
||||
return Err(Error::NotFound { what: format!("tile journal {}", path.display()) });
|
||||
}
|
||||
let flags = OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX;
|
||||
let mut last = None;
|
||||
for immutable in [false, true] {
|
||||
let attempt = Connection::open_with_flags(sqlite_uri(path, immutable), flags).and_then(|conn| {
|
||||
conn.busy_timeout(Duration::from_secs(30))?;
|
||||
conn.query_row("SELECT COUNT(*) FROM tile WHERE seq < 0", [], |_| Ok(()))?;
|
||||
Ok(conn)
|
||||
});
|
||||
match attempt {
|
||||
Ok(conn) => return Ok(Journal { conn: Mutex::new(conn), path: path.to_path_buf() }),
|
||||
Err(e) => {
|
||||
let message = e.to_string().to_lowercase();
|
||||
let retryable = message.contains("unable to open") || message.contains("readonly");
|
||||
last = Some(e);
|
||||
if !retryable {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(Error::Io { path: path.to_path_buf(), source: std::io::Error::other(last.expect("an attempt ran")) })
|
||||
}
|
||||
|
||||
/// Blocking. The tiles of one block as of `max_seq`, in digest order (`tiles.block_rows`).
|
||||
pub fn block_rows(&self, style: &str, dataset: &str, z: u64, bx: u64, by: u64, max_seq: u64) -> Result<Vec<JournalRow>> {
|
||||
let shift = crate::manifest::BLOCK_SHIFT;
|
||||
let (z, bx, by, max_seq) = (z as i64, bx as i64, by as i64, max_seq as i64);
|
||||
let conn = self.conn.lock().unwrap_or_else(|p| p.into_inner());
|
||||
let run = || -> rusqlite::Result<Vec<JournalRow>> {
|
||||
let mut stmt = conn.prepare_cached(
|
||||
"SELECT z, x, y, ext, size, sha256 FROM tile \
|
||||
WHERE style = ? AND dataset = ? AND z = ? AND x BETWEEN ? AND ? AND y BETWEEN ? AND ? AND seq <= ? \
|
||||
ORDER BY z, x, y, ext",
|
||||
)?;
|
||||
let rows = stmt.query_map(
|
||||
rusqlite::params![
|
||||
style, dataset, z,
|
||||
bx << shift, ((bx + 1) << shift) - 1,
|
||||
by << shift, ((by + 1) << shift) - 1,
|
||||
max_seq
|
||||
],
|
||||
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?, row.get(5)?)),
|
||||
)?;
|
||||
rows.collect()
|
||||
};
|
||||
run().map_err(|e| Error::Io { path: self.path.clone(), source: std::io::Error::other(e) })
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,272 @@
|
||||
//! One tile block: fetch it (site tar stream or drive journal copy) into a staging
|
||||
//! directory, check every name and the block digest from the bytes, then move the tiles in.
|
||||
|
||||
use std::io::Read;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use bytes::{Buf, Bytes};
|
||||
use futures_util::StreamExt;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use crate::copy::{copy_hashing, unique_hex};
|
||||
use crate::error::{Error, IntegrityKind, Result};
|
||||
use crate::progress::Reporter;
|
||||
use crate::source::{Journal, Site};
|
||||
use crate::tiles::{block_digest, block_of, parse_tile_name, tile_relpath, IndexBlock, TileRow};
|
||||
|
||||
const TILE_BUF: usize = 256 * 1024;
|
||||
|
||||
/// `<set_dir>/../.incoming/<hex>`, removed however the block ends.
|
||||
struct Staging {
|
||||
path: PathBuf,
|
||||
}
|
||||
|
||||
impl Staging {
|
||||
fn create(set_dir: &Path) -> Result<Staging> {
|
||||
let root = set_dir.parent().unwrap_or(set_dir).join(".incoming");
|
||||
std::fs::create_dir_all(&root).map_err(Error::io(&root))?;
|
||||
loop {
|
||||
let path = root.join(unique_hex());
|
||||
match std::fs::create_dir(&path) {
|
||||
Ok(()) => return Ok(Staging { path }),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
|
||||
Err(e) => return Err(Error::Io { path, source: e }),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Staging {
|
||||
fn drop(&mut self) {
|
||||
let _ = std::fs::remove_dir_all(&self.path);
|
||||
}
|
||||
}
|
||||
|
||||
fn tile_path(root: &Path, z: u64, x: u64, y: u64, ext: &str) -> PathBuf {
|
||||
root.join(z.to_string()).join(x.to_string()).join(format!("{y}{ext}"))
|
||||
}
|
||||
|
||||
/// Digest and count check, then move every tile from staging into the set.
|
||||
fn finish_block(what: &str, block: &IndexBlock, rows: &[TileRow], staging: &Path, set_dir: &Path) -> Result<()> {
|
||||
let got = block_digest(rows);
|
||||
if got != block.digest || rows.len() as u64 != block.files {
|
||||
return Err(Error::integrity(
|
||||
IntegrityKind::BlockDigest,
|
||||
what,
|
||||
format!(
|
||||
"block digest {got} with {} tiles, signed index says {} with {}",
|
||||
rows.len(),
|
||||
block.digest,
|
||||
block.files
|
||||
),
|
||||
));
|
||||
}
|
||||
for (z, x, y, ext, _, _) in rows {
|
||||
let target = tile_path(set_dir, *z, *x, *y, ext);
|
||||
if let Some(parent) = target.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(Error::io(parent))?;
|
||||
}
|
||||
let staged = tile_path(staging, *z, *x, *y, ext);
|
||||
std::fs::rename(&staged, &target).map_err(Error::io(&target))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ---- site: tar stream ---------------------------------------------------------
|
||||
|
||||
/// Blocking `Read` over body chunks sent from the async side.
|
||||
struct ChannelReader {
|
||||
rx: tokio::sync::mpsc::Receiver<std::result::Result<Bytes, String>>,
|
||||
current: Bytes,
|
||||
}
|
||||
|
||||
impl Read for ChannelReader {
|
||||
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
||||
while self.current.is_empty() {
|
||||
match self.rx.blocking_recv() {
|
||||
Some(Ok(chunk)) => self.current = chunk,
|
||||
Some(Err(message)) => return Err(std::io::Error::other(message)),
|
||||
None => return Ok(0),
|
||||
}
|
||||
}
|
||||
let n = buf.len().min(self.current.len());
|
||||
buf[..n].copy_from_slice(&self.current[..n]);
|
||||
self.current.advance(n);
|
||||
Ok(n)
|
||||
}
|
||||
}
|
||||
|
||||
/// Download `url_path` (an uncompressed tar of `z/x/y.ext`) and install it as `block`.
|
||||
pub(crate) async fn fetch_site_block(
|
||||
site: &Site,
|
||||
url_path: String,
|
||||
block: IndexBlock,
|
||||
set_dir: PathBuf,
|
||||
reporter: Arc<Reporter>,
|
||||
cancel: CancellationToken,
|
||||
) -> Result<()> {
|
||||
let (response, url) = site.get(&url_path, None, &cancel).await?;
|
||||
let (tx, rx) = tokio::sync::mpsc::channel::<std::result::Result<Bytes, String>>(16);
|
||||
|
||||
let extractor = tokio::task::spawn_blocking({
|
||||
let (url, cancel) = (url.clone(), cancel.clone());
|
||||
move || extract_tar(ChannelReader { rx, current: Bytes::new() }, &url, &block, &set_dir, &reporter, &cancel)
|
||||
});
|
||||
|
||||
let feeder = async {
|
||||
let mut stream = response.bytes_stream();
|
||||
loop {
|
||||
let next = tokio::select! {
|
||||
_ = cancel.cancelled() => {
|
||||
let _ = tx.send(Err("cancelled".into())).await;
|
||||
return Some(Error::Cancelled);
|
||||
}
|
||||
next = stream.next() => next,
|
||||
};
|
||||
match next {
|
||||
Some(Ok(chunk)) => {
|
||||
if tx.send(Ok(chunk)).await.is_err() {
|
||||
return None; // the extractor is done (tar end reached, or it failed)
|
||||
}
|
||||
}
|
||||
Some(Err(e)) => {
|
||||
let message = format!("body cut short: {e}");
|
||||
let _ = tx.send(Err(message.clone())).await;
|
||||
return Some(Error::Http { url: url.clone(), status: None, message });
|
||||
}
|
||||
None => return None,
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let feed_error = feeder.await;
|
||||
drop(tx);
|
||||
let extracted = match extractor.await {
|
||||
Ok(result) => result,
|
||||
Err(e) if e.is_panic() => std::panic::resume_unwind(e.into_panic()),
|
||||
Err(_) => Err(Error::Cancelled),
|
||||
};
|
||||
match extracted {
|
||||
Ok(()) => Ok(()),
|
||||
// A network error or a cancel explains a broken tar better than the tar error does.
|
||||
Err(e) => Err(feed_error.unwrap_or(e)),
|
||||
}
|
||||
}
|
||||
|
||||
fn extract_tar<R: Read>(
|
||||
reader: R,
|
||||
what: &str,
|
||||
block: &IndexBlock,
|
||||
set_dir: &Path,
|
||||
reporter: &Reporter,
|
||||
cancel: &CancellationToken,
|
||||
) -> Result<()> {
|
||||
let staging = Staging::create(set_dir)?;
|
||||
let tar_error = |e: std::io::Error| Error::integrity(IntegrityKind::TarEntry, what, format!("malformed tar stream: {e}"));
|
||||
let mut archive = tar::Archive::new(reader);
|
||||
let mut rows: Vec<TileRow> = Vec::new();
|
||||
let mut buf = vec![0u8; TILE_BUF];
|
||||
for entry in archive.entries().map_err(tar_error)? {
|
||||
if cancel.is_cancelled() {
|
||||
return Err(Error::Cancelled);
|
||||
}
|
||||
let mut entry = entry.map_err(tar_error)?;
|
||||
let name = String::from_utf8_lossy(&entry.path_bytes()).into_owned();
|
||||
let kind = entry.header().entry_type();
|
||||
let parsed = if kind.is_file() || kind == tar::EntryType::Continuous { parse_tile_name(&name) } else { None };
|
||||
let (z, x, y, ext) = parsed.ok_or_else(|| {
|
||||
Error::integrity(IntegrityKind::TarEntry, what, format!("unexpected tar entry {name:?}"))
|
||||
})?;
|
||||
if block_of(z, x, y) != (block.z, block.bx, block.by) {
|
||||
return Err(Error::integrity(IntegrityKind::TarEntry, what, format!("tile {name} is outside this block")));
|
||||
}
|
||||
let target = tile_path(&staging.path, z, x, y, &ext);
|
||||
if let Some(parent) = target.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(Error::io(parent))?;
|
||||
}
|
||||
let (size, sha) = copy_hashing(&mut entry, &target, &mut buf, reporter, cancel, &tar_error)?;
|
||||
rows.push((z, x, y, ext, size, sha));
|
||||
}
|
||||
finish_block(what, block, &rows, &staging.path, set_dir)
|
||||
}
|
||||
|
||||
// ---- drive: journal + tile store --------------------------------------------------
|
||||
|
||||
pub(crate) struct DriveBlock<'a> {
|
||||
pub journal: &'a Journal,
|
||||
/// `<drive>/tiles/<style>/<dataset>`.
|
||||
pub set_root: &'a Path,
|
||||
pub style: &'a str,
|
||||
pub dataset: &'a str,
|
||||
pub journal_seq: u64,
|
||||
}
|
||||
|
||||
/// Blocking: copy one block's tiles (journal rows at `seq <= journal_seq`) from the drive.
|
||||
pub(crate) fn copy_drive_block(
|
||||
drive: &DriveBlock<'_>,
|
||||
what: &str,
|
||||
block: &IndexBlock,
|
||||
set_dir: &Path,
|
||||
reporter: &Reporter,
|
||||
cancel: &CancellationToken,
|
||||
) -> Result<()> {
|
||||
let journal_rows = drive.journal.block_rows(drive.style, drive.dataset, block.z, block.bx, block.by, drive.journal_seq)?;
|
||||
let mut expected: Vec<TileRow> = Vec::with_capacity(journal_rows.len());
|
||||
for (z, x, y, ext, size, sha) in journal_rows {
|
||||
let unsafe_row = || {
|
||||
Error::integrity(IntegrityKind::TarEntry, what, format!("the drive's tile journal holds an unsafe tile {z}/{x}/{y}{ext:?}"))
|
||||
};
|
||||
if z < 0 || x < 0 || y < 0 || size < 0 {
|
||||
return Err(unsafe_row());
|
||||
}
|
||||
let (z, x, y, size) = (z as u64, x as u64, y as u64, size as u64);
|
||||
// The same name rule the site path applies to tar entries.
|
||||
if parse_tile_name(&tile_relpath(z, x, y, &ext)) != Some((z, x, y, ext.clone())) {
|
||||
return Err(unsafe_row());
|
||||
}
|
||||
expected.push((z, x, y, ext, size, sha));
|
||||
}
|
||||
// The update service's pre-stream check (its 409): the journal must match the index.
|
||||
let total: u64 = expected.iter().map(|row| row.4).sum();
|
||||
let journal_digest = block_digest(&expected);
|
||||
if expected.len() as u64 != block.files || total != block.bytes || journal_digest != block.digest {
|
||||
return Err(Error::integrity(
|
||||
IntegrityKind::BlockDigest,
|
||||
what,
|
||||
format!(
|
||||
"the drive's tile journal does not match the signed index: {} tiles, {total} bytes, digest {journal_digest}; \
|
||||
index says {} tiles, {} bytes, digest {}",
|
||||
expected.len(),
|
||||
block.files,
|
||||
block.bytes,
|
||||
block.digest
|
||||
),
|
||||
));
|
||||
}
|
||||
|
||||
let staging = Staging::create(set_dir)?;
|
||||
let mut rows: Vec<TileRow> = Vec::with_capacity(expected.len());
|
||||
let mut buf = vec![0u8; TILE_BUF];
|
||||
for (z, x, y, ext, _, _) in &expected {
|
||||
if cancel.is_cancelled() {
|
||||
return Err(Error::Cancelled);
|
||||
}
|
||||
let source = tile_path(drive.set_root, *z, *x, *y, ext);
|
||||
let mut file = std::fs::File::open(&source).map_err(|e| {
|
||||
if e.kind() == std::io::ErrorKind::NotFound {
|
||||
Error::NotFound { what: source.display().to_string() }
|
||||
} else {
|
||||
Error::Io { path: source.clone(), source: e }
|
||||
}
|
||||
})?;
|
||||
let target = tile_path(&staging.path, *z, *x, *y, ext);
|
||||
if let Some(parent) = target.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(Error::io(parent))?;
|
||||
}
|
||||
let read_error = |e: std::io::Error| Error::Io { path: source.clone(), source: e };
|
||||
let (size, sha) = copy_hashing(&mut file, &target, &mut buf, reporter, cancel, &read_error)?;
|
||||
rows.push((*z, *x, *y, ext.clone(), size, sha));
|
||||
}
|
||||
finish_block(what, block, &rows, &staging.path, set_dir)
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
//! Tile blocks, block digests and frozen tile indexes (the Python module `odian_packs.tiles`).
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use crate::canonical::canonical_bytes;
|
||||
use crate::error::{Error, Result};
|
||||
use crate::hashing::hex;
|
||||
use crate::manifest::BLOCK_SHIFT;
|
||||
|
||||
/// One tile as the digest sees it: `(z, x, y, ext, size, sha256)`.
|
||||
pub type TileRow = (u64, u64, u64, String, u64, String);
|
||||
|
||||
pub fn block_of(z: u64, x: u64, y: u64) -> (u64, u64, u64) {
|
||||
(z, x >> BLOCK_SHIFT, y >> BLOCK_SHIFT)
|
||||
}
|
||||
|
||||
/// `"{z}/{x}/{y}{ext} {size} {sha256}\n"`.
|
||||
pub fn tile_line(z: u64, x: u64, y: u64, ext: &str, size: u64, sha256: &str) -> String {
|
||||
format!("{z}/{x}/{y}{ext} {size} {sha256}\n")
|
||||
}
|
||||
|
||||
pub fn tile_relpath(z: u64, x: u64, y: u64, ext: &str) -> String {
|
||||
format!("{z}/{x}/{y}{ext}")
|
||||
}
|
||||
|
||||
/// Digest of tile rows; sorts them by `(z, x, y, ext)` first, like Python.
|
||||
pub fn block_digest(rows: &[TileRow]) -> String {
|
||||
let mut sorted: Vec<&TileRow> = rows.iter().collect();
|
||||
sorted.sort_by(|a, b| (a.0, a.1, a.2, a.3.as_bytes()).cmp(&(b.0, b.1, b.2, b.3.as_bytes())));
|
||||
let mut digest = Sha256::new();
|
||||
for (z, x, y, ext, size, sha) in sorted {
|
||||
digest.update(tile_line(*z, *x, *y, ext, *size, sha).as_bytes());
|
||||
}
|
||||
hex(&digest.finalize())
|
||||
}
|
||||
|
||||
/// sha256 of an index document's canonical JSON.
|
||||
pub fn index_sha256(index: &Value) -> Result<String> {
|
||||
Ok(hex(&Sha256::digest(canonical_bytes(index)?)))
|
||||
}
|
||||
|
||||
/// `(z, x, y, ext)` of a tile name `z/x/y.ext` (the installer's tar-entry rule:
|
||||
/// ASCII digits, and an extension of 1 to 8 ASCII letters or digits).
|
||||
pub fn parse_tile_name(name: &str) -> Option<(u64, u64, u64, String)> {
|
||||
let mut parts = name.split('/');
|
||||
let (zs, xs, last) = (parts.next()?, parts.next()?, parts.next()?);
|
||||
if parts.next().is_some() {
|
||||
return None;
|
||||
}
|
||||
let dot = last.find('.')?;
|
||||
let (ys, ext) = last.split_at(dot);
|
||||
let ext_body = &ext[1..];
|
||||
if ext_body.is_empty() || ext_body.len() > 8 || !ext_body.bytes().all(|b| b.is_ascii_alphanumeric()) {
|
||||
return None;
|
||||
}
|
||||
Some((digits(zs)?, digits(xs)?, digits(ys)?, ext.to_string()))
|
||||
}
|
||||
|
||||
fn digits(s: &str) -> Option<u64> {
|
||||
if s.is_empty() || !s.bytes().all(|b| b.is_ascii_digit()) {
|
||||
return None;
|
||||
}
|
||||
// Leading zeros are allowed (Python's int() takes them); the value is what counts.
|
||||
let trimmed = s.trim_start_matches('0');
|
||||
if trimmed.is_empty() {
|
||||
return Some(0);
|
||||
}
|
||||
trimmed.parse().ok()
|
||||
}
|
||||
|
||||
/// One block of a frozen tile index.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct IndexBlock {
|
||||
pub z: u64,
|
||||
pub bx: u64,
|
||||
pub by: u64,
|
||||
pub files: u64,
|
||||
pub bytes: u64,
|
||||
pub digest: String,
|
||||
}
|
||||
|
||||
impl IndexBlock {
|
||||
/// The key recorded in `.blocks-done.json`: `"{z}/{bx}/{by}:{digest}"`.
|
||||
pub fn done_key(&self) -> String {
|
||||
format!("{}/{}/{}:{}", self.z, self.bx, self.by, self.digest)
|
||||
}
|
||||
}
|
||||
|
||||
/// A frozen tile index document (format 1).
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct TileIndex {
|
||||
pub format: u64,
|
||||
pub style: String,
|
||||
pub dataset: String,
|
||||
pub block_shift: u64,
|
||||
pub journal_seq: u64,
|
||||
pub files: u64,
|
||||
pub bytes: u64,
|
||||
pub blocks: Vec<IndexBlock>,
|
||||
}
|
||||
|
||||
/// Parse index bytes and check them against the manifest's `index_sha256`.
|
||||
pub fn verify_index(raw: &[u8], expected_sha256: &str, what: &str) -> Result<TileIndex> {
|
||||
let value: Value = serde_json::from_slice(raw)
|
||||
.map_err(|e| Error::manifest(format!("{what}: tile index is not valid JSON: {e}")))?;
|
||||
let got = index_sha256(&value)?;
|
||||
if got != expected_sha256 {
|
||||
return Err(Error::integrity(
|
||||
crate::error::IntegrityKind::IndexHash,
|
||||
what,
|
||||
format!("index sha256 {got}, the signed manifest says {expected_sha256}"),
|
||||
));
|
||||
}
|
||||
let index: TileIndex = serde_json::from_value(value)
|
||||
.map_err(|e| Error::manifest(format!("{what}: signed tile index is malformed: {e}")))?;
|
||||
if index.format != 1 || index.block_shift != BLOCK_SHIFT as u64 {
|
||||
return Err(Error::manifest(format!("{what}: tile index must be format 1 with block_shift {BLOCK_SHIFT}")));
|
||||
}
|
||||
Ok(index)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn tile_names() {
|
||||
assert_eq!(parse_tile_name("7/63/5.png"), Some((7, 63, 5, ".png".into())));
|
||||
assert_eq!(parse_tile_name("07/1/2.pbf"), Some((7, 1, 2, ".pbf".into())));
|
||||
for bad in ["7/1/2", "7/1/2.", "7/1/2.tar.gz", "7/1/x.png", "7/-1/2.png", "7/1/2/3.png", "../1/2.png", "7/1/2.abcdefghi"] {
|
||||
assert_eq!(parse_tile_name(bad), None, "{bad}");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user