Three nodes, one Dad: messaging and shard placement in Effect and Pekko

In part 1 every chat got its own Dad, and the race that made the naive webhook repeat jokes was gone. But it all ran on one node, so “cluster” was a promise, not a fact.

This post follows Effect’s part 2. Dad learns some manners: he waits until the chat has been quiet for 2.5 seconds, then makes one joke about everything that was said. The cluster grows to three nodes, which raises the question this series is really about: who decides where a Dad lives? Effect and Pekko answer it in completely different ways.

Dad waits for the chat to go quiet

Effect’s Dad turns NewMessage into a queue offer. A loop forked into the entity’s scope takes the first message, keeps taking until 2.5 seconds pass without one, and then tells one pun about the batch. The Pekko Dad does the same thing as a state machine with three states:

export const DadLayer = Dad.toLayer(
  Effect.gen(function* () {
    const address = yield* Entity.CurrentAddress
    const punGenerator = yield* PunGenerator
    const chat = yield* Chat
    const told = new Set<string>()
    const unread = yield* Queue.unbounded<{ from: string; text: string }>()

    // Wait for a message, then keep reading until the chat goes quiet
    const nextBatch = Effect.gen(function* () {
      const batch = [yield* Queue.take(unread)]
      while (true) {
        const next = yield* Queue.take(unread).pipe(Effect.timeoutOption("2500 millis"))
        if (Option.isNone(next)) return batch
        batch.push(next.value)
      }
    })

    yield* Effect.gen(function* () {
      const messages = yield* nextBatch
      const joke = yield* punGenerator.generate(messages, { told })
      told.add(joke)
      yield* chat.reply(address.entityId, joke)
    }).pipe(Effect.forever, Effect.forkScoped)

    return {
      NewMessage: ({ payload }) => Queue.offer(unread, payload).pipe(Effect.asVoid),
    }
  }),
)

object Dad:
  sealed trait Command
  final case class NewMessage(from: String, text: String) extends Command with CborSerializable
  private case object Quiet extends Command with NoSerializationVerificationNeeded
  private final case class JokeReady(joke: String) extends Command with NoSerializationVerificationNeeded
  private final case class JokeFailed(reason: String) extends Command with NoSerializationVerificationNeeded

private final class Dad(chatId: String, puns: PunGenerator, chat: Chat, quietPeriod: FiniteDuration,
    ctx: ActorContext[Dad.Command], timers: TimerScheduler[Dad.Command]):

  // Same key: every message restarts the quiet period.
  private def waitForQuiet(): Unit = timers.startSingleTimer(Quiet, Quiet, quietPeriod)

  def idle(told: Set[String]): Behavior[Command] =
    Behaviors.receiveMessagePartial { case NewMessage(from, text) =>
      waitForQuiet()
      collecting(told, Vector(ChatMessage(from, text)))
    }

  private def collecting(told: Set[String], batch: Vector[ChatMessage]): Behavior[Command] =
    Behaviors.receiveMessagePartial {
      case NewMessage(from, text) =>
        waitForQuiet()
        collecting(told, batch :+ ChatMessage(from, text))
      case Quiet =>
        ctx.pipeToSelf(puns.generate(batch, told)) {
          case Success(joke)  => JokeReady(joke)
          case Failure(cause) => JokeFailed(cause.getMessage)
        }
        thinking(told, Vector.empty)
    }

  private def thinking(told: Set[String], unread: Vector[ChatMessage]): Behavior[Command] =
    Behaviors.receiveMessagePartial {
      case NewMessage(from, text) => thinking(told, unread :+ ChatMessage(from, text))
      case JokeReady(joke) =>
        chat.reply(chatId, joke)
        next(told + joke, unread)
      case JokeFailed(reason) =>
        ctx.log.warn("Dad for {} lost his train of thought: {}", chatId, reason)
        next(told, unread)
    }

  private def next(told: Set[String], unread: Vector[ChatMessage]): Behavior[Command] =
    if unread.isEmpty then idle(told)
    else
      waitForQuiet()
      collecting(told, unread)

