Scope existence preconditions to their table; lint against name-only checks

A foreignKeyConstraintExists precondition with only foreignKeyName (or
indexExists with only indexName) forces Liquibase to snapshot the entire
schema to answer the check -- ~2s per precondition on a large postgres
appdb. 27 such preconditions (24 FK, 3 index; 21 in one v59 workspace
block) added ~46s to every full migration: 50.2s of total changeset time,
of which 24 changesets cost 45.9s and the other 1,182 cost 3.7s.

Add the scoping key to all 27. Measured on a fresh postgres migration:
total changeset time drops 50.2s -> 5.1s, with zero changesets over 1s.

Editing released changesets' preconditions is safe: Liquibase checksums
cover the changes, not preConditions. Verified two ways -- ChangeSet
.generateCheckSum is byte-identical before/after for all 27 (v9
checksums), and a clone of an already-migrated appdb re-validates
cleanly against the edited files.

lint-migrations-file: generalize require-primary-key-exists-has-table-name
into require-table-scoped-existence-preconditions, covering
primaryKeyExists/indexExists/uniqueConstraintExists (tableName) and
foreignKeyConstraintExists (foreignKeyTableName), walking the full
precondition tree in both its flat-map and list forms.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01G7FfdpzW5pr7CX9aNwa98f

.clj-kondo/config.edn

@@ -465,6 +465,7 @@
   metabase.actions.test-util/with-actions                                               clojure.core/let
   metabase.transforms.test-util/with-transform-cleanup!                                 clojure.core/let
   metabase.api.common/let-404                                                           clojure.core/let
+  metabase.app-db.connection/with-unshared-connection                                  clojure.core/fn
   metabase.app-db.custom-migrations.util/with-temp-scheduler!                           clojure.core/fn
   metabase.app-db.custom-migrations/define-migration                                    clj-kondo.lint-as/def-catch-all
   metabase.app-db.custom-migrations/define-reversible-migration                         clj-kondo.lint-as/def-catch-all

bin/lint-migrations-file/src/lint_migrations_file.cljc

@@ -207,16 +207,42 @@
                         (str/join ", " bad-ids))
                 {:invalid-ids (vec bad-ids)}))))))
 
-(defn require-primary-key-exists-has-table-name
-  "Ensures that all primaryKeyExists preconditions specify a tableName."
+(def ^:private table-scoped-preconditions
+  "Existence preconditions mapped to the key that scopes them to a single table. Without that key,
+  Liquibase answers the check by snapshotting the whole schema — ~2 seconds per precondition on a
+  large postgres appdb, which is how 27 name-only preconditions once added ~50s to every full
+  migration."
+  {:primaryKeyExists           :tableName
+   :indexExists                :tableName
+   :uniqueConstraintExists     :tableName
+   :foreignKeyConstraintExists :foreignKeyTableName})
+
+(defn- unscoped-preconditions
+  "Walk a preConditions tree (maps nested under and/or/not, in either flat-map or list form) and
+  return [precondition-key required-key] pairs for every table-scoped existence check that is
+  missing its table key."
+  [form]
+  (cond
+    (map? form)        (concat
+                        (for [[precondition required] table-scoped-preconditions
+                              :let [check (get form precondition)]
+                              :when (and (map? check) (not (contains? check required)))]
+                          [precondition required])
+                        (mapcat unscoped-preconditions (vals form)))
+    (sequential? form) (mapcat unscoped-preconditions form)
+    :else              nil))
+
+(defn require-table-scoped-existence-preconditions
+  "Ensures that primaryKeyExists/indexExists/uniqueConstraintExists preconditions specify a
+  tableName, and foreignKeyConstraintExists a foreignKeyTableName."
   [change-log]
   (doseq [{:keys [changeSet]} change-log
           :when changeSet
-          :let [s (pr-str (:preConditions changeSet))]
-          :when (and (str/includes? s ":primaryKeyExists")
-                     (not (str/includes? s ":tableName")))]
+          :let [[precondition required] (first (unscoped-preconditions (:preConditions changeSet)))]
+          :when precondition]
     (throw (validation-error
-            (format "Migration '%s' has primaryKeyExists precondition without tableName" (:id changeSet))
+            (format "Migration '%s' has %s precondition without %s: without it Liquibase snapshots the entire schema to answer the check (~2s per precondition on a large postgres appdb)"
+                    (:id changeSet) (name precondition) (name required))
             {:id (:id changeSet)}))))
 
 (s/def ::changeSet
@@ -292,7 +318,7 @@
   (require-no-bare-blob-or-text-types change-log)
   (require-no-bare-boolean-types change-log file)
   (require-no-datetime-type change-log file)
-  (require-primary-key-exists-has-table-name change-log)
+  (require-table-scoped-existence-preconditions change-log)
   (let [{:keys [changeSet]
          :as all} (group-by only-key change-log)]
     (when-not (set/subset? (set (keys all)) #{:property :objectQuotingStrategy :changeSet :logicalFilePath})

bin/lint-migrations-file/test/lint_migrations_file_test.clj

@@ -398,3 +398,40 @@
                                                                                   :remarks "none"
                                                                                   :type "timestamp with time zone"}}]}}]
                                      :rollback nil))))))
