bug fixes

scheduler.rb

@@ -1,4 +1,4 @@
-#!/usr/bin/env ruby
+#!/usr/bin/env jruby
 # scheduler.rb - Clock-driven show/stream scheduler for the radio automation.
 #
 # Designed to run every minute from cron as the liquidsoap user:
@@ -9,13 +9,17 @@
 # <storage>/queues/current_queue.txt so station.liq picks it up within ~2s.
 #
 # Resolution rules:
-#   stream -> the URL is written verbatim. RUNLENGTH (seconds) is carried in a
-#             comment line; liquidsoap plays the live stream until the next
-#             scheduled event overwrites the file.
+#   stream -> ffmpeg captures the live URL to a finite MP3 in captures/ for
+#             RUNLENGTH seconds (default 3600) and that LOCAL path is queued.
 #   show   -> look up the slug in subscriptions.db, find the next unplayed
 #             episode in played.db (prefer local_path, fall back to url),
 #             mark it played, and write that path/url.
 #
+# Expiry: each queued item gets an end time in queues/current_queue.meta
+# (RUNLENGTH for shows, capped at 4h without one; RUNLENGTH+300s for stream
+# captures). Every cron tick clears current_queue.txt once its item has
+# expired, so background music resumes and nothing loops.
+#
 # Idempotent: a fire marker prevents re-firing the same entry twice in one
 # minute if cron double-runs or the script is invoked manually.
 
@@ -23,13 +27,11 @@ require 'json'
 require 'sequel'
 require 'fileutils'
 require 'time'
+require 'uri'
 
 SCRIPT_DIR = File.expand_path(File.dirname(__FILE__))
 CONFIG_PATH = File.join(SCRIPT_DIR, "config.json")
 SCHEDULE_FILE = File.join(SCRIPT_DIR, "schedule.txt")
-QUEUE_DIR = nil   # derived below
-QUEUE_FILE = nil  # derived below
-FIRE_MARKER_DIR = nil
 
 CFG = JSON.parse(File.read(CONFIG_PATH))
 STORAGE = CFG["storage"]
@@ -39,11 +41,32 @@ PLAYED_DB = File.join(STATE_DIR, "played.db")
 LOG_DIR = File.join(STORAGE, "logs")
 
 QUEUE_DIR = File.join(STORAGE, "queues")
-QUEUE_FILE = File.join(QUEUE_DIR, "current_queue.txt")
+# Single queue file, consumed by one request.dynamic in station.liq. Both show
+# and stream entries write their URI here; whichever is due overwrites it.
+QUEUE_FILE      = File.join(QUEUE_DIR, "current_queue.txt")
+# Sidecar for QUEUE_FILE: a single epoch integer = the wall-clock second at
+# which the queued item stops airing. Written whenever a URI is queued with a
+# TTL, deleted whenever the queue is cleared. Each cron tick checks it and
+# clears the queue once expired, so a finished episode never replays and
+# background music resumes on its own.
+QUEUE_META      = File.join(QUEUE_DIR, "current_queue.meta")
 FIRE_MARKER_DIR = File.join(STATE_DIR, "fire_markers")
+# Live-capture output dir: ffmpeg writes scheduled stream slots here as finite
+# MP3s that station.liq plays like any other local file.
+CAPTURE_DIR = File.join(STORAGE, "captures")
 
+# Create dirs and make them world-writable-ish (0775) so the cron-run process
+# can write into them even though the installer's chown ran before queues/ existed.
 FileUtils.mkdir_p(QUEUE_DIR)
 FileUtils.mkdir_p(FIRE_MARKER_DIR)
+FileUtils.mkdir_p(CAPTURE_DIR)
+begin
+  FileUtils.chmod(0o775, QUEUE_DIR)
+  FileUtils.chmod(0o775, FIRE_MARKER_DIR)
+  FileUtils.chmod(0o775, CAPTURE_DIR)
+rescue => e
+  log("WARN: could not chmod queue/fire-marker/capture dirs: #{e.message}")
+end
 
 def log(msg)
   puts "#{Time.now.strftime('%Y-%m-%d %H:%M:%S')} #{msg}"
@@ -142,12 +165,23 @@ end
 
 # --- Database helpers ----------------------------------------------------------
 
+# JRuby: Sequel.sqlite() loads the CRuby-only 'sqlite3' gem and raises
+# AdapterNotFound. Use the JDBC form exactly as fetch_podcasts.rb does
+# (verified-working there), and set PRAGMAs via Database#execute.
 def open_subs_db
