v2.7.1

Fixed

• [Oban.Telemetry] Restore the :job key to [:oban, :job, :start] events. It was mistakenly removed during a refactoring.

v2.7.0

Pluggable Notifier

The PubSub functionality Oban relies on for starting, stopping, scaling, and pausing queues is now pluggable. The previous notifier was hard-coded to use Postgres for PubSub via LISTEN/NOTIFY. Unfortunately, LISTEN/NOTIFY doesn't work for PG Bouncer in "Transaction Mode." While relatively few users run in "Transaction Mode," it is increasingly common and deserved a work-around.

Oban.Notifier defines a minimal behaviour and Oban ships with a default implementation, Oban.PostgresNotifier, based on the old LISTEN/NOTIFY module. You can specify a different notifier in your Oban configuration:

config :my_app, Oban,
  notifier: MyApp.CustomNotifier,
  repo: MyApp.Repo,
  ...

An alternate pg/pg2 based notifier is available in Oban Pro.

Replacing Opts on Unique Conflict

The replace_args option for updating args on unique conflict is replaced with a highly flexible replace option. Using replace, you can update any job field when there is a conflict. Replace works for any of the following fields: args, max_attempts, meta, priority, queue, scheduled_at, tags, worker.

For example, to change the priority to 0 and increase max_attempts to 5 when there is a conflict:

BusinessWorker.new(
  args,
  max_attempts: 5,
  priority: 0,
  replace: [:max_attempts, :priority]
)

Another example is bumping the scheduled time to 1 second in the future on conflict. Either scheduled_at or schedule_in values will work, but the replace option is always scheduled_at.

UrgentWorker.new(args, schedule_in: 1, replace: [:scheduled_at])

Added

• [Oban.Job] A new conflict? field indicates whether a unique job conflicted with an existing job on insert.

• [Oban.Telemetry] Always populate the unsaved_error field for error or discard events. Previously, the field was empty for discard events.

• [Oban.Queue.Engine] Define a cancel_job/2 callback in the engine and move cancellation from the query module to the BasicEngine.

Changed

• [Oban.Testing] Thanks to a slight refactoring, the perform_job/3 helper now emits the same telemetry events that you'd have in production. The refactoring also exposed and fixed an inconsistency around invalid snooze handlers.

• [Oban.Testing] Expose Testing.all_enqueued/0, an alternate clause that doesn't require any options. This new clause makes it possible to check for any enqueued jobs without filters.

• [Oban.Plugins.Stager] Limit the number of jobs staged at one time to prevent timeout errors while staging a massive number of jobs.

Fixed

• [Oban.Repo] Respect ongoing transactions when using get_dynamic_repo. Prior to this fix, calls that use Oban.Repo, such as Oban.insert/2, could use a different repo than the one used in the current transaction.

v2.6.1

Fixes

• [Oban.Drainer] Always use the BasicEngine for draining, regardless of the
  currently configured engine.

v2.6.0

Pluggable Queue Engines

Queue producers now use an "engine" callback module to manage demand and work
with the database. The engine provides a clean way to expand and enhance the
functionality of queue producers without modifying the solid foundation of queue
supervision trees. Engines make enhanced functionality such as global
concurrency, local rate limiting and global rate limiting possible!

The BasicEngine is the default, and it retains the exact functionality of
previous Oban versions. To specify the engine explicitly, or swap in an
alternate engine, set it in your configuration:

config :my_app, Oban,
  engine: Oban.Queue.BasicEngine,
  ...

See the v2.6 upgrade guide for instructions on swapping in an alternate
engine.

Recursive Queue Draining

The ever-handy Oban.drain_queue/1,2 function gained a new with_recursion
option, which makes it possible to test jobs that insert more jobs when they
execute. When with_recursion is enabled the queue will keep executing until no
jobs are available. It even composes with with_scheduled!

In practice, this is especially useful for testing dependent workflows.

Gossip Plugin for Queue Monitoring

The new gossip plugin uses PubSub to efficiently exchange queue state
information between all interested nodes. This allows Oban instances to
broadcast state information regardless of which engine they are using, and
without storing anything in the database.

See the v2.6 upgrade guide for details on switching to the gossip
plugin.

Added

• [Oban.Job] Inject the current conf into the jobs struct as virtual field,
  making the complete conf available within perform/1.

• [Oban.Notifier] Add unlisten/2, used to stop a process from receiving
  notifications for one or more channels.

Changed

• [Oban.Telemetry] Stop emitting circuit events for queue producers. As
  producers no longer poll, the circuit breaker masked real errors more than it
  guarded against sporatic issues.

• [Oban.Telemetry] Delay [:oban, :supervisor, :init] event until the complete
  supervision tree finishes initialization.

• [Oban.Migration] Stop creating an oban_beats table entirely.

Fixed

• [Oban.Plugins.Cron] Prevent schedule overflow right before midnight

v2.5.0

Top of the Minute Cron

Rather than scheduling cron evaluation 60 seconds after the server starts, evaluation is now scheduled at the top of the next minute. This yields several improvements:

• More predictable timing for cron jobs now that they are inserted at the top of the minute. Note that jobs may execute later depending on how busy queues are.
• Minimize contention between servers inserting jobs, thanks to the transaction lock acquired by each plugin.
• Prevent duplicate inserts for jobs that omit the completed state (when server start time is staggered the transaction lock has no effect)

Repeater Plugin for Transaction Pooled Databases

Environments that can't make use of PG notifications, i.e. because they use PgBouncer with transaction pooling, won't process available jobs reliably. The new Repeater plugin provides a workaround that simulates polling functionality for producers.

Simply include the Repeater in your plugin list:

config :my_app, Oban,
  plugins: [
    Oban.Plugins.Pruner,
    Oban.Plugins.Stager,
    Oban.Plugins.Repeater
  ],
  ...

Include Perform Result in Job Telemetry Metadata

The metadata for [:oban, :job, :stop] telemetry events now include the job's perform/1 return value. That makes it possible to extract job output from other processes:

def handle_event([:oban, :job, :stop], _timing, meta, _opts) do
  IO.inspect(meta.result, label: "return from #{meta.job.worker}")

  :ok
end

Added

• [Oban] Support a changeset function in addition to a changeset in insert_all/4. Inserting jobs within a multi is even more powerful.

• [Oban.Queue.Executor] Stash a timed job's pid to enable inner process messaging, notably via telemetry events.

Changed

• [Oban] Upgrade minimum Elixir version from 1.8 to 1.9.

• [Oban.Plugins.Cron] Individually validate crontab workers and options, providing more descriptive errors at the collection and entry level.

• [Oban] Simplify the job cancelling flow for consistency. Due to race conditions it was always possible for a job to complete before it was cancelled, now that flow is much simpler.

• [Oban.Queue.Producer] Replace dispatch cooldown with a simpler debounce mechanism.

• [Oban.Plugins.Stager] Limit the number of notify events to one per queue, rather than one per available job.

Fixed

• [Oban.Queue.Producer] Handle unclean exits without an error and stack, which prevents "zombie" jobs. This would occur after multiple hot upgrades when the worker module changed, as the VM would forcibly terminate any processes running the old modules.

• [Oban] Specify that the changeset_wrapper type allows keys other than :changesets, fixing a dialyzer type mismatch.

Removed

• [Oban] Remove the undocumented version/0 function in favor of using the standard Application.spec/2