Feature/nrmi 257 address deadlock error

app/jobs/cleanup_submission_entries_job.rb

@@ -1,9 +1,11 @@
 class CleanupSubmissionEntriesJob < ApplicationJob
+  retry_on ActiveRecord::Deadlocked, wait: :exponentially_longer, attempts: 8
+  retry_on PG::TRDeadlockDetected, wait: :exponentially_longer, attempts: 8
+
   # rubocop:disable Metrics/CyclomaticComplexity
   # rubocop:disable Metrics/PerceivedComplexity
   def perform(dry_run: false, max_run_time: 5.hours, task_batch_size: 100, entries_batch_size: 500)
     start_time = Time.current
-    total_deleted_entries = 0
 
     tasks_with_unprocessed_submissions = Task.joins(:submissions)
                                              .where(submissions: { aasm_state: 'validation_failed',
@@ -22,29 +24,39 @@ def perform(dry_run: false, max_run_time: 5.hours, task_batch_size: 100, entries
 
         next if failed_submissions.empty?
 
-        deleted_for_task = 0
-
         failed_submissions.find_each do |submission|
-          begin
-            submission.entries.in_batches(of: entries_batch_size) do |entries_batch|
-              if dry_run
-                Rollbar.info("Dry run: would delete #{entries_batch.count} entries for Submission ID #{submission.id}.")
-              else
-                deleted_count = entries_batch.delete_all
-                deleted_for_task += deleted_count
-                total_deleted_entries += deleted_count
-                # rubocop:disable Layout/LineLength
-                Rollbar.info("Task ID #{task.id}: Processed #{failed_submissions.count} failed submissions, deleted #{deleted_for_task} entries.")
-                # rubocop:enable Layout/LineLength
+          break if Time.current - start_time >= max_run_time
+
+          Submission.transaction do
+            submission.lock!
+
+            if dry_run
+              entries_count = submission.entries.count
+              staging_entries_count = submission.staging_entries.count
+              # rubocop:disable Layout/LineLength
+              Rollbar.info("Dry run: would delete #{entries_count + staging_entries_count} entries for Submission ID #{submission.id}.")
+              # rubocop:enable Layout/LineLength
+            else
+              deleted_entries = 0
+              deleted_staging_entries = 0
+
+              submission.entries.in_batches(of: entries_batch_size) do |entries_batch|
+                deleted_entries += entries_batch.delete_all
               end
-            end
 
-            submission.update(cleanup_processed: true) unless dry_run
-          rescue StandardError => e
-            Rollbar.error(e, "Error processing Submission ID #{submission.id} for Task ID #{task.id}: #{e.message}")
-          end
+              submission.staging_entries.in_batches(of: entries_batch_size) do |staging_entries_batch|
+                deleted_staging_entries += staging_entries_batch.delete_all
+              end
 
-          break if Time.current - start_time >= max_run_time
+              submission.update!(cleanup_processed: true)
+
+              # rubocop:disable Layout/LineLength
+              Rollbar.info("Task ID #{task.id}: Processed Submission ID #{submission.id}, deleted #{deleted_entries + deleted_staging_entries} entries.")
+              # rubocop:enable Layout/LineLength
+            end
+          end
+        rescue StandardError => e
+          Rollbar.error(e, "Error processing Submission ID #{submission.id} for Task ID #{task.id}: #{e.message}")
         end
       end
 

app/jobs/submission_ingestion_job.rb

@@ -3,6 +3,9 @@ class IngestFailed < StandardError; end
 
   queue_as :ingest
 
+  retry_on ActiveRecord::Deadlocked, wait: :exponentially_longer, attempts: 8
+  retry_on PG::TRDeadlockDetected, wait: :exponentially_longer, attempts: 8
+
   discard_on(Ingest::Loader::MissingColumns) do |job, exception|
     handle_unretryable_job_failure(job, exception)
   end

app/models/submission.rb

@@ -2,7 +2,7 @@ class Submission < ApplicationRecord
   ERRORED_ROW_LIMIT = 10
   REPLACEMENT_STATES = %i[validation_failed ingest_failed in_review completed].freeze
 
-  after_save :cleanup_prev_failed_entries, if: :state_changed_to_replacement?
+  after_save :mark_prev_failed_entries_for_cleanup, if: :state_changed_to_replacement?
 
   include AASM
 
@@ -126,11 +126,13 @@ def call_workday(workday_reference)
 
   private
 
-  def cleanup_prev_failed_entries
-    task.submissions.where(aasm_state: 'validation_failed').where.not(id: id).find_each do |s|
-      s.entries.destroy_all
-      s.staging_entries.destroy_all
-    end
+  def mark_prev_failed_entries_for_cleanup
+    # rubocop:disable Rails/SkipsModelValidations
+    task.submissions
+        .where(aasm_state: 'validation_failed')
+        .where.not(id: id)
+        .update_all(cleanup_processed: false)
+    # rubocop:enable Rails/SkipsModelValidations
   end
 
   def state_changed_to_replacement?

spec/jobs/cleanup_submission_entries_job_spec.rb

@@ -5,12 +5,11 @@
     let!(:task) { FactoryBot.create(:task) }
     let!(:failed_submission1) { FactoryBot.create(:submission_with_invalid_entries, task: task) }
     let!(:failed_submission2) { FactoryBot.create(:submission_with_invalid_entries, task: task) }
-    let!(:submission_entry) { FactoryBot.create(:submission_entry, submission: failed_submission1) }
 
     context 'when dry_run is true' do
       it 'does not delete entries and logs the intended deletions' do
         # rubocop:disable Layout/LineLength
-        expect(Rollbar).to receive(:info).with("Dry run: would delete 1 entries for Submission ID #{failed_submission1.id}.")
+        expect(Rollbar).to receive(:info).with("Dry run: would delete 3 entries for Submission ID #{failed_submission1.id}.")
         # rubocop:enable Layout/LineLength
         expect do
           described_class.perform_now(dry_run: true)
@@ -22,10 +21,12 @@
 
     context 'when dry_run is false' do
       it 'deletes entries and marks submissions as processed' do
-        expect(Rollbar).to receive(:info).with("Task ID #{task.id}: Processed 1 failed submissions, deleted 1 entries.")
+        # rubocop:disable Layout/LineLength
+        expect(Rollbar).to receive(:info).with("Task ID #{task.id}: Processed Submission ID #{failed_submission1.id}, deleted 3 entries.")
+        # rubocop:enable Layout/LineLength
         expect do
           described_class.perform_now(dry_run: false)
-        end.to change(SubmissionEntry, :count).by(-1)
+        end.to change(SubmissionEntry, :count).by(-3)
         expect(failed_submission1.reload.cleanup_processed).to be_truthy
       end
     end

spec/models/submission_spec.rb

@@ -15,11 +15,10 @@
     let!(:failed_submission) { FactoryBot.create(:submission_with_invalid_entries, task: task) }
 
     Submission::REPLACEMENT_STATES.each do |new_state|
-      it "deletes entries and staging entries for the failed submission when the new state is #{new_state}" do
+      it "marks prev failed entries for clean up when the new state is #{new_state}" do
         FactoryBot.create(:submission, aasm_state: new_state, task: task)
 
-        expect(failed_submission.entries.reload).to be_empty
-        expect(failed_submission.staging_entries.reload).to be_empty
+        expect(failed_submission.reload.cleanup_processed).to be false
       end
     end
   end