Class: SolidObjects::EffectExecutor

Inherits:
Object
  • Object
show all
Defined in:
lib/solid_objects/effect_executor.rb,
sig/generated/lib/solid_objects/effect_executor.rbs

Constant Summary collapse

ACTOR_MESSAGE_EFFECT =

Returns:

  • (::String) 
"__actor_message__"

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(process_registry: ProcessRegistry.new, database_adapter: SolidObjects.database_adapter) ⇒ EffectExecutor

Returns a new instance of EffectExecutor.

RBS:

  • (?process_registry: ProcessRegistry, ?database_adapter: DatabaseAdapter) -> void

Parameters:

  • process_registry: (ProcessRegistry) (defaults to: ProcessRegistry.new)
  • database_adapter: (DatabaseAdapter) (defaults to: SolidObjects.database_adapter) 


18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/solid_objects/effect_executor.rb', line 18

def initialize(
  process_registry: ProcessRegistry.new,
  database_adapter: SolidObjects.database_adapter
)
  @process_registry = process_registry
  @database_adapter = database_adapter
  process_registry.register(kind: "effect")
  @stopped = false
  @shutdown_requested = false
  @polling_backoff = PollingBackoff.new(
    minimum_interval: SolidObjects.configuration.polling_interval,
    maximum_interval: SolidObjects.configuration.idle_polling_interval,
    on_change: ->(transition) do
      SolidObjects.instrument(
        :"polling.interval_changed",
        role: "effects",
        **transition
      )
    end
  )
end

Instance Attribute Details

#database_adapter ⇒ Object (readonly)

Returns the value of attribute database_adapter.

Returns:

  • (Object) 


112
113
114
 
# File 'lib/solid_objects/effect_executor.rb', line 112

def database_adapter
  @database_adapter
end

#polling_backoff ⇒ Object (readonly)

Returns the value of attribute polling_backoff.

Returns:

  • (Object) 


112
113
114
 
# File 'lib/solid_objects/effect_executor.rb', line 112

def polling_backoff
  @polling_backoff
end

#process_registry ⇒ Object (readonly)

Returns the value of attribute process_registry.

Returns:

  • (Object) 


112
113
114
 
# File 'lib/solid_objects/effect_executor.rb', line 112

def process_registry
  @process_registry
end

Instance Method Details

#claim_next ⇒ Effect?

RBS:

  • () -> Effect?

Returns:



115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/solid_objects/effect_executor.rb', line 115

def claim_next
  database_adapter.transaction do
    now = database_adapter.database_now
    effect = database_adapter.lock_candidates(
      Effect.where(status: "pending", available_at: ..now).order(:available_at, :id)
    ).first
    next unless effect

    effect.update!(
      status: "processing",
      attempt_count: effect.attempt_count + 1,
      claimed_by: process_registry.process_record.id,
      claimed_at: now
    )
    effect
  end
end

#complete(effect, result) ⇒ void

This method returns an undefined value.

RBS:

  • (Effect, untyped) -> void

Parameters:



174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/solid_objects/effect_executor.rb', line 174

def complete(effect, result)
  serialized_result = Serialization.dump(
    result,
    max_bytes: SolidObjects.configuration.max_result_bytes
  )
  result_message = nil
  database_adapter.transaction do
    locked_effect = Effect.lock.find(effect.id)
    verify_claim!(locked_effect)
    result_message = enqueue_result_message(
      effect: locked_effect,
      operation: locked_effect.success_operation,
      outcome: "success",
      arguments: {
        "effect_id" => locked_effect.effect_id,
        "arguments" => locked_effect.arguments,
        "result" => serialized_result
      }
    )
    locked_effect.update!(
      status: "completed",
      result: serialized_result,
      error: nil,
      completed_at: database_adapter.database_now,
      claimed_by: nil,
      claimed_at: nil
    )
  end
  Mailbox.new.announce(result_message) if result_message
  SolidObjects.instrument(
    :"effect.completed",
    effect_id: effect.effect_id,
    effect_name: effect.name,
    message_id: effect.message_id,
    attempt: effect.attempt_count
  )
end

#current_polling_interval ⇒ Float

RBS:

  • () -> Float

Returns:

  • (Float) 


106
107
108
 
# File 'lib/solid_objects/effect_executor.rb', line 106

def current_polling_interval
  polling_backoff.current_interval
end

#deliver(effect) ⇒ Object

RBS:

  • (Effect) -> untyped

Parameters:

Returns:

  • (Object) 


134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/solid_objects/effect_executor.rb', line 134

def deliver(effect)
  return deliver_actor_message(effect) if effect.name == ACTOR_MESSAGE_EFFECT

  handler = SolidObjects.effect_registry.fetch(effect.name)
  context = EffectContext.new(
    id: effect.effect_id,
    attempt: effect.attempt_count,
    source_message_id: effect.message_id,
    actor_type: effect.instance.actor_type,
    actor_id: effect.instance.actor_id
  )
  handler.call(effect.arguments, context)
end

#deliver_actor_message(effect) ⇒ Integer

RBS:

  • (Effect) -> Integer

Parameters:

Returns:

  • (Integer) 


