Bootstrap a release version — clone the unchanged ingests from the previous version

ver in libs/paths.R has only ever moved once (v7 → v8), and that bump rewrote every ingest. A bump on the same grid (v8 → v9: same global05 cells, same r_cellid_global.tif) must reuse the ingests nothing changed — AquaMaps alone is 50 GB and 40 min on the server, BirdLife and IUCN are hours — and it must do so reproducibly: which datasets were carried forward, from where, and whether they are the surfaces the content-addressed manifests checkpointed.

This notebook clones {dir_big}/{ver_prev}/marine-atlas/dist/dataset=* (+ model_*.csv) into this version’s dist/. On macOS/APFS that is cp -c (clonefile: instant, zero extra bytes, copy-on-write); on Linux it is a hardlink tree (cp -al). The ingest notebooks then resume against the clone — every Parquet already present — so their manifests do not change and nothing downstream re-runs for the wrong reason. Datasets this version re-ingests (v9: nothing; a future AquaMaps refresh would be BOOTSTRAP_SKIP_DS=am) are simply not cloned.

Verification: by default file count + byte total per dataset must equal ver_prev’s; BOOTSTRAP_VERIFY=1 re-hashes each cloned surface with msens::hash_parquet() and asserts equality with that ingest’s data/manifests/ingest_*.json content_hash — proof that what was cloned is the checkpointed surface and not a stale directory.

1 Design

flowchart LR
  prev[("{ver_prev}/marine-atlas/dist/<br/>dataset=* + model_*.csv")] -->|"cp -c (APFS clonefile)<br/>cp -al (Linux)"| cur[("{ver}/marine-atlas/dist/")]
  cur --> chk["count + bytes == ver_prev<br/>BOOTSTRAP_VERIFY=1: hash_parquet == ingest manifest"]
  chk --> mf["manifest: what was cloned,<br/>from where, how"]

flowchart LR
  prev[("{ver_prev}/marine-atlas/dist/<br/>dataset=* + model_*.csv")] -->|"cp -c (APFS clonefile)<br/>cp -al (Linux)"| cur[("{ver}/marine-atlas/dist/")]
  cur --> chk["count + bytes == ver_prev<br/>BOOTSTRAP_VERIFY=1: hash_parquet == ingest manifest"]
  chk --> mf["manifest: what was cloned,<br/>from where, how"]

Figure 1: Same-grid version bump: clone the unchanged ingests (copy-on-write) so they resume, verify against the ingest checkpoints

2 Setup

librarian::shelf(DBI, dplyr, duckdb, fs, glue, here, jsonlite, logger, purrr, readr, tibble, yaml, quiet = TRUE)
source(here("libs/paths.R"))   # ver, ver_prev, dir_big, dir_big_v
source(here("libs/vars.R"))    # bootstrap_verify
options(readr.show_col_types = FALSE)

dist_prev <- path.expand(glue("{dir_big}/{ver_prev}/marine-atlas/dist"))
dist_cur  <- path.expand(glue("{dir_big_v}/marine-atlas/dist"))
manifest  <- here("data/manifests/bootstrap_version.json")
skip_ds   <- trimws(strsplit(Sys.getenv("BOOTSTRAP_SKIP_DS", ""), ",")[[1]])
skip_ds   <- skip_ds[nzchar(skip_ds)]
dir_create(c(dist_cur, path_dir(manifest)))

versions <- read_csv(here("data/versions.csv"))
stopifnot(
  "this version must have a row in data/versions.csv before anything is built for it" = ver %in% versions$ver,
  "ver_prev has no marine-atlas dist/ to clone from" = dir_exists(dist_prev),
  "ver and ver_prev must be on the same grid — a grid change is a re-ingest, not a clone" =
    identical(msens::grid_for_ver(ver), msens::grid_for_ver(ver_prev)))
log_info("bootstrap {ver} from {ver_prev}: {dist_prev} -> {dist_cur}; skip: {if (length(skip_ds)) paste(skip_ds, collapse=', ') else 'none'}")

3 Clone the unchanged ingests (copy-on-write)

# copy-on-write clone of one directory (or file). macOS: `cp -c` = clonefile(2) on APFS —
# instant and zero extra bytes until either side is modified. Linux: hardlinks (`cp -al`),
# which the ingests never break because they only ever ADD files (resume) or delete + rebuild
# their whole dataset dir under REDO_*. rsync would COPY 60 GB; that is what this avoids.
clone <- function(from, to) {
  if (Sys.info()[["sysname"]] == "Darwin") system2("cp", c("-Rc", shQuote(from), shQuote(to)))
  else                                     system2("cp", c("-al", shQuote(from), shQuote(to)))
}
tally <- function(d) {
  f <- dir_ls(d, recurse = TRUE, type = "file")
  tibble(n_files = length(f), bytes = sum(file_size(f)))
}

