PubSub messages and readiness

The library starts a Phoenix.PubSub server named MishkaInstaller.PubSub with its application. It sends every message to the topic "mishka:plugin:" <> channel, where the channel is "event", "installer" or "mnesia". Subscribe with Phoenix.PubSub.subscribe/2. This process logs each plugin that is stopped:

defmodule MyApp.PluginWatcher do
  use GenServer
  require Logger

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)

  @impl true
  def init(_opts) do
    :ok = Phoenix.PubSub.subscribe(MishkaInstaller.PubSub, "mishka:plugin:event")
    {:ok, %{}}
  end

  @impl true
  def handle_info(%{channel: "event", status: :stop, data: %{name: name}}, state) do
    Logger.info("plugin stopped: #{inspect(name)}")
    {:noreply, state}
  end

  def handle_info(_message, state), do: {:noreply, state}
end

A message is a plain map with three keys: :status, an atom; :channel, the channel string; and :data, whose shape depends on the status. The tables below give the data of each one.

MishkaInstaller.subscribe/1 and MishkaInstaller.unsubscribe/1 do the same for you, but both are marked @doc false. They are not in the library's HexDocs, and the library does not promise them. Phoenix.PubSub.subscribe/2 belongs to phoenix_pubsub, a dependency of the library, and its documentation covers it. The function that sends the messages, MishkaInstaller.broadcast/4, is @doc false as well.

The topic is mishka:plugin:event. The plugin registry, the plugin processes and the event handler send these messages. Every running plugin process subscribes to this channel too, and it reads its stored status again when it receives :start, :restart, :enable, :stop, :unregister or :re_event.

A call that names a whole event, such as Event.start(:event, "name"), and the start of every event on boot send only the empty-map message, with no record. Installer.async_install/1 also sends :start with an empty map on this channel when it queues the install.

A plugin that starts under a supervisor sends these five messages, usually in this order:

%{status: :register, channel: "event", data: %MishkaInstaller.Event.Event{status: :registered}}
%{status: :start, channel: "event", data: %{}}
%{status: :start, channel: "event", data: %MishkaInstaller.Event.Event{status: :started}}
%{status: :purge_create, channel: "event", data: "event_c"}
%{status: :cluster_recompile, channel: "event", data: %{event: "event_c", origin: :nonode@nohost}}

The request with an empty map comes first. With queue set to true, the event handler builds the module in its own process, so :purge_create and :cluster_recompile can arrive before or after the record message. A restart that changes nothing sends no :purge_create. For the record fields, read the event reference documentation.

The topic is mishka:plugin:installer. Neither message is sent when the call returns an error. The replay of a stored library on boot sends nothing here and emits [:mishka_installer, :installer, :replay] telemetry instead.

The record fields are in the installer reference documentation.

The topic is mishka:plugin:mnesia. These messages move the boot from the store to the install handler, and then to the event handler.

The [:mishka_installer, :mnesia, :synchronized] and [:mishka_installer, :mnesia, :cluster, :nodeup] telemetry events go with these messages. The telemetry documentation lists them, and the deployment documentation covers the cluster.

Three signals say that the library started every event. Check one before you read the registry in your own boot, or before you rely on the plugins of an event.

The event handler sets the flag first. Then it logs, sends the message and emits the telemetry event, so the flag is already set when the other two arrive. The install handler sets the :compile_status flag to "ready" after it loads the stored libraries, and the event handler starts the events only after that, so :event_status also means the stored libraries are loaded. A plugin process reads both flags before it registers.

The flag exists because the other two signals may be over before you listen. Your application starts after the library's, and the library finishes on its own schedule. Mishka CMS attaches a telemetry handler and then reads the flag, so it reacts whether the sync comes after the attach or came before it:

@handler_id "my-app-ready"
@event [:mishka_installer, :event, :synchronized]

def attach(on_ready) do
  :telemetry.detach(@handler_id)
  :ok = :telemetry.attach(@handler_id, @event, &__MODULE__.handle_telemetry/4, on_ready)

  if :persistent_term.get(:event_status, nil) == "ready", do: on_ready.()
  :ok
end

def handle_telemetry(@event, _measurements, _metadata, on_ready) do
  on_ready.()
  :ok
end

To wait inside a process instead, subscribe first and read the flag second. A message sent between the two steps then reaches you:

def await_ready(timeout) do
  topic = "mishka:plugin:mnesia"
  :ok = Phoenix.PubSub.subscribe(MishkaInstaller.PubSub, topic)

  result =
    if :persistent_term.get(:event_status, nil) == "ready" do
      :ready
    else
      receive do
        %{status: :event_status, channel: "mnesia", data: "ready"} -> :ready
      after
        timeout -> :timeout
      end
    end

  Phoenix.PubSub.unsubscribe(MishkaInstaller.PubSub, topic)
  result
end

The event handler starts the events each time it receives :synchronized from the store, when :compile_status is already "ready", and each time it receives :compile_synchronized. Every start unregisters the plugins whose code is gone, starts every event, sets the flag again, and sends :event_status and the telemetry event again. The install handler also loads every stored library again on each :synchronized.

So the ready signal comes once on a normal boot, and again when the store announces itself again:

• The supervisor restarts the process of the Mnesia store.
• A node that booted :pending, because its seed was not reachable, joins the cluster when the seed connects.
• Another node of the cluster does either of the two. Its :synchronized message reaches this node, and the handlers do not check where it came from.
• On a boot, the event handler can start the events for both messages, if the install handler finishes before the event handler reads the store's message.

On a repeat the stored libraries are already running, so the replay of each one answers an error. The install handler logs [mishka_installer.installer] skipped re-loading and emits [:mishka_installer, :installer, :replay] with result: :error. The libraries keep running. When the event handler cannot read the event table, it sets no flag and sends no message or telemetry event.

A handler you attach must be safe to run again. The reconcile of Mishka CMS is: it moves only the registrations that sit under the wrong event, so a second run finds nothing to move.

• Telemetry and log reference: the events and the log lines.
• How events and plugins work: what happens on boot.
• Test plugins and events: wait for these signals in a test.
• Deploy to production: cluster and split-brain handling.