-  Sequel.sqlite(SUBS_DB, database_options: { timeout: 5 })
+  db = Sequel.connect("jdbc:sqlite:" + SUBS_DB)
+  db.execute("PRAGMA journal_mode=WAL;")
+  db.execute("PRAGMA busy_timeout=5000;")
+  db.execute("PRAGMA foreign_keys=OFF;")
+  db
 end
 
 def open_played_db
-  Sequel.sqlite(PLAYED_DB, database_options: { timeout: 5 })
+  db = Sequel.connect("jdbc:sqlite:" + PLAYED_DB)
+  db.execute("PRAGMA journal_mode=WAL;")
+  db.execute("PRAGMA busy_timeout=5000;")
+  db.execute("PRAGMA foreign_keys=OFF;")
+  db
 end
 
 # Resolve a show slug to the next unplayed episode's playable path/url.
@@ -162,16 +196,26 @@ def resolve_show(slug)
     return nil
   end
 
-  ep = played_db[:episodes]
-        .where(show_guid: show[:guid], played: 0)
-        .order(Sequel.desc(:created_at))
-        .first
+  total_eps   = played_db[:episodes].where(show_guid: show[:guid]).count
+  unplayed_ep = played_db[:episodes]
+                   .where(show_guid: show[:guid], played: 0)
+                   .order(Sequel.desc(:created_at))
+                   .first
 
-  if ep.nil?
-    log("INFO: no unplayed episodes for show '#{slug}' (#{show[:title]})")
+  if unplayed_ep.nil?
+    # Distinguish the two failure modes so a silent "no audio" is diagnosable:
+    #   * zero episodes at all  -> fetch never populated this show's feed
+    #   * episodes but all played -> archive-cycle reset isn't running
+    if total_eps.zero?
+      log("WARN: show '#{slug}' (#{show[:title]}) has ZERO episodes in played.db; run fetch_podcasts.rb --fetch-all to populate it")
+    else
+      log("WARN: show '#{slug}' (#{show[:title]}) has #{total_eps} episode(s) but ALL are marked played; nothing to queue until new episodes arrive or flags reset")
+    end
     return nil
   end
 
+  ep = unplayed_ep
+
   playable = ep[:local_path] || ep[:url]
   if playable.nil? || playable.to_s.strip.empty?
     log("WARN: episode #{ep[:guid]} for '#{slug}' has neither local_path nor url")
@@ -188,26 +232,207 @@ ensure
   played_db&.disconnect
 end
 
-# --- Queue writer --------------------------------------------------------------
-# Writes the URI (and optionally a runlength hint) atomically: write to a temp
-# file in the same directory, then rename over the target. This prevents
-# station.liq from reading a partially-written file.
+# --- Queue writer ------------------------------------------------------------
+# The single queue file must contain ONLY the bare URI, nothing else. station.liq
+# trims whitespace before handing it to request.create, but a second line would
+# still corrupt the URI, so we keep it to exactly one token.
+# Atomic write: temp file in the same dir, then rename over the target so
+# station.liq never reads a partially-written file.
 
