fix: silent-failure fixes from the acdc deployment (version normalisation, idempotent indexd, --allow-prod, commons URL)

src/g3dt/cli/_internal/safety.py

@@ -33,18 +33,6 @@ def require_test_env(env: str) -> None:
         raise typer.Exit(2)
 
 
-def abort_if_prod(env: str) -> None:
-    """Hard abort for bulk operations that must never touch production."""
-    if is_prod(env):
-        typer.secho(
-            f"Refusing bulk operation against a production environment "
-            f"('{env}').",
-            fg=typer.colors.RED,
-            err=True,
-        )
-        raise typer.Exit(2)
-
-
 def confirm_destructive(action: str, target: str, env: str, assume_yes: bool) -> None:
     """Gate a destructive operation with an appropriate confirmation.
 

src/g3dt/cli/delete_cmds.py

@@ -12,6 +12,8 @@
 """
 from __future__ import annotations
 
+import re
+
 import typer
 
 from g3dt.cli._internal import dispatch, safety
@@ -22,6 +24,37 @@
 
 _DELETE_METADATA = "services/delete/delete_metadata.sh"
 
+#: A version token in the form the Athena ``version`` column stores it. The
+#: uploader writes ``group(1)`` of this same pattern (metadata_submitter's
+#: ``_find_version_from_path``), i.e. WITHOUT any leading ``v``. The delete
+#: query interpolates the string straight into SQL, so a ``v``-prefixed version
+#: matches zero rows and is reported as "skipped" rather than as an error — a
+#: silent no-op that reads as a clean run. Normalising here closes that.
+_VERSION_RE = re.compile(r"^v?(\d+\.\d+\.\d+)$", re.IGNORECASE)
+
+
+def _normalise_version(raw: str, where: str) -> str:
+    """Canonicalise one version token to the form stored in Athena.
+
+    ``all`` in any case becomes ``all``; ``v1.5.4`` and ``1.5.4`` both become
+    ``1.5.4``. Anything else is a usage error: the column only ever holds
+    three-part semver, so a truncated version like ``0.9`` would match nothing
+    and be counted as a skip.
+    """
+    token = raw.strip()
+    if token.lower() == "all":
+        return "all"
+    match = _VERSION_RE.match(token)
+    if not match:
+        typer.secho(
+            f"Invalid version '{raw}' {where}: expected x.y.z (e.g. 0.9.8) "
+            "or 'all'.",
+            fg=typer.colors.RED,
+            err=True,
+        )
+        raise typer.Exit(2)
+    return match.group(1)
+
 
 @app.command()
 def metadata(
@@ -54,10 +87,12 @@ def metadata(
         )
         raise typer.Exit(2)
 
+    version = _normalise_version(version, "for --version")
+
     names = [s.strip() for s in studies.split(",") if s.strip()]
     keys = [study_of(name, env).key for name in names]
     target = ",".join(keys)
-    all_versions = version.strip().lower() == "all"
+    all_versions = version == "all"
 
     if all_versions:
         # Deleting every version is the most destructive path: always prompt

src/g3dt/cli/indexd_cmds.py

@@ -27,10 +27,18 @@ def register(
     dry_run: bool = typer.Option(
         False, "--dry-run", help="Scan + write file_metadata only; skip indexd."
     ),
+    force: bool = typer.Option(
+        False, "--force",
+        help="Re-register files already in the registry with the same md5.",
+    ),
     on: Target = typer.Option(Target.local, "--on", help="Run local or on ec2."),
 ) -> None:
     """Scan S3 prefixes and register the files with Gen3 indexd.
 
+    Files already registered for this study at this endpoint with an unchanged
+    md5 are skipped (each re-registration would create a new indexd revision
+    and duplicate the registry). ``--force`` re-registers everything.
+
     Examples:
       g3dt indexd register --s3-paths s3://bucket/edcad/ --study edcad --env staging
       g3dt indexd register --s3-paths s3://b/a/ --s3-paths s3://b/c/ --study edcad --env staging --on ec2
@@ -41,6 +49,8 @@ def build_args(env_name):
         a = ["--s3-paths", *s3_paths, "--study", s.key, "--env", env_name]
         if dry_run:
             a.append("--dry-run")
+        if force:
+            a.append("--force")
         return a
 
     def remote_cli(env_name):
@@ -50,6 +60,8 @@ def remote_cli(env_name):
         a += ["--study", study, "--env", env_name]
         if dry_run:
             a.append("--dry-run")
+        if force:
+            a.append("--force")
         return a
 
     dispatch.run_or_dispatch(

src/g3dt/cli/metadata.py

@@ -7,7 +7,7 @@
 
 import typer
 
-from g3dt.cli._internal import dispatch
+from g3dt.cli._internal import dispatch, safety
 from g3dt.cli._internal.dispatch import Target
 from g3dt.cli._internal.resolve import study_of
 
@@ -55,20 +55,55 @@ def upload_all(
         ..., "--studies", help="Comma-separated studies, e.g. ausdiab,caughtcad."
     ),
     env: str = typer.Option(..., "--env", "-e", help="Environment, e.g. test."),
+    allow_prod: bool = typer.Option(
+        False, "--allow-prod",
+        help="Allow bulk upload against production (typed confirmation required).",
+    ),
+    prod_confirmed: bool = typer.Option(
+        False, "--prod-confirmed", hidden=True,
+        help="Internal: set by the remote re-entry after the typed "
+        "confirmation already happened locally. Never pass by hand.",
+    ),
     on: Target = typer.Option(Target.local, "--on", help="Run local or on ec2."),
 ) -> None:
     """Upload several studies sequentially (wraps upload_all_studies.sh).
 
-    The wrapped script aborts on any 'prod' environment.
+    Production needs ``--allow-prod`` AND a typed confirmation of the env
+    name. The confirmation happens locally, before any EC2 dispatch (SSM has
+    no TTY, so a remote prompt would abort); the remote re-entry carries the
+    hidden ``--prod-confirmed`` marker instead of re-prompting, and
+    ``--allow-prod`` is forwarded so the wrapped script's own guard passes.
     """
     names = [s.strip() for s in studies.split(",") if s.strip()]
     keys = [study_of(name, env).key for name in names]
 
+    # Prod is detected on the resolved study keys as well as on --env:
+    # `--env staging --studies ausdiab_prod` is a production write.
+    if safety.is_prod(env) or any(safety.is_prod(k) for k in keys):
+        if not allow_prod:
+            typer.secho(
+                "Refusing bulk upload against a production environment. "
+                "Re-run with --allow-prod to confirm interactively.",
+                fg=typer.colors.RED,
+                err=True,
+            )
+            raise typer.Exit(2)
+        if not prod_confirmed:
+            safety.confirm_prod_strict("bulk metadata upload", env)
+
     def build_args(env_name):
-        return ["--studies", ",".join(keys), "--env", env_name]
+        a = ["--studies", ",".join(keys), "--env", env_name]
+        if allow_prod:
+            a.append("--allow-prod")
+        return a
 
     def remote_cli(env_name):
-        return ["metadata", "upload-all", "--studies", studies, "--env", env_name]
+        a = ["metadata", "upload-all", "--studies", studies, "--env", env_name]
+        if allow_prod:
+            # The typed confirmation already happened locally above; the box
+            # has no TTY, so the re-entry must not prompt again.
+            a += ["--allow-prod", "--prod-confirmed"]
+        return a
 
     dispatch.run_or_dispatch(
         on, env, _UPLOAD_ALL, build_args, "metadata-upload-all",

src/g3dt/indexd/indexd_registrar.py

@@ -115,17 +115,108 @@ def scan_s3_files(
 
 # ---------- indexd registration ----------
 
+def fetch_registered_files(
+    database: str,
+    table: str,
+    study_id: str,
+    indexd_endpoint: str,
+    athena_s3_output: str,
+    workgroup: str = "primary",
+    boto3_session: Optional[boto3.Session] = None,
+) -> pd.DataFrame:
+    """Return the ``file_name``/``md5`` pairs already registered for a study.
+
+    Scoped to one study and one indexd endpoint so a staging registration can
+    never mask a missing prod one (and vice versa).
+
+    A missing table means nothing has ever been registered for this
+    environment, which is a normal first-run state, not an error — an empty
+    frame is returned so the caller registers everything.
+
+    Returns
+    -------
+    pd.DataFrame
+        Columns ``file_name`` and ``md5``; empty if the table does not exist.
+    """
+    sql = f"""
+        SELECT DISTINCT file_name, md5
+        FROM "{database}"."{table}"
+        WHERE study_id = '{study_id}'
+          AND indexd_endpoint = '{indexd_endpoint}'
+    """
+    try:
+        return wr.athena.read_sql_query(
+            sql,
+            database=database,
+            ctas_approach=False,
+            workgroup=workgroup,
+            s3_output=athena_s3_output,
+            boto3_session=boto3_session,
+        )
+    except Exception as exc:
+        logger.warning(
+            "Could not read existing registrations from %s.%s (%s). "
+            "Treating every scanned file as unregistered.",
+            database, table, exc,
+        )
+        return pd.DataFrame(columns=["file_name", "md5"])
+
+
+def filter_unregistered(
+    df: pd.DataFrame,
+    registered: pd.DataFrame,
+) -> pd.DataFrame:
+    """Drop scanned files already registered with the same name and md5.
+
+    Re-submitting a file to indexd does not overwrite it: because ``baseid``
+    is derived from the filename, indexd creates a *new revision with a new
+    did* every time. Those revisions accumulate in the registry table (which
+    merges on ``did``), so an unfiltered re-run doubles the corpus and forces
+    every downstream join to de-duplicate. Skipping files whose content has
+    not changed makes a re-run a cheap no-op.
+
+    An md5 that differs from the registered one means the file genuinely
+    changed, so it is *not* skipped — that is exactly when a new revision is
+    wanted.
+
+    Parameters
+    ----------
+    df : pd.DataFrame
+        Scanned files; must have ``file_name`` and ``md5``.
+    registered : pd.DataFrame
+        Existing registrations (see :func:`fetch_registered_files`).
+
+    Returns
+    -------
+    pd.DataFrame
+        The subset of *df* still needing registration.
+    """
+    if df.empty or registered.empty:
+        return df
+
+    already = set(
+        zip(registered["file_name"].astype(str), registered["md5"].astype(str))
+    )
+    keep = [
+        (str(name), str(md5)) not in already
+        for name, md5 in zip(df["file_name"], df["md5"])
+    ]
+    return df[pd.Series(keep, index=df.index)]
+
+
 def register_files_with_indexd(
     index: Gen3Index,
     df: pd.DataFrame,
     authz: List[str],
 ) -> pd.DataFrame:
     """Register files with Gen3 indexd and return results.
 
-    For each row in *df*, calls ``Gen3Index.create_record`` with the
-    file's hashes, size, S3 URL, baseid, and authz.  Re-submitting
-    the same baseid will create a new revision in indexd, making this
-    safe to re-run (idempotent at the API level).
+    For each row in *df*, calls ``Gen3Index.create_record`` with the file's
+    hashes, size, S3 URL, baseid, and authz. Registers exactly what it is
+    given: re-submitting a baseid creates a NEW revision with a new ``did``
+    (indexd never overwrites), so callers must pre-filter already-registered
+    files with :func:`fetch_registered_files` + :func:`filter_unregistered`
+    unless duplicate revisions are intended.
 
     Parameters
     ----------

src/g3dt/services/indexd/register_indexd.py

@@ -30,6 +30,8 @@
 )
 from g3dt.indexd.indexd_registrar import (
     scan_s3_files,
+    fetch_registered_files,
+    filter_unregistered,
     register_files_with_indexd,
     write_to_glue,
 )
