Module: Pgbus::Batch::Sweep

Defined in:
lib/pgbus/batch/sweep.rb

Overview

Repairs batches the regular completion path cannot finish: worker crash between archive and row-delete, enqueue crash between row-insert and send, a pending batch whose enqueue block never returned, or a processing batch whose finish UPDATE rolled back after callbacks failed to enqueue.

Constant Summary collapse

STALL_THRESHOLD =

seconds; solid_queue default stalled_for: 5.minutes

300

Class Method Summary collapse

Class Method Details

.run(stalled_for: Pgbus.configuration.batch_stall_threshold, batch_size: 500, client: Pgbus.client) ⇒ Object 



13
14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/pgbus/batch/sweep.rb', line 13

def run(stalled_for: Pgbus.configuration.batch_stall_threshold, batch_size: 500, client: Pgbus.client)
  return unless Batch.executions_migrated?

  payload = { stale_executions: 0, orphan_rows: 0, started_batches: 0, finished_batches: 0,
              stalled_for: stalled_for }
  Instrumentation.instrument("pgbus.batch_sweep", payload) do |p|
    p[:stale_executions] = sweep_stale_executions(batch_size: batch_size, client: client, stalled_for: stalled_for)
    p[:orphan_rows] = sweep_orphan_rows(stalled_for: stalled_for, batch_size: batch_size, client: client)
    p[:started_batches] = start_stalled_pending(stalled_for: stalled_for, batch_size: batch_size)
    p[:finished_batches] = finish_stalled_processing(batch_size: batch_size)
  end
end