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) 


15
16
17
18
19
20
21
22
23
24
 
# File 'lib/solid_objects/effect_executor.rb', line 15

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
end

Instance Attribute Details

#database_adapter ⇒ Object (readonly)

Returns the value of attribute database_adapter.

Returns:

  • (Object) 


81
82
83
 
# File 'lib/solid_objects/effect_executor.rb', line 81

def database_adapter
  @database_adapter
end

#process_registry ⇒ Object (readonly)

Returns the value of attribute process_registry.

Returns:

  • (Object) 


81
82
83
 
# File 'lib/solid_objects/effect_executor.rb', line 81

def process_registry
  @process_registry
end

Instance Method Details

#claim_next ⇒ Effect?

RBS:

  • () -> Effect?

Returns:



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/solid_objects/effect_executor.rb', line 84

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:



143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/solid_objects/effect_executor.rb', line 143

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(
      locked_effect,
      locked_effect.success_message_name,
      "success",
      { "effect_id" => locked_effect.effect_id, "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

#deliver(effect) ⇒ Object

RBS:

  • (Effect) -> untyped

Parameters:

Returns:

  • (Object) 


103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/solid_objects/effect_executor.rb', line 103

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) 


118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/solid_objects/effect_executor.rb', line 118

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)
  message_name = arguments.fetch("message_name")
  unless actor_class.definition.messages.key?(message_name.to_sym)
    raise UnknownMessage, "unknown actor-to-actor message #{message_name.inspect}"
  end

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

#enqueue_result_message(effect, message_name, outcome, arguments) ⇒ Message?

RBS:

  • (Effect, String?, String, Hash[String, untyped]) -> Message?

Parameters:

  • (Effect)
  • (String, nil)
  • (String)
  • (Hash[String, untyped])

Returns:



220
221
222
223
224
225
226
227
228
229
230
231
232
233
 
# File 'lib/solid_objects/effect_executor.rb', line 220

def enqueue_result_message(effect, message_name, outcome, arguments)
  return unless message_name

  Mailbox.new.enqueue_in_transaction(
    Reference.new(
      actor_type: effect.instance.actor_type,
      actor_id: effect.instance.actor_id
    ),
    message_name,
    arguments,
    kind: "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:



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
 
# File 'lib/solid_objects/effect_executor.rb', line 178

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(
        locked_effect,
        locked_effect.failure_message_name,
        "failure",
        { "effect_id" => locked_effect.effect_id, "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



64
65
66
67
 
# File 'lib/solid_objects/effect_executor.rb', line 64

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

#run ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



52
53
54
55
56
57
58
59
60
61
 
# File 'lib/solid_objects/effect_executor.rb', line 52

def run
  until shutdown_requested?
    worked = run_once
    next if worked

    SolidObjects.wake_up.wait(timeout: SolidObjects.configuration.polling_interval)
  end
ensure
  stop
end

#run_once ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/solid_objects/effect_executor.rb', line 27

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) 


75
76
77
 
# File 'lib/solid_objects/effect_executor.rb', line 75

def shutdown_requested?
  @shutdown_requested
end

#stop ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



43
44
45
46
47
48
49
 
# File 'lib/solid_objects/effect_executor.rb', line 43

def stop
  return if stopped?

  @stopped = true
  process_registry.start_draining
  process_registry.stop
end

#stopped? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


70
71
72
 
# File 'lib/solid_objects/effect_executor.rb', line 70

def stopped?
  @stopped
end

#verify_claim!(effect) ⇒ void

This method returns an undefined value.

RBS:

  • (Effect) -> void

Parameters:



212
213
214
215
216
217
 
# File 'lib/solid_objects/effect_executor.rb', line 212

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

  raise LostActivation, "effect claim changed"
end