Class: Hammer::CronServer

Inherits:
Object
  • Object
show all
Defined in:
lib/hammer/cron_server.rb

Overview

The hammer h:cron job server. Discovers every task that declares a cron '<expr>' schedule, ticks once per minute and runs each due job in its own subprocess (a fresh hammer ns:task, exactly like the macOS GUI runs tasks), streaming stdout+stderr into a per-job log. A small web UI (Hammer::CronWeb) serves job status and logs on localhost.

On-disk layout, relative to the Hammerfile dir:

log/hammer/<slug>.log      current job log (db:backup -> db-backup.log)
log/hammer/<slug>.log.1    previous log, rotation keeps max 2 files
tmp/hammer/cron.state.json last-run timestamps + daemon pid/port

Defined Under Namespace

Classes: Job

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(root_klass, port:, pass: nil) ⇒ CronServer

Returns a new instance of CronServer.



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/hammer/cron_server.rb', line 36

def initialize(root_klass, port:, pass: nil)
  @root_dir   = Dir.pwd   # Hammer.cli already chdir-ed to the Hammerfile dir
  @port       = port
  @pass       = pass && !pass.to_s.empty? ? pass.to_s : nil
  @hammer_bin = begin
    File.realpath($PROGRAM_NAME)
  rescue Errno::ENOENT
    File.expand_path($PROGRAM_NAME)
  end
  @log_dir    = File.join(@root_dir, 'log/hammer')
  @state_path = File.join(@root_dir, 'tmp/hammer/cron.state.json')
  @mutex      = Mutex.new
  @threads    = []

  @jobs = {}
  root_klass.each_command(include_builtins: false) do |path, cmd|
    next unless cmd.cron
    @jobs[path] = Job.new(path: path, command: cmd, schedule: cmd.cron_schedule)
  end
  if @jobs.empty?
    raise Hammer::Error,
          "no tasks declare a cron schedule - add `cron '*/10 * * * *'` (or '10m', '@daily') inside a task block"
  end

  load_state
end

Instance Attribute Details

#jobs ⇒ Object (readonly)

Returns the value of attribute jobs.



34
35
36
 
# File 'lib/hammer/cron_server.rb', line 34

def jobs
  @jobs
end

#port ⇒ Object (readonly)

Returns the value of attribute port.



34
35
36
 
# File 'lib/hammer/cron_server.rb', line 34

def port
  @port
end

#root_dir ⇒ Object (readonly)

Returns the value of attribute root_dir.



34
35
36
 
# File 'lib/hammer/cron_server.rb', line 34

def root_dir
  @root_dir
end

Instance Method Details

#job?(path) ⇒ Boolean

Returns:

  • (Boolean) 


293
294
295
 
# File 'lib/hammer/cron_server.rb', line 293

def job?(path)
  @jobs.key?(path)
end

#launch(job, at, trigger:) ⇒ Object

Fire one run in a subprocess. Overlap policy: if the previous run of the SAME job is still alive, skip and note it in the job log - jobs that outlive their interval must not pile up. Different jobs run freely in parallel. The child runs the full normal hammer pipeline (Bundler, dotenv, before hooks, needs) in @root_dir with stdin closed, so anything interactive fails fast instead of hanging the scheduler.



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/hammer/cron_server.rb', line 130

