Release marine-atlas — versioned Parquet → S3, view-DB serving, STAC

Published

2026-07-15 18:02:43

Publish the v8 marine-atlas and stand up its serving, reproducibly:

  1. Data → S3 s3://oceanmetrics.io-public/marine-atlas/{ver}/:
    • tables/ — derived scoring tables (cell, taxon, dataset, model, cell_metric, zone*, metric).
    • dist_merged/ — merged per-taxon surfaces (ms_merge|…).
    • serve/model_cell.parquet — the serving-optimized surface: one file, globally sorted by mdl_key (row-group zone-map pruning) so a titiler tile does an HTTP-range point read; single file ⇒ anonymous GET (no S3 LIST/creds).
    • registry/dataset/model Parquet.
    • dist/ — raw per-dataset surfaces (~74G) — opt-in (RELEASE_RAW=1).
  2. View DB — a tiny serve.duckdb whose tables are views over the S3 Parquet (path-style HTTPS URLs). Titiler reads this; the big data never rsyncs to the server.
  3. STACstac_build(version) → deploy the catalog to the file host /stac.
  4. Deploy (RELEASE_DEPLOY=1) — rsync the KB view DB + STAC to msens1, (re)build the parallel titiler-v8 service, restart caddy. v7 serving is untouched (A/B).

1 Setup

Code
librarian::shelf(DBI, dplyr, duckdb, fs, glue, here, jsonlite, logger, quiet = T)
source(here("libs/paths.R"))
dir_atlas <- glue("{dir_big_v}/marine-atlas")
tbl_dir   <- glue("{dir_atlas}/tables");  dir_create(tbl_dir)
serve_dir <- glue("{dir_atlas}/serve");   dir_create(serve_dir)
manifest  <- here("data/manifests/release_marine-atlas.json"); dir_create(path_dir(manifest))
s3_ver    <- glue("{s3_atlas}/{ver}")                      # s3://…/marine-atlas/v8
s3_http   <- glue("https://s3.us-east-1.amazonaws.com/{sub('^s3://', '', s3_ver)}")  # path-style
do_s3     <- !nzchar(Sys.getenv("RELEASE_NO_S3"))
do_raw    <-  nzchar(Sys.getenv("RELEASE_RAW"))
do_deploy <-  nzchar(Sys.getenv("RELEASE_DEPLOY"))
do_apps   <-  do_deploy || nzchar(Sys.getenv("DEPLOY_APPS"))        # (re)deploy the v8 Shiny apps
do_s3_tbl <-  do_s3     || nzchar(Sys.getenv("RELEASE_S3_TABLES"))  # push derived tables (native_asset) without the full serve cutover
# server targets (parallel titiler-v8)
srv_repo  <- "/share/github/MarineSensitivity/server"
srv_viewdb<- glue("/share/data/big/{ver}/serve.duckdb")
srv_stac  <- glue("/share/data/derived/stac/{ver}")
titiler_v8<- glue("https://titiler-{ver}.marinesensitivity.org/msens")
stopifnot("run build_registry first" = {
  con0 <- dbConnect(duckdb(sdm_db, read_only = TRUE)); on.exit(dbDisconnect(con0, shutdown = TRUE))
  all(c("dataset", "model", "model_cell") %in% dbListTables(con0)) })

2 Stage derived tables + serving surface → Parquet

Code
con <- dbConnect(duckdb(sdm_db, read_only = TRUE))
dbExecute(con, "PRAGMA memory_limit='12GB'"); dbExecute(con, "PRAGMA threads=6")
[1] 0
[1] 0
Code
dbExecute(con, glue("PRAGMA temp_directory='{dir_atlas}/duckdb_tmp'"))
[1] 0
Code
# small/derived scoring tables (model_cell is released as dist_merged + serve/, not here).
# native_asset (mdl_key -> COG/PMTiles input surfaces) is included when publish_native has run.
rel_tables <- c("cell", "taxon", "dataset", "model", "metric",
                "cell_metric", "zone", "zone_cell", "zone_metric", "native_asset")
rel_tables <- intersect(rel_tables, dbListTables(con))
for (t in rel_tables)
  msens::copy_atlas_parquet(con, glue("SELECT * FROM {t}"), glue("{tbl_dir}/{t}.parquet"))

