Class: Nexo::RunStore::Memory

Inherits:
Object
  • Object
show all
Defined in:
lib/nexo/run_store.rb

Overview

In-memory backend used by the plain-Ruby path and the offline test suite. Runs are held in a process-wide Hash keyed by their UUID id so that a run created by Workflow.run is still findable through a later RunStore.default call (e.g. Workflow.logs) — each call builds a fresh Memory instance, but they all share the same underlying store, mirroring how the ActiveRecord backend shares one database. Nothing is persisted to disk; the store lives only for the process.

Defined Under Namespace

Classes: Run

Class Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Class Attribute Details

.mutex ⇒ Object (readonly)

The store-wide lock guarding runs and every run mutation. Not reentrant, so no method that holds it calls another that grabs it.



96
97
98
 
# File 'lib/nexo/run_store.rb', line 96

def mutex
  @mutex
end

.runs ⇒ Object (readonly)

The shared run table. Ids are UUIDs, so runs from independent callers never collide.



92
93
94
 
# File 'lib/nexo/run_store.rb', line 92

def runs
  @runs
end

Class Method Details

.reset! ⇒ Object

Clears the shared table. Intended for test isolation.



99
100
101
 
# File 'lib/nexo/run_store.rb', line 99

def reset!
  @mutex.synchronize { @runs = {} }
end

Instance Method Details

#claim_for_resume!(run) ⇒ Object

Atomically claims a "suspended" run for resume: flips it to "running" and returns true only if it was still suspended, so two concurrent resumes can't both re-enter #call (Spec 13 double-execution guard). The status flip is direct (not via #update!) to avoid re-entering the non-reentrant mutex.



130
131
132
133
134
135
136
137
 
# File 'lib/nexo/run_store.rb', line 130

def claim_for_resume!(run)
  self.class.mutex.synchronize do
    return false unless run.status == "suspended"

    run.status = "running"
    true
  end
end

#create(workflow_class:, payload:) ⇒ Object

Builds a fresh "pending" Run (UUID id, empty events/artifacts/state) and stores it in the process-wide table, returning it.



106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/nexo/run_store.rb', line 106

def create(workflow_class:, payload:)
  run = Run.new(
    id: Nexo.generate_run_id,
    workflow_class: workflow_class,
    status: "pending",
    payload: payload,
    result: nil,
    error: nil,
    events: [],
    artifacts: [],
    state: {}
  )
  self.class.mutex.synchronize { self.class.runs[run.id] = run }
end

#find(id) ⇒ Object

Fetches a run by its UUID string id. A miss raises KeyError, which is acceptable for v1.



123
 
# File 'lib/nexo/run_store.rb', line 123

def find(id) = self.class.mutex.synchronize { self.class.runs.fetch(id) }