Class: Pgbus::BatchExecution

Inherits:
BusRecord
  • Object
show all
Defined in:
app/models/pgbus/batch_execution.rb

Class Method Summary collapse

Methods inherited from BusRecord

disconnect_all_pools!

Class Method Details

.backfill!(job_id, msg_id:, queue_name:) ⇒ Object 



24
25
26
 
# File 'app/models/pgbus/batch_execution.rb', line 24

def self.backfill!(job_id, msg_id:, queue_name:)
  where(job_id: job_id).update_all(msg_id: msg_id, queue_name: queue_name)
end

.insert_for!(batch_id:, job_id:, queue_name: nil) ⇒ Object

One row per outstanding batched job. Inserted before send_message so a crash cannot produce an untracked in-flight job. ON CONFLICT DO NOTHING makes a retry re-enqueue of the same ActiveJob id a no-op. Raw SQL rather than insert_all(unique_by:) — Rails resolves unique_by through the schema cache (issue #401). queue_name is the logical queue at insert (so the sweep can probe for a live message before the msg_id backfill lands); backfill! overwrites it with the physical target after send.



15
16
17
18
19
20
21
22
 
# File 'app/models/pgbus/batch_execution.rb', line 15

def self.insert_for!(batch_id:, job_id:, queue_name: nil)
  connection.exec_query(
    "INSERT INTO #{table_name} (batch_id, job_id, queue_name, created_at) " \
    "VALUES ($1, $2, $3, $4) ON CONFLICT (job_id) DO NOTHING",
    "BatchExecution Insert",
    [batch_id, job_id, queue_name, Time.current]
  )
end