DEV Community: krishnadaspc

Building Distributed Systems in Elixir: Part 6 — Named Processes

krishnadaspc — Wed, 19 Aug 2026 15:52:31 +0000

In the previous part of this series, we built a tiny supervisor from scratch.

When a worker crashed, the supervisor started a replacement. That replacement had a new PID:

old worker -> #PID<0.102.0>
new worker -> #PID<0.105.0>

This reveals an important limitation of sharing PIDs as a public interface.

A PID identifies one running incarnation of a process. It is excellent for sending a reply, setting up a monitor, or creating a link. It is not a stable address for a service that may stop and later be replaced.

In this part, we'll use named processes to give a worker a discoverable address:

:worker

We'll build three small examples using:

Process.register/2
Process.whereis/1
:global.register_name/2
:global.whereis_name/1
send/2

No GenServer.

No OTP Registry.

The goal is to understand the lookup problem that registries solve before reaching for those abstractions.

The PID-Sharing Problem

Suppose one process starts a worker and gives its PID to a client:

worker = spawn(fn -> worker_loop() end)
send(client, {:worker_started, worker})

The client can now send work directly:

send(worker, {:work, self(), "hello"})

This works while that particular worker process is alive.

But process IDs are temporary. If the worker exits, the PID is no longer a route to the service:

Client                         Worker

