From 3e589ce4b1eb16734c292abbe9784a784f8b80ff Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 12:14:43 -0400
Subject: [PATCH 01/10] Upgrade pg_registry from 0.2.2 to 0.4

PgRegistry 0.3+ replaced the Erlang :pg backend with a self-contained
Elixir port. This migrates stagehand off the old API:

- Replace :pg scope with a PgRegistry :stagehand scope
- Use PgRegistry.register/3 in init instead of :via tuples for cluster
  group membership (producers and cron)
- Use local Registry names for producer addressing (consumer subscription)
- Replace :pg.monitor with PgRegistry.monitor/2 for join/leave events
- Replace get_members/2 with lookup/2 (returns [{pid, value}])
- Update tests for new API shapes
---
 lib/stagehand/plugins/cron.ex        | 11 ++++++-----
 lib/stagehand/queue/pipeline.ex      | 10 +++++-----
 lib/stagehand/queue/producer.ex      | 18 ++++++++++--------
 lib/stagehand/supervisor.ex          |  8 ++++++++
 mix.exs                              |  2 +-
 mix.lock                             |  2 +-
 test/stagehand/plugins/cron_test.exs | 10 ++++++----
 test/test_helper.exs                 |  2 +-
 8 files changed, 38 insertions(+), 25 deletions(-)

diff --git a/lib/stagehand/plugins/cron.ex b/lib/stagehand/plugins/cron.ex
index eec84ca..0b29dcc 100644
--- a/lib/stagehand/plugins/cron.ex
+++ b/lib/stagehand/plugins/cron.ex
@@ -2,7 +2,7 @@ defmodule Stagehand.Plugins.Cron do
   @moduledoc """
   Plugin for scheduling periodic/cron jobs.
 
-  Uses highlander pattern — only the leader node (first in sorted :pg members)
+  Uses highlander pattern — only the leader node (first in sorted PgRegistry members)
   fires cron jobs, preventing duplicate scheduling across the cluster.
 
   ## Usage
@@ -32,14 +32,13 @@ defmodule Stagehand.Plugins.Cron do
   defstruct [:conf, :crontab, :timer_ref]
 
   def start_link(opts) do
-    conf = opts[:conf]
-    name = {:via, PgRegistry, {:pg, {:stagehand_cron, conf.name}}}
-    GenServer.start_link(__MODULE__, opts, name: name)
+    GenServer.start_link(__MODULE__, opts)
   end
 
   @impl true
   def init(opts) do
     conf = opts[:conf]
+    {:ok, _} = PgRegistry.register(:stagehand, {:stagehand_cron, conf.name}, nil)
 
     crontab =
       opts
@@ -74,7 +73,9 @@ defmodule Stagehand.Plugins.Cron do
   # -- Private --
 
   defp leader?(%{conf: conf}) do
-    case PgRegistry.get_members(:pg, {:stagehand_cron, conf.name}) do
+    pids = for {pid, _} <- PgRegistry.lookup(:stagehand, {:stagehand_cron, conf.name}), do: pid
+
+    case pids do
       [] -> true
       members -> self() == Enum.min(members)
     end
diff --git a/lib/stagehand/queue/pipeline.ex b/lib/stagehand/queue/pipeline.ex
index 120777e..464c657 100644
--- a/lib/stagehand/queue/pipeline.ex
+++ b/lib/stagehand/queue/pipeline.ex
@@ -43,9 +43,9 @@ defmodule Stagehand.Queue.Pipeline do
         type: :supervisor,
         shutdown: shutdown_grace
       },
-      # Producer starts second — {:via, PgRegistry, ...} joins the pg
-      # group automatically. On shutdown (reverse order) it stops first,
-      # leaves the group in terminate, then drains executing jobs.
+      # Producer starts second and registers with PgRegistry in init/1.
+      # On shutdown (reverse order) it stops first, leaves the group in
+      # terminate, then drains executing jobs.
       %{
         id: Producer,
         start:
@@ -71,7 +71,7 @@ defmodule Stagehand.Queue.Pipeline do
   """
   @spec producer_name(atom(), atom() | binary()) :: {:via, module(), term()}
   def producer_name(stagehand_name, queue) do
-    {:via, PgRegistry, {:pg, {:stagehand, stagehand_name, :producers, to_string(queue)}}}
+    {:via, Registry, {Module.concat(stagehand_name, Registry), {:producer, to_string(queue)}}}
   end
 
   @doc """
@@ -79,7 +79,7 @@ defmodule Stagehand.Queue.Pipeline do
   """
   @spec producers_for_queue(atom(), atom() | binary()) :: [pid()]
   def producers_for_queue(stagehand_name, queue) do