@@ -81,6 +83,12 @@ def main():
         help="Scan and write file_metadata only; skip indexd "
         "registration",
     )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="Re-register files even if already recorded in the "
+        "registry with the same md5",
+    )
 
     args = parser.parse_args()
 
@@ -186,6 +194,36 @@ def main():
         logger.info("Dry run — skipping indexd registration.")
         sys.exit(0)
 
+    # --- Skip files already registered at this endpoint (same name + md5) ---
+    # Re-submitting creates a NEW indexd revision per run (baseid never
+    # overwrites), so an unfiltered re-run duplicates the corpus. A changed
+    # md5 is deliberately NOT skipped — that is when a new revision is wanted.
+    if args.force:
+        logger.info(
+            "--force given: re-registering all %d scanned file(s).",
+            len(file_df),
+        )
+    else:
+        registered = fetch_registered_files(
+            database=reg_database,
+            table=reg_table,
+            study_id=args.study,
+            indexd_endpoint=indexd_endpoint,
+            athena_s3_output=athena_s3_output,
+            workgroup=workgroup,
+            boto3_session=session,
+        )
+        before = len(file_df)
+        file_df = filter_unregistered(file_df, registered)
+        logger.info(
+            "Skipping %d / %d file(s) already registered with the same md5 "
+            "(use --force to re-register).",
+            before - len(file_df), before,
+        )
+        if file_df.empty:
+            logger.info("Nothing new to register. Done.")
+            sys.exit(0)
+
     auth = Gen3Auth(refresh_token=api_key)
     index = Gen3Index(auth)