Class: ChronoForge::BranchMergeJob

Inherits:
ActiveJob::Base
  • Object
show all
Defined in:
lib/chrono_forge/branch_merge_job.rb

Overview

Lightweight poller that joins one or more branches. NOT a workflow — it holds no lock, does no replay, and carries no context. It exists so the heavy parent workflow is replayed only twice per merge (kick off + completion wake).

DEPLOY NOTE — queue placement matters. merge_branches enqueues this poller AFTER dispatching the branch's children, so if it runs on the SAME queue as a large fan-out's children it is starved behind the whole backlog and only gets a worker slot near the end. It then polls once, at pending≈0, with no prior sample (rate 0) and backs off to max_interval — so the parent's convergence lags by up to max_interval and no mid-drain throughput sample is ever recorded. Set ChronoForge.config.branch_merge_queue to a queue NOT saturated by the fan-out's own children so it polls throughout the drain (ETA cadence then converges tightly). See ChronoForge::Configuration and docs/fanout-scale-test.md.

Constant Summary collapse

ETA_FRACTION =

poll at this fraction of the projected time-to-drain

0.5
REKICK_AFTER = 
5.minutes
REKICK_BATCH =

bound per-run rekicks; later polls handle the rest

200

Instance Method Summary collapse

Instance Method Details

#perform(parent_key, parent_job_class, branch_log_ids, min_interval, max_interval, token = nil) ⇒ Object

Raises:

  • (ArgumentError) 


38
39
40
41
42
43
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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/chrono_forge/branch_merge_job.rb', line 38

