Class: Nexo::RunStore::Disk

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

Overview

File-backed backend for a host with no database: one JSON document per run under dir, rewritten whenever the run changes. This is what makes durability available outside Rails — checkpoint, suspend/resume and a run's recorded artifacts all read state back from the store, and Memory dies with the process.

Nexo.config.run_store = Nexo::RunStore::Disk.new(dir: "~/.local/state/myapp/runs")

The Run is Memory's, subclassed: the shape and every read helper are shared, so a host written against one store behaves identically on the other. Only the save_* hooks differ — they write the document. push_event/push_artifact deliberately do NOT, because Workflow always calls the matching save_* right after, and persisting twice would double the writes for nothing.

Scope, honestly: rewriting the whole document per change is fine for the run sizes this is meant for (a CLI's own history) and wrong for a busy multi-worker queue — that is what the ActiveRecord backend is for. claim_for_resume! is atomic within a process but NOT across processes; two machines resuming the same run concurrently is out of scope. Runs are never pruned here: retention is the host's policy, and dir is a plain directory it can manage.

Defined Under Namespace

Classes: Run

Constant Summary collapse

MUTEX =

Serializes writes within the process, mirroring Memory's store-wide lock.

Mutex.new

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(dir:) ⇒ Disk

dir is created on demand; ~ is expanded.



232
233
234
 
# File 'lib/nexo/run_store.rb', line 232

def initialize(dir:)
  @dir = ::File.expand_path(dir)
end

Instance Attribute Details

#dir ⇒ Object (readonly)

The directory runs are written to.



237
238
239
 
# File 'lib/nexo/run_store.rb', line 237

def dir
  @dir
end

Instance Method Details

#all ⇒ Object

Every run in the directory, newest-written first. Not part of the store contract Workflow calls — it is here because a host with a run directory inevitably wants to list it ("what did the last run produce?").



262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/nexo/run_store.rb', line 262

def all
  return [] unless ::File.directory?(@dir)

  Dir.glob(::File.join(@dir, "*.json"))
    .sort_by { |p| -::File.mtime(p).to_f }
    .filter_map do |p|
      hydrate(JSON.parse(::File.read(p)), p)
    rescue JSON::ParserError, SystemCallError
      # A half-written or hand-edited document must not break the listing.
      nil
    end
end

#claim_for_resume!(run) ⇒ Object

Atomically claims a "suspended" run for resume, re-reading from disk so a stale in-memory copy cannot win. Within one process only — see the class note.



278
279
280
281
282
283
284
285
286
287
288
 
# File 'lib/nexo/run_store.rb', line 278

def claim_for_resume!(run)
  MUTEX.synchronize do
    current = find(run.id)
    return false unless current.status == "suspended"

    run.status = "running"
    run.path ||= path_for(run.id)
    run.persist!
    true
  end
end

#create(workflow_class:, payload:) ⇒ Object

Builds a fresh "pending" Run, writes it, and returns it.



240
241
242
243
244
245
246
247
248
249
 
# File 'lib/nexo/run_store.rb', line 240

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: {}
  )
  ::FileUtils.mkdir_p(@dir)
  run.path = path_for(run.id)
  run.persist!
  run
end

#find(id) ⇒ Object

Reads a run back by id. A miss raises KeyError, matching Memory.

Raises:

  • (KeyError) 


252
253
254
255
256
257
 
# File 'lib/nexo/run_store.rb', line 252

def find(id)
  path = path_for(id)
  raise KeyError, "run not found: #{id}" unless ::File.exist?(path)

  hydrate(JSON.parse(::File.read(path)), path)
end