# serving surface: Hive-PARTITIONED by the integer mdl_id (from the model registry), which REPLACES
# the 40-byte mdl_key string in the stored rows (compact: rows are just cell_id,val; mdl_id lives in
# the path). A titiler tile reads exactly ONE partition (serve/model_cell/mdl_id={id}/data_0.parquet)
# by exact path -> anonymous GET, no S3 LIST, no global ORDER BY (avoids the ~500GB sort spill). The
# STABLE public identifier stays mdl_key: titiler resolves mdl_key->mdl_id from the model registry,
# so mdl_id (which renumbers as models are added) never appears in a URL.
# skip the rebuild if it exists (model_cell rarely changes); RELEASE_REDO_SERVE=1 forces.
serve_mc <- glue("{serve_dir}/model_cell")
if (!dir_exists(serve_mc) || nzchar(Sys.getenv("RELEASE_REDO_SERVE"))) {
  if (dir_exists(serve_mc)) dir_delete(serve_mc)
  dbExecute(con, "SET partitioned_write_max_open_files=1024")   # ~one partition per merged model
  msens::copy_atlas_parquet(
    con, "SELECT m.mdl_id, c.cell_id, c.val FROM model_cell c JOIN model m USING (mdl_key)",
    serve_mc, partition_by = "mdl_id")
} else {
  log_info("serve/model_cell/ exists — skipping repartition (RELEASE_REDO_SERVE=1 to force)")
}
# content fingerprint of the served surface (source table = deterministic, cheaper than scanning parts)
serve_hash <- msens::hash_query(con, "SELECT mdl_key, cell_id, val FROM model_cell")
n_parts    <- length(dir_ls(serve_mc, type = "directory"))
dbDisconnect(con, shutdown = TRUE)
msens::report_table(tibble(
  file = c(glue("tables/{rel_tables}.parquet"), glue("serve/model_cell/ ({n_parts} mdl_id partitions)")),
  mb   = round(c(file_size(glue("{tbl_dir}/{rel_tables}.parquet")),
                 sum(file_size(dir_ls(serve_mc, recurse = TRUE, glob = "*.parquet")))) / 1e6, 1)),
  caption = "released Parquet (V2/zstd/80MB)")
released Parquet (V2/zstd/80MB)
file mb
tables/cell.parquet 391.9
tables/taxon.parquet 1.0
tables/dataset.parquet 0.0
tables/model.parquet 1.1
tables/metric.parquet 0.0
tables/cell_metric.parquet 68.3
tables/zone.parquet 0.0
tables/zone_cell.parquet 0.1
tables/zone_metric.parquet 0.0
tables/native_asset.parquet 0.5
serve/model_cell/ (17763 mdl_id partitions) 3266.5

3 Sync data → S3

Code
sync <- function(local, prefix, del = FALSE, tries = 4) {
  if (!dir_exists(local)) { log_warn("skip (missing): {local}"); return(invisible()) }
  args <- c("s3", "sync", local, glue("{s3_ver}/{prefix}"), "--only-show-errors", "--no-progress")
  if (del) args <- c(args, "--delete")   # prune stale S3 keys not present locally
  # retry transient network errors: `aws s3 sync` is resumable (re-uploads only missing/changed),
  # so retrying the same command picks up where a dropped connection left off.
  for (i in seq_len(tries)) {
    st <- system2("aws", args, stdout = TRUE, stderr = TRUE)
    if (is.null(attr(st, "status")) || attr(st, "status") == 0) {
      log_info("synced {path_file(local)} -> {s3_ver}/{prefix}{if (del) ' (--delete)' else ''}")
      return(invisible())
    }
    log_warn("sync {prefix} attempt {i}/{tries} failed (transient?); retrying …")
  }
  stop(paste(tail(st, 20), collapse = "\n"))
}
# derived tables (incl native_asset, which the species-app native/model toggle reads) can be
# pushed on their own via RELEASE_S3_TABLES, without the multi-GB serve/dist_merged cutover.
if (do_s3_tbl) sync(tbl_dir, "tables")
if (do_s3) {
  # serve is Hive-partitioned by mdl_id, which RENUMBERS when the model set changes (e.g. the
  # bird crosswalk merged 25 taxa) -> --delete prunes stale partitions + the legacy single file.
  sync(serve_dir,                       "serve", del = TRUE)
  sync(glue("{dir_atlas}/dist_merged"), "dist_merged")
  sync(glue("{dir_atlas}/registry"),    "registry")
  if (do_raw) sync(glue("{dir_atlas}/dist"), "dist")
} else if (do_s3_tbl) {
  log_info("RELEASE_S3_TABLES: synced tables/ only (serve/dist_merged/registry NOT pushed)")
} else log_info("RELEASE_NO_S3 set — staged locally only")

