Class: RubyReactor::Storage::RedisAdapter

Inherits:
Adapter
  • Object
show all
Includes:
RedisLocking, RedisOrderedLocking
Defined in:
lib/ruby_reactor/storage/redis_adapter.rb

Constant Summary

Constants included from RedisOrderedLocking

RubyReactor::Storage::RedisOrderedLocking::ADVANCE_SCRIPT, RubyReactor::Storage::RedisOrderedLocking::ASSIGN_SCRIPT, RubyReactor::Storage::RedisOrderedLocking::CAN_PROCEED_SCRIPT, RubyReactor::Storage::RedisOrderedLocking::HEARTBEAT_SCRIPT, RubyReactor::Storage::RedisOrderedLocking::SKIP_SCRIPT

Constants included from RedisLocking

RubyReactor::Storage::RedisLocking::LOCK_ACQUIRE_SCRIPT, RubyReactor::Storage::RedisLocking::LOCK_EXTEND_SCRIPT, RubyReactor::Storage::RedisLocking::LOCK_RELEASE_SCRIPT, RubyReactor::Storage::RedisLocking::RATE_LIMIT_SCRIPT, RubyReactor::Storage::RedisLocking::SEMAPHORE_TTL, RubyReactor::Storage::RedisLocking::SEM_ACQUIRE_SCRIPT, RubyReactor::Storage::RedisLocking::SEM_RELEASE_SCRIPT

Instance Method Summary collapse

Methods included from RedisOrderedLocking

#ordered_lock_advance, #ordered_lock_assign, #ordered_lock_can_proceed, #ordered_lock_heartbeat, #ordered_lock_keys, #ordered_lock_peek, #ordered_lock_reset, #ordered_lock_skip

Methods included from RedisLocking

#lock_acquire, #lock_extend, #lock_held?, #lock_info, #lock_release, #lock_ttl, #period_mark, #period_marker?, #period_seen?, #period_ttl, #rate_limit_check_and_increment, #rate_limit_count, #rate_limit_ttl, #semaphore_acquire, #semaphore_exists?, #semaphore_init, #semaphore_release, #semaphore_reset, #semaphore_state

Constructor Details

#initialize(redis_config) ⇒ RedisAdapter

Returns a new instance of RedisAdapter.



12
13
14
15
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 12

def initialize(redis_config)
  super()
  @redis = Redis.new(redis_config)
end

Instance Method Details

#count_map_results(map_id, reactor_class_name) ⇒ Object

rubocop:enable Lint/UnusedMethodArgument



307
308
309
310
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 307

def count_map_results(map_id, reactor_class_name)
  key = map_results_key(map_id, reactor_class_name)
  @redis.hlen(key)
end

#decrement_map_counter(map_id, reactor_class_name) ⇒ Object 



127
128
129
130
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 127

def decrement_map_counter(map_id, reactor_class_name)
  key = map_counter_key(map_id, reactor_class_name)
  @redis.decr(key)
end

#delete_context(context_id, reactor_class_name) ⇒ Object 



172
173
174
175
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 172

def delete_context(context_id, reactor_class_name)
  key = context_key(context_id, reactor_class_name)
  @redis.del(key)
end

#delete_correlation_id(correlation_id, reactor_class_name) ⇒ Object 



167
168
169
170
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 167

def delete_correlation_id(correlation_id, reactor_class_name)
  key = correlation_id_key(correlation_id, reactor_class_name)
  @redis.del(key)
end

#determine_status(data) ⇒ Object 



233
234
235
236
237
238
239
240
241
242
243
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 233

def determine_status(data)
  status = data["status"].to_s
  return status if status && %w[failed paused completed running skipped pending].include?(status)
  return "cancelled" if data["cancelled"]
  # Heuristic
  return "failed" if data["retry_count"]&.positive? && !data["current_step"].nil?
  return "running" if data["current_step"]
  return "completed" if execution_evidence?(data)

  "pending"
end

#execution_evidence?(data) ⇒ Boolean

Returns:

  • (Boolean) 


245
246
247
248
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 245

def execution_evidence?(data)
  (data["execution_trace"] || []).any? ||
    (data["intermediate_results"] || {}).any?
end

