Module: Pgbus::VisibilityHeartbeat

Defined in:
lib/pgbus/visibility_heartbeat.rb

Overview

Keeps a running job's PGMQ message invisible for as long as the job is actually running.

PGMQ hands a message to one reader for visibility_timeout seconds. A job that runs longer is redelivered while it is still running: a second copy starts, read_ct climbs on every redelivery, and after max_retries the message is dead-lettered — all without the job ever raising. The heartbeat extends the VT of every in-flight message on a fixed cadence, so the timeout only ever fires for a job whose process is gone (crash, SIGKILL), which is the case it exists for.

One background thread per process, started lazily by the first tracked job and stopped by Worker#shutdown. Entries are keyed by physical queue + msg_id and hold the client that read the message, so async and threaded execution modes both work: the extension runs on this thread, never inside the job's fiber. A fork forgets the parent's entries.

Pgbus::VisibilityHeartbeat.track(client:, queue_name:, msg_id:) { job.perform_now }

Disable globally with config.visibility_heartbeat = false, tune the cadence with config.visibility_heartbeat_interval, or opt a job class out with pgbus_visibility_heartbeat false.

Defined Under Namespace

Modules: JobMixin Classes: Entry

Class Method Summary collapse

Class Method Details

.reset! ⇒ Object

Forget every entry and stop the thread. Test helper.



102
103
104
105
 
# File 'lib/pgbus/visibility_heartbeat.rb', line 102

def reset!
  stop
  synchronize { @entries = {} }
end

.stop ⇒ Object

Stop the background thread. Tracked entries are kept: a job still running during shutdown can drive tick! itself, and Worker#shutdown only calls this once the pool has drained.



88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/pgbus/visibility_heartbeat.rb', line 88

def stop
  thread = synchronize do
    @running = false
    current = @thread
    @thread = nil
    current
  end
  return unless thread

  thread.wakeup if thread.alive?
  thread.join(1)
end

.tick!(now: monotonic_now, config: Pgbus.configuration) ⇒ Object

Extend every tracked message whose last extension is older than the heartbeat interval. Public so tests and callers without the thread can drive it.



74
75
76
77
78
79
 
# File 'lib/pgbus/visibility_heartbeat.rb', line 74

def tick!(now: monotonic_now, config: Pgbus.configuration)
  interval = config.effective_visibility_heartbeat_interval
  due = synchronize { entries.values.select { |entry| now - entry.extended_at >= interval } }
  due.each { |entry| extend!(entry, now: now, config: config) }
  due.size
end

.track(client:, queue_name:, msg_id:, prefixed: true, job_class: nil, config: Pgbus.configuration) ⇒ Object

Track the message for the duration of the block.

Parameters:

  • client (Pgbus::Client) —

    the client that read the message

  • queue_name (String) —

    logical name, or physical when prefixed: false

  • msg_id (Integer)
  • prefixed (Boolean) (defaults to: true) —

    whether queue_name still needs the prefix

  • job_class (String, nil) (defaults to: nil) —

    for logging and instrumentation

  • config (Pgbus::Configuration) (defaults to: Pgbus.configuration) 


58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/pgbus/visibility_heartbeat.rb', line 58

def track(client:, queue_name:, msg_id:, prefixed: true, job_class: nil, config: Pgbus.configuration)
  return yield unless config.visibility_heartbeat

  entry = Entry.new(client: client, queue_name: queue_name, prefixed: prefixed, msg_id: msg_id.to_i,
                    job_class: job_class, extended_at: monotonic_now, extensions: 0)
  register(entry, config)
  begin
    yield
  ensure
    unregister(entry)
  end
end

.tracked_count ⇒ Object 



81
82
83
 
# File 'lib/pgbus/visibility_heartbeat.rb', line 81

def tracked_count
  synchronize { entries.size }
end