Align history-audit trigger SQL between migration and runtime installer

mvdbeek · mvdbeek · commit 4ba4a9033cc6 · 2026-04-17T15:26:57.000+02:00
The follow-up pg_notify migration b8d5e2f9a1c7 and runtime installer
update_audit_table.py carried two independent copies of the plpgsql
trigger function body, and disagreed on the STATEMENT-vs-ROW threshold:

- c716ee82337b (last trigger migration) used `version &gt; 10`, so on PG 10
  it installed a ROW trigger + ROW-variant function.
- b8d5e2f9a1c7 used `&gt;= 10`, so on PG 10 its upgrade() overwrote the
  function body with the STATEMENT variant (references `new_table`)
  while the trigger left behind by c716ee82337b was still ROW — the
  next audit write would raise "missing FROM-clause entry for
  new_table".
- update_audit_table.py (fresh install) used `&gt;= 10`, which is
  internally consistent for fresh installs but diverges from the
  upgrade path.

Extract `build_trigger_fn(function_name, id_field, use_statement,
with_notify)` and `use_statement_trigger(version)` from
update_audit_table.py, and have the migration import them. Now both
paths share one SQL body and one predicate, and the predicate matches
c716ee82337b so upgraded PG 10 databases keep a consistent ROW
trigger + ROW-variant function. (PG 10 is long past EOL, but drift
between runtime install and migration is exactly the class of bug
this migration was written to repair.)
diff --git a/lib/galaxy/model/migrations/alembic/versions_gxy/b8d5e2f9a1c7_add_pg_notify_to_history_audit_triggers.py b/lib/galaxy/model/migrations/alembic/versions_gxy/b8d5e2f9a1c7_add_pg_notify_to_history_audit_triggers.py
@@ -14,6 +14,10 @@
 affected history id, matching `model/triggers/update_audit_table.py` used on
 fresh installs.
 
