Module: RubyReactor::Storage::RedisStepResults

Included in:
RedisAdapter
Defined in:
lib/ruby_reactor/storage/redis_step_results.rb

Overview

The durable outcome of one async_step, keyed by (parent context, step name). It lives OUTSIDE the parent's serialized context blob on purpose: a worker writes it concurrently with the still-running parent, and two writers on one blob race.

The dispatched record is written before the job is enqueued, so it also serves as the re-attach marker on recovery — "a record exists" is exactly the question "was this already dispatched?".

Instance Method Summary collapse

Instance Method Details

#retrieve_step_result(context_id, step_name, reactor_class_name) ⇒ Object 



21
22
23
24
25
26
 
# File 'lib/ruby_reactor/storage/redis_step_results.rb', line 21

def retrieve_step_result(context_id, step_name, reactor_class_name)
  json = @redis.get(step_result_key(context_id, step_name, reactor_class_name))
  return nil unless json

  JSON.parse(json)
end

#scan_step_results(count: 1000) ⇒ Object

Every record, for StepSweeper — a unit whose job was lost leaves nothing else behind to find it by, since the parent only parks on the read side.



30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/ruby_reactor/storage/redis_step_results.rb', line 30

def scan_step_results(count: 1000)
  records = []

  @redis.scan_each(match: "reactor:*:context:*:step_result:*", count: 100) do |key|
    json = @redis.get(key)
    records << JSON.parse(json) if json
    return records if records.size >= count
  end

  records
end

#store_step_result(context_id, step_name, record, reactor_class_name) ⇒ Object 



14
15
16
17
18
19
 
# File 'lib/ruby_reactor/storage/redis_step_results.rb', line 14

def store_step_result(context_id, step_name, record, reactor_class_name)
  key = step_result_key(context_id, step_name, reactor_class_name)
  # Shares context_ttl with the parent: the record must not outlive what it
  # belongs to, and must not expire before it either.
  @redis.set(key, JSON.generate(record), ex: durability_ttl)
end