def launch(job, at, trigger:)
  @mutex.synchronize do
    if job.running?
      append_line(job, "[h:cron] #{at.strftime('%F %T')} skipped (#{trigger}) - previous run still active")
      return
    end
    job.pid      = :starting
    job.status   = 'running'
    job.last_run = at
    save_state
  end

  @threads << Thread.new do
    begin
      rotate_log(job)
      FileUtils.mkdir_p(@log_dir)
      File.open(log_path(job), 'a') do |log|
        log.sync = true
        log.puts "===== #{job.path} | #{trigger} | #{at.strftime('%F %T')} ====="
        t0  = Process.clock_gettime(Process::CLOCK_MONOTONIC)
        pid = Process.spawn(
          { 'NO_COLOR' => '1', 'HAMMER_QUIET' => '1' },
          @hammer_bin, job.path,
          in: File::NULL, out: log, err: log, chdir: @root_dir
        )
        # Persist the real child pid right away - it's what lets a
        # restarted server find this run again if we die mid-flight.
        @mutex.synchronize do
          job.pid = pid
          save_state
        end
        _, status = Process.wait2(pid)
        dur = Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0
        log.puts "===== exit #{status.exitstatus.inspect} | #{format('%.1f', dur)}s ====="
        @mutex.synchronize do
          job.pid       = nil
          job.exit_code = status.exitstatus
          job.duration  = dur
          job.status    = status.success? ? 'ok' : 'failed'
          save_state
        end
      end
    rescue => e
      @mutex.synchronize do
        job.pid    = nil
        job.status = 'error'
        save_state
      end
      append_line(job, "[h:cron] run failed: #{e.class}: #{e.message}")
    end
  end
end

#load_state ⇒ Object

Tolerates a missing or corrupt file - scheduling state is best-effort, a fresh start is always safe. Jobs no longer declared in the Hammerfile simply are not restored (and get pruned on the next save).



229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
 
# File 'lib/hammer/cron_server.rb', line 229

def load_state
  data = begin
    JSON.parse(File.read(@state_path))
  rescue Errno::ENOENT
    return
  rescue JSON::ParserError
    warn Shell.paint("h:cron: ignoring corrupt #{@state_path}", :gray)
    return
  end
  @saved_pid = data['pid']
  (data['jobs'] || {}).each do |path, j|
    job = @jobs[path] or next
    job.last_run  = j['last_run'] ? Time.at(j['last_run']) : nil
    job.exit_code = j['exit']
    job.duration  = j['duration']
    job.status    = j['status']
    # A job saved as 'running' means the previous server went away
    # mid-run. If its subprocess is still alive we keep reporting it
    # as running (and the overlap guard keeps skipping new runs);
    # otherwise the run's exit status is lost - mark it 'unknown'
    # instead of leaving a stale 'running' badge forever.
    next unless job.status == 'running'
    if j['pid'] && process_alive?(j['pid'])
      job.pid = j['pid']
    else
      job.pid    = nil
      job.status = 'unknown'
    end
  end
end

#log_path(job) ⇒ Object

----- logs -----------------------------------------------------------



185
186
187
 
# File 'lib/hammer/cron_server.rb', line 185

def log_path(job)
  File.join(@log_dir, "#{job.file_slug}.log")
end

#print_service_unit ⇒ Object

Print a launchd plist (macOS) or systemd user unit (everything else) for running h:cron supervised. Unit text goes to stdout so it can be redirected into a file; install hints go to stderr.



265
266
267
268
269
270
271
272
273
274
275
276
 
# File 'lib/hammer/cron_server.rb', line 265

def print_service_unit
  if RUBY_PLATFORM.include?('darwin')
    label = "com.lux-hammer.cron.#{File.basename(@root_dir)}"
    puts launchd_service_unit(label)
    warn Shell.paint("# save to ~/Library/LaunchAgents/#{label}.plist, then:", :gray)
    warn Shell.paint("#   launchctl load ~/Library/LaunchAgents/#{label}.plist", :gray)
  else
    puts systemd_service_unit
    warn Shell.paint('# save to ~/.config/systemd/user/hammer-cron.service, then:', :gray)
    warn Shell.paint('#   systemctl --user enable --now hammer-cron', :gray)
  end
end

#print_startup_summary ⇒ Object

Table of scheduled jobs + next runs; doubles as h:cron --list.



96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/hammer/cron_server.rb', line 96