+The STATEMENT-vs-ROW decision must match the trigger DEFINITION installed by
+`c716ee82337b` so the function body references the right context (`new_table`
+vs `NEW`); both use `version > 10` (and treat offline mode as STATEMENT).
+
 SQLite installations use the poll-only path and require no change.
 """
 
@@ -23,16 +27,18 @@
     _is_sqlite,
     transaction,
 )
+from galaxy.model.triggers.update_audit_table import (
+    build_trigger_fn,
+    fn_prefix,
+    use_statement_trigger,
+)
 
 revision = "b8d5e2f9a1c7"
 down_revision = "f5e9e4bca542"
 branch_labels = None
 depends_on = None
 
 
-CHANNEL = "galaxy_history_update"
-
-
 def upgrade():
     if _is_sqlite():
         return
@@ -47,58 +53,10 @@ def downgrade():
         _install_functions(with_notify=False)
 
 
-def _install_functions(with_notify: bool):
+def _install_functions(with_notify: bool) -> None:
     version_info = op.get_bind().engine.dialect.server_version_info
-    use_statement = version_info is None or version_info[0] >= 10
-    builder = _statement_trigger_fn if use_statement else _row_trigger_fn
+    # Offline mode (no live connection) matches c716ee82337b: assume STATEMENT.
+    statement = version_info is None or use_statement_trigger(version_info[0])
     for id_field in ("history_id", "id"):
-        op.execute(builder(f"fn_audit_history_by_{id_field}", id_field, with_notify))
-
-
-def _statement_trigger_fn(function_name: str, id_field: str, with_notify: bool) -> str:
-    notify_block = (
-        f"""
-                FOR _history_id IN SELECT DISTINCT {id_field} FROM new_table WHERE {id_field} IS NOT NULL
-                LOOP
-                    PERFORM pg_notify('{CHANNEL}', _history_id::text);
-                END LOOP;
-        """
-        if with_notify
-        else ""
-    )
-    declare_block = "DECLARE _history_id integer;" if with_notify else ""
-    return f"""
-        CREATE OR REPLACE FUNCTION {function_name}()
-            RETURNS TRIGGER
-            LANGUAGE 'plpgsql'
-        AS $BODY$
-            {declare_block}
-            BEGIN
-                INSERT INTO history_audit (history_id, update_time)
-                SELECT DISTINCT {id_field}, clock_timestamp() AT TIME ZONE 'UTC'
-                FROM new_table
-                WHERE {id_field} IS NOT NULL
-                ON CONFLICT DO NOTHING;
-                {notify_block}
-                RETURN NULL;
-            END;
-        $BODY$
-    """
-
-
-def _row_trigger_fn(function_name: str, id_field: str, with_notify: bool) -> str:
-    notify_stmt = f"PERFORM pg_notify('{CHANNEL}', NEW.{id_field}::text);" if with_notify else ""
-    return f"""
-        CREATE OR REPLACE FUNCTION {function_name}()
-            RETURNS TRIGGER
-            LANGUAGE 'plpgsql'
-        AS $BODY$
-            BEGIN
-                INSERT INTO history_audit (history_id, update_time)
-                VALUES (NEW.{id_field}, clock_timestamp() AT TIME ZONE 'UTC')
-                ON CONFLICT DO NOTHING;
-                {notify_stmt}
-                RETURN NULL;
-            END;
-        $BODY$
-    """
+        fn_name = f"{fn_prefix}_{id_field}"
+        op.execute(build_trigger_fn(fn_name, id_field, use_statement=statement, with_notify=with_notify))
diff --git a/lib/galaxy/model/triggers/update_audit_table.py b/lib/galaxy/model/triggers/update_audit_table.py
@@ -3,6 +3,9 @@
 # function name prefix
 fn_prefix = "fn_audit_history_by"
 
+# channel used by pg_notify so HistoryAuditMonitor can LISTEN for updates
+NOTIFY_CHANNEL = "galaxy_history_update"
+
 # map between source table and associated incoming id field
 trigger_config = {
     "history_dataset_association": "history_id",
@@ -11,6 +14,68 @@
 }
 
 
+def use_statement_trigger(version: int) -> bool:
+    """Return True when the postgres version supports the STATEMENT variant.
+
+    Fresh installs and the pg_notify migration share this predicate to ensure
+    the trigger function body (STATEMENT references new_table, ROW references NEW)
+    matches the trigger definition installed at that version.
+    """
+    return version > 10
+
+
+def build_trigger_fn(function_name: str, id_field: str, *, use_statement: bool, with_notify: bool = True) -> str:
+    """Build the plpgsql CREATE OR REPLACE FUNCTION body for an audit trigger.
+
+    Shared between runtime install (update_audit_table.install) and alembic
+    migrations so the two cannot drift.
+    """
+    if use_statement:
+        notify_block = (
+            f"""
+                    FOR _history_id IN SELECT DISTINCT {id_field} FROM new_table WHERE {id_field} IS NOT NULL
+                    LOOP
+                        PERFORM pg_notify('{NOTIFY_CHANNEL}', _history_id::text);
+                    END LOOP;
+            """
+            if with_notify
+            else ""
+        )
+        declare_block = "DECLARE _history_id integer;" if with_notify else ""
+        return f"""
+            CREATE OR REPLACE FUNCTION {function_name}()
+                RETURNS TRIGGER
+                LANGUAGE 'plpgsql'
+            AS $BODY$
+                {declare_block}
+                BEGIN
+                    INSERT INTO history_audit (history_id, update_time)
+                    SELECT DISTINCT {id_field}, clock_timestamp() AT TIME ZONE 'UTC'
+                    FROM new_table
+                    WHERE {id_field} IS NOT NULL
+                    ON CONFLICT DO NOTHING;
+                    {notify_block}
+                    RETURN NULL;
+                END;
+            $BODY$
+        """
+    notify_stmt = f"PERFORM pg_notify('{NOTIFY_CHANNEL}', NEW.{id_field}::text);" if with_notify else ""
+    return f"""
+        CREATE OR REPLACE FUNCTION {function_name}()
+            RETURNS TRIGGER
+            LANGUAGE 'plpgsql'
+        AS $BODY$
+            BEGIN
+                INSERT INTO history_audit (history_id, update_time)
+                VALUES (NEW.{id_field}, clock_timestamp() AT TIME ZONE 'UTC')
+                ON CONFLICT DO NOTHING;
+                {notify_stmt}
+                RETURN NULL;
+            END;
+        $BODY$
+    """
+
+
 def install(engine):
     """Install history audit table triggers"""
     sql = _postgres_install(engine) if "postgres" in engine.name else _sqlite_install()
@@ -41,54 +106,6 @@ def _postgres_install(engine):
 
     sql = []
 
-    # PostgreSQL trigger function template
-    # need to make separate functions purely because the incoming history_id field name will be
-    # different for different source tables. There may be a fancier way to dynamically choose
-    # between incoming fields, but having 2 triggers fns seems straightforward
-
-    def statement_trigger_fn(id_field):
-        fn = f"{fn_prefix}_{id_field}"
-
-        return f"""
-            CREATE OR REPLACE FUNCTION {fn}()
-                RETURNS TRIGGER
-                LANGUAGE 'plpgsql'
-            AS $BODY$
-                DECLARE
-                    _history_id integer;
-                BEGIN
-                    INSERT INTO history_audit (history_id, update_time)
-                    SELECT DISTINCT {id_field}, clock_timestamp() AT TIME ZONE 'UTC'
-                    FROM new_table
-                    WHERE {id_field} IS NOT NULL
-                    ON CONFLICT DO NOTHING;
-                    FOR _history_id IN SELECT DISTINCT {id_field} FROM new_table WHERE {id_field} IS NOT NULL
-                    LOOP
-                        PERFORM pg_notify('galaxy_history_update', _history_id::text);
-                    END LOOP;
-                    RETURN NULL;
-                END;
-            $BODY$
-        """
-
-    def row_trigger_fn(id_field):
-        fn = f"{fn_prefix}_{id_field}"
-
-        return f"""
-            CREATE OR REPLACE FUNCTION {fn}()
-                RETURNS TRIGGER
-                LANGUAGE 'plpgsql'
-            AS $BODY$
-                BEGIN
-                    INSERT INTO history_audit (history_id, update_time)
-                    VALUES (NEW.{id_field}, clock_timestamp() AT TIME ZONE 'UTC')
-                    ON CONFLICT DO NOTHING;
-                    PERFORM pg_notify('galaxy_history_update', NEW.{id_field}::text);
-                    RETURN NULL;
-                END;
-            $BODY$
-        """
-
     def trigger_def(source_table: str, id_field: str, operation: str, version: int, when: str = "AFTER") -> str:
         fn = f"{fn_prefix}_{id_field}"
         # PostgreSQL supports many triggers per operation/table so the label can
@@ -100,7 +117,7 @@ def trigger_def(source_table: str, id_field: str, operation: str, version: int,
         # The use of the keyword PROCEDURE here is historical and deprecated (https://www.postgresql.org/docs/11/sql-createtrigger.html).
         function_keyword = "FUNCTION" if version >= 11 else "PROCEDURE"
         create_or_replace = "CREATE OR REPLACE" if version >= 14 else "CREATE"
-        if version >= 10 and when == "AFTER":
+        if use_statement_trigger(version) and when == "AFTER":
             return f"""
                 {create_or_replace} TRIGGER {trigger_name}
                 AFTER {operation}
@@ -121,10 +138,11 @@ def trigger_def(source_table: str, id_field: str, operation: str, version: int,
 
     # pick row or statement triggers depending on postgres version
     version = engine.dialect.server_version_info[0]
-    trigger_fn = statement_trigger_fn if version >= 10 else row_trigger_fn
+    statement = use_statement_trigger(version)
 
     for id_field in ["history_id", "id"]:
-        sql.append(trigger_fn(id_field))
+        fn_name = f"{fn_prefix}_{id_field}"
+        sql.append(build_trigger_fn(fn_name, id_field, use_statement=statement, with_notify=True))
 
     for source_table, id_field in trigger_config.items():
         for operation in ["UPDATE", "INSERT"]:
