mbuhot

EctoJob - a transactional job queue built with Ecto, PostgreSQL and GenStage

EctoJob

A transactional job queue built with Ecto, PostgreSQL and GenStage

Available on Hex.pm: ecto_job | Hex
Docs: API Reference — ecto_job v3.1.0
Github: GitHub - mbuhot/ecto_job: Transactional job queue with Ecto, PostgreSQL and GenStage · GitHub

Goals

Transactional job processing
Retries
Scheduled jobs
Multiple queues
Low latency concurrent processing
Avoid frequent database polling
Library of functions, not a full OTP application

Details

One of the distinguishing features of ecto_job is that the API encourages transactional job processing through Ecto.Multi. Job handlers are passed an Ecto.Multi parameter that must be passed to Repo.transaction to complete the job.

def perform(multi = %Ecto.Multi{}, job = %{}) do
  multi
  |> do_first_thing(job["customer_id"])
  |> do_second_thing(job["product_id"])
  |> MyApp.Repo.transaction()
end

Similarly, there are helpers to encourage transactional job creation by adding to an Ecto.Multi

Multi.new()
|> Multi.insert(:add_user, User.insert_changeset(%{name: "Joe", email: "joe@gmail.com"}))
|> JobQueue.enqueue(:email_job, %{"type" => "SendEmail", "address" => "joe@gmail.com", "body" => "Welcome!"})
|> MyApp.Repo.transaction()

Jobs are processed using a GenStage producer and ConsumerSupervisor that will execute jobs concurrently up to a configurable max_demand.

Postgrex.Notifications is used to trigger job processors as soon as the transaction that adds a job is committed.

Job workers can be run on a separate node, without requiring any beam clustering, just PostgreSQL listen/notify.

Job completion is also signalled using listen/notify enabling websocket or chunked HTTP responses (on another node) to be triggered once a job completes.

Similar libraries:

22 comments

#library #ectojob #job-queue

20 5654 22

2024-08-20 07:31:59 UTC

First Post!

outlog

awesome - look forward to battle testing it.

Post #1

Most Liked

mbuhot

EctoJob version 2.0.0 has been published to Hex.pm.

EctoJob 2.0 depends on Ecto 3.0, but is otherwise backwards compatible with existing code.

Other changes in this release:

Timestamp options on job tables can be customized.
Job tables can be declared with custom schema_prefix.
EctoJob will always use a :bigserial primary key instead of relying on the ecto default.

Hex Package
Docs
Source

Post #8

mbuhot

EctoJob version 2.1.0 has been released.

Version 2.1 adds support for requeing jobs, fixes to the job reservation algorithm and dialyzer warnings.

Changelog
Hex Package
Docs

Thank you to everyone that contributed to this version!

I’d also like to invite any users of EctoJob to please raise issues on the GitHub page for any features you’d like to see added or would be willing to work on (admin UI, Telemetry, Pre/Post job hooks, etc..)

Post #13

mbuhot

EctoJob version 3.1 has been released

Changelog
Hex Package
Docs

Version 3.1.0 adds support for MySQL 8 and storing job params as an Elixir/Erlang term.
You can insert any arbitrary Elixir/Erlang term into the queue:

{"SendEmail", "jonas@gmail.com", "Welcome!"} 
|> MyApp.JobQueue.new() 
|> MyApp.Repo.insert()

You should use the option :params_type when defining your queue module:

defmodule MyJobQueue do 
  use EctoJob.JobQueue, table_name: "jobs", params_type: :binary # ... 
end

Possible values of the option are: :map (default) and :binary (for storing Elixir/Erlang terms).

You should use the same option when setting up the migration:

@ecto_job_version 3 
def up do 
  EctoJob.Migrations.Install.up() 
  EctoJob.Migrations.up("jobs", version: @ecto_job_version, params_type: :binary) 
end

We are looking forward to releasing a new major version soon with support for completed job retention and job idempotency. Stay tuned