Background jobs and scheduled work

There is no job queue and no thread. There is one durable object with one timer, and it is enough for more than it sounds like.

This is the question every new melee app runs into. Thread.new does not exist. There is no Sidekiq, no ActiveJob, no after_commit ... later, and no cron. The build step will refuse Thread and Queue by name.

The reason is the process model: a request is handled by a process that exits when the response is written. A thread you started would be killed with it, and there is nothing left running to hand work to.

What there is: a durable object — a named, long-lived object in the app’s worker process, with its own SQLite database and one timer. The server knows when each object’s timer is next due and starts the app at that moment, even if nobody has visited for days.

flowchart LR
  req["request child<br/>exits when the response is sent"] -- "record the work" --> store[("the object's storage")]
  timer(["the object's timer fires<br/>server starts the app for it"]) --> work["on_timer<br/>does the work"]
  work --> store
  work -- "re-arm" --> timer

Decide which of four things you have

1. Do it in the request

The commonest right answer. A request is its own process with a 30-second budget, so a 200 ms API call or a slow query is not hurting anyone else. There is no shared runtime to protect.

post "/subscribe" do
  res = HTTP.post("https://api.example/subscribe", body: params.fetch(:email), timeout: 5)
  halt 502, "Upstream is unhappy" unless res.ok?
  redirect "/thanks"
end

The limits to respect: the request times out at 30 seconds (the process group is killed and the caller gets a 504), and your app runs four requests at a time by default, so a slow route does eat a slot.

2. A schedule

class Refresher < Durable
  def setup = timer(after: 0)               # run once, now, when the object is first created

  def on_timer
    refresh
    timer after: 600                        # then every ten minutes
  end

  def refresh
    body = Melee::HTTP.get("https://example.com/feed.json", timeout: 10).body
    storage.put("feed", JSON.parse(body))
    storage.put("refreshed_at", Time.now.to_i)
  rescue Melee::HTTP::Error => e
    log.warn "refresh failed", error: e.message
  end

  def feed = storage.get("feed")
end

get "/" do
  render :index, feed: Refresher.get("main").feed
end

Four things about this shape are deliberate:

• on_timer re-arms. There is no repeating timer. That is a feature: a run that raises does not schedule the next one, so a broken job stops rather than hammering. If you want it to keep trying regardless, re-arm first and do the work after, or re-arm in an ensure.
• The rescue is inside. An exception out of on_timer is logged by the worker, but the timer is gone. Catch what you expect.
• State goes through storage. An instance variable survives only while the worker keeps the object loaded, which it stops doing on idle, deploy or restart.
• The route reads, the timer writes. Visitors get a cheap read of something already fetched. This is the main reason to reach for a timer at all.

For “at 3am”, compute the next occurrence rather than counting seconds:

def on_timer
  do_the_nightly_thing
  timer at: next_3am
end

def next_3am
  now = Time.now
  today = Time.local(now.year, now.month, now.day).to_i + 3 * 3600
  today > now.to_i ? today : today + 86_400
end

There is no Date and no Time.parse in this dialect — Time.local, Time.at, Time.now and strftime are what you have.

3. A queue

When a request should cause work that need not block the response, write the work down and let a timer drain it. The object’s storage is a SQLite database; a queue is a table.

-- db/objects/outbox/001_queue.sql
CREATE TABLE jobs (
  id         INTEGER PRIMARY KEY,
  kind       TEXT NOT NULL,
  payload    TEXT NOT NULL,
  attempts   INTEGER NOT NULL DEFAULT 0,
  created_at INTEGER NOT NULL
);

class Outbox < Durable
  MAX_ATTEMPTS = 5

  def setup = timer(after: 0)

  def push(kind, payload)
    storage.run "INSERT INTO jobs (kind, payload, created_at) VALUES (?, ?, ?)",
                kind, JSON.generate(payload), Time.now.to_i
    timer after: 1
    nil
  end

  def on_timer
    drain
    timer after: pending.zero? ? 60 : 5
  end

  def pending = storage.first("SELECT COUNT(*) AS n FROM jobs")["n"].to_i

  private

  def drain
    storage.query("SELECT id, kind, payload, attempts FROM jobs ORDER BY id LIMIT 20").each { |job| run_one(job) }
  end

  def run_one(job)
    deliver(job["kind"].to_s, JSON.parse(job["payload"].to_s))
    storage.run "DELETE FROM jobs WHERE id = ?", job["id"].to_i
  rescue StandardError => e
    attempts = job["attempts"].to_i + 1
    log.warn "job failed", id: job["id"], kind: job["kind"], attempts: attempts, error: e.message
    if attempts >= MAX_ATTEMPTS
      storage.run "DELETE FROM jobs WHERE id = ?", job["id"].to_i
      log.error "job given up", id: job["id"], kind: job["kind"]
    else
      storage.run "UPDATE jobs SET attempts = ? WHERE id = ?", attempts, job["id"].to_i
    end
  end

  def deliver(kind, payload)
    case kind                       # `case`, not send: nothing dispatches on a computed name here
    when "welcome" then Melee::HTTP.post("https://api.example/mail", body: JSON.generate(payload), timeout: 10)
    else log.warn "unknown job kind", kind: kind
    end
  end
end

From a route:

post "/signup" do
  db.run "INSERT INTO users (email) VALUES (?)", params.fetch(:email)
  Outbox.get("main").push("welcome", { "email" => params.fetch(:email) })
  redirect "/thanks"
end

Note timer after: 1 inside push: the object pulls its own timer forward so a new job is picked up in about a second rather than at the next scheduled sweep.

The honest caveat. push writes to the object’s database and the user row went into the app’s database. Those are two files and two transactions, so a crash between them can lose the job. If exactly-once matters, keep both sides in one database — put the users table in the object’s storage too, and write the row and the job in a single storage.transaction.

4. Retries and idempotence

Anything a timer does can run twice: the worker can be killed mid-call, the object reloaded, the timer fired again. Make the work idempotent — a DELETE after success, an upsert keyed on something stable, an external call with an idempotency key — rather than assuming it happens once.

What this costs you

Be clear-eyed about the ceilings before you build something big on it.

• One call at a time, per app. The worker hosting every durable object in the app is single-threaded. A job that takes ten seconds delays every other object call in that app for ten seconds, including the ones a route is waiting on. Keep durable methods short; do the slow part in chunks across timer fires if you have to.
• One pending timer per object. Not one per job. If you need several schedules, that is several objects — they each get their own timer and their own database.
• No later and no every. Both are named as future work; today you write the re-arm yourself.
• No fan-out. Nothing runs two jobs in parallel.
• Timer precision is “soon after”, not “on the tick”. The server starts the app for the timer; a fire is not to-the-millisecond, and an app under a deploy or a restart picks its timer up afterwards.
• A timer keeps the app awake. An object with a 5-second timer means the app is started every 5 seconds, forever, which is the opposite of melee’s whole economics. Prefer minutes.

Things that are not the answer

See also

• Durable objects — the full API: setup, destroy, migrations, the JSON-shape rules, melee objects.
• Working with the database — transactions and the one-writer rule.
• What melee cannot do.