mule/registry

Local process storage for Mule instances — a port of Oban.Registry.

Elixir’s registry is a Registry of unique via-tuple keys; gleam_otp has no Registry/via-tuple equivalent and its static_supervisor exposes no child lookup, so the honest substitute is a named public ETS table per instance (FFI in mule_registry_ffi, the mule_cell_ffi crash-free style) plus a small owner actor. The owner creates the table, monitors every registrant, and deletes a registrant’s row on DOWN — deleting only when the row still holds the dead pid, so a restarted process that re-registered first is never wiped by the old registration’s DOWN. lookup additionally filters dead pids, closing the gap between a crash and the owner’s cleanup.

A persistent_term cell is deliberately NOT used here: registry rows are created and deleted at runtime (queue start/stop, crashes), need enumeration (check_all_queues), and need dead-entry cleanup — all things persistent_term cannot do without global-GC cost and read-modify-write races.

Handles are stored untyped in ETS; each role’s handle type is fixed by its registrant (see Role), and typed accessors over the generic lookup live in the modules that own the handle types (producer.subject, mule.config) — the same discipline as the peer/Sonar cells.

Types

Message

opaque </>

pub opaque type Message

Role

</>

The row key, mirroring Oban.Registry’s Oban.name() | {Oban.name(), role()} keys: the instance itself is Instance; everything else is a role. Each role’s handle type is fixed by its registrant:

Instance — handle: the assembled Config, pid: root supervisor
Notifier / Peer / Sonar / Stager / Foreman — handle: Nil
Producer(q) — handle: Subject(producer.Message)
Queue(q) — handle: Nil, pid: the queue’s supervisor
QueueForeman(q) — handle: foreman.Foreman (the task factory)
Watchman(q) — handle: Subject(watchman.Message)
Midwife — handle: Subject(midwife.Message)

pub type Role {
  Instance
  Foreman
  Midwife
  Notifier
  Peer
  Sonar
  Stager
  Queue(queue: String)
  QueueForeman(queue: String)
  Producer(queue: String)
  Watchman(queue: String)
  Owner
}

Constructors

```
Instance
```
```
Foreman
```
```
Midwife
```
```
Notifier
```
```
Peer
```
```
Sonar
```
```
Stager
```
```
Queue(queue: String)
```
```
QueueForeman(queue: String)
```
```
Producer(queue: String)
```
```
Watchman(queue: String)
```
```
Owner
```

Values

child

</>

pub fn child(
  instance_name instance_name: String,
  instance_handle instance_handle: handle,
) -> supervision.ChildSpecification(process.Subject(Message))

Child for the root supervisor, FIRST in the tree (so it survives notifier crashes under RestForOne; a registry crash restarts everything after it, which re-creates the table and re-registers every process). The start closure runs in the root supervisor’s process, so process.self() there IS the instance pid — captured and stored as the Instance row by the owner’s initialiser, which also creates the table. instance_handle is the assembled Config, held generically so this module imports nothing from mule/config.

lookup

</>

pub fn lookup(
  instance_name instance_name: String,
  role role: Role,
) -> Result(#(process.Pid, handle), Nil)

The generic lookup (Oban.Registry.lookup/2). Returns the row only when the registered pid is alive, closing the gap between a crash and the owner’s DOWN cleanup. The handle type is fixed per role by its registrant (see Role); annotating a different type is a caller bug — the same discipline as the peer/Sonar cells.

producers

</>

pub fn producers(
  instance_name instance_name: String,
) -> List(#(String, process.Pid, handle))

Every live producer row for the instance, as #(queue, pid, handle) — check_all_queues’s enumeration (mule.ex uses a Registry.select match spec; a tab2list + filter is equivalent at these table sizes).

register

</>

pub fn register(
  instance_name instance_name: String,
  role role: Role,
  handle handle: handle,
) -> Nil

Register the CALLING process under role (producers call this in their own initialisers, like via-tuple registration). Inserts the ETS row directly, then asks the owner to monitor the caller for DOWN cleanup. A total no-op when the instance’s table does not exist (a producer started outside mule.start, as in tests).

register_child

</>

pub fn register_child(
  child child: supervision.ChildSpecification(data),
  instance_name instance_name: String,
  role role: Role,
) -> supervision.ChildSpecification(data)

Wrap a child spec so the started pid is registered under role with a Nil handle — used on the notifier/peer/sonar/stager children for whereis parity.

register_pid

</>

pub fn register_pid(
  instance_name instance_name: String,
  role role: Role,
  pid pid: process.Pid,
  handle handle: handle,
) -> Nil

unregister

</>

pub fn unregister(
  instance_name instance_name: String,
  role role: Role,
) -> Nil

Delete a role’s row (the midwife’s stop_queue cleanup).

whereis

</>

pub fn whereis(
  instance_name instance_name: String,
  role role: Role,
) -> option.Option(process.Pid)

The pid of a live registered process (Oban.Registry.whereis/2).