-def write_queue(uri, runlength_sec = nil)
-  tmp = "#{QUEUE_FILE}.tmp"
-  content = uri
-  if runlength_sec && runlength_sec > 0
-    content += "\n# runlength=#{runlength_sec}"
+# ttl_sec: when > 0, stamp QUEUE_META with an expiry time (now + ttl_sec) so a
+# later cron tick clears the queue once the slot is over. When nil/0, leave any
+# existing meta untouched (the queue keeps whatever expiry it already had).
+def write_uri_file(path, uri, ttl_sec = nil)
+  tmp = "#{path}.tmp"
+  begin
+    File.write(tmp, uri)
+    File.rename(tmp, path)
+  rescue Errno::EACCES => e
+    log("WARN: atomic write to #{path} blocked (#{e.message}); falling back to direct write")
+    File.write(path, uri)
+    File.chmod(0o664, path) if File.exist?(path)
+  ensure
+    File.delete(tmp) if File.exist?(tmp)
+  end
+  if ttl_sec && ttl_sec.to_i > 0
+    write_queue_meta(Time.now.to_i + ttl_sec.to_i)
+  end
+end
+
+# Best-effort: expiry bookkeeping must never block a fire.
+def write_queue_meta(epoch)
+  tmp = "#{QUEUE_META}.tmp"
+  begin
+    File.write(tmp, epoch.to_i.to_s)
+    File.rename(tmp, QUEUE_META)
+  rescue => e
+    log("WARN: could not write queue expiry meta: #{e.message}; queue will persist until next overwrite")
+  ensure
+    begin
+      File.delete(tmp) if File.exist?(tmp)
+    rescue StandardError
+      nil
+    end
+  end
+end
+
+# Returns the queue expiry as a Time, or nil when no (valid) meta exists.
+def queue_expiry
+  return nil unless File.exist?(QUEUE_META)
+  raw = File.read(QUEUE_META).strip
+  if raw.match?(/\A\d+\z/)
+    Time.at(raw.to_i)
+  else
+    Time.parse(raw)
+  end
+rescue ArgumentError
+  nil
+end
+
+# Lazy expiry check, run at the top of every cron tick: if the queued item is
+# past its end time, clear the queue. station.liq's request.dynamic then goes
+# fallible and the fallback drops back to background music within ~2s.
+# Worst-case overshoot is one cron minute.
+def expire_queue_if_due(now)
+  exp = queue_expiry
+  return if exp.nil?
+  return if now.to_i < exp.to_i
+  log("EXPIRED: queued item past its end time (#{exp.strftime('%Y-%m-%d %H:%M:%S')}); clearing queue -> background music")
+  clear_uri_file(QUEUE_FILE)
+end
+
+def clear_uri_file(path)
+  begin
+    File.delete(QUEUE_META) if File.exist?(QUEUE_META)
+  rescue => e
+    log("WARN: could not delete #{QUEUE_META}: #{e.message}")
+  end
+  return unless File.exist?(path)
+  begin
+    File.write("#{path}.empty", "")
+    File.rename("#{path}.empty", path)
+  rescue => e
+    log("WARN: could not clear #{path}: #{e.message}")
+  end
+end
+
+# Fire a SHOW entry: put the episode path/url into the single queue file.
+# runlength_sec (from schedule.txt, when given) becomes the queue TTL so the
+# entry self-expires at the end of its slot; without one we cap at 4 hours so
+# even an unbounded show eventually hands back to background music.
+def fire_show(playable, runlength_sec = nil)
+  ttl = (runlength_sec && runlength_sec > 0) ? runlength_sec : 4 * 3600
+  write_uri_file(QUEUE_FILE, playable, ttl)
+  log("QUEUED show: #{playable} (expires in #{ttl}s)")
+end
+
+# Build a deterministic capture path from the stream URL + fire time so we can
+# tell captures apart and avoid clobbering a still-playing one.
+def capture_path_for(url, now)
+  safe_host = URI.parse(url).host.to_s.gsub(/[^a-zA-Z0-9.-]/, "_")
+  ts        = now.strftime("%Y%m%d_%H%M%S")
+  File.join(CAPTURE_DIR, "#{safe_host}_#{ts}.mp3")
+end
+
+# Prune captured MP3s older than 2 days so the dir doesn't grow unbounded.
+def prune_old_captures(now)
+  cutoff = now - 172_800 # 2 days
+  Dir.glob(File.join(CAPTURE_DIR, "*.mp3")).each do |f|
+    begin
+      File.delete(f) if File.mtime(f) < cutoff
+    rescue => e
+      log("WARN: could not prune capture #{f}: #{e.message}")
+    end
+  end
+end
+
+# Fire a STREAM entry by capturing it locally with ffmpeg, then queuing the
+# LOCAL PATH (not the live URL). This sidesteps liquidsoap's inability to relay
+# an infinite Icecast mount: station.liq just plays a finite MP3 like any show.
+#   * ffmpeg runs in the background, stops after RUNLENGTH seconds (-t).
+#   * If no runlength is given, default to 3600s (1h) so the capture is bounded.
+#   * We wait ~5s for ffmpeg to actually start writing before queuing, so
+#     station.liq picks up a real (non-empty) file. If ffmpeg fails to start,
+#     we fall back to queuing nothing (background music carries the slot).
+def fire_stream(url, runlength_sec = nil)
+  dur = (runlength_sec && runlength_sec > 0) ? runlength_sec : 3600
+  cap = capture_path_for(url, Time.now)
+
+  # JRuby's native spawn launcher does NOT accept an Array command combined
+  # with a redirect hash (raises "wrong first argument"), and it does not
+  # reliably accept open IO objects as redirect values ("wrong exec redirect").
+  # The documented-working form on JRuby is: a SINGLE shell String for the
+  # command, plus plain FILENAME strings for :out/:err. We quote every path
+  # and the URL so shell word-splitting can't break them.
+  sh_cmd = %(ffmpeg -hide_banner -loglevel error -i "#{url}" -t #{dur} -acodec libmp3lame -b:a 128k -y "#{cap}")
+  fflog  = "#{cap}.fflog"
+  pid    = spawn(sh_cmd, out: fflog, err: fflog)
+  log("STREAM CAPTURE started: #{url} -> #{cap} (pid=#{pid}, dur=#{dur}s)")
+
+  # Give ffmpeg a moment to open the input and start writing. We judge success
+  # purely by whether the capture file has real bytes in it -- that's the actual
+  # signal station.liq needs. (Process.waitpid2 with WNOHANG returns nil on
+  # JRuby for a freshly spawned child, so we don't use it for liveness.)
+  sleep 5
+  size = File.exist?(cap) ? File.size(cap) : 0
+
+  if size > 0
+    # TTL = capture length + 5 min slack: ffmpeg stops writing after dur, and
+    # the queue self-clears shortly after the capture has finished playing.
+    write_uri_file(QUEUE_FILE, cap, dur + 300)
+    log("QUEUED stream capture: #{cap} (size=#{size} bytes so far, expires in #{dur + 300}s)")
+  else
+    # ffmpeg wrote nothing within 5s -- clear the queue so background music
+    # carries the slot instead of liquidsoap replaying a stale/broken URI.
+    log("WARN: capture not ready (size=#{size}); clearing queue -> background music")
+    clear_uri_file(QUEUE_FILE)
+    ff_err = File.read("#{cap}.fflog") rescue "<no log>"
+    log("ffmpeg stderr: #{ff_err.strip[0, 400]}")
   end
-  File.write(tmp, content)
-  File.rename(tmp, QUEUE_FILE)
-  log("QUEUED: #{uri}#{runlength_sec ? " (runlength=#{runlength_sec}s)" : ''}")
 end
 
 # --- Main ----------------------------------------------------------------------
 
 def main
   now = Time.now
+
+  # Lazy expiry first: whatever was queued earlier stops airing at its slot
+  # end (or the 4h show cap / capture+5min stream cap) without needing a new
+  # scheduled event to overwrite it.
+  expire_queue_if_due(now)
+
+  # Dev hook: --dryfire-stream URL [SECS] fires one stream capture immediately,
+  # bypassing the clock match. Lets us test the ffmpeg-capture path on demand
+  # instead of waiting for a scheduled minute. Example:
+  #   ./run_radio.sh scheduler.rb --dryfire-stream "https://streams.kpfa.org:8443/kpfa" 300
+  if ARGV[0] == "--dryfire-stream"
+    url = ARGV[1]
+    secs = ARGV[2]&.to_i
+    raise ArgumentError, "--dryfire-stream requires a URL" if url.nil? || url.empty?
+    log("[dryfire] triggering stream capture now: #{url} (#{secs || 3600}s)")
+    fire_stream(url, secs)
+    log("[dryfire] done.")
+    return
+  end
+
+  # Dev hook: --dryfire-show SLUG runs the exact same resolve_show -> fire_show
+  # path a scheduled "show" entry uses, but immediately, without waiting for the
+  # clock match. Marks the chosen episode played=1 just like a real fire.
+  #   ./run_radio.sh scheduler.rb --dryfire-show the_history_of_rome
+  if ARGV[0] == "--dryfire-show"
+    slug = ARGV[1]
+    raise ArgumentError, "--dryfire-show requires a slug" if slug.nil? || slug.empty?
+    log("[dryfire] resolving show now: #{slug}")
+    uri = resolve_show(slug)
+    if uri
+      fire_show(uri)
+      log("[dryfire] queued -- station.liq should pick it up on its next poll")
+    else
+      log("[dryfire] nothing queued (see WARN above); background music continues")
+    end
+    return
+  end
+
   log("--- scheduler tick #{now.strftime('%Y-%m-%d %H:%M:%S')} ---")
 
   unless File.exist?(SCHEDULE_FILE)
@@ -216,6 +441,7 @@ def main
   end
 
   prune_old_markers(now)
+  prune_old_captures(now)
   entries = parse_schedule(SCHEDULE_FILE)
   log("Loaded #{entries.length} schedule entries")
 
@@ -235,14 +461,14 @@ def main
 
     case entry[:type]
     when "stream"
-      write_queue(entry[:target], entry[:runlength])
+      fire_stream(entry[:target], entry[:runlength])
       mark_fired(key)
       fired_any = true
 
     when "show"
       uri = resolve_show(entry[:target])
       if uri
-        write_queue(uri)
+        fire_show(uri, entry[:runlength])
         mark_fired(key)
         fired_any = true
       else