-    PgRegistry.get_members(:pg, {:stagehand, stagehand_name, :producers, to_string(queue)})
+    for {pid, _} <- PgRegistry.lookup(:stagehand, {:stagehand, stagehand_name, :producers, to_string(queue)}), do: pid
   end
 
   @doc """
diff --git a/lib/stagehand/queue/producer.ex b/lib/stagehand/queue/producer.ex
index 90f7b12..7c37144 100644
--- a/lib/stagehand/queue/producer.ex
+++ b/lib/stagehand/queue/producer.ex
@@ -100,8 +100,9 @@ defmodule Stagehand.Queue.Producer do
     conf = opts[:conf]
 
     pg_group = {:stagehand, conf.name, :producers, queue}
-    existing = :pg.get_members(:pg, pg_group)
-    :pg.monitor(:pg, pg_group)
+    {:ok, _} = PgRegistry.register(:stagehand, pg_group, nil)
+    {_ref, existing} = PgRegistry.monitor(:stagehand, pg_group)
+    existing = for {pid, _} <- existing, do: pid
 
     # If there are existing producers on other nodes, tell Unique to block
     # check_and_insert until all of them have synced their entries to us.
@@ -239,14 +240,15 @@ defmodule Stagehand.Queue.Producer do
     {:stop, :normal, state}
   end
 
-  def handle_info({_ref, :join, _group, pids}, state) do
+  def handle_info({_ref, :join, _group, entries}, state) do
     pg_key = {:stagehand, state.conf.name, :producers, state.queue}
-    sync_unique_entries(state, PgRegistry.get_members(:pg, pg_key))
+    members = for {pid, _} <- PgRegistry.lookup(:stagehand, pg_key), do: pid
+    sync_unique_entries(state, members)
 
     # Signal sync complete to new producers' Unique servers
     unique_name = Module.concat(state.conf.name, Stagehand.Unique)
 
-    for pid <- pids, node(pid) != node() do
+    for {pid, _} <- entries, node(pid) != node() do
       try do
         Stagehand.Unique.sync_complete({unique_name, node(pid)})
       catch
@@ -257,7 +259,7 @@ defmodule Stagehand.Queue.Producer do
     {:noreply, [], state}
   end
 
-  def handle_info({_ref, :leave, _group, _pids}, state) do
+  def handle_info({_ref, :leave, _group, _entries}, state) do
     {:noreply, [], state}
   end
 
@@ -277,10 +279,10 @@ defmodule Stagehand.Queue.Producer do
     pg_key = {:stagehand, state.conf.name, :producers, state.queue}
 
     # Snapshot membership while we're still in the group (needed for hash ring)
-    all_producers = PgRegistry.get_members(:pg, pg_key)
+    all_producers = for {pid, _} <- PgRegistry.lookup(:stagehand, pg_key), do: pid
 
     # Leave so no new jobs are routed to us
-    PgRegistry.unregister_name({:pg, pg_key})
+    PgRegistry.unregister(:stagehand, pg_key)
 
     # Drain any in-flight enqueue messages that arrived before we left
     state = drain_mailbox(state)
diff --git a/lib/stagehand/supervisor.ex b/lib/stagehand/supervisor.ex
index 494dfba..a4c2997 100644
--- a/lib/stagehand/supervisor.ex
+++ b/lib/stagehand/supervisor.ex
@@ -17,6 +17,14 @@ defmodule Stagehand.Supervisor do
     registry_name = Module.concat(conf.name, Registry)
     unique_name = Module.concat(conf.name, Stagehand.Unique)
 
+    # Ensure the shared PgRegistry scope is running. If it's already
+    # started (by another Stagehand instance or the application), this
+    # is a no-op.
+    case PgRegistry.start_link(:stagehand) do
+      {:ok, _} -> :ok
+      {:error, {:already_started, _}} -> :ok
+    end
+
     children =
       [
         {Registry, keys: :unique, name: registry_name},
diff --git a/mix.exs b/mix.exs
index 3ffd3c4..d657970 100644
--- a/mix.exs
+++ b/mix.exs
@@ -41,7 +41,7 @@ defmodule Stagehand.MixProject do
       {:crontab, "~> 1.1"},
       {:gen_stage, "~> 1.2"},
       {:libring, "~> 1.7"},
-      {:pg_registry, "~> 0.2.2"},
+      {:pg_registry, "~> 0.4"},
       {:telemetry, "~> 1.0"},
       {:credo, "~> 1.7", only: [:dev, :test], runtime: false},
       {:styler, "~> 1.11", only: [:dev, :test], runtime: false},
diff --git a/mix.lock b/mix.lock
index 9e151e7..33b95e4 100644
--- a/mix.lock
+++ b/mix.lock
@@ -8,7 +8,7 @@
   "gen_stage": {:hex, :gen_stage, "1.3.2", "7c77e5d1e97de2c6c2f78f306f463bca64bf2f4c3cdd606affc0100b89743b7b", [:mix], [], "hexpm", "0ffae547fa777b3ed889a6b9e1e64566217413d018cabd825f786e843ffe63e7"},
   "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"},
   "libring": {:hex, :libring, "1.7.0", "4f245d2f1476cd7ed8f03740f6431acba815401e40299208c7f5c640e1883bda", [:mix], [], "hexpm", "070e3593cb572e04f2c8470dd0c119bc1817a7a0a7f88229f43cf0345268ec42"},
-  "pg_registry": {:hex, :pg_registry, "0.2.2", "1dd46a90dd53d9af47ace311c767b2d2ebe621b9f77a7eea072283d8ca713a91", [:mix], [], "hexpm", "22b83dd2e9ffe970d3fbb380ab56cb98cf88a486798a9ecbe6cc6392fa1fc5cd"},
+  "pg_registry": {:hex, :pg_registry, "0.4.0", "edb4de796e2a0224e69d29edb3883c7a2be90a026261fcda2f1d0c1e19189df3", [:mix], [], "hexpm", "2a34ebdce4ce9ea43995aeda7981b0b954165d444f6aa5ce1784e7c5b1708274"},
   "styler": {:hex, :styler, "1.11.0", "35010d970689a23c2bcc8e97bd8bf7d20e3561d60c49be84654df5c37d051a9c", [:mix], [], "hexpm", "70f36165d0cf238a32b7a456fdef6a9c72e77e657d7ac4a0ace33aeba3f2b8c0"},
   "telemetry": {:hex, :telemetry, "1.4.1", "ab6de178e2b29b58e8256b92b382ea3f590a47152ca3651ea857a6cae05ac423", [:rebar3], [], "hexpm", "2172e05a27531d3d31dd9782841065c50dd5c3c7699d95266b2edd54c2dafa1c"},
 }
diff --git a/test/stagehand/plugins/cron_test.exs b/test/stagehand/plugins/cron_test.exs
index 9121960..fd13a78 100644
--- a/test/stagehand/plugins/cron_test.exs
+++ b/test/stagehand/plugins/cron_test.exs
@@ -48,7 +48,10 @@ defmodule Stagehand.Plugins.CronTest do
   end
 
   defp cron_pid(name) do
-    PgRegistry.whereis_name({:pg, {:stagehand_cron, name}})
+    case PgRegistry.lookup(:stagehand, {:stagehand_cron, name}) do
+      [{pid, _}] -> pid
+      [] -> nil
+    end
   end
 
   defp tick(pid) do
@@ -132,10 +135,9 @@ defmodule Stagehand.Plugins.CronTest do
           crontab: [{"* * * * *", EveryMinuteWorker}]
         )
 
-      :pg.join(:pg, {:stagehand_cron, name}, cron2)
-
-      members = PgRegistry.get_members(:pg, {:stagehand_cron, name})
+      members = for {pid, _} <- PgRegistry.lookup(:stagehand, {:stagehand_cron, name}), do: pid
       assert length(members) == 2
+      assert cron2 in members
 
       leader = Enum.min(members)
       non_leader = Enum.max(members)
diff --git a/test/test_helper.exs b/test/test_helper.exs
index ec7c2e0..acdc6cf 100644
--- a/test/test_helper.exs
+++ b/test/test_helper.exs
@@ -1,2 +1,2 @@
-:pg.start(:pg)
+PgRegistry.Pg.start(:stagehand)
 ExUnit.start()

From 28d3268037ac504a6ae9f59d293a7f101045b424 Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 12:18:34 -0400
Subject: [PATCH 02/10] Rename PgRegistry scope from :stagehand to
 Stagehand.PgRegistry

Follow the Elixir convention of module-style atoms for named processes
(e.g. MyApp.Registry, MyApp.PubSub).
---
 lib/stagehand/plugins/cron.ex        |  4 ++--
 lib/stagehand/queue/pipeline.ex      |  2 +-
 lib/stagehand/queue/producer.ex      | 10 +++++-----
 lib/stagehand/supervisor.ex          |  2 +-
 test/stagehand/plugins/cron_test.exs |  4 ++--
 test/test_helper.exs                 |  2 +-
 6 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/lib/stagehand/plugins/cron.ex b/lib/stagehand/plugins/cron.ex
index 0b29dcc..185c4a8 100644
--- a/lib/stagehand/plugins/cron.ex
+++ b/lib/stagehand/plugins/cron.ex
@@ -38,7 +38,7 @@ defmodule Stagehand.Plugins.Cron do
   @impl true
   def init(opts) do
     conf = opts[:conf]
-    {:ok, _} = PgRegistry.register(:stagehand, {:stagehand_cron, conf.name}, nil)
+    {:ok, _} = PgRegistry.register(Stagehand.PgRegistry, {:stagehand_cron, conf.name}, nil)
 
     crontab =
       opts
@@ -73,7 +73,7 @@ defmodule Stagehand.Plugins.Cron do
   # -- Private --
 
   defp leader?(%{conf: conf}) do
-    pids = for {pid, _} <- PgRegistry.lookup(:stagehand, {:stagehand_cron, conf.name}), do: pid
+    pids = for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, {:stagehand_cron, conf.name}), do: pid
 
     case pids do
       [] -> true
diff --git a/lib/stagehand/queue/pipeline.ex b/lib/stagehand/queue/pipeline.ex
index 464c657..4943f92 100644
--- a/lib/stagehand/queue/pipeline.ex
+++ b/lib/stagehand/queue/pipeline.ex
@@ -79,7 +79,7 @@ defmodule Stagehand.Queue.Pipeline do
   """
   @spec producers_for_queue(atom(), atom() | binary()) :: [pid()]
   def producers_for_queue(stagehand_name, queue) do