#expire(key, seconds) ⇒ Object 



185
186
187
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 185

def expire(key, seconds)
  @redis.expire(key, seconds)
end

#find_context_by_id(context_id) ⇒ Object 



216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 216

def find_context_by_id(context_id)
  # We don't know the reactor class, so we search for the ID
  pattern = "reactor:*:context:#{context_id}"
  keys = []
  @redis.scan_each(match: pattern, count: 1) do |key|
    keys << key
    break
  end
  return nil if keys.empty?

  key = keys.first
  json = @redis.get(key)
  return nil unless json

  JSON.parse(json)
end

#increment_last_queued_index(map_id, reactor_class_name) ⇒ Object 



137
138
139
140
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 137

def increment_last_queued_index(map_id, reactor_class_name)
  key = map_last_queued_index_key(map_id, reactor_class_name)
  @redis.incr(key)
end

#increment_map_counter(map_id, reactor_class_name) ⇒ Object 



121
122
123
124
125
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 121

def increment_map_counter(map_id, reactor_class_name)
  key = map_counter_key(map_id, reactor_class_name)
  @redis.incr(key)
  @redis.expire(key, durability_ttl)
end

#increment_map_offset(map_id, increment, reactor_class_name) ⇒ Object 



292
293
294
295
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 292

def increment_map_offset(map_id, increment, reactor_class_name)
  key = map_offset_key(map_id, reactor_class_name)
  @redis.incrby(key, increment)
end

#initialize_map_operation(map_id, count, parent_reactor_class_name, reactor_class_info:, strict_ordering: true, parent_context_id: nil, step_name: nil, parent_is_map_element: false, outer_map_id: nil, outer_index: nil) ⇒ Object

rubocop:disable Metrics/ParameterLists



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 67

def initialize_map_operation(map_id, count, parent_reactor_class_name, reactor_class_info:, strict_ordering: true,
                             parent_context_id: nil, step_name: nil, parent_is_map_element: false,
                             outer_map_id: nil, outer_index: nil)
  # Ensure counter is set
  set_map_counter(map_id, count, parent_reactor_class_name)

  # Store metadata. parent_context_id/step_name let the map sweeper recover
  # without re-deriving the map_id (which is brittle to split on ':'). The
  # nested-map fields (parent_is_map_element + outer_map_id/outer_index)
  # record which liveness lock the parent actually holds (N1): a nested
  # map's parent is itself a map element running under a `map_element:` lock,
  # not an `async:` lock.
  key = "reactor:#{parent_reactor_class_name}:map:#{map_id}:metadata"
  metadata = {
    map_id: map_id,
    count: count,
    strict_ordering: strict_ordering,
    reactor_class_info: reactor_class_info,
    parent_context_id: parent_context_id,
    parent_reactor_class_name: parent_reactor_class_name,
    step_name: step_name,
    parent_is_map_element: parent_is_map_element,
    outer_map_id: outer_map_id,
    outer_index: outer_index,
    created_at: Time.now.to_i
  }
  @redis.set(key, metadata.to_json, ex: durability_ttl)
end

#missing_map_indices(map_id, count, reactor_class_name) ⇒ Object

Indices that have NO stored result yet: the authoritative, idempotent signal for what the map sweeper must (re)dispatch.



55
56
57
58
59
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 55

def missing_map_indices(map_id, count, reactor_class_name)
  key = map_results_key(map_id, reactor_class_name)
  present = @redis.hkeys(key).map(&:to_i)
  (0...count).to_a - present
end

#publish(channel, message) ⇒ Object 



181
182
183
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 181

def publish(channel, message)
  @redis.publish(channel, message)
end

#retrieve_context(context_id, reactor_class_name) ⇒ Object 



25
26
27
28
29
30
31
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 25

def retrieve_context(context_id, reactor_class_name)
  key = context_key(context_id, reactor_class_name)
  json = @redis.get(key)
  return nil unless json

  JSON.parse(json)
end

#retrieve_context_id_by_correlation_id(correlation_id, reactor_class_name) ⇒ Object 



162
163
164
165
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 162

def retrieve_context_id_by_correlation_id(correlation_id, reactor_class_name)
  key = correlation_id_key(correlation_id, reactor_class_name)
  @redis.get(key)