+
+(deftest ^:parallel require-table-scoped-existence-preconditions-test
+  (testing "foreignKeyConstraintExists without foreignKeyTableName is rejected"
+    (is-thrown-with-error-info?
+     "Migration 'v49.2024-01-01T10:30:00' has foreignKeyConstraintExists precondition without foreignKeyTableName: without it Liquibase snapshots the entire schema to answer the check (~2s per precondition on a large postgres appdb)"
+     {:id "v49.2024-01-01T10:30:00"}
+     (validate
+      (mock-change-set
+       :preConditions [{:onFail "MARK_RAN"}
+                       {:not [{:foreignKeyConstraintExists {:foreignKeyName "fk_my_table_other_id"}}]}]))))
+  (testing "foreignKeyConstraintExists with foreignKeyTableName passes"
+    (is (= :ok
+           (validate
+            (mock-change-set
+             :preConditions [{:onFail "MARK_RAN"}
+                             {:not [{:foreignKeyConstraintExists {:foreignKeyName      "fk_my_table_other_id"
+                                                                  :foreignKeyTableName "my_table"}}]}])))))
+  (testing "indexExists without tableName is rejected, including in flat-map not form"
+    (is-thrown-with-error-info?
+     "Migration 'v49.2024-01-01T10:30:00' has indexExists precondition without tableName: without it Liquibase snapshots the entire schema to answer the check (~2s per precondition on a large postgres appdb)"
+     {:id "v49.2024-01-01T10:30:00"}
+     (validate
+      (mock-change-set
+       :preConditions [{:not {:indexExists {:indexName "idx_my_table_other_id"}}}]))))
+  (testing "indexExists with tableName passes"
+    (is (= :ok
+           (validate
+            (mock-change-set
+             :preConditions [{:not {:indexExists {:indexName "idx_my_table_other_id"
+                                                  :tableName "my_table"}}}])))))
+  (testing "primaryKeyExists without tableName is still rejected (previous rule subsumed)"
+    (is-thrown-with-error-info?
+     "Migration 'v49.2024-01-01T10:30:00' has primaryKeyExists precondition without tableName: without it Liquibase snapshots the entire schema to answer the check (~2s per precondition on a large postgres appdb)"
+     {:id "v49.2024-01-01T10:30:00"}
+     (validate
+      (mock-change-set
+       :preConditions [{:primaryKeyExists {:primaryKeyName "pk_my_table"}}])))))

deps.edn

@@ -80,7 +80,7 @@
   hiccup/hiccup                             {:mvn/version "2.0.0"}              ; HTML templating
   inflections/inflections                   {:mvn/version "0.15.0"}             ; Clojure/Script library used for prularizing words
   instaparse/instaparse                     {:mvn/version "1.5.0"}              ; Make your own parser
-  io.github.camsaul/toucan2                 {:mvn/version "1.0.571"}
+  io.github.camsaul/toucan2                 {:mvn/version "1.0.572"}
   io.github.eerohele/pp                     {#_#_:git/tag "2024-11-13.77"       ; super fast pretty-printing library
                                              :git/sha "ffc4edc482c057cd9412c62f018a41e9140c618b"
                                              :git/url "https://github.com/eerohele/pp"}

enterprise/backend/src/metabase_enterprise/audit_app/audit.clj

@@ -338,9 +338,7 @@
      - the `instance_analytics_views` SQL files have changed since the last successful sync,
        meaning a migration may have added a new view that isn't yet in `metabase_table`.
    The two triggers share one sync because they both want the same operation. Runs synchronously