-    for {pid, _} <- PgRegistry.lookup(:stagehand, {:stagehand, stagehand_name, :producers, to_string(queue)}), do: pid
+    for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, {:stagehand, stagehand_name, :producers, to_string(queue)}), do: pid
   end
 
   @doc """
diff --git a/lib/stagehand/queue/producer.ex b/lib/stagehand/queue/producer.ex
index 7c37144..95f464a 100644
--- a/lib/stagehand/queue/producer.ex
+++ b/lib/stagehand/queue/producer.ex
@@ -100,8 +100,8 @@ defmodule Stagehand.Queue.Producer do
     conf = opts[:conf]
 
     pg_group = {:stagehand, conf.name, :producers, queue}
-    {:ok, _} = PgRegistry.register(:stagehand, pg_group, nil)
-    {_ref, existing} = PgRegistry.monitor(:stagehand, pg_group)
+    {:ok, _} = PgRegistry.register(Stagehand.PgRegistry, pg_group, nil)
+    {_ref, existing} = PgRegistry.monitor(Stagehand.PgRegistry, pg_group)
     existing = for {pid, _} <- existing, do: pid
 
     # If there are existing producers on other nodes, tell Unique to block
@@ -242,7 +242,7 @@ defmodule Stagehand.Queue.Producer do
 
   def handle_info({_ref, :join, _group, entries}, state) do
     pg_key = {:stagehand, state.conf.name, :producers, state.queue}
-    members = for {pid, _} <- PgRegistry.lookup(:stagehand, pg_key), do: pid
+    members = for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, pg_key), do: pid
     sync_unique_entries(state, members)
 
     # Signal sync complete to new producers' Unique servers
@@ -279,10 +279,10 @@ defmodule Stagehand.Queue.Producer do
     pg_key = {:stagehand, state.conf.name, :producers, state.queue}
 
     # Snapshot membership while we're still in the group (needed for hash ring)
-    all_producers = for {pid, _} <- PgRegistry.lookup(:stagehand, pg_key), do: pid
+    all_producers = for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, pg_key), do: pid
 
     # Leave so no new jobs are routed to us
-    PgRegistry.unregister(:stagehand, pg_key)
+    PgRegistry.unregister(Stagehand.PgRegistry, pg_key)
 
     # Drain any in-flight enqueue messages that arrived before we left
     state = drain_mailbox(state)
diff --git a/lib/stagehand/supervisor.ex b/lib/stagehand/supervisor.ex
index a4c2997..14c7552 100644
--- a/lib/stagehand/supervisor.ex
+++ b/lib/stagehand/supervisor.ex
@@ -20,7 +20,7 @@ defmodule Stagehand.Supervisor do
     # Ensure the shared PgRegistry scope is running. If it's already
     # started (by another Stagehand instance or the application), this
     # is a no-op.
-    case PgRegistry.start_link(:stagehand) do
+    case PgRegistry.start_link(Stagehand.PgRegistry) do
       {:ok, _} -> :ok
       {:error, {:already_started, _}} -> :ok
     end
diff --git a/test/stagehand/plugins/cron_test.exs b/test/stagehand/plugins/cron_test.exs
index fd13a78..a37b0fe 100644
--- a/test/stagehand/plugins/cron_test.exs
+++ b/test/stagehand/plugins/cron_test.exs
@@ -48,7 +48,7 @@ defmodule Stagehand.Plugins.CronTest do
   end
 
   defp cron_pid(name) do
-    case PgRegistry.lookup(:stagehand, {:stagehand_cron, name}) do
+    case PgRegistry.lookup(Stagehand.PgRegistry, {:stagehand_cron, name}) do
       [{pid, _}] -> pid
       [] -> nil
     end
@@ -135,7 +135,7 @@ defmodule Stagehand.Plugins.CronTest do
           crontab: [{"* * * * *", EveryMinuteWorker}]
         )
 
-      members = for {pid, _} <- PgRegistry.lookup(:stagehand, {:stagehand_cron, name}), do: pid
+      members = for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, {:stagehand_cron, name}), do: pid
       assert length(members) == 2
       assert cron2 in members
 
diff --git a/test/test_helper.exs b/test/test_helper.exs
index acdc6cf..16dbbeb 100644
--- a/test/test_helper.exs
+++ b/test/test_helper.exs
@@ -1,2 +1,2 @@
-PgRegistry.Pg.start(:stagehand)
+PgRegistry.Pg.start(Stagehand.PgRegistry)
 ExUnit.start()

From 423d7bf2653411cd1b830f71b2790d31a84ae1b7 Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 12:25:34 -0400
Subject: [PATCH 03/10] Use Highlander for cron singleton, rename PgRegistry
 scope

- Replace manual leader election in Cron with Highlander, which
  ensures a single cron process runs cluster-wide via :global
- Remove cron's PgRegistry dependency entirely
- Register cron in the local Registry for addressability
- Rename PgRegistry scope from Stagehand.PgRegistry to
  Stagehand.ProducerRegistry since it is now only used for
  producer cluster discovery
---
 lib/stagehand/plugins/cron.ex        | 31 +++++++++--------
 lib/stagehand/queue/pipeline.ex      |  2 +-
 lib/stagehand/queue/producer.ex      | 10 +++---
 lib/stagehand/supervisor.ex          |  2 +-
 mix.exs                              |  1 +
 mix.lock                             |  1 +
 test/stagehand/plugins/cron_test.exs | 50 ++++++----------------------
 test/test_helper.exs                 |  2 +-
 8 files changed, 38 insertions(+), 61 deletions(-)

diff --git a/lib/stagehand/plugins/cron.ex b/lib/stagehand/plugins/cron.ex
index 185c4a8..2d39685 100644
--- a/lib/stagehand/plugins/cron.ex
+++ b/lib/stagehand/plugins/cron.ex
@@ -2,8 +2,8 @@ defmodule Stagehand.Plugins.Cron do
   @moduledoc """
   Plugin for scheduling periodic/cron jobs.
 
