fix(recheck): see image digest changes, never crash on a bad header, and lock the ClinVar refresh

bin/owngenome_recheck.py

@@ -22,6 +22,8 @@
   in-pipeline-release  a newer pin is in a pipeline release: update the pipeline
   upstream-only        upstream has a newer release that no pipeline release pins
   no-newer-release     the lookup answered and nothing is newer
+  unknown              a pin with no upstream lookup that the run's manifest
+                       does not name: nothing was compared
   error                the lookup failed (never read as up to date); exit 1
 The steps an item feeds come from the reads column of assets/steps.tsv, then
 every step downstream of them through its upstream column, then the reports.
@@ -37,7 +39,9 @@
 check named, and lists the named script steps (04b, 14, 29, 03a-03d), which
 skip when their output exists, as not redone. --refresh-clinvar runs
 setup.sh --refresh clinvar first, and refuses while a run of any sample of
-GENOME_DIR is going on, since they share the ClinVar files. Standard library only.
+GENOME_DIR is going on, since they share the ClinVar files. It holds
+GENOME_DIR/clinvar/.refresh.lock while it refreshes: run-all.sh and step 06
+wait on it. Standard library only.
 """
 import argparse
 import fcntl
@@ -66,8 +70,11 @@
 REPORT_STEP = "24"  # the HTML report; the text report is written after it
 # ClinVar's ##fileDate is the release; NCBI's Last-Modified comes a day or two later.
 CLINVAR_SLACK_DAYS = 2
-CLASSES = ("rerun-ready", "refresh-available", "in-pipeline-release", "upstream-only", "no-newer-release", "error")
+CLASSES = ("rerun-ready", "refresh-available", "in-pipeline-release", "upstream-only", "no-newer-release", "unknown",
+           "error")
 LOG_HEADER = "utc\tevent\titem\tclass\tdetail\n"
+# Held exclusive by --refresh-clinvar, shared by scripts/run-all.sh and scripts/06-clinvar-screen.sh.
+CLINVAR_LOCK = "clinvar/.refresh.lock"
 CLINVAR_FILES = ("clinvar/clinvar_pathogenic_chr.vcf.gz", "clinvar/clinvar_chr.vcf.gz", "clinvar/clinvar.vcf.gz")
 # The pinned items: (manifest key, label, freshness lookup, how to read its upstream).
 PINNED = (
@@ -290,7 +297,10 @@ def pgs(self, pgs_id):
         except self.fr.LOOKUP_ERRORS as e:
             raise Failed(str(e))
         import email.utils
-        return email.utils.parsedate_to_datetime(lm).strftime("%Y%m%d")
+        try:  # a malformed date is a failed lookup of this score, never a crash of the whole check
+            return email.utils.parsedate_to_datetime(lm).strftime("%Y%m%d")
+        except (TypeError, ValueError, AttributeError, OverflowError):
+            raise Failed(f"cannot read the Last-Modified of {url}: {lm!r}")
 
     def release(self):
         try:
@@ -334,7 +344,10 @@ def pgs(self, pgs_id):
 
     def release(self):
         e = self._get("release")
-        return e["tag"], dict(e.get("pins") or {})
+        tag, pins = e.get("tag"), e.get("pins") or {}
+        if not isinstance(tag, str) or not tag or not isinstance(pins, dict):
+            raise Failed("the offline answers' release needs a tag and a pins object")
+        return tag, dict(pins)
 
 
 # --- the check -----------------------------------------------------------------------------
@@ -362,6 +375,10 @@ def classify(item, release_tag, release_pins, release_error, clinvar=False):
     ran, inst, up = item.get("ran_with"), item.get("installed"), item.get("upstream")
     if inst and ran and ver(inst) > ver(ran):
         return "rerun-ready"
+    # An image re-pinned at the same tag (a new digest): the tags compare equal, the references do not.
+    ran_ref, inst_ref = item.get("ran_with_ref"), item.get("installed_ref")
+    if inst and ran and ver(inst) == ver(ran) and ran_ref and inst_ref and ran_ref != inst_ref:
+        return "rerun-ready"
     # ClinVar's date is in the manifest whenever step 06 had a file, so a run without one cannot be compared
     # with the file here. A pin with no run value (HLA data not installed then) compares this checkout's.
     if clinvar and not ran:
@@ -391,6 +408,21 @@ def classify(item, release_tag, release_pins, release_error, clinvar=False):
     return "upstream-only"
 
 
+def steps_not_run(sample, sample_dir):
+    """The steps this run did not run: the ones run-all.sh recorded as skipped (no BAM, missing
+    data), and the ones it wrote no row for, which is how it leaves out an opt-in step not in TOOLS
+    (run_state's "not planned"). Empty when logs/run_status.tsv records no step at all (a run of an
+    older version), so nothing is left out on a guess."""
+    try:
+        status = run_state.read_run_status(os.path.join(sample_dir, "logs", "run_status.tsv"))
+        if not (status or {}).get("steps"):
+            return set()
+        return {s["step"] for s in run_state.collect(sample, sample_dir)["steps"]
+                if s["state"] in ("skipped", "not planned")}
+    except (OSError, ValueError, KeyError):
+        return set()
+
+
 def action(item, sample):
     c, key = item["class"], item["item"]
     if c == "rerun-ready":
@@ -415,6 +447,9 @@ def action(item, sample):
         return "no pipeline release pins it yet; nothing to do until one does"
     if c == "error":
         return "the lookup failed; check again later"
+    if c == "unknown":
+        return ("the run's manifest does not name this pin (a run of an older version): nothing to compare; "
+                "the next run records it")
     return ""
 
 
@@ -446,18 +481,31 @@ def add(item, fn, *args, clinvar=False):
         ran = pin_value(key, man.get(("data", key)) if key == "hla_database" else version.get(key))
         inst = this_hla if key == "hla_database" else pin_value(key, this_pins.get(key))
         item = {"item": key, "label": label, "kind": "pin", "ran_with": ran, "installed": inst}
+        if key.endswith("_IMAGE"):
+            item["ran_with_ref"], item["installed_ref"] = version.get(key) or None, this_pins.get(key) or None
         if key == "hla_database" and not this_hla:
             item["error"] = "could not read this checkout's HLA_DB_RELEASE (scripts/lib/common.sh)"
         add(item, answers.pinned, key, lookup, pattern)
     # The other versions.env pins a step reads (steps 50 to 53): no freshness source exists for them, so
-    # nothing is looked up; a pin here that differs from the run's makes its step rerun-ready.
+    # nothing is looked up; a pin here that differs from the run's (an image by its full reference, so a
+    # new digest at the same tag counts) makes its step rerun-ready. A pin the run's manifest lacks is
+    # unknown, never no-newer-release. A pin read only by steps this run did not run (an opt-in step not
+    # in TOOLS, or one skipped for missing data) is left out: a rerun would not run that step either.
     known = {k for k, *_ in PINNED}
+    skipped = steps_not_run(sample, sample_dir)
     for key in dict.fromkeys(r[:-5] if r.endswith("@self") else r for s in steps for r in reads_of(s)):
         if key in known or key not in this_pins:
             continue
-        ran, inst = pin_value(key, version.get(key)), pin_value(key, this_pins.get(key))
+        readers = [s["step"] for s in steps if any(key_matches(key, r) for r in reads_of(s))]
+        if readers and all(n in skipped for n in readers):
+            continue
+        ran_ref, inst_ref = version.get(key) or None, this_pins.get(key) or None
+        ran, inst = pin_value(key, ran_ref), pin_value(key, inst_ref)
+        cls = "unknown" if not (ran_ref and inst_ref) else "rerun-ready" if ran_ref != inst_ref else "no-newer-release"
         item = {"item": key, "label": f"{key} (no upstream lookup)", "kind": "pin", "ran_with": ran,
-                "installed": inst, "class": "rerun-ready" if ran and inst and ran != inst else "no-newer-release"}
+                "installed": inst, "class": cls}
+        if key.endswith("_IMAGE"):
+            item["ran_with_ref"], item["installed_ref"] = ran_ref, inst_ref
         item["steps"] = affected_steps(steps, [key])
         item["action"] = action(item, sample)
         items.append(item)
@@ -520,6 +568,9 @@ def print_check(state, out=None):
     w(f"Database re-check of {state['sample']} ({state['checked_utc']}):\n")
     for i in state["items"]:
         detail = i.get("error") or f"run used {i.get('ran_with') or 'unknown'}, here {i.get('installed') or 'unknown'}, upstream {i.get('upstream') or 'unknown'}"
+        if not i.get("error") and i.get("ran_with_ref") and i.get("installed_ref") and i["ran_with_ref"] != i["installed_ref"] \
+                and i.get("ran_with") == i.get("installed"):
+            detail += f"; same tag, new image: run used {i['ran_with_ref']}, here {i['installed_ref']}"
         w(f"  {i['label']}: {i['class']} ({detail})\n")
         if i["class"] not in ("no-newer-release",) and i.get("action"):
             w(f"    {i['action']}\n")
@@ -624,6 +675,31 @@ def running_samples(genome_dir):
             and run_state.collect(n, os.path.join(genome_dir, n))["run"]["state"] == "running"]
 
 
+def refresh_clinvar(genome_dir):
+    """setup.sh --refresh clinvar under the ClinVar lock of GENOME_DIR (CLINVAR_LOCK). The ClinVar files
+    are shared by every sample, so a run of any of them blocks the refresh: run-all.sh and step 06 hold
+    the lock shared while they read the files, and wait while this holds it. A run with no lock (a bare
+    nextflow run, an older checkout) is caught by its logs/run_status.tsv. The lock is let go before
+    --rerun starts run-all.sh, which takes it shared."""
+    os.makedirs(os.path.join(genome_dir, "clinvar"), exist_ok=True)
+    with open(os.path.join(genome_dir, CLINVAR_LOCK), "a") as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            print(f"ERROR: a run or step 06 holds {os.path.join(genome_dir, CLINVAR_LOCK)}: not refreshing the "
+                  "ClinVar files it reads.", file=sys.stderr)
+            return 3
+        busy = running_samples(genome_dir)
+        if busy:
+            print(f"ERROR: a run of {', '.join(busy)} is going on: not refreshing the ClinVar files it reads.",
+                  file=sys.stderr)
+            return 3
+        r = subprocess.run([os.path.join(ROOT, "scripts", "setup.sh"), "--refresh", "clinvar", genome_dir]).returncode
+        if r != 0:
+            print("ERROR: setup.sh --refresh clinvar failed: see above.", file=sys.stderr)
+        return r
+
+
 def main(argv=None):
     ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
     ap.add_argument("--sample")
@@ -659,16 +735,8 @@ def main(argv=None):
         print(f"ERROR: no {sample_dir}/run_manifest.tsv: run the pipeline on {a.sample} first.", file=sys.stderr)
         return 2
     if a.refresh_clinvar:
-        # The ClinVar files are shared by every sample of GENOME_DIR, so a run of any of them blocks the
-        # refresh. This narrows the window; a run that starts during the refresh is not stopped.
-        busy = running_samples(a.genome_dir)
-        if busy:
-            print(f"ERROR: a run of {', '.join(busy)} is going on: not refreshing the ClinVar files it reads.",
-                  file=sys.stderr)
-            return 3
-        r = subprocess.run([os.path.join(ROOT, "scripts", "setup.sh"), "--refresh", "clinvar", a.genome_dir]).returncode
+        r = refresh_clinvar(a.genome_dir)
         if r != 0:
-            print("ERROR: setup.sh --refresh clinvar failed: see above.", file=sys.stderr)
             return r
     fr = load_freshness()
     this_pins, _ = fr.read_pins(os.path.join(ROOT, "versions.env"))

docs/recheck.md

@@ -23,6 +23,8 @@ Standard library only, so the plain `python3` on the host is enough. It needs th
 | IPD-IMGT/HLA | `hla_database` | The `release_version.txt` of the IPD-IMGT/HLA repository | 08 |
 | Each PGS score | `pgs:<ID>`, its `HmPOS_date` | The PGS Catalog's REST API, then the `Last-Modified` date of the harmonised GRCh38 file | 25, then 26 |
 
+The pins of steps 50 to 53 (`MEGANE_IMAGE`, `MOCHA_PANEL_VERSION`, `MOCHA_IMAGE`, `VNTYPER_IMAGE`, `LOCITYPER_PANEL_VERSION`, `LOCITYPER_IMAGE`) have no upstream lookup: each is compared only with this checkout's `versions.env`. An image is compared by its full reference, so a new `@sha256` digest at the same tag counts as a change, for these and for `PHARMCAT_IMAGE`. A pin read only by steps this run did not run (an opt-in step not in `TOOLS`, or one skipped for missing data) is not listed: a rerun would not run that step either.
+
 The HTML report (24) and the text report are written again after any of them. The ClinVar release inside the PCGR bundle (`cpsr_clinvar_release`) moves with the PCGR bundle, never with step 06's ClinVar file.
 
 The lookups are the ones the monthly CI check uses (`scripts/ci/freshness.py`). It also reads the newest release of this pipeline (`--owngenome-repo`, default `GeiserX/OwnGenome`) and that release's `versions.env`, to say whether a newer pin is already in a release.
@@ -37,11 +39,12 @@ Each item gets one class:
 
 | Class | Meaning | What to do |
 |---|---|---|
-| `rerun-ready` | What is installed here is newer than what the run used: a newer ClinVar file or PGS file on disk, or a pin in this checkout | Rerun (for a pin, run `./scripts/setup.sh` first so its data is there) |
-| `refresh-available` | ClinVar only: NCBI's file is more than 2 days newer than the one installed | `./scripts/setup.sh --refresh clinvar $GENOME_DIR`, then rerun; `--rerun --refresh-clinvar` does both, and refuses while any sample of `$GENOME_DIR` is running, since they share the ClinVar files |
+| `rerun-ready` | What is installed here is newer than what the run used: a newer ClinVar file or PGS file on disk, or a pin in this checkout (an image re-pinned to a new digest at the same tag too) | Rerun (for a pin, run `./scripts/setup.sh` first so its data is there) |
+| `refresh-available` | ClinVar only: NCBI's file is more than 2 days newer than the one installed | `./scripts/setup.sh --refresh clinvar $GENOME_DIR`, then rerun; `--rerun --refresh-clinvar` does both, and refuses while any sample of `$GENOME_DIR` is running, since they share the ClinVar files ([the ClinVar lock](#the-clinvar-lock)) |
 | `in-pipeline-release` | A pipeline release pins a newer version | Update the pipeline to that release, run `setup.sh`, then rerun |
 | `upstream-only` | Upstream has a newer release that no pipeline release pins | Nothing yet for a pinned item; for a PGS score, download it again ([step 25](25-prs.md)) |
 | `no-newer-release` | The lookup answered and nothing is newer | Nothing |
+| `unknown` | A pin of steps 50 to 53 that the run's manifest does not name (a run of an older version): nothing was compared | Nothing; the next run records it. It does not make the command exit 1 |
 | `error` | The lookup failed | Check again later. A failed lookup is never read as "up to date", and the command exits 1 |
 
 The 2-day margin for ClinVar is there because NCBI dates the file a day or two after the release in its `##fileDate`. A newer PGS file date means the catalog wrote the harmonised file again; it does not prove the weights changed.
@@ -67,6 +70,10 @@ When the run ends, it compares the steps that ran a task again with the steps th
 
 A sample last run by an older `run-all.sh` has no `meta command` row: run the same command once by hand.
 
+### The ClinVar lock
+
+The ClinVar files are shared by every sample of `$GENOME_DIR`. `--refresh-clinvar` holds `$GENOME_DIR/clinvar/.refresh.lock` exclusive while `setup.sh --refresh clinvar` replaces them, and lets go before the rerun starts. `run-all.sh` holds the same lock shared from its start to its end, and `scripts/06-clinvar-screen.sh` while it runs: a run that starts during a refresh waits for it ("Waiting for the ClinVar refresh ..."), and a refresh refuses (exit 3) while a run holds the lock. It also refuses while any sample's `logs/run_status.tsv` says a run is going on, which covers a run of an older checkout. The shell side needs `flock` (part of util-linux on Linux; on macOS `brew install flock`); without it nothing waits. A bare `nextflow run`, and `setup.sh --refresh clinvar` run by hand, take no lock.
+
 ### A bare Nextflow run
 
 A bare `nextflow run main.nf` writes no run manifest, so there is nothing to check. To get the command back, `--launch-dir` prints the last command of that launch directory (its `.nextflow/history`) with `-resume` added. It never runs it.
@@ -77,9 +84,10 @@ python3 bin/owngenome_recheck.py --launch-dir /path/to/launch/dir
 
 ## The files it writes
 
-- `<sample>/recheck/state.json`, replaced whole on each check: `checked_utc`, the hosts asked, the pipeline release it read, one entry per item (`item`, `label`, `ran_with`, `installed`, `upstream`, `class`, `steps`, `action`, and `error` or `in_owngenome_release` when they apply), `affected_steps` (the steps a rerun would redo for the `rerun-ready` and `refresh-available` items), and after a rerun a `rerun` block (`started_utc`, `finished_utc`, `rc`, `command`, `predicted_vs_redone`). The last rerun stays until another one replaces it.
+- `<sample>/recheck/state.json`, replaced whole on each check: `checked_utc`, the hosts asked, the pipeline release it read, one entry per item (`item`, `label`, `ran_with`, `installed`, `upstream`, `class`, `steps`, `action`, and `error` or `in_owngenome_release` when they apply; an image item also has `ran_with_ref` and `installed_ref`, the full references with any digest), `affected_steps` (the steps a rerun would redo for the `rerun-ready` and `refresh-available` items), and after a rerun a `rerun` block (`started_utc`, `finished_utc`, `rc`, `command`, `predicted_vs_redone`). The last rerun stays until another one replaces it.
 - `<sample>/recheck/log.tsv`, only ever appended: one row per item and check, and a row as each rerun starts and ends.
 - `<sample>/recheck/rerun.lock`, an empty file each `--rerun` locks while it runs.
+- `$GENOME_DIR/clinvar/.refresh.lock`, an empty file: [the ClinVar lock](#the-clinvar-lock). `run-all.sh` and step 06 make it too when it is missing.
 
 ## In the reports and on the live site
 

renovate.json

@@ -63,6 +63,14 @@
       "extractVersionTemplate": "^v(?<version>.+)$",
       "versioningTemplate": "regex:^(?<major>\\d+)\\.(?<minor>\\d+)\\.(?<patch>\\d+)$"
     },
+    {
+      "customType": "regex",
+      "description": "nf-weblog, the Nextflow plugin of the live site: NF_WEBLOG_PLUGIN in scripts/run-all.sh and the plugins block docs/live-site.md shows, in one PR. Its releases are tagged without a v.",
+      "managerFilePatterns": ["/^scripts/run-all\\.sh$/", "/^docs/live-site\\.md$/"],
+      "matchStrings": ["NF_WEBLOG_PLUGIN=nf-weblog@(?<currentValue>\\d+\\.\\d+\\.\\d+)", "id 'nf-weblog@(?<currentValue>\\d+\\.\\d+\\.\\d+)'"],
+      "depNameTemplate": "nextflow-io/nf-weblog",
+      "datasourceTemplate": "github-releases"
+    },
     {
       "customType": "regex",
       "description": "actionlint, downloaded by version in the lint workflow.",

scripts/06-clinvar-screen.sh

@@ -61,6 +61,18 @@ done
 
 echo "=== ClinVar Pathogenic Screen: ${SAMPLE} ==="
 
+# bin/owngenome_recheck.py --refresh-clinvar holds this lock while it replaces the ClinVar files every sample
+# shares; the screen holds it shared on fd 7 until it ends, so it waits for a refresh going on and a refresh
+# refuses while it reads. Without flock (macOS: brew install flock) nothing waits here.
+CVLOCK="${GENOME_DIR}/clinvar/.refresh.lock"
+if [ -d "${GENOME_DIR}/clinvar" ] && command -v flock >/dev/null 2>&1; then
+  [ -e "$CVLOCK" ] || : 2>/dev/null >> "$CVLOCK" || true
+  if [ -r "$CVLOCK" ]; then
+    exec 7<"$CVLOCK"
+    flock -s -n 7 || { echo "Waiting for the ClinVar refresh that holds ${CVLOCK} to end ..."; flock -s 7; }
+  fi
+fi
+
 for f in "$VCF" "${VCF}.tbi" "$CLINVAR" "${CLINVAR}.tbi" "$REF" "${REF}.fai" "$SITES"; do
   if [ ! -f "$f" ]; then
     echo "ERROR: File not found: ${f}" >&2