Class: SolidObjects::ProcessRegistry

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

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ProcessRegistry

Returns a new instance of ProcessRegistry.

RBS:

  • () -> void



76
77
78
79
 
# File 'lib/solid_objects/process_registry.rb', line 76

def initialize
  @process_record = nil
  @last_heartbeat_at = nil
end

Instance Attribute Details

#process_record ⇒ Object (readonly)

RBS:

  • @process_record: Process?

  • @last_heartbeat_at: Float?

Returns:

  • (Object) 


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

def process_record
  @process_record
end

Class Method Details

.cleanup_dead(now: SolidObjects.database_adapter.database_now) ⇒ Integer

RBS:

  • (?now: Time) -> Integer

Parameters:

  • now: (Time) (defaults to: SolidObjects.database_adapter.database_now)

Returns:

  • (Integer) 


9
10
11
12
13
14
15
16
17
18
 
# File 'lib/solid_objects/process_registry.rb', line 9

def cleanup_dead(now: SolidObjects.database_adapter.database_now)
  stale_at = now - SolidObjects.configuration.process_alive_threshold
  dead_processes = Process
    .where.not(shutdown_state: "stopped")
    .where(last_heartbeat_at: ..stale_at)
    .to_a

  dead_processes.each { |process_record| cleanup_process(process_record, now) }
  dead_processes.length
end

.deregister(process_record, now: SolidObjects.database_adapter.database_now) ⇒ Boolean

RBS:

  • (Process, ?now: Time) -> bool

Parameters:

  • (Process)
  • now: (Time) (defaults to: SolidObjects.database_adapter.database_now)

Returns:

  • (Boolean) 


21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/solid_objects/process_registry.rb', line 21

def deregister(process_record, now: SolidObjects.database_adapter.database_now)
  SolidObjects.database_adapter.transaction do
    Instance.where(activation_owner_id: process_record.id).update_all(
      activation_owner_id: nil,
      activation_token: nil,
      activation_expires_at: nil
    )
    ClaimedMessage.where(process_id: process_record.id).update_all(
      process_id: nil,
      activation_token: nil
    )
    Effect.where(claimed_by: process_record.id).update_all(
      status: "pending",
      claimed_by: nil,
      claimed_at: nil,
      available_at: now
    )
    Reminder.where(claimed_by: process_record.id).update_all(
      claimed_by: nil,
      claimed_at: nil
    )
    Broadcast.where(claimed_by: process_record.id).update_all(
      status: "pending",
      claimed_by: nil,
      claimed_at: nil,
      available_at: now
    )
    process_record.update!(
      shutdown_state: "stopped",
      stopped_at: now
    )
  end
  true
end

Instance Method Details

#default_metadata ⇒ Hash[Symbol, String]

RBS:

  • () -> Hash[Symbol, String]

Returns:

  • (Hash[Symbol, String]) 


134
135
136
137
138
139
 
# File 'lib/solid_objects/process_registry.rb', line 134

def default_metadata
  {
    solid_objects_version: SolidObjects::VERSION,
    ruby_version: RUBY_VERSION
  }
end

#heartbeat ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/solid_objects/process_registry.rb', line 101

def heartbeat
  return false unless process_record
  return false if heartbeat_recent?

  updated = SolidObjects.database_adapter.with_lock_retry do
    process_record.update(
      last_heartbeat_at: SolidObjects.database_adapter.database_now
    )
  end
  @last_heartbeat_at = monotonic_now if updated
  updated
end

#heartbeat_recent? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


142
143
144
145
146
147
 
# File 'lib/solid_objects/process_registry.rb', line 142

def heartbeat_recent?
  return false unless @last_heartbeat_at

  monotonic_now - @last_heartbeat_at <
    SolidObjects.configuration.process_heartbeat_interval
end

#monotonic_now ⇒ Float

RBS:

  • () -> Float

Returns:

  • (Float) 


150
151
152
 
# File 'lib/solid_objects/process_registry.rb', line 150

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

#register(kind: "worker", metadata: {}) ⇒ Process

RBS:

  • (?kind: String, ?metadata: Hash[String | Symbol, untyped]) -> Process

Parameters:

  • kind: (String) (defaults to: "worker")
  • metadata: (Hash[String | Symbol, untyped]) (defaults to: {})

Returns:



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

def register(kind: "worker", metadata: {})
  process_record = SolidObjects.database_adapter.with_lock_retry do
    now = SolidObjects.database_adapter.database_now
    Process.create!(
      id: SecureRandom.uuid,
      kind:,
      hostname: Socket.gethostname,
      pid: ::Process.pid,
      started_at: now,
      last_heartbeat_at: now,
      metadata: Serialization.dump(default_metadata.merge(metadata))
    )
  end
  @process_record = process_record
  @last_heartbeat_at = monotonic_now
  process_record
end

#start_draining ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


115
116
117
118
119
120
121
122
 
# File 'lib/solid_objects/process_registry.rb', line 115

def start_draining
  return false unless process_record

  process_record.update(
    shutdown_state: "draining",
    shutdown_requested_at: SolidObjects.database_adapter.database_now
  )
end

#stop ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


125
126
127
128
129
 
# File 'lib/solid_objects/process_registry.rb', line 125

def stop
  return false unless process_record

  self.class.deregister(process_record)
end