ds_dirs <- dir_ls(dist_prev, type = "directory", regexp = "dataset=")
ds_keys <- sub("^dataset=", "", path_file(ds_dirs))
todo    <- setdiff(ds_keys, skip_ds)

rows <- map_dfr(todo, function(ds) {
  from <- path(dist_prev, glue("dataset={ds}")); to <- path(dist_cur, glue("dataset={ds}"))
  method <- if (dir_exists(to)) "exists" else {
    st <- clone(from, to)
    if (!identical(st, 0L)) stop(glue("clone of {ds} failed (exit {st})"))
    if (Sys.info()[["sysname"]] == "Darwin") "clonefile" else "hardlink"
  }
  tp <- tally(from); tc <- tally(to)
  tibble(ds_key = ds, method = method,
         n_files_prev = tp$n_files, n_files = tc$n_files,
         gb_prev = round(tp$bytes / 1e9, 2), gb = round(tc$bytes / 1e9, 2),
         same = tp$n_files == tc$n_files && tp$bytes == tc$bytes)
})
# the per-dataset species crosswalks travel with their surfaces
for (f in dir_ls(dist_prev, glob = "*model_*.csv")) {
  to <- path(dist_cur, path_file(f))
  if (!file_exists(to) && sub("^model_(.+)\\.csv$", "\\1", path_file(f)) %in% todo) file_copy(f, to)
}
# a dataset that was cloned earlier and has since diverged (an ingest ADDED models to it) is
# legitimately different; a dataset that is SMALLER than its source is a truncated clone
short <- rows |> filter(n_files < n_files_prev)
stopifnot("cloned dataset(s) have FEWER files than ver_prev — truncated clone" = nrow(short) == 0)
knitr::kable(rows)

ds_key	method	n_files_prev	n_files	gb_prev	gb	same
am	exists	23699	23699	53.83	53.83	TRUE
bl	exists	10995	10995	6.65	6.65	TRUE
ca_nmfs	exists	1	1	0	0	TRUE
ch_fws	exists	29	29	0	0	TRUE
ch_nmfs	exists	38	38	0	0	TRUE
gm	exists	217	217	20.48B	20.48B	TRUE
nc	exists	313	313	81.92B	81.92B	TRUE
rng_fws	exists	106	106	10.24B	10.24B	TRUE
rng_iucn	exists	6246	6246	5.5	5.5	TRUE
rng_turtle_swot_dps	exists	14	14	286.72NA	286.72NA	TRUE

4 Clone the unchanged NATIVE assets (+ their published copies, BOOTSTRAP_PUBLISH=1)

publish_native.qmd builds per-model COGs / PMTiles for every served input and is resumable by file existence — so a version directory without native/ rebuilds all 18,710 AquaMaps COGs, the 0.5° originals, the vector PMTiles and the 7 GB IUCN GeoPackage (hours), for inputs that did not change. Clone them from ver_prev the same way as dist/. The merged COGs (merged, _merged_parts) are deliberately NOT cloned: they are what a new version repaints.