Effect’s background loop is a fiber. In Pekko there is no loop: the “waiting” is a timer, and each state is the behavior for the next message.

• idle: the first message starts the quiet timer.
• collecting: every message restarts it. Starting a timer with the same key cancels the old one, and Pekko guarantees that the old Quiet is never delivered, even if it was already in the mailbox.
• thinking: the pun is on its way through pipeToSelf. Messages that arrive now become the next batch, like messages that pile up in Effect’s unread queue while the loop is busy.

The stash from part 1 is gone. Part 1 said that Pekko makes you choose what to do with messages that arrive while Dad is busy: stash them, drop them, or collect them. Part 2 collects them.

The run below starts three nodes in one JVM and sends each message through a different node, as if a load balancer had picked it:

$ sbt part2-messaging/run
14:15:55.218  #engineering · Sebastian: Why is this test flaky?  (via node-0)
14:15:55.435  Dad for #engineering has entered the chat on node-0
14:15:56.235  #engineering · Mattia: It's flaky for me too  (via node-1)
14:15:56.240  #random · Tim: Is CI down?  (via node-2)
14:15:56.701  Dad for #random has entered the chat on node-1
14:16:01.279  #engineering · DadBot: I'd tell you a UDP joke, but you might not get it.
14:16:01.279  #random · DadBot: I'd tell you a UDP joke, but you might not get it.
14:16:01.367  #engineering → shard 320 → node-0
14:16:01.377  #random → shard 878 → node-1

Sebastian and Mattia get one joke between them. It comes about five seconds after Mattia’s message: 2.5 seconds of quiet, plus the two seconds the pun generator takes to think. That’s the 4.5-second gap in Effect’s output too. The extra half second here is routing. Mattia’s message entered through node-1, which didn’t know yet where shard 320 lived and had to ask (more on that below).

The protocol crosses the wire

With one node, NewMessage never left the JVM. Now it does, so it needs a serializer. In Effect the protocol is a schema: Rpc.make("NewMessage", { payload: { from: Schema.String, text: Schema.String } }) is serializable by construction. In Pekko a message is a plain case class, and serialization is configured separately. You bind a marker trait to a serializer:

// Nothing to add: Schema is the protocol.

pekko.actor.serialization-bindings {
  "dadbot.part2.CborSerializable" = jackson-cbor
}

Only NewMessage extends CborSerializable. Dad’s private messages extend NoSerializationVerificationNeeded instead, which says that they never leave the node. The cluster test runs with pekko.actor.serialize-messages = on, which serializes every message even locally, so a message that couldn’t cross the network fails a test instead of production.

That’s more ceremony than Effect, and it’s a real trade-off. You get to choose: Jackson for convenience, Protobuf for schema evolution, a custom serializer for speed. You also have to remember to choose. Without a binding, Pekko refuses to send the message rather than fall back to Java serialization.

Where does #engineering’s Dad live?

Both frameworks first map the chat to a shard. That part is easy:

Both warn you not to change the shard count later, because every entity would move. The interesting question is the next step: which node owns shard 320?

Effect: every runner computes the answer. Runners register in a shared runner table. Each one reads the table and puts runners and shards on the same hash ring, with 128 virtual nodes per runner and bounded loads. Every runner comes to the same conclusion without asking anyone. Because runners can read the table at slightly different moments, a runner must hold a shard’s lock in the database before it starts that shard’s entities. There is no leader, but there is a database.

Pekko: one coordinator decides. The ShardCoordinator is a cluster singleton on the oldest node. When a node’s ShardRegion sees a message for a shard it doesn’t know, it asks the coordinator once (GetShardHome) and caches the answer. That’s the half second Mattia’s message spent above. The coordinator’s state is replicated to the other nodes with Distributed Data (state-store-mode = ddata), so a new coordinator can take over when the oldest node leaves. There is no database, but there is a leader.

