Merge pull request #2167 from broadinstitute/development

Release 1.85.0

Author: eweitz

app/models/delete_queue_job.rb

@@ -175,6 +175,50 @@ def self.prepare_file_for_deletion(study_file_id)
                            file_type: 'DELETE')
   end
 
+  # do a targeted cleanup of data to allow a file to retry ingest after OOM failure
+  def self.prepare_file_for_retry(study_file, action, cluster_name: nil)
+    study_file.reload
+    return false if study_file.queued_for_deletion
+
+    study = study_file.study
+    case action.to_sym
+    when :ingest_anndata
+      # delete raw counts 'all cells' array since this may be extracted during this phase
+      query = {
+        name: 'h5ad_frag.matrix.raw.mtx.gz Cells', array_type: 'cells', linear_data_type: 'Study',
+        linear_data_id: study.id, study_id: study.id, study_file_id: study_file.id, cluster_group_id: nil,
+        subsample_annotation: nil, subsample_threshold: nil
+      }
+      DataArray.where(query).delete_all
+    when :ingest_cell_metadata
+      metadata = CellMetadatum.where(study:, study_file:)
+      delete_arrays_by_query(study, study_file, 'CellMetadatum', metadata.pluck(:id))
+      metadata.delete_all
+      # now delete 'all cells' array
+      query = {
+        name: 'All Cells', array_type: 'cells', linear_data_type: 'Study', linear_data_id: study.id,
+        study_id: study.id, study_file_id: study_file.id, cluster_group_id: nil, subsample_annotation: nil,
+        subsample_threshold: nil
+      }
+      DataArray.where(query).delete_all
+    when :ingest_expression
+      genes = Gene.where(study:, study_file:)
+      delete_arrays_by_query(study, study_file, 'Gene', genes.pluck(:id))
+      genes.delete_all
+    when :ingest_cluster
+      cluster_group = study.cluster_groups.by_name(cluster_name)
+      delete_arrays_by_query(study, study_file, 'ClusterGroup', [cluster_group.id])
+      cluster_group.delete
+    else
+      # ingest_subsample and differential_expression require no extra data cleanup
+      return false
+    end
+  end
+
+  def self.delete_arrays_by_query(study, study_file, linear_data_type, ids = [])
+    DataArray.where(study:, study_file:, linear_data_type:, :linear_data_id.in => ids).delete_all
+  end
+
   private
 
   # remove a study_file from a study_file_bundle, and clean original_file_list up as necessary

app/models/ingest_job.rb

@@ -398,20 +398,19 @@ def poll_for_completion(run_at: 1.minute.from_now)
       log_to_mixpanel # log before queuing file for deletion to preserve properties
       # don't delete files or notify users if this is a 'special action', like DE or image pipeline jobs
       subject = "Error: #{study_file.file_type} file: '#{study_file.upload_file_name}' parse has failed"
-      retry_job = should_retry?
-      handle_ingest_failure(subject, retry_job:) unless special_action?
+      handle_ingest_failure(subject) unless (special_action? || should_retry?)
+
       admin_email_content = generate_error_email_body(email_type: :dev)
-      if retry_job && params_object&.next_machine_type
+      if should_retry? && params_object&.next_machine_type
         new_machine = params_object.next_machine_type
         params_object.machine_type = new_machine
-        cloned_file = study_file.clone
-        # free up filename and other values so cloned_file can save properly, including deleting nested documents
-        # this prevents Frozen BSON::Document error
-        DeleteQueueJob.prepare_file_for_deletion(study_file.id)
-        cloned_file.update!(parse_status: 'parsing')
-        file_identifier = "#{cloned_file.upload_file_name}:#{cloned_file.id} (#{study.accession})"
+        # run a selective cleanup to allow file to retry ingest on the next machine type
+        # this leaves any prior ingested valid data in place, and only removes data associated with this exact run
+        DeleteQueueJob.prepare_file_for_retry(study_file, action, cluster_name: params_object.try(:name))
+        study_file.update!(parse_status: 'parsing')
+        file_identifier = "#{study_file.upload_file_name}:#{study_file.id} (#{study.accession})"
         Rails.logger.info "Retrying #{action} after #{exit_code} failure for #{file_identifier} with machine_type: #{new_machine}"
