Class: SolidObjects::Activation

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(lease:) ⇒ Activation

Returns a new instance of Activation.

RBS:

  • (lease: Lease) -> void

Parameters:



14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/solid_objects/activation.rb', line 14

def initialize(lease:)
  @lease = lease
  instance = SolidObjects.database_adapter.with_lock_retry do
    Instance.find(lease.instance_id)
  end
  @actor_class = SolidObjects.registry.fetch(instance.actor_type)
  @actor = build_actor(instance)
  @last_used_at = monotonic_now
  @pass_exhausted = false
  actor.activate
  SolidObjects.instrument(
    :"activation.started",
    instance_id: instance.id,
    actor_type: instance.actor_type,
    actor_id: instance.actor_id,
    owner_id: lease.owner_id,
    generation: lease.generation
  )
end

Instance Attribute Details

#actor ⇒ Object (readonly)

RBS:

  • @lease: Lease

  • @actor_class: Class

  • @actor: Actor

  • @last_used_at: Float

  • @pass_exhausted: bool

Returns:

  • (Object) 


11
12
13
 
# File 'lib/solid_objects/activation.rb', line 11

def actor
  @actor
end

#actor_class ⇒ Object (readonly)

Returns the value of attribute actor_class.

Returns:

  • (Object) 


121
122
123
 
# File 'lib/solid_objects/activation.rb', line 121

def actor_class
  @actor_class
end

#lease ⇒ Object (readonly)

RBS:

  • @lease: Lease

  • @actor_class: Class

  • @actor: Actor

  • @last_used_at: Float

  • @pass_exhausted: bool

Returns:

  • (Object) 


11
12
13
 
# File 'lib/solid_objects/activation.rb', line 11

def lease
  @lease
end

Instance Method Details

#build_actor(instance) ⇒ Actor

RBS:

  • (Instance) -> Actor

Parameters:

Returns:



151
152
153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/solid_objects/activation.rb', line 151

def build_actor(instance)
  state_data = ApplicationWriteGuard.call(
    actor_type: instance.actor_type,
    actor_id: instance.actor_id,
    operation: "state_migration"
  ) do
    actor_class.definition.migrate_state(instance.state_version, instance.state)
  end
  actor_class.new(
    actor_id: instance.actor_id,
    state: State.new(actor_class.definition.state_definition, state_data)
  )
end

#claim_next_message ⇒ Message?

RBS:

  • () -> Message?

Returns:



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

def claim_next_message
  lease.fenced_transaction do |instance|
    recover_stale_claim(instance)
    ready_message = ReadyMessage
      .where(instance_id: instance.id)
      .order(:sequence)
      .first
    next unless ready_message

    now = SolidObjects.database_adapter.database_now
    next if ready_message.available_at > now

    message = Message.lock.find(ready_message.message_id)
    ready_message.destroy!
    message.update!(
      attempt_count: message.attempt_count + 1,
      started_at: now
    )
    ClaimedMessage.create!(
      message:,
      instance:,
      process_id: lease.owner_id,
      activation_token: lease.activation_token,
      activation_generation: lease.generation,
      claimed_at: now
    )
    message
  end
end

#deactivate ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/solid_objects/activation.rb', line 96

def deactivate
  actor.deactivate
rescue => error
  SolidObjects.instrument(
    :"activation.deactivation_failed",
    instance_id: lease.instance_id,
    actor_type: actor.class.actor_type,
    actor_id: actor.actor_id,
    owner_id: lease.owner_id,
    generation: lease.generation,
    error_class: error.class.name,
    error_message: error.message
  )
  SolidObjects.configuration.logger.error(
    "SolidObjects activation deactivation failed " \
      "actor_type=#{actor.class.actor_type.inspect} " \
      "actor_id=#{actor.actor_id.inspect} " \
      "error_class=#{error.class.name}"
  )
ensure
  release_lease
end

#drain ⇒ Integer

RBS:

  • () -> Integer

Returns:

  • (Integer) 


35
36
37
 
# File 'lib/solid_objects/activation.rb', line 35

def drain
  drain_messages
end

#drain_messages(message_id: nil, deadline: nil) ⇒ Integer

RBS:

  • (?message_id: Integer?, ?deadline: Float?) -> Integer

Parameters:

  • message_id: (Integer, nil) (defaults to: nil)
  • deadline: (Float, nil) (defaults to: nil)

Returns:

  • (Integer) 


