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



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

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) 


69
70
71
 
# File 'lib/solid_objects/process_registry.rb', line 69

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
 
# 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_expires_at: nil
    )
    ClaimedMessage.where(process_id: process_record.id).update_all(process_id: 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]) 


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

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

#heartbeat ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


94
95
96
97
98
99
100
101
 
# File 'lib/solid_objects/process_registry.rb', line 94

def heartbeat
  return false unless process_record
  return false if heartbeat_recent?

  process_record.update(
    last_heartbeat_at: SolidObjects.database_adapter.database_now
  ).tap { @last_heartbeat_at = monotonic_now }
end

#heartbeat_recent? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


131
132
133
134
135
136
 
# File 'lib/solid_objects/process_registry.rb', line 131

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) 


139
140
141
 
# File 'lib/solid_objects/process_registry.rb', line 139

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:



78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/solid_objects/process_registry.rb', line 78

def register(kind: "worker", metadata: {})
  now = SolidObjects.database_adapter.database_now
  @process_record = 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))
  )
  @last_heartbeat_at = monotonic_now
  @process_record
end

#start_draining ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


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

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) 


114
115
116
117
118
 
# File 'lib/solid_objects/process_registry.rb', line 114

def stop
  return false unless process_record

  self.class.deregister(process_record)
end