How Pekko places shards: LeastShardAllocationStrategy

Pekko has no hash ring for shards. The default and only built-in allocation strategy is LeastShardAllocationStrategy, plus ExternalShardAllocationStrategy if your own code wants to decide. Akka added a ConsistentHashingShardAllocationStrategy in 2023, after Pekko forked from Akka 2.6. It’s meant for keeping the same shard id of different entity types together, not for the even spreading Effect’s ring does. ShardAllocationStrategy is a public interface, so you could write a ring yourself. Here’s what you get out of the box:

• Shards are allocated lazily. A shard gets a home only when its first message arrives, so the strategy only ever deals with shards that are in use. Effect’s ring assigns all 300 shards up front, whether any Dad lives there or not.
• A new shard goes to the node with the fewest shards. Before counting, it prefers nodes that aren’t on their way out (Leaving, Exiting, Down) and nodes with the newest appVersion, so a rolling update moves work towards new code. There is no hashing, so the same chat can land on a different node in the next run. Only id → shard is stable.
• Rebalancing is a periodic, throttled decision. Every rebalance-interval (10 s), the coordinator asks the strategy which shards to move. It skips the round while nodes are joining, unreachable, or running mixed app versions.

The rebalance algorithm has a trap. The newer algorithm computes the optimal count per node, ceil(shards / nodes). It moves the excess away from nodes above it, then evens out nodes two or more below it, with at most min(rebalance-absolute-limit, 10% of shards) per round. But it’s only enabled when rebalance-absolute-limit > 0, and the default is 0. With the default you get the legacy algorithm: rebalance-threshold = 1 and max-simultaneous-rebalance = 3, which move one shard per round. If a fourth node joins three nodes with 300 active shards, about 75 shards should move, and that takes 75 rounds, more than 12 minutes. reference.conf recommends the new algorithm, and DadBot turns it on:

# 0 (the default) silently selects the legacy algorithm, which moves one shard per round.
pekko.cluster.sharding.least-shard-allocation-strategy.rebalance-absolute-limit = 20

Every node is an edge

As in Effect’s diagram, every node runs the same thing: the webhook, the sharding region, and Pekko Management. A load balancer can send Slack’s request to any of them:

yield* makeDad(chat).NewMessage({ from, text }, { discard: true })

// Runs on every node.
path("chats" / Segment / "messages") { chatId =>
  post {
    entity(as[Message]) { message =>
      sharding.entityRefFor(Dad.TypeKey, chatId) ! Dad.NewMessage(message.from, message.text)
      complete(StatusCodes.Accepted)
    }
  }
}

The webhook answers 202 Accepted without waiting for Dad. ! is Pekko’s equivalent of { discard: true }. If shard 320 lives on another node, the local region forwards the message over remoting.

In the three-node test, Sebastian, Mattia and Tim each posted through a different node, one after another. Dad got all three in one batch, but not in that order. Pekko guarantees ordering per sender–receiver pair, and these messages came from three different regions, each of which had to learn where shard 320 lives first. With several entry points, nothing orders messages across them. If the order of a chat matters, the chat has to provide it, for example with Slack’s message timestamps.

Every node being an edge isn’t the only option. With node roles, Entity(Dad.TypeKey)(...).withRole("dad") starts Dads only on nodes with the dad role. A gateway node without the role still calls ClusterSharding.init, but gets a proxy-only region that routes messages and never hosts a Dad. That separates the HTTP edge from the workers, so they scale and deploy independently, at the cost of an extra process and an extra hop. Effect’s equivalent is a runner in client-only mode (runnerAddress: Option.none() in ShardingConfig): it sends to entities but never hosts them.

Moving a shard

When a node leaves gracefully, its shards have to move. Here the run above continues: node-0 hosts shard 320, and it leaves the cluster.

