Merge pull request #732 from coursemology-collab/lowjoel/deadlock-fixes

 ActiveRecord::Base#wait deadlock fix, part 2

Gemfile

@@ -34,6 +34,8 @@ gem 'validates_hostname'
 gem 'workflow'
 # Add creator_id and updater_id attributes to models
 gem 'activerecord-userstamp', '>= 3.0.2'
+# Allow actions to be deferred until after a record is committed.
+gem 'after_commit_action'
 # Allow declaring the calculated attributes of a record
 gem 'calculated_attributes', '>= 0.1.3'
 # Squeel as an SQL-like DSL

Gemfile.lock

@@ -98,6 +98,9 @@ GEM
       thread_safe (~> 0.3, >= 0.3.4)
       tzinfo (~> 1.1)
     addressable (2.4.0)
+    after_commit_action (1.0.1)
+      activerecord (>= 3.0.0)
+      activesupport (>= 3.0.0)
     anbt-sql-formatter (0.0.3)
     arel (6.0.3)
     ast (2.2.0)
@@ -514,6 +517,7 @@ DEPENDENCIES
   active_record-acts_as!
   activerecord-userstamp (>= 3.0.2)
   acts_as_tenant!
+  after_commit_action
   autoprefixer-rails
   bootstrap-sass
   bootstrap-sass-extras!

app/models/course/assessment/answer.rb

@@ -77,9 +77,14 @@ def validate_consistent_grade
   # there can be a concurrent creation of such a record across two processes, and this can only
   # be detected at the database level.
   #
+  # The additional transaction is in place because a RecordNotUnique will cause the active
+  # transaction to be considered as errored, and needing a rollback.
+  #
   # @return [Course::Assessment::Answer::AutoGrading]
   def ensure_auto_grading!
-    auto_grading || create_auto_grading!
+    ActiveRecord::Base.transaction do
+      auto_grading || create_auto_grading!
+    end
   rescue ActiveRecord::RecordInvalid, ActiveRecord::RecordNotUnique => e
     raise e if e.is_a?(ActiveRecord::RecordInvalid) && e.record.errors[:answer_id].empty?
     association(:auto_grading).reload

app/models/course/assessment/question/programming.rb

@@ -5,7 +5,6 @@ class Course::Assessment::Question::Programming < ActiveRecord::Base
   acts_as :question, class_name: Course::Assessment::Question.name
 
   after_save :process_new_package, if: :attachment_changed?
-  after_commit :save, if: :import_job_id_changed?
 
   validates :memory_limit, :time_limit, numericality: { greater_then: 0 }
 
@@ -41,7 +40,10 @@ def copy_template_files_to(answer)
 
   # Queues the new question package for processing.
   def process_new_package
-    self.import_job_id =
-      Course::Assessment::Question::ProgrammingImportJob.perform_later(self, attachment).job_id
+    execute_after_commit do
+      import_job =
+        Course::Assessment::Question::ProgrammingImportJob.perform_later(self, attachment)
+      update(import_job_id: import_job.job_id)
+    end
   end
 end

app/models/course/assessment/submission.rb

@@ -115,6 +115,8 @@ def publish(_ = nil)
   def auto_grade_submission
     return unless workflow_state_changed?
 
-    auto_grade!
+    execute_after_commit do
+      auto_grade!
+    end
   end
 end

app/models/course/discussion/topic.rb

@@ -16,10 +16,15 @@ def subscribed_by?(user)
 
   # Create subscription for a user
   #
+  # The additional transaction is in place because a RecordNotUnique will cause the active
+  # transaction to be considered as errored, and needing a rollback.
+  #
   # @param [User] user The user who needs to subscribe to this topic
   def ensure_subscribed_by(user)
     return true if self.subscribed_by?(user)
-    subscriptions.build(user: user).save!
+    ActiveRecord::Base.transaction do
+      subscriptions.build(user: user).save!
+    end
   rescue ActiveRecord::RecordInvalid, ActiveRecord::RecordNotUnique => e
     return true if e.is_a?(ActiveRecord::RecordInvalid) &&
                    e.record.errors[:topic_id].any? && e.record.errors[:user_id].any?

lib/autoload/active_job/queue_adapters/background_thread_adapter.rb

@@ -10,11 +10,13 @@
 class ActiveJob::QueueAdapters::BackgroundThreadAdapter < ActiveJob::QueueAdapters::InlineAdapter
   class << self
     def enqueue(job) #:nodoc:
-      Thread.new do
+      thread = Thread.new do
         ActiveRecord::Base.connection_pool.with_connection do
           ActiveJob::Base.execute(job.serialize)
         end
       end
+
+      thread.abort_on_exception = true
     end
   end
 end

lib/autoload/trackable_job.rb

@@ -22,6 +22,7 @@ class Job < ActiveRecord::Base
   end
 
   included do
+    before_enqueue :save_job
     rescue_from(StandardError) do |exception|
       rescue_tracked(exception)
     end
@@ -41,11 +42,11 @@ def wait(timeout = nil)
     fail Timeout::Error if wait_result.nil?
   end
 
-  # Implements +initialize+, creating the job in the database.
+  # Implements +initialize+, creating or loading the job from the database.
   def initialize(*args)
     super
 
-    @job = Job.create!(id: job_id)
+    @job = Job.find_or_initialize_by(id: job_id)
   end
 
   def perform(*args)
@@ -55,9 +56,11 @@ def perform(*args)
   end
 
   def job_id=(job_id)
-    super
-    @job.destroy
-    @job = Job.find(job_id)
+    super.tap do
+      next unless @job
+      @job.destroy
+      @job = Job.find(job_id)
+    end
   end
 
   protected
@@ -78,6 +81,13 @@ def rescue_tracked(exception)
 
   private
 
+  # Saves the job to the database.
+  #
+  # The job is not saved until queueing because the job ID might not be decided.
+  def save_job
+    @job.save!
+  end
+
   # Specifies that the job should redirect to the given path.
   def redirect_to(path)
     @job.redirect_to = path

lib/extensions/after_commit_action.rb

@@ -0,0 +1,4 @@
+module Extensions::AfterCommitAction; end
+ActiveRecord::Base.class_eval do
+  include AfterCommitAction
+end

spec/libraries/trackable_job_spec.rb

@@ -68,6 +68,10 @@ def self.validate_absence_of_error
     it 'is persisted to the database' do
       expect(TrackableJob::Job.find(subject.job_id)).to eq(subject.job)
     end
+
+    it 'only creates one job' do
+      expect { subject }.to change { TrackableJob::Job.count }.by(1)
+    end
   end
 
   context 'when the job is completed' do
@@ -118,10 +122,12 @@ def subject.perform_tracked
   end
 
   describe '#perform_tracked' do
-    subject { self.class::NoOpJob.perform_later }
+    with_active_job_queue_adapter(:inline) do
+      subject { self.class::NoOpJob.perform_later }
 
-    it 'fails with NotImplementedError' do
-      expect { subject.send(:perform_tracked) }.to raise_error(NotImplementedError)
+      it 'fails with NotImplementedError' do
+        expect { subject.send(:perform_tracked) }.to raise_error(NotImplementedError)
+      end
     end
   end