Class: Magick::Adapters::Registry

Inherits:
Object
  • Object
show all
Defined in:
lib/magick/adapters/registry.rb

Constant Summary collapse

CACHE_INVALIDATION_CHANNEL = 
'magick:cache:invalidate'
MAX_INVALIDATION_PAYLOAD_BYTES =

An invalidation message is a small JSON object; anything larger than this did not come from us and is not worth parsing.

512
FEATURE_NAME_PATTERN =

Accept only conservative feature identifiers coming off the wire. Anything outside this alphabet (newlines, spaces, Unicode punctuation, 200-char garbage) is a sign of a malformed or malicious publisher and must not be fed back into Magick.features / feature.reload.

/\A[a-zA-Z0-9_\-.:]{1,120}\z/.freeze
DEFAULT_REFRESH_INTERVAL =

How long a process may go without re-reading the shared backend. Pub/Sub is the fast path; this is the bound on staleness when that path is broken or bypassed — a write made outside a gem process, a Redis user without pub/sub permission, a proxy that drops SUBSCRIBE, a subscriber connection silently killed by a NAT/LB. Before it existed, a registered feature that missed one invalidation stayed stale until restart, and nothing said so. See ADR-0002.

30.0
SUBSCRIBER_FAILURE_REPORT_INTERVAL =

A subscriber that cannot subscribe retries every 5s. The first failure is reported at once; while it keeps failing, at most one report per this many seconds, so a permanently broken Redis does not flood the log.

300.0
SUBSCRIBER_RETRY_DELAY =

Seconds between attempts to (re)subscribe after a failure.

5

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(memory_adapter, redis_adapter = nil, active_record_adapter: nil, circuit_breaker: nil, async: false, primary: nil, async_queue_limit: nil, async_enqueue_timeout: nil, refresh_interval: DEFAULT_REFRESH_INTERVAL) ⇒ Registry

Returns a new instance of Registry.



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/magick/adapters/registry.rb', line 38

def initialize(memory_adapter, redis_adapter = nil, active_record_adapter: nil, circuit_breaker: nil,
               async: false, primary: nil, async_queue_limit: nil, async_enqueue_timeout: nil,
               refresh_interval: DEFAULT_REFRESH_INTERVAL)
  @memory_adapter = memory_adapter
  @redis_adapter = redis_adapter
  @active_record_adapter = active_record_adapter
  @circuit_breaker = circuit_breaker || Magick::CircuitBreaker.new
  @async = async
  @async_queue_limit = async_queue_limit
  @async_enqueue_timeout = async_enqueue_timeout
  @async_writer = nil
  @write_order_mutex = Mutex.new
  @primary = primary || :memory # :memory, :redis, or :active_record
  @subscriber_thread = nil
  @subscriber = nil
  @subscriber_pid = nil # process that opened @subscriber; a fork must not touch its parent's
  @subscriber_generation = 0 # bumped to retire a running subscriber (reconfigure, shutdown)
  @subscribed = false
  @subscriber_last_error = nil
  @subscriber_failure_reported_at = nil
  @refresh_interval = normalize_refresh_interval(refresh_interval)
  @refresh_gate = Mutex.new # claims the once-per-interval slot
  @source_read_mutex = Mutex.new # serializes the bulk read + apply
  @last_refresh_at = nil # monotonic; gates the next refresh
  @last_refresh_time = nil # wall clock; reported by #health
  @source_snapshot = nil # feature => data as last read from the shared backend
  @stopping = false
  @shutdown_mutex = Mutex.new
  @owner_pid = Process.pid
  @identity_mutex = Mutex.new
  @publisher_id_pid = Process.pid
  @publisher_id = generate_publisher_id
  # Only start Pub/Sub subscriber if Redis is available
  # In memory-only mode, each process has isolated cache (no cross-process invalidation)
  start_cache_invalidation_subscriber if redis_adapter
end

Instance Attribute Details

#active_record_adapter ⇒ Object (readonly)

Public so Versioning can apply tiered retention: hot window written to memory/Redis, unlimited archive written to ActiveRecord only.



526
527
528
 
# File 'lib/magick/adapters/registry.rb', line 526

def active_record_adapter
  @active_record_adapter
end

#async_writer ⇒ Object (readonly)

The serialized writer draining async Redis writes, or nil when this registry has not made one yet (sync mode, or no write has happened).



530
531
532
 
# File 'lib/magick/adapters/registry.rb', line 530

def async_writer
  @async_writer
end

#memory_adapter ⇒ Object (readonly)

Public so Versioning can apply tiered retention: hot window written to memory/Redis, unlimited archive written to ActiveRecord only.



