Module: Pgbus::Concurrency::BlockedExecution

Defined in:
lib/pgbus/concurrency/blocked_execution.rb

Class Method Summary collapse

Class Method Details

.count_for(concurrency_key) ⇒ Object

Count blocked executions for a given key. Useful for testing/monitoring.



65
66
67
 
# File 'lib/pgbus/concurrency/blocked_execution.rb', line 65

def count_for(concurrency_key)
  Pgbus::BlockedExecution.where(concurrency_key: concurrency_key).count
end

.expire_stale ⇒ Object

Delete blocked executions that have expired. Returns the count of deleted rows.



60
61
62
 
# File 'lib/pgbus/concurrency/blocked_execution.rb', line 60

def expire_stale
  Pgbus::BlockedExecution.expired(Time.current).delete_all
end

.insert(concurrency_key:, queue_name:, payload:, duration:, priority: 0) ⇒ Object

Insert a blocked execution for a job that hit the concurrency limit.



10
11
12
13
14
15
16
17
18
 
# File 'lib/pgbus/concurrency/blocked_execution.rb', line 10

def insert(concurrency_key:, queue_name:, payload:, duration:, priority: 0)
  Pgbus::BlockedExecution.create!(
    concurrency_key: concurrency_key,
    queue_name: queue_name,
    payload: JSON.generate(payload),
    priority: priority,
    expires_at: Time.current + duration
  )
end

.promote_next(concurrency_key, client:, delay: 0) ⇒ Object

Atomically promote the next blocked execution: delete the row and enqueue the job in a single transaction. Returns true if a job was promoted, false otherwise. This avoids losing a blocked row if enqueue fails.



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/pgbus/concurrency/blocked_execution.rb', line 29

def promote_next(concurrency_key, client:, delay: 0)
  released = nil
  msg_id = nil
  Pgbus::BlockedExecution.transaction do
    released = release_next(concurrency_key)
    raise ActiveRecord::Rollback unless released

    actual_delay = resolve_delay(released[:payload], delay)
    # Carry the enqueuer's priority through: under priority routing it
    # picks the _pN sub-queue, not just the release order (issue #423).
    msg_id = client.send_message(released[:queue_name], released[:payload],
                                 delay: actual_delay, priority: released[:priority])
  end

  if released && msg_id
    begin
      Batch.backfill_execution(released[:payload], msg_id,
                               client.target_queue(released[:queue_name], released[:priority]))
    rescue StandardError => e
      Pgbus.logger.warn { "[Pgbus] Batch execution backfill failed after promote: #{e.message}" }
    end
  end

  !!released
rescue StandardError => e
  Pgbus.logger.warn { "[Pgbus] Promote blocked execution failed for #{concurrency_key}: #{e.message}" }
  false
end

.release_next(concurrency_key) ⇒ Object

Release the next blocked execution for a given concurrency key. Returns the released row (queue_name, payload) or nil if none.



22
23
24
 
# File 'lib/pgbus/concurrency/blocked_execution.rb', line 22

def release_next(concurrency_key)
  Pgbus::BlockedExecution.release_next!(concurrency_key)
end