def print_startup_summary
  now  = Time.now
  rows = @jobs.values.map do |job|
    [job.path, job.schedule.source, format_time(job.last_run),
     format_time(job.schedule.next_run(now, last_run: job.last_run))]
  end
  widths = ['task', 'schedule', 'last run', 'next run'].each_with_index.map do |h, i|
    [h.length, *rows.map { |r| r[i].length }].max
  end
  header = ['task', 'schedule', 'last run', 'next run']
  Shell.say header.each_with_index.map { |h, i| h.ljust(widths[i]) }.join('  '), :yellow
  rows.each do |r|
    Shell.say r.each_with_index.map { |v, i| v.ljust(widths[i]) }.join('  ')
  end
end

#rotate_log(job) ⇒ Object

Called before each run: past 1 MB the current log rolls to .log.1 (POSIX rename clobbers the previous .1) and the run starts fresh. Max two files per job, ever.



192
193
194
195
196
197
198
199
200
 
# File 'lib/hammer/cron_server.rb', line 192

def rotate_log(job)
  path = log_path(job)
  size = begin
    File.size(path)
  rescue Errno::ENOENT
    0
  end
  File.rename(path, "#{path}.1") if size > MAX_LOG_BYTES
end

#run! ⇒ Object

Run scheduler + web UI until SIGINT/SIGTERM. Ticks on minute boundaries (or every second when a sub-minute interval job exists, see tick_step). The scheduler owns the main thread (so Ctrl-C lands in the thread that is sleeping); the HTTP accept loop runs in a background thread. Trap handlers only flip a flag - no IO or locking is allowed inside a trap.



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/hammer/cron_server.rb', line 71

def run!
  ensure_single_instance!
  @stop = false
  trap('INT')  { @stop = true }
  trap('TERM') { @stop = true }

  save_state
  @jobs.each_value { |job| watch_orphan(job) if job.running? }
  require_relative 'cron_web'
  @web = CronWeb.new(self, port: @port, pass: @pass).start
  print_startup_summary
  Shell.say "web ui: http://127.0.0.1:#{@port}#{' (password protected)' if @pass}", :cyan
  Shell.say 'Ctrl-C to stop', :gray

  until @stop
    tick = wait_for_tick or break
    each_job_snapshot do |job|
      launch(job, tick, trigger: 'cron') if job.schedule.due?(tick, job.last_run)
    end
  end

  shutdown!
end

#run_now(path) ⇒ Object

Trigger one job outside its schedule (web UI "run now" button). Same code path as a scheduled run; the log header says 'manual'. Returns false when the job is unknown.



115
116
117
118
119
 
# File 'lib/hammer/cron_server.rb', line 115

def run_now(path)
  job = @jobs[path] or return false
  launch(job, Time.now, trigger: 'manual')
  true
end

#save_state ⇒ Object

tmp/hammer/cron.state.json remembers last runs across restarts so a relaunched scheduler neither re-fires jobs that just ran nor forgets interval clocks. Also records our pid/port so a second h:cron (and the web UI) can see who is running.



208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
 
# File 'lib/hammer/cron_server.rb', line 208

def save_state
  data = {
    schema:     1,
    pid:        Process.pid,
    port:       @port,
    started_at: (@started_at ||= Time.now).to_i,
    jobs:       @jobs.transform_values do |j|
      { last_run: j.last_run&.to_i, exit: j.exit_code, duration: j.duration,
        status: j.status, pid: (j.pid if j.pid.is_a?(Integer)) }
    end
  }
  FileUtils.mkdir_p(File.dirname(@state_path))
  tmp = "#{@state_path}.tmp"
  File.write(tmp, JSON.pretty_generate(data))
  File.rename(tmp, @state_path)
end

#snapshot ⇒ Object

Plain-value copy of every job, taken under the mutex, so the web thread never renders half-updated state.



282
283
284
285
286
287
288
289
290
291
 
# File 'lib/hammer/cron_server.rb', line 282

def snapshot
  @mutex.synchronize do
    @jobs.values.map do |j|
      { path: j.path, desc: j.command.brief, cron: j.schedule.source,
        last_run: j.last_run, status: j.status, exit_code: j.exit_code,
        duration: j.duration, running: j.running?,
        next_run: j.schedule.next_run(Time.now, last_run: j.last_run) }
    end
  end
end