526
527
528
 
# File 'lib/magick/adapters/registry.rb', line 526

def memory_adapter
  @memory_adapter
end

#redis_adapter ⇒ Object

Public so Versioning can apply tiered retention: hot window written to memory/Redis, unlimited archive written to ActiveRecord only.



526
527
528
 
# File 'lib/magick/adapters/registry.rb', line 526

def redis_adapter
  @redis_adapter
end

#refresh_interval ⇒ Object

Seconds between periodic re-reads of the shared backend, or nil when the periodic refresh is off and Pub/Sub is the only thing that updates a registered feature (the behavior before the periodic refresh existed).



372
373
374
 
# File 'lib/magick/adapters/registry.rb', line 372

def refresh_interval
  @refresh_interval
end

Instance Method Details

#all_features ⇒ Object 



197
198
199
200
201
202
203
204
205
 
# File 'lib/magick/adapters/registry.rb', line 197

def all_features
  features = []
  features += safely([]) { memory_adapter&.all_features } || []
  features += redis_read { |redis| redis.all_features } || []
  features += safely([]) { active_record_adapter&.all_features } || []
  # Version history and audit history are stored under reserved
  # pseudo-feature namespaces; they are bookkeeping, not features.
  features.uniq.reject { |f| reserved_store_name?(f) }
end

#authoritative_get_all_data(feature_name) ⇒ Object

Read a feature's complete data straight from the shared, authoritative backend — ActiveRecord first (it is written synchronously on every set/set_all_data), then Redis — bypassing this process's local memory cache, and refresh memory with the result.

The Admin UI uses this so a toggle is reflected immediately on whichever process/container serves the (load-balanced) request after the write, instead of rendering this process's possibly-stale memory cache while it waits for Pub/Sub invalidation to arrive.



247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/magick/adapters/registry.rb', line 247

def authoritative_get_all_data(feature_name)
  data = read_from_source(feature_name)
  if data && !data.empty?
    memory_adapter&.set_all_data(feature_name, data)
    return data
  end

  # Source unavailable (Redis/AR down, or feature absent there) — fall back
  # to whatever this process already has rather than wiping a usable cache.
  memory_adapter ? memory_adapter.get_all_data(feature_name) : {}
end

#delete(feature_name) ⇒ Object 



171
172
173
174
175
176
177
178
179
180
181
182
183
184
 
# File 'lib/magick/adapters/registry.rb', line 171

def delete(feature_name)
  memory_adapter&.delete(feature_name)

  if redis_adapter
    # Behind the breaker like every other Redis write, and the peers are
    # only told to reload once Redis has actually forgotten the feature.
    deleted = write_to_redis(:delete, feature_name) { redis_adapter.delete(feature_name) }
    publish_cache_invalidation(feature_name) if deleted
  end

  return unless active_record_adapter

  write_to_active_record(:delete, feature_name) { active_record_adapter.delete(feature_name) }
end

#delete_key(feature_name, key) ⇒ Object

Remove one key from a feature across every configured layer.

No local-write bookkeeping is needed: self-invalidation is keyed on the publisher of each message, so this process's own echo is recognised by identity — and #delete_key does not publish at all.



333
334
335
336
337
338
339
 
# File 'lib/magick/adapters/registry.rb', line 333

def delete_key(feature_name, key)
  deleted = false
  [memory_adapter, redis_adapter, active_record_adapter].compact.each do |adapter|
    deleted = true if safely_delete_key(adapter, feature_name, key)
  end
  deleted
end

#ensure_subscriber! ⇒ Object

Restart the Pub/Sub subscriber after a fork. The subscriber thread is not carried into child processes, so a worker inheriting a stale reference must re-create its own subscription. Safe to call on every request; it only does work when Process.pid changes.



79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/magick/adapters/registry.rb', line 79

def ensure_subscriber!
  return if @owner_pid == Process.pid

  @shutdown_mutex.synchronize do
    return if @owner_pid == Process.pid

    # The thread did not survive the fork and the connection object is the
    # parent's: dropped, never closed (see #close_subscriber_connection).
    @subscriber_thread = nil
    @subscriber = nil
    @subscribed = false
    @owner_pid = Process.pid
    @stopping = false
  end

  start_cache_invalidation_subscriber if redis_adapter
end

#exists?(feature_name) ⇒ Boolean

Fail-safe by contract: an unreachable backend means "not found", never an exception. Callers reach this from outside the fail-safe evaluation path, where a raise would surface as a 500 rather than a disabled flag.

Returns:

  • (Boolean) 


189
190
191
192
193
194
195
 
# File 'lib/magick/adapters/registry.rb', line 189

