Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 44 additions & 28 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -44,63 +44,73 @@ bundle install
## Quickstart

**1. Define job classes.** A job is a plain class that subclasses
`Benchwarmer::Job` and implements `#perform`. Positional `perform` arguments come
straight from the enqueue call (round-tripped through JSON).
`Benchwarmer::Job` and implements `#perform`. `perform` can take **named
parameters** (delivered from an enqueued instance's constructor kwargs) or
**positional arguments** (splatted from an `args:` array) — either way they
round-trip through JSON.

```ruby
require "benchwarmer"

class GreetJob < Benchwarmer::Job
def self.description = "Say hello (shown in the UI enqueue form)"

def perform(name)
def perform(name:, language: "en")
append_output("Hello, #{name}!\n") # streams into the job's row
end
end
```

**2. Build a registry.** The registry is a plain `Hash` mapping each job's type
key to its class. `Job.job_type` derives the key from the class name
(`"GreetJob"` -> `"greet"`, dropping a trailing `Job` and underscoring), so you
can build the hash from your classes:
Every subclass **registers itself automatically** under its `job_type`
(`"GreetJob"` -> `"greet"`, dropping a trailing `Job` and underscoring). There is
no registry to build or pass around — the worker and web UI read this global
registry by default.

```ruby
REGISTRY = [GreetJob].to_h { |klass| [klass.job_type, klass] }
# => { "greet" => GreetJob }
```

**3. Enqueue.** The queue owns its own SQLite file:
**2. Enqueue.** The queue owns its own SQLite file. Enqueue by passing a job
**class** (with positional `args:`) or a job **instance** (carrying its keyword
arguments):

```ruby
queue = Benchwarmer::Queue.new(db_path: "db/benchwarmer.db")
queue.enqueue("greet", args: ["world"])

queue.enqueue(GreetJob.new(name: "World")) # kwargs -> perform(name:)
queue.enqueue(GreetJob.new(name: "World", language: "fr"))
queue.enqueue(PositionalJob, args: ["a", 1]) # array -> perform(*args)

# enqueue also takes: delay: (seconds), run_at:, parent_id:, after:,
# max_attempts:, scheduled_job_id:
```

**4. Run a worker.** `Worker#run` blocks, polling for runnable jobs and running
Instance kwargs are serialized as a JSON **object** and delivered back as keyword
arguments; an `args:` array is serialized as a JSON **array** and splatted as
positional arguments. Args must be JSON-serializable — anything else raises an
`ArgumentError` at enqueue time, not later inside the worker. (You can also
enqueue by type **string** — `queue.enqueue("greet", args: {name: "World"})` —
which is how the web form submits; the class/instance forms are preferred in
code.)

**3. Run a worker.** `Worker#run` blocks, polling for runnable jobs and running
them one at a time. It traps `INT`/`TERM` to shut down cleanly (the in-flight job
always finishes first).
always finishes first). No `registry:` needed — it defaults to the global
auto-registry.

```ruby
Benchwarmer::Worker.new(
db_path: "db/benchwarmer.db",
registry: REGISTRY,
poll_interval: 5 # seconds between polls
).run
```

**5. Mount the web UI.** The web app is a separate entry point,
`require "benchwarmer/web"`, mounted from a `config.ru`. Config is injected via
Sinatra `set`:
**4. Mount the web UI.** The web app is a separate entry point,
`require "benchwarmer/web"`, mounted from a `config.ru`. Only the queue factory
is injected; the enqueue form's job list comes from the auto-registry:

```ruby
# config.ru
require "benchwarmer/web"
require_relative "jobs" # your job classes + REGISTRY
require_relative "jobs" # your job classes (auto-registered on load)

Benchwarmer::Web.set :queue_factory, -> { Benchwarmer::Queue.new(db_path: "db/benchwarmer.db") }
Benchwarmer::Web.set :registry, REGISTRY

run Benchwarmer::Web
```
Expand Down Expand Up @@ -179,12 +189,13 @@ Why it's built this way:

The runner is injected through the job's constructor. The worker builds jobs with
its `job_factory:` (default `->(klass) { klass.new }`), which is the seam for
passing real collaborators in production and fakes in tests:
passing real collaborators in production and fakes in tests. Collaborators are
dependency-injected here; a job's serialized params (the kwargs it was enqueued
with) are separate data, delivered straight to `#perform`:

```ruby
Benchwarmer::Worker.new(
db_path: "db/benchwarmer.db",
registry: REGISTRY,
job_factory: ->(klass) { klass.new(runner: RealRunner.new) }
)
```
Expand All @@ -193,13 +204,14 @@ Benchwarmer::Worker.new(

A parent job enqueues children; pass `after:` a sibling's id to order a child
after that sibling **succeeds**. Inside a job, `enqueue_child` sets `parent_id`
for you and returns the child's id:
for you and returns the child's id. It takes the same targets as `enqueue` — a
job class, a job instance, or a type string:

```ruby
class NightlyJob < Benchwarmer::Job
def perform
sync_id = enqueue_child("sync")
enqueue_child("build", after: sync_id) # runs only after sync succeeds
sync_id = enqueue_child(SyncJob)
enqueue_child(BuildJob, after: sync_id) # runs only after sync succeeds
end
end
```
Expand All @@ -224,10 +236,14 @@ name updates it in place). The worker's scheduler tick enqueues due schedules an
recomputes the next run:

```ruby
queue.schedule(name: "nightly", type: "nightly", cron: "30 2 * * *")
queue.schedule(name: "nightly", type: NightlyJob, cron: "30 2 * * *") # or type: "nightly"
queue.unschedule(name: "nightly")
```

`type:` accepts a job class (stored as its `job_type`) or the type string
directly, and `args:` may be a JSON array (positional) or object (named), just
like `enqueue`.

The built-in parser is a 5-field cron (`minute hour day-of-month month
day-of-week`) supporting:

Expand Down
28 changes: 20 additions & 8 deletions examples/app.rb
Original file line number Diff line number Diff line change
Expand Up @@ -75,14 +75,27 @@ def perform
end

# The pipeline parent (type "nightly"): enqueue sync, then build ordered after
# sync succeeds. The parent succeeds as soon as the children are enqueued; its
# detail page shows the batch. Each step is its own row with its own retry.
# sync succeeds. Children are enqueued by class -- no registry to consult. The
# parent succeeds as soon as the children are enqueued; its detail page shows
# the batch. Each step is its own row with its own retry.
class NightlyJob < Benchwarmer::Job
def self.description = "Nightly pipeline: sync, then build"

def perform
sync_id = enqueue_child("sync")
enqueue_child("build", after: sync_id)
sync_id = enqueue_child(SyncJob)
enqueue_child(BuildJob, after: sync_id)
end
end

# A job with named parameters (issue #26): enqueue an instance carrying kwargs
# (GreetJob.new(name: "World")) or post {"name": "World"} into the UI enqueue
# form -- either way they arrive as keyword arguments to #perform.
class GreetJob < Benchwarmer::Job
def self.description = "Say hello (demonstrates named parameters)"

def perform(name:, language: "en")
greeting = (language == "fr") ? "Bonjour" : "Hello"
append_output("#{greeting}, #{name}!\n")
end
end

Expand All @@ -95,7 +108,6 @@ def perform
end
end

# The registry maps each job's type key to its class. Job.job_type derives the
# key from the class name ("SyncJob" -> "sync", "NightlyJob" -> "nightly").
REGISTRY = [NightlyJob, SyncJob, BuildJob, FlakyJob]
.to_h { |klass| [klass.job_type, klass] }
# No registry to build: every Benchwarmer::Job subclass registers itself under
# its job_type as soon as it's defined, and the worker + web UI read that global
# registry by default.
6 changes: 4 additions & 2 deletions examples/bin/worker
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,12 @@
require_relative "../app"

queue = Benchwarmer::Queue.new(db_path: DB_PATH)
queue.schedule(name: "nightly", type: "nightly", cron: "30 2 * * *")
# schedule accepts a job class for type: (stored as its job_type string).
queue.schedule(name: "nightly", type: NightlyJob, cron: "30 2 * * *")

# No registry: passed -- the worker defaults to the global auto-registry that
# every Benchwarmer::Job subclass populates when it's defined.
Benchwarmer::Worker.new(
db_path: DB_PATH,
registry: REGISTRY,
poll_interval: 1
).run
3 changes: 2 additions & 1 deletion examples/config.ru
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,8 @@
require_relative "app"
require "benchwarmer/web"

