Class: Pgbus::MCP::HealthAnalyzer

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/mcp/health_analyzer.rb

Overview

Computes the top-level pgbus health verdict (OK / DEGRADED / STALLED) from the existing DataSource read layer. This is the single signal that catches the silent-worker-wedge class of incident (#179, #174, #181): a queue with visible messages and no claim progress while a subscribing worker is heart-beating with idle capacity.

Verdict semantics (issue #180 acceptance criteria):

STALLED  — backlog (visible > 0) AND at least one worker is heart-beating
         but its claim loop has stopped advancing (status :stalled),
         OR backlog with live-but-idle workers and zero claim progress.
DEGRADED — something is wrong but not the wedge: stale processes, a
         paused queue holding a backlog, growing DLQ, or MVCC horizon
         pinned by a long-running transaction.
OK       — draining normally / nothing actionable.

Constant Summary collapse

WORKER_KIND =

A worker is considered to have idle capacity unless its metadata explicitly reports it is saturated. We treat the presence of any live worker as "has capacity" because a wedged worker reports healthy heartbeats while doing no work — exactly the case we must catch.

"worker"
DRAINING_KINDS =

Process kinds that claim from queues: workers drain job queues, consumers drain EventBus handler queues. drained_queue_names unions both kinds' queues, so the wedge signal must count both kinds as liveness evidence — otherwise a consumer-only fleet with a wedged handler queue is missed (the worker set is empty and the branch never runs).

%w[worker consumer].freeze
NAME_LIMIT =

Cap on how many queue names a reason string lists before truncating to "(+N more)" — keeps a flagged fleet of hundreds of queues from producing a multi-KB log line (issue #367 minor).

10

Instance Method Summary collapse

Constructor Details

#initialize(data_source) ⇒ HealthAnalyzer

Returns a new instance of HealthAnalyzer.



38
39
40
 
# File 'lib/pgbus/mcp/health_analyzer.rb', line 38

def initialize(data_source)
  @data_source = data_source
end

Instance Method Details

#verdict ⇒ Object

Returns a machine-readable verdict hash suitable for both interactive agent use and automated alerting.



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/pgbus/mcp/health_analyzer.rb', line 44

def verdict
  queues          = @data_source.queues_with_metrics
  processes       = @data_source.processes
  stream_names    = @data_source.stream_queue_names
  health, health_error = safe_queue_health

  # Partition queues once: the operational set excludes DLQs and
  # registered stream queues, and the backlog is the subset with
  # visible, claimable messages. Stream queues are excluded like DLQs
  # (issue #359): stream delivery is a non-consuming peek, so their
  # messages are permanently visible with read_ct=0 — exactly the wedge
  # signature — and counting them makes the verdict scream STALLED on
  # every streams-heavy install. Paused queues are removed from the
  # STALLED backlog (an intentional pause is not the wedge — it's
  # reported under DEGRADED), but kept in `non_dlq` for the summary.
  non_dlq = queues.reject { |q| dlq?(q) || stream_names.include?(q[:name].to_s) }
  backlog = non_dlq.select { |q| q[:queue_visible_length].to_i.positive? }
  active_backlog = backlog.reject { |q| q[:paused] }

  # Split the active backlog by whether some configured capsule or handler
  # drains the queue (issue #367 defect 2). Live workers say nothing about
  # a queue they can never claim from, so only the drained side can be the
  # silent-worker-wedge; the undrained side is its own DEGRADED hazard.
  drained, undrained = partition_by_drain(active_backlog)

  stalled  = stalled_reasons(active_backlog, drained, processes)
  degraded = degraded_reasons(queues, backlog, undrained, processes, health, health_error)

  status = if stalled.any?
             "STALLED"
           elsif degraded.any?
             "DEGRADED"
           else
             "OK"
           end

  {
    status: status,
    reasons: stalled + degraded,
    checked_at: Time.now.utc.iso8601,
    summary: build_summary(queues, non_dlq, processes, health)
  }
end