diff --git a/docs/LOGGING.md b/docs/LOGGING.md index fd474080..a31f419e 100644 --- a/docs/LOGGING.md +++ b/docs/LOGGING.md @@ -23,12 +23,16 @@ implementation, and example of a structured record (step name, phase, elapsed time, RSS/ cgroup memory snapshot — logged at the start and end of every pipeline step). A step can log its own internal sub-steps the same way, under a -dotted name (e.g. `import_osm.fetch`, `import_osm.assemble`) — not a +dotted name (e.g. `fetch_inputs.osm`, `import_osm.assemble`) — not a second logging mechanism, the exact same `run_step` helper, just called again from inside a step for finer timing resolution than that step’s -own start/end pair alone would give. `import_osm` does this for its -fetch/open/prune/assemble/index-build phases; see -[`src/pipeline/osm/mod.rs`](../src/pipeline/osm/mod.rs). +own start/end pair alone would give. `fetch_inputs` does this for +each input it downloads (`fetch_inputs.atp`, `fetch_inputs.osm`; these +run concurrently, so their timings overlap, and a fetch cancelled +because another one failed logs an `end` record plus a "cancelled" +warning), see [`src/pipeline/inputs.rs`](../src/pipeline/inputs.rs); +`import_osm` does it for its open/prune/assemble/index-build phases, +see [`src/pipeline/osm/mod.rs`](../src/pipeline/osm/mod.rs). ## Where weekly-run logs end up diff --git a/docs/PRODUCTION.md b/docs/PRODUCTION.md index b73badcd..8c17f979 100644 --- a/docs/PRODUCTION.md +++ b/docs/PRODUCTION.md @@ -27,6 +27,11 @@ gave: | 6g | 2h27m53s | 10m17s | 2h46m28s | | 4g | 4h02m46s | 10m36s | 4h21m36s | +(Measured before [#830](https://github.com/brawer/osmdiffs/issues/830), +when `import_osm` still included the ~28-minute planet download and a +separate hashing pass; both now happen in `fetch_inputs`, with the hash +computed during the download.) + **6GB is the measured floor** on this CPU count: 8g and 6g are indistinguishable from a comfortable 12g baseline; 4g still completes correctly (identical row counts, all `validate` hard checks pass — the @@ -361,9 +366,12 @@ get started: cache, not heap) — if that ratio drops, something’s changed about the access pattern this design depends on. - **Step timings drifting**: `import_osm`’s own sub-steps - (`import_osm.fetch`/`.open`/`.prune`/`.assemble`/`.index`, logged + (`import_osm.open`/`.prune`/`.assemble`/`.index`, logged individually as of - [#761](https://github.com/brawer/osmdiffs/pull/761)) are worth + [#761](https://github.com/brawer/osmdiffs/pull/761)) and the planet + download (`fetch_inputs.osm`, part of `import_osm` as + `import_osm.fetch` before + [#830](https://github.com/brawer/osmdiffs/issues/830)) are worth watching over time as the planet grows — a slow drift is expected; a sudden jump on an otherwise-unchanged config is worth investigating the way this document’s own `--mem-limit` sweep did. diff --git a/docs/TECHNICAL_DESIGN.md b/docs/TECHNICAL_DESIGN.md index da88f30b..e5ab169c 100644 --- a/docs/TECHNICAL_DESIGN.md +++ b/docs/TECHNICAL_DESIGN.md @@ -179,10 +179,18 @@ graph TD ATP_SRC@{ shape: cloud, label: "AllThePlaces weekly dump" } OSM_SRC@{ shape: cloud, label: "OpenStreetMap planet" } - ATP_SRC --> IMPORT_ATP(import_atp) --> ATP_PARQUET[alltheplaces.parquet] + subgraph INPUTS [Inputs] + ATP_SRC + OSM_SRC + end + INPUTS --> FETCH_INPUTS("fetch_inputs
(all in parallel)") + FETCH_INPUTS --> ATP_ZIP[alltheplaces.zip] + FETCH_INPUTS --> OSM_PBF[planet-latest.osm.pbf] + + ATP_ZIP --> IMPORT_ATP(import_atp) --> ATP_PARQUET[alltheplaces.parquet] ATP_PARQUET --> COLLECT_WIKI(collect_wikidata_ids) --> WIKIDATA[alltheplaces.wikidata-ids] - OSM_SRC --> IMPORT_OSM(import_osm) --> OSM_INDEX[osm-features.index] + OSM_PBF --> IMPORT_OSM(import_osm) --> OSM_INDEX[osm-features.index] ATP_PARQUET --> CONFLATE(conflate) OSM_INDEX --> CONFLATE @@ -210,7 +218,7 @@ graph TD UPLOAD_BOM --> UPLOAD_DATAPACKAGE(upload_datapackage
data/datapackage.json) --> S3A classDef process fill:#fce4ec,stroke:#ad1457,stroke-width:2px,color:#4a0e28,font-weight:bold; - class IMPORT_ATP,COLLECT_WIKI,IMPORT_OSM,CONFLATE,UPLOAD_CONFLATED,EXTRACT_CONFLATED_LAYERS,RENDER_CONFLATED_OVERVIEW,RENDER_CONFLATED_DETAIL,JOIN_CONFLATED_TILES,UPLOAD_CONFLATED_TILES,SUGGEST_EDITS,RENDER_TILES,UPLOAD_TILES,UPLOAD_BOM,UPLOAD_DATAPACKAGE process; + class FETCH_INPUTS,IMPORT_ATP,COLLECT_WIKI,IMPORT_OSM,CONFLATE,UPLOAD_CONFLATED,EXTRACT_CONFLATED_LAYERS,RENDER_CONFLATED_OVERVIEW,RENDER_CONFLATED_DETAIL,JOIN_CONFLATED_TILES,UPLOAD_CONFLATED_TILES,SUGGEST_EDITS,RENDER_TILES,UPLOAD_TILES,UPLOAD_BOM,UPLOAD_DATAPACKAGE process; ``` (Pink boxes are processing steps; plain rectangles are the files they @@ -251,9 +259,29 @@ worth refreshing occasionally, not a guarantee — every run’s own [`LOGGING.md`](LOGGING.md)), so up-to-date numbers are always one log fetch away. +- **`fetch_inputs`** ([`src/pipeline/inputs.rs`](../src/pipeline/inputs.rs)) + — downloads every external input in parallel, before any processing + step runs: AllThePlaces’ latest published run + ([`atp/fetch.rs`](../src/pipeline/atp/fetch.rs), logged as sub-step + `fetch_inputs.atp`) and the OpenStreetMap planet dump + ([`osm/fetch.rs`](../src/pipeline/osm/fetch.rs), `fetch_inputs.osm`; + over plain HTTPS, a redirect straight to a well-provisioned cloud + object store, not BitTorrent — see + [#755](https://github.com/brawer/osmdiffs/pull/755) for why that + switch happened). Each input is hashed (SHA-256) while it streams in, + and its provenance written to a `*.meta.json` sidecar. If any + download fails, the others are cancelled and the run aborts within + minutes; an interrupted planet download is resumed on the next + attempt. An input already in `--workdir` is reused rather than + downloaded: when its sidecar is there along with either the raw file + or the artifact built from it (`alltheplaces.parquet`, + `osm-features.index`), so deleting the ~90 GB planet once its index + is built doesn’t cost a re-download. A planet `.pbf` dropped in by + hand without a sidecar (e.g. a regional extract for testing) is + accepted too. Dominated by the planet download: **~28 minutes** on + the cpx42 run below; the other inputs download alongside it. - **`import_atp`** ([`src/pipeline/atp/`](../src/pipeline/atp/)) — - downloads AllThePlaces’ latest published run (`fetch.rs`) and parses - every spider’s GeoJSON output out of the zip, in parallel, filtering + parses every spider’s GeoJSON output out of the AllThePlaces zip, in parallel, filtering out any dataset not usable for OSM by its declared license or an explicit `use:openstreetmap` marker (`is_usable_for_osm()` in [`src/pipeline/atp/mod.rs`](../src/pipeline/atp/mod.rs)), and writing @@ -266,10 +294,7 @@ fetch away. tracks elsewhere, [#682](https://github.com/brawer/osmdiffs/issues/682)); not consumed by anything yet. Not separately timed. - **`import_osm`** ([`src/pipeline/osm/`](../src/pipeline/osm/)) — - downloads the OpenStreetMap planet dump over plain HTTPS (`fetch.rs`; - a redirect straight to a well-provisioned cloud object store, not - BitTorrent — see [#755](https://github.com/brawer/osmdiffs/pull/755) - for why that switch happened); does a first pass over it that + does a first pass over the planet dump that decides, by tag, which nodes/ways/relations are even worth fully assembling, and which node coordinates and relation members they’ll need (`prune.rs`); builds real OGC geometry (point/line/polygon) for @@ -284,7 +309,9 @@ fetch away. figure. On the newer, HTTPS-based cpx42 run, the download itself took **~28 minutes** — roughly 10x faster than the old BitTorrent baseline — with the whole step (download + SHA-256 hash + prune/ - assemble/index-build) completing in **2h24m12s**. The two runs’ + assemble/index-build) completing in **2h24m12s**; the download and + hash have since moved to `fetch_inputs`, with the hash now computed + while downloading instead of in a separate pass. The two runs’ compute-heavy portions (prune/assemble/index-build) aren’t a clean apples-to-apples comparison, since they ran under different CPU counts and memory limits. diff --git a/scripts/test-on-hetzner/remote/fetch_test_extract.sh b/scripts/test-on-hetzner/remote/fetch_test_extract.sh index 76e37d23..4475f990 100755 --- a/scripts/test-on-hetzner/remote/fetch_test_extract.sh +++ b/scripts/test-on-hetzner/remote/fetch_test_extract.sh @@ -2,7 +2,7 @@ # Pushed to the remote VM and run there by `cloud_test.py start # --containerized --regional-extract`. Downloads a Geofabrik regional # extract straight to the workdir's planet-latest.osm.pbf, skipping the -# multi-hour BitTorrent download entirely -- import_osm's fetch_planet() +# full-planet download entirely -- fetch_inputs's fetch_planet() # computes its own .meta.json sidecar for whatever valid .pbf shows up # (src/pipeline/osm/fetch.rs), so no sidecar needs to be faked here. set -euo pipefail diff --git a/src/pipeline/atp/fetch.rs b/src/pipeline/atp/fetch.rs index 2c6f8671..9ebcc581 100644 --- a/src/pipeline/atp/fetch.rs +++ b/src/pipeline/atp/fetch.rs @@ -1,4 +1,5 @@ use crate::make_download_bar; +use crate::pipeline::inputs::Input; use crate::utils::to_hex; use anyhow::{Context, Ok, Result, anyhow}; use aws_lc_rs::digest::{Context as DigestContext, SHA256}; @@ -85,15 +86,29 @@ pub struct AtpMetadata { pub sha256: Option, } +/// Makes the latest usable AllThePlaces run available as a local input +/// in `workdir`, downloading it (discovered via `url`, i.e. +/// `history.json`) only if nothing in `workdir` can stand in for it. In +/// order: +/// +/// 1. `alltheplaces.zip` exists: reused, with the metadata read from its +/// sidecar -- a hard error if that sidecar is missing or unreadable, +/// since there's no way to recover which run the zip came from. +/// 2. The zip is gone, but its sidecar and the `alltheplaces.parquet` +/// built from it (see `import_atp`) are both still there: only the +/// metadata is returned, and the returned path does not exist. +/// `import_atp` then reuses the parquet file. +/// 3. Otherwise: downloaded, hashed while streaming, and its metadata +/// persisted alongside. pub async fn fetch_atp( url: &str, client: &Client, progress: &MultiProgress, workdir: &Path, -) -> Result<(PathBuf, AtpMetadata)> { - let out_path: PathBuf = workdir.join("alltheplaces.zip"); +) -> Result> { + let out_path: PathBuf = workdir.join(super::ATP_ZIP_FILENAME); let meta_json_path = workdir.join(META_JSON_FILENAME); - if out_path.exists() { + if tokio::fs::try_exists(&out_path).await? { let metadata = read_meta_json(&meta_json_path).await.with_context(|| { format!( "{} exists, but its metadata could not be read from {}", @@ -101,13 +116,27 @@ pub async fn fetch_atp( meta_json_path.display() ) })?; - return Ok((out_path, metadata)); + return Ok(Input { + path: out_path, + metadata, + }); + } + if tokio::fs::try_exists(workdir.join(super::ATP_PARQUET_FILENAME)).await? + && let Result::Ok(metadata) = read_meta_json(&meta_json_path).await + { + return Ok(Input { + path: out_path, + metadata, + }); } let tmp_path = workdir.join("alltheplaces.zip.tmp"); let metadata = download_atp(url, client, progress, &tmp_path, &meta_json_path).await?; - std::fs::rename(&tmp_path, &out_path)?; // atomic file system operation - Ok((out_path, metadata)) + tokio::fs::rename(&tmp_path, &out_path).await?; // atomic file system operation + Ok(Input { + path: out_path, + metadata, + }) } async fn download_atp( @@ -368,7 +397,7 @@ mod tests { let client = test_client(&server); let progress = MultiProgress::with_draw_target(ProgressDrawTarget::hidden()); let workdir = TempDir::new()?; - let (path, metadata) = + let Input { path, metadata } = fetch_atp(&mock_history_url, &client, &progress, workdir.path()).await?; mock_history.assert_async().await; mock_atp_data.assert_async().await; @@ -399,11 +428,58 @@ mod tests { // Fetching again must not hit the network a second time, and // must return the metadata read back from disk. - let (path2, metadata2) = - fetch_atp(&mock_history_url, &client, &progress, workdir.path()).await?; - assert_eq!(path2, path); - assert_eq!(metadata2, metadata); + let again = fetch_atp(&mock_history_url, &client, &progress, workdir.path()).await?; + assert_eq!(again.path, path); + assert_eq!(again.metadata, metadata); + + Ok(()) + } + #[tokio::test] + async fn test_fetch_atp_reuses_parquet_when_zip_was_deleted() -> Result<()> { + // The zip is gone, but its sidecar and the parquet file built + // from it are still there: no download (the history URL points + // to nothing, so any attempt would fail). + let workdir = TempDir::new()?; + let metadata = AtpMetadata { + run_id: "2026-03-04-15-16-17".to_string(), + output_url: "https://example.org/output.zip".to_string(), + history_url: ATP_RUN_HISTORY_URL.to_string(), + start_time: UtcDateTime::from_unix_timestamp(1772637377)?, + end_time: UtcDateTime::from_unix_timestamp(1772647377)?, + spiders: MIN_SPIDERS, + total_lines: MIN_TOTAL_LINES, + size_bytes: MIN_SIZE_BYTES, + sha256: Some("00".repeat(32)), + }; + write_meta_json(&metadata, &workdir.path().join(META_JSON_FILENAME)).await?; + tokio::fs::write(workdir.path().join("alltheplaces.parquet"), b"").await?; + + let client = Client::builder().no_proxy().build()?; + let progress = MultiProgress::with_draw_target(ProgressDrawTarget::hidden()); + let input = fetch_atp( + "http://127.0.0.1:9/history.json", + &client, + &progress, + workdir.path(), + ) + .await?; + assert_eq!(input.metadata, metadata); + assert!(!input.path.exists()); + + // Without the sidecar, the parquet file alone isn't enough: there + // would be no way to know which run it came from. + tokio::fs::remove_file(workdir.path().join(META_JSON_FILENAME)).await?; + assert!( + fetch_atp( + "http://127.0.0.1:9/history.json", + &client, + &progress, + workdir.path(), + ) + .await + .is_err() + ); Ok(()) } diff --git a/src/pipeline/atp/mod.rs b/src/pipeline/atp/mod.rs index 611a5982..16a0cba4 100644 --- a/src/pipeline/atp/mod.rs +++ b/src/pipeline/atp/mod.rs @@ -7,7 +7,6 @@ use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; use memmap2::Mmap; use piz::ZipArchive; use rayon::prelude::*; -use reqwest::Client; use std::fs::{File, rename}; use std::io::{BufRead, BufReader, Read}; use std::path::{Path, PathBuf}; @@ -18,6 +17,7 @@ use time::{ format_description::well_known::{Iso8601, Rfc3339}, }; +use super::inputs::Input; use crate::places::{ParquetWriter, Place}; use crate::{PROGRESS_BAR_STYLE, matchers::MatchMask}; @@ -26,24 +26,50 @@ mod wikidata_ids; // Only these two re-exported crate-wide (rather than making all of // `fetch` pub(crate)): provenance needs them to assemble this -// pipeline's provenance BOM, nothing outside `atp` needs the rest of -// fetch's API (fetch_atp, ATP_RUN_HISTORY_URL, ...). +// pipeline's provenance BOM. pub(crate) use fetch::{AtpMetadata, read_cached_metadata}; +// For `pipeline::inputs`, which fetches every pipeline input up front; +// nothing else outside `atp` needs the rest of fetch's API. +pub(super) use fetch::{ATP_RUN_HISTORY_URL, fetch_atp}; + pub use wikidata_ids::collect_wikidata_ids; -pub async fn import_atp( - client: &Client, +/// Filename, within `workdir`, of the AllThePlaces dump `fetch_atp` +/// downloads. +const ATP_ZIP_FILENAME: &str = "alltheplaces.zip"; + +/// Filename, within `workdir`, of the file `import_atp` builds from +/// [`ATP_ZIP_FILENAME`]. +const ATP_PARQUET_FILENAME: &str = "alltheplaces.parquet"; + +/// Builds (or reuses) `alltheplaces.parquet` from the AllThePlaces +/// `input` that `pipeline::inputs::fetch_inputs` already made available +/// locally -- this step itself never touches the network. +/// +/// An existing `alltheplaces.parquet` is reused unless the zip is there +/// and newer than it (i.e. was fetched again after the parquet file was +/// built). With the zip gone, there's nothing to compare against, and +/// `fetch_atp` deliberately didn't download it again: the parquet file +/// is reused as is. +pub fn import_atp( + input: &Input, progress: &MultiProgress, workdir: &Path, ) -> Result { - let out = workdir.join("alltheplaces.parquet"); + let out = workdir.join(ATP_PARQUET_FILENAME); if out.exists() { - return Ok(out); + let stale = match std::fs::metadata(&input.path).and_then(|m| m.modified()) { + std::result::Result::Ok(zip_modified) => super::modified(&out)? < zip_modified, + Err(_) => false, + }; + if !stale { + return Ok(out); + } } - let (input_zip, atp_metadata) = - fetch::fetch_atp(fetch::ATP_RUN_HISTORY_URL, client, progress, workdir).await?; + let input_zip = &input.path; + let atp_metadata = &input.metadata; let atp_start_time = atp_metadata .start_time .format(&Rfc3339) @@ -70,7 +96,7 @@ pub async fn import_atp( let (tx, rx) = sync_channel(50_000); let (ra, rb) = std::thread::scope(|s| { let r1 = s.spawn(|| process_places(rx, progress, workdir, &out)); - let r2 = s.spawn(|| process_zip(&input_zip, progress, tx)); + let r2 = s.spawn(|| process_zip(input_zip, progress, tx)); (r1.join().unwrap(), r2.join().unwrap()) }); ra?; diff --git a/src/pipeline/datapackage.rs b/src/pipeline/datapackage.rs index b54080e1..a864af74 100644 --- a/src/pipeline/datapackage.rs +++ b/src/pipeline/datapackage.rs @@ -67,7 +67,7 @@ pub struct PublishedFile { } /// Builds the descriptor. `anchor` is the pipeline's input-anchored -/// timestamp (`provenance::read_anchor`): `version` is its calendar +/// timestamp (`provenance::anchor_timestamp`): `version` is its calendar /// date, `created` its full RFC 3339 form. pub fn build_datapackage( workdir: &std::path::Path, diff --git a/src/pipeline/inputs.rs b/src/pipeline/inputs.rs new file mode 100644 index 00000000..a49cbed7 --- /dev/null +++ b/src/pipeline/inputs.rs @@ -0,0 +1,242 @@ +//! The pipeline's first step, `fetch_inputs`: makes every external input +//! available locally in `workdir`, downloading whatever is missing, all +//! in parallel. Every later step works only on local files, so a run +//! that can't get its inputs fails within minutes instead of hours into +//! processing (see ). +//! +//! # When a cached input is reused +//! +//! Per input, nothing is downloaded if `workdir` already has the input's +//! metadata sidecar (`*.meta.json`), *and* either +//! +//! - the raw input file itself, or +//! - the artifact derived from it (`alltheplaces.parquet` for +//! AllThePlaces, `osm-features.index` for OpenStreetMap), so the raw +//! file was presumably deleted by hand to save disk space once it had +//! served its purpose. +//! +//! In the second case, the returned [`Input::path`] doesn't exist; the +//! downstream step reuses its derived artifact instead, and provenance +//! only needs the metadata. One exception: a planet `.pbf` without a +//! sidecar (e.g. a regional extract dropped in for testing) is accepted +//! too, with its metadata computed on the spot. See +//! [`super::atp::fetch_atp`] and [`super::osm::fetch_planet`] for each +//! input's exact rule. + +use anyhow::Result; +use indicatif::MultiProgress; +use std::path::{Path, PathBuf}; + +use super::{AtpMetadata, OsmMetadata, atp, osm}; + +/// One pipeline input, available locally once `fetch_inputs` is done. +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct Input { + /// Where the raw input lives in `workdir`. Might not exist if the + /// artifact derived from it is reused instead; see the module + /// documentation. + pub path: PathBuf, + + /// Provenance of the input, as persisted in its metadata sidecar. + pub metadata: M, +} + +/// Every external input of the pipeline. +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct Inputs { + pub atp: Input, + pub osm: Input, +} + +/// Where to fetch the pipeline's inputs from. Only ever not the +/// [`Default`] in tests. +pub(crate) struct Sources { + /// AllThePlaces' `history.json`, which points to the actual dump. + pub atp_history_url: String, + + /// The OpenStreetMap planet PBF. + pub osm_planet_url: String, +} + +impl Default for Sources { + fn default() -> Self { + Sources { + atp_history_url: atp::ATP_RUN_HISTORY_URL.to_string(), + osm_planet_url: osm::OSM_PLANET_URL.to_string(), + } + } +} + +/// Fetches every input in parallel, each logged as its own sub-step +/// (`fetch_inputs.atp`, `fetch_inputs.osm`). If any fetch fails, the +/// others are cancelled right away (a cancelled planet download leaves a +/// `.part` file that the next attempt resumes), and the error is +/// returned. See the module documentation for when a cached input is +/// reused instead of downloaded. +pub(crate) async fn fetch_inputs( + http_client: &reqwest::Client, + progress: &MultiProgress, + workdir: &Path, + sources: &Sources, +) -> Result { + let (atp, osm) = futures_util::future::try_join( + super::run_step_async( + "fetch_inputs.atp", + atp::fetch_atp(&sources.atp_history_url, http_client, progress, workdir), + ), + super::run_step_async( + "fetch_inputs.osm", + osm::fetch_planet(&sources.osm_planet_url, http_client, progress, workdir), + ), + ) + .await?; + Ok(Inputs { atp, osm }) +} + +#[cfg(test)] +mod tests { + use super::*; + use indicatif::ProgressDrawTarget; + use mockito::Server; + use std::os::unix::fs::symlink; + use std::time::Duration; + + fn fixture(name: &str) -> PathBuf { + let mut p = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + p.push("tests/test_data"); + p.push(name); + p + } + + fn test_client() -> reqwest::Client { + reqwest::Client::builder() + // Test servers run on 127.0.0.1; no proxy needed. + .no_proxy() + .build() + .expect("failed to build test client") + } + + fn hidden_progress() -> MultiProgress { + MultiProgress::with_draw_target(ProgressDrawTarget::hidden()) + } + + /// Sources nothing listens on: any attempt to download from them + /// fails, so a test passing with these shows nothing was downloaded. + fn unreachable_sources() -> Sources { + Sources { + atp_history_url: "http://127.0.0.1:9/history.json".to_string(), + osm_planet_url: "http://127.0.0.1:9/planet.osm.pbf".to_string(), + } + } + + #[tokio::test] + async fn test_complete_workdir_downloads_nothing() -> Result<()> { + let workdir = tempfile::tempdir()?; + for (src, dst) in [ + ("alltheplaces.zip", "alltheplaces.zip"), + ("alltheplaces.meta.json", "alltheplaces.meta.json"), + ("zugerland.osm.pbf", "planet-latest.osm.pbf"), + ( + "planet-latest.osm.pbf.meta.json", + "planet-latest.osm.pbf.meta.json", + ), + ] { + symlink(fixture(src), workdir.path().join(dst))?; + } + + let inputs = fetch_inputs( + &test_client(), + &hidden_progress(), + workdir.path(), + &unreachable_sources(), + ) + .await?; + + assert_eq!(inputs.atp.path, workdir.path().join("alltheplaces.zip")); + assert_eq!( + inputs.atp.metadata, + super::super::read_cached_atp_metadata(workdir.path())? + ); + assert_eq!( + inputs.osm.path, + workdir.path().join("planet-latest.osm.pbf") + ); + assert_eq!( + inputs.osm.metadata, + super::super::read_cached_metadata(workdir.path())? + ); + Ok(()) + } + + #[tokio::test] + async fn test_derived_artifacts_stand_in_for_deleted_inputs() -> Result<()> { + // Both raw inputs were deleted after their derived artifacts + // were built: their sidecars and those artifacts are enough. + let workdir = tempfile::tempdir()?; + for name in ["alltheplaces.meta.json", "planet-latest.osm.pbf.meta.json"] { + symlink(fixture(name), workdir.path().join(name))?; + } + for name in [ + "alltheplaces.parquet", + "osm-features.index", + "osm-features.index.inverted", + "osm-assemble.strings", + ] { + std::fs::write(workdir.path().join(name), b"")?; + } + + let inputs = fetch_inputs( + &test_client(), + &hidden_progress(), + workdir.path(), + &unreachable_sources(), + ) + .await?; + assert!(!inputs.atp.path.exists()); + assert!(!inputs.osm.path.exists()); + Ok(()) + } + + #[tokio::test] + async fn test_failing_fetch_cancels_the_others() -> Result<()> { + // AllThePlaces fails right away... + let mut server = Server::new_async().await; + let atp_history = server + .mock("GET", "/history.json") + .with_status(503) + .create_async() + .await; + + // ...while the planet server accepts connections but never + // answers. With no request timeout on the planet download (it + // takes hours), only cancellation can end that fetch. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let planet_addr = listener.local_addr()?; + let hang = tokio::spawn(async move { + let mut connections = Vec::new(); + while let std::result::Result::Ok((socket, _)) = listener.accept().await { + connections.push(socket); + } + }); + + let workdir = tempfile::tempdir()?; + let sources = Sources { + atp_history_url: format!("{}/history.json", server.url()), + osm_planet_url: format!("http://{planet_addr}/planet.osm.pbf"), + }; + let result = tokio::time::timeout( + Duration::from_secs(30), + fetch_inputs(&test_client(), &hidden_progress(), workdir.path(), &sources), + ) + .await + .expect("a failed fetch should cancel the others, not wait for them"); + hang.abort(); + + let err = result.expect_err("fetch_inputs should fail if any fetch fails"); + assert!(format!("{err:#}").contains("history.json"), "{err:#}"); + atp_history.assert_async().await; + assert!(!workdir.path().join("alltheplaces.zip").exists()); + assert!(!workdir.path().join("planet-latest.osm.pbf").exists()); + Ok(()) + } +} diff --git a/src/pipeline/mod.rs b/src/pipeline/mod.rs index 5f5e7d96..7eabac87 100644 --- a/src/pipeline/mod.rs +++ b/src/pipeline/mod.rs @@ -35,6 +35,7 @@ mod conflate; mod conflated_tiles; mod datapackage; mod edits; +mod inputs; mod logging; mod memstats; mod osm; @@ -195,11 +196,35 @@ fn run_pipeline_steps( progress: &indicatif::MultiProgress, ) -> Result<()> { crate::geometry::init_geospatial_stats()?; - let runtime = tokio::runtime::Builder::new_multi_thread() - .enable_all() - .build()?; + + // Every external input first, in parallel, so a run that can't get + // its inputs fails within minutes, before any processing; every + // later step only reads local files. See `inputs`. + let inputs = run_step("fetch_inputs", || { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()?; + let result = runtime.block_on(inputs::fetch_inputs( + http_client, + progress, + workdir, + &inputs::Sources::default(), + )); + // A fetch cancelled because another one failed might still have + // blocking work (hashing) running; don't wait for it. + runtime.shutdown_background(); + result + })?; + + // The input-anchored release timestamp (see `provenance`): its + // calendar date names every published `data/` object + // (`--.`), and it becomes the manifest's + // `version`. Depends only on the inputs' metadata. + let anchor = provenance::anchor_timestamp(&inputs.atp.metadata, &inputs.osm.metadata); + let date = anchor.date().to_string().replace('-', ""); + let atp = run_step("import_atp", || { - runtime.block_on(atp::import_atp(http_client, progress, workdir)) + atp::import_atp(&inputs.atp, progress, workdir) })?; // Not consumed by anything yet -- see atp::collect_wikidata_ids for // what this is for. @@ -216,20 +241,13 @@ fn run_pipeline_steps( // beyond) for no reason. let conflated = { let osm_features = run_step("import_osm", || { - osm::import_osm(http_client, progress, workdir) + osm::import_osm(&inputs.osm, progress, workdir) })?; run_step("conflate", || { conflate::conflate(&atp, &osm_features, progress, workdir, pipeline_run_id) })? }; - // The input-anchored release timestamp (see `provenance`): its - // calendar date names every published `data/` object - // (`--.`), and it becomes the manifest's - // `version`. Available now that both inputs' metadata is in `workdir`. - let anchor = provenance::read_anchor(workdir)?; - let date = anchor.date().to_string().replace('-', ""); - // Each published file's manifest entry, collected as it's uploaded; // `datapackage.json` is built from these and uploaded last. let mut published = Vec::new(); @@ -344,7 +362,9 @@ fn run_pipeline_steps( /// also calls this directly (via plain Rust module-tree visibility -- /// this function isn't `pub`, but `osm` is a descendant module of /// `pipeline`, so it can see it) for its own internal sub-steps -/// (`import_osm.fetch`, `.open`, `.prune`, `.assemble`, `.index`), +/// (`import_osm.open`, `.prune`, `.assemble`, `.index`), as does +/// `inputs::fetch_inputs` via [`run_step_async`] (`fetch_inputs.atp`, +/// `.osm`), /// dotted-named to stay visually distinct from top-level steps in /// `pipeline.log` while sharing the exact same logging shape -- one /// mechanism, not two, and no separate parsing convention needed for @@ -360,16 +380,65 @@ fn run_pipeline_steps( /// ever surface via `main()`'s default unwind on the way out of the /// process, never in `pipeline.log` itself. fn run_step(name: &str, step: impl FnOnce() -> Result) -> Result { - let start = Instant::now(); - log_snapshot(name, "start", None); + let guard = StepLog::start(name); let result = step(); - log_snapshot(name, "end", Some(start.elapsed().as_secs_f64())); - if let Err(e) = &result { - log::error!(step = name; "{name} failed: {e:#}"); - } + guard.finish(&result); result } +/// [`run_step`] for an async step, e.g. one of several input fetches +/// running concurrently (see `inputs::fetch_inputs`), with the same +/// logging. A step whose future gets dropped before completing -- i.e. +/// cancelled, because a sibling step failed -- still gets its "end" +/// snapshot, plus a warning saying it was cancelled. +async fn run_step_async( + name: &str, + step: impl std::future::Future>, +) -> Result { + let guard = StepLog::start(name); + let result = step.await; + guard.finish(&result); + result +} + +/// The start/end logging shared by [`run_step`] and [`run_step_async`]. +/// Dropped without [`StepLog::finish`] means the step was cancelled. +struct StepLog<'a> { + name: &'a str, + start: Instant, + finished: bool, +} + +impl<'a> StepLog<'a> { + fn start(name: &'a str) -> Self { + log_snapshot(name, "start", None); + StepLog { + name, + start: Instant::now(), + finished: false, + } + } + + fn finish(mut self, result: &Result) { + self.finished = true; + let name = self.name; + log_snapshot(name, "end", Some(self.start.elapsed().as_secs_f64())); + if let Err(e) = result { + log::error!(step = name; "{name} failed: {e:#}"); + } + } +} + +impl Drop for StepLog<'_> { + fn drop(&mut self) { + if !self.finished { + let name = self.name; + log_snapshot(name, "end", Some(self.start.elapsed().as_secs_f64())); + log::warn!(step = name; "{name} cancelled"); + } + } +} + /// Above this fraction of the cgroup memory limit, [`log_snapshot`] logs /// a warning in addition to its usual info-level snapshot -- an early /// signal ahead of an actual OOM-kill, which the kernel triggers once diff --git a/src/pipeline/osm/fetch.rs b/src/pipeline/osm/fetch.rs index 3fa36d7f..3e3a7689 100644 --- a/src/pipeline/osm/fetch.rs +++ b/src/pipeline/osm/fetch.rs @@ -1,7 +1,11 @@ use crate::make_download_bar; +use crate::pipeline::inputs::Input; +use crate::utils::to_hex; use anyhow::{Context, Ok, Result, anyhow}; +use aws_lc_rs::digest::{Context as DigestContext, SHA256}; use futures_util::StreamExt; use indicatif::MultiProgress; +use std::io::Read; use std::path::{Path, PathBuf}; use std::time::Duration; use tokio::io::AsyncWriteExt; @@ -13,7 +17,8 @@ use super::OsmMetadata; /// bucket (`osm-planet-eu-central-1.s3.dualstack.eu-central-1.amazonaws.com`, /// confirmed by hand with `curl -sIL`) -- `reqwest` follows the /// redirect transparently. -const OSM_PLANET_URL: &str = "https://planet.openstreetmap.org/pbf/planet-latest.osm.pbf"; +pub(crate) const OSM_PLANET_URL: &str = + "https://planet.openstreetmap.org/pbf/planet-latest.osm.pbf"; /// How long a single chunk read is allowed to sit with no data arriving /// before this gives up and lets the caller retry (via `fetch_planet`'s @@ -23,37 +28,79 @@ const OSM_PLANET_URL: &str = "https://planet.openstreetmap.org/pbf/planet-latest /// guarantee failure for a file this size). const STALL_TIMEOUT: Duration = Duration::from_secs(60); -pub fn fetch_planet( +/// Makes the OpenStreetMap planet available as a local input in +/// `workdir`, downloading it from `url` only if nothing in `workdir` can +/// stand in for it. In order: +/// +/// 1. The `.pbf` exists, with its `.meta.json` sidecar: reused as is. +/// 2. The `.pbf` exists without a sidecar -- e.g. a regional extract +/// dropped in for cloud testing (`scripts/test-on-hetzner`), rather +/// than a file this function downloaded itself: its metadata is +/// computed (header + full SHA-256) and persisted, no download. +/// 3. The `.pbf` is gone, but its sidecar and the index built from it +/// (`osm-features.index`, see `import_osm`) are both still there -- +/// e.g. the PBF was deleted by hand to free its ~90GB once the index +/// was built: only the metadata is returned, and the returned path +/// does not exist. `import_osm` then reuses the index. +/// 4. Otherwise: downloaded (resuming a `.part` file left over from an +/// interrupted attempt, if any), hashed while streaming. +/// +/// Async, with all blocking file work in `spawn_blocking`, so it can run +/// concurrently with the other input fetches (see `pipeline::inputs`). +pub async fn fetch_planet( + url: &str, http_client: &reqwest::Client, progress: &MultiProgress, workdir: &Path, -) -> Result<(PathBuf, OsmMetadata)> { +) -> Result> { let pbf_path: PathBuf = workdir.join(super::PLANET_PBF_FILENAME); - if pbf_path.exists() { - if let Result::Ok(metadata) = super::read_cached_metadata(workdir) { - return Ok((pbf_path, metadata)); - } - // The .pbf exists, but its .meta.json sidecar is missing or - // unreadable -- e.g. a regional extract dropped in for cloud - // testing (scripts/test-on-hetzner), rather than a file this - // function downloaded itself. Compute the metadata now, the - // same way a fresh download does below, instead of failing or - // re-downloading a file that's already there. - let metadata = super::compute_and_persist_metadata(&pbf_path, workdir, progress)?; - return Ok((pbf_path, metadata)); - } - tokio::runtime::Builder::new_multi_thread() - .enable_all() - .build()? - .block_on(download_osm_planet(http_client, progress, &pbf_path))?; + let cached = { + let (pbf_path, workdir, progress) = + (pbf_path.clone(), workdir.to_path_buf(), progress.clone()); + tokio::task::spawn_blocking(move || cached_planet(&pbf_path, &workdir, &progress)).await?? + }; + if let Some(metadata) = cached { + return Ok(Input { + path: pbf_path, + metadata, + }); + } - // Not async: a plain buffered read/hash of the (potentially tens of - // GB) downloaded file, outside the tokio runtime used for the - // download above. - let metadata = super::compute_and_persist_metadata(&pbf_path, workdir, progress)?; + let sha256 = download_osm_planet(url, http_client, progress, &pbf_path).await?; + let metadata = { + let (pbf_path, workdir) = (pbf_path.clone(), workdir.to_path_buf()); + tokio::task::spawn_blocking(move || super::persist_metadata(&pbf_path, &workdir, sha256)) + .await?? + }; + Ok(Input { + path: pbf_path, + metadata, + }) +} - Ok((pbf_path, metadata)) +/// Cases 1-3 of [`fetch_planet`]'s doc comment: the planet's metadata, if +/// `workdir` already has what it takes to skip the download; `None` if +/// it has to be downloaded. Blocking. +fn cached_planet( + pbf_path: &Path, + workdir: &Path, + progress: &MultiProgress, +) -> Result> { + let cached_metadata = super::read_cached_metadata(workdir); + if pbf_path.exists() { + if let Result::Ok(metadata) = cached_metadata { + return Ok(Some(metadata)); + } + let metadata = super::compute_and_persist_metadata(pbf_path, workdir, progress)?; + return Ok(Some(metadata)); + } + if let Result::Ok(metadata) = cached_metadata + && super::index_exists(workdir) + { + return Ok(Some(metadata)); + } + Ok(None) } /// Downloads the planet file over plain HTTPS instead of BitTorrent. @@ -72,45 +119,58 @@ pub fn fetch_planet( /// picks up from wherever that file left off via an HTTP `Range` /// request if one already exists on disk (e.g. after a restarted /// pipeline run) -- S3 advertises `Accept-Ranges: bytes` for this file. +/// Equally, dropping this future mid-download (e.g. because another +/// input's fetch failed, see `pipeline::inputs`) just leaves a `.part` +/// file for the next attempt to resume. +/// +/// Returns the SHA-256 of the whole file, as lowercase hex, computed +/// while streaming rather than in a second read pass over ~90GB; on a +/// resumed download, the already-downloaded prefix is hashed first. async fn download_osm_planet( + url: &str, http_client: &reqwest::Client, progress: &MultiProgress, pbf_path: &Path, -) -> Result<()> { +) -> Result { let part_path = pbf_path.with_extension("pbf.part"); let resume_from = tokio::fs::metadata(&part_path) .await .map(|m| m.len()) .unwrap_or(0); - let mut request = http_client.get(OSM_PLANET_URL); + let mut request = http_client.get(url); if resume_from > 0 { request = request.header(reqwest::header::RANGE, format!("bytes={resume_from}-")); } let response = request .send() .await - .with_context(|| format!("Failed to GET {OSM_PLANET_URL}"))? + .with_context(|| format!("Failed to GET {url}"))? .error_for_status() - .with_context(|| format!("Server returned error for {OSM_PLANET_URL}"))?; + .with_context(|| format!("Server returned error for {url}"))?; // A server that doesn't support (or ignores) the Range request // sends the whole file back from byte 0 instead of a 206 -- the // status code is the one reliable way to tell "here's the rest" // from "here's everything, again" apart. - let (mut file, already_have) = + let (mut file, already_have, mut hasher) = if resume_from > 0 && response.status() == reqwest::StatusCode::PARTIAL_CONTENT { + let hasher = { + let (part_path, progress) = (part_path.clone(), progress.clone()); + tokio::task::spawn_blocking(move || hash_prefix(&part_path, resume_from, &progress)) + .await?? + }; let file = tokio::fs::OpenOptions::new() .append(true) .open(&part_path) .await .with_context(|| format!("Failed to open {}", part_path.display()))?; - (file, resume_from) + (file, resume_from, hasher) } else { let file = tokio::fs::File::create(&part_path) .await .with_context(|| format!("Failed to create file {}", part_path.display()))?; - (file, 0) + (file, 0, DigestContext::new(&SHA256)) }; let total = response.content_length().map(|len| len + already_have); @@ -126,6 +186,7 @@ async fn download_osm_planet( break; }; let chunk = chunk.context("Error reading chunk from response stream")?; + hasher.update(&chunk); file.write_all(&chunk) .await .context("Failed to write chunk to disk")?; @@ -146,13 +207,46 @@ async fn download_osm_planet( ) })?; - Ok(()) + Ok(to_hex(hasher.finish().as_ref())) +} + +/// SHA-256 state after hashing the first `len` bytes of `path` -- the +/// already-downloaded prefix of a resumed download, so +/// [`download_osm_planet`] can carry on hashing the rest as it streams +/// in. Blocking. +fn hash_prefix(path: &Path, len: u64, progress: &MultiProgress) -> Result { + let file = std::fs::File::open(path).with_context(|| format!("could not open `{path:?}`"))?; + let mut reader = file.take(len); + let bar = make_download_bar(progress, "osm.hash ", Some(len)); + let mut hasher = DigestContext::new(&SHA256); + let mut buf = vec![0u8; 8 * 1024 * 1024]; // 8 MiB + let mut total: u64 = 0; + loop { + let n = reader + .read(&mut buf) + .with_context(|| format!("could not read `{path:?}`"))?; + if n == 0 { + break; + } + hasher.update(&buf[..n]); + total += n as u64; + bar.inc(n as u64); + } + bar.finish(); + if total != len { + return Err(anyhow!( + "`{path:?}` shrank while resuming its download: expected {len} bytes, read {total}" + )); + } + Ok(hasher) } #[cfg(test)] mod tests { use super::*; use crate::pipeline::osm::{PLANET_PBF_FILENAME, read_cached_metadata}; + use indicatif::ProgressDrawTarget; + use mockito::Server; fn test_data_path(filename: &str) -> PathBuf { let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR")); @@ -163,11 +257,24 @@ mod tests { } fn test_client() -> reqwest::Client { - reqwest::Client::new() + reqwest::Client::builder() + // Mockito runs on 127.0.0.1; no proxy needed. + .no_proxy() + .build() + .expect("failed to build test client") + } + + fn hidden_progress() -> MultiProgress { + MultiProgress::with_draw_target(ProgressDrawTarget::hidden()) } - #[test] - fn test_fetch_planet_computes_metadata_for_preexisting_pbf_without_sidecar() -> Result<()> { + /// A URL nothing listens on: any attempt to download from it fails, + /// so a test passing with it shows nothing was downloaded. + const UNREACHABLE_URL: &str = "http://127.0.0.1:9/planet.osm.pbf"; + + #[tokio::test] + async fn test_fetch_planet_computes_metadata_for_preexisting_pbf_without_sidecar() -> Result<()> + { // A .pbf dropped into the workdir without its .meta.json sidecar // -- e.g. a regional extract fetched for cloud testing, rather // than a file this function downloaded itself -- should have its @@ -178,11 +285,16 @@ mod tests { std::fs::copy(test_data_path("zugerland.osm.pbf"), &pbf_path)?; assert!(read_cached_metadata(workdir.path()).is_err()); - let progress = MultiProgress::new(); - let client = test_client(); - let (returned_path, metadata) = fetch_planet(&client, &progress, workdir.path())?; + let input = fetch_planet( + UNREACHABLE_URL, + &test_client(), + &hidden_progress(), + workdir.path(), + ) + .await?; - assert_eq!(returned_path, pbf_path); + assert_eq!(input.path, pbf_path); + let metadata = input.metadata; // Must agree with what mod.rs's own test_read_header reports for // the same fixture. assert_eq!( @@ -198,4 +310,100 @@ mod tests { Ok(()) } + + #[tokio::test] + async fn test_fetch_planet_reuses_index_when_pbf_was_deleted() -> Result<()> { + // The PBF was deleted by hand after the index was built from it: + // its sidecar and the index stand in for it, no download. + let workdir = tempfile::tempdir()?; + std::fs::copy( + test_data_path("planet-latest.osm.pbf.meta.json"), + workdir + .path() + .join(format!("{PLANET_PBF_FILENAME}.meta.json")), + )?; + for name in [ + "osm-features.index", + "osm-features.index.inverted", + "osm-assemble.strings", + ] { + std::fs::write(workdir.path().join(name), b"")?; + } + + let input = fetch_planet( + UNREACHABLE_URL, + &test_client(), + &hidden_progress(), + workdir.path(), + ) + .await?; + assert_eq!(input.metadata, read_cached_metadata(workdir.path())?); + assert!(!input.path.exists()); + Ok(()) + } + + #[tokio::test] + async fn test_fetch_planet_downloads_without_index() -> Result<()> { + // A sidecar alone, with neither the PBF nor the index, isn't + // enough to skip the download: import_osm would have nothing to + // work from. + let fixture = std::fs::read(test_data_path("zugerland.osm.pbf"))?; + let mut server = Server::new_async().await; + let mock = server + .mock("GET", "/planet.osm.pbf") + .with_status(200) + .with_body(&fixture) + .create_async() + .await; + let workdir = tempfile::tempdir()?; + std::fs::copy( + test_data_path("planet-latest.osm.pbf.meta.json"), + workdir + .path() + .join(format!("{PLANET_PBF_FILENAME}.meta.json")), + )?; + + let url = format!("{}/planet.osm.pbf", server.url()); + let input = fetch_planet(&url, &test_client(), &hidden_progress(), workdir.path()).await?; + mock.assert_async().await; + + assert_eq!(std::fs::read(&input.path)?, fixture); + let expected_sha256 = crate::utils::hash_file(&input.path, &hidden_progress(), "", false)?; + assert_eq!(input.metadata.sha256, Some(expected_sha256.sha256)); + assert_eq!(input.metadata.writing_program, Some("osmx".to_owned())); + assert_eq!(read_cached_metadata(workdir.path())?, input.metadata); + assert!(!workdir.path().join("planet-latest.osm.pbf.part").exists()); + Ok(()) + } + + #[tokio::test] + async fn test_fetch_planet_resumes_and_hashes_whole_file() -> Result<()> { + // A `.part` file left over from an interrupted download is + // resumed with a Range request; the hash must still cover the + // whole file, not just the bytes fetched this time. + let fixture = std::fs::read(test_data_path("zugerland.osm.pbf"))?; + let split = fixture.len() / 3; + let mut server = Server::new_async().await; + let mock = server + .mock("GET", "/planet.osm.pbf") + .match_header("range", format!("bytes={split}-").as_str()) + .with_status(206) + .with_body(&fixture[split..]) + .create_async() + .await; + let workdir = tempfile::tempdir()?; + std::fs::write( + workdir.path().join("planet-latest.osm.pbf.part"), + &fixture[..split], + )?; + + let url = format!("{}/planet.osm.pbf", server.url()); + let input = fetch_planet(&url, &test_client(), &hidden_progress(), workdir.path()).await?; + mock.assert_async().await; + + assert_eq!(std::fs::read(&input.path)?, fixture); + let expected_sha256 = crate::utils::hash_file(&input.path, &hidden_progress(), "", false)?; + assert_eq!(input.metadata.sha256, Some(expected_sha256.sha256)); + Ok(()) + } } diff --git a/src/pipeline/osm/mod.rs b/src/pipeline/osm/mod.rs index 2a268b23..dd7ea800 100644 --- a/src/pipeline/osm/mod.rs +++ b/src/pipeline/osm/mod.rs @@ -12,6 +12,7 @@ use std::time::SystemTime; use time::UtcDateTime; use time::format_description::well_known::Rfc3339; +use super::inputs::Input; use crate::tables::OsmFeatures; mod assemble; @@ -22,6 +23,9 @@ mod prune; use prune::Prunings; +// For `pipeline::inputs`, which fetches every pipeline input up front. +pub(super) use fetch::{OSM_PLANET_URL, fetch_planet}; + /// Encodes an OpenStreetMap element's `(type, id)` as a `Feature.id` -- /// `id * 10 + 1` for nodes, `+ 2` for ways, `+ 3` for relations. The same /// encoding is also used for relation member references @@ -60,30 +64,31 @@ pub(crate) fn osm_type_str(member_type: RelationMemberType) -> &'static str { } } +/// Builds (or reuses) `osm-features.index` from the planet `input` +/// that `pipeline::inputs::fetch_inputs` already made available +/// locally -- this step itself never touches the network. pub fn import_osm<'a>( - http_client: &reqwest::Client, + input: &Input, progress: &MultiProgress, workdir: &Path, ) -> Result> { assert!(workdir.exists()); - let osm_index_path = workdir.join("osm-features.index"); + let osm_index_path = index_path(workdir); let strings_path = assemble::strings_path(workdir); if OsmFeatures::exists(&osm_index_path, &strings_path) { let features = OsmFeatures::open(&osm_index_path, &strings_path)?; // Compare against the planet PBF's own mtime, if it's still - // around -- a stat, not a fetch (see `fetch::fetch_planet`), so - // this doesn't cost the download this shortcut exists to avoid. - // The PBF can legitimately be gone (e.g. deleted by hand to free - // its ~90GB once this index was built): with nothing to compare - // against, there's no way to detect staleness, so this falls - // back to the previous existence-only behavior rather than - // erroring out. - let stale = - match std::fs::metadata(workdir.join(PLANET_PBF_FILENAME)).and_then(|m| m.modified()) { - std::result::Result::Ok(pbf_modified) => features.modified()? < pbf_modified, - Err(_) => false, - }; + // around. The PBF can legitimately be gone (e.g. deleted by hand + // to free its ~90GB once this index was built -- in which case + // `fetch::fetch_planet` deliberately didn't download it again): + // with nothing to compare against, there's no way to detect + // staleness, so this falls back to existence-only behavior + // rather than erroring out. + let stale = match std::fs::metadata(&input.path).and_then(|m| m.modified()) { + std::result::Result::Ok(pbf_modified) => features.modified()? < pbf_modified, + Err(_) => false, + }; if !stale { return Ok(features); } @@ -93,25 +98,21 @@ pub fn import_osm<'a>( // memstats logging shape as this crate's top-level pipeline steps // (see `super::run_step`, which this reuses directly rather than a // second, parallel logging mechanism) -- just under a dotted name - // ("import_osm.fetch", not "fetch") so a log consumer can tell a + // ("import_osm.prune", not "prune") so a log consumer can tell a // sub-step from a top-level one at a glance, and so `import_osm`'s // own already-existing top-level start/end pair (logged by // whichever `run_step` call wraps this whole function, see // `pipeline::run_pipeline_steps`) keeps meaning "the whole step", - // not "the whole step minus its sub-steps". Before this, only two - // points inside `import_osm` were distinguishable at all -- - // "opened OpenStreetMap planet file" (after fetch+hash+blob-scan) - // and the outer step's own end -- which lumps prune/assemble/ - // index-build into one undifferentiated number; not enough - // resolution to tell which of those actually needs the memory a - // tight `--mem-limit` run runs short on (see #711's investigation, - // e.g. brawer/osmdiffs#711's comments for a real case where - // that distinction mattered). - let (pbf, fetch_metadata) = super::run_step("import_osm.fetch", || { - fetch::fetch_planet(http_client, progress, workdir) - })?; + // not "the whole step minus its sub-steps". Without these, prune/ + // assemble/index-build would be lumped into one undifferentiated + // number; not enough resolution to tell which of those actually + // needs the memory a tight `--mem-limit` run runs short on (see + // brawer/osmdiffs#711's comments for a real case where that + // distinction mattered). Downloading the planet used to be a sub-step + // here too (`import_osm.fetch`); it's now `fetch_inputs.osm`. + let pbf = &input.path; let pbf_error = || format!("could not open file `{:?}`", pbf); - let mut file = File::open(&pbf).with_context(pbf_error)?; + let mut file = File::open(pbf).with_context(pbf_error)?; let mut reader = super::run_step("import_osm.open", || { BlobReader::open(&mut file).with_context(pbf_error) })?; @@ -124,7 +125,7 @@ pub fn import_osm<'a>( replication_timestamp = replication_timestamp.as_str(), source = header.source.as_deref(), writing_program = header.writing_program.as_deref(), - sha256 = fetch_metadata.sha256.as_deref(); + sha256 = input.metadata.sha256.as_deref(); "opened OpenStreetMap planet file" ); @@ -143,6 +144,18 @@ pub fn import_osm<'a>( }) } +/// Path, within `workdir`, of the spatial index `import_osm` builds. +fn index_path(workdir: &Path) -> PathBuf { + workdir.join("osm-features.index") +} + +/// Whether `import_osm`'s output -- the index plus its string pool -- +/// exists in `workdir`. Used by `fetch::fetch_planet` to decide whether +/// a deleted planet PBF needs downloading again. +fn index_exists(workdir: &Path) -> bool { + OsmFeatures::exists(&index_path(workdir), &assemble::strings_path(workdir)) +} + /// Filename, within `workdir`, that `fetch::fetch_planet` downloads the /// OSM planet PBF to. Shared with `pipeline::provenance`, which needs to /// find it again (via `read_header`) without re-fetching. Matches @@ -218,8 +231,9 @@ pub(crate) fn read_cached_metadata(workdir: &Path) -> Result { serde_json::from_str(&data).with_context(|| format!("Failed to parse {}", path.display())) } -/// Computes a fully-populated [`OsmMetadata`] for a freshly downloaded -/// planet file at `pbf_path` -- its header (cheap, via [`read_header`]) +/// Computes a fully-populated [`OsmMetadata`] for a planet file at +/// `pbf_path` that has no sidecar yet (e.g. a regional extract dropped +/// into `workdir` by hand, see `fetch::fetch_planet`) -- its header (cheap, via [`read_header`]) /// and the SHA-256 of its entire contents (not cheap: one dedicated /// sequential read pass over the whole file, see /// [`crate::utils::hash_file`]) -- and persists it to `workdir`, so a @@ -230,8 +244,16 @@ pub(crate) fn compute_and_persist_metadata( workdir: &Path, progress: &MultiProgress, ) -> Result { - let header = read_header(pbf_path)?; let sha256 = crate::utils::hash_file(pbf_path, progress, "osm.hash ", false)?.sha256; + persist_metadata(pbf_path, workdir, sha256) +} + +/// Builds the [`OsmMetadata`] for the planet file at `pbf_path` from its +/// header and an already-computed `sha256` of its entire contents (e.g. +/// hashed while downloading), and persists it to `workdir` the same way +/// [`compute_and_persist_metadata`] does. +fn persist_metadata(pbf_path: &Path, workdir: &Path, sha256: String) -> Result { + let header = read_header(pbf_path)?; let metadata = OsmMetadata { sha256: Some(sha256), ..header diff --git a/src/pipeline/provenance.rs b/src/pipeline/provenance.rs index 4e160ef4..559e85c5 100644 --- a/src/pipeline/provenance.rs +++ b/src/pipeline/provenance.rs @@ -40,18 +40,6 @@ pub(crate) struct OutputFileRef<'a> { pub distribution_url: &'a str, } -/// The input-anchored timestamp every reproducible artifact keys off: -/// `max(AllThePlaces run start, OSM planet replication)`, read back from -/// the metadata `import_atp`/`import_osm` left in `workdir`. See -/// [`build_bom_for_conflated_parquet`]. -pub(crate) fn read_anchor(workdir: &Path) -> Result { - let atp = pipeline::read_cached_atp_metadata(workdir) - .context("could not read AllThePlaces provenance")?; - let osm = pipeline::read_cached_metadata(workdir) - .context("could not read OpenStreetMap provenance")?; - Ok(anchor_timestamp(&atp, &osm)) -} - /// Standard OSM attribution notice, distinct from the license itself: /// ODbL requires reproducing this in any produced/derivative work, and /// that requirement propagates to `conflated.parquet` the same way the @@ -195,11 +183,14 @@ pub fn build_bom_for_conflated_parquet( })) } -/// The BOM's single time anchor: the freshness of whichever input was -/// updated most recently. Used for every timestamp in the document -/// instead of the wall clock, so the BOM is reproducible from the cached -/// inputs alone (see [`build_bom_for_conflated_parquet`]). -fn anchor_timestamp(atp: &AtpMetadata, osm: &OsmMetadata) -> UtcDateTime { +/// The input-anchored timestamp every reproducible artifact keys off: +/// `max(AllThePlaces run start, OSM planet replication)`, i.e. the +/// freshness of whichever input was updated most recently. The BOM uses +/// it for every timestamp in the document instead of the wall clock, so +/// the BOM is reproducible from the cached inputs alone (see +/// [`build_bom_for_conflated_parquet`]); `pipeline::run_pipeline_steps` +/// uses its date to name every published file. +pub(crate) fn anchor_timestamp(atp: &AtpMetadata, osm: &OsmMetadata) -> UtcDateTime { atp.start_time.max(osm.replication_timestamp) }