-  Uses highlander pattern — only the leader node (first in sorted PgRegistry members)
-  fires cron jobs, preventing duplicate scheduling across the cluster.
+  Uses `Highlander` to ensure a single cron process runs across the
+  cluster, preventing duplicate scheduling.
 
   ## Usage
 
@@ -31,14 +31,27 @@ defmodule Stagehand.Plugins.Cron do
 
   defstruct [:conf, :crontab, :timer_ref]
 
+  def child_spec(opts) do
+    conf = opts[:conf]
+
+    inner = %{
+      id: {__MODULE__, conf.name},
+      start: {__MODULE__, :start_link, [opts]}
+    }
+
+    Highlander.child_spec(inner)
+  end
+
   def start_link(opts) do
-    GenServer.start_link(__MODULE__, opts)
+    conf = opts[:conf]
+    name = {:via, Registry, {Module.concat(conf.name, Registry), :cron}}
+    GenServer.start_link(__MODULE__, opts, name: name)
   end
 
   @impl true
   def init(opts) do
+    Process.flag(:trap_exit, true)
     conf = opts[:conf]
-    {:ok, _} = PgRegistry.register(Stagehand.PgRegistry, {:stagehand_cron, conf.name}, nil)
 
     crontab =
       opts
@@ -59,7 +72,6 @@ defmodule Stagehand.Plugins.Cron do
     now = NaiveDateTime.utc_now()
 
     for {cron, worker, opts} <- state.crontab,
-        leader?(state),
         Crontab.DateChecker.matches_date?(cron, now) do
       job = worker.new(%{}, Keyword.put(opts, :meta, %{"cron" => true}))
       Stagehand.insert(state.conf.name, job)
@@ -72,15 +84,6 @@ defmodule Stagehand.Plugins.Cron do
 
   # -- Private --
 
-  defp leader?(%{conf: conf}) do
-    pids = for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, {:stagehand_cron, conf.name}), do: pid
-
-    case pids do
-      [] -> true
-      members -> self() == Enum.min(members)
-    end
-  end
-
   defp parse_entry({expression, worker}), do: parse_entry({expression, worker, []})
 
   defp parse_entry({expression, worker, opts}) do