4 Build the tiny view DB (views over the S3 Parquet)

Code
viewdb <- glue("{serve_dir}/serve.duckdb")
if (file_exists(viewdb)) file_delete(viewdb)
cv <- dbConnect(duckdb(viewdb))
# view sources: the released S3 Parquet (production) OR the local staged Parquet when
# RELEASE_NO_S3 (a local dry-run — nothing was pushed, so an S3 view would point at nothing).
if (do_s3 || do_s3_tbl) {   # S3 views whenever the tables live on S3 (even a tables-only push)
  dbExecute(cv, "INSTALL httpfs; LOAD httpfs;")
  # tables: single files via explicit path-style HTTPS -> anonymous GET + HTTP range (the dotted
  # bucket breaks virtual-hosted TLS, so never use s3://… for these).
  tbl_src <- glue("{s3_http}/tables")
  # model_cell glob needs S3 LIST -> the aws extension + credential chain (server has creds).
  dbExecute(cv, "INSTALL aws; LOAD aws; SET s3_url_style='path';")
  dbExecute(cv, "CREATE OR REPLACE SECRET atlas_s3 (TYPE s3, PROVIDER credential_chain, REGION 'us-east-1');")
  mc_glob <- glue("s3://{sub('^s3://', '', s3_ver)}/serve/model_cell/*/*.parquet")
} else {
  tbl_src <- tbl_dir                                 # local staged tables
  mc_glob <- glue("{serve_mc}/*/*.parquet")          # local partitioned serve
}
# scoring tables store the metric in `val`; the Shiny apps still reference `value` (the pipeline
# value->val rename never fully propagated), so expose `val AS value` on these views as a
# back-compat alias — both column names resolve, no per-app query rewrites needed.
val_tbls <- c("cell_metric", "zone_metric", "zone")
for (t in rel_tables) {
  extra <- if (t %in% val_tbls) ", val AS value" else ""
  dbExecute(cv, glue("CREATE VIEW {t} AS SELECT *{extra} FROM read_parquet('{tbl_src}/{t}.parquet')"))
}
# model_cell is Hive-partitioned by the integer mdl_id (stored rows are cell_id,val). The HOT tile
# path reads one partition by EXACT path in titiler (factory mdl_key->mdl_id fast-path, anonymous
# GET, no LIST). This glob VIEW is for ad-hoc/statistics queries; it joins the `model` view back to
# the STABLE mdl_key so ad-hoc queries can select by mdl_key, not the internal id.
dbExecute(cv, glue("CREATE VIEW model_cell AS
  SELECT m.mdl_key, mc.mdl_id, mc.cell_id, mc.val, mc.val AS value
  FROM read_parquet('{mc_glob}', hive_partitioning = true) mc
  JOIN model m USING (mdl_id)"))
[1] 0
Code
dbDisconnect(cv, shutdown = TRUE)
log_info("view DB {viewdb} ({round(file_size(viewdb)/1e3,1)} KB) — {if (do_s3 || do_s3_tbl) 'S3' else 'LOCAL'} sources, model_cell partitioned by mdl_id (public key = mdl_key)")

6 Deploy to msens1 (parallel titiler-{ver})

Code
if (!do_deploy) {
  log_info("RELEASE_DEPLOY unset — skipping server deploy (S3 data + local view DB/STAC ready)")
} else {
# 1. the KB view DB + the STAC version subtree (data stays on S3 — no big rsync)
system2("ssh", c("msens", glue("mkdir -p {path_dir(srv_viewdb)} {srv_stac}")))
system2("rsync", c("-aq", viewdb, glue("msens:{srv_viewdb}")))
system2("rsync", c("-aq", glue("{stac_dir}/{ver}/"), glue("msens:{srv_stac}/")))
# add this version as a child of the STAC root (keep other versions)
system2("ssh", c("msens", shQuote(glue(
  "python3 - <<'PY'\nimport json\np='/share/data/derived/stac/catalog.json'\nc=json.load(open(p))\nh=[l['href'] for l in c['links'] if l.get('rel')=='child']\nif './{ver}/collection.json' not in h:\n c['links'].append({{'rel':'child','href':'./{ver}/collection.json'}}); json.dump(c,open(p,'w'),indent=2)\nPY"))))
# 2. update the server repo + REBUILD & FORCE-RECREATE titiler-{ver} (so a new factory always
# takes effect even if compose sees no config change), restart caddy. Explicit fetch + ff-only
# with `set -e` so a stale/failed pull FAILS LOUDLY instead of silently serving the old factory
# (a silent `git pull --quiet` once left the mdl_key factory undeployed for days).
deploy_out <- system2("ssh", c("msens", shQuote(glue(
  "set -e; cd {srv_repo} && git fetch --quiet origin && git merge --ff-only origin/main && ",
  "docker compose up -d --build --force-recreate titiler-{ver} && docker compose restart caddy && ",
  "echo DEPLOY_OK $(git rev-parse --short HEAD)"))), stdout = TRUE, stderr = TRUE)
log_info("titiler-{ver} deploy: {paste(tail(deploy_out, 2), collapse=' | ')}")
if (!any(grepl("DEPLOY_OK", deploy_out))) stop(paste(tail(deploy_out, 15), collapse = "\n"))
# 3. smoke test — the stable-mdl_key fast-path (factory resolves mdl_key->mdl_id, reads one
# partition). Hit tilejson.json (not /statistics, which requires a geometry) → expect 200.
c0 <- dbConnect(duckdb(sdm_db, read_only = TRUE))
mk <- dbGetQuery(c0, "SELECT mdl_key FROM model WHERE ds_key='ms_merge' ORDER BY mdl_id LIMIT 1")$mdl_key
dbDisconnect(c0, shutdown = TRUE)
code <- system2("curl", c("-s","-o","/dev/null","-w","%{http_code}","--retry","3",
  glue("{titiler_v8}/tilejson.json?mdl_key={URLencode(mk, reserved = TRUE)}")), stdout = TRUE)
log_info("titiler-{ver} /tilejson.json?mdl_key={mk} smoke test: HTTP {code} (expect 200)")
if (!identical(code, "200")) log_warn("titiler-{ver} merged-surface smoke test != 200 — check the factory")
}

7 Deploy the v8 Shiny apps (reload only apps_v8; v7 untouched)

Reproducible app deploy — no ad-hoc ssh. Pull the apps_v8 checkout (MarineSensitivity/apps@main, symlinked as /share/shiny_apps/{species,scores}_v8) and reload only the v8 apps via Shiny Server’s per-app restart.txt. The v7 apps are a separate checkout (/share/shiny_apps/{species,scores}mapsp/mapgl) sharing the same container; they are not restarted, so v7 serving is uninterrupted. Gated by RELEASE_DEPLOY (full release) or the granular DEPLOY_APPS=1.

Code
if (!do_apps) {
  log_info("RELEASE_DEPLOY/DEPLOY_APPS unset — skipping Shiny app deploy")
} else {
  apps_co  <- "/share/github/MarineSensitivity/apps_v8"   # main branch; symlinked as *_v8
  msens_co <- "/share/github/MarineSensitivity/msens"
  v8_apps  <- c("species", "scores")
  touches  <- paste(sprintf("touch %s/%s/restart.txt", apps_co, v8_apps), collapse = "; ")
  # `set -e` + explicit ff-only fetch (a silent `git pull --quiet` once left the app + msens
  # stale for hours). ALSO reinstall msens into the shiny/rstudio container: a stale container
  # msens lacked `cell_tile_url(mdl_key=)` and crashed the merged surface. Reload only the v8 apps.
  cmd <- glue("set -e; ",
              "cd {apps_co}  && git fetch --quiet origin && git merge --ff-only origin/main; ",
              "cd {msens_co} && git fetch --quiet origin && git merge --ff-only origin/main && ",
              "docker exec rstudio R CMD INSTALL --no-multiarch {msens_co} >/dev/null 2>&1; ",
              "{touches}; echo APPS_OK $(git -C {apps_co} rev-parse --short HEAD)")
  out <- system2("ssh", c("msens", shQuote(cmd)), stdout = TRUE, stderr = TRUE)
  if (!any(grepl("APPS_OK", out))) stop(paste(tail(out, 15), collapse = "\n"))
  log_info("deployed v8 apps [{paste(v8_apps, collapse=', ')}] + reinstalled msens; v7 apps untouched: {tail(out,1)}")
}

8 Manifest

Code
# content-addressed: fingerprint of the served model_cell surface (run-mode flags like
# pushed_s3/deployed are side effects, not content, so they're excluded).
msens::write_manifest(
  manifest, target = "release_marine_atlas", content_hash = serve_hash,
  stats = list(ver = ver, s3 = s3_ver, n_tables = length(rel_tables),
               tables = paste(sort(rel_tables), collapse = ","),
               serve = "serve/model_cell/ (partitioned by mdl_id)", n_partitions = n_parts,
               titiler = titiler_v8),
  force = msens::force_target("release_marine_atlas"))