From 5a51cab50be9c80a9ffad989a492e84a6c76757e Mon Sep 17 00:00:00 2001 From: Parker Selbert Date: Mon, 19 Aug 2024 20:39:29 -0500 Subject: [PATCH] Link queue supervisor and midwife for restarts When a producer crashes it brings the queue's supervisor down with it. With enough database errors, the producer may crash repeatedly enough to exhaust restarts and bring down the DynamicSupervisor in charge of all queues. Now the supervisor is linked to the midwife to ensure that the midwife restarts as well, and it restarts all of the queues. --- lib/oban.ex | 15 +++++++-------- lib/oban/nursery.ex | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 8 deletions(-) create mode 100644 lib/oban/nursery.ex diff --git a/lib/oban.ex b/lib/oban.ex index 4cf51ac5..a7c14178 100644 --- a/lib/oban.ex +++ b/lib/oban.ex @@ -26,7 +26,7 @@ defmodule Oban do use Supervisor alias Ecto.{Changeset, Multi} - alias Oban.{Config, Engine, Job, Midwife, Notifier, Peer, Registry, Sonar, Stager} + alias Oban.{Config, Engine, Job, Notifier, Nursery, Peer, Registry, Sonar, Stager} alias Oban.Queue.{Drainer, Producer} @typedoc """ @@ -465,14 +465,13 @@ defmodule Oban do def whereis(name), do: Registry.whereis(name) @impl Supervisor - def init(%Config{plugins: plugins} = conf) do + def init(%Config{name: name, plugins: plugins} = conf) do children = [ - {Notifier, conf: conf, name: Registry.via(conf.name, Notifier)}, - {DynamicSupervisor, name: Registry.via(conf.name, Foreman), strategy: :one_for_one}, - {Peer, conf: conf, name: Registry.via(conf.name, Peer)}, - {Sonar, conf: conf, name: Registry.via(conf.name, Sonar)}, - {Midwife, conf: conf, name: Registry.via(conf.name, Midwife)}, - {Stager, conf: conf, name: Registry.via(conf.name, Stager)} + {Notifier, conf: conf, name: Registry.via(name, Notifier)}, + {Nursery, conf: conf, name: Registry.via(name, Nursery)}, + {Peer, conf: conf, name: Registry.via(name, Peer)}, + {Sonar, conf: conf, name: Registry.via(name, Sonar)}, + {Stager, conf: conf, name: Registry.via(name, Stager)} ] children = children ++ Enum.map(plugins, &plugin_child_spec(&1, conf)) diff --git a/lib/oban/nursery.ex b/lib/oban/nursery.ex new file mode 100644 index 00000000..e623f036 --- /dev/null +++ b/lib/oban/nursery.ex @@ -0,0 +1,33 @@ +defmodule Oban.Nursery do + @moduledoc false + + use Supervisor + + alias Oban.{Config, Midwife, Registry} + + @type opts :: [conf: Config.t(), name: GenServer.name()] + + @spec start_link(opts()) :: Supervisor.on_start() + def start_link(opts) when is_list(opts) do + Supervisor.start_link(__MODULE__, opts, name: opts[:name]) + end + + @spec child_spec(opts()) :: Supervisor.child_spec() + def child_spec(opts) do + name = Keyword.fetch!(opts, :name) + + %{super(opts) | id: name} + end + + @impl Supervisor + def init(opts) do + conf = Keyword.fetch!(opts, :conf) + + children = [ + {DynamicSupervisor, name: Registry.via(conf.name, Foreman)}, + {Midwife, conf: conf, name: Registry.via(conf.name, Midwife)} + ] + + Supervisor.init(children, max_restarts: 5, max_seconds: 30, strategy: :rest_for_one) + end +end