-        retry_job = IngestJob.new(study:, study_file: cloned_file, user:, params_object:, action:, persist_on_fail:)
+        retry_job = IngestJob.new(study:, study_file:, user:, params_object:, action:, persist_on_fail:)
         retry_job.push_remote_and_launch_ingest
         # notify admins that the parse failed for visibility purposes
         SingleCellMailer.notify_admin_parse_fail(user.email, subject, admin_email_content).deliver_now
@@ -428,7 +427,7 @@ def poll_for_completion(run_at: 1.minute.from_now)
   # will automatically clean up data and notify user
   # in case of subsampling, only subsampled data cleanup is run and all other data is left in place
   # this reduces churn for study owners as full-resolution data is still valid
-  def handle_ingest_failure(email_subject, retry_job: false)
+  def handle_ingest_failure(email_subject)
     if action.to_sym == :ingest_subsample
       study_file.update(parse_status: 'parsed') # reset parse flag
       cluster_name = cluster_name_by_file_type
@@ -439,17 +438,15 @@ def handle_ingest_failure(email_subject, retry_job: false)
       create_study_file_copy
       study_file.update(parse_status: 'failed')
       DeleteQueueJob.new(study_file).delay.perform
-      unless persist_on_fail || retry_job
+      unless persist_on_fail
         ApplicationController.firecloud_client.delete_workspace_file(study.bucket_id, study_file.bucket_location)
         study_file.bundled_files.each do |bundled_file|
           ApplicationController.firecloud_client.delete_workspace_file(study.bucket_id, bundled_file.bucket_location)
         end
       end
     end
-    unless retry_job
-      user_email_content = generate_error_email_body
-      SingleCellMailer.notify_user_parse_fail(user.email, email_subject, user_email_content, study).deliver_now
-    end
+    user_email_content = generate_error_email_body
+    SingleCellMailer.notify_user_parse_fail(user.email, email_subject, user_email_content, study).deliver_now
   end
 
   # TODO (SCP-4709, SCP-4710) Processed and Raw expression files

test/models/delete_queue_job_test.rb

@@ -228,9 +228,9 @@ class DeleteQueueJobTest < ActiveSupport::TestCase
     assert_not DataArray.where(study_id: @basic_study.id, name: 'All Cells').exists?
   end
 
-  test 'should prepare file for deletion to allow cloning' do
+  test 'should prepare file for deletion' do
     study = FactoryBot.create(:detached_study,
-                              name_prefix: 'Clone Test',
+                              name_prefix: 'Deletion Prepare Test',
                               user: @user,
                               test_array: @@studies_to_clean)
     matrix = FactoryBot.create(:ann_data_file,
@@ -247,12 +247,43 @@ class DeleteQueueJobTest < ActiveSupport::TestCase
                                  'phex' => [['A', 0.3], ['B', 1.0], ['C', 0.5], ['D', 0.1]]
                                })
 
-    cloned_matrix = matrix.clone
     DeleteQueueJob.prepare_file_for_deletion(matrix.id)
     matrix.reload
     assert matrix.is_deleting?
