refactor(pipeline): Fetch all inputs up front, in parallel

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
 

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.

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<br/>(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<br/>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.

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

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,29 +86,57 @@ pub struct AtpMetadata {
     pub sha256: Option<String>,
 }
 
+/// 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<Input<AtpMetadata>> {
+    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 {}",
                 out_path.display(),
                 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(())
     }
 

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<AtpMetadata>,
     progress: &MultiProgress,
     workdir: &Path,
 ) -> Result<PathBuf> {
-    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?;

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,