Class: GenevaDrive::Signal

Inherits:
ActiveRecord::Base
  • Object
show all
Defined in:
lib/geneva_drive/signal.rb

Overview

A per-workflow event record. Signals are how the outside world tells a workflow that something happened: a webhook landed, a human clicked "approve", another system finished its part of the job.

A signal is always addressed to one workflow instance and is persisted before any processing happens, so it can never be lost between arriving and being noticed - a signal that arrives before the waiting step even exists simply sits in the table until the step's gate picks it up.

The lifecycle, in the state column, is pending -> claimed -> consumed:

  • pending: persisted, nobody is working on it
  • claimed: attached to at least one step execution, at least one of which has not resolved yet
  • consumed: every attached chain resolved cleanly; no new execution may attach

Alongside the state there are two counters, claimed and consumed: how many executions have attached to this event, and how many attached chains have resolved cleanly.

Examples:

Delivering a signal

workflow = OrderFulfillmentWorkflow.ongoing.for_hero(order).first
workflow.signal!(:payment_confirmed, payload: {amount_cents: 12_500})

Constant Summary collapse

ACTIVE_EXECUTION_STATES =

Step execution states that mean "this chain has not come to rest yet".

%w[waiting scheduled in_progress].freeze
CLEAN_OUTCOMES =

Outcomes that resolve an attached chain cleanly, so the signal can be considered handled by it.

%w[success skipped].freeze

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.reset_table_available_cache! ⇒ void

This method returns an undefined value.

Clears the cached table detection result. Call this in tests or after running migrations in-process so the next access re-checks.



83
84
85
 
# File 'lib/geneva_drive/signal.rb', line 83

def reset_table_available_cache!
  remove_instance_variable(:@_table_available) if defined?(@_table_available)
end

.serialize_payload(value) ⇒ Object?

Serializes a payload using ActiveJob serializers (handles Date, Time, and other types) and enforces GenevaDrive.max_signal_payload_size on the serialized JSON. The single serialization path for payload writes.

Parameters:

  • value (Object, nil) —

    the payload

Returns:

  • (Object, nil) —

    the serialized payload

Raises:



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/geneva_drive/signal.rb', line 94

def serialize_payload(value)
  return nil if value.nil?

  serialized = ActiveJob::Arguments.serialize([value]).first

  limit = GenevaDrive.max_signal_payload_size
  if limit
    bytesize = JSON.generate(serialized).bytesize
    if bytesize > limit
      raise GenevaDrive::SignalPayloadTooLargeError,
        "Serialized signal payload is #{bytesize} bytes, exceeding " \
        "GenevaDrive.max_signal_payload_size (#{limit} bytes). A payload describes the " \
        "event, it is not a place to ship the data the step should be working on - " \
        "write what matters onto the hero instead. Set " \
        "GenevaDrive.max_signal_payload_size to nil to disable this check."
    end
  end

  serialized
end

.table_available? ⇒ Boolean

Lazily checks whether the signals table has been migrated. Never hits the database at class definition time - only on the first runtime call.

Deployments usually ship the gem update before running migrations, so everything except actually sending or waiting for a signal must keep working when the table is absent.

Returns:

  • (Boolean) 


71
72
73
74
75
76
77
 
# File 'lib/geneva_drive/signal.rb', line 71

def table_available?
  if defined?(@_table_available)
    return @_table_available
  end

  @_table_available = table_exists?
end

Instance Method Details

#claim! ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Records one execution attaching to this signal, bumping the claimed count. The pending -> claimed state flip and claimed_at happen on the first claim only, so the timestamp keeps meaning "when this event started being handled". A retry attaching to a still-claimed signal counts as a new claim; a successor continuing the same chain does not (it carries the pin over rather than acquiring it).



180
181
182
183
184
185
186
187
 
# File 'lib/geneva_drive/signal.rb', line 180

def claim!
  attrs = {claimed: claimed + 1}
  if state == "pending"
    attrs[:state] = "claimed"
    attrs[:claimed_at] = Time.current
  end
  update!(attrs)
end

#claimed? ⇒ Boolean

Whether at least one execution has ever attached to this signal. Reads the counter, not the lifecycle state, so it stays true after the signal has been consumed - "this event was picked up" rather than "is being handled right now".

Returns:

  • (Boolean) 


162
 
# File 'lib/geneva_drive/signal.rb', line 162

def claimed? = claimed > 0

#consumed? ⇒ Boolean

Whether at least one attached chain has resolved cleanly. Reads the counter, so for a one-to-many dispatch it goes true with the first resolved branch, while the state only flips once the last one resolves.

Returns:

  • (Boolean) 


169
 
# File 'lib/geneva_drive/signal.rb', line 169

def consumed? = consumed > 0

#duplicate_delivery! ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Flags this instance as the result of a deduplicated delivery.



152
153
154
 
# File 'lib/geneva_drive/signal.rb', line 152

def duplicate_delivery!
  @duplicate_delivery = true
end

#duplicate_delivery? ⇒ Boolean

Whether this instance was returned from a deduplicated delivery - that is, Workflow#signal! found an existing row with the same idempotency key instead of inserting a new one. Not a column: it is a fact about this call, not about the record.

Returns:

  • (Boolean) 


144
145
146
 
# File 'lib/geneva_drive/signal.rb', line 144

def duplicate_delivery?
  !!@duplicate_delivery
end

#fully_resolved?(resolved_step_names: []) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Whether every execution chain attached to this signal has come to a clean stop. A chain that is still running (or parked), and one whose latest attached execution failed or was canceled, both keep the state at claimed - the failed chain's retry re-attaches to it and reads the same payload.

Only the most recent attached execution per step counts: the earlier ones in a chain end with continued or reattempted precisely because they handed the work to the next one.

Parameters:

  • resolved_step_names (Array<String, Symbol>) (defaults to: []) —

    steps to treat as resolved whatever their executions say. Skipping a step is a decision that its chain is over, including when that chain ended in a failure - no future execution of it is coming to resolve it cleanly.

Returns:

  • (Boolean) 


224
225
226
227
228
229
230
231
232
233
234
235
 
# File 'lib/geneva_drive/signal.rb', line 224

def fully_resolved?(resolved_step_names: [])
  settled = Array(resolved_step_names).map(&:to_s)
  attached = step_executions.order(created_at: :asc, id: :asc).to_a
  return true if attached.empty?

  unsettled = attached.reject { |execution| settled.include?(execution.step_name) }
  return false if unsettled.any? { |execution| ACTIVE_EXECUTION_STATES.include?(execution.state) }

  unsettled.group_by(&:step_name).all? do |_step_name, chain|
    CLEAN_OUTCOMES.include?(chain.last.outcome)
  end
end

#payload ⇒ Object?

Returns the deserialized payload. Hashes come back with indifferent access, because webhook senders produce string keys while Ruby senders produce symbols and matchers should not have to know which.

Returns:

  • (Object, nil) —

    the payload



121
122
123
124
125
126
127
 
# File 'lib/geneva_drive/signal.rb', line 121

def payload
  raw = self[:payload]
  return nil if raw.nil?

  value = ActiveJob::Arguments.deserialize([raw]).first
  value.is_a?(Hash) ? ActiveSupport::HashWithIndifferentAccess.new(value) : value
end

#payload=(value) ⇒ void

This method returns an undefined value.

Sets the payload, serializing it through ActiveJob serializers.

Parameters:

  • value (Object, nil) —

    the payload

Raises:



134
135
136
 
# File 'lib/geneva_drive/signal.rb', line 134

def payload=(value)
  self[:payload] = self.class.serialize_payload(value)
end

#record_consumption!(resolved_step_names: []) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Records one attached execution chain resolving cleanly, bumping the consumed count. The state flips to consumed - closing the signal to new attachments - only once every attached chain has resolved, which for a single claimant is the same moment.

Parameters:

  • resolved_step_names (Array<String, Symbol>) (defaults to: []) —

    steps whose chains are resolved by decree rather than by outcome - a skip settles the chain it skips past even when that chain's last execution failed



199
200
201
202
203
204
205
206
 
# File 'lib/geneva_drive/signal.rb', line 199

def record_consumption!(resolved_step_names: [])
  attrs = {consumed: consumed + 1}
  if state == "claimed" && fully_resolved?(resolved_step_names: resolved_step_names)
    attrs[:state] = "consumed"
    attrs[:consumed_at] = Time.current
  end
  update!(attrs)
end

#step_executions ⇒ ActiveRecord::Relation<GenevaDrive::StepExecution>

Step executions pinned to this signal.

Returns:



240
241
242
 
# File 'lib/geneva_drive/signal.rb', line 240

def step_executions
  GenevaDrive::StepExecution.where(signal_id: id)
end