def exists?(feature_name)
  return true if safely(false) { memory_adapter&.exists?(feature_name) }
  return true if redis_read { |redis| redis.exists?(feature_name) } == true
  return true if safely(false) { active_record_adapter&.exists?(feature_name) } == true

  false
end

#get(feature_name, key) ⇒ Object 



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/magick/adapters/registry.rb', line 128

def get(feature_name, key)
  # Try memory first (fastest) - no Redis calls needed thanks to Pub/Sub invalidation
  value = memory_adapter.get(feature_name, key) if memory_adapter
  return value unless value.nil?

  # Fall back to Redis if available
  value = redis_read { |redis| redis.get(feature_name, key) }
  if !value.nil? && memory_adapter
    memory_adapter.set(feature_name, key, value)
    return value
  end

  # Fall back to Active Record if available
  if active_record_adapter
    begin
      value = active_record_adapter.get(feature_name, key)
      memory_adapter.set(feature_name, key, value) if !value.nil? && memory_adapter
      return value
    rescue StandardError, AdapterError
      nil
    end
  end

  nil
end

#get_all_data(feature_name) ⇒ Object

Load all keys for a single feature in one call instead of N separate get() calls



208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/magick/adapters/registry.rb', line 208

def get_all_data(feature_name)
  # Try memory first
  if memory_adapter
    data = memory_adapter.get_all_data(feature_name)
    return data unless data.nil? || data.empty?
  end

  # Fall back to Redis
  data = redis_read { |redis| redis.get_all_data(feature_name) }
  if data && !data.empty?
    memory_adapter.set_all_data(feature_name, data) if memory_adapter
    return data
  end

  # Fall back to Active Record
  if active_record_adapter
    begin
      data = active_record_adapter.get_all_data(feature_name)
      if data && !data.empty?
        memory_adapter.set_all_data(feature_name, data) if memory_adapter
        return data
      end
    rescue StandardError, AdapterError
      # AR failed
    end
  end

  {}
end

#health ⇒ Object

A snapshot for host health checks: is this process listening for invalidations, when did it last confirm its view against the shared backend, and is anything queued. Plain values, safe to render as JSON.



452
453
454
455
456
457
458
459
460
461
462
 
# File 'lib/magick/adapters/registry.rb', line 452

def health
  {
    redis: redis_available?,
    active_record: !active_record_adapter.nil?,
    subscriber_running: subscriber_running?,
    subscriber_last_error: @subscriber_last_error,
    refresh_interval: @refresh_interval,
    last_source_refresh_at: @last_refresh_time,
    pending_async_writes: pending_async_writes
  }
end

#invalidate_cache(feature_name) ⇒ Object

Explicitly trigger cache invalidation for a feature This is useful for targeting updates that need immediate cache invalidation Invalidates memory cache in current process AND publishes to Redis for other processes



361
362
363
364
365
366
367
 
# File 'lib/magick/adapters/registry.rb', line 361

def invalidate_cache(feature_name)
  # Invalidate memory cache in current process immediately
  memory_adapter&.delete(feature_name)

  # Publish to Redis Pub/Sub to invalidate cache in other processes
  publish_cache_invalidation(feature_name)
end

#next_sequence(feature_name, key, floor: 0) ⇒ Object

Allocate a number from a shared counter. Unlike #set this deliberately does NOT fan out to every layer: a counter must have exactly one authority, or two processes reading different layers would be handed the same number. ActiveRecord is preferred (durable, row-locked), then Redis (atomic HINCRBY), then this process's memory as a last resort.



346
347
348
349
350
351
352
353
354
355
356
 
# File 'lib/magick/adapters/registry.rb', line 346

def next_sequence(feature_name, key, floor: 0)
  [active_record_adapter, redis_adapter, memory_adapter].compact.each do |adapter|
    value = begin
      adapter.next_sequence(feature_name, key, floor: floor)
    rescue StandardError, AdapterError, NotImplementedError
      nil
    end
    return value if value
  end
  nil
end

#pending_async_writes ⇒ Object

Writes accepted by the async writer but not yet sent to Redis.



533
534
535
 
# File 'lib/magick/adapters/registry.rb', line 533

def pending_async_writes
  @async_writer ? @async_writer.pending : 0
end

#preload! ⇒ Object

Bulk load ALL features into memory cache in minimal queries. Call this after configuration to warm the cache.

Version snapshots and audit history are skipped, exactly as #all_features skips them: both are unbounded bookkeeping, and pulling the version archive into every worker's memory cache at boot cost tens of megabytes per worker on installs with a few thousand versions.



278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
 
# File 'lib/magick/adapters/registry.rb', line 278