14:16:01.381  --- node-0 leaves the cluster ---
14:16:02.626  #engineering · Sebastian: Still flaky?  (via node-1)
14:16:02.957  Dad for #engineering has entered the chat on node-2
14:16:07.499  #engineering · DadBot: I'd tell you a UDP joke, but you might not get it.
Repeated joke in #engineering!

Node-0 was also the oldest node, so the coordinator moved too, and it all took about a second. The handoff is a protocol run by the coordinator:

1. BeginHandOff to every region. Each forgets where the shard lives and acknowledges. From now on, messages for that shard are buffered in the sending region (up to buffer-size, 100,000) instead of sent.
2. HandOff to the hosting region, which stops the shard’s entities.
3. ShardStopped back to the coordinator, which allocates the shard again. The regions flush their buffers to the new home.

Effect gets the same guarantee, never two Dads, from the shard lock. The old runner stops its entities and releases the lock, the new owner acquires it, and senders retry in the meantime. Pekko has no locks, because the coordinator is the only one deciding.

Dad moved, and he forgot everything. His history lived in memory on node-0, so the new Dad on node-2 tells the UDP joke again. Effect’s post ends on the same note: state that survives a move is part 3.

When a node dies

A graceful leave is the easy case. For the hard case, I ran the three nodes as separate JVMs (scripts/node.sh 0|1|2) and killed the one hosting #random with taskkill /F. Pekko Management showed where the shards were:

$ curl localhost:8551/cluster/shards/Dad
{"regions":[{"numEntities":1,"shardId":"878"}]}

Then node-1 died, and Tim posted to #random through node-2:

14:30:07.618  node-1 killed
14:30:07.670  POST :8082/chats/%23random/messages → 202
14:30:11.503  node-0: Marking node as UNREACHABLE [127.0.0.1:25521]
14:30:12.516  node-2: Marking node as UNREACHABLE [127.0.0.1:25521]
14:30:31.812  SBR took decision DownUnreachable and is downing [127.0.0.1:25521], [1] unreachable of [3] members
              ...Tim never gets an answer
14:32:08.510  POST :8082/chats/%23random/messages → 202
14:32:08.596  Dad for #random has entered the chat on node-2
14:32:13.144  #random · DadBot: I'd tell you a UDP joke, but you might not get it.

About four seconds pass before the failure detector marks node-1 unreachable. Then the SBR waits stable-after (20 s) for the membership to settle and downs it. Only then is shard 878 free to move. The next message starts a fresh Dad on node-2, and he tells the UDP joke again.

Tim’s first message is gone. Node-2’s region had cached “shard 878 is on node-1” and handed the message to remoting, and remoting dropped it when node-1 was quarantined. I repeated the experiment and posted eight seconds after the kill, when the node was already marked unreachable but not yet downed. That message was lost too. Pekko sharding delivers at most once: for roughly 25 seconds, every message to a dead node’s shards is lost.

Why the 20 seconds? Without a database, the only way to know that a node is really dead and not just partitioned is for the cluster to agree on it. If the SBR downed nodes too quickly, a network blip could leave two halves of the cluster each running their own #random Dad. Effect avoids that with the lock in the database: a lock expires unless its owner keeps refreshing it, so a runner that can’t reach the database loses its shards. Both frameworks make you wait. They differ in what you wait for: a quorum of peers, or a lease in a database.

Trade-offs in part 2

What’s next

Dad now runs on three nodes. He handles a graceful move in about a second and comes back after a crash. But twice in this post he forgot what he’d already told the chat, and once a message simply disappeared. Part 3 follows Effect’s post on delivery: Effect’s persisted messages and message storage, set against Pekko’s reliable delivery controllers and event-sourced entities.

The code for this part is in part2-messaging. sbt part2-messaging/run plays the scripted run. scripts/node.sh 0|1|2 starts real nodes to kill.