-   (not in a background future) so it stays inside the caller's cross-node cluster lock and
-   transaction — a sync on another thread would escape the lock and, on a transactional appdb,
-   deadlock against the caller's uncommitted `metabase_table` writes."
+   (not in a background future) so it stays serialized under the caller's cross-node cluster lock."
   [audit-db engine-changed?]
   (let [current      (views-checksum)
         views-stale? (and current (not= current (audit-app.settings/last-analytics-views-checksum)))]
@@ -478,11 +476,19 @@
   ;; serialize install+adjust+load+sync+reconcile across nodes so a rolling upgrade can't run an adjust or sync
   ;; against a half-adjusted schema (`with-duplicate-ops-prevented` is per-process only). The install runs inside the
   ;; lock too, so a node that acquires it after the installer committed sees the DB already exists and falls through
-  ;; to a no-op rather than colliding on the audit DB primary key. The sync runs synchronously inside the lock so it
-  ;; stays in the lock's transaction. If another node already holds the lock it is doing this same work, so we skip
-  ;; rather than fail the boot.
+  ;; to a no-op rather than colliding on the audit DB primary key. The sync runs synchronously so it stays serialized
+  ;; under the lock. If another node already holds the lock it is doing this same work, so we skip rather than fail
+  ;; the boot.
+  ;;
+  ;; The lock is :detached? so the pipeline does NOT run as one long transaction on the lock's connection: the serdes
+  ;; load's per-entity transactions commit incrementally (restoring #74412's per-entity isolation) instead of piling
+  ;; up thousands of savepoints/subtransaction XIDs and holding every row lock until one big commit at the end —
+  ;; which stalled first boot for many minutes and blocked concurrent work (#78238). Safe because this whole pipeline
+  ;; is idempotent and self-healing: the checksum only advances on completion and reconcile runs every boot, so a
+  ;; crash mid-pipeline is repaired by the next boot.
   (try
-    (cluster-lock/with-cluster-lock {:lock audit-db-cluster-lock :timeout-seconds 5 :retry-config {:max-retries 2}}
+    (cluster-lock/with-cluster-lock {:lock audit-db-cluster-lock :timeout-seconds 5 :retry-config {:max-retries 2}
+                                     :detached? true}
       (u/prog1 (maybe-install-audit-db!)
         (when-let [audit-db (t2/select-one :model/Database :is_audit true)]
           ((sync-util/with-duplicate-ops-prevented

resources/migrations/001_update_migrations.yaml

@@ -11733,6 +11733,7 @@ databaseChangeLog:
         - not:
             - indexExists:
                 indexName: idx_security_advisory_acknowledged_by
+                tableName: security_advisory
       changes:
         - createIndex:
             tableName: security_advisory
@@ -11750,6 +11751,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_security_advisory_acknowledged_by
+                foreignKeyTableName: security_advisory
       changes:
         - addForeignKeyConstraint:
             baseTableName: security_advisory
@@ -11768,6 +11770,7 @@ databaseChangeLog:
         - not:
             - indexExists:
                 indexName: idx_security_advisory_match_status
+                tableName: security_advisory
       changes:
         - createIndex:
             tableName: security_advisory

resources/migrations/057_update_migrations.yaml

@@ -2240,6 +2240,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_document_made_public_by_id
+                foreignKeyTableName: document
       changes:
         - addForeignKeyConstraint:
             baseTableName: document

resources/migrations/059_update_migrations.yaml

@@ -118,6 +118,7 @@ databaseChangeLog:
         - not:
             foreignKeyConstraintExists:
               foreignKeyName: fk_transform_collection_id
+              foreignKeyTableName: transform
       changes:
         - addForeignKeyConstraint:
             baseTableName: transform
@@ -1526,6 +1527,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_collection_workspace_id
+                foreignKeyTableName: collection
       changes:
         - addForeignKeyConstraint:
             baseTableName: collection
@@ -1629,6 +1631,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_log_workspace_id
+                foreignKeyTableName: workspace_log
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_log
@@ -1864,6 +1867,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_transform_workspace_id
+                foreignKeyTableName: workspace_transform
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_transform
@@ -1882,6 +1886,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_transform_global_id
+                foreignKeyTableName: workspace_transform
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_transform
@@ -1900,6 +1905,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_transform_collection_id
+                foreignKeyTableName: workspace_transform
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_transform
@@ -1918,6 +1924,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_transform_creator_id
+                foreignKeyTableName: workspace_transform
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_transform
@@ -2119,6 +2126,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_input_workspace_id
+                foreignKeyTableName: workspace_input
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_input
@@ -2137,6 +2145,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_input_table_id
+                foreignKeyTableName: workspace_input
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_input
@@ -2307,6 +2316,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_output_workspace_id
+                foreignKeyTableName: workspace_output
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_output
@@ -2325,6 +2335,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_output_ref_id
+                foreignKeyTableName: workspace_output
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_output
@@ -2415,6 +2426,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_merge_workspace_id
+                foreignKeyTableName: workspace_merge
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_merge
@@ -2433,6 +2445,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_merge_creator_id
+                foreignKeyTableName: workspace_merge
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_merge
@@ -2555,6 +2568,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_merge_transform_workspace_merge_id
+                foreignKeyTableName: workspace_merge_transform
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_merge_transform
@@ -2573,6 +2587,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_merge_transform_workspace_id
+                foreignKeyTableName: workspace_merge_transform
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_merge_transform
@@ -2591,6 +2606,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_merge_transform_transform_id
+                foreignKeyTableName: workspace_merge_transform
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_merge_transform
@@ -2609,6 +2625,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_merge_transform_merging_user_id
+                foreignKeyTableName: workspace_merge_transform
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_merge_transform
@@ -2837,6 +2854,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_output_external_workspace_id
+                foreignKeyTableName: workspace_output_external
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_output_external
@@ -2855,6 +2873,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_output_external_transform_id
+                foreignKeyTableName: workspace_output_external
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_output_external
@@ -2989,6 +3008,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_input_external_workspace_id
+                foreignKeyTableName: workspace_input_external
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_input_external
@@ -3103,6 +3123,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_workspace_graph_workspace_id
+                foreignKeyTableName: workspace_graph
       changes:
         - addForeignKeyConstraint:
             baseTableName: workspace_graph

resources/migrations/060/20260501_metabot_source_feedback.yaml

@@ -171,6 +171,7 @@ databaseChangeLog:
         - not:
             - foreignKeyConstraintExists:
                 foreignKeyName: fk_metabot_source_feedback_message_id
+                foreignKeyTableName: metabot_source_feedback
       changes:
         - addForeignKeyConstraint:
             baseTableName: metabot_source_feedback

resources/migrations/062/20260606_task_run_notification_id.yaml

@@ -40,6 +40,7 @@ databaseChangeLog:
         - not:
             - indexExists:
                 indexName: idx_task_run_notification_id_started_at
+                tableName: task_run
       changes:
         - createIndex:
             tableName: task_run

src/metabase/app_db/cluster_lock.clj

@@ -105,21 +105,73 @@
               result-set (.executeQuery stmt)]
     (when-not (.next result-set)
       ;; this record will not be visible until the tx commits, so there's no need to lock it
-      ;; we instead rely on concurrent threads having constraint violation trying to insert their own record
-      (t2/query-one {:insert-into [:metabase_cluster_lock]
-                     :columns [:lock_name]
-                     :values [[lock-name-str]]})))
+      ;; we instead rely on concurrent threads having constraint violation trying to insert their own record.
+      ;; the insert must go on `conn` explicitly: in detached mode there is no ambient transaction, and an
+      ;; insert on a pooled connection would commit immediately without us holding the row lock
+      (t2/query-one conn {:insert-into [:metabase_cluster_lock]
+                          :columns [:lock_name]
+                          :values [[lock-name-str]]})))
   (log/debugf "Obtained cluster lock: %s (%s)" lock-name-str mode))
 
+(def ^:private ^:dynamic *detached-locks-held*
+  "Map of lock-name-str -> mode for detached cluster locks held by the current thread. Used to make
+  re-acquisition of a held detached lock a no-op: a second acquisition would run on a second dedicated
+  connection and deadlock against our own row lock."
+  {})
+
+(defn- held-mode-covers?
+  "Whether an already-held detached lock in `held-mode` satisfies a request for `requested-mode`."
+  [held-mode requested-mode]
+  (or (= held-mode :exclusive)
+      (= held-mode requested-mode)))
+
+(defn- detached-held? [{:keys [lock-name-str mode]}]
+  (when-let [held-mode (*detached-locks-held* lock-name-str)]
+    (held-mode-covers? held-mode mode)))
+
 (defn- do-with-cluster-locks*
   "Acquire all `locks` (each a `{:lock-name-str, :mode}` map) inside a single
-  transaction, then run `thunk`."
+  transaction, then run `thunk`. Locks this thread already holds detached are skipped -- acquiring them
+  here would block against our own dedicated connection."
   [locks timeout-seconds thunk]
   (t2/with-transaction [conn]
-    (doseq [{:keys [lock-name-str mode]} locks]
+    (doseq [{:keys [lock-name-str mode] :as lock} locks
+            :when (not (detached-held? lock))]
       (acquire-lock-row! conn lock-name-str timeout-seconds mode))
     (thunk)))
 
+(defn- do-with-detached-cluster-locks*
+  "Like [[do-with-cluster-locks*]], but holds the lock rows on a dedicated unshared connection
+  (see [[metabase.app-db.connection/with-unshared-connection]]) while `thunk` runs on ordinary pooled
+  connections: `thunk`'s toucan work commits incrementally in its own short transactions instead of
+  riding one long transaction on the lock's connection. The locks are released when `thunk` completes
+  (commit) or throws (the pool rolls the dedicated connection back on check-in) -- but `thunk`'s own
+  already-committed work is NOT rolled back on a throw.
+
+  Requesting `:exclusive` while already holding the same lock detached in `:share` mode is not
+  supported (upgrading would self-deadlock) and throws."
+  [locks timeout-seconds thunk]
+  (let [needed (remove (fn [{:keys [lock-name-str mode] :as lock}]
+                         (when (and (contains? *detached-locks-held* lock-name-str)
+                                    (not (detached-held? lock)))
+                           (throw (ex-info "Cannot upgrade a held detached cluster lock from :share to :exclusive"
+                                           {:lock-name lock-name-str :held (*detached-locks-held* lock-name-str) :requested mode})))
+                         (detached-held? lock))
+                       locks)]
+    (if (empty? needed)
+      (thunk)
+      (mdb.connection/with-unshared-connection [conn]
+        (.setAutoCommit ^Connection conn false)
+        (doseq [{:keys [lock-name-str mode]} needed]
+          (acquire-lock-row! conn lock-name-str timeout-seconds mode))
+        (let [result (binding [*detached-locks-held* (into *detached-locks-held*
+                                                           (map (juxt :lock-name-str :mode))
+                                                           needed)]
+                       (thunk))]
+          ;; releases the row locks and persists any lock rows we inserted
+          (.commit ^Connection conn)
+          result)))))
+
 ;; ---------- h2 in-process rw locks ----------
 ;;
 ;; h2 does not respect query timeout when taking SELECT ... FOR UPDATE locks, and
@@ -178,18 +230,24 @@
     {:locks [(normalize-lock-spec opts)]}
 
     (map? opts)
-    (let [{:keys [lock locks timeout-seconds retry-config retry-transient?]} opts]
+    (let [{:keys [lock locks timeout-seconds retry-config retry-transient? detached?]} opts]
       (when (and lock locks)
         (throw (ex-info "Cluster-lock opts must specify exactly one of :lock or :locks"
                         {:opts opts})))
       (when-not (or lock locks)
         (throw (ex-info "Cluster-lock opts must specify :lock or :locks" {:opts opts})))
+      (when (and detached? retry-transient?)
+        ;; :retry-transient? re-runs the body assuming a rollback undid its work; detached bodies commit
+        ;; incrementally, so a re-run would double-apply everything before the failure
+        (throw (ex-info "Cluster-lock opts :detached? and :retry-transient? are mutually exclusive"
+                        {:opts opts})))
       (cond-> {:locks            (if lock
                                    [(normalize-lock-spec (if (map? lock)
                                                            lock
                                                            {:lock lock :mode (or (:mode opts) :exclusive)}))]
                                    (mapv normalize-lock-spec locks))
-               :retry-transient? (boolean retry-transient?)}
+               :retry-transient? (boolean retry-transient?)
+               :detached?        (boolean detached?)}
         timeout-seconds (assoc :timeout-seconds timeout-seconds)
         retry-config    (assoc :retry-config retry-config)))
 
@@ -227,7 +285,16 @@
   replicated across nodes, so the lock can't serialize writers and the conflicting
   commit comes back as a deadlock. Only opt in when the body is safe to re-run from
   scratch — idempotent, with no side effects outside the appdb transaction (a
-  rolled-back deadlock undoes only the db writes, not external calls)."
+  rolled-back deadlock undoes only the db writes, not external calls).
+
+  `:detached?` (default false) holds the lock rows on a dedicated connection instead
+  of the caller's transaction: the body's toucan work then runs on ordinary pooled
+  connections in its own short transactions, committing incrementally, rather than
+  riding one long transaction that holds the lock. Use for long-running locked work
+  (e.g. the audit DB install/load/sync at boot) whose body is idempotent/self-healing —
+  a failure mid-body does NOT roll back its already-committed work. Re-acquiring a
+  lock this thread already holds detached is a no-op (in both detached and
+  transactional form). Mutually exclusive with `:retry-transient?`."
   [opts :- [:or
             :keyword
             [:map
@@ -240,22 +307,25 @@
              [:mode             {:optional true} [:enum :exclusive :share]]
              [:timeout-seconds  {:optional true} :int]
              [:retry-config     {:optional true} [:ref ::retry/retry-overrides]]
-             [:retry-transient? {:optional true} :boolean]]]
+             [:retry-transient? {:optional true} :boolean]
+             [:detached?        {:optional true} :boolean]]]
    thunk :- ifn?]
-  (let [{:keys [locks timeout-seconds retry-config retry-transient?]
+  (let [{:keys [locks timeout-seconds retry-config retry-transient? detached?]
          :or   {timeout-seconds cluster-lock-timeout-seconds}} (parse-opts opts)]
     (cond
       ;; h2 does not respect the query timeout when taking the lock and is not cross-process,
-      ;; so we fall back to an in-process ReentrantReadWriteLock per lock name.
+      ;; so we fall back to an in-process ReentrantReadWriteLock per lock name. The h2 locks never
+      ;; hold a transaction open, so :detached? is already the behavior and needs no special casing.
       (= (mdb.connection/db-type) :h2)
       (do-with-h2-cluster-locks* locks thunk)
 
       :else
       (let [config (assoc (merge default-retry-config retry-config)
-                          :retry-if (fn [_ e] (retry-if-error? retry-transient? e)))]
+                          :retry-if (fn [_ e] (retry-if-error? retry-transient? e)))
+            impl   (if detached? do-with-detached-cluster-locks* do-with-cluster-locks*)]
         (try
           (retry/with-retry config
-            (do-with-cluster-locks* locks timeout-seconds thunk))
+            (impl locks timeout-seconds thunk))
           (catch Throwable e
             ;; only a genuine lock-acquisition failure gets the "Failed to obtain cluster lock" wrapper;
             ;; an exhausted transient body error (e.g. deadlock) propagates raw so the message stays truthful.

src/metabase/app_db/connection.clj

@@ -300,6 +300,37 @@
         (run-after-commit-callbacks! callbacks))
       result)))
 
+;;;; Unshared connections
+;;;;
+;;;; A connection that holds fragile long-lived state -- a streaming result set (postgres portal), a row
+;;;; lock held while other work runs -- must not be picked up by ambient connection reuse: a borrower that
+;;;; inherits it via [[toucan2.connection/*current-connectable*]] and commits destroys that state (see
+;;;; #78238 and the revert of #76645). [[toucan2.connection/unshared-connection!]] makes toucan use such a
+;;;; connection only when passed it explicitly, and never advertise it ambiently -- ambient work inside
+;;;; `body` resolves to the default connectable and runs on other pooled connections.
+
+(defn do-with-unshared-connection
+  "Impl for [[with-unshared-connection]]."
+  [f]
+  (with-open [conn (.getConnection (data-source))]
+    (f (t2.conn/unshared-connection! conn))))
+
+(defmacro with-unshared-connection
+  "Execute `body` with `conn-binding` bound to a fresh app-db connection that ambient connection reuse can
+  never pick up: toucan runs queries and transactions on it when `body` passes it explicitly, but never
+  binds it as `*current-connectable*`, so toucan calls that resolve their connection ambiently -- including
+  mid-reduction of a long-running query on this connection -- run (and commit) on other pooled connections.
+
+  In every other respect this is an ordinary JDBC connection, configured by `body`: e.g. call
+  `.setAutoCommit false` for a streaming result set (postgres portal/cursor) or a held row lock. It is
+  closed when `body` ends; the pool resets its state (rolling back any unresolved transaction) on check-in.
+
+    (mdb/with-unshared-connection [conn]
+      (.setAutoCommit conn false)
+      (reduce rf init (t2/reducible-query conn a-huge-query)))"
+  [[conn-binding] & body]
+  `(do-with-unshared-connection (fn [~conn-binding] ~@body)))
+
 (methodical/defmethod t2.pipeline/transduce-query :before :default
   "Make sure application database calls are not done inside core.async dispatch pool threads. This is done relatively
   early in the pipeline so the stacktrace when this fails isn't super enormous."

src/metabase/app_db/core.clj

@@ -42,7 +42,8 @@
   in-transaction?
   quoting-style
   unique-identifier
-  transaction-state]
+  transaction-state
+  with-unshared-connection]
  [mdb.connection-pool-setup
   recent-activity?]
  [mdb.data-source

test/metabase/app_db/cluster_lock_test.clj

@@ -5,6 +5,7 @@
    [metabase.app-db.cluster-lock :as sut]
    [metabase.app-db.core :as mdb]
    [metabase.app-db.transient-error :as transient-error]
+   [metabase.test :as mt]
    [metabase.test.fixtures :as fixtures]
    [metabase.util :as u]
    [toucan2.core :as t2])
@@ -226,3 +227,61 @@
         (is (= [[:a :failed :timeout]
                 [:b :done]]
                @results))))))
+
+(deftest detached-lock-basic-test
+  (testing "returns the body's value and works non-concurrently"
+    (is (= :ok (sut/with-cluster-lock {:lock ::detached-basic :detached? true} :ok))))
+  (testing ":detached? and :retry-transient? are mutually exclusive"
+    (is (thrown-with-msg? clojure.lang.ExceptionInfo #"mutually exclusive"
+                          (sut/with-cluster-lock {:lock ::detached-basic :detached? true :retry-transient? true}
+                            :nope)))))
+
+(deftest detached-lock-body-commits-incrementally-test
+  (let [email (mt/random-email)]
+    (try
+      (testing "body work commits as it goes and survives a later throw in the body"
+        (is (thrown-with-msg? clojure.lang.ExceptionInfo #"boom"
+                              (sut/with-cluster-lock {:lock ::detached-commit :detached? true}
+                                (t2/insert! :model/User (assoc (mt/with-temp-defaults :model/User) :email email))
+                                (throw (ex-info "boom" {})))))
+        (is (t2/exists? :model/User :email email)))
+      (finally
+        (t2/delete! :model/User :email email)))))
+
+(deftest detached-lock-reentrant-test
+  (testing "re-acquiring a held detached lock is a no-op instead of a self-deadlock"
+    (is (= :ok (sut/with-cluster-lock {:lock ::detached-reentrant :detached? true}
+                 (sut/with-cluster-lock {:lock ::detached-reentrant :detached? true
+                                         :timeout-seconds 1 :retry-config {:max-retries 0}}
+                   :ok)))))
+  (when (not= (mdb/db-type) :h2)
+    (testing "a transactional acquisition of a lock this thread holds detached is also a no-op"
+      (is (= :ok (sut/with-cluster-lock {:lock ::detached-reentrant :detached? true}
+                   (sut/with-cluster-lock {:lock ::detached-reentrant
+                                           :timeout-seconds 1 :retry-config {:max-retries 0}}
+                     :ok)))))
+    (testing "upgrading a held detached :share lock to :exclusive throws instead of self-deadlocking"
+      (is (thrown-with-msg? clojure.lang.ExceptionInfo #"Cannot upgrade"
+                            (sut/with-cluster-lock {:lock ::detached-reentrant :mode :share :detached? true}
+                              (sut/with-cluster-lock {:lock ::detached-reentrant :detached? true}
+                                :nope)))))))
+
+(deftest detached-lock-mutual-exclusion-test
+  ;; h2 uses the in-process rw-lock path where :detached? is a no-op; real row-lock semantics only.
+  (when (not= (mdb/db-type) :h2)
+    ;; warm up the row first so we're not racing on the initial INSERT
+    (sut/with-cluster-lock {:lock ::detached-mutex :detached? true} :warm)
+    (let [held (run-with-lock {:lock ::detached-mutex :detached? true})]
+      (is (.await ^CountDownLatch (:entered held) 3 TimeUnit/SECONDS))
+      (testing "a detached holder blocks another detached acquirer"
+        (is (not (acquirable-within? {:lock ::detached-mutex :detached? true
+                                      :timeout-seconds 1 :retry-config {:max-retries 0}}
+                                     3000))))
+      (testing "a detached holder blocks a transactional acquirer"
+        (is (not (acquirable-within? {:lock ::detached-mutex
+                                      :timeout-seconds 1 :retry-config {:max-retries 0}}
+                                     3000))))
+      (.countDown ^CountDownLatch (:release held))
+      (is (= :ok (deref (:done held) 3000 :timeout)))
+      (testing "released once the body completes"
+        (is (acquirable-within? {:lock ::detached-mutex :detached? true} 3000))))))

test/metabase/app_db/connection_test.clj

@@ -5,7 +5,8 @@
    [metabase.test :as mt]
    [metabase.test.fixtures :as fixtures]
    [toucan2.connection :as t2.connection]
-   [toucan2.core :as t2])
+   [toucan2.core :as t2]
+   [toucan2.jdbc.options :as t2.jdbc.options])
   (:import
    (java.sql Connection)
    (java.util.concurrent Semaphore)))
@@ -339,3 +340,67 @@
            (throw (ex-info "force savepoint rollback" {})))))
     (is (= {:outer-key "outer-val"} @mdb.connection/*transaction-state*)
         "data stashed by a rolled-back nested transaction is discarded from transaction-state")))
+
+(deftest unshared-connection-not-bound-during-use-test
+  (mdb.connection/with-unshared-connection [conn]
+    (testing "explicitly passing the unshared connection works, but it is never bound as *current-connectable*"
+      (let [rows (reduce (fn [acc row]
+                           (is (not (identical? conn t2.connection/*current-connectable*))
+                               "the unshared connection must not be ambiently visible mid-reduction")
+                           (conj acc (into {} row)))
+                         []
+                         (t2/reducible-query conn ["select 1 as x"]))]
+        (is (= 1 (count rows)))))
+    (testing "a transaction opened while the unshared connection is in use gets its own connection"
+      (reduce (fn [_ _row]
+                (t2/with-transaction [tx-conn]
+                  (is (not (identical? tx-conn conn)))))
+              nil
+              (t2/reducible-query conn ["select 1 as x"])))))
+
+(deftest unshared-connection-borrower-work-commits-independently-test
+  (let [email (mt/random-email)]
+    (testing "toucan work while the unshared connection is in use commits on its own connection"
+      (mdb.connection/with-unshared-connection [conn]
+        ;; open a transaction on the unshared connection that is never committed: if the insert below
+        ;; wrongly rode this connection, it would be rolled back at pool check-in and the user would not exist
+        (.setAutoCommit ^Connection conn false)
+        (reduce (fn [_ _row]
+                  (t2/insert! :model/User (assoc (mt/with-temp-defaults :model/User) :email email)))
+                nil
+                (t2/reducible-query conn ["select 1 as x"])))
+      (is (t2/exists? :model/User :email email))
+      (t2/delete! :model/User :email email))))
+
+(deftest unshared-connection-postgres-portal-survival-test
+  ;; postgres streams a result set through a portal that dies if anything commits on the connection
+  ;; mid-iteration -- the failure mode that forced the revert of #76645. Other appdbs buffer, so there is
+  ;; nothing to kill.
+  (when (= (mdb.connection/db-type) :postgres)
+    (let [portal-death? (fn [e]
+                          (loop [e e]
+                            (cond
+                              (nil? e)                                    false
+                              (re-find #"portal" (str (ex-message e)))    true
+                              :else                                       (recur (ex-cause e)))))
+          stream!       (fn [^Connection conn]
+                          ;; fetch-size + autocommit off = pg streams through a portal
+                          (.setAutoCommit conn false)
+                          (binding [t2.jdbc.options/*options* {:fetch-size 50}]
+                            (reduce (fn [n _row]
+                                      ;; unrelated toucan work mid-stream, resolved ambiently, that commits
+                                      (when (= n 30)
+                                        (t2/with-transaction [_]
+                                          (t2/query ["select 1"])))
+                                      (inc n))
+                                    0
+                                    (t2/reducible-query conn ["select g from generate_series(1, 1000) g"]))))]
+      (testing "control: an ordinary connection is ambiently visible mid-reduction, so the borrower commit kills the portal"
+        (with-open [^Connection raw (.getConnection (mdb.connection/data-source))]
+          (let [e (try (stream! raw) nil (catch Throwable e e))]
+            (is (portal-death? e)))
+          (.rollback raw)
+          (.setAutoCommit raw true)))
+      (testing "an unshared connection is invisible to the borrower; the portal survives"
+        (mdb.connection/with-unshared-connection [conn]
+          (is (= 1000 (stream! conn))))))))