def preload!
  all_data = {}

  # Load from ActiveRecord first (source of truth for persistence)
  if active_record_adapter
    begin
      all_data = load_features_only(active_record_adapter)
    rescue StandardError, AdapterError
      # AR failed, try Redis
    end
  end

  # Merge/override with Redis data (more up-to-date than AR in most setups)
  redis_data = redis_read { |redis| load_features_only(redis) }
  redis_data&.each do |feature_name, data|
    all_data[feature_name] ||= {}
    all_data[feature_name].merge!(data)
  end

  # Reserved bookkeeping namespaces (version snapshots, audit history)
  # are not features: keep their blobs out of the feature cache.
  all_data = all_data.reject { |feature_name, _| reserved_store_name?(feature_name) }

  # Populate memory cache in bulk
  if memory_adapter && !all_data.empty?
    all_data.each do |feature_name, data|
      memory_adapter.set_all_data(feature_name, data)
    end
  end

  all_data
end

#publish_cache_invalidation(feature_name) ⇒ Object

Publish cache invalidation message to Redis Pub/Sub (without deleting local memory cache) This is useful when you've just updated the cache and want to notify other processes but keep the local memory cache intact



479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
 
# File 'lib/magick/adapters/registry.rb', line 479

def publish_cache_invalidation(feature_name)
  return unless redis_adapter

  begin
    # Through the breaker for two reasons: PUBLISH is a Redis round-trip
    # like any other and must not block a request thread on a dead
    # server, and an open circuit means this process cannot have written
    # the new value — inviting peers to reload would hand them the
    # pre-toggle state.
    circuit_breaker.call do
      redis_client = redis_adapter.client
      redis_client&.publish(CACHE_INVALIDATION_CHANNEL, invalidation_message(feature_name))
    end
  rescue CircuitOpenError
    AdapterFailure.report(backend: :redis, operation: :publish_cache_invalidation,
                          feature_name: feature_name, reason: 'circuit breaker open')
  rescue StandardError => e
    # Best effort for this process, but a dropped invalidation leaves every
    # OTHER process holding a stale value, so it is reported like any other
    # failed write rather than swallowed.
    AdapterFailure.report(backend: :redis, operation: :publish_cache_invalidation,
                          feature_name: feature_name, error: e)
  end
end

#publisher_id ⇒ Object

This registry's identity on the invalidation channel. Every message it publishes carries it, and it is the ONLY thing the subscriber suppresses on: a message is ignored when we published it, and acted on otherwise.

Per-registry rather than per-process, so two registries sharing a process still see each other's writes; random, because PIDs collide across containers.



511
512
513
514
515
516
517
518
519
520
521
522
 
# File 'lib/magick/adapters/registry.rb', line 511

def publisher_id
  @identity_mutex.synchronize do
    # A forked child inherits the parent's identity along with the rest of
    # its memory. Re-mint it so parent and child do not each mistake the
    # other's invalidations for their own echo and ignore them.
    if @publisher_id_pid != Process.pid
      @publisher_id_pid = Process.pid
      @publisher_id = generate_publisher_id
    end
    @publisher_id
  end
end

#redis_available? ⇒ Boolean

Check if Redis adapter is available

Returns:

  • (Boolean) 


465
466
467
 
# File 'lib/magick/adapters/registry.rb', line 465

def redis_available?
  !redis_adapter.nil?
end

#redis_client ⇒ Object

Get Redis client (public method for use by other classes)



470
471
472
473
474
 
# File 'lib/magick/adapters/registry.rb', line 470

def redis_client
  return nil unless redis_adapter

  redis_adapter.client
end

#refresh_all_from_source ⇒ Object

Bulk variant of #authoritative_get_all_data: refresh the local memory cache for EVERY feature from the shared backend in 1-2 queries. Returns the loaded data. Used by the Admin UI index so the full list reflects authoritative state regardless of which container serves it.



263
264
265
266
267
268
269
 
# File 'lib/magick/adapters/registry.rb', line 263

def refresh_all_from_source
  data = load_all_from_source
  if memory_adapter && !data.empty?
    data.each { |feature_name, feature_data| memory_adapter.set_all_data(feature_name, feature_data) }
  end
  data
end

#refresh_from_source! ⇒ Object

Re-read every feature from the shared backend and bring this process in line with it: features whose stored data changed since the last read are written into memory and, when registered here, reloaded. ActiveRecord is read first (written synchronously on every set), then Redis — the same source the Admin UI treats as authoritative.