124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/solid_objects/activation.rb', line 124

def drain_messages(message_id: nil, deadline: nil)
  processed_count = 0
  started_at = monotonic_now
  @pass_exhausted = false

  loop do
    break if deadline && monotonic_now >= deadline

    if processed_count >= SolidObjects.configuration.max_messages_per_activation_pass ||
        monotonic_now - started_at >= SolidObjects.configuration.max_activation_duration
      @pass_exhausted = true
      break
    end

    message = claim_next_message
    break unless message

    Executor.new(activation: self, message:).call
    processed_count += 1
    @last_used_at = monotonic_now
    break if message.id == message_id
  end

  processed_count
end

#drain_until(message_id:, deadline:) ⇒ Integer

RBS:

  • (message_id: Integer, deadline: Float) -> Integer

Parameters:

  • message_id: (Integer)
  • deadline: (Float)

Returns:

  • (Integer) 


40
41
42
 
# File 'lib/solid_objects/activation.rb', line 40

def drain_until(message_id:, deadline:)
  drain_messages(message_id:, deadline:)
end

#idle? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


62
63
64
 
# File 'lib/solid_objects/activation.rb', line 62

def idle?
  monotonic_now - @last_used_at >= SolidObjects.configuration.idle_deactivation_timeout
end

#lease_renewal_due? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


72
73
74
75
 
# File 'lib/solid_objects/activation.rb', line 72

def lease_renewal_due?
  lease.expires_at - SolidObjects.database_adapter.database_now <=
    SolidObjects.configuration.lease_renewal_interval
end

#monotonic_now ⇒ Float

RBS:

  • () -> Float

Returns:

  • (Float) 


228
229
230
 
# File 'lib/solid_objects/activation.rb', line 228

def monotonic_now
  ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
end

#pass_exhausted? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


57
58
59
 
# File 'lib/solid_objects/activation.rb', line 57

def pass_exhausted?
  @pass_exhausted
end

#ready? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


45
46
47
48
49
50
51
52
53
54
 
# File 'lib/solid_objects/activation.rb', line 45

def ready?
  now = SolidObjects.database_adapter.database_now
  ReadyMessage.where(instance_id: lease.instance_id, available_at: ..now).exists? ||
    ClaimedMessage.where(
      instance_id: lease.instance_id,
      process_id: lease.owner_id,
      activation_token: lease.activation_token,
      activation_generation: lease.generation
    ).exists?
end

#recover_stale_claim(instance) ⇒ void

This method returns an undefined value.

RBS:

  • (Instance) -> void

Parameters:



204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'lib/solid_objects/activation.rb', line 204

def recover_stale_claim(instance)
  claimed_message = ClaimedMessage
    .joins(:message)
    .where(instance_id: instance.id)
    .order("#{Message.table_name}.sequence")
    .first
  return unless claimed_message
  return if claimed_message.process_id == lease.owner_id &&
    claimed_message.activation_token == lease.activation_token &&
    claimed_message.activation_generation == lease.generation

  message = claimed_message.message
  claimed_message.destroy!
  return if message.completed? || message.dead?

  ReadyMessage.create!(
    message:,
    instance:,
    sequence: message.sequence,
    available_at: SolidObjects.database_adapter.database_now
  )
end

#release_lease ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



166
167
168
169
170
 
# File 'lib/solid_objects/activation.rb', line 166

def release_lease
  lease.release
rescue LostActivation
  nil
end

#renew_lease ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



67
68
69
 
# File 'lib/solid_objects/activation.rb', line 67

def renew_lease
  @lease = lease.renew
end

#restore_state(state_data) ⇒ void

This method returns an undefined value.

RBS:

  • (Hash[String, untyped]) -> void

Parameters:

  • (Hash[String, untyped]) 


88
89
90
91
92
93
 
# File 'lib/solid_objects/activation.rb', line 88

def restore_state(state_data)
  @actor = actor_class.new(
    actor_id: actor.actor_id,
    state: State.new(actor_class.definition.state_definition, state_data)
  )
end

#yield_ready_messages ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



78
79
80
81
82
83
84
85
 
# File 'lib/solid_objects/activation.rb', line 78

def yield_ready_messages
  SolidObjects.database_adapter.transaction do
    now = SolidObjects.database_adapter.database_now
    ReadyMessage
      .where(instance_id: lease.instance_id, available_at: ..now)
      .update_all(available_at: now)
  end
end