Published copies are per version on S3 ({ver}/native/…) and on the file host (/share/data/derived/pmtiles/{ver}), so with BOOTSTRAP_PUBLISH=1 they are copied server-side (aws s3 sync s3://…/{ver_prev}/native/x s3://…/{ver}/native/x, and a hard-link copy on the file host) — no bytes leave the laptop, and publish_native’s own sync then finds every key present.

nat_prev <- path.expand(glue("{dir_big}/{ver_prev}/marine-atlas/native"))
nat_cur  <- path.expand(glue("{dir_big_v}/marine-atlas/native")); dir_create(nat_cur)
# unchanged INPUT assets only; merged/_merged_parts are this version's to repaint
nat_items <- c("am", "am_native", "pmtiles", "vec_grid", "src", "gm", "gm_cog_registry.csv",
               "_am_parts", "_am_parts.map.csv")
nat_rows <- map_dfr(nat_items, function(it) {
  from <- path(nat_prev, it); to <- path(nat_cur, it)
  if (!file_exists(from)) return(tibble(item = it, method = "absent in ver_prev", n_files = 0L, gb = 0))
  method <- if (file_exists(to)) "exists" else {
    st <- clone(from, to); if (!identical(st, 0L)) stop(glue("clone of native/{it} failed (exit {st})"))
    if (Sys.info()[["sysname"]] == "Darwin") "clonefile" else "hardlink" }
  t <- if (is_dir(to)) tally(to) else tibble(n_files = 1L, bytes = file_size(to))
  tibble(item = it, method = method, n_files = t$n_files, gb = round(t$bytes / 1e9, 2))
})
knitr::kable(nat_rows, caption = "native assets carried forward from ver_prev (copy-on-write)")

native assets carried forward from ver_prev (copy-on-write)

item	method	n_files	gb
am	clonefile	18710	5.85
am_native	clonefile	18703	153.6B
pmtiles	clonefile	6753	7.09
vec_grid	clonefile	6753	153.6NA
src	clonefile	1	9.07
gm	clonefile	17	0
gm_cog_registry.csv	clonefile	1	0
_am_parts	clonefile	28601	49.83
_am_parts.map.csv	clonefile	1	0

if (.flag("BOOTSTRAP_PUBLISH")) {
  s3_prev <- glue("{s3_atlas}/{ver_prev}"); s3_cur <- glue("{s3_atlas}/{ver}")
  for (p in c("am", "am_native", "pmtiles", "vec_grid")) {
    st <- system2("aws", c("s3", "sync", shQuote(glue("{s3_prev}/native/{p}")), shQuote(glue("{s3_cur}/native/{p}")),
                           "--only-show-errors", "--no-progress"), stdout = TRUE, stderr = TRUE)
    if (!is.null(attr(st, "status")) && attr(st, "status") != 0) stop(paste(st, collapse = "\n"))
    n <- length(system2("aws", c("s3", "ls", shQuote(glue("{s3_cur}/native/{p}/")), "--recursive"), stdout = TRUE))
    log_info("S3 native/{p}: {ver_prev} -> {ver} server-side copy, {n} objects")
  }
  # PMTiles on the file host: hard-link copy of ver_prev's tree (instant, no transfer)
  out <- system2("ssh", c("msens", shQuote(glue(
    "set -e; cd /share/data/derived/pmtiles && if [ ! -d {ver} ]; then cp -al {ver_prev} {ver}; fi; ",
    "echo PMTILES_OK $(find {ver} -type f | wc -l)"))), stdout = TRUE, stderr = TRUE)
  stopifnot("file-host PMTiles copy failed" = any(grepl("PMTILES_OK", out)))
  log_info("file host pmtiles/{ver}: {grep('PMTILES_OK', out, value = TRUE)}")
} else log_info("BOOTSTRAP_PUBLISH unset — published native copies (S3, file host) not made")

5 Verify against the ingest checkpoints (BOOTSTRAP_VERIFY=1)

The count/bytes check above says the clone is complete; it cannot say the source was the surface the pipeline checkpointed. Re-hashing does — the ingest manifests are content-addressed (msens::hash_parquet, order-independent), so equality is proof, not suggestion. Off by default because hashing 60 GB takes a while; on when a version is being built for release.

# ingest target -> ds_key, from each ingest notebook's msens: block (the DAG parser keeps only
# target/type/deps/output, so read the dataset block the way build_registry.qmd does)
ing <- map_dfr(dir_ls(here(), glob = "*.qmd"), \(f) {
  txt <- readLines(f, warn = FALSE); d <- which(txt == "---")
  if (length(d) < 2) return(NULL)
  m <- tryCatch(yaml::yaml.load(paste(txt[(d[1] + 1):(d[2] - 1)], collapse = "\n")), error = \(e) NULL)$msens
  if (is.null(m) || !identical(m$workflow_type, "ingest") || is.null(m$dataset$ds_key)) return(NULL)
  tibble(target_name = m$target_name, ds_key = m$dataset$ds_key, output = m$output)
})

verified <- tibble()
if (bootstrap_verify) {
  con <- dbConnect(duckdb())
  verified <- map_dfr(seq_len(nrow(ing)), function(i) {
    ds <- ing$ds_key[i]; if (!ds %in% rows$ds_key) return(NULL)
    mf <- here(ing$output[i])
    expect <- if (file_exists(mf)) fromJSON(mf)$content_hash else NA_character_
    got <- msens::hash_parquet(glue("{dist_cur}/dataset={ds}/*.parquet"), con)
    tibble(ds_key = ds, target = ing$target_name[i], expected = expect, got = got,
           ok = !is.na(expect) && identical(expect, got))
  })
  dbDisconnect(con, shutdown = TRUE)
  knitr::kable(verified)
  stopifnot("a cloned dataset does not hash to its ingest checkpoint" = all(verified$ok))
} else log_info("BOOTSTRAP_VERIFY unset — clone checked by count + bytes only")

6 Manifest

# content-addressed on WHAT was cloned (dataset set + sizes + method), not when
h <- digest::digest(rows |> select(ds_key, n_files, gb, method), algo = "xxhash64")
msens::write_manifest(
  manifest, target = "bootstrap_version", content_hash = h,
  stats = list(ver = ver, ver_prev = ver_prev, n_datasets = nrow(rows), n_native = sum(nat_rows$n_files),
               datasets = paste(rows$ds_key, collapse = ","), skipped = paste(skip_ds, collapse = ","),
               gb_total = sum(rows$gb), method = paste(unique(rows$method), collapse = ","),
               verified = if (nrow(verified)) all(verified$ok) else NA),
  force = msens::force_target("bootstrap_version"))