diff --git a/Cargo.lock b/Cargo.lock index d3f85316e..5f525a9d1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1786,7 +1786,9 @@ dependencies = [ "causal-edge", "cpic", "deno_core", + "futures", "futures-core", + "hex", "include_dir", "lance-graph", "lance-graph-contract", @@ -1795,12 +1797,14 @@ dependencies = [ "memmap2", "ndarray 0.17.2", "notebook-query", + "object_store", "osm-soa-bake", "quarto-core", "quarto-system-runtime", "serde", "serde_json", "serde_v8", + "sha2 0.10.9", "tokio", "tower-http 0.5.2", "tracing", diff --git a/claude-notes/plans/2026-08-08-osm-soa-cockpit-wiring.md b/claude-notes/plans/2026-08-08-osm-soa-cockpit-wiring.md index 209957d72..0537ab919 100644 --- a/claude-notes/plans/2026-08-08-osm-soa-cockpit-wiring.md +++ b/claude-notes/plans/2026-08-08-osm-soa-cockpit-wiring.md @@ -721,10 +721,13 @@ and Berlin's densest hip tile holds 3,844, so a z13 tile is bounded above by Berlin-class bake. (The bound is itself slack: it assumes 64 adjacent maximum-density tiles, where the one measured z14 tile holds 15,016.) -- [ ] Build the hip-cell representative form for overview zooms and compare it +- [x] Build the hip-cell representative form for overview zooms and compare it against the stride at equal budget — coverage, and what a user actually loses. That comparison is what promotes `OVERVIEW_ROW_BUDGET` from CONJECTURE, and it is the "dynamic compression bucket threshold" thread. + **Done in Phase 5** — generalised to a per-tile prefix depth rather than + fixed-at-hip (hip is the `zz=16` case); stride keeps 14/316 isolated + features at z8 where the cell form keeps 316/316. - [ ] Re-measure `CITY_ZOOM_FLOOR` against a denser extract than Berlin before treating 13 as anything but a policy floor. @@ -812,8 +815,10 @@ tiles. The old 5,000 cap masked it. It is survivable at the load the evidence supports (measured working below), so it is recorded rather than fixed in this pass. -- [ ] `render()` is O(tiles x markers) — rebuild incrementally, or diff, rather - than clearing `#tiles` on every tile arrival. +- [x] `render()` is O(tiles x markers) — rebuild incrementally, or diff, rather + than clearing `#tiles` on every tile arrival. **Done in the same phase** + — `paintFeatures()` + `drawnCells`, measured 1,550,957 -> 64,707 appends + (23.97x -> 1.00x), wall-clock 1.72x, behaviour identical. - [ ] The viewport bound is the honest one; `OVERVIEW_ROW_BUDGET` is its per-tile share at one window size. A budget derived from the actual tiles-in-view count would not drift with window size. @@ -913,3 +918,90 @@ in `osm_features.rs` is grounded in — so the real envelope is at least 1.6x what those constants assume. Not acted on: this was at city zoom with different fetch timing, and a budget should not be widened on one incidental observation. It does mean `OVERVIEW_ROW_BUDGET = 3_000` is conservative rather than tight. + +## Phase 6 — S3 → volume01 hydration (2026-08-11) + +**Operator: "railway has exactly the same S3 variables like you."** That single +fact retired the blocker. Phase 4 was recorded as *"verifiable only on deploy"* +because `/volume01` does not exist in this sandbox — but the volume is a +**path**, and the part that actually needed proving (do these credentials reach +that bucket, and does a 1.29 GiB round trip work) is verifiable right here. +`OSM_SLAB_CACHE_DIR` substitutes for the mount, so everything except the +literal mount point is exercised locally. + +**Operator, on shape:** *"S3 → lancedb → volume01"*, *"persist the lance file +to volume01"*, *"it's just a convenience to persist lance across rebuilds."* +The third message is the one that fixes the semantics: **the volume is a cache, +S3 is the source of truth.** Deleting the volume costs a re-download and +nothing else. + +### Three assumptions I had written down, all wrong + +The Phase 4 sketch claimed this needed ~5 new dependencies (`reqwest` + rustls, +`sha2`, `hex`, `chrono`, `tokio` sync), hand-rolled SigV4, and a sync→async +conversion of `open_slab()`. Checked rather than repeated: + +| claimed | measured | +|---|---| +| needs `reqwest` + a new HTTP stack | **`object_store` v0.13.2 is already in the graph** via `datafusion`/`lance`; adding it direct pulls no new transitive tree | +| needs hand-rolled SigV4 | **`AmazonS3Builder::from_env()`** reads it all — and `aws_endpoint_url` is an accepted alias for the endpoint key (`builder.rs:498`), so a non-AWS endpoint needs no special casing | +| `open_slab()` must become async | **`main()` is already `async`** — hydrating there, before the listener binds, leaves the read path untouched and stops the first request paying for the download | + +Net new direct deps: `object_store` (already in lockfile at this exact +version), `sha2`, `hex`, `futures` — all already resolved. + +### The artifact question, and why it did not block + +`openstreetmap-website-rs/Cargo.toml` says in its own header: *"no lance, no +datafusion: the ABI row is a byte layout, not an engine."* The bake emits raw +`.soa` + `.books` and has no Lance writer — and `lance_abi` was **removed** +from that repo (`a27b06a`) for doing a full-table materialising read. Lance +lives on the **consumer** side, where `cockpit-server` already has the stack. + +So the hydration layer is written **format-agnostic**: it moves named objects +from a prefix onto the volume and verifies them. Whether the artifact travels +as `.soa` or as a Lance dataset is a separate decision it does not prejudge. +That is also why this could ship without resolving it. + +Worth carrying from the `lance_abi` removal, because it names the correct +shape if the Lance path is taken later: *"this crate already has fragment.rs +(64k-row fragments, so a tile prefix lands in a fragment) and +`RowSlab::tile_range`. The module used neither."* `object_store` exposes +`get_range(location, Range)` — the range read that removal asked for. + +### What shipped + +`crates/cockpit-server/src/osm_slab_hydrate.rs`, called once from `main()`: + +1. `OSM_SLAB_PATH` set and a real file ⇒ use it. Local dev never touches S3. +2. Else `AWS_S3_BUCKET_NAME` + (`OSM_SLAB_CACHE_DIR` | `RAILWAY_VOL`) ⇒ + for each of `berlin.soa`, `berlin.books`: cache hit **verified by + checksum**, else stream from S3 → hash while writing → rename into place. +3. Else `None` ⇒ the endpoint answers 503 exactly as before. Absent + configuration is not an error. + +**Checksum pinning, on the cache hit too.** `SHA256SUMS` is fetched from the +same prefix; a missing entry is a refusal, not a warning. Verifying on a hit +(not only after download) guards the half-written file left by a container +killed mid-download — precisely the case a "we already have it" check waves +through. Downloads land on `.part` and are renamed only after the hash +matches, so a kill leaves nothing that looks complete; the hit-side check +stays anyway because the volume outlives this code. + +### Bucket conventions — followed, not invented + +The bucket already had 50+ objects, all MedCare-rs, with a clear shape: +`/bakes//` plus `SHA256SUMS`, and Lance datasets +shipped as `.lance.tar` (`all-lanes.lance.tar`, 378 MB). q2's upload follows +it exactly at `q2/bakes/berlin-v1/` and touches nothing under `MedCare-rs/`. + +| key | bytes | +|---|---| +| `q2/bakes/berlin-v1/berlin.soa` | 1,292,826,624 | +| `q2/bakes/berlin-v1/berlin.books` | 58,763,338 | +| `q2/bakes/berlin-v1/SHA256SUMS` | 156 | + +`sha256(berlin.soa) = cbf5989ab45bc921d8a85fdbdb71c8e5029cd904a3d230a898c2b5eb81d7ebe7` + +- [x] `OSM_SLAB_PATH` gains an S3 sibling: hydrate → volume, then mmap. +- [x] Checksum-pin the object; no unverified-fetch path. diff --git a/crates/cockpit-server/Cargo.toml b/crates/cockpit-server/Cargo.toml index b3500d87d..c8dd47238 100644 --- a/crates/cockpit-server/Cargo.toml +++ b/crates/cockpit-server/Cargo.toml @@ -70,6 +70,15 @@ osm-soa-bake = { path = "../../../openstreetmap-website-rs" } # memory). Pinned to the same 0.5 line osm-soa-bake itself uses, so the # dependency tree carries one memmap2 version, not two. memmap2 = "0.5" +# S3 -> volume hydration for the OSM slab (`osm_slab_hydrate`). `object_store` +# is already in the graph via lance/datafusion at this exact version, so this +# adds no new transitive tree — and its `AmazonS3Builder::from_env()` reads the +# AWS_* vars directly, so there is no hand-rolled SigV4 here. +object_store = { version = "0.13.2", features = ["aws"] } +sha2 = "0.10" +hex = "0.4" +futures = "0.3" + # ── Quarto rendering pipeline ─────────────────────────────────────── # pampa parses .qmd, quarto-core renders, deno_core executes JS/TS cells diff --git a/crates/cockpit-server/src/main.rs b/crates/cockpit-server/src/main.rs index 788ee630f..bf3b2c818 100644 --- a/crates/cockpit-server/src/main.rs +++ b/crates/cockpit-server/src/main.rs @@ -43,6 +43,7 @@ mod osint_classview; mod osm_tiles; mod osm; mod osm_features; +mod osm_slab_hydrate; // ── Embed the Vite build at compile time ───────────────────────────────────── // The cockpit/ directory is built by `cd cockpit && npm run build` which @@ -165,9 +166,36 @@ struct McpError { #[tokio::main] async fn main() { tracing_subscriber::fmt() - .with_env_filter("cockpit_server=info,tower_http=info") + // The binary target is `q2-cockpit`, so `module_path!()` on this + // crate's own logs starts with `q2_cockpit` — NOT `cockpit_server`, + // which is the *package* name. The filter said `cockpit_server=info`, + // so every `tracing::*` call in this binary was silently discarded. + // Found when the S3 hydrate ran correctly and logged nothing: on a + // deploy that would have been 503s with no explanation. Both names are + // listed so a future rename of either cannot silence it again, and + // `RUST_LOG` wins when set, which is the operational control a deploy + // needs. + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| { + tracing_subscriber::EnvFilter::new( + "q2_cockpit=info,cockpit_server=info,tower_http=info", + ) + }), + ) .init(); + // Hydrate the OSM slab BEFORE the listener binds: S3 is the source of + // truth, `$RAILWAY_VOL` is a cache that survives container rebuilds. Doing + // it here rather than lazily keeps `osm_features::open_slab` synchronous — + // it still just mmaps `OSM_SLAB_PATH` — and stops the first request paying + // for a 1.29 GiB download. A `None` here is not an error: the endpoint + // answers 503 exactly as it did before, and local dev sets OSM_SLAB_PATH + // directly and never reaches S3. + if let Some(path) = osm_slab_hydrate::ensure_slab_local().await { + // SAFETY: single-threaded startup, before any task or listener exists. + unsafe { std::env::set_var("OSM_SLAB_PATH", &path) }; + } + let (tx, _rx) = broadcast::channel::(256); let scene_state = shader_stream::new_scene_state(); let state = Arc::new(AppState { tx, scene_state }); diff --git a/crates/cockpit-server/src/osm_slab_hydrate.rs b/crates/cockpit-server/src/osm_slab_hydrate.rs new file mode 100644 index 000000000..4d92c219b --- /dev/null +++ b/crates/cockpit-server/src/osm_slab_hydrate.rs @@ -0,0 +1,345 @@ +//! Boot-time hydration of the baked OSM slab from S3 onto a persistent volume. +//! +//! # The topology +//! +//! ```text +//! S3 (durable source of truth) +//! │ s3://$AWS_S3_BUCKET_NAME//{berlin.soa, berlin.books, SHA256SUMS} +//! ▼ +//! $RAILWAY_VOL (persistence across container rebuilds — a CACHE, not truth) +//! │ /osm/{berlin.soa, berlin.books} +//! ▼ +//! mmap ([`crate::osm_features::open_slab`], unchanged) +//! ``` +//! +//! The volume exists so the 1.29 GiB artifact survives a rebuild — that is the +//! whole of its job. Deleting it costs a re-download and nothing else, which is +//! why every decision here treats the volume as disposable and S3 as +//! authoritative. +//! +//! # Why this runs at boot and not on first request +//! +//! `main()` is already `async`, so hydration is a step before the listener +//! binds. That keeps [`crate::osm_features::open_slab`] **synchronous and +//! untouched** — it still just mmaps `OSM_SLAB_PATH`. The alternative (hydrate +//! lazily inside the `OnceLock`) would have forced the whole read path async +//! for a one-shot boot concern, and would have made the first request pay a +//! multi-minute download while holding the initializer. +//! +//! # Checksum pinning +//! +//! `SHA256SUMS` is fetched from the same prefix and every file is verified +//! against it before use — on download AND on a cache hit. There is no +//! unverified path: a corrupt or truncated volume copy is re-fetched rather +//! than mmap'd. This mirrors `MedCare-rs`'s `scripts/fetch-frontend-assets.sh` +//! discipline, where a URL bump requires a checksum bump in the same edit. +//! +//! Verifying on a cache HIT (not only after download) is deliberate: the +//! failure this guards is a half-written file from a container killed +//! mid-download, which is exactly the case a "we already have it" check would +//! otherwise wave through. Downloads land on a `.part` file and are renamed +//! only after the hash matches, so a killed container leaves no file that +//! looks complete — but the volume outlives this code, so the check stays. +//! +//! # Absent configuration is not an error +//! +//! No bucket, no volume, no credentials ⇒ `None`, and the endpoint keeps +//! answering 503 exactly as it did before this module existed. Local +//! development sets `OSM_SLAB_PATH` directly and never reaches S3. + +use std::path::{Path, PathBuf}; + +use object_store::aws::AmazonS3Builder; +use object_store::{ObjectStore, ObjectStoreExt}; +use sha2::{Digest, Sha256}; + +/// Default S3 prefix holding the bake. Overridable with `OSM_SLAB_S3_PREFIX`. +const DEFAULT_PREFIX: &str = "q2/bakes/berlin-v1"; + +/// The slab and its codebook sidecar. Both are required: `RowSlab` can read +/// positions without the books, but identity resolution needs them, and a slab +/// without its books is what the bake itself calls unreadable. +const ARTIFACTS: [&str; 2] = ["berlin.soa", "berlin.books"]; + +/// Where the hydrated copy lives, given the volume root. +fn cache_dir(vol: &str) -> PathBuf { + Path::new(vol).join("osm") +} + +/// Resolve a local, verified slab path, hydrating from S3 if needed. +/// +/// Returns the path to set as `OSM_SLAB_PATH`, or `None` when the feature is +/// not configured — never panics, and never returns a path that failed its +/// checksum. +pub async fn ensure_slab_local() -> Option { + // 1. An explicit local path always wins. This is the local-dev and + // already-hydrated case, and it must not require S3 credentials. + if let Ok(p) = std::env::var("OSM_SLAB_PATH") { + let path = PathBuf::from(&p); + if path.is_file() { + tracing::info!(path = %p, "osm slab: using OSM_SLAB_PATH directly"); + return Some(path); + } + tracing::warn!(path = %p, "osm slab: OSM_SLAB_PATH set but not a file; trying S3"); + } + + // 2. Otherwise hydrate. Both a bucket and a destination are required. + let bucket = std::env::var("AWS_S3_BUCKET_NAME").ok()?; + // `OSM_SLAB_CACHE_DIR` overrides the volume — it is what makes this + // testable off-Railway, where /volume01 does not exist. + let vol = std::env::var("OSM_SLAB_CACHE_DIR") + .or_else(|_| std::env::var("RAILWAY_VOL")) + .ok()?; + let prefix = std::env::var("OSM_SLAB_S3_PREFIX").unwrap_or_else(|_| DEFAULT_PREFIX.to_string()); + + let dir = cache_dir(&vol); + if let Err(e) = std::fs::create_dir_all(&dir) { + tracing::error!(dir = %dir.display(), error = %e, "osm slab: cannot create cache dir"); + return None; + } + + // `from_env()` reads AWS_ENDPOINT_URL / AWS_ACCESS_KEY_ID / + // AWS_SECRET_ACCESS_KEY / AWS_DEFAULT_REGION with no glue — `aws_endpoint_url` + // is an accepted alias for the endpoint key, so a non-AWS S3 endpoint needs + // no special casing. + let store = match AmazonS3Builder::from_env() + .with_bucket_name(&bucket) + .build() + { + Ok(s) => s, + Err(e) => { + tracing::error!(error = %e, "osm slab: S3 client build failed"); + return None; + } + }; + + let sums = match fetch_sums(&store, &prefix).await { + Some(s) => s, + None => return None, + }; + + for name in ARTIFACTS { + let want = match sums.iter().find(|(k, _)| k == name).map(|(_, h)| h.clone()) { + Some(h) => h, + None => { + tracing::error!(artifact = name, "osm slab: no checksum pinned; refusing"); + return None; + } + }; + let dest = dir.join(name); + + // Cache hit, but only if it still hashes correctly — see module docs. + if dest.is_file() { + match sha256_file(&dest) { + Ok(got) if got == want => { + tracing::info!(artifact = name, "osm slab: cache hit, checksum verified"); + continue; + } + Ok(got) => tracing::warn!( + artifact = name, %got, %want, + "osm slab: cached copy failed its checksum; re-fetching" + ), + Err(e) => { + tracing::warn!(artifact = name, error = %e, "osm slab: cannot hash cached copy; re-fetching") + } + } + } + + if !download_verified(&store, &prefix, name, &dest, &want).await { + return None; + } + } + + let slab = dir.join(ARTIFACTS[0]); + tracing::info!(path = %slab.display(), "osm slab: hydrated and verified"); + Some(slab) +} + +/// Fetch and parse `SHA256SUMS` — ` ` per line, the `sha256sum` +/// format the bucket already uses for the MedCare bakes. +async fn fetch_sums(store: &impl ObjectStore, prefix: &str) -> Option> { + let path = object_store::path::Path::from(format!("{prefix}/SHA256SUMS")); + let bytes = match store.get(&path).await { + Ok(r) => match r.bytes().await { + Ok(b) => b, + Err(e) => { + tracing::error!(error = %e, "osm slab: SHA256SUMS body read failed"); + return None; + } + }, + Err(e) => { + tracing::error!(error = %e, %prefix, "osm slab: SHA256SUMS not readable"); + return None; + } + }; + Some(parse_sums(&String::from_utf8_lossy(&bytes))) +} + +/// Parse `sha256sum` output. Tolerates the `*name` binary marker and blank +/// lines; ignores anything that is not ` `. +fn parse_sums(text: &str) -> Vec<(String, String)> { + text.lines() + .filter_map(|line| { + let mut it = line.split_whitespace(); + let hash = it.next()?; + let name = it.next()?.trim_start_matches('*'); + if hash.len() == 64 && hash.chars().all(|c| c.is_ascii_hexdigit()) { + Some((name.to_string(), hash.to_ascii_lowercase())) + } else { + None + } + }) + .collect() +} + +/// Stream one object to `.part`, hash while writing, and rename into +/// place only if it matches. A mismatch leaves no file behind. +async fn download_verified( + store: &impl ObjectStore, + prefix: &str, + name: &str, + dest: &Path, + want: &str, +) -> bool { + use futures::StreamExt; + use std::io::Write; + + let path = object_store::path::Path::from(format!("{prefix}/{name}")); + let result = match store.get(&path).await { + Ok(r) => r, + Err(e) => { + tracing::error!(artifact = name, error = %e, "osm slab: download failed"); + return false; + } + }; + + let part = dest.with_extension("part"); + let mut file = match std::fs::File::create(&part) { + Ok(f) => f, + Err(e) => { + tracing::error!(artifact = name, error = %e, "osm slab: cannot create .part"); + return false; + } + }; + + let mut hasher = Sha256::new(); + let mut stream = result.into_stream(); + let mut written: u64 = 0; + while let Some(chunk) = stream.next().await { + let chunk = match chunk { + Ok(c) => c, + Err(e) => { + tracing::error!(artifact = name, error = %e, "osm slab: stream error"); + let _ = std::fs::remove_file(&part); + return false; + } + }; + hasher.update(&chunk); + if let Err(e) = file.write_all(&chunk) { + tracing::error!(artifact = name, error = %e, "osm slab: write error"); + let _ = std::fs::remove_file(&part); + return false; + } + written += chunk.len() as u64; + } + if let Err(e) = file.flush() { + tracing::error!(artifact = name, error = %e, "osm slab: flush error"); + let _ = std::fs::remove_file(&part); + return false; + } + drop(file); + + let got = hex::encode(hasher.finalize()); + if got != want { + tracing::error!(artifact = name, %got, %want, "osm slab: checksum mismatch; discarding"); + let _ = std::fs::remove_file(&part); + return false; + } + if let Err(e) = std::fs::rename(&part, dest) { + tracing::error!(artifact = name, error = %e, "osm slab: rename into place failed"); + let _ = std::fs::remove_file(&part); + return false; + } + tracing::info!( + artifact = name, + bytes = written, + "osm slab: downloaded and verified" + ); + true +} + +/// SHA-256 of a file, streamed — the artifact is 1.29 GiB and must not be read +/// into memory to be hashed. +fn sha256_file(path: &Path) -> std::io::Result { + use std::io::Read; + let mut f = std::fs::File::open(path)?; + let mut hasher = Sha256::new(); + let mut buf = vec![0u8; 1 << 20]; + loop { + let n = f.read(&mut buf)?; + if n == 0 { + break; + } + hasher.update(&buf[..n]); + } + Ok(hex::encode(hasher.finalize())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_sums_reads_the_sha256sum_format() { + let text = "\ +cbf5989ab45bc921d8a85fdbdb71c8e5029cd904a3d230a898c2b5eb81d7ebe7 berlin.soa +d12bc8a15270f9a61290fb7117c92621e1f88229a85bdfdfc4d217481addde7f *berlin.books + +not-a-hash junk.txt +"; + let got = parse_sums(text); + assert_eq!(got.len(), 2, "the junk line must not parse as a pin"); + assert_eq!( + got[0], + ( + "berlin.soa".to_string(), + "cbf5989ab45bc921d8a85fdbdb71c8e5029cd904a3d230a898c2b5eb81d7ebe7".to_string() + ) + ); + // The `*` binary marker is stripped, not treated as part of the name. + assert_eq!(got[1].0, "berlin.books"); + } + + /// A short hex string is not a pin. Without this, a truncated SHA256SUMS + /// could yield an entry that silently matches nothing and reads as "no + /// checksum for this artifact" — which `ensure_slab_local` refuses on, but + /// only because the entry is absent rather than malformed. + #[test] + fn parse_sums_rejects_a_short_hash() { + assert!(parse_sums("abc123 berlin.soa").is_empty()); + // ...and accepts it at exactly 64, so the length check is the reason + // and not some other property of the fixture. + let sixty_four = "a".repeat(64); + assert_eq!(parse_sums(&format!("{sixty_four} berlin.soa")).len(), 1); + } + + #[test] + fn sha256_file_matches_the_known_digest_of_its_bytes() { + let dir = std::env::temp_dir().join("q2-hydrate-test"); + std::fs::create_dir_all(&dir).unwrap(); + let p = dir.join("probe.bin"); + std::fs::write(&p, b"abc").unwrap(); + // The published SHA-256 of "abc" — an external anchor, not a value + // this code produced. + assert_eq!( + sha256_file(&p).unwrap(), + "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad" + ); + std::fs::remove_file(&p).ok(); + } + + #[test] + fn cache_dir_is_under_the_volume_root() { + assert_eq!(cache_dir("/volume01"), PathBuf::from("/volume01/osm")); + } +}