Remove txn from with_statement_timeout block (#15088)

Turns out this caused a *different* problem. It would work for the first
delete batch in `BulkSqlDelete`, but subsequent `DELETE` queries would
then get their `statement_timeout`s reset to the global setting. The
issue was that `BulkSqlDelete` uses methods that don't respect nested
transactions, and we were already resetting the value of the setting to
its previous value anyway, so there was really no reason to run it
inside a transaction.

The only failure mode I know of here is that the `ensure` block breaks
because `connection` is severed. In that case, the value is reset anyway
when the connection pool replaces the connection so the failure does not
propagate.

app/models/application_record.rb

@@ -60,28 +60,14 @@ class ApplicationRecord < ActiveRecord::Base
       .seconds / 1000
   end
 
-  def self.with_statement_timeout(duration)
-    # Using SET LOCAL only works inside a transaction.
-    transaction do
-      original_timeout = statement_timeout
-      milliseconds = (duration.to_f * 1000).to_i
-      connection.execute "SET LOCAL statement_timeout = #{milliseconds}"
-      yield
-    ensure
-      # Typically this value is reset after a transaction is completed (whether
-      # successfully or unsuccessfully), but a nested transaction will not reset
-      # this value after the inner transaction is complete because they are not
-      # isolated from the outer transaction the way the outer transaction is
-      # from transactions on other connections. Nested transactions are
-      # implemented simply as savepoints inside the top-level transaction. For
-      # this reason, we need to explicitly reset the statement timeout so that
-      # when the block completes the statement timeout is reset to its previous
-      # value.
-      #
-      # See: https://www.postgresql.org/docs/11/sql-savepoint.html
-      milliseconds = original_timeout.to_i * 1000
-      connection.execute "SET LOCAL statement_timeout = #{milliseconds}"
-    end
+  def self.with_statement_timeout(duration, connection: self.connection)
+    original_timeout = statement_timeout
+    milliseconds = (duration.to_f * 1000).to_i
+    connection.execute "SET statement_timeout = #{milliseconds}"
+    yield
+  ensure
+    milliseconds = original_timeout.to_i * 1000
+    connection.execute "SET statement_timeout = #{milliseconds}"
   end
 
   def errors_as_sentence

app/services/bulk_sql_delete.rb

@@ -1,11 +1,15 @@
 class BulkSqlDelete
+  STATEMENT_TIMEOUT = ENV.fetch("STATEMENT_TIMEOUT_BULK_DELETE", 10_000).to_i.seconds / 1_000.to_f
+
   def self.delete_in_batches(sql)
     ActiveRecord::Base.connection_pool.with_connection do |connection|
       perform_and_log(sql) do
         unless Rails.env.test?
           connection.begin_db_transaction
         end
-        result = connection.exec_delete(sql)
+        result = ApplicationRecord.with_statement_timeout STATEMENT_TIMEOUT, connection: connection do
+          connection.exec_delete(sql)
+        end
         connection.commit_db_transaction unless Rails.env.test?
 
         result

app/workers/emails/remove_old_emails_worker.rb

@@ -4,12 +4,8 @@ module Emails
 
     sidekiq_options queue: :low_priority, retry: 10
 
-    STATEMENT_TIMEOUT = ENV.fetch("STATEMENT_TIMEOUT_REMOVE_OLD_EMAILS", 10_000).to_i.seconds / 1_000.to_f
-
     def perform
-      ApplicationRecord.with_statement_timeout STATEMENT_TIMEOUT do
-        EmailMessage.fast_destroy_old_retained_email_deliveries
-      end
+      EmailMessage.fast_destroy_old_retained_email_deliveries
     end
   end
 end