end

#retrieve_map_element_context_id(map_id, reactor_class_name, index: -1)) ⇒ Object 



261
262
263
264
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 261

def retrieve_map_element_context_id(map_id, reactor_class_name, index: -1)
  key = map_element_contexts_key(map_id, reactor_class_name)
  @redis.lindex(key, index)
end

#retrieve_map_element_context_ids(map_id, reactor_class_name) ⇒ Object 



256
257
258
259
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 256

def retrieve_map_element_context_ids(map_id, reactor_class_name)
  key = map_element_contexts_key(map_id, reactor_class_name)
  @redis.lrange(key, 0, -1)
end

#retrieve_map_failed_context_id(map_id, reactor_class_name) ⇒ Object 



272
273
274
275
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 272

def retrieve_map_failed_context_id(map_id, reactor_class_name)
  key = map_failed_context_key(map_id, reactor_class_name)
  @redis.get(key)
end

#retrieve_map_metadata(map_id, reactor_class_name) ⇒ Object 



113
114
115
116
117
118
119
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 113

def retrieve_map_metadata(map_id, reactor_class_name)
  key = "reactor:#{reactor_class_name}:map:#{map_id}:metadata"
  json = @redis.get(key)
  return nil unless json

  JSON.parse(json)
end

#retrieve_map_offset(map_id, reactor_class_name) ⇒ Object 



287
288
289
290
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 287

def retrieve_map_offset(map_id, reactor_class_name)
  key = map_offset_key(map_id, reactor_class_name)
  @redis.get(key)
end

#retrieve_map_results(map_id, reactor_class_name, strict_ordering: true) ⇒ Object 



45
46
47
48
49
50
51
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 45

def retrieve_map_results(map_id, reactor_class_name, strict_ordering: true)
  # rubocop:enable Lint/UnusedMethodArgument
  key = map_results_key(map_id, reactor_class_name)
  results = @redis.hgetall(key)
  # Index-keyed for both modes; sort by index so reads are deterministic.
  results.keys.sort_by(&:to_i).map { |k| JSON.parse(results[k]) }
end

#retrieve_map_results_batch(map_id, reactor_class_name, offset:, limit:, strict_ordering: true) ⇒ Object

rubocop:disable Lint/UnusedMethodArgument



298
299
300
301
302
303
304
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 298

def retrieve_map_results_batch(map_id, reactor_class_name, offset:, limit:, strict_ordering: true)
  # Always index-keyed now (Phase 5): HMGET the contiguous index window.
  key = map_results_key(map_id, reactor_class_name)
  fields = (offset...(offset + limit)).map(&:to_s)
  results = @redis.hmget(key, *fields)
  results.compact.map { |r| JSON.parse(r) }
end

#scan_maps(count: 1000) ⇒ Object

