Class: Ductwork::Branch

Inherits:
Record
  • Object
show all
Defined in:
lib/ductwork/models/branch.rb

Overview

rubocop:todo Metrics/ClassLength

Defined Under Namespace

Classes: TransitionError

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.with_latest_claimed(pipeline_klass) ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/ductwork/models/branch.rb', line 50

def self.with_latest_claimed(pipeline_klass)
  branch_claim = Ductwork::BranchClaim.new(pipeline_klass)
  branch = branch_claim.latest

  if branch.present?
    Ductwork::FaultInjection.checkpoint(:after_branch_claim)

    yield branch, branch_claim.transition, branch_claim.advancement

    true
  else
    false
  end
ensure
  advancement = branch_claim.advancement

  if advancement&.persisted? && advancement.completed_at.nil?
    advancement.thread_crashed!(branch_claim.token)
  end
end

Instance Method Details

#advance!(transition, advancement) ⇒ Object 



103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/ductwork/models/branch.rb', line 103

def advance!(transition, advancement)
  # NOTE: capture the claim token now, while it is guaranteed live, so the
  # fence can survive an in-block mutation + rollback (see `with_claim_fence`)
  @claim_fence_token = claim_token
  step = latest_step
  max_crash = Ductwork.configuration.pipeline_advancer_max_crash

  # NOTE: the crash cap is checked first as the true backstop against a
  # poison branch that repeatedly crashes the advancer process/thread. In a
  # normal failed-step halt no advancer crashes have accrued, so this only
  # fires on a genuine crash loop.
  if advancement.crash_count >= max_crash
    halt_branch_and_resolve_run!(transition, advancement, "advancer_crashes_exhausted")
  elsif step.failed?
    halt_branch_and_resolve_run!(transition, advancement, failed_step_halt_reason(step))
  else
    route_by_edge(transition, advancement)
  end

  # NOTE: this is a no-op unless this advancement just halted the whole run
  run.dispatch_on_halt!
end

#complete! ⇒ Object 



126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/ductwork/models/branch.rb', line 126

def complete!
  update!(
    completed_at: Time.current,
    status: "completed",
    claimed_for_advancing_at: nil,
    claim_token: nil,
    last_advanced_at: Time.current
  )

  Ductwork.logger.info(
    msg: "Branch completed",
    branch_id: id,
    role: :pipeline_advancer
  )
end

#halt!(halt_reason) ⇒ Object 



142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/ductwork/models/branch.rb', line 142

def halt!(halt_reason)
  self.halt_reason = halt_reason

  update!(
    status: "halted",
    claimed_for_advancing_at: nil,
    claim_token: nil,
    last_advanced_at: Time.current
  )

  Ductwork.logger.info(
    msg: "Branch halted",
    branch_id: id,
    role: :pipeline_advancer
  )
end

#latest_step ⇒ Object 



159
160
161
 
# File 'lib/ductwork/models/branch.rb', line 159

def latest_step
  steps.order(started_at: :desc, id: :desc).limit(1).first
end

#release!(expected_token = @claim_fence_token || claim_token) ⇒ Object

NOTE: the default expected_token is the token captured in advance!, not the live claim_token attribute, for the same reason as with_claim_fence: complete!/halt! null the in-memory attribute mid-transition, and after a rollback (e.g. a run-row deadlock) the DB token is restored while the attribute stays nil. Releasing with the stale nil would match zero rows and silently leave the branch stranded in advancing. Callers outside an advancement (fresh branch objects, explicit tokens) are unaffected.



170
171
172
173
174
175
176
177
178
179
 
# File 'lib/ductwork/models/branch.rb', line 170

def release!(expected_token = @claim_fence_token || claim_token)
  Ductwork::Branch
    .where(id: id, claim_token: expected_token, status: :advancing)
    .update_all(
      claimed_for_advancing_at: nil,
      claim_token: nil,
      status: :in_progress,
      last_advanced_at: Time.current
    )
end

#with_claim_fence(&block) ⇒ Object

NOTE: claim divergence (the reaper released the branch and another advancer reclaimed it) is the expected outcome of a race, not an error, so the fence treats it as such: it logs and returns false rather than raising. Every branch/run mutation must run inside the fence. The row is locked for the block's duration, so divergence can only be observed at entry, never mid-block, and nested fences on the same row always hold. Returns true when the block ran, false when the claim had diverged.

NOTE: the fence compares against @claim_fence_token (captured in advance! before any mutation), NOT the live claim_token attribute. complete!/halt! null the in-memory claim_token mid-block; if the surrounding transaction then rolls back (e.g. a deadlock on the run row), the DB restores the real token but the in-memory attribute stays nil. A later fence in the same advancement (the error-recovery path) would then compare the real DB token against nil, wrongly conclude the claim diverged, skip recovery, and strand the branch in advancing. Anchoring to the captured token still detects a genuine divergence (another advancer reclaimed it => the DB token changed to a value we never held).



89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/ductwork/models/branch.rb', line 89

def with_claim_fence(&block)
  fence_token = @claim_fence_token || claim_token

  Ductwork::Record.transaction do
    if self.class.where(id:).lock.pick(:claim_token) == fence_token
      block.call
      true
    else
      log_claim_diverged
      false
    end
  end
end