def perform(parent_key, parent_job_class, branch_log_ids, min_interval, max_interval, token = nil)
  raise ArgumentError, "branch_log_ids must not be empty" if branch_log_ids.empty?

  # Fencing: every merge_branches pass mints a fresh token and writes it onto
  # the branch logs, so a poller from a superseded chain (parent replay /
  # re-enqueue) holds a stale token. It stops quietly — no poll, no wake, no
  # reschedule — leaving only the newest chain to drive the merge. (A nil token
  # is a pre-upgrade job enqueued before fencing existed; it runs unfenced.)
  logs = ExecutionLog.where(id: branch_log_ids).to_a
  return if superseded?(logs, token)

  # Per-branch probe (kept as maps so we can persist each branch's own state,
  # not just the merge aggregate). Same query count as a plain sum/all?.
  # The pending count is UNCAPPED: it feeds the drain signal below (a change in
  # pending since the prior poll), which a CAP would flatten into a false
  # "not draining" for large branches.
  prev_pending_by_branch = logs.to_h { |l| [l.id, l.metadata&.dig("poll", "pending")] }
  pending_by_branch = branch_log_ids.to_h { |id| [id, BranchProbe.incomplete(id).count] }
  sealed_by_branch = branch_log_ids.to_h { |id| [id, BranchProbe.sealed?(id)] }
  pending = pending_by_branch.values.sum
  sealed = sealed_by_branch.values.all?

  # Total children spawned per branch. Immutable once the branch is SEALED
  # (dispatch done), so we count it exactly ONCE and cache it on the metadata;
  # every later poll (and the dashboard) reuses the cached value, never recounting.
  # Unsealed (mid-spawn, count still climbing) => nil, and the dashboard falls back
  # to its capped live count until the seal freezes the total.
  logs_by_id = logs.index_by(&:id)
  spawned_by_branch = branch_log_ids.to_h do |id|
    cached = logs_by_id[id]&.metadata&.dig("poll", "spawned")
    [id, cached || (sealed_by_branch[id] ? BranchProbe.spawned(id).count : nil)]
  end

  if sealed && pending.zero?
    record_poll!(pending_by_branch, sealed_by_branch, token, next_poll_at: nil, interval: nil,
      rate_by_branch: {}, never_started_by_branch: {}, spawned_by_branch: spawned_by_branch, rekicked_by_branch: {})
    parent_job_class.constantize.perform_later(parent_key)
    return
  end

  # DISPATCHED (never-started) count per branch — the rekick drain signal. A
  # drop since the prior poll means workers are consuming this branch's queue,
  # so a still-queued child is in line; a flat count with stale never-started
  # children is a dropped job to recover. Keyed off this, NOT total pending,
  # which a wait/wait_until child completing would drop without any never-started
  # child moving (masking a genuinely-dropped one behind staggered waits).
  prev_never_started_by_branch = logs.to_h { |l| [l.id, l.metadata&.dig("poll", "never_started")] }
  never_started_by_branch = branch_log_ids.to_h { |id| [id, BranchProbe.never_started(id).count] }

  rekicked_by_branch = rekick_dropped_jobs(branch_log_ids, never_started_by_branch, prev_never_started_by_branch)

  # Cadence is driven by ESTIMATED TIME-TO-DRAIN, measured from the prior
  # poll's persisted pending. `motion` (EXISTS probes) is the fallback signal
  # when nothing completed this interval: :running => a live worker is
  # executing a child (hold the floor, it'll finish); :never_started => the only
  # motion is a queued/rekicked-but-unpicked child (back off exponentially,
  # it may never be picked up); :none => blocked/waiting (max backstop).
  # See reschedule_delay. Computed lazily below, only off the drain path.
  prior = logs.map { |l| l.metadata&.dig("poll") }
  # Only trust the AGGREGATE prev_pending when every requested branch log is
  # loaded AND carries a prior sample — otherwise `pending` (over all
  # branch_log_ids) and prev_pending (over loaded logs) would cover different
  # sets and yield a bogus aggregate rate. Missing/partial => no sample =>
  # bootstrap. Per-branch rate below is independently safe (missing => nil => 0).
  complete_prior = logs.size == branch_log_ids.size && prior.all?
  prev_pending = (prior.sum { |p| p["pending"].to_i } if complete_prior)
  prev_polled_at = prior.filter_map { |p| p && p["last_polled_at"] }.map { |s| Time.zone.parse(s) }.min
  elapsed = prev_polled_at && (Time.current - prev_polled_at)
  prev_delay = prior.filter_map { |p| p && p["interval"] }.max

  # Drain rate = children completed / second since the prior poll — THIS is the
  # throughput surfaced on the dashboard. Per branch for display; aggregated for
  # the ETA. Zero unless the branch actually drained (a no-headway / cold poll).
  # NOTE: the aggregate ETA blurs a heterogeneous multi-branch merge; acceptable
  # (the common case is single-branch; clamp + per-poll re-estimate bound any
  # skew, and only poll timing is affected — the parent is still woken).
  drained = ->(pend, prev) { prev && elapsed && elapsed > 0 && pend < prev }
  rate_by_branch = pending_by_branch.to_h do |id, pend|
    prev = prev_pending_by_branch[id]
    [id, drained.call(pend, prev) ? (prev - pend) / elapsed.to_f : 0.0]
  end
  rate = drained.call(pending, prev_pending) ? (prev_pending - pending) / elapsed.to_f : 0.0

  # Only needed when the ETA branch won't be taken (rate == 0); computing the
  # EXISTS probes lazily keeps them off the hot drain path. See reschedule_delay.
  motion = if rate > 0 then nil
  elsif branch_log_ids.any? { |id| BranchProbe.running?(id) } then :running
  elsif never_started_by_branch.values.any?(&:positive?) then :never_started
  else :none
  end

  delay = reschedule_delay(pending, rate, motion, prev_delay, min_interval, max_interval)
  record_poll!(pending_by_branch, sealed_by_branch, token, next_poll_at: delay.seconds.from_now,
    interval: delay, rate_by_branch: rate_by_branch, never_started_by_branch: never_started_by_branch,
    spawned_by_branch: spawned_by_branch, rekicked_by_branch: rekicked_by_branch)
  self.class.set(wait: delay.seconds)
    .perform_later(parent_key, parent_job_class, branch_log_ids, min_interval, max_interval, token)
end