Enumerate active map operations for the map sweeper (Phase 5d). Returns the parsed metadata hash for each (includes map_id, count, parent_context_id, step_name, parent_reactor_class_name, and the nested-map lock fields). Bounded by ‘count` to keep a sweep cheap.



101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 101

def scan_maps(count: 1000)
  results = []
  @redis.scan_each(match: "reactor:*:map:*:metadata", count: 100) do |key|
    json = @redis.get(key)
    next unless json

    results << JSON.parse(json)
    return results if results.size >= count
  end
  results
end

#scan_reactors(pattern: "reactor:*:context:*", count: 50) ⇒ Object

New methods for API



190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 190

def scan_reactors(pattern: "reactor:*:context:*", count: 50)
  # Use SCAN to find keys matching the pattern
  results = []
  batch_keys = []

  # scan_each yields keys. We buffer them to use MGET efficiently.
  # We request a batch size from Redis (count: 100) to reduce roundtrips.
  @redis.scan_each(match: pattern, count: 100) do |key|
    batch_keys << key

    # specific batch size for MGET processing
    if batch_keys.size >= 50
      results.concat(fetch_and_filter_reactors(batch_keys))
      batch_keys = []

      # Stop if we have enough results
      return results.take(count) if results.size >= count
    end
  end

  # Process remaining keys
  results.concat(fetch_and_filter_reactors(batch_keys)) if batch_keys.any?

  results.take(count)
end

#set_last_queued_index(map_id, index, reactor_class_name) ⇒ Object 



132
133
134
135
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 132

def set_last_queued_index(map_id, index, reactor_class_name)
  key = map_last_queued_index_key(map_id, reactor_class_name)
  @redis.set(key, index, ex: durability_ttl)
end

#set_map_counter(map_id, count, reactor_class_name) ⇒ Object 



61
62
63
64
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 61

def set_map_counter(map_id, count, reactor_class_name)
  key = map_counter_key(map_id, reactor_class_name)
  @redis.set(key, count, ex: durability_ttl)
end

#set_map_offset(map_id, offset, reactor_class_name) ⇒ Object 



277
278
279
280
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 277

def set_map_offset(map_id, offset, reactor_class_name)
  key = map_offset_key(map_id, reactor_class_name)
  @redis.set(key, offset, ex: durability_ttl)
end

#set_map_offset_if_not_exists(map_id, offset, reactor_class_name) ⇒ Object 



282
283
284
285
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 282

def set_map_offset_if_not_exists(map_id, offset, reactor_class_name)
  key = map_offset_key(map_id, reactor_class_name)
  @redis.set(key, offset, nx: true, ex: durability_ttl)
end

#store_context(context_id, serialized_context, reactor_class_name) ⇒ Object 



17
18
19
20
21
22
23
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 17

def store_context(context_id, serialized_context, reactor_class_name)
  key = context_key(context_id, reactor_class_name)
  # Use standard SET for compatibility (ReJSON not strictly required for full docs).
  # TTL is re-stamped on every write so long-running / snoozed contexts
  # never expire mid-flight (Phase 4).
  @redis.set(key, serialized_context, ex: durability_ttl)
end

#store_correlation_id(correlation_id, context_id, reactor_class_name) ⇒ Object

Raises:



142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 142

def store_correlation_id(correlation_id, context_id, reactor_class_name)
  key = correlation_id_key(correlation_id, reactor_class_name)
  # Store mapping correlation_id -> context_id
  # Try to set if not exists
  success = @redis.set(key, context_id, nx: true, ex: durability_ttl)

  return if success

  # If it exists, check if it's the same context_id
  existing_context_id = @redis.get(key)

  if existing_context_id == context_id
    # Refresh TTL
    @redis.expire(key, durability_ttl)
    return
  end

  raise Error::ValidationError, "Correlation ID '#{correlation_id}' already exists"
end

#store_map_element_context_id(map_id, context_id, reactor_class_name) ⇒ Object 



250
251
252
253
254
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 250

def store_map_element_context_id(map_id, context_id, reactor_class_name)
  key = map_element_contexts_key(map_id, reactor_class_name)
  @redis.rpush(key, context_id)
  @redis.expire(key, durability_ttl)
end

#store_map_failed_context_id(map_id, context_id, reactor_class_name) ⇒ Object 



266
267
268
269
270
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 266

def store_map_failed_context_id(map_id, context_id, reactor_class_name)
  key = map_failed_context_key(map_id, reactor_class_name)
  # Only store the first failure (nx: true)
  @redis.set(key, context_id, nx: true, ex: durability_ttl)
end

#store_map_result(map_id, index, serialized_result, reactor_class_name, strict_ordering: true) ⇒ Object

Durable map storage is ALWAYS index-keyed (HSET), regardless of strict_ordering. The index->slot mapping makes completion recoverable (missing = (0…count) - HKEYS) and re-dispatch idempotent (re-running index i overwrites slot i, never duplicates). strict_ordering is now only a read-order convenience, not a storage-layout switch (Phase 5). rubocop:disable Lint/UnusedMethodArgument



39
40
41
42
43
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 39

def store_map_result(map_id, index, serialized_result, reactor_class_name, strict_ordering: true)
  key = map_results_key(map_id, reactor_class_name)
  @redis.hset(key, index.to_s, serialized_result.to_json)
  @redis.expire(key, durability_ttl)
end

#subscribe(channel, &block) ⇒ Object 



177
178
179
 
# File 'lib/ruby_reactor/storage/redis_adapter.rb', line 177

def subscribe(channel, &block)
  @redis.subscribe(channel, &block)
end