-    assert cloned_matrix.valid?
-    cloned_matrix.save!
-    assert cloned_matrix.persisted?
+  end
+
+  test 'should prepare file for retry' do
+    study = FactoryBot.create(:detached_study,
+                              name_prefix: 'Retry Prepare Test',
+                              user: @user,
+                              test_array: @@studies_to_clean)
+    matrix = FactoryBot.create(:ann_data_file,
+                               name: 'matrix.h5ad',
+                               study:,
+                               cell_input: %w[A B C D],
+                               annotation_input: [
+                                 { name: 'disease', type: 'group', values: %w[cancer cancer normal normal] }
+                               ],
+                               coordinate_input: [
+                                 { tsne: { x: [1, 2, 3, 4], y: [5, 6, 7, 8] } }
+                               ],
+                               expression_input: {
+                                 'phex' => [['A', 0.3], ['B', 1.0], ['C', 0.5], ['D', 0.1]]
+                               })
+    assert_equal 1, study.cell_metadata.count
+    assert_equal 1, study.genes.count
+    assert_equal 1, study.cluster_groups.count
+    DeleteQueueJob.prepare_file_for_retry(matrix, :ingest_cell_metadata)
+    assert_empty CellMetadatum.where(study:, study_file: matrix)
+    assert_empty CellMetadatum.where(study:, study_file: matrix, linear_data_type: 'CellMetadatum')
+    DeleteQueueJob.prepare_file_for_retry(matrix, :ingest_expression)
+    assert_empty Gene.where(study:, study_file: matrix)
+    assert_empty CellMetadatum.where(study:, study_file: matrix, linear_data_type: 'Gene')
+    DeleteQueueJob.prepare_file_for_retry(matrix, :ingest_cluster, cluster_name: 'tsne')
+    assert_empty ClusterGroup.where(study:, study_file: matrix, name: 'tsne')
+    assert_empty CellMetadatum.where(study:, study_file: matrix, linear_data_type: 'ClusterGroup')
+    DeleteQueueJob.prepare_file_for_retry(matrix, :ingest_anndata)
+    study.reload
+    assert_empty study.expression_matrix_cells(matrix, matrix_type: 'raw')
   end
 end

test/models/ingest_job_test.rb

@@ -929,12 +929,12 @@ class IngestJobTest < ActiveSupport::TestCase
     }.with_indifferent_access
     # first pipeline mock represents the failed OOM job
     pipeline_mock = Minitest::Mock.new
-    8.times { pipeline_mock.expect :metadata, mock_metadata }
-    5.times { pipeline_mock.expect :done?, true }
+    9.times { pipeline_mock.expect :metadata, mock_metadata }
+    6.times { pipeline_mock.expect :done?, true }
     3.times { pipeline_mock.expect :error, { code: 137, message: 'OOM exception' } }
     # must mock life_sciences_api_client getting pipeline metadata
     client_mock = Minitest::Mock.new
-    16.times { client_mock.expect :get_pipeline, pipeline_mock, [{ name: pipeline_name }] }
+    18.times { client_mock.expect :get_pipeline, pipeline_mock, [{ name: pipeline_name }] }
     # new pipeline mock is resubmitted job with larger machine_type
     new_pipeline = Minitest::Mock.new
     new_op = Google::Apis::LifesciencesV2beta::Operation.new(name: 'oom-retry')
@@ -947,17 +947,14 @@ class IngestJobTest < ActiveSupport::TestCase
     # also prevents mock 'unexpected arguments' errors that can happen
     client_mock.expect :run_pipeline, new_op do |args|
       args[:study_file].upload_file_name == study_file.upload_file_name &&
-        args[:study_file].id.to_s != study_file.id.to_s && # this should be a new file with the same name
+        args[:study_file].id.to_s == study_file.id.to_s && # this should be the exact same file
         args[:action] == :ingest_anndata &&
         args[:params_object].machine_type == 'n2d-highmem-16'
     end
     terra_mock = Minitest::Mock.new
     terra_mock.expect :get_workspace_file,
                       Google::Cloud::Storage::File.new,
                       [bucket, study_file.bucket_location]
-    terra_mock.expect :execute_gcloud_method,
-                      Google::Cloud::Storage::File.new,
-                      [:copy_workspace_file, 0, bucket, study_file.bucket_location, study_file.parse_fail_bucket_location]
     terra_mock.expect :workspace_file_exists?,
                       false,
                       [bucket, String]