The comparison is against the previous read of the source, not against memory. A local write in flight (memory ahead of the store, e.g. an async Redis write still queued) therefore never gets reverted: the source has not changed, so the feature is left alone, and when the write lands it shows up as a change whose reload is a no-op. The first read, with no previous one to compare to, is measured against memory instead.

Features that vanished from the source are deliberately NOT evicted. A backend that answers with a partial view (one adapter reachable, the other not; ActiveRecord half backfilled) must not be able to strip targeting off live features. Deletion through the gem still publishes an invalidation, which does evict.

Returns the names of the features that changed, or nil when no source answered (nothing is touched in that case). Never raises.



416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
 
# File 'lib/magick/adapters/registry.rb', line 416

def refresh_from_source!
  return nil unless source_adapter?

  # Serialized so a manual Magick.refresh! racing the scheduled read
  # cannot interleave two snapshots; the evaluation path never waits here
  # because #refresh_if_stale! admits one caller per interval.
  @source_read_mutex.synchronize do
    data = read_source_features
    return nil if data.nil?

    changed = changed_since_last_read(data)
    changed.each do |feature_name|
      memory_adapter&.set_all_data(feature_name, data[feature_name])
      reload_registered_feature(feature_name)
    end

    @source_snapshot = data
    @last_refresh_time = Time.now
    changed
  end
rescue StandardError => e
  AdapterFailure.report(backend: source_backend, operation: :refresh, error: e)
  nil
end

#refresh_if_stale! ⇒ Object

Hot-path entry point, called on every feature evaluation. Almost always a clock read and a comparison; once per interval, in exactly one thread, it re-reads the shared backend (see #refresh_from_source!). Concurrent callers skip rather than queue, and the slot is claimed before the read so a slow or failing source is probed once per interval, not once per evaluation. Never raises. Returns what #refresh_from_source! returned, or nil when nothing was done.



385
386
387
388
389
390
391
392
393
 
# File 'lib/magick/adapters/registry.rb', line 385

def refresh_if_stale!
  interval = @refresh_interval
  return nil unless interval && source_adapter? && refresh_due?(interval)
  return nil unless claim_refresh_slot(interval)

  refresh_from_source!
rescue StandardError
  nil
end

#set(feature_name, key, value) ⇒ Object 



154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/magick/adapters/registry.rb', line 154

def set(feature_name, key, value)
  update_redis = proc do
    write_to_redis(:set, feature_name) { redis_adapter.set(feature_name, key, value) }
  end

  # Memory is always updated synchronously; Redis follows either inline
  # or through the serialized async writer.
  write_through(feature_name, update_redis) do
    memory_adapter&.set(feature_name, key, value)
  end

  # Always update Active Record if available (as fallback/persistence layer)
  return unless active_record_adapter

  write_to_active_record(:set, feature_name) { active_record_adapter.set(feature_name, key, value) }
end

#set_all_data(feature_name, data_hash) ⇒ Object

Bulk set multiple keys for a feature in one call (1 query instead of N)



312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
 
# File 'lib/magick/adapters/registry.rb', line 312

def set_all_data(feature_name, data_hash)
  update_redis = proc do
    write_to_redis(:set_all_data, feature_name) { redis_adapter.set_all_data(feature_name, data_hash) }
  end

  write_through(feature_name, update_redis) do
    memory_adapter&.set_all_data(feature_name, data_hash)
  end

  return unless active_record_adapter

  write_to_active_record(:set_all_data, feature_name) do
    active_record_adapter.set_all_data(feature_name, data_hash)
  end
end

#shutdown(timeout: 5) ⇒ Object

Gracefully terminate the Pub/Sub subscriber thread and its Redis connection. Without this, Ruby/Puma shutdown waits on the blocking subscribe call.



110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/magick/adapters/registry.rb', line 110

def shutdown(timeout: 5)
  @shutdown_mutex.synchronize do
    return if @stopping

    @stopping = true
  end

  # Drain queued writes while the Redis connection is still usable.
  drain_async_writer(timeout)

  stop_subscriber!(timeout: timeout)
  true
end

#stopping? ⇒ Boolean

Returns:

  • (Boolean) 


124
125
126
 
# File 'lib/magick/adapters/registry.rb', line 124

def stopping?
  @stopping == true
end

#subscriber_running? ⇒ Boolean

True while this process holds a live subscription on the invalidation channel — the thread is alive AND Redis has acknowledged the SUBSCRIBE. A thread stuck in the retry loop, or one whose connection was torn down, answers false.

Returns:

  • (Boolean) 


445
446
447
 
# File 'lib/magick/adapters/registry.rb', line 445

def subscriber_running?
  @subscribed == true && !@subscriber_thread.nil? && @subscriber_thread.alive?
end