# No :registry to set -- the web UI defaults to the global auto-registry, so the
# enqueue form lists every Benchwarmer::Job subclass defined in app.rb.
Benchwarmer::Web.set :queue_factory, -> { Benchwarmer::Queue.new(db_path: DB_PATH) }
Benchwarmer::Web.set :registry, REGISTRY

run Benchwarmer::Web
2 changes: 1 addition & 1 deletion lib/benchwarmer/components/enqueue_form.rb
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ def render_type_field

def render_args_field
div(class: "mb-2") do
label(for: "enqueue-args", class: "form-label") { "Args (JSON array)" }
label(for: "enqueue-args", class: "form-label") { "Args (JSON array or object)" }
textarea(name: "args", id: "enqueue-args", class: "form-control", rows: 3,
data_test_id: "enqueue-args") { @args }
end
Expand Down
48 changes: 41 additions & 7 deletions lib/benchwarmer/job.rb
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,28 @@ module Benchwarmer
# row (via #bind) immediately before #perform is called, so #append_output can
# stream captured output back into the row through the queue.
class Job
# Auto-registry: every named subclass registers itself here (keyed by its
# job_type) the moment it is defined, so consuming apps never build a
# registry by hand -- Worker and Web default to this hash. It stays a
# mutable seam: tests (and apps that want isolation) can still inject an
# explicit registry to Queue/Worker/Web, which overrides the global.
REGISTRY = {}

class << self
# The shared auto-registry (see REGISTRY). Defined as a method so it reads
# the single Benchwarmer::Job constant even when called on a subclass.
def registry = REGISTRY

# Register each named subclass under its job_type. Anonymous classes
# (Class.new(Job), used in some tests) have no name yet and are skipped --
# they can always be run through an explicitly injected registry.
def inherited(subclass)
super
REGISTRY[subclass.job_type] = subclass if subclass.name
end

# Registry key derived from the class name: "SyncJob" -> "sync"
# (drop a trailing "Job", then underscore). Consuming apps build their
# registry Hash from this.
# (drop a trailing "Job", then underscore).
def job_type
base = name.split("::").last.sub(/Job\z/, "")
base.gsub(/([A-Z]+)([A-Z][a-z])/, '\1_\2')
Expand All @@ -28,22 +46,38 @@ def description = ""

attr_reader :queue, :job_row

# Named parameters captured at construction (e.g. GreetJob.new(name: "World"))
# become the job's payload: Queue#enqueue serializes them into the args
# column as a JSON object, and the worker delivers them back as keyword
# arguments to #perform. Subclasses that override #initialize for dependency
# injection (and don't call super) simply carry no params -- collaborators
# are DI, params are data.
def initialize(**params)
@params = params
end

# The captured named parameters (see #initialize). Always a Hash, even when
# a subclass's own #initialize skipped super.
def params = @params || {}

# Called by the worker to wire dependencies before #perform. Returns self.
def bind(queue:, job_row:)
@queue = queue
@job_row = job_row
self
end

def perform(*args)
def perform(*, **)
raise NotImplementedError, "#{self.class} must implement #perform"
end

# Enqueue a child job whose parent_id points back at this job's row, so the
# batch is queryable (Queue#children). Pass after: a sibling's id to order
# this child after that sibling succeeds. Returns the child job id.
def enqueue_child(type, args: [], after: nil)
queue.enqueue(type, args: args, parent_id: job_row["id"], after: after)
# batch is queryable (Queue#children). The first argument has the same
# flexibility as Queue#enqueue: a Job class, a Job instance (carrying its
# own kwargs), or a type string. Pass after: a sibling's id to order this
# child after that sibling succeeds. Returns the child job id.
def enqueue_child(job, args: [], after: nil)
queue.enqueue(job, args: args, parent_id: job_row["id"], after: after)
end

# Stream text into this job's captured output, keeping only the trailing
Expand Down
59 changes: 50 additions & 9 deletions lib/benchwarmer/queue.rb
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
require "time"
require_relative "db"
require_relative "cron"
require_relative "job"