diff --git a/lib/stagehand/queue/pipeline.ex b/lib/stagehand/queue/pipeline.ex
index 4943f92..21b8f19 100644
--- a/lib/stagehand/queue/pipeline.ex
+++ b/lib/stagehand/queue/pipeline.ex
@@ -79,7 +79,7 @@ defmodule Stagehand.Queue.Pipeline do
   """
   @spec producers_for_queue(atom(), atom() | binary()) :: [pid()]
   def producers_for_queue(stagehand_name, queue) do
-    for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, {:stagehand, stagehand_name, :producers, to_string(queue)}), do: pid
+    for {pid, _} <- PgRegistry.lookup(Stagehand.ProducerRegistry, {:stagehand, stagehand_name, :producers, to_string(queue)}), do: pid
   end
 
   @doc """
diff --git a/lib/stagehand/queue/producer.ex b/lib/stagehand/queue/producer.ex
index 95f464a..97ba9af 100644
--- a/lib/stagehand/queue/producer.ex
+++ b/lib/stagehand/queue/producer.ex
@@ -100,8 +100,8 @@ defmodule Stagehand.Queue.Producer do
     conf = opts[:conf]
 
     pg_group = {:stagehand, conf.name, :producers, queue}
-    {:ok, _} = PgRegistry.register(Stagehand.PgRegistry, pg_group, nil)
-    {_ref, existing} = PgRegistry.monitor(Stagehand.PgRegistry, pg_group)
+    {:ok, _} = PgRegistry.register(Stagehand.ProducerRegistry, pg_group, nil)
+    {_ref, existing} = PgRegistry.monitor(Stagehand.ProducerRegistry, pg_group)
     existing = for {pid, _} <- existing, do: pid
 
     # If there are existing producers on other nodes, tell Unique to block
@@ -242,7 +242,7 @@ defmodule Stagehand.Queue.Producer do
 
   def handle_info({_ref, :join, _group, entries}, state) do
     pg_key = {:stagehand, state.conf.name, :producers, state.queue}
-    members = for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, pg_key), do: pid
+    members = for {pid, _} <- PgRegistry.lookup(Stagehand.ProducerRegistry, pg_key), do: pid
     sync_unique_entries(state, members)
 
     # Signal sync complete to new producers' Unique servers
@@ -279,10 +279,10 @@ defmodule Stagehand.Queue.Producer do
     pg_key = {:stagehand, state.conf.name, :producers, state.queue}
 
     # Snapshot membership while we're still in the group (needed for hash ring)
-    all_producers = for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, pg_key), do: pid
+    all_producers = for {pid, _} <- PgRegistry.lookup(Stagehand.ProducerRegistry, pg_key), do: pid
 
     # Leave so no new jobs are routed to us
-    PgRegistry.unregister(Stagehand.PgRegistry, pg_key)
+    PgRegistry.unregister(Stagehand.ProducerRegistry, pg_key)
 
     # Drain any in-flight enqueue messages that arrived before we left
     state = drain_mailbox(state)
diff --git a/lib/stagehand/supervisor.ex b/lib/stagehand/supervisor.ex
index 14c7552..ccbb021 100644
--- a/lib/stagehand/supervisor.ex
+++ b/lib/stagehand/supervisor.ex
@@ -20,7 +20,7 @@ defmodule Stagehand.Supervisor do
     # Ensure the shared PgRegistry scope is running. If it's already
     # started (by another Stagehand instance or the application), this
     # is a no-op.
-    case PgRegistry.start_link(Stagehand.PgRegistry) do
+    case PgRegistry.start_link(Stagehand.ProducerRegistry) do
       {:ok, _} -> :ok
       {:error, {:already_started, _}} -> :ok
     end
diff --git a/mix.exs b/mix.exs
index d657970..92618c9 100644
--- a/mix.exs
+++ b/mix.exs
@@ -41,6 +41,7 @@ defmodule Stagehand.MixProject do
       {:crontab, "~> 1.1"},
       {:gen_stage, "~> 1.2"},
       {:libring, "~> 1.7"},
+      {:highlander, "~> 0.2"},
       {:pg_registry, "~> 0.4"},
       {:telemetry, "~> 1.0"},
       {:credo, "~> 1.7", only: [:dev, :test], runtime: false},
diff --git a/mix.lock b/mix.lock
index 33b95e4..ca2ea86 100644
--- a/mix.lock
+++ b/mix.lock
@@ -6,6 +6,7 @@
   "erlex": {:hex, :erlex, "0.2.8", "cd8116f20f3c0afe376d1e8d1f0ae2452337729f68be016ea544a72f767d9c12", [:mix], [], "hexpm", "9d66ff9fedf69e49dc3fd12831e12a8a37b76f8651dd21cd45fcf5561a8a7590"},
   "file_system": {:hex, :file_system, "1.1.1", "31864f4685b0148f25bd3fbef2b1228457c0c89024ad67f7a81a3ffbc0bbad3a", [:mix], [], "hexpm", "7a15ff97dfe526aeefb090a7a9d3d03aa907e100e262a0f8f7746b78f8f87a5d"},
   "gen_stage": {:hex, :gen_stage, "1.3.2", "7c77e5d1e97de2c6c2f78f306f463bca64bf2f4c3cdd606affc0100b89743b7b", [:mix], [], "hexpm", "0ffae547fa777b3ed889a6b9e1e64566217413d018cabd825f786e843ffe63e7"},
+  "highlander": {:hex, :highlander, "0.2.1", "e59b459f857e89daf73f2598bf2b2c0479a435481e6101ea389fd3625919b052", [:mix], [], "hexpm", "5ba19a18358803d82a923511acec8ee85fac30731c5ca056f2f934bc3d3afd9a"},
   "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"},
   "libring": {:hex, :libring, "1.7.0", "4f245d2f1476cd7ed8f03740f6431acba815401e40299208c7f5c640e1883bda", [:mix], [], "hexpm", "070e3593cb572e04f2c8470dd0c119bc1817a7a0a7f88229f43cf0345268ec42"},
   "pg_registry": {:hex, :pg_registry, "0.4.0", "edb4de796e2a0224e69d29edb3883c7a2be90a026261fcda2f1d0c1e19189df3", [:mix], [], "hexpm", "2a34ebdce4ce9ea43995aeda7981b0b954165d444f6aa5ce1784e7c5b1708274"},