holds #PID<0.102.0>            #PID<0.102.0>
     |                               |
     |                               X exits
     |
     | send(#PID<0.102.0>, work)
     |------------------------------> no worker receives it

Sending to a dead local PID does not raise an error and does not restart a process. The message is simply not delivered to a living worker.

One answer is to tell every client about every new PID after a restart. That spreads lifecycle knowledge throughout the system.

Another answer is to make clients depend on a name and resolve that name when sending.

Registering a Local Name

Our first worker waits for a stop message:

defmodule Worker do
  def start do
    spawn(fn ->
      receive do
        :stop -> :ok
      end
    end)
  end
end

Starting it gives us a PID:

pid = Worker.start()

We can associate that PID with a local name:

Process.register(pid, :worker)

The registration belongs to the current BEAM node.

local process registry

:worker  ->  #PID<0.102.0>

The atom :worker is now a registered local name. We can retrieve its current PID with:

Process.whereis(:worker)

In the example, the returned PID is the same PID that we registered:

registered_pid = Process.whereis(:worker)

Process.whereis/1 returns nil if the name is not registered on this node.

Sending to the Name

There is often no need to perform a lookup yourself. send/2 accepts a local registered name:

send(:worker, :stop)

The runtime resolves the name and delivers the message to its current owner.

Caller                         Local Registry                  Worker
  |                                  |                           |
  | send(:worker, :stop)             |                           |
  |--------------------------------->| :worker -> worker PID     |
  |                                  |-------------------------->|
  |                                  |                           | receives :stop

This is the first benefit of a name: code that sends the message need not carry a PID around.

Run the local example:

elixir 01-process-registration.exs

Typical output is:

Worker PID: #PID<...>
Registered as :worker
Lookup using name: #PID<...>
Stopping using name: #PID<...>

The PID values vary from run to run. The name in the code does not.

Name Scope Matters

Process.register/2 is local to one node.

If two disconnected nodes both run:

Process.register(pid, :worker)

there is no conflict:

Node A: :worker -> #PID<...>
Node B: :worker -> #PID<...>

Each node has its own local process registry. Therefore:

Process.whereis(:worker)

answers only the question:

Which local process on this node is registered as :worker?

It does not discover a process on another BEAM node.

That is exactly right for many services. A local cache, a local connection manager, or a supervisor-owned worker usually needs only node-local lookup.

For cluster-wide lookup, the system needs a broader naming mechanism.

A Global Name

Erlang supplies the :global module for a namespace coordinated across connected nodes.

The second example registers the worker with:

:global.register_name(:worker, pid)

and looks it up with:

:global.whereis_name(:worker)

The example then sends a message to the PID returned by that lookup:

send(:global.whereis_name(:worker), :stop)

Run it with:

elixir 02-global-lookup.exs

The script itself runs on a single node, so its output looks much like the local case. The distinction becomes meaningful after nodes are named and connected.

Conceptually, a connected cluster can coordinate one name:

                       connected Erlang nodes

Node A                                                Node B
 Client                                               Worker
   |                                                    |
   | :global.whereis_name(:worker)                      |
   |--------------------> global name coordination -----|
   |<-------------------- worker PID -------------------|
   |                                                    |
   | send(worker_pid, message)                          |
   |--------------------------------------------------->|

:global (written as :global in code) provides a useful primitive, but global coordination is not free. It has failure behavior and operational trade-offs that matter in a real cluster.

Hiding the PID From Clients

The third example changes the startup API.

Instead of returning the spawned PID, the worker registers itself and returns its public name:

def start do
  pid =
    spawn(fn ->
      receive do
        {:work, from, request} ->
          send(from, {:reply, request, "processed: #{request}"})
      end
    end)

  Process.register(pid, :worker)
  :worker
end

The caller receives:

worker = Worker.start()
# :worker

It does not receive or retain the worker PID.

The client sends its request to the name:

send(worker_name, {:work, self(), request})

The message protocol is still explicit:

{:work, caller_pid, request}

The worker replies directly to the caller:

send(from, {:reply, request, "processed: #{request}"})

The name identifies the service endpoint. The caller PID identifies where that particular reply must go.

Client                                          :worker
  |                                                 |
  | {:work, client_pid, "hello"}                    |
  |------------------------------------------------>|
  |                                                 |
  | {:reply, "hello", "processed: hello"}          |
  |<------------------------------------------------|

Run it with:

elixir 03-avoiding-pid-sharing.exs

Expected output:

Worker available as: :worker
Response: processed: hello

Names Are Not Supervision

It is tempting to think that a name solves the worker-restart problem. It does not.

When a registered process terminates, its registration is removed:

:worker -> #PID<0.102.0>

worker exits

:worker -> no registered process

A replacement worker still must be started and registered:

supervisor starts new worker
         |
         v
new worker registers :worker
         |
         v
:worker -> #PID<0.105.0>

Names address discovery. Supervisors address lifecycle. They are complementary.

In an OTP application, a common pattern is to start a supervised process with a name, often using :via and Registry. The supervisor creates the current process; the registry lets callers find it without depending on its old PID.

One Name Means One Owner

A local registered name has one owner on its node. This is useful for a singleton service, but it also means the simple mechanism does not describe a worker pool.

For example, this is a single endpoint:

:worker -> one PID

It cannot represent three workers under the same local atom without another routing process or a richer registry design:

:workers -> router process -> worker_1, worker_2, worker_3

That distinction will matter when we build a worker pool later in the series.

Correlating Replies

The client waits for:

{:reply, ^request, response} -> response

Pinning request ensures it accepts a reply with the request value it sent.

For this small demonstration, the string itself is adequate. In a client that might have several requests in flight with the same value, use a unique reference instead:

ref = make_ref()
send(:worker, {:work, self(), ref, request})

receive do
  {:reply, ^ref, response} -> response
end

This combines named lookup from this lesson with the correlated request-reply technique from Part 2.

Limits of Global Registration

:global makes node-wide lookup possible, but it does not remove the hard parts of distributed systems.

A system still needs to decide what happens when:

nodes cannot communicate because of a network partition
a process dies and needs replacement
two nodes attempt to claim the same service role
clients have already looked up a PID that then exits

There is no universally correct answer. The appropriate naming and discovery strategy depends on whether the service must be local, replicated, partition-tolerant, or strongly coordinated.

For the small examples in this repository, the central idea is simpler:

Use a process name when callers should depend on a service role rather than on one temporary PID.

That small separation between identity and process incarnation is one of the building blocks of a resilient BEAM system.

Source code link: https://github.com/pckrishnadas88/elixir-distributed-systems-lab/tree/main/06-named-processes

Series

← Previous: Part 5 — Supervisor From Scratch

Next: Part 7 — Worker Pool (coming soon) →

Content License

This article is licensed under CC BY-NC 4.0. You may share and adapt it with attribution for non-commercial purposes. The accompanying source code remains licensed under the MIT License.

Building Distributed Systems in Elixir: Part 5 — Supervisor From Scratch

krishnadaspc — Tue, 18 Aug 2026 05:41:06 +0000

In the previous part of this series, we explored process links.

A linked worker that crashes sends an exit signal to the process linked to it. By default, that failure propagates and can terminate both processes.

We also saw that a process can trap exits:

Process.flag(:trap_exit, true)

Once exit trapping is enabled, an incoming exit signal becomes a mailbox message:

{:EXIT, pid, reason}

Receiving that message tells us that a linked process terminated, but detection is only the beginning.

What should happen next?

If the failed process was performing essential work, we may need to start a replacement.

In this part, we'll build a small supervisor from scratch using:

spawn/1
spawn_link/1
Process.flag(:trap_exit, true)
send/2
receive/1

Our supervisor will manage two workers. When one crashes, it will restart only that worker while leaving the healthy sibling unchanged.

No GenServer.

No OTP Supervisor.

The goal is not to replace OTP. The goal is to understand the mechanism and policy that an OTP supervisor provides.

The Problem

Imagine a long-running worker process:

worker =
  spawn(fn ->
    worker_loop()
  end)

The worker receives jobs and keeps private state in its recursive loop.

If it crashes, the process disappears:

Client                       Worker

  |                            |
  | {:work, value}             |
  |--------------------------->|
  |                            |
  |                         crash
  |                            X
  |
  | {:work, another_value}
  |---------------------------> no process

Messages sent to a dead local PID do not bring that process back. The application needs another process responsible for detecting the failure and deciding whether to start a replacement.

We could periodically call:

Process.alive?(worker)

But polling has several problems:

failure detection is delayed until the next check
restart logic becomes scattered through the application
the code must repeatedly check every worker
a liveness check does not tell us why the worker stopped

The BEAM already gives us event-driven failure detection through links and exit signals.

We need a process that uses those primitives to apply a restart policy.

The Design

We will create one manual supervisor and two workers:

                 +------------------------+
                 |   Manual Supervisor    |
                 |------------------------|
                 | trap_exit = true       |
                 | child PID -> name map  |
                 | restart loop           |
                 +------------------------+
                    /                  \
              link /                    \ link
                  v                      v
        +------------------+    +------------------+
        |     worker_1     |    |     worker_2     |
        |------------------|    |------------------|
        | receive loop     |    | receive loop     |
        | private state    |    | private state    |
        +------------------+    +------------------+

The supervisor will:

enable exit trapping
start and link both workers
record the relationship between each PID and logical worker name
wait for messages
restart a worker after an abnormal exit
leave healthy siblings running
avoid restarting a worker after a normal exit

This is a minimal one-for-one restart strategy.

The Worker

Our worker starts with a logical name and a processed-job count:

defmodule Worker do
  def start(name) do
    spawn_link(fn -> loop(name, 0) end)
  end
end

The call to spawn_link/1 does two things atomically:

start process + create link

The link connects the worker's failure lifecycle to the process that starts it. In this example, that process is our manual supervisor.

The worker then enters a receive loop:

defp loop(name, count) do
  receive do
    {:work, caller, value} ->
      new_count = count + 1
      send(caller, {:done, name, value, new_count})
      loop(name, new_count)

    :crash ->
      raise "#{name} crashed"

    :stop ->
      IO.puts("#{name} stopping normally")
      :ok
  end
end

The worker supports three messages.

Processing Work

{:work, caller, value}

For each job, the worker increments its private counter and replies:

{:done, name, value, new_count}

The recursive call carries the new state into the next loop:

loop(name, new_count)

Crashing Intentionally

:crash

This branch raises an exception:

raise "#{name} crashed"

The exception terminates the worker with an abnormal exit reason. Because the worker is linked to the supervisor, an exit signal travels across that link.

Stopping Normally

:stop

Returning :ok lets the worker function finish normally. The worker exits with reason:

:normal

That distinction will determine whether our supervisor restarts it.

Starting the Manual Supervisor

The public start/1 function creates the supervisor process:

defmodule ManualSupervisor do
  def start(worker_names) do
    spawn(fn -> init(worker_names) end)
  end
end

We use spawn/1, rather than spawn_link/1, here because this example focuses on links between the supervisor and its children. The caller running the demonstration is not part of that supervision relationship.

The new supervisor begins in init/1:

defp init(worker_names) do
  Process.flag(:trap_exit, true)

  children =
    Map.new(worker_names, fn name ->
      pid = start_child(name)
      {pid, name}
    end)

  loop(children)
end

The order here matters.

The supervisor enables exit trapping before it starts linked children:

Process.flag(:trap_exit, true)

If a child crashes immediately after it starts, the supervisor is already prepared to receive the exit as a message instead of being terminated by it.

Tracking Child Identity

A restarted process receives a new PID.

That means a PID cannot be the permanent identity of a logical worker:

worker_1 before crash -> #PID<0.102.0>
worker_1 after crash  -> #PID<0.104.0>

The name remains :worker_1, but the process incarnation changes.

Our supervisor stores its children in a map:

%{
  worker_1_pid => :worker_1,
  worker_2_pid => :worker_2
}

The PID is the key because the exit message identifies the terminated process by PID:

{:EXIT, pid, reason}

With the map, the supervisor can recover the logical name:

name = Map.fetch!(children, pid)

It can then start a replacement using that name.

Starting Children

Child creation is kept in one helper:

defp start_child(name) do
  pid = Worker.start(name)
  IO.puts("supervisor: started #{name} as #{inspect(pid)}")
  pid
end

Worker.start/1 uses spawn_link/1, so the calling supervisor becomes linked to the new worker.

After initialization, the relationship looks like this:

ManualSupervisor
    |
    +---- link ---- worker_1
    |
    +---- link ---- worker_2

The supervisor then carries the child map into its recursive message loop:

loop(children)

Detecting an Abnormal Exit

Suppose the demonstration sends this message:

send(worker_1, :crash)

The worker raises an exception and terminates.

Because the supervisor traps exits, it receives:

{:EXIT, worker_1_pid, reason}

The abnormal-exit branch handles that message:

{:EXIT, pid, reason} ->
  name = Map.fetch!(children, pid)
  IO.puts("supervisor: #{name} exited with #{inspect(reason)}")

  children = Map.delete(children, pid)
  new_pid = start_child(name)
  IO.puts("supervisor: restarted #{name} as #{inspect(new_pid)}")

  loop(Map.put(children, new_pid, name))

There are four important steps here.

1. Identify the Logical Worker

name = Map.fetch!(children, pid)

The exit message contains the old PID. The child map tells us that this PID belonged to :worker_1.

2. Remove the Dead PID

children = Map.delete(children, pid)

The old PID will never become alive again, so it must not remain in the supervisor's state.

3. Start a Replacement

new_pid = start_child(name)

This creates another linked process with the same logical name and a new PID.

4. Store the Replacement

loop(Map.put(children, new_pid, name))

The recursive loop continues with the updated child map.

The Restart Message Flow

The complete interaction looks like this:

Demo                  Supervisor                 worker_1

send(worker_1, :crash)
----------------------------------------------->

                                               raises
                                               exits

                      {:EXIT, pid, reason}
                      <-------------------------- BEAM

                      find name for old PID
                      remove old PID
                      spawn_link replacement

children(supervisor)
--------------------->

{:children, ref,
 %{worker_1: new_pid,
   worker_2: same_pid}}
<---------------------

The important result is:

worker_1 crashes -> worker_1 is replaced
worker_2 survives -> worker_2 keeps the same PID

That is the meaning of one-for-one supervision in this small example.

Healthy Siblings Stay Alive

When worker_1 crashes, the supervisor does not rebuild its entire child set.

It removes and replaces only the matching PID:

children = Map.delete(children, worker_1_pid)
children = Map.put(children, new_worker_1_pid, :worker_1)

The worker_2 entry is untouched.

Before

worker_1 -> #PID<0.102.0>
worker_2 -> #PID<0.103.0>

After worker_1 crashes

worker_1 -> #PID<0.104.0>  changed
worker_2 -> #PID<0.103.0>  unchanged

This matters because a healthy worker may contain useful in-memory state. Restarting it unnecessarily would discard that state and interrupt work it could have continued performing.

OTP also provides other restart strategies for cases where child lifecycles are related, but our implementation supports only one-for-one replacement.

Normal Exit Is Not a Failure

Our supervisor has a separate pattern for a normal exit:

{:EXIT, pid, :normal} ->
  name = Map.fetch!(children, pid)
  IO.puts("supervisor: #{name} stopped normally; not restarting")
  loop(Map.delete(children, pid))

When a worker receives :stop, it finishes normally:

:stop ->
  IO.puts("#{name} stopping normally")
  :ok

The supervisor removes that child but does not replace it.

This gives our restart policy a simple rule:

reason == :normal -> do not restart
reason != :normal -> restart

This is closest to OTP's :transient restart policy, which restarts a child only after an abnormal exit. It is still deliberately simplified. OTP child specifications provide explicit :permanent, :transient, and :temporary policies instead of forcing every application to encode the decision directly in a receive loop.

Asking the Supervisor for Its Children

The demonstration needs a safe way to discover current child PIDs, including a replacement PID after a crash.

The public function uses a correlated request-reply protocol:

def children(supervisor) do
  ref = make_ref()
  send(supervisor, {:children, self(), ref})

  receive do
    {:children, ^ref, children} -> children
  after
    1_000 -> {:error, :timeout}
  end
end

This reuses an idea from Part 2.

The request includes:

{:children, caller_pid, reference}

The supervisor transforms its internal PID-to-name map into a caller-friendly name-to-PID map:

{:children, caller, ref} ->
  by_name = Map.new(children, fn {pid, name} -> {name, pid} end)
  send(caller, {:children, ref, by_name})
  loop(children)

The reply is:

{:children, reference, %{
  worker_1: worker_1_pid,
  worker_2: worker_2_pid
}}

The pinned reference ensures that the caller accepts only the response belonging to its request:

{:children, ^ref, children}

Our supervisor is already combining several earlier primitives:

mailbox loop
    +
request/reply correlation
    +
process links
    +
trapped exits
    +
restart policy

Replacement Means Fresh State

Each worker begins with a count of zero:

spawn_link(fn -> loop(name, 0) end)

Suppose worker_1 processes three jobs and then crashes:

old worker_1 count = 3

The replacement is a completely new process:

new worker_1 count = 0

Supervision restores the process, not its private memory.

That distinction is fundamental:

Restarting a worker restores availability only if the worker can begin again from valid initial state or reconstruct its state from somewhere durable.

Real applications may rebuild state from a database, an event log, another process, a cache, or the arguments in a child specification.

Our worker intentionally starts from zero so that this behavior is visible. After restart, its first completed job replies with a count of 1:

{:done, :worker_1, :job_after_restart, 1}

Stopping the Supervisor

The supervisor also handles its own :stop message:

:stop ->
  Enum.each(Map.keys(children), &send(&1, :stop))
  wait_for_children(Map.keys(children))

It asks each current child to stop normally, then waits for all corresponding exit messages:

defp wait_for_children([]), do: :ok

defp wait_for_children(children) do
  receive do
    {:EXIT, pid, _reason} ->
      wait_for_children(List.delete(children, pid))
  end
end

Because exit trapping remains enabled, normal child exits arrive as :EXIT messages. Waiting for them ensures that the supervisor does not finish before its children have processed the shutdown request.

This is only a minimal shutdown protocol. It has no timeout and cannot force a child that ignores :stop to terminate.

OTP supervisors implement more complete shutdown behavior.

The Complete Example

defmodule Worker do
  def start(name) do
    spawn_link(fn -> loop(name, 0) end)
  end

  defp loop(name, count) do
    receive do
      {:work, caller, value} ->
        new_count = count + 1
        send(caller, {:done, name, value, new_count})
        loop(name, new_count)

      :crash ->
        raise "#{name} crashed"

      :stop ->
        IO.puts("#{name} stopping normally")
        :ok
    end
  end
end

defmodule ManualSupervisor do
  def start(worker_names) do
    spawn(fn -> init(worker_names) end)
  end

  def children(supervisor) do
    ref = make_ref()
    send(supervisor, {:children, self(), ref})

    receive do
      {:children, ^ref, children} -> children
    after
      1_000 -> {:error, :timeout}
    end
  end

  defp init(worker_names) do
    Process.flag(:trap_exit, true)

    children =
      Map.new(worker_names, fn name ->
        pid = start_child(name)
        {pid, name}
      end)

    loop(children)
  end

  defp loop(children) do
    receive do
      {:EXIT, pid, :normal} ->
        name = Map.fetch!(children, pid)
        IO.puts("supervisor: #{name} stopped normally; not restarting")
        loop(Map.delete(children, pid))

      {:EXIT, pid, reason} ->
        name = Map.fetch!(children, pid)
        IO.puts("supervisor: #{name} exited with #{inspect(reason)}")

        children = Map.delete(children, pid)
        new_pid = start_child(name)
        IO.puts("supervisor: restarted #{name} as #{inspect(new_pid)}")

        loop(Map.put(children, new_pid, name))

      {:children, caller, ref} ->
        by_name = Map.new(children, fn {pid, name} -> {name, pid} end)
        send(caller, {:children, ref, by_name})
        loop(children)

      :stop ->
        Enum.each(Map.keys(children), &send(&1, :stop))
        wait_for_children(Map.keys(children))
    end
  end

  defp start_child(name) do
    pid = Worker.start(name)
    IO.puts("supervisor: started #{name} as #{inspect(pid)}")
    pid
  end

  defp wait_for_children([]), do: :ok

  defp wait_for_children(children) do
    receive do
      {:EXIT, pid, _reason} ->
        wait_for_children(List.delete(children, pid))
    end
  end
end

defmodule SupervisorDemo do
  def run do
    supervisor = ManualSupervisor.start([:worker_1, :worker_2])
    children_before = ManualSupervisor.children(supervisor)

    worker_1_before = children_before.worker_1
    worker_2_before = children_before.worker_2

    send(worker_1_before, :crash)

    children_after =
      wait_for_restart(supervisor, :worker_1, worker_1_before)

    IO.puts(
      "worker_1 restarted: #{children_after.worker_1 != worker_1_before}"
    )

    IO.puts(
      "worker_2 unchanged: #{children_after.worker_2 == worker_2_before}"
    )

    send(
      children_after.worker_1,
      {:work, self(), :job_after_restart}
    )

    receive do
      message -> IO.inspect(message, label: "reply")
    after
      1_000 -> IO.puts("reply timed out")
    end

    send(supervisor, :stop)
  end

  defp wait_for_restart(supervisor, name, old_pid, attempts \\ 50)

  defp wait_for_restart(_supervisor, _name, _old_pid, 0) do
    raise "worker was not restarted"
  end

  defp wait_for_restart(supervisor, name, old_pid, attempts) do
    children = ManualSupervisor.children(supervisor)

    case Map.get(children, name) do
      pid when is_pid(pid) and pid != old_pid ->
        children

      _ ->
        Process.sleep(10)
        wait_for_restart(supervisor, name, old_pid, attempts - 1)
    end
  end
end

SupervisorDemo.run()

Running the Example

From the 05-supervisor-from-scratch directory, run:

elixir supervisor.exs

Example output:

supervisor: started worker_1 as #PID<0.102.0>
supervisor: started worker_2 as #PID<0.103.0>

10:30:00.000 [error] Process #PID<0.102.0> raised an exception
** (RuntimeError) worker_1 crashed

supervisor: worker_1 exited with {%RuntimeError{message: "worker_1 crashed"}, ...}
supervisor: started worker_1 as #PID<0.104.0>
supervisor: restarted worker_1 as #PID<0.104.0>
worker_1 restarted: true
worker_2 unchanged: true
reply: {:done, :worker_1, :job_after_restart, 1}
worker_1 stopping normally
worker_2 stopping normally

The exact PIDs, timestamps, stack trace, and output order may vary because the processes execute concurrently.

The exception report is expected. We deliberately crash worker_1 to exercise the restart path.

The important observations are:

worker_1 has a new PID
worker_2 has its original PID
replacement worker_1 accepts work
replacement worker_1 begins with fresh state

Why the Demo Polls for the New PID

After sending :crash, the caller cannot assume that the supervisor has already processed the resulting exit signal.

These operations are concurrent:

caller sends :crash
worker receives :crash
worker exits
supervisor receives :EXIT
supervisor starts replacement

The caller therefore asks for the children repeatedly until worker_1 has a different PID:

case Map.get(children, name) do
  pid when is_pid(pid) and pid != old_pid ->
    children

  _ ->
    Process.sleep(10)
    wait_for_restart(supervisor, name, old_pid, attempts - 1)
end

This polling exists only in the demonstration code so it can verify the asynchronous restart. The supervisor itself does not poll for failures; it reacts to linked exit messages.

What Makes This a Supervisor?

Our process performs the smallest useful form of supervision:

start children
      |
link to children
      |
observe exits
      |
apply restart policy
      |
start replacements

The mechanism comes from BEAM primitives:

links connect process lifecycles
exit trapping converts exit signals into messages
the mailbox delivers failure information
recursion preserves the current child set

The policy comes from our own code:

normal exits are not restarted
abnormal exits are restarted
only the failed child is restarted
replacements begin with the same logical name

OTP's Supervisor separates and formalizes these responsibilities through child specifications and restart strategies.

The Restart Loop Is Not Enough

It is tempting to look at this example and conclude that supervision is simply:

receive do
  {:EXIT, pid, reason} ->
    start_replacement(pid)
end

But production supervision must answer many more questions.

What If a Worker Crashes Forever?

Our supervisor restarts every abnormal exit without a limit:

start -> crash -> restart -> crash -> restart -> crash -> ...

That loop can consume CPU, flood logs, and hide a persistent configuration or programming error.

A real supervisor needs restart intensity limits, such as a maximum number of restarts within a time window.

What If Children Depend on Each Other?

One-for-one is not always correct.

If a producer and consumer share a protocol state, restarting only one could leave them inconsistent. Other strategies may restart multiple children together or restart children in a defined order.

What If Shutdown Hangs?

Our supervisor waits forever if a worker ignores :stop:

wait_for_children(children)

Production shutdown needs timeouts and a way to force termination after the grace period expires.

What If Child Startup Fails?

Our start_child/1 assumes that starting a worker succeeds immediately.

Real child startup may need to validate configuration, open resources, or return an error. The supervisor must handle partial initialization and report why startup failed.

In What Order Should Processes Start and Stop?

Applications often require dependencies to start before their consumers and stop after them. A plain map does not represent that ordering.

How Should Children Be Configured?

Our child configuration is only a list of names:

[:worker_1, :worker_2]

Real supervision needs structured information about how to start, restart, identify, and shut down each child.

These requirements are why OTP supervisors exist.

Why Use OTP Supervisor in Real Applications?

Our manual supervisor is a learning exercise. It makes the underlying process model visible, but it is not production-ready.

OTP's Supervisor provides tested implementations of concerns including:

child specifications
one-for-one and other restart strategies
restart intensity and period limits
deterministic startup and shutdown behavior
configurable shutdown timeouts
supervisor initialization guarantees
child introspection
structured error reporting
supervision trees

The value of OTP is not that these behaviors are impossible to write ourselves.

The value is that failure handling contains subtle edge cases, and OTP provides a standard implementation that has been exercised across decades of BEAM systems.

Understanding the raw primitives helps us use that abstraction deliberately.

Limitations of Our Implementation

Our manual supervisor intentionally leaves out many production requirements:

a repeatedly crashing worker is restarted forever
there is no restart delay or exponential backoff
there is no restart-frequency limit
only one-for-one replacement is supported
restarted workers lose all private in-memory state
child startup cannot return a structured error
shutdown has no timeout
an unresponsive child can block supervisor shutdown forever
there is no nested supervision tree
the child registry is a private map
unknown linked exit messages would crash Map.fetch!/2
the supervisor is not linked to the process that starts it

These are not reasons to keep expanding the custom implementation until it becomes a second supervisor library.

They are evidence for using OTP's existing one.

What We Built

Without using GenServer or OTP Supervisor, we now have:

Process
   +
Mailbox
   +
Request / Reply
   +
Request References
   +
Process Monitoring
   +
Process Linking
   +
Trapped Exits
   +
Child Tracking
   +
One-for-One Restart

The central idea from this part is:

A supervisor combines failure detection with a restart policy.

Links and trapped exits tell the supervisor that a child terminated.

The supervisor's policy decides whether to restart it, which process to replace, and what happens to the remaining children.

Our one-for-one policy replaces only the failed worker and preserves its healthy sibling.

What's Next?

Our caller currently discovers workers by asking the supervisor for its private child map.

That works inside this small example, but passing and refreshing PIDs becomes awkward as a system grows.

In Part 6, we'll explore named processes and process registration:

logical name
     |
     v
current process PID

This will let processes find one another through stable names instead of sharing raw PIDs everywhere.

Source code link: https://github.com/pckrishnadas88/elixir-distributed-systems-lab/tree/main/05-supervisor-from-scratch

Series

← Previous: Part 4 — Process Linking

Next: Part 6 - Named processes →

Content License

This article is licensed under CC BY-NC 4.0. You may share and adapt it with attribution for non-commercial purposes. The accompanying source code remains licensed under the MIT License.

Building Distributed Systems in Elixir: Part 4 — Process Linking

krishnadaspc — Fri, 14 Aug 2026 12:02:52 +0000

In the previous part of this series, we used process monitors to observe when another process terminated.

A monitor sends a message like this:

{:DOWN, ref, :process, pid, reason}

The monitoring process stays alive and decides what to do with that information.

But sometimes observation is not enough.

Two processes may be part of the same operation. If one of them fails, allowing the other to continue could leave the system in an invalid state.

For that relationship, the BEAM provides process links.

In this part, we'll build two small examples using:

spawn_link/1
Process.flag(:trap_exit, true)

No GenServer.

No Supervisor.

The goal is to understand how failures propagate between processes before introducing OTP supervision.

The Problem

Imagine a parent process starts a worker:

worker =
  spawn(fn ->
    perform_work()
  end)

The two processes execute independently.

If the worker crashes, the parent is unaffected unless it explicitly monitors the worker or checks its state.

Parent Process                 Worker Process

      alive                       working
        |                            |
        |                            X crash
        |
      alive

That independence is useful when the processes do unrelated work.

But suppose the worker is essential to the parent's operation. Continuing without it would not make sense.

What we want is a shared failure relationship:

worker fails
     |
     v
parent is affected

This is what process links provide.

Creating a Linked Process

We can create and link a process in one operation:

worker =
  spawn_link(fn ->
    perform_work()
  end)

Conceptually, spawn_link/1 combines:

spawn/1
Process.link/1

The important difference is that spawning and linking happen atomically. There is no gap in which the child could terminate before the link is established.

After spawn_link/1, the processes have a bidirectional relationship:

Parent Process <----------> Worker Process
                  link

The word bidirectional matters.

The link does not represent ownership or a one-way subscription. An abnormal exit from either process can affect the other.

Exit Propagation

Let's start a linked worker that intentionally crashes:

worker =
  spawn_link(fn ->
    IO.puts("Worker started: #{inspect(self())}")
    Process.sleep(500)

    raise "Worker crashed"
  end)

The worker raises an exception:

raise "Worker crashed"

That causes the worker to terminate with an abnormal exit reason.

Because the worker is linked to the parent, the BEAM sends an exit signal across the link.

Parent Process                    Worker Process

spawn_link(fn -> ... end)
--------------------------------->

                            worker starts

                            Process.sleep(500)

                            raise "Worker crashed"

                            worker terminates
                            abnormally
                                  X
                                  |
                     exit signal  |
                    <-------------+
        |
        X parent terminates

By default, an abnormal exit signal terminates the receiving process too.

The parent does not receive an ordinary mailbox message in this mode. The exit signal directly affects its lifecycle.

A Propagation Example

Here is the first mode from our script:

def run("propagate") do
  IO.puts("Parent started: #{inspect(self())}")

  worker =
    spawn_link(fn ->
      IO.puts("Worker started: #{inspect(self())}")
      Process.sleep(500)

      raise "Worker crashed"
    end)

  IO.puts("Parent linked to worker: #{inspect(worker)}")
  IO.puts("The worker crash will terminate the parent.")

  Process.sleep(:infinity)
end

The parent sleeps forever here so that it remains alive while the worker runs:

Process.sleep(:infinity)

Without that line, the parent function could finish normally before the worker reaches the intentional crash.

The sleep is not responsible for detecting the failure. It simply keeps the parent alive long enough for the linked exit signal to arrive.

Run the example with:

elixir process_linking.exs propagate

Example output:

Parent started: #PID<0.95.0>
Parent linked to worker: #PID<0.102.0>
Worker started: #PID<0.102.0>
The worker crash will terminate the parent.
** (EXIT from #PID<0.95.0>) an exception was raised:
    ** (RuntimeError) Worker crashed

The exact PIDs, stack trace, and output order may vary between runs because the processes execute concurrently.

The important result is the same:

worker crashes -> parent terminates

Normal and Abnormal Exits

Links propagate exit signals, but the exit reason matters.

A process that finishes normally exits with:

:normal

Normal exits do not ordinarily take down linked processes that are not trapping exits.

An exception produces an abnormal reason. In our example, that reason contains the RuntimeError and stack trace.

Other important reasons include:

:shutdown
:killed

So the simplified rule for an ordinary process is:

linked process exits normally    -> remain alive
linked process exits abnormally  -> terminate

There are special cases—particularly untrappable :kill signals—but this model is enough to understand the example and the motivation for supervisors.

Trapping Exits

Sometimes a linked process needs to react to a failure instead of immediately terminating.

The current process can enable exit trapping:

Process.flag(:trap_exit, true)

This changes an incoming link exit signal into a regular mailbox message:

{:EXIT, from_pid, reason}

Now the process can handle the exit using receive.

receive do
  {:EXIT, ^worker, reason} ->
    IO.puts("Parent received an EXIT message.")
    IO.puts("Reason: #{inspect(reason)}")
end

The worker still crashes.

Exit trapping changes the receiving parent's response to that crash.

Without trap_exit              With trap_exit

exit signal                    exit signal
     |                              |
     v                              v
parent terminates              {:EXIT, pid, reason}
                                     |
                                     v
                               parent's mailbox

A Trapped-Exit Example

Here is the second mode:

def run("trap") do
  Process.flag(:trap_exit, true)

  IO.puts("Parent started: #{inspect(self())}")
  IO.puts("Parent is trapping exits")

  worker =
    spawn_link(fn ->
      IO.puts("Worker started: #{inspect(self())}")
      Process.sleep(500)

      raise "Worker crashed"
    end)

  IO.puts("Parent linked to worker: #{inspect(worker)}")

  receive do
    {:EXIT, ^worker, reason} ->
      IO.puts("Parent received an EXIT message.")
      IO.puts("Worker: #{inspect(worker)}")
      IO.puts("Reason: #{inspect(reason)}")
      IO.puts("Parent is still alive.")
  end
end

The call to Process.flag/2 affects the process that calls it:

Process.flag(:trap_exit, true)

In this script, that is the parent process executing run/1.

It must be enabled before the linked worker crashes. Otherwise, the abnormal signal can terminate the parent before it enters receive.

Understanding the `:EXIT` Message

The trapped exit message has this shape:

{:EXIT, from_pid, reason}

Let's break it down.

`:EXIT`

Identifies the value as an exit signal converted into a mailbox message.

`from_pid`

Identifies the linked process that terminated.

`reason`

Describes why that process terminated.

In our example, the reason contains the exception and stack trace because the worker terminates by raising:

raise "Worker crashed"

Why Use `^worker`?

The receive pattern contains:

{:EXIT, ^worker, reason}

The variable was already bound when we created the process:

worker = spawn_link(fn -> ... end)

The pin operator means:

Match an exit message from this particular worker.

Without the pin:

{:EXIT, worker, reason}

the pattern could bind worker to the PID found in any matching :EXIT message.

Pinning matters when a process has several links and its mailbox may contain exit messages from more than one process.

Running the Trap Example

Run:

elixir process_linking.exs trap

Example output:

Parent started: #PID<0.95.0>
Parent is trapping exits
Parent linked to worker: #PID<0.102.0>
Worker started: #PID<0.102.0>

11:40:13.501 [error] Process #PID<0.102.0> raised an exception
** (RuntimeError) Worker crashed

Parent received an EXIT message.
Worker: #PID<0.102.0>
Reason: {%RuntimeError{message: "Worker crashed"}, [...]}
Parent is still alive.

The error report is expected. Trapping an exit does not prevent or hide the worker's exception.

It prevents that linked abnormal exit from automatically terminating the parent and makes the reason available as a message.

The Complete Example

defmodule ProcessLinking do
  def run("propagate") do
    IO.puts("Parent started: #{inspect(self())}")

    worker =
      spawn_link(fn ->
        IO.puts("Worker started: #{inspect(self())}")
        Process.sleep(500)

        raise "Worker crashed"
      end)

    IO.puts("Parent linked to worker: #{inspect(worker)}")
    IO.puts("The worker crash will terminate the parent.")

    Process.sleep(:infinity)
  end

  def run("trap") do
    Process.flag(:trap_exit, true)

    IO.puts("Parent started: #{inspect(self())}")
    IO.puts("Parent is trapping exits")

    worker =
      spawn_link(fn ->
        IO.puts("Worker started: #{inspect(self())}")
        Process.sleep(500)

        raise "Worker crashed"
      end)

    IO.puts("Parent linked to worker: #{inspect(worker)}")

    receive do
      {:EXIT, ^worker, reason} ->
        IO.puts("Parent received an EXIT message.")
        IO.puts("Worker: #{inspect(worker)}")
        IO.puts("Reason: #{inspect(reason)}")
        IO.puts("Parent is still alive.")
    end
  end

  def run(_) do
    IO.puts("""
    Usage:

      elixir process_linking.exs propagate
      elixir process_linking.exs trap
    """)
  end
end

mode = List.first(System.argv())

ProcessLinking.run(mode)

The script has two modes because an untrapped linked crash terminates the process running the example. Keeping the demonstrations separate makes each behavior clear.

Links vs Monitors

Links and monitors both provide information about process termination, but they express different relationships.

Monitor
-------
one-way observation
monitored process terminates
observer receives {:DOWN, ...}
observer stays alive

Link
----
bidirectional lifecycle relationship
linked process terminates abnormally
exit signal propagates
receiver normally terminates too

With exit trapping, a link can also produce a mailbox message, but it is still not the same as a monitor.

Monitor notification

{:DOWN, ref, :process, pid, reason}

Trapped link exit

{:EXIT, pid, reason}

A monitor includes a unique reference and is one-way.

A link is bidirectional and connects the failure behavior of both processes.

A useful question is:

Do I need to observe this process?
    -> monitor

Should our failure lifecycles be connected?
    -> link

Links Are Not Supervision

At this point, the trapped parent knows that its worker failed:

{:EXIT, worker, reason}

But it still has to decide what happens next.

It might:

log the failure
stop itself
start a replacement worker
retry with a limit
apply a restart policy

Our example does only the first step: it receives and prints the failure.

A production-quality restart system must answer more questions:

Which processes should restart?

How many restart attempts are allowed?

What state should a replacement receive?

What if restarts repeatedly fail?

Should sibling processes restart too?

Those decisions are why OTP provides supervisors.

What We Built

Without using GenServer or Supervisor, we now have:

Process
   +
Mailbox
   +
Request / Reply
   +
Request References
   +
Process Monitoring
   +
Process Linking
   +
Trapped Exits

The important idea from this part is:

Links turn process failure into a relationship between process lifecycles.

Without exit trapping, an abnormal exit propagates and normally terminates the linked process.

With exit trapping, that signal becomes an :EXIT message that the process can handle.

What's Next?

We can now detect a worker failure and receive its reason.

The next step is to respond by starting a replacement.

In Part 5, we'll build a small supervisor from scratch and explore:

worker crashes
      |
      v
supervisor receives exit
      |
      v
supervisor starts replacement

That will show why restart strategies, restart intensity limits, and OTP supervisors exist.

Series

← Previous: Part 3 — Process Monitoring

Next: Part 5 - Supervisor from scratch →

Source code link : https://github.com/pckrishnadas88/elixir-distributed-systems-lab/tree/main/04-process-linking

Building Distributed Systems in Elixir: Part 3 — Process Monitoring

krishnadaspc — Thu, 13 Aug 2026 14:44:20 +0000

In the previous parts of this series, we built a stateful process using a mailbox and then added correlated request–reply using references.

Now we have another problem.

What happens when the process we are communicating with dies?

A worker might finish normally, crash because of an exception, or terminate for some other reason.

Instead of repeatedly checking whether a process is alive, the BEAM provides process monitoring.

In this part, we'll build process monitoring directly using:

Process.monitor/1

No GenServer.

No Supervisor.

The goal is to understand the primitive underneath OTP's failure-handling mechanisms.

The Problem

Imagine we start a worker:

worker = Worker.start("worker-1")

That worker is an independent BEAM process.

At some point it could stop:

send(worker, :stop)

or crash:

send(worker, :crash)

Another process may need to know that the worker is gone.

One option would be repeatedly checking:

Process.alive?(worker)

But polling isn't what we want.

Instead, we can ask the BEAM to notify us when the process terminates.

Monitoring a Process

Monitoring starts with one line:

ref = Process.monitor(worker)

The important detail is:

The process calling Process.monitor/1 becomes the monitoring process.

Process.monitor/1 returns a unique reference.

If the worker later terminates, the BEAM sends the monitoring process a message:

{:DOWN, ref, :process, worker, reason}

This is just a message.

That means we can handle it using the same receive primitive we've already been using.

A Simple Worker

Let's create a worker that understands three messages:

defmodule Worker do
  def start(name) do
    spawn(fn -> loop(name) end)
  end

  defp loop(name) do
    receive do
      {:work, from, value} ->
        IO.puts("#{name} received #{value}")
        send(from, {:done, name, value})
        loop(name)

      :stop ->
        IO.puts("#{name} stopping normally")
        :ok

      :crash ->
        raise "#{name} crashed"
    end
  end
end

The process starts here:

spawn(fn -> loop(name) end)

The newly spawned process executes loop/1 and waits at:

receive do

If it receives work:

{:work, from, value}

it handles the message and calls:

loop(name)

again.

That keeps the process alive.

But look at :stop:

:stop ->
  IO.puts("#{name} stopping normally")
  :ok

There is no call to loop/1.

The function finishes, so the process terminates normally.

Detecting a Normal Exit

Now let's monitor the worker:

def normal_exit do
  worker = Worker.start("worker-1")

  ref = Process.monitor(worker)

  send(worker, :stop)

  receive do
    {:DOWN, ^ref, :process, ^worker, reason} ->
      IO.puts("worker-1 terminated")
      IO.puts("reason: #{inspect(reason)}")
  end
end

The execution flow looks like this:

Monitoring Process                 Worker Process

       |                                |
       | -------- spawn ------------->  |
       |                                |
       |                             receive
       |                             waiting
       |
       | Process.monitor(worker)
       |
       | -------- :stop ------------->  |
       |                                |
       |                         receives :stop
       |                                |
       |                          returns :ok
       |                                |
       |                           terminates
       |                                X
       |
       |      BEAM detects termination
       |
       | <---- {:DOWN, ..., :normal}
       |
    receive
       |
       v
 reason = :normal

There are two separate messages here.

First:

:stop

is sent to the worker's mailbox.

Later:

{:DOWN, ref, :process, worker, :normal}

is sent by the BEAM to the monitoring process's mailbox.

This distinction is important.

Understanding the `:DOWN` Message

The notification has this shape:

{:DOWN, ref, :process, pid, reason}

Let's break it down.

`:DOWN`

Identifies the message as a monitor notification.

`ref`

The unique monitor reference returned by:

Process.monitor(worker)

`:process`

Indicates that the monitored entity was a process.

`pid`

The PID of the process that terminated.

`reason`

Describes why the process terminated.

For a normal exit:

:normal

For a crash, it contains information about the failure.

Why Use `^ref`?

Our receive pattern contains:

{:DOWN, ^ref, :process, ^worker, reason}

The pin operator ^ tells Elixir to match against an existing value.

We already have:

ref = Process.monitor(worker)

So:

^ref

means:

Match the :DOWN message belonging to this particular monitor.

Likewise:

^worker

means:

Match the PID of this particular worker.

This lets us precisely identify the termination notification we're waiting for.

Detecting a Crash

Monitoring also works when a worker crashes.

def crash_exit do
  worker = Worker.start("worker-2")

  ref = Process.monitor(worker)

  send(worker, :crash)

  receive do
    {:DOWN, ^ref, :process, ^worker, reason} ->
      IO.puts("worker-2 terminated")
      IO.puts("reason: #{inspect(reason)}")
  end
end

The worker receives:

:crash

and executes:

raise "#{name} crashed"

The worker dies.

But the monitoring process does not die with it.

Instead, the BEAM sends it a :DOWN message containing the failure reason.

This gives us an important property:

Monitoring lets one process observe another process's failure without automatically propagating that failure.

Monitoring Multiple Workers

Real systems usually have more than one worker.

Let's start three:

worker1 = Worker.start("worker-1")
worker2 = Worker.start("worker-2")
worker3 = Worker.start("worker-3")

Then monitor each one:

ref1 = Process.monitor(worker1)
ref2 = Process.monitor(worker2)
ref3 = Process.monitor(worker3)

Each monitor gets a different reference.

We can keep track of them using a map:

%{
  ref1 => "worker-1",
  ref2 => "worker-2",
  ref3 => "worker-3"
}

Think of this map as:

workers whose termination we are still waiting for

ref1 -> worker-1
ref2 -> worker-2
ref3 -> worker-3

Now send different termination messages:

send(worker1, :stop)
send(worker2, :crash)
send(worker3, :stop)

One worker crashes while the other two stop normally.

Waiting for All Workers

We can write a small recursive function to collect the monitor notifications:

defp wait_for_workers(monitors) when map_size(monitors) == 0 do
  IO.puts("all workers terminated")
end

defp wait_for_workers(monitors) do
  receive do
    {:DOWN, ref, :process, pid, reason} ->
      case Map.pop(monitors, ref) do
        {nil, monitors} ->
          wait_for_workers(monitors)

        {name, remaining} ->
          IO.puts("#{name} terminated")
          IO.puts("pid: #{inspect(pid)}")
          IO.puts("reason: #{inspect(reason)}")
          IO.puts("")

          wait_for_workers(remaining)
      end
  end
end

The function has a simple purpose:

Wait for each worker to terminate and log what happened.

Conceptually:

3 workers
    |
    | worker terminates
    v
receive :DOWN
    |
    v
log termination
    |
    v
remove monitor
    |
    v
2 workers
    |
    | worker terminates
    v
receive :DOWN
    |
    v
1 worker
    |
    | worker terminates
    v
receive :DOWN
    |
    v
0 workers
    |
    v
finished

The order is not guaranteed.

Even though we send:

send(worker1, :stop)
send(worker2, :crash)
send(worker3, :stop)

the workers execute concurrently.

Any of them may terminate first.

The Complete Example

defmodule Worker do
  def start(name) do
    spawn(fn -> loop(name) end)
  end

  defp loop(name) do
    receive do
      {:work, from, value} ->
        IO.puts("#{name} received #{value}")
        send(from, {:done, name, value})
        loop(name)

      :stop ->
        IO.puts("#{name} stopping normally")
        :ok

      :crash ->
        raise "#{name} crashed"
    end
  end
end

defmodule MonitorDemo do
  def normal_exit do
    worker = Worker.start("worker-1")
    ref = Process.monitor(worker)

    send(worker, :stop)

    receive do
      {:DOWN, ^ref, :process, ^worker, reason} ->
        IO.puts("worker-1 terminated")
        IO.puts("reason: #{inspect(reason)}")
    end
  end

  def crash_exit do
    worker = Worker.start("worker-2")
    ref = Process.monitor(worker)

    send(worker, :crash)

    receive do
      {:DOWN, ^ref, :process, ^worker, reason} ->
        IO.puts("worker-2 terminated")
        IO.puts("reason: #{inspect(reason)}")
    end
  end

  def multiple_workers do
    worker1 = Worker.start("worker-1")
    worker2 = Worker.start("worker-2")
    worker3 = Worker.start("worker-3")

    ref1 = Process.monitor(worker1)
    ref2 = Process.monitor(worker2)
    ref3 = Process.monitor(worker3)

    send(worker1, :stop)
    send(worker2, :crash)
    send(worker3, :stop)

    wait_for_workers(%{
      ref1 => "worker-1",
      ref2 => "worker-2",
      ref3 => "worker-3"
    })
  end

  defp wait_for_workers(monitors) when map_size(monitors) == 0 do
    IO.puts("all workers terminated")
  end

  defp wait_for_workers(monitors) do
    receive do
      {:DOWN, ref, :process, pid, reason} ->
        case Map.pop(monitors, ref) do
          {nil, monitors} ->
            wait_for_workers(monitors)

          {name, remaining} ->
            IO.puts("#{name} terminated")
            IO.puts("pid: #{inspect(pid)}")
            IO.puts("reason: #{inspect(reason)}")
            IO.puts("")

            wait_for_workers(remaining)
        end
    end
  end
end

MonitorDemo.normal_exit()
MonitorDemo.crash_exit()
MonitorDemo.multiple_workers()

Run it with:

elixir monitor.exs

Monitor vs Polling

We could ask:

Process.alive?(worker)

But that only tells us whether the process was alive at that particular moment.

The process could terminate immediately afterward.

With monitoring:

Process.monitor(worker)

we instead ask the BEAM to notify us about termination.

monitor worker
      |
      v
worker runs independently
      |
      v
worker terminates
      |
      v
BEAM detects termination
      |
      v
{:DOWN, ...}
      |
      v
monitoring process

This fits naturally with the BEAM's message-passing model.

What We Built

Without using GenServer or Supervisor, we now have:

Process
   +
Mailbox
   +
Request / Reply
   +
Request References
   +
Process Monitoring

We're gradually building the primitives needed for fault-tolerant process systems.

The important idea from this part is:

Failure detection itself can be represented as message passing.

A worker terminates, and the monitoring process receives a message describing what happened.

What's Next?

Monitoring lets us observe another process's failure.

But sometimes processes need a stronger relationship.

What if the failure of one process should affect another process?

That's where process links come in.

In Part 4, we'll explore:

spawn_link/1

and see how failures propagate between linked processes.

The examples intentionally use low-level process primitives first so we can understand what OTP abstractions are solving before using them.

Source code of the same can be found at: https://github.com/pckrishnadas88/elixir-distributed-systems-lab

Series

Building a Stateful Process in Elixir Without GenServer
Correlated Request/Reply
Process Monitoring
Process Linking
Supervisor from scratch

The examples intentionally use low-level process primitives first so we can understand what OTP abstractions are solving before using them.

← Previous: Part 2 — Correlated Request/Reply

Next: Part 4 — Process Linking →

Content License

This article is licensed under CC BY-NC 4.0. You may share and adapt it with attribution for non-commercial purposes. The accompanying source code remains licensed under the MIT License.

Building a Distributed System in Go: Part 1 — In-Process Message Passing & CSP Primitives

krishnadaspc — Thu, 13 Aug 2026 06:16:11 +0000

Welcome to Part 1 of the Go Distributed Systems Lab series! Over the course of 20 hands-on projects, we are building core distributed systems primitives from the ground up using Go 1.22+ and the standard library (net, sync, context, log/slog, encoding/binary).

Before jumping into raw socket framing, gossip protocols, or Raft consensus, we need to master the foundational concurrency building blocks inside a single process: Goroutines, Channels, and Communicating Sequential Processes (CSP).

💡 The Philosophy: Share Memory by Communicating

In traditional concurrent programming (like C++ or Java), thread synchronization often relies on shared memory protected by mutexes, lock-free queues, or read-write locks.

Go flips this model with a core design principle:

"Do not communicate by sharing memory; instead, share memory by communicating."

By passing ownership of data structures through Go channels, each pipeline stage operates on isolated memory. This eliminates data races by design without requiring explicit lock management (sync.Mutex).

🏗️ Architecture & Component Design

In this first module (01-message-passing), we construct a 3-stage data processing pipeline:

+------------------+         Job Channel          +------------------+        Result Channel       +-------------------+
|     Producer     |  ------------------------->  |      Worker      |  ------------------------>  |     Collector     |
| (Generates Jobs) |     (Buffered, cap=10)       | (Isolated State) |    (Buffered, cap=10)      | (Aggregates Data) |
+------------------+                              +------------------+                             +-------------------+

1. Ingestion Stage (Producer)

Generates typed Job values and pushes them into a direction-constrained buffered channel (chan<- Job). When generation finishes, it closes the channel to broadcast an end-of-stream signal.

2. Processing Stage (Worker)

Consumes from <-chan Job using Go's for job := range in construct. The worker maintains internal execution metrics (e.g., processedCount) entirely within its local stack scope—no locks required.

3. Collector Stage

Receives output items from <-chan Result and processes them asynchronously until the channel is closed.

💻 Full Implementation

Here is the complete standard-library implementation using Go 1.22+:

package main

import (
    "fmt"
    "log/slog"
    "os"
    "sync"
    "time"
)

// Job represents a unit of work passed between isolated pipeline stages.
type Job struct {
    ID        int
    Payload   string
    Processed bool
}

// Result represents the output produced by the processor stage.
type Result struct {
    JobID     int
    Output    string
    Timestamp time.Time
}

func main() {
    // Initialize standard structured logger (log/slog)
    logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
    slog.SetDefault(logger)

    slog.Info("Starting message passing pipeline", "stage", "initialization")

    // Channels acting as bounded queues
    jobs := make(chan Job, 10)
    results := make(chan Result, 10)

    var wg sync.WaitGroup

    // Stage 1: Ingestion Producer
    wg.Add(1)
    go producer(jobs, 5, &wg)

    // Stage 2: Processing Worker (Isolated State Pipeline)
    wg.Add(1)
    go worker(1, jobs, results, &wg)

    // Stage 3: Aggregator / Collector
    done := make(chan struct{})
    go collector(results, done)

    // Synchronize pipeline termination
    wg.Wait()
    close(results) // Signal collector to exit once upstream workers complete

    <-done // Block until collector drains remaining results
    slog.Info("Pipeline processing complete", "status", "success")
}

// producer generates work items and sends them through the buffered channel.
func producer(out chan<- Job, count int, wg *sync.WaitGroup) {
    defer wg.Done()
    defer close(out) // Closing channel signals EOF to downstream consumers

    for i := 1; i <= count; i++ {
        job := Job{
            ID:      i,
            Payload: fmt.Sprintf("task-payload-%d", i),
        }
        slog.Info("Producing job", "job_id", job.ID)
        out <- job
        time.Sleep(50 * time.Millisecond)
    }
}

// worker encapsulates isolated processing state without shared memory locks.
func worker(id int, in <-chan Job, out chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()

    // Isolated state local to this goroutine - thread-safe without mutexes!
    processedCount := 0

    for job := range in {
        processedCount++
        slog.Info("Worker processing job", 
            "worker_id", id, 
            "job_id", job.ID, 
            "local_total", processedCount,
        )

        // Simulate computation / processing overhead
        time.Sleep(100 * time.Millisecond)

        out <- Result{
            JobID:     job.ID,
            Output:    fmt.Sprintf("PROCESSED[%s]", job.Payload),
            Timestamp: time.Now(),
        }
    }
}

// collector receives final stage outputs until the stream is exhausted.
func collector(in <-chan Result, done chan<- struct{}) {
    for res := range in {
        slog.Info("Result collected", 
            "job_id", res.JobID, 
            "output", res.Output, 
            "time", res.Timestamp.Format(time.RFC3339Nano),
        )
    }
    close(done)
}

🔍 Key Go Concurrency Patterns Explained

1. Directional Channel Types

Notice the parameter signatures in our pipeline functions:

out chan<- Job: Send-only channel. Writing into this is allowed; reading or closing from inside the caller scope (if not intended) produces a compile-time error.
in <-chan Job: Receive-only channel. The worker can only consume from it.

Restricting channel direction at API boundaries prevents accidental closed-channel writes or unauthorized channel closures.

2. Clean Channel Drain Mechanics

When producer() executes close(out), it doesn't delete existing values inside the buffer. Instead, it marks the channel as closed.

Downstream in worker():

for job := range in { ... }

The for range loop continuously extracts items until the channel is empty AND closed, at which point the loop cleanly exits.

3. Structured Logging with `log/slog`

Since Go 1.21, log/slog provides structured key-value logging natively. In distributed systems, plain text strings become impossible to query. Using slog.Info("...", "job_id", job.ID) ensures consistent log parsing across asynchronous routines.

⚠️ Architectural Limitations of In-Process Message Passing

While in-process channels provide clean abstractions, they have clear boundaries when designing real-world distributed systems:

Process Boundary Restrictions: Channels are strictly in-memory data structures managed by the Go runtime scheduler (m:n goroutine multiplexing). They cannot cross host, network, or process boundaries.
Volatile Memory: If the application panics or crashes, all messages currently sitting inside buffered channels are lost permanently.
Coordinated Backpressure: Channels handle backpressure synchronously (blocking on send when full). However, if an upstream system floods the channel faster than workers can consume, memory usage grows up to channel capacity before blocking cascades backward.
No Error Return Channels: This basic pattern lacks a mechanism for workers to report execution errors or request task retries.

🚀 What's Next?

In Part 2, we will address the asynchronous communication problem by building a Correlated Request/Reply mechanism. We will implement unique correlation IDs, response channels, and request multiplexing to turn one-way pipelines into interactive distributed calls!

What concurrency pattern do you use most in your Go services? Drop a comment below! 👇

Source code of the same can be found at : https://github.com/pckrishnadas88/go-distributed-systems-lab/tree/main/01-message-passing

Building Message Passing in Node.js from Scratch

krishnadaspc — Wed, 12 Aug 2026 19:34:55 +0000

"Learn the foundation of distributed systems by passing messages between a Node.js main thread and worker thread."

When we use RabbitMQ, Kafka, or another message broker, it is easy to focus on the API and miss the fundamental idea underneath:

One component owns some state, and other components communicate with it by sending messages.

In this first project of my Node.js Distributed Systems Lab, we will build that idea using only Node.js core APIs.

We will not use a framework, database, or message broker. We will use two small JavaScript files and the built-in worker_threads module.

What we will build

Our program has two parts:

The main thread starts a worker and sends commands.
The worker thread owns a counter and processes those commands.

The main thread will ask the worker to:

Increment the counter by 1.
Increment it by 4.
Return the current count.
Stop.

The important rule is that the main thread never modifies the counter directly. It can change the counter only by sending a message.

Project structure

01-message-passing/
├── main.js
└── worker.js

This example requires no npm dependencies. Node.js 20 or newer is enough.

The main thread

Create main.js:

const path = require("node:path");
const { Worker } = require("node:worker_threads");

const workerPath = path.join(__dirname, "worker.js");
const worker = new Worker(workerPath);

worker.on("message", (message) => {
  console.log("Main received:", message);
});

worker.on("error", (error) => {
  console.error("Worker error:", error);
});

worker.on("exit", (code) => {
  console.log(`Worker exited with code ${code}`);
});

worker.postMessage({
  type: "increment",
  amount: 1,
});

worker.postMessage({
  type: "increment",
  amount: 4,
});

worker.postMessage({
  type: "get-count",
});

worker.postMessage({
  type: "stop",
});

Let us look at the important parts.

const worker = new Worker(workerPath);

This starts worker.js in a separate worker thread.

worker.postMessage({
  type: "increment",
  amount: 1,
});

postMessage() does not directly call a function in worker.js. It sends a message to the worker's message port. The worker receives it asynchronously.

worker.on("message", (message) => {
  console.log("Main received:", message);
});

The worker can also send messages back. The main thread listens for those messages using the message event.

The worker thread

Create worker.js:

const { parentPort } = require("node:worker_threads");

let count = 0;

parentPort.on("message", (message) => {
  if (message.type === "increment") {
    count += message.amount;

    parentPort.postMessage({
      type: "count-changed",
      count: count,
    });
  }

  if (message.type === "get-count") {
    parentPort.postMessage({
      type: "current-count",
      count: count,
    });
  }

  if (message.type === "stop") {
    parentPort.close();
  }
});

The counter is declared inside the worker:

let count = 0;

The worker owns this state. The main thread does not import or share the variable.

The worker listens for messages from its parent:

parentPort.on("message", (message) => {
  // Handle the message.
});

When it receives an increment message, it updates its state and reports the new value:

count += message.amount;

parentPort.postMessage({
  type: "count-changed",
  count: count,
});

The object shapes form a small message protocol. Both sides must agree on what type, amount, and count mean.

Running the project

Run the main file:

node main.js

You should see:

Main received: { type: 'count-changed', count: 1 }
Main received: { type: 'count-changed', count: 5 }
Main received: { type: 'current-count', count: 5 }
Worker exited with code 0

The messages sent through the same port are handled in order. The worker first adds 1, then adds 4, and only then handles get-count.

What this example teaches

1. State ownership

Only the worker owns and modifies count. Other parts of the program must send messages to request a change.

This is easier to reason about than allowing several components to mutate the same state directly.

2. Asynchronous communication

The main thread sends a message and continues executing. The worker processes the message separately and may respond later.

This is the same mental model used by many queue-based and distributed systems, although the transport mechanism will be different.

3. Message protocols

The sender and receiver need a shared agreement about message structure.

For example:

{
  type: "increment",
  amount: 4
}

As a system grows, designing and versioning these message formats becomes an important engineering problem.

4. Isolation

Worker threads execute JavaScript separately. If we avoid shared memory, they can coordinate using messages instead of directly sharing mutable objects.

Two easy mistakes

While building this example, I encountered two simple mistakes that show why a clear protocol matters.

Mismatched property names produce `NaN`

If main.js sends this:

{
  type: "increment",
  amount: 4
}

but worker.js reads message.value, the value is undefined:

count += message.value;

Adding undefined to a number produces NaN. The property must have the same name on both sides.

`parentPort.stop()` does not exist

To close the worker's message port, use:

parentPort.close();

There is no parentPort.stop() method.

Is this a distributed system?

No—not yet.

The main thread and worker thread run inside one Node.js process on one machine. This project demonstrates the message-passing foundation without first adding networking, serialization, timeouts, or machine failures.

In later projects, we can keep the same basic model while changing the transport:

Worker thread messages
Messages between processes
TCP messages between machines
Task queues
Kafka events

Once communication crosses a network, we must handle new problems such as timeouts, duplicate messages, partial failures, retries, and unavailable nodes.

What comes next

This example works because we have only one caller and a small number of messages. But imagine sending several get-count requests concurrently. When a response arrives, how do we know which request it belongs to?

Project 02 will solve that problem using correlation IDs to build a proper request/reply protocol.

The complete roadmap and source code are available in the node-distributed-systems-lab repository: https://github.com/pckrishnadas88/node-distributed-systems-lab/tree/main

Building Distributed Systems with Elixir — 02: Correlated Request–Reply

krishnadaspc — Thu, 06 Aug 2026 18:00:34 +0000

In the previous example, we built a minimal stateful process using spawn, send, and receive.

A client could send a request:

{:get, self()}

and the server could reply:

{:counter_value, value}

That gives us basic request–reply communication.

But there is an important question we haven't answered yet:

If a process makes multiple requests, how do we know which reply belongs to which request?

This is the problem we'll explore in this example.

We will continue using raw Elixir process primitives without GenServer, Task, or other OTP abstractions.

The Problem

Every Elixir process has a PID.

A client can include its PID in a request:

send(server, {:request, self(), value})

The server can then use that PID to send the response:

send(from, {:reply, result})

For a simple request this works.

But a PID identifies a process, not an individual request.

The same process can make many requests during its lifetime.

Client #PID<0.95.0>

    request 10
    request 20
    request 30

All three requests come from the same process.

So we have two separate questions:

Where should the reply go?

Which request does the reply belong to?

A PID answers the first question.

We need something else to answer the second.

Creating a Request ID

Elixir provides make_ref/0.

ref = make_ref()

It creates a unique reference.

For example:

#Reference<0.2323213673.1614020613.213691>

Calling it again produces another reference.

ref1 = make_ref()
ref2 = make_ref()

These references can be used as request identifiers.

So instead of sending:

{:request, self(), value}

we send:

{:request, self(), ref, value}

Our protocol now contains both pieces of information:

self() -> where should the response go?

ref    -> which request is this?

The Server

The server is intentionally small.

defmodule Server do
  def start do
    spawn(fn -> loop() end)
  end

  defp loop do
    receive do
      {:request, from, ref, value} ->
        IO.puts("Server received request: #{value}")

        result = value * 2

        send(from, {:reply, ref, result})

        loop()

      :stop ->
        IO.puts("Server stopped")
        :ok
    end
  end
end

The important part is the protocol.

The server receives:

{:request, from, ref, value}

It processes the request:

result = value * 2

and returns the same reference:

send(from, {:reply, ref, result})

The server doesn't generate a new reference.

It simply returns the request identifier supplied by the client.

Conceptually:

Client                              Server

       {:request, pid, ref1, 10}
---------------------------------->

                                  10 * 2

          {:reply, ref1, 20}
<----------------------------------

The request and response are now correlated by ref1.

The Client

The client creates the reference.

defmodule Client do
  def request(server, value) do
    ref = make_ref()

    IO.puts("Client #{inspect(self())} sending #{value}")
    IO.puts("Request ref: #{inspect(ref)}")

    send(server, {:request, self(), ref, value})

    receive do
      {:reply, ^ref, result} ->
        IO.puts(
          "Client #{inspect(self())} received matching reply :#{result}"
        )

        {:ok, result}
    end
  end
end

There is one particularly important line here:

{:reply, ^ref, result}

Why ^ref instead of just ref?

Matching the Correct Reply

Before entering receive, the client already created a reference:

ref = make_ref()

Suppose conceptually its value is:

ref2

Now imagine the mailbox contains:

{:reply, ref1, 20}
{:reply, ref2, 40}
{:reply, ref3, 60}

The client is interested only in the response for ref2.

That's what the pin operator does:

receive do
  {:reply, ^ref, result} ->
    {:ok, result}
end

^ref means:

Match against the value already stored in ref.

So:

{:reply, ref1, 20}  -> doesn't match
{:reply, ref2, 40}  -> matches
{:reply, ref3, 60}  -> doesn't match

The process can select the matching message while unmatched messages remain in its mailbox.

This is selective receive.

PID vs Request Reference

This distinction was the most useful part of this example for me.

Consider three requests made by the same process:

Client #PID<0.95.0>

request 10 -> ref1
request 20 -> ref2
request 30 -> ref3

The PID doesn't change.

The request reference does.

So they represent different things:

PID = process identity

REF = request identity

Or from the server's perspective:

PID -> where do I send the reply?

REF -> which request am I replying to?

A process and an operation are not the same thing.

Running the Example

The project is kept deliberately small:

02-correlated-request-reply/
├── client.ex
├── server.ex
├── example.exs
└── README.md

The example starts the server and makes several requests.

server = Server.start()

IO.inspect(Client.request(server, 10))
IO.inspect(Client.request(server, 20))
IO.inspect(Client.request(server, 30))

send(server, :stop)

Run:

elixir example.exs

Example output:

Client #PID<0.95.0> sending 10
Request ref: #Reference<...>
Server received request: 10
Client #PID<0.95.0> received matching reply :20
{:ok, 20}

Client #PID<0.95.0> sending 20
Request ref: #Reference<...>
Server received request: 20
Client #PID<0.95.0> received matching reply :40
{:ok, 40}

Client #PID<0.95.0> sending 30
Request ref: #Reference<...>
Server received request: 30
Client #PID<0.95.0> received matching reply :60
{:ok, 60}

Server stopped

Something interesting is visible in this output.

The PID remains:

#PID<0.95.0>

while every request gets a different reference.

That's exactly what we wanted.

What About Multiple Client Processes?

BEAM processes are lightweight, so we can also have multiple clients communicating with the same server.

for value <- [40, 50, 60] do
  spawn(fn ->
    IO.inspect(Client.request(server, value))
  end)
end

Now the architecture looks more like:

Client A #PID<0.101.0> ---- refA ----\
                                      \
Client B #PID<0.102.0> ---- refB -----> Server
                                      /
Client C #PID<0.103.0> ---- refC ----/

Each client has its own PID.

Each request also has its own reference.

The server doesn't need to know anything about how those clients were created.

It understands only the protocol:

{:request, from, ref, value}

and:

{:reply, ref, result}

If Every Request Has Its Own Process, Do We Still Need a Request ID?

This was a question I had while building the example.

Suppose we create one process for every request.

Each process has a unique PID.

Couldn't the PID itself identify the request?

For a very simple one-request-per-process design, it sometimes could.

But that mixes two different concepts.

A process represents a unit of execution and isolation.

A reference identifies an operation.

Process   = WHO

Reference = WHICH REQUEST

A process may eventually have several operations in flight:

One Client Process

    ├── request ref1
    ├── request ref2
    ├── request ref3
    └── request ref4

If our protocol already contains request identifiers, it doesn't depend on the assumption that every request needs a new process.

Processes should be created when we want concurrency or isolation, not merely as a substitute for request IDs.

Why This Pattern Matters

The idea itself isn't specific to Elixir.

The general pattern is:

Request
   |
   +---- request_id
   |
   v
Server
   |
   +---- same request_id
   |
   v
Response

Whenever requests and responses are asynchronous, some form of correlation can be useful.

The identifier might be:

a BEAM reference
a UUID
an integer
a protocol-specific request ID

The implementation changes, but the underlying problem is the same:

Which response belongs to which request?

What We Have So Far

After the first two examples, our tiny protocol has evolved from:

Client
   |
   | message
   v
Process

into:

Client
   |
   | request + PID + request ID
   v
Server
   |
   | response + request ID
   v
Client

We now understand:

process mailboxes
send
receive
request–reply communication
make_ref/0
request IDs
matching replies
selective receive
concurrent clients

But there is still a much bigger problem.

What happens if the process we're communicating with dies?

Client
   |
   | request
   v
Server
   |
   X

How does another process know that it terminated?

That's where process monitoring comes in.

Next: Process Monitoring

In the next example I'll explore:

Process.monitor(pid)

and the :DOWN messages delivered when a monitored process terminates.

That will be the first step from basic message passing toward understanding how the BEAM detects and reacts to process failures.

The full source for this example is available in the repository: https://github.com/pckrishnadas88/elixir-distributed-systems-lab/tree/main/02-correlated-request-reply

Series

Building a Stateful Process in Elixir Without GenServer
Correlated Request/Reply
Process Monitoring
Process Linking

← Previous: Building a Stateful Process in Elixir Without GenServer

Next: Part 3 — Process Monitoring →

Content License

This article is licensed under CC BY-NC 4.0. You may share and adapt it with attribution for non-commercial purposes. The accompanying source code remains licensed under the MIT License.

1Million DB Updates, 3 Days, 200 Lines of Go: A Legacy Rescue Story

krishnadaspc — Sat, 01 Aug 2026 07:27:07 +0000

The Context

I wasn't part of the original team. I didn't know their codebase, their business logic, or their deployment process. I was brought in because the internal team had tried everything and the system was still failing daily.

A legacy PHP application served as the backend for a marketing dashboard. Business users would select a large segment of customers and click a button (e.g., "Send Campaign" or "Mark as Processed").

The Original Flow:

User clicks the dashboard button.
PHP receives the HTTP request.
PHP executes a SELECT * FROM users WHERE status = 'pending' (fetching up to 1 million rows).
PHP attempts to process these rows and update a status flag synchronously.
The HTTP request times out (30s), the database CPU spikes to 100%, and the system fails daily.

The application had no message queue and no asynchronous worker. The database crash caused partial updates, leading to duplicate actions on subsequent retries.

Why Re-Architecture Wasn't Possible

When you're staring at a burning production system, the ideal solution—a full rewrite, microservices, Kubernetes, event sourcing—is a luxury you simply don't have. Here is why we couldn't just "do it properly":

1. The System Was 8 Years Old

The PHP monolith had accumulated years of business logic, edge cases, and undocumented behavior. No one on the team fully understood every code path. A rewrite would have taken months and introduced new bugs.

2. No Automated Test Coverage

There were no integration tests, no unit tests, no staging environment that mirrored production. Any change to the core business logic was a gamble. We had to minimize the surface area of our changes.

3. The Business Couldn't Wait

The system was failing daily at 3 PM. Marketing campaigns were delayed, users were getting duplicate SMS, and support tickets were piling up. The business needed a fix now, not in 6 months.

4. Infrastructure Was Constrained

The database and application ran on the same 64GB server.
There was no budget for additional RDS instances, read replicas, or load balancers.
The hosting environment was locked down—installing new tools required weeks of approvals (if they were even allowed).

5. Team Skills Were Specialized

The existing team was PHP-heavy. Introducing a new language or a complex orchestration tool would have created a knowledge gap and maintenance burden. We needed a solution that was simple enough to hand over.

6. Risk Aversion

The company had been burned by previous "big bang" rewrites that failed. Management was highly risk-averse. Any solution had to be incremental, reversible, and low-risk.

The Bottom Line

We couldn't rewrite the monolith. We couldn't introduce Kubernetes. We couldn't wait for a greenfield project. We needed a minimally invasive, high-impact fix that could be deployed in days, not months.

The Fix: Decoupling the Request Lifecycle

To prevent the HTTP timeout and stabilize the database, we decoupled the request:

PHP (Producer): Accepts the request, selects the user IDs, publishes them to a queue, and returns 202 Accepted immediately.
RabbitMQ: Acts as a persistent buffer, absorbing the load spike.
Go (Consumer): A stateless, dedicated worker that consumes messages in batches and performs atomic database updates.

┌─────────────┐       ┌──────────────┐       ┌──────────────────────┐
│  Dashboard  │───HTTP─▶│ PHP Backend  │──────▶│    RabbitMQ          │
│  (User)     │◀──202──│ (Producer)   │       │    (Buffer)          │
└─────────────┘       └──────────────┘       └──────────┬───────────┘
                                                          │
                                                          ▼
                                                  ┌──────────────────────┐
                                                  │   Go Worker          │
                                                  │ (Batch Updater)      │
                                                  └──────────┬───────────┘
                                                             │
                                                             ▼
                                                     ┌──────────────┐
                                                     │  MySQL  │
                                                     │  (Stable)    │
                                                     └──────────────┘

Why this was the right minimal intervention:

PHP changes: Minimal (~50 lines to publish to RabbitMQ).
Go service: A single binary, easy to deploy, no dependencies.
RabbitMQ: Already available in the infrastructure (used for other purposes).
Rollback plan: If it failed, we could simply stop the Go service and revert the PHP changes in minutes.

The Initial Implementation (Static)

Here is the core logic that was deployed to stop the daily crashes. It is intentionally simple, stateless, and idempotent.

package main

import (
    "database/sql"
    "time"
)

func processBatch(messages []StatusMessage) error {
    // 1. Fixed batch size of 50,000 messages.
    batch := messages[:50000]

    // 2. Begin an atomic transaction. (ALL or NOTHING)
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // 3. Process updates within the transaction.
    for _, msg := range batch {
        // CRITICAL: The WHERE clause enforces idempotency.
        _, err := tx.Exec(`
            UPDATE users 
            SET status = 'processed' 
            WHERE id = $1 
            AND status = 'pending'  -- << Guard Rail
        `, msg.UserID)

        if err != nil {
            // Rollback the entire 50k batch on any error.
            return err
        }
    }

    // 4. Commit the batch atomically.
    if err := tx.Commit(); err != nil {
        return err
    }

    // 5. Acknowledge messages to RabbitMQ ONLY after commit.
    for _, msg := range batch {
        msg.Ack()
    }

    // 6. Static Backpressure: Sleep for 2 seconds.
    // Without this, the DB CPU saturates at 100%.
    time.Sleep(2 * time.Second)

    return nil
}

Why This Worked (The Principles)

1. Idempotency

RabbitMQ delivers at-least-once. The WHERE status = 'pending' ensures replaying the same batch is a no-op. Zero duplicate updates.

2. Atomicity

Wrapping 50k updates in a single transaction guarantees consistency. If the service crashes mid-batch, the DB rolls back, and the messages remain unacked for safe redelivery.

3. Backpressure

The static time.Sleep(2s) gave the MySQL wal_writer time to flush, dropping CPU from 100% to a stable 68%.

4. Separation of Concerns

The Go worker does not read the database and does not know about business logic. It is a pure "flag updater," making it highly resilient.

Performance Outcomes

Metric	Before (Sync)	After (Async Static)
HTTP Response Time	~30s (Timeout)	< 200ms (`202 Accepted`)
DB CPU	100% (Crash)	65-70% (Stable)
Failures	Daily	Zero failures (18 months)

Future Iterations: Moving from Static to Adaptive

The system worked perfectly for 18 months. However, if I were to build this today, I would move from a static configuration to an adaptive one.

Here is exactly how I would improve it for the next iteration:

1. Dynamic Backpressure (CPU-Aware Sleep)

Instead of a hard-coded time.Sleep(2s), we can query the database CPU utilization and adjust the throttling in real-time.

type AdaptiveConfig struct {
    TargetCPU      float64       // e.g., 0.7 (70%)
    MaxBatchSize   int           // 50000
    MinBatchSize   int           // 10000
    CurrentBatch   int           // Starts at 50000
    SleepDuration  time.Duration // Starts at 2s
}

func (a *AdaptiveConfig) Adjust() {
    cpu := getDBCPU() // Read pg_stat_database or /proc/stat

    if cpu > a.TargetCPU {
        // Back off: Smaller batch, longer sleep
        a.CurrentBatch = int(float64(a.CurrentBatch) * 0.9)
        if a.CurrentBatch < a.MinBatchSize {
            a.CurrentBatch = a.MinBatchSize
        }
        a.SleepDuration = time.Duration(float64(a.SleepDuration) * 1.2)
    } else if cpu < a.TargetCPU - 0.1 {
        // Speed up: Larger batch, shorter sleep
        a.CurrentBatch = int(float64(a.CurrentBatch) * 1.05)
        if a.CurrentBatch > a.MaxBatchSize {
            a.CurrentBatch = a.MaxBatchSize
        }
        a.SleepDuration = time.Duration(float64(a.SleepDuration) * 0.9)
    }
}

Why this matters: Peak vs. off-peak hours. The system becomes self-tuning—processing faster when the DB is idle, and slowing down automatically when it gets busy.

2. Sub-Batching Inside the Transaction

A 50k-row transaction holds locks for a long time. To reduce contention, we can break the work into smaller chunks within the same atomic transaction:

func processBatch(messages []StatusMessage) error {
    tx, _ := db.Begin()
    defer tx.Rollback()

    // Sub-batch of 1000 rows inside the 50k transaction
    for i := 0; i < len(messages); i += 1000 {
        end := min(i+1000, len(messages))
        chunk := messages[i:end]

        // Single statement for 1000 rows (faster than 1000 individual updates)
        tx.Exec(`
            UPDATE users 
            SET status = 'processed' 
            WHERE id = ANY($1::int[]) 
            AND status = 'pending'
        `, chunk.IDs)
    }

    tx.Commit() // All 50k still commit atomically
    // ... ACK messages ...
}

Result: Locks are held for 1000 rows at a time, but the atomicity of the overall 50k is preserved.

3. Observability (Prometheus Metrics)

Without metrics, we were flying blind. Next time, I would export:

var (
    batchDuration = prometheus.NewHistogram(...)  // How long per batch
    batchSize     = prometheus.NewGauge(...)      // Current batch size (dynamic)
    dbCPU         = prometheus.NewGauge(...)      // DB CPU %
    queueLag      = prometheus.NewGauge(...)      // Time since message creation
)

4. Graceful Shutdown

To avoid losing in-flight batches during deployments:

func main() {
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGTERM, syscall.SIGINT)

    go func() {
        <-sigChan
        log.Println("Shutting down gracefully...")
        consumer.Stop() // Stop pulling new messages
        worker.Wait()   // Finish current 50k batch
        os.Exit(0)
    }()
}

5. Dead Letter Queue (DLX)

If a message is malformed (e.g., missing user_id), retrying it forever is wasteful. Next time, I'd configure RabbitMQ to route bad messages to a DLX for manual inspection.

The Takeaway

The initial static solution was the perfect "firefighter" tool—it stopped daily outages in 3 days. It worked within the constraints of a legacy system, a risk-averse business, and a team that needed a simple handover.

Sometimes, being an outsider is an advantage. The internal team had been fighting this problem for weeks—they were too close to see the simple solution. I walked in, looked at the data flow, identified the bottleneck, and built a minimal intervention.

No heroics. No complex architecture. Just 200 lines of Go, a 50k batch limit, and a 2-second sleep.

Simplicity works for immediate rescue. Adaptability works for long-term scale.

Key Lessons

Don't let perfection be the enemy of stability. A simple fix deployed today beats a perfect solution delivered next year.
Understand your constraints. Re-architecting wasn't possible, but isolating the bottleneck was.
Idempotency is your best friend. When dealing with message queues, always design for at-least-once delivery.
Backpressure saves databases. Sometimes, slowing down is the fastest way to finish.
Build for handover. Your fix isn't done until someone else can maintain it.
Sometimes, the outsider sees the solution first. Familiarity breeds blindness. Fresh eyes can spot the obvious bottleneck that everyone else has been walking past for weeks.

Building Distributed Systems in Elixir: Part 1 - Building a Stateful Process in Elixir Without GenServer

krishnadaspc — Thu, 30 Jul 2026 18:12:12 +0000

Series: Distributed Systems from First Principles — Part 1

GenServer is one of the most widely used OTP behaviours in Elixir. It provides a convenient abstraction for building long-running stateful processes, but behind the abstraction lies a surprisingly small set of ideas.

At its core, a stateful server is simply:

A process
That owns private state
Receives messages
Updates its state
Sends replies
Waits for the next message

This article builds a minimal stateful counter using only:

spawn/1
send/2
receive
recursion

No GenServer, Agent, or other OTP behaviours are used.

The objective is not to replace OTP, but to understand the message-passing model it builds upon.

The Problem

Imagine multiple clients need access to a shared counter.

In many programming languages, the counter might simply be a mutable object.

Counter
-------
value = 0

Multiple threads can read or modify this value directly.

The BEAM follows a completely different design.

Processes never share memory.

Each process owns its own private state, and the only way to interact with another process is by sending it a message.

Instead of representing the counter as a shared variable, the counter itself becomes a process.

High-Level Design

The counter process owns the state.

Clients never access that state directly.

Instead, they communicate using a simple protocol.

                    +---------------------------+
                    |     Counter Process       |
                    |---------------------------|
                    | value = 0                 |
                    |                           |
                    | Mailbox                  |
                    +---------------------------+
                             ▲
                             │
                  Requests   │   Replies
                             │
                    +--------------------------+
                    |      Client Process      |
                    +--------------------------+

The communication protocol consists of three request messages.

{:get, caller_pid}
{:increment, caller_pid}
:stop

The counter responds with:

{:counter_value, value}

Notice that clients never call functions inside the counter process.

Instead, they exchange messages.

Complete Implementation

defmodule Counter do
  def start(initial_value \\ 0) do
    spawn(fn ->
      loop(initial_value)
    end)
  end

  def get(counter_pid) do
    send(counter_pid, {:get, self()})

    receive do
      {:counter_value, value} ->
        {:ok, value}
    after
      1000 ->
        {:error, :timeout}
    end
  end

  def increment(counter_pid) do
    send(counter_pid, {:increment, self()})

    receive do
      {:counter_value, value} ->
        {:ok, value}
    after
      1000 ->
        {:error, :timeout}
    end
  end

  def stop(counter_pid) do
    send(counter_pid, :stop)
  end

  defp loop(value) do
    receive do
      {:get, from_pid} ->
        send(from_pid, {:counter_value, value})
        loop(value)

      {:increment, from_pid} ->
        new_value = value + 1
        send(from_pid, {:counter_value, new_value})
        loop(new_value)

      :stop ->
        IO.puts("Counter stopped")

      unknown ->
        IO.inspect(unknown, label: "Unknown message")
        loop(value)
    end
  end
end

counter = Counter.start()

IO.inspect(Counter.get(counter), label: "Initial value")
IO.inspect(Counter.increment(counter), label: "After increment")
IO.inspect(Counter.increment(counter), label: "After second increment")
IO.inspect(Counter.get(counter), label: "Current value")

Counter.stop(counter)

Expected output:

Initial value: {:ok, 0}
After increment: {:ok, 1}
After second increment: {:ok, 2}
Current value: {:ok, 2}
Counter stopped

Understanding the Implementation

1. Starting the Server

def start(initial_value \\ 0) do
  spawn(fn ->
    loop(initial_value)
  end)
end

spawn/1 creates a new lightweight BEAM process.

That process immediately starts executing loop/1.

The value passed to loop/1 becomes the process's private state.

The returned PID is not the counter value.

It is simply the address used to communicate with the counter process.

2. Reading the Counter

The implementation of get/1 surprises many developers.

send(counter_pid, {:get, self()})

At first glance, it feels strange that a function called get() sends a message.

The reason is simple.

The client cannot access another process's memory.

Instead of reading a variable, it asks the process that owns the state.

The message

{:get, self()}

contains two pieces of information:

:get → the requested operation
self() → the PID of the caller

The caller PID is included because the counter needs to know where to send the reply.

Immediately after sending the request, the client waits.

receive do
  {:counter_value, value} ->
    {:ok, value}
end

This receive does not read the message that was just sent.

Instead, it waits for a new message created by the counter process.

3. Updating the Counter

Incrementing the value follows exactly the same request–reply pattern.

The client sends:

{:increment, caller_pid}

The counter updates its private state and sends the updated value back.

Notice that the client never performs the increment itself.

Only the counter process can modify its own state.

4. The Receive Loop

The receive loop is the heart of the server.

receive do
  {:get, from_pid} ->
    ...

  {:increment, from_pid} ->
    ...

  :stop ->
    ...
end

Each incoming message is pattern matched.

The atom (:get, :increment, :stop) identifies which operation should be performed.

Once the operation completes, the process calls loop/1 again.

loop(0)

↓

loop(1)

↓

loop(2)

The state is never mutated.

Instead, every new state is passed into the next iteration of the loop.

Message Flow

The interaction between the client and the counter looks like this.

Client Process                     Counter Process

{:get, caller_pid}
-------------------------------------->

                           Reads current value

                           {:counter_value, 0}

<--------------------------------------

{:increment, caller_pid}
-------------------------------------->

                           value = value + 1

                           {:counter_value, 1}

<--------------------------------------

Two messages are exchanged.

The client sends a request.

The counter sends a reply.

The client never reads the message it originally sent.

Why This Design?

Using message passing instead of shared memory provides an important guarantee.

Every request enters the process mailbox.

Mailbox

1. {:increment, pid1}
2. {:increment, pid2}
3. {:get, pid3}

The counter processes one message at a time.

No two clients can modify the state simultaneously.

The process itself serializes access to its private state.

This model avoids many of the synchronization problems commonly associated with shared mutable memory.

What Does GenServer Add?

This implementation already demonstrates the core behaviour of a stateful server.

GenServer builds on the same ideas while providing:

Standard request handling
Request correlation
Timeouts
Monitoring
Supervision
Debugging support
System messages
Consistent APIs

Understanding the raw implementation makes these abstractions much easier to appreciate.

Summary

This small example demonstrates several fundamental ideas behind the BEAM.

Processes own private state.
Processes communicate only through messages.
State is maintained through recursive receive loops.
A get() operation is a request–reply conversation, not a memory read.
A stateful server is simply a process implementing a message protocol.

Although intentionally minimal, this example forms the foundation for understanding GenServer, supervision trees, distributed messaging, and many other OTP concepts explored in later articles.

Source Code

The complete runnable example is available on GitHub:

Repository: https://github.com/pckrishnadas88/elixir-distributed-systems-lab/tree/main/01-process-mailbox

Series

Building a Stateful Process in Elixir Without GenServer
Correlated Request/Reply
Process Monitoring
Process Linking

Next: Part 2 — Correlated Request/Reply →

Content License

This article is licensed under CC BY-NC 4.0. You may share and adapt it with attribution for non-commercial purposes. The accompanying source code remains licensed under the MIT License.

Part 1: Taming Asynchronous JavaScript: How to Build a "Mailbox" Queue

krishnadaspc — Sat, 23 May 2026 10:13:18 +0000

Have you ever tried to catch water from a fire hydrant with a paper cup?

That is exactly what it feels like when you are building a JavaScript app and data starts coming in way faster than your code can process it. Maybe you are handling a flood of incoming webhooks, reading a massive file, or listening to a busy WebSocket.

If your data sender is faster than your data receiver, things break.

What you need is a waiting room. A place where incoming data can chill out until your app is ready to handle it. In computer science, this is called a "Queue" or a "Channel." Today, we are going to build one from scratch. Let's call it our Mailbox.

The Idea Behind the Mailbox
Think of a literal physical mailbox.

The Mail Carrier (The Producer): Drops letters into the box. They don't care if you are home; they just drop the mail and leave.

You (The Consumer): You check the mailbox. If there is mail, you take it. If the box is empty, you just wait until the mail carrier shows up.

We can recreate this exact relationship in JavaScript using Promises. Here is the code. It might look a little magical at first, but we will break it down right after!

export class Mailbox {
  constructor() {
    this.messages = []
    this.waiters = []
    this.closed = false
  }

  push(message) {
    if (this.closed) {
      throw new Error("Mailbox is closed")
    }

    // Deliver directly to waiting consumer
    if (this.waiters.length > 0) {
      const resolve = this.waiters.shift()
      resolve(message)
      return
    }

    this.messages.push(message)
  }

  async pop() {
    // Message already available
    if (this.messages.length > 0) {
      return this.messages.shift()
    }

    // Closed mailbox
    if (this.closed) {
      return null
    }

    // Wait for future message
    return new Promise((resolve) => {
      this.waiters.push(resolve)
    })
  }

  close() {
    this.closed = true

    // Wake all waiting consumers
    while (this.waiters.length > 0) {
      const resolve = this.waiters.shift()
      resolve(null)
    }
  }

  get size() {
    return this.messages.length
  }

  async *[Symbol.asyncIterator]() {
    while (true) {
      const msg = await this.pop()

      if (msg === null) {
        break
      }

      yield msg
    }
  }
}

The "Aha!" Moment: Pausing JavaScript

The coolest part of this code is inside the pop() method.

Usually, when we use a Promise in JavaScript, it's for something like fetch()

You make a request, and eventually, it resolves.

But here, we are doing something sneaky. If the mailbox is empty, we create a new Promise, but we take its resolve function and shove it into our this.waiters array. We are essentially bottling up the ability to finish the Promise for later.

Your code effectively pauses. It just sits there, waiting.

Then, when the push() method gets called, it looks inside this.waiters, pulls out that bottled-up resolve function, and triggers it with the new message. Boom! Your paused code instantly wakes up with the data.

How to Use It

Because we added that weird looking [Symbol.asyncIterator] at the bottom of the class, using our Mailbox is beautifully simple:

const mailbox = new Mailbox();

// 1. You: waiting for mail
async function readMail() {
  // This loop will naturally pause and wait for new messages!
  for await (const msg of mailbox) {
    console.log("Just got:", msg);
  }
  console.log("No more mail coming!");
}
readMail();

// 2. The Mail Carrier: dropping off mail at random times
mailbox.push("Letter 1");
setTimeout(() => mailbox.push("Letter 2"), 1000);
setTimeout(() => mailbox.close(), 2000);

This setup works flawlessly for everyday tasks. But there is a hidden monster in this code.

If you get too popular—say, someone drops 1,000,000 letters into your mailbox at once—this exact code will completely freeze your server for minutes. In Part 2, we are going to find out exactly why JavaScript hates huge arrays, and how to fix our Mailbox to handle millions of messages in a fraction of a second.

The link to the repo of full source code of this: https://github.com/pckrishnadas88/mailbox

How CSRF Token protects your web application

krishnadaspc — Mon, 20 Sep 2021 08:48:54 +0000

We are going to look how CSRF tokens works in a real node express application and how it protects the app from cross site request forgery. If you would like to see the video version of this post you can watch it here.
Watch video here

What is cross site request forgery?

When a web server is receives a request from a client without any mechanism for verifying that it was intentionally sent, then it might be possible for a hacker to send requests to the web server which will be treated as a genuine request. This can be done via a Form submit, URL, image load, XMLHttpRequest, etc. and can result in data breach or unintended code execution. In this post we are going to explain a form submit kind of attack and how CSRF tokens prevents this.

Code set up

There are two fresh express applications running. One is running on port 3000 which is App1 and another is running on port 5000 which is App2. Both of these applications are created using the express generator.

App1 Code snippets

Install the npm csurf

Enable the csrf middleware for the application in app.js

var csrf = require('csurf')
// setup route middlewares
app.use(cookieParser('fsgdesgsdYYFCCXXX'));
app.use(csrf({ cookie: true }))

Setting up the routes for the App1. Code from routes/index.js

var express = require('express');
var router = express.Router();

/* GET home page. */
router.get('/', function(req, res, next) {
  res.render('index', { title: "App1, CSRF Demo", csrfToken: req.csrfToken() })
});
router.post('/process', function (req, res) {
  res.send('csrf is valid, data is being processed')
})

In the above code we are generating the csrfToken and passing it to the view where we load our form with csrfToken: req.csrfToken()

In the view we use handlebars as our templating engine and csrf token is added as a hidden input field.

<h1>{{title}}</h1>
<p>Welcome to {{title}}</p>
<form action="/process" method="POST">
  <input type="hidden" name="_csrf" value="{{csrfToken}}">

  name: <input type="text" name="name">
  <button type="submit">Submit</button>
</form>

When we start the App1 we can see a form loaded with the generated csrf token if you check the html view source of the page.

and submit the form with some data. You can see this result as csrf token is validated and correct.

How token protects the app?

Now let's remove the token generation and see how this CSRF middle-ware protects our application. Change the code in app1/routes/index.js like this. Now we pass csrf as an empty string.

/* GET home page. */
router.get('/', function(req, res, next) {
  res.render('index', { title: "App1, CSRF Demo", csrfToken: '' })
});

Now if you submit the form you will get an error message like this.

Now server rejects the request as it cannot find a valid token in the request and now we are protected from such kind of attacks.

Attack request from App2

Now let's mimic an attack from another domain/application. In this example which is App2

For the time being just disable the csrf middleware in App1. Comment these lines in app1/app.js

// app.use(csrf({ cookie: true }))

In App2 home route we also have the same form but the form submit action is the url of App1

<h1>{{title}}</h1>
<p>Welcome to {{title}}</p>


<form action="http://localhost:3000/process" method="POST">

  name: <input type="text" name="name">
  <button type="submit">Submit</button>
</form>

Start the server App2 which is running on port 5000. You can see the form but now when you submit the form it is accepted as App1 currently do not have CSRF middleware enabled.

Now just re enable our csrf middleware in App1. Un comment this line

app.use(csrf({ cookie: true }))

And now when you submit the form from App2 you will get this error page only. So we are now protected.

That's how a CSRF token protects in a real application. You can find the full source code here Github Code Link

You can follow me on my twitter here KrishnadasPC

DEV Community: krishnadaspc

Building Distributed Systems in Elixir: Part 6 — Named Processes

The PID-Sharing Problem

Registering a Local Name

Sending to the Name

Name Scope Matters

A Global Name

Hiding the PID From Clients

Names Are Not Supervision

One Name Means One Owner

Correlating Replies

Limits of Global Registration

Series

Content License

Building Distributed Systems in Elixir: Part 5 — Supervisor From Scratch

The Problem

The Design

The Worker

Processing Work

Crashing Intentionally

Stopping Normally

Starting the Manual Supervisor

Tracking Child Identity

Starting Children

Detecting an Abnormal Exit

1. Identify the Logical Worker

2. Remove the Dead PID

3. Start a Replacement

4. Store the Replacement

The Restart Message Flow

Healthy Siblings Stay Alive

Normal Exit Is Not a Failure

Asking the Supervisor for Its Children

Replacement Means Fresh State

Stopping the Supervisor

The Complete Example

Running the Example

Why the Demo Polls for the New PID

What Makes This a Supervisor?

The Restart Loop Is Not Enough

What If a Worker Crashes Forever?

What If Children Depend on Each Other?

What If Shutdown Hangs?

What If Child Startup Fails?

In What Order Should Processes Start and Stop?

How Should Children Be Configured?

Why Use OTP Supervisor in Real Applications?

Limitations of Our Implementation

What We Built

What's Next?

Series

Content License

Building Distributed Systems in Elixir: Part 4 — Process Linking

The Problem

Creating a Linked Process

Exit Propagation

A Propagation Example

Normal and Abnormal Exits

Trapping Exits

A Trapped-Exit Example

Understanding the :EXIT Message

:EXIT

from_pid

reason

Why Use ^worker?

Running the Trap Example

The Complete Example

Links vs Monitors

Monitor notification

Trapped link exit

Links Are Not Supervision

What We Built

What's Next?

Series

Building Distributed Systems in Elixir: Part 3 — Process Monitoring

The Problem

Monitoring a Process

A Simple Worker

Detecting a Normal Exit

Understanding the :DOWN Message

Understanding the `:EXIT` Message

`:EXIT`

`from_pid`

`reason`

Why Use `^worker`?

Understanding the `:DOWN` Message

`:DOWN`

`ref`

`:process`

`pid`

`reason`

Why Use `^ref`?

3. Structured Logging with `log/slog`

Mismatched property names produce `NaN`

`parentPort.stop()` does not exist