Module: Legion::Extensions::Llm::Vllm::Runners::DiscoveryRefresh::InstanceLifecycle

Included in:
Legion::Extensions::Llm::Vllm::Runners::DiscoveryRefresh
Defined in:
lib/legion/extensions/llm/vllm/runners/discovery_refresh/instance_lifecycle.rb

Overview

Tick entrypoint, catalog reconcile, and removal for the vLLM discovery runner. Mixed into DiscoveryRefresh.

Instance Method Summary collapse

Instance Method Details

#physical_id_changed?(state, instance_cfg) ⇒ Boolean

Returns:

  • (Boolean) 


77
78
79
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/instance_lifecycle.rb', line 77

def physical_id_changed?(state, instance_cfg)
  state[:instance_key].physical_id != derive_physical_id(instance_cfg: instance_cfg)
end

#reconcile_instances ⇒ Object

Re-scan the configured catalog each tick (D4 reconcile): claim instances that appeared since boot, remove ones that vanished, and refresh/probe every known instance. Vllm.discover_instances already applies the D3 filtering (no synthetic :default, no endpoint-less entries).

Instance identity is the operator's CONFIG NAME (the key the router uses for instances. settings lookups). The derived host:port/ak physical id is a secondary field only — two config names pointing at the same endpoint remain distinct instances.



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/instance_lifecycle.rb', line 32

def reconcile_instances
  configured = Vllm.discover_instances
  configured_ids = configured.keys.map(&:to_s)

  tracked_ids = state_mutex.synchronize { instance_states.keys }
  tracked_ids.each do |instance_id|
    next if configured_ids.include?(instance_id)

    remove_instance_state(instance_id)
  end

  configured.each do |name, instance_cfg|
    update_instance(name: name.to_s, instance_cfg: instance_cfg)
  rescue StandardError => e
    handle_exception(e, level: :warn, operation: 'vllm.runner.discovery.instance',
                        instance_name: name.to_s)
  end
  observe_dormant_weights
end

#refresh ⇒ Object

── Entrypoint (called by the Every actor each tick) ───────────



14
15
16
17
18
19
20
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/instance_lifecycle.rb', line 14

def refresh(**)
  reconcile_instances
  { success: true }
rescue StandardError => e
  handle_exception(e, level: :warn, operation: 'vllm.runner.discovery.refresh')
  { success: false }
end

#remove_all_instances ⇒ Object 



52
53
54
55
56
57
58
59
60
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/instance_lifecycle.rb', line 52

def remove_all_instances(**)
  tracked_ids = state_mutex.synchronize { instance_states.keys }
  tracked_ids.each { |instance_id| remove_instance_state(instance_id) }
  state_mutex.synchronize do
    instance_states.clear
    dormant_weight_tracker.clear!
  end
  { success: true }
end

#remove_instance_state(instance_id) ⇒ Object 



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/instance_lifecycle.rb', line 81

def remove_instance_state(instance_id)
  state = state_mutex.synchronize do
    tracked = instance_states[instance_id]
    next unless tracked

    publisher.remove_instance(
      instance_id: instance_id, publisher_token: tracked[:publisher_token]
    )
    instance_states.delete(instance_id) if instance_states[instance_id].equal?(tracked)
    tracked
  end
  return unless state

  clear_settings_health(name: state[:name])
rescue StandardError => e
  handle_exception(e, level: :warn, operation: 'vllm.runner.discovery.remove_instance',
                      instance_id: instance_id)
end

#update_instance(name:, instance_cfg:) ⇒ Object 



62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/instance_lifecycle.rb', line 62

def update_instance(name:, instance_cfg:)
  state = state_mutex.synchronize { instance_states[name] }
  if state && physical_id_changed?(state, instance_cfg)
    # The physical target moved (endpoint or API key changed) —
    # the captured callable points at the old endpoint, so drop
    # and re-claim under the same config-name identity.
    remove_instance_state(name)
    claim_and_activate_instance(name: name, instance_cfg: instance_cfg)
  elsif state
    refresh_instance(instance_id: name, instance_cfg: instance_cfg)
  else
    claim_and_activate_instance(name: name, instance_cfg: instance_cfg)
  end
end