diff --git a/test/stagehand/plugins/cron_test.exs b/test/stagehand/plugins/cron_test.exs
index a37b0fe..8003451 100644
--- a/test/stagehand/plugins/cron_test.exs
+++ b/test/stagehand/plugins/cron_test.exs
@@ -1,5 +1,5 @@
 defmodule Stagehand.Plugins.CronTest do
-  use ExUnit.Case, async: true
+  use ExUnit.Case, async: false
 
   alias Stagehand.Plugins.Cron
 
@@ -48,10 +48,8 @@ defmodule Stagehand.Plugins.CronTest do
   end
 
   defp cron_pid(name) do
-    case PgRegistry.lookup(Stagehand.PgRegistry, {:stagehand_cron, name}) do
-      [{pid, _}] -> pid
-      [] -> nil
-    end
+    [{pid, _}] = Registry.lookup(Module.concat(name, Registry), :cron)
+    pid
   end
 
   defp tick(pid) do
@@ -64,7 +62,6 @@ defmodule Stagehand.Plugins.CronTest do
     test "inserts jobs matching the current minute" do
       name = start_stagehand(crontab: [{"* * * * *", EveryMinuteWorker}])
 
-      # Trigger a tick manually instead of waiting for the real timer
       tick(cron_pid(name))
 
       jobs = Stagehand.Testing.all_enqueued(name, worker: EveryMinuteWorker)
@@ -105,8 +102,6 @@ defmodule Stagehand.Plugins.CronTest do
   describe "aliases" do
     test "@daily alias is parsed" do
       name = start_stagehand(crontab: [{"@daily", HourlyWorker}])
-
-      # If parsing failed, start_supervised would have crashed
       assert is_pid(cron_pid(name))
     end
 
@@ -116,39 +111,16 @@ defmodule Stagehand.Plugins.CronTest do
     end
   end
 
-  describe "leader election" do
-    test "only the leader fires jobs" do
+  describe "singleton" do
+    test "cron is registered globally via Highlander" do
       name = start_stagehand(crontab: [{"* * * * *", EveryMinuteWorker}])
+      pid = cron_pid(name)
 
-      # Start a second cron process and manually join the same pg group.
-      # Can't use {:via, PgRegistry, ...} since the name is already taken.
-      conf = %Stagehand.Config{
-        name: name,
-        queues: [default: 5],
-        shutdown_grace_period: @timeout,
-        testing: :manual
-      }
-
-      {:ok, cron2} =
-        GenServer.start_link(Cron,
-          conf: conf,
-          crontab: [{"* * * * *", EveryMinuteWorker}]
-        )
-
-      members = for {pid, _} <- PgRegistry.lookup(Stagehand.PgRegistry, {:stagehand_cron, name}), do: pid
-      assert length(members) == 2
-      assert cron2 in members
-
-      leader = Enum.min(members)
-      non_leader = Enum.max(members)
-
-      # Only the leader should insert
-      tick(leader)
-      assert length(Stagehand.Testing.all_enqueued(name, worker: EveryMinuteWorker)) == 1
-
-      # The non-leader should not insert
-      tick(non_leader)
-      assert length(Stagehand.Testing.all_enqueued(name, worker: EveryMinuteWorker)) == 1
+      # Highlander registers under {Highlander, child_spec.id}
+      highlander_pid = :global.whereis_name({Highlander, {Cron, name}})
+      assert is_pid(highlander_pid)
+      assert is_pid(pid)
+      assert pid != highlander_pid
     end
   end
 
diff --git a/test/test_helper.exs b/test/test_helper.exs
index 16dbbeb..0df4f24 100644
--- a/test/test_helper.exs
+++ b/test/test_helper.exs
@@ -1,2 +1,2 @@
-PgRegistry.Pg.start(Stagehand.PgRegistry)
+PgRegistry.Pg.start(Stagehand.ProducerRegistry)
 ExUnit.start()

From 4feac7854e5f1b825bdcd9f47cfc2f0221b1a35d Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 12:30:16 -0400
Subject: [PATCH 04/10] Start ProducerRegistry in application supervision tree

Move PgRegistry scope startup from Stagehand.Supervisor (with
idempotent case/match) to a proper Application module. The scope
is now started once when the :stagehand application boots.
---
 lib/stagehand/application.ex | 14 ++++++++++++++
 lib/stagehand/supervisor.ex  |  8 --------
 mix.exs                      |  3 ++-
 test/test_helper.exs         |  1 -
 4 files changed, 16 insertions(+), 10 deletions(-)
 create mode 100644 lib/stagehand/application.ex

diff --git a/lib/stagehand/application.ex b/lib/stagehand/application.ex
new file mode 100644
index 0000000..084310e
--- /dev/null
+++ b/lib/stagehand/application.ex
@@ -0,0 +1,14 @@
+defmodule Stagehand.Application do
+  @moduledoc false
+
+  use Application
+
+  @impl true
+  def start(_type, _args) do
+    children = [
+      {PgRegistry, Stagehand.ProducerRegistry}
+    ]
+
+    Supervisor.start_link(children, strategy: :one_for_one, name: Stagehand.ApplicationSupervisor)
+  end
+end
diff --git a/lib/stagehand/supervisor.ex b/lib/stagehand/supervisor.ex
index ccbb021..494dfba 100644
--- a/lib/stagehand/supervisor.ex
+++ b/lib/stagehand/supervisor.ex
@@ -17,14 +17,6 @@ defmodule Stagehand.Supervisor do
     registry_name = Module.concat(conf.name, Registry)
     unique_name = Module.concat(conf.name, Stagehand.Unique)
 
