Class: Ductwork::Run

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

Instance Method Summary collapse

Instance Method Details

#dispatch_on_halt! ⇒ Object

NOTE: Fires the host on_halt handler (at-most) once after the run has halted

The handler is an observable side effect (it may page, refund, or emit an external event), so it must run only after the halt is durably committed and outside the advancer's with_claim_fence transaction; otherwise a rolled-back commit (deadlock victim, dropped connection) would leave it spuriously fired and double-firing on re-claim. Branch#advance! calls this once the advancement has committed.

The dispatch is claimed with an atomic UPDATE ... WHERE: only the advancer that flips on_halt_dispatched_at from NULL runs the handler, and only while the run is actually halted (a rolled-back halt never persists status). That makes the handler at-most-once even across concurrent advancers and re-claims.



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/ductwork/models/run.rb', line 82

def dispatch_on_halt!
  klass = parsed_definition.dig(:metadata, :on_halt, :klass)

  return if klass.blank?

  claimed = self.class
                .where(id: id, status: "halted", on_halt_dispatched_at: nil)
                .update_all(on_halt_dispatched_at: Time.current)

  return if claimed.zero?

  begin
    reasons = branches.halted.pluck(:halt_reason)

    Object.const_get(klass).new(reasons).execute
  rescue StandardError => e
    Ductwork.logger.error(
      msg: "on_halt handler errored",
      run_id: id,
      error_klass: e.class.to_s,
      error_message: e.message
    )
  end
end

#parsed_definition ⇒ Object 



35
36
37
 
# File 'lib/ductwork/models/run.rb', line 35

def parsed_definition
  @parsed_definition ||= JSON.parse(definition).with_indifferent_access
end

#resolve_terminal_state! ⇒ Object 



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
 
# File 'lib/ductwork/models/run.rb', line 39

def resolve_terminal_state!
  Ductwork::Record.transaction do
    lock_for_terminal_resolution!

    next if halted? || completed?
    next if branches.where.not(status: %w[completed halted]).exists?

    if branches.halted.exists?
      pipeline.update!(status: "halted")
      update!(status: "halted", halted_at: Time.current)

      Ductwork.logger.warn(
        msg: "Pipeline halted",
        pipeline_id: pipeline.id,
        run_id: id
      )
    else
      pipeline.update!(status: "completed")
      update!(status: "completed", completed_at: Time.current)

      Ductwork.logger.info(
        msg: "Pipeline completed",
        pipeline_id: pipeline.id,
        run_id: id
      )
    end
  end
end