Fixes #3329 - Scheduler jobs don't ensure current ticket information if they're running a long time

app/models/job.rb

@@ -53,50 +53,15 @@ job.run(true)
   def run(force = false, start_at = Time.zone.now)
     logger.debug { "Execute job #{inspect}" }
 
-    tickets = nil
+    ticket_ids = start_job(start_at, force)
-    Transaction.execute(reset_user_id: true) do
-      if !executable?(start_at) && force == false
-        if next_run_at && next_run_at <= Time.zone.now
-          save!
-        end
-        return
-      end
 
-      # find tickets to change
+    return if ticket_ids.nil?
-      matching = matching_count
-      if self.matching != matching
-        self.matching = matching
-        save!
-      end
 
-      if !in_timeplan?(start_at) && force == false
+    ticket_ids&.each_slice(10) do |slice|
-        if next_run_at && next_run_at <= Time.zone.now
+      run_slice(slice)
-          save!
-        end
-        return
-      end
-
-      ticket_count, tickets = Ticket.selectors(condition, limit: 2_000, execution_time: true)
-
-      logger.debug { "Job #{name} with #{ticket_count} tickets" }
-
-      self.processed = ticket_count || 0
-      self.running = true
-      self.last_run_at = Time.zone.now
-      save!
     end
 
-    tickets&.each do |ticket|
+    finish_job
-      Transaction.execute(disable_notification: disable_notification, reset_user_id: true) do
-        ticket.perform_changes(self, 'job')
-      end
-    end
-
-    Transaction.execute(reset_user_id: true) do
-      self.running = false
-      self.last_run_at = Time.zone.now
-      save!
-    end
   end
 
   def executable?(start_at = Time.zone.now)
@@ -144,10 +109,79 @@ job.run(true)
     self.next_run_at = next_run_at_calculate
   end
 
-  def match_minutes(minutes)
+  def finish_job
-    return 0 if minutes < 10
+    Transaction.execute(reset_user_id: true) do
-
+      mark_as_finished
-    "#{minutes.to_s.gsub(%r{(\d)\d}, '\\1')}0".to_i
+    end
   end
 
+  def mark_as_finished
+    self.running = false
+    self.last_run_at = Time.zone.now
+    save!
+  end
+
+  def start_job(start_at, force)
+    Transaction.execute(reset_user_id: true) do
+      if start_job_executable?(start_at, force) && start_job_ensure_matching_count && start_job_in_timeplan?(start_at, force)
+        ticket_count, tickets = Ticket.selectors(condition, limit: 2_000, execution_time: true)
+
+        logger.debug { "Job #{name} with #{ticket_count} tickets" }
+
+        mark_as_started(ticket_count)
+
+        tickets&.pluck(:id) || []
+      end
+    end
+  end
+
+  def start_job_executable?(start_at, force)
+    return true if executable?(start_at) || force
+
+    if next_run_at && next_run_at <= Time.zone.now
+      save!
+    end
+
+    false
+  end
+
+  def start_job_ensure_matching_count
+    matching = matching_count
+
+    if self.matching != matching
+      self.matching = matching
+      save!
+    end
+
+    true
+  end
+
+  def start_job_in_timeplan?(start_at, force)
+    return true if in_timeplan?(start_at) || force
+
+    if next_run_at && next_run_at <= Time.zone.now
+      save!
+    end
+
+    false
+  end
+
+  def mark_as_started(ticket_count)
+    self.processed = ticket_count || 0
+    self.running = true
+    self.last_run_at = Time.zone.now
+    save!
+  end
+
+  def run_slice(slice)
+    Transaction.execute(disable_notification: disable_notification, reset_user_id: true) do
+      _, tickets = Ticket.selectors(condition, limit: 2_000, execution_time: true)
+
+      tickets
+        &.where(id: slice)
+        &.each do |ticket|
+          ticket.perform_changes(self, 'job')
+        end
+    end
+  end
 end

spec/models/job_spec.rb

@@ -487,4 +487,49 @@ RSpec.describe Job, type: :model do
       end
     end
   end
+
+  # when running a very large job, tickets may change during the job
+  # if tickets are fetched once, their action may be performed later on
+  # when it no longer matches the conditions
+  # https://github.com/zammad/zammad/issues/3329
+  context 'job re-checks conditions' do
+    let(:job)    { create(:job, condition: condition, perform: perform) }
+    let(:ticket) { create(:ticket, title: initial_title) }
+    let(:initial_title) { 'initial 3329'  }
+    let(:changed_title) { 'performed 3329' }
+
+    let(:condition) do
+      { 'ticket.title' => { 'value' => initial_title, 'operator' => 'is' } }
+    end
+
+    let(:perform) do
+      { 'ticket.title' => { 'value'=> changed_title } }
+    end
+
+    it 'condition matches ticket' do
+      ticket
+      expect(job.send(:start_job, Time.zone.now, true)).to eq [ticket.id]
+    end
+
+    it 'action is performed' do
+      ticket
+
+      ticket_ids = job.send(:start_job, Time.zone.now, true)
+      job.send(:run_slice, ticket_ids)
+
+      expect(ticket.reload.title).to eq changed_title
+    end
+
+    it 'checks conditions' do
+      ticket
+
+      ticket_ids = job.send(:start_job, Time.zone.now, true)
+
+      ticket.update! title: 'another title'
+
+      job.send(:run_slice, ticket_ids)
+
+      expect(ticket.reload.title).not_to eq changed_title
+    end
+  end
 end