Class: Legion::Extensions::Llm::Inventory::Registry::Store

Inherits:
Object
  • Object
show all
Includes:
Logging::Helper
Defined in:
lib/legion/extensions/llm/inventory/registry.rb

Overview

The synchronized internal store. All validation-against-current plus the root swap happen under @mutation_mutex; snapshot reads are lock-free. It never calls a provider callable, disconnect, probe enqueue handle, logger callback, or transport while holding the mutex; those side effects run after the swap. See section 12.2.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Store

Returns a new instance of Store.



39
40
41
42
43
44
45
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 39

def initialize
  @scopes = Concurrent::Map.new
  @issued_probe_tokens = Concurrent::Map.new
  @generation = 0
  @mutation_mutex = Mutex.new
  @snapshot_ref = Concurrent::AtomicReference.new(build_snapshot_locked)
end

Instance Method Details

#acquire(callable_handle:) ⇒ Object

Raises:



173
174
175
176
177
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 173

def acquire(callable_handle:)
  raise Errors::UnknownCallableError, 'acquire requires a CallableHandle' unless callable_handle.is_a?(CallableHandle)

  callable_handle.acquire
end

#activate_instance_snapshot(publisher_token:, instance_key:, offerings:, sequence:, probe_token:) ⇒ Object 



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 74

def activate_instance_snapshot(publisher_token:, instance_key:, offerings:, sequence:, probe_token:)
  validate_instance_key!(instance_key)
  drafts = ensure_drafts!(offerings)
  prepared = prepare_records(instance_key, drafts)
  @mutation_mutex.synchronize do
    scope = @scopes[instance_key]
    guard = guard_current(scope, publisher_token, instance_key)
    return guard if guard
    raise Errors::InvalidTransitionError, 'activate requires an initializing claim' unless scope.publication_status.state == :initializing
    return stale_mutation(scope, instance_key) unless prepared_matches?(prepared, scope)

    consume_probe_token!(probe_token, scope, instance_key)
    validate_sequence!(sequence, scope.last_sequence)
    apply_activation(scope, instance_key, prepared, sequence, probe_token)
  end
end

#claim_instance(instance_key:, callable:, probe_request_handle:) ⇒ Object

Raises:



51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 51

def claim_instance(instance_key:, callable:, probe_request_handle:)
  validate_instance_key!(instance_key)
  raise Errors::ValidationError, 'callable must not be nil' if callable.nil?
  raise Errors::ValidationError, 'probe_request_handle must not be nil' if probe_request_handle.nil?

  handle = CallableHandle.new(handle_id: "call:v1:#{SecureRandom.uuid}", callable: callable)
  token = PublisherToken.issue(instance_key: instance_key)
  old_handle = install_claim(instance_key, callable, handle, probe_request_handle, token)
  old_handle&.retire
  token
end

#dispatch_instance_unavailable(instance_key:, publisher_token_id:, reason:) ⇒ Object 



136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 136

def dispatch_instance_unavailable(instance_key:, publisher_token_id:, reason:)
  validate_instance_key!(instance_key)
  side_effect = nil
  result =
    @mutation_mutex.synchronize do
      scope = @scopes[instance_key]
      return absent_stale(instance_key) if scope.nil?
      return stale_mutation(scope, instance_key) unless scope.publisher_token.publisher_token_id == publisher_token_id
      raise Errors::InvalidTransitionError, 'dispatch_instance_unavailable requires an activated instance' unless activated?(scope)

      mutation, side_effect = apply_dispatch_unavailable(scope, instance_key, reason)
      mutation
    end
  enqueue_probe(*side_effect) if side_effect
  result
end

#readiness_failed(instance_key:, probe_token:, reason:) ⇒ Object 



122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 122

def readiness_failed(instance_key:, probe_token:, reason:)
  validate_instance_key!(instance_key)
  @mutation_mutex.synchronize do
    scope = @scopes[instance_key]
    return absent_stale(instance_key) if scope.nil?

    probe_guard = guard_probe_publisher(scope, probe_token, instance_key)
    return probe_guard if probe_guard

    consume_probe_token!(probe_token, scope, instance_key)
    apply_readiness_failure(scope, instance_key, probe_token, reason)
  end
end

#readiness_probe_started(instance_key:, publisher_token:) ⇒ Object 



63
64
65
66
67
68
69
70
71
72
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 63

def readiness_probe_started(instance_key:, publisher_token:)
  validate_instance_key!(instance_key)
  @mutation_mutex.synchronize do
    scope = @scopes[instance_key]
    classification = classify_publisher(scope, publisher_token, instance_key)
    raise Errors::FencedPublisherError, 'superseded or invalid publisher token' unless classification == :current

    issue_probe_token(scope)
  end
end

#readiness_succeeded(instance_key:, probe_token:) ⇒ Object 



107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 107

def readiness_succeeded(instance_key:, probe_token:)
  validate_instance_key!(instance_key)
  @mutation_mutex.synchronize do
    scope = @scopes[instance_key]
    return absent_stale(instance_key) if scope.nil?

    probe_guard = guard_probe_publisher(scope, probe_token, instance_key)
    return probe_guard if probe_guard
    raise Errors::InvalidTransitionError, 'readiness_succeeded before activation' if scope.publication_status.state == :initializing

    consume_probe_token!(probe_token, scope, instance_key)
    apply_readiness_success(scope, instance_key, probe_token)
  end
end

#remove_instance(instance_key:, publisher_token:) ⇒ Object 



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 153

def remove_instance(instance_key:, publisher_token:)
  validate_instance_key!(instance_key)
  old_handle = nil
  result =
    @mutation_mutex.synchronize do
      scope = @scopes[instance_key]
      return absent_removed(instance_key) if scope.nil?

      guard = guard_current(scope, publisher_token, instance_key)
      return guard if guard

      old_handle = scope.callable_handle
      @scopes.delete(instance_key)
      bump_and_snapshot!
      MutationResult.new(applied: true, reason: :removed, generation: @generation, instance_key: instance_key)
    end
  old_handle&.retire
  result
end

#replace_instance_snapshot(publisher_token:, instance_key:, offerings:, sequence:) ⇒ Object 



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 91

def replace_instance_snapshot(publisher_token:, instance_key:, offerings:, sequence:)
  validate_instance_key!(instance_key)
  drafts = ensure_drafts!(offerings)
  prepared = prepare_records(instance_key, drafts)
  @mutation_mutex.synchronize do
    scope = @scopes[instance_key]
    guard = guard_current(scope, publisher_token, instance_key)
    return guard if guard
    raise Errors::InvalidTransitionError, 'replace requires an activated instance' unless activated?(scope)
    return stale_mutation(scope, instance_key) unless prepared_matches?(prepared, scope)

    validate_sequence!(sequence, scope.last_sequence)
    apply_replacement(scope, instance_key, prepared, sequence)
  end
end

#retire_all_handles ⇒ Object 



179
180
181
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 179

def retire_all_handles
  @scopes.each_value { |scope| scope.callable_handle.retire }
end

#snapshot ⇒ Object 



47
48
49
 
# File 'lib/legion/extensions/llm/inventory/registry.rb', line 47

def snapshot
  @snapshot_ref.get
end