module Benchwarmer
# The Queue owns ALL SQL. It can be constructed either with a db_path: (opens
Expand All @@ -22,12 +23,21 @@ def initialize(db_path: nil, db: nil, clock: -> { Time.now.utc }, registry: nil)

@db = db || DB.connect(db_path: db_path)
@clock = clock
@registry = registry || {}
# Default to the global auto-registry so callers never wire one by hand;
# an explicit registry: still overrides it (tests inject isolated ones).
@registry = registry || Job.registry
end

# Enqueue a job. Returns the new job id.
def enqueue(type, args: [], delay: nil, run_at: nil, parent_id: nil, after: nil,
# Enqueue a job. The first argument may be:
# * a Job subclass -- Queue.enqueue(GreetJob, args: ["world"])
# * a Job instance -- Queue.enqueue(GreetJob.new(name: "World"))
# (its constructor kwargs become the payload, serialized as a JSON object)
# * a type string -- Queue.enqueue("greet", args: ["world"])
# (the web form posts strings; args: may be a JSON array or object)
# Returns the new job id.
def enqueue(job, args: [], delay: nil, run_at: nil, parent_id: nil, after: nil,
max_attempts: nil, scheduled_job_id: nil)
type, payload, klass = coerce_job(job, args)
now = now_iso
run_at_iso =
if run_at
Expand All @@ -38,11 +48,11 @@ def enqueue(type, args: [], delay: nil, run_at: nil, parent_id: nil, after: nil,
now
end

# When max_attempts: isn't given, fall back to the registered job class's
# max_attempts (so a class's self.max_attempts is honored at enqueue time),
# then to the jobs-table default of 1 when the type isn't in the registry.
max = max_attempts || @registry[type]&.max_attempts || 1
params = [type, JSON.generate(args), run_at_iso, parent_id, after,
# When max_attempts: isn't given, take it from the job class (available
# directly for a class/instance, else via the registry for a string type),
# falling back to the jobs-table default of 1.
max = max_attempts || (klass || @registry[type])&.max_attempts || 1
params = [type, json_args(payload), run_at_iso, parent_id, after,
max, scheduled_job_id, now]
@db.execute(<<~SQL, params)
INSERT INTO jobs (type, args, status, run_at, parent_id, after_job_id,
Expand Down Expand Up @@ -72,8 +82,9 @@ def retry(job_id)
# (re)computes next_run_at from the clock's current time. Returns the
# scheduled_jobs row id.
def schedule(name:, type:, cron:, args: [])
type = type.job_type if type.is_a?(Class)
next_run_at = Cron.new(cron).next_after(@clock.call)
params = [name, type, JSON.generate(args), cron, to_iso(next_run_at), now_iso]
params = [name, type, json_args(args), cron, to_iso(next_run_at), now_iso]
rows = @db.execute(<<~SQL, params)
INSERT INTO scheduled_jobs (name, type, args, cron, next_run_at, created_at)
VALUES (?, ?, ?, ?, ?, ?)
Expand Down Expand Up @@ -152,6 +163,36 @@ def append_output(id, text)

private

# Normalize the enqueue target into [type_string, payload, klass]. payload is
# the value serialized into the args column: a positional array (from args:)
# for a class/string, or the instance's captured kwargs (a Hash) for an
# instance. klass is the resolved job class (nil for a string type not in the
# registry), used only to source max_attempts.
def coerce_job(job, args)
case job
when Class
[job.job_type, args, job]
when Job
[job.class.job_type, job.params, job.class]
when String
[job, args, @registry[job]]
else
raise ArgumentError,
"enqueue expects a Job class, a Job instance, or a type string, got #{job.class}"
end
end

# Serialize an args payload (Array or Hash) to JSON, failing fast with a
# clear ArgumentError if it contains anything that isn't JSON-serializable --
# better here at enqueue time than as a surprise inside the worker. strict:
# true rejects arbitrary Ruby objects (Symbols are still allowed and become
# strings, so keyword args survive the round trip).
def json_args(payload)
JSON.generate(payload, strict: true)
rescue JSON::GeneratorError => e
raise ArgumentError, "job args must be JSON-serializable (#{payload.inspect}): #{e.message}"
end

def now_iso
@clock.call.utc.iso8601
end
Expand Down
Loading
Loading