Keep the log history through schema_consolidate (#354)

CLAUDE.md

@@ -22,7 +22,7 @@ Experimental package — breaking all the time and loving the learning curve. St
 - **The recompute takes 154 min at 241 groups and `-j4`** (median 2.0 min per group, longest 16 min), not the ~1 h the plan assumed.
 - **A present species with no barrier in a group is accessible** (`access_<sp>` 1 or 2) **and has `has_barriers_<sp>_dnstr` FALSE**, so its mapping code is built. Zero barrier rows mean NA ("no data") only when nothing says the species is present. Before `91a968ab` the flag stayed NA and UPET RB shipped with `mapping_code_rb` "" on every segment; UPET RB is the only such case in the 241.
 - **`fresh_default` is 31 GB** (14 GB tables, 17 GB indexes; 11.39 M segments), up from 7.3 GB. Summing `pg_total_relation_size()` over every `pg_class` row counts each index twice (48 GB). The fwapg volume is 600 GiB with 253 G free after the run.
-- **The run replaced the August KOTL, LARL and SLOC builds and the 2026-10-12 DUNC build** (status below), which published floodplains items had come from. Those four old builds survive in a machine-local `pg_dump`, the driver's pre-consolidate backup: it was taken after m1 had rebuilt its own 76 groups but before the cloud hosts' groups (all four were on job3) were copied in, so it is **not** the whole pre-run schema. The pre-run state of the other pre-run groups m1 rebuilt is gone. Operator, 2026-10-17: no floodplains item needs the old builds restored.
+- **The run replaced the August KOTL, LARL and SLOC builds and the 2026-10-12 DUNC build** (status below), which published floodplains items had come from. Those four old builds survive in a machine-local `pg_dump`, the driver's pre-consolidate backup: it was taken after m1 had rebuilt its own 76 groups but before the cloud hosts' groups (all four were on job3) were copied in, so it is **not** the whole pre-run schema. The pre-run state of the other pre-run groups m1 rebuilt is gone. Operator, 2026-10-17: no floodplains item needs the old builds restored. **Their log rows were restored from that dump (#354):** DUNC, KOTL, LARL, SLOC and PINE (also cypher-built, not named before) each hold their earlier `log` row again, and their `log_input` rows (5 + 60; `data-raw/logs/consolidate_history_354/`). Since #354 consolidate keeps the dispatcher's history in the three log tables it copies (`log`, `log_input`, `log_recompute`), replacing their rows there by primary key.
 
 ## Status (2026-10-16) — `default_tuned` reads wet's discharge; BT's MAD minima 0.043 / 0.043 (#331)
 

R/lnk_log.R

@@ -492,19 +492,23 @@ cols_log_input <- c(
 }
 
 
-#' Create the provenance tables. Idempotent.
+#' The provenance tables: name, columns and primary key.
 #'
-#' Called from [lnk_persist_init()] so a standalone init produces a complete
-#' schema, and again from the run-start path so a run never fails for want of
-#' a table.
+#' These tables are history, not per-group state, and no primary key here
+#' includes `watershed_group_code`. Of them, `data-raw/schema_consolidate.R`
+#' copies the three that carry `watershed_group_code` (`log`, `log_input`,
+#' `log_recompute`; the config-snapshot tables never leave a host). It
+#' replaces their rows on the destination by primary key and never deletes
+#' them there by watershed group, so a group copied home from another host
+#' keeps the rows of its earlier builds (link#354).
 #'
+#' Read by [.lnk_log_create_tables()] and by `schema_consolidate.R`, so the two
+#' agree on which tables are history. (`lnk_persist_init()`'s drift check keeps
+#' its own list of log tables.)
+#'
+#' @return A list of specs named by table.
 #' @noRd
-.lnk_log_create_tables <- function(conn, schema) {
-  # The log is opened BEFORE lnk_persist_init runs (the open row must predate
-  # any write, so wsg_upstream reflects the state the run started from), so on
-  # a brand-new persist schema nothing has created it yet.
-  .lnk_db_execute(conn, sprintf("CREATE SCHEMA IF NOT EXISTS %s", schema))
-
+.lnk_log_specs <- function() {
   specs <- list(
     list(table = "log", cols = cols_log, pk = "run_id"),
     list(table = "log_recompute", cols = cols_log_recompute,
@@ -519,6 +523,25 @@ cols_log_input <- c(
     list(table = "log_dimensions", cols = .lnk_cols_log_dimensions(),
          pk = c("config_hash", "species"))
   )
+  names(specs) <- vapply(specs, `[[`, character(1), "table")
+  specs
+}
+
+
+#' Create the provenance tables. Idempotent.
+#'
+#' Called from [lnk_persist_init()] so a standalone init produces a complete
+#' schema, and again from the run-start path so a run never fails for want of
+#' a table.
+#'
+#' @noRd
+.lnk_log_create_tables <- function(conn, schema) {
+  # The log is opened BEFORE lnk_persist_init runs (the open row must predate
+  # any write, so wsg_upstream reflects the state the run started from), so on
+  # a brand-new persist schema nothing has created it yet.
+  .lnk_db_execute(conn, sprintf("CREATE SCHEMA IF NOT EXISTS %s", schema))
+
+  specs <- .lnk_log_specs()
 
   for (s in specs) {
     .lnk_db_execute(conn, sprintf(

RUNBOOK.md

@@ -504,8 +504,14 @@ EXCEPT SELECT watershed_group_code FROM <persist>.log;
     persist schema). This table is what is loaded here, read by runs into
     every schema, and `state_clean.sh` drops `fresh` but not `provenance`, so
     loader rows that cannot be re-created (link#270's FWA rows) survive it.
-- `log` and `log_input` carry `watershed_group_code` so `schema_consolidate.R`
-  auto-discovers them; `log_parameters_fresh` / `log_parameters_habitat_thresholds` /
+- `log`, `log_input` and `log_recompute` carry `watershed_group_code` so
+  `schema_consolidate.R` auto-discovers them. They are history, not per-group
+  state: consolidate replaces their rows on the dispatcher by primary key (one
+  key is one build) and never deletes them there by group (the source's copy
+  is still cleared once copied), so a group copied home from a cloud host
+  keeps its earlier builds' rows (link#354; before it, every cloud-built group
+  kept only its newest row). The set is `.lnk_log_specs()`, read by the DDL
+  and by consolidate. `log_parameters_fresh` / `log_parameters_habitat_thresholds` /
   `log_dimensions` deliberately
   do not (they key on `config_hash`), so **they do not yet travel between
   hosts** — follow-up PR.
@@ -586,10 +592,10 @@ and their rows would be indistinguishable from a real post-consolidate
 recompute — so the verify script's model-vs-recompute check would pass on
 benchmark noise.
 
-**Known exposure:** `schema_consolidate.R` DELETEs its bucket from every
-discovered table before COPYing. `log_recompute` is populated *after*
-consolidate, so a second or resumed consolidate over the same bucket will wipe
-those rows. Re-run the recompute if that happens.
+**A second or resumed consolidate keeps these rows.** `log_recompute` is
+populated *after* consolidate, and consolidate leaves the log tables out of its
+bucket DELETE: their rows are replaced by primary key, in one transaction per
+table (link#354). Before #354 a re-consolidate over the same bucket wiped them.
 
 ### The bcfp reference pin, tunnel-free
 
@@ -1194,10 +1200,11 @@ what stops `=== READY (install stage only; ...)` satisfying a full-prep check.
 
 ### Two post-conditions, and why they are not optional
 
-`schema_consolidate` DELETEs the destination bucket
-(`schema_consolidate.R:272-276`) and *then* COPYs (`:313-316`). A host that
-produced nothing therefore does not merely fail to add rows — it **removes** the
-rows already there for those WSGs and returns `ok = TRUE`. So:
+`schema_consolidate` DELETEs the destination bucket from every persist table
+(its step 3.5) and *then* COPYs (step 4). A host that produced nothing
+therefore does not merely fail to add rows — it **removes** the rows already
+there for those WSGs and returns `ok = TRUE`. (The log tables are left out of
+that DELETE, so a group's log history survives it; link#354.) So:
 
 - before consolidate, every host must account for its whole bucket, read from
   its **database**: `<schema>.log` rows for the run_uid with `date_end` set,
@@ -1266,9 +1273,19 @@ To recover a kept host: run its `ip_command`, consolidate its `groups` with
 `groups_unknown`, and only where `groups_measured` is `yes`**: `groups` is
 then the finished work still on the host. `unsettled` means wait until
 nothing runs there and re-measure. `schema_consolidate` deletes its bucket on the dispatcher before it
-copies, so consolidating a group the host no longer holds (already
-consolidated, then deleted from the source), or holds only in part, loses the
-dispatcher's copy. Where it says `no`, find out what is left first. If the
+copies, so consolidating a group the host holds only in part loses the
+dispatcher's copy of its persist rows (its log rows survive, link#354).
+Where both sides have a `log` table, consolidate refuses a source outright
+(`stage = "dest_log_newer"`, nothing deleted) when a group has a log row on
+the dispatcher and the source's newest is older or absent: the persist tables
+carry no run id, so that newest row is what says which build they hold. So a
+group the host no longer holds at all (already consolidated, then deleted from
+the source, log rows included) is refused rather than lost, and a group re-run
+on the dispatcher during recovery is never overwritten by the kept host's
+older build. A group with persist rows but **no** log row on the dispatcher
+(built before the log tables, link#127: 59 of `fresh`'s 98 groups, 0 in
+`fresh_default`, 2026-10-17) is not protected: it is deleted and, if the host
+no longer holds it, not replaced. Where it says `no`, find out what is left first. If the
 run's recompute ran, it ran without the kept groups: recompute each one and
 each run group upstream of it (`wsg_recompute_one.R`). If it did not run
 (consolidate failed, coverage FATAL), recompute every group of the run; the

data-raw/README.md

@@ -90,7 +90,7 @@ Run-adjacent helpers (planning, consolidation across hosts).
 | Script | Purpose |
 |--------|---------|
 | `buckets_balance.R` | Standalone LPT planner for the 3-host case. Reads per-host wall times from prior runs and prints buckets ready to paste into `wsgs_dispatch.sh --m4-bucket=…`. **Superseded for the N-host orchestrator** — `wsgs_dispatch.sh` now computes the LPT plan inline at dispatch time using the same algorithm. Kept here for one-off planning + cross-checks. Dedups `(wsg, host)` and across hosts before LPT so multi-run CSV accumulation doesn't double-assign WSGs. |
-| `schema_consolidate.R` | pg_dump from M1 + cypher → scp to M4 → pg_restore --data-only. Bucket-aware destination cleanup (DELETEs each source host's WSG bucket from destination tables before restore — avoids duplicate-key violations on re-consolidation). `ok = TRUE` requires pg_restore rc=0 AND post-restore row count > 0; rc=0 with empty schema flags as failure. |
+| `schema_consolidate.R` | Per-table, bucket-filtered `COPY` from each source host over ssh into the dispatcher. Deletes each source's bucket from the destination's persist tables before copying; the three log tables it copies (`log`, `log_input`, `log_recompute`) are history, so on the destination their rows are replaced by primary key and never deleted by group, and a source is refused if a group that has a destination log row has an older or no log row on the source (#354). `ok = TRUE` requires every table's COPY to succeed and the row count to rise. |
 | `runs_archive.sh` | Moves the current top-level `_per_wsg_times.csv` + `*.rds` + `*_annotated.csv` artifacts in `provincial_<bundle>/` to `archive/<TS>/`. Operator cadence: run between provincial runs when you want the LPT planner to use the most recent run only. Skip to median-over multiple recent runs. |
 | `trifecta_smoke.sh` | Thin shim over `wsgs_dispatch.sh` — one small WSG per host (m4→DEAD, m1→ELKR, cyN→ADMS/BABL/BULL). ~3 min wall. Exercises every orchestrator code path (preflight, dispatch, tunnel, RDS pull-back, annotation) before committing to a 200-WSG run. All flags pass through (e.g. `--cy-workspaces=`, `--mapping-code`). |
 

data-raw/habitat_diff_segments.R

@@ -161,7 +161,7 @@ git_sha <- tryCatch(system2("git", c("rev-parse", "--short", "HEAD"), stdout = T
                     error = function(e) NA_character_)
 logrow <- q(sprintf("SELECT run_id, date_end, link_version, link_sha, fresh_version, config_name
                      FROM %s.log WHERE watershed_group_code = %s
-                     ORDER BY date_end DESC LIMIT 1", schema, wsg_lit))
+                     ORDER BY date_end DESC NULLS LAST LIMIT 1", schema, wsg_lit))
 view_n <- q("SELECT count(*) AS n FROM fresh.streams_vw_bcfp")$n
 writeLines(c(
   sprintf("run: %s (UTC)", format(Sys.time(), "%Y-%m-%dT%H:%MZ", tz = "UTC")),

data-raw/logs/consolidate_history_354/README.md

@@ -0,0 +1,33 @@
+# Restoring the log rows #246's consolidate deleted (#354)
+
+2026-10-17, local fwapg, `fresh_default`. Operator call at the #354 plan gate: restore.
+
+Run `20261017_004619-3f4c713b`'s consolidate deleted each cloud-built group's earlier `log` /
+`log_input` rows on the dispatcher. The driver's pre-consolidate backup
+(`/tmp/fresh_default_pre_consolidate_202610162242.dump`, machine-local, taken after m1's own
+groups and before the cloud hosts' were copied in) still held them. Against the current schema
+(an anti-join on each table's primary key) it had exactly **5 `log` rows and 60 `log_input` rows** whose primary key was gone, and an
+empty `log_recompute`:
+
+| group | run_id | built | link |
+|---|---|---|---|
+| DUNC | `m1-20261012T153135.985-b33d3a` | 2026-10-12 | 0.63.1 |
+| KOTL | `m1-20260827T145219.059-18b6fe` | 2026-08-27 | 0.45.1 |
+| LARL | `m1-20260827T144852.009-9d37a9` | 2026-08-27 | 0.45.1 |
+| PINE | `m1-20260806T184954.325-afdf9b` | 2026-08-06 | 0.44.3 |
+| SLOC | `m1-20260827T145652.093-c69f2d` | 2026-08-27 | 0.45.1 |
+
+PINE was not named in the issue; it was built on `cypher-job2` in the run and lost its August
+row the same way. PARS had no earlier row. The later backup (`…_202610162328.dump`, taken
+before the hand consolidate of the 18 groups) has nothing missing in any of the three tables,
+so the 22:42 dump alone was complete.
+
+- `restore_log_rows.sql` → `restore_log_rows.out`: additive `INSERT … ON CONFLICT DO NOTHING`,
+  `log` 241 → 246, `log_input` 2,892 → 2,952, 12 `log_input` rows per restored build.
+- `study_area_verify_after_restore.out`, `verify_extra_after_restore.out`: both pass at
+  `run_uid=20261017_004619-3f4c713b`, `expected_n=241`.
+- `lnk_log_read()`: 241 groups, all 241 newest rows from this run; 246 rows with
+  `latest = FALSE`.
+
+The persist rows of those old builds are not restored (operator, 2026-10-17: no floodplains
+item needs them). The scratch schema `zz354_restore` was dropped after the restore.

data-raw/logs/consolidate_history_354/restore_log_rows.out

@@ -0,0 +1,23 @@
+BEGIN
+   at   | log | log_input 
+--------+-----+-----------
+ before | 241 |      2892
+(1 row)
+
+INSERT 0 5
+INSERT 0 60
+  at   | log | log_input 
+-------+-----+-----------
+ after | 246 |      2952
+(1 row)
+
+ watershed_group_code |            run_id             | date_start | link_version | log_input_rows 
+----------------------+-------------------------------+------------+--------------+----------------
+ DUNC                 | m1-20261012T153135.985-b33d3a | 2026-10-12 | 0.63.1       |             12
+ KOTL                 | m1-20260827T145219.059-18b6fe | 2026-08-27 | 0.45.1       |             12
+ LARL                 | m1-20260827T144852.009-9d37a9 | 2026-08-27 | 0.45.1       |             12
+ PINE                 | m1-20260806T184954.325-afdf9b | 2026-08-06 | 0.44.3       |             12
+ SLOC                 | m1-20260827T145652.093-c69f2d | 2026-08-27 | 0.45.1       |             12
+(5 rows)
+
+COMMIT

data-raw/logs/consolidate_history_354/restore_log_rows.sql

@@ -0,0 +1,46 @@
+-- Restore the log rows run_uid 20261017_004619-3f4c713b's consolidate deleted
+-- from fresh_default (link#354). Source: the driver's pre-consolidate backup
+-- /tmp/fresh_default_pre_consolidate_202610162242.dump (machine-local), its
+-- log / log_input / log_recompute data loaded into zz354_restore first. The
+-- scratch tables are `LIKE fresh_default.<t>`, so the `SELECT *` below is
+-- positional over identical column lists:
+--
+--   CREATE TABLE zz354_restore.<t> (LIKE fresh_default.<t>);
+--   pg_restore --data-only -t <t> -f <t>.sql <dump>
+--   sed 's/^COPY fresh_default\./COPY zz354_restore./' <t>.sql | psql
+--
+-- Before it ran, every one of the 5 log and 60 log_input rows was absent from
+-- fresh_default by primary key (an anti-join, README), so ON CONFLICT skipped
+-- nothing: the INSERT counts below are those numbers.
+--
+-- Additive: a row whose primary key is present is left alone, so re-running
+-- this changes nothing. Columns added since the dump take NULL.
+\set ON_ERROR_STOP on
+BEGIN;
+
+SELECT 'before' AS at,
+       (SELECT count(*) FROM fresh_default.log)       AS log,
+       (SELECT count(*) FROM fresh_default.log_input) AS log_input;
+
+INSERT INTO fresh_default.log
+SELECT * FROM zz354_restore.log
+ON CONFLICT (run_id) DO NOTHING;
+
+INSERT INTO fresh_default.log_input
+SELECT * FROM zz354_restore.log_input
+ON CONFLICT (run_id, table_name) DO NOTHING;
+
+SELECT 'after' AS at,
+       (SELECT count(*) FROM fresh_default.log)       AS log,
+       (SELECT count(*) FROM fresh_default.log_input) AS log_input;
+
+-- Every restored log_input row belongs to a restored log row.
+SELECT l.watershed_group_code, l.run_id, l.date_start::date, l.link_version,
+       (SELECT count(*) FROM fresh_default.log_input i
+         WHERE i.run_id = l.run_id) AS log_input_rows
+  FROM fresh_default.log l
+ WHERE l.run_id IN (SELECT run_id FROM zz354_restore.log)
+   AND l.run_uid IS DISTINCT FROM '20261017_004619-3f4c713b'
+ ORDER BY 1;
+
+COMMIT;