@mbuhot This is awesome. Thank you!
typicalpixel
danbruder
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.
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..)
6
santoshbt
Hi,
The library is very good and works fine for my delayed email sending functionality. But I am facing a problem, when I add the child worker in supervisor tree, like this
def start(_type, _args) do
# List all child processes to be supervised
children = [
# Start the Ecto repository
MyApp.Repo,
# Start the Endpoint (http/https)
MyAppWeb.Endpoint,
{MyApp.JobQueue, repo: MyApp.Repo, max_demand: 100},
{Task.Supervisor, [name: MyApp.TaskSupervisor]}
]
I am unable to run any tests. Getting the error like this
I am getting the error
** (DBConnection.OwnershipError) cannot find ownership process for pid<0.847.0>.
When using ownership, you must manage connections in one
of the four ways:
- By explicitly checking out a connection
- By explicitly allowing a spawned process
- By running the pool in shared mode
- By using :caller option with allowed process
The first two options require every new process to explicitly
check a connection out or be allowed by calling checkout or
allow respectively.
The third option requires a {:shared, pid} mode to be set.
If using shared mode in tests, make sure your tests are not
async.
The fourth option requires [caller: pid] to be used when
checking out a connection from the pool. The caller process
should already be allowed on a connection.
If you are reading this error, it means you have not done one
of the steps above or that the owner process has crashed.
See Ecto.Adapters.SQL.Sandbox docs for more information.
(ecto_sql 3.3.3) lib/ecto/adapters/sql.ex:609: Ecto.Adapters.SQL.raise_sql_call_error/1
(ecto_sql 3.3.3) lib/ecto/adapters/sql.ex:545: Ecto.Adapters.SQL.execute/5
(ecto 3.3.2) lib/ecto/repo/queryable.ex:192: Ecto.Repo.Queryable.execute/4
(ecto_job 3.0.0) lib/ecto_job/producer.ex:184: EctoJob.Producer.dispatch_jobs/2
(gen_stage 0.14.3) lib/gen_stage.ex:2103: GenStage.noreply_callback/3
(stdlib 3.11.1) gen_server.erl:637: :gen_server.try_dispatch/4
(stdlib 3.11.1) gen_server.erl:711: :gen_server.handle_msg/6
(stdlib 3.11.1) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
Last message: {:DOWN, reference<0.1630123645.3685482498.192412>, :process, pid<0.847.0>, {%DBConnection.OwnershipError{message: “cannot find ownership process for pid<0.847.0>.\n\nWhen using ownership, you must manage connections in one\nof the four ways:\n\n* By explicitly checking out a connection\n* By explicitly allowing a spawned process\n* By running the pool in shared mode\n* By using :caller option with allowed process\n\nThe first two options require every new process to explicitly\ncheck a connection out or be allowed by calling checkout or\nallow respectively.\n\nThe third option requires a {:shared, pid} mode to be set.\nIf using shared mode in tests, make sure your tests are not\nasync.\n\nThe fourth option requires [caller: pid] to be used when\nchecking out a connection from the pool. The caller process\nshould already be allowed on a connection.\n\nIf you are reading this error, it means you have not done one\nof the steps above or that the owner process has crashed.\n\nSee Ecto.Adapters.SQL.Sandbox docs for more information.”}, [{Ecto.Adapters.SQL, :raise_sql_call_error, 1, [file: ‘lib/ecto/adapters/sql.ex’, line: 609]}, {Ecto.Adapters.SQL, :execute, 5, [file: ‘lib/ecto/adapters/sql.ex’, line: 545]}, {Ecto.Repo.Queryable, :execute, 4, [file: ‘lib/ecto/repo/queryable.ex’, line: 192]}, {EctoJob.Producer, :dispatch_jobs, 2, [file: ‘lib/ecto_job/producer.ex’, line: 184]}, {GenStage, :noreply_callback, 3, [file: ‘lib/gen_stage.ex’, line: 2103]}, {:gen_server, :try_dispatch, 4, [file: ‘gen_server.erl’, line: 637]}, {:gen_server, :handle_msg, 6, [file: ‘gen_server.erl’, line: 711]}, {:proc_lib, :init_p_do_apply, 3, [file: ‘proc_lib.erl’, line: 249]}]}}
I have tried many ways to resolve this, but unable to succeed.
Please help me.
0
mbuhot
Hi @santoshbt
Take a look at this github issue: pg_notify not working when testing · Issue #36 · mbuhot/ecto_job · GitHub
0
santoshbt
Hi,
It was accidentally jobs table is deleted from my local database. I am unable to re-run the migration.
It always gives the error
execute “CREATE FUNCTION fn_notify_inserted()\n RETURNS trigger AS $$\nDECLARE\nBEGIN\n PERFORM pg_notify(TG_TABLE_NAME, ‘’);\n RETURN NEW;\nEND;\n$$ LANGUAGE plpgsql\n”
** (Postgrex.Error) ERROR 42723 (duplicate_function) function “fn_notify_inserted” already exists with same argument types.
Please let me know how to handle it.
0
mbuhot
@santoshbt use the postgresql drop function command to remove the function before running the migration.
the CASCADE option will also remove any triggers that depend on that function.
1
mbuhot
EctoJob version 3.1 has been released 
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 
5
towhans
Is it expected that each queue creates a new DB connection independent of the connection pool maintained by Ecto? It is how I see EctoJob behave in our production and it seems very weird to me. So in effect there is pool_size + number_of_queues connections.
0
mbuhot
Yes a dedicated connection is required to listen for notifications of new jobs in each queue table. (at ecto_job/lib/ecto_job/supervisor.ex at v3.1.0 · mbuhot/ecto_job · GitHub)
The Ecto APIs like Repo.checkout only allows a connection to be checked out for the duration of a function call, so EctoJob requires an additional connection.
it seems very weird to me
How so?
0