Class: Xeno::ScheduleJob

Inherits:
ApplicationJob show all
Defined in:
app/jobs/xeno/schedule_job.rb

Overview

Fires one scheduled run: a fire-and-forget task-mode session under the app principal. Solid Queue recurring entries (rake xeno:schedules:sync) and the dev dispatch endpoint both enqueue this.

At-least-once safe: queues redeliver, and Solid Queue's recurring (task, run-at) dedup only covers the enqueue, not the execution. Each enqueue carries one ActiveJob job_id; the dedup row claims it in the SAME transaction that opens the session, so a redelivered tick can never double-open — it re-enqueues the recorded turn instead (idle no-op if the first delivery finished: the claim CAS decides).

Constant Summary collapse

DEDUP_SCOPE = 
"schedule_run".freeze

Instance Method Summary collapse

Instance Method Details

#perform(name) ⇒ Object

Raises:

  • (ArgumentError) 


19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'app/jobs/xeno/schedule_job.rb', line 19

def perform(name)
  definition = Xeno.definition
  schedule = definition.schedules[name]
  raise ArgumentError, "unknown schedule: #{name}" unless schedule

  turn = nil
  claimed = Session.transaction do
    Dedup.claim(DEDUP_SCOPE, job_id, metadata: { "schedule" => name }).tap do |row|
      next unless row

      session = Session.open!(
        message: schedule.prompt,
        channel: "schedule",
        principal: { "type" => "app", "schedule" => name },
        definition: definition
      )
      turn = session.turns.order(:sequence).last
      row.update!(metadata: row.metadata.merge("session_id" => session.id, "turn_id" => turn.id))
    end
  end

  turn ||= redelivered_turn(claimed)
  TurnJob.perform_now(turn.id) if turn
end