149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/solid_objects/effect_executor.rb', line 149

def deliver_actor_message(effect)
  arguments = effect.arguments
  reference = Reference.new(
    actor_type: arguments.fetch("actor_type"),
    actor_id: arguments.fetch("actor_id")
  )
  actor_class = SolidObjects.registry.fetch(reference.actor_type)
  operation = arguments.fetch("operation")
  unless actor_class.definition.messages.key?(operation.to_sym)
    raise UnknownMessage, "unknown actor-to-actor operation #{operation.inspect}"
  end

  available_at = arguments["available_at"]
  available_at = Time.iso8601(available_at) if available_at
  Mailbox.new.enqueue(
    reference:,
    operation:,
    arguments: arguments.fetch("arguments"),
    delivery_mode: "internal",
    available_at:,
    idempotency_key: effect.effect_id
  ).id
end

#enqueue_result_message(effect:, operation:, outcome:, arguments:) ⇒ Message?

RBS:

  • (effect: Effect, operation: String?, outcome: String, arguments: Hash[String, untyped]) -> Message?

Parameters:

  • effect: (Effect)
  • operation: (String, nil)
  • outcome: (String)
  • arguments: (Hash[String, untyped])

Returns:



259
260
261
262
263
264
265
266
267
268
269
270
271
272
 
# File 'lib/solid_objects/effect_executor.rb', line 259

def enqueue_result_message(effect:, operation:, outcome:, arguments:)
  return unless operation

  Mailbox.new.enqueue_in_transaction(
    reference: Reference.new(
      actor_type: effect.instance.actor_type,
      actor_id: effect.instance.actor_id
    ),
    operation:,
    arguments:,
    delivery_mode: "internal",
    idempotency_key: "effect:#{effect.effect_id}:#{outcome}"
  )
end

#fail_effect(effect, error) ⇒ void

This method returns an undefined value.

RBS:

  • (Effect, Exception) -> void

Parameters:



213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
 
# File 'lib/solid_objects/effect_executor.rb', line 213

def fail_effect(effect, error)
  result_message = nil
  database_adapter.transaction do
    locked_effect = Effect.lock.find(effect.id)
    verify_claim!(locked_effect)
    dead = locked_effect.attempt_count >= locked_effect.max_attempts
    error_details = {
      "class" => error.class.name,
      "message" => error.message.to_s.byteslice(0, 8_192),
      "backtrace" => Array(error.backtrace).first(50)
    }
    if dead
      result_message = enqueue_result_message(
        effect: locked_effect,
        operation: locked_effect.failure_operation,
        outcome: "failure",
        arguments: {
          "effect_id" => locked_effect.effect_id,
          "arguments" => locked_effect.arguments,
          "error" => error_details
        }
      )
    end
    locked_effect.update!(
      status: dead ? "dead" : "pending",
      available_at: database_adapter.database_now +
        SolidObjects.configuration.retry_delay.call(locked_effect.attempt_count),
      error: error_details,
      claimed_by: nil,
      claimed_at: nil
    )
  end
  Mailbox.new.announce(result_message) if result_message
rescue ActiveRecord::RecordNotFound, LostActivation
  nil
end

#request_shutdown ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



90
91
92
93
 
# File 'lib/solid_objects/effect_executor.rb', line 90

def request_shutdown
  @shutdown_requested = true
  SolidObjects.wake_up.signal
end

#run ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/solid_objects/effect_executor.rb', line 66

def run
  ProcessRegistry.warn_if_polling_is_only_cross_process_wake_up

  until shutdown_requested?
    wake_up = SolidObjects.wake_up
    watch = wake_up.respond_to?(:watch) ? wake_up.watch : wake_up
    worked = run_once
    if worked
      polling_backoff.reset(:work)
      next
    end

    notified = watch.wait(timeout: current_polling_interval)
    if notified == false
      polling_backoff.record_idle
    else
      polling_backoff.reset(:wake_up)
    end
  end
ensure
  stop
end

#run_once ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/solid_objects/effect_executor.rb', line 41

def run_once
  return false if stopped?

  process_registry.heartbeat
  effect = claim_next
  return false unless effect

  result = deliver(effect)
  complete(effect, result)
  true
rescue => error
  fail_effect(effect, error) if effect
  false
end

#shutdown_requested? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


101
102
103
 
# File 'lib/solid_objects/effect_executor.rb', line 101

def shutdown_requested?
  @shutdown_requested
end

#stop ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



57
58
59
60
61
62
63
 
# File 'lib/solid_objects/effect_executor.rb', line 57

def stop
  return if stopped?

  @stopped = true
  process_registry.start_draining
  process_registry.stop
end

#stopped? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


96
97
98
 
# File 'lib/solid_objects/effect_executor.rb', line 96

def stopped?
  @stopped
end

#verify_claim!(effect) ⇒ void

This method returns an undefined value.

RBS:

  • (Effect) -> void

Parameters:



251
252
253
254
255
256
 
# File 'lib/solid_objects/effect_executor.rb', line 251

def verify_claim!(effect)
  return if effect.status == "processing" &&
    effect.claimed_by == process_registry.process_record.id

  raise LostActivation, "effect claim changed"
end