-    # Ensure the shared PgRegistry scope is running. If it's already
-    # started (by another Stagehand instance or the application), this
-    # is a no-op.
-    case PgRegistry.start_link(Stagehand.ProducerRegistry) do
-      {:ok, _} -> :ok
-      {:error, {:already_started, _}} -> :ok
-    end
-
     children =
       [
         {Registry, keys: :unique, name: registry_name},
diff --git a/mix.exs b/mix.exs
index 92618c9..9769471 100644
--- a/mix.exs
+++ b/mix.exs
@@ -22,7 +22,8 @@ defmodule Stagehand.MixProject do
 
   def application do
     [
-      extra_applications: [:logger]
+      extra_applications: [:logger],
+      mod: {Stagehand.Application, []}
     ]
   end
 
diff --git a/test/test_helper.exs b/test/test_helper.exs
index 0df4f24..869559e 100644
--- a/test/test_helper.exs
+++ b/test/test_helper.exs
@@ -1,2 +1 @@
-PgRegistry.Pg.start(Stagehand.ProducerRegistry)
 ExUnit.start()

From e74199466c865be503d03b65f85c0f1c4872aaae Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 12:41:54 -0400
Subject: [PATCH 05/10] Remove redundant local Registry name from producers

Producers are discovered cluster-wide via PgRegistry and addressed by
pid (consumer subscribes via self(), router uses producers_for_queue).
The local Registry name was unused. Producer still accepts an optional
name: for direct unit tests.
---
 lib/stagehand/queue/pipeline.ex | 12 +-----------
 lib/stagehand/queue/producer.ex |  4 ++--
 2 files changed, 3 insertions(+), 13 deletions(-)

diff --git a/lib/stagehand/queue/pipeline.ex b/lib/stagehand/queue/pipeline.ex
index 21b8f19..d430f48 100644
--- a/lib/stagehand/queue/pipeline.ex
+++ b/lib/stagehand/queue/pipeline.ex
@@ -23,7 +23,6 @@ defmodule Stagehand.Queue.Pipeline do
     conf = opts[:conf]
     limit = opts[:limit] || 10
 
-    producer_name = producer_name(conf.name, queue)
     consumer_name = consumer_name(conf.name, queue)
 
     shutdown_grace = conf.shutdown_grace_period + 1_000
@@ -52,7 +51,6 @@ defmodule Stagehand.Queue.Pipeline do
           {Producer, :start_link,
            [
              [
-               name: producer_name,
                queue: queue,
                conf: conf,
                consumer: consumer_name,
@@ -67,15 +65,7 @@ defmodule Stagehand.Queue.Pipeline do
   end
 
   @doc """
-  Local producer name (for consumer subscription on the same node).
-  """
-  @spec producer_name(atom(), atom() | binary()) :: {:via, module(), term()}
-  def producer_name(stagehand_name, queue) do
-    {:via, Registry, {Module.concat(stagehand_name, Registry), {:producer, to_string(queue)}}}
-  end
-
-  @doc """
-  Get all producer pids for a queue across the cluster.
+  Returns all producer pids for a queue across the cluster.
   """
   @spec producers_for_queue(atom(), atom() | binary()) :: [pid()]
   def producers_for_queue(stagehand_name, queue) do
diff --git a/lib/stagehand/queue/producer.ex b/lib/stagehand/queue/producer.ex
index 97ba9af..8d2f82d 100644
--- a/lib/stagehand/queue/producer.ex
+++ b/lib/stagehand/queue/producer.ex
@@ -19,8 +19,8 @@ defmodule Stagehand.Queue.Producer do
   ]
 
   def start_link(opts) do
-    name = opts[:name]
-    GenStage.start_link(__MODULE__, opts, name: name)
+    server_opts = if opts[:name], do: [name: opts[:name]], else: []
+    GenStage.start_link(__MODULE__, opts, server_opts)
   end
 
   @doc """

From 0d966f52ee0e872e50ffa69be8f9e46fe0d836bb Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 12:45:07 -0400
Subject: [PATCH 06/10] Remove local Registry name from cron process
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Cron doesn't need a local name — it's a Highlander singleton found
through :global. Test helper walks the Highlander supervision tree
to find the cron pid.
---
 lib/stagehand/plugins/cron.ex        | 4 +---
 test/stagehand/plugins/cron_test.exs | 4 +++-
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/lib/stagehand/plugins/cron.ex b/lib/stagehand/plugins/cron.ex
index 2d39685..6a0305a 100644
--- a/lib/stagehand/plugins/cron.ex
+++ b/lib/stagehand/plugins/cron.ex
@@ -43,9 +43,7 @@ defmodule Stagehand.Plugins.Cron do
   end
 
   def start_link(opts) do
-    conf = opts[:conf]
-    name = {:via, Registry, {Module.concat(conf.name, Registry), :cron}}
-    GenServer.start_link(__MODULE__, opts, name: name)
+    GenServer.start_link(__MODULE__, opts)
   end
 
   @impl true
diff --git a/test/stagehand/plugins/cron_test.exs b/test/stagehand/plugins/cron_test.exs
index 8003451..cdedac7 100644
--- a/test/stagehand/plugins/cron_test.exs
+++ b/test/stagehand/plugins/cron_test.exs
@@ -48,7 +48,9 @@ defmodule Stagehand.Plugins.CronTest do
   end
 
   defp cron_pid(name) do
-    [{pid, _}] = Registry.lookup(Module.concat(name, Registry), :cron)
+    highlander_pid = :global.whereis_name({Highlander, {Cron, name}})
+    %{pid: sup_pid} = :sys.get_state(highlander_pid)
+    [{_, pid, _, _}] = Supervisor.which_children(sup_pid)
     pid
   end
 

From 25f076e4c4b95d2a7fc6f23c40b472e8147415d8 Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 12:50:15 -0400
Subject: [PATCH 07/10] Replace libring with rendezvous hashing
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Rendezvous (HRW) hashing needs only the member list and a key —
no separate ring structure to build and maintain. The implementation
is a single Enum.max_by over :erlang.phash2({node, key}).

Removes the libring dependency.
---
 lib/stagehand/queue/producer.ex | 4 +---
 lib/stagehand/router.ex         | 8 ++++++--
 mix.exs                         | 1 -
 3 files changed, 7 insertions(+), 6 deletions(-)

diff --git a/lib/stagehand/queue/producer.ex b/lib/stagehand/queue/producer.ex
index 8d2f82d..cb7aa9c 100644
--- a/lib/stagehand/queue/producer.ex
+++ b/lib/stagehand/queue/producer.ex
@@ -323,11 +323,9 @@ defmodule Stagehand.Queue.Producer do
   defp do_sync_unique(_unique_name, _entries, producers) when length(producers) < 2, do: :ok
 
   defp do_sync_unique(unique_name, entries, producers) do
-    ring = Enum.reduce(producers, HashRing.new(), &HashRing.add_node(&2, &1))
-
     remote =
       for {fp, _job, _ts} = entry <- entries,
-          owner = HashRing.key_to_node(ring, fp),
+          owner = Stagehand.Router.rendezvous(producers, fp),
           owner != self(),
           node(owner) != node(),
           reduce: %{} do
diff --git a/lib/stagehand/router.ex b/lib/stagehand/router.ex
index 304527c..f0b8a9d 100644
--- a/lib/stagehand/router.ex
+++ b/lib/stagehand/router.ex
@@ -43,8 +43,7 @@ defmodule Stagehand.Router do
     unique_server = Module.concat(conf.name, Unique)
     fingerprint = Unique.fingerprint(job)
 
-    ring = Enum.reduce(producers, HashRing.new(), &HashRing.add_node(&2, &1))
-    producer_pid = HashRing.key_to_node(ring, fingerprint)
+    producer_pid = rendezvous(producers, fingerprint)
 
     case Unique.check_and_insert(unique_server, fingerprint, job) do
       {:ok, job} ->
@@ -55,6 +54,11 @@ defmodule Stagehand.Router do
     end
   end
 
+  @doc false
+  def rendezvous(nodes, key) do
+    Enum.max_by(nodes, fn node -> :erlang.phash2({node, key}) end)
+  end
+
   defp schedule_delay(%Job{scheduled_at: nil}), do: 0
 
   defp schedule_delay(%Job{scheduled_at: scheduled_at}) do
diff --git a/mix.exs b/mix.exs
index 9769471..c88f3f3 100644
--- a/mix.exs
+++ b/mix.exs
@@ -41,7 +41,6 @@ defmodule Stagehand.MixProject do
     [
       {:crontab, "~> 1.1"},
       {:gen_stage, "~> 1.2"},
-      {:libring, "~> 1.7"},
       {:highlander, "~> 0.2"},
       {:pg_registry, "~> 0.4"},
       {:telemetry, "~> 1.0"},

From 8d5069c2141d4d104ee8ab61fa53e6d3ff707e68 Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 12:53:01 -0400
Subject: [PATCH 08/10] Remove unnecessary trap_exit from cron init

---
 lib/stagehand/plugins/cron.ex | 1 -
 1 file changed, 1 deletion(-)

diff --git a/lib/stagehand/plugins/cron.ex b/lib/stagehand/plugins/cron.ex
index 6a0305a..89a1368 100644
--- a/lib/stagehand/plugins/cron.ex
+++ b/lib/stagehand/plugins/cron.ex
@@ -48,7 +48,6 @@ defmodule Stagehand.Plugins.Cron do
 
   @impl true
   def init(opts) do
-    Process.flag(:trap_exit, true)
     conf = opts[:conf]
 
     crontab =

From 943c3b1de4c7be28028d27eb8f74ef2019526a97 Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 13:03:38 -0400
Subject: [PATCH 09/10] Hash on node(pid) instead of pid in rendezvous function

A pid's internal representation differs between the local node
(where it shows as <0.N.C>) and remote nodes (where the first
element is a node-specific identifier). Hashing on node(pid)
ensures all nodes in the cluster compute the same winner for a
given key.
---
 lib/stagehand/router.ex | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/stagehand/router.ex b/lib/stagehand/router.ex
index f0b8a9d..40a66ef 100644
--- a/lib/stagehand/router.ex
+++ b/lib/stagehand/router.ex
@@ -55,8 +55,8 @@ defmodule Stagehand.Router do
   end
 
   @doc false
-  def rendezvous(nodes, key) do
-    Enum.max_by(nodes, fn node -> :erlang.phash2({node, key}) end)
+  def rendezvous(pids, key) do
+    Enum.max_by(pids, fn pid -> :erlang.phash2({node(pid), key}) end)
   end
 
   defp schedule_delay(%Job{scheduled_at: nil}), do: 0

From 31632b7aa67014f2e92dbbc5f3fd1ebe639de2e5 Mon Sep 17 00:00:00 2001
From: Tony Winn <hi@tonywinn.me>
Date: Fri, 10 Apr 2026 13:06:31 -0400
Subject: [PATCH 10/10] Format pipeline.ex

---
 lib/stagehand/queue/pipeline.ex | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/lib/stagehand/queue/pipeline.ex b/lib/stagehand/queue/pipeline.ex
index d430f48..019770b 100644
--- a/lib/stagehand/queue/pipeline.ex
+++ b/lib/stagehand/queue/pipeline.ex
@@ -69,7 +69,9 @@ defmodule Stagehand.Queue.Pipeline do
   """
   @spec producers_for_queue(atom(), atom() | binary()) :: [pid()]
   def producers_for_queue(stagehand_name, queue) do
-    for {pid, _} <- PgRegistry.lookup(Stagehand.ProducerRegistry, {:stagehand, stagehand_name, :producers, to_string(queue)}), do: pid
+    for {pid, _} <-
+          PgRegistry.lookup(Stagehand.ProducerRegistry, {:stagehand, stagehand_name, :producers, to_string(queue)}),
+        do: pid
   end
 
   @doc """