Multi-broker solution

[ddrcode]

Status

Fairly good understanding of the behavior, but the implementation details are TBD.
This is the core missing functionality of Maiko. Crucial part of 0.4.0 release.
Don't implement until all points from to be decided section are answered.

Problem

Currently there is only a single broker, that means - all traffic goes through a single task. For larger amount of actors or for heavy traffic - this is a major limitation. However - current solution is implemented having multiple brokers in mind. Means, conceptually, singe broker is a special case of multi-broker architecture. Current solutions, like overflow policy, don't need to change.

Decisions (so far)

1. Topics can be organized in topic groups (one topic can belong to only one topic group).
2. Each topic group has exactly one broker.

The above guarantees event ordering withing a group, but not across the groups. In the most extreme case can be one group per topic.

To be decided.

Number of channels. Current channel arithmetic is: actor -> broker channel (stage 1) and broker -> actor channel (stage 2) - per actor. That gives 2A channels in total. With multi-broker it will be 2AB (2 * actors * brokers). For 10 actors and 4 topic groups that's 80 channels. Manageable, but not small, and it will be growing quickly. The question is whether we want to optimize it. if so, then the situation is complicating, because:

• we only know to which incoming topics an actor subscribes to,
• that means every actor can send any type of event (topic).

Three options here

1. one shared sender per Stage 1, shared across actors. For stage 2, as the point below
2. we optimize the number of stage 2 channels only - easier as we can do it statically during registration, as we know subscribed topics,
3. we optimze stage 1 and stage 2 by wiring actors with brokers dynamically - on first event

1 is the easiest, 3 the hardest. 2 can work in pair with any stage 1 solution.

For option 3 - we could reuse our command channel. Send request for new channel, specific broker picks it, creates new reader, and sends command with sender. That means first event send is always bit delayed.

Alternative - producers declare what type of events they produce.

Thoughts?

[vnermolaev]

I'd love to comment but we need to set the terminology straight.
Previously you referred to Stage 1 and Stage 2 as reception of the first event and as draining of the channel in ActorController (#99), respectively. Here, it seems, you refere to something else. Can you pls elaborate?

[ddrcode]

Hi, thanks for the question.

Yes the "stage" term in #99 was "reused" for that specific scenario. Here the stages mean something different.

In helicopter view - we always have a broker actively participating in message delivery. It must be like that because by design: a) actors are decoupled, b) they subscribe to topics. In reality we often have one-to-many message transport (if many actors subscribe to the same topic), so that also requires the intermediary (the broker).

In that scenario, from event producer perspective:
Stage 1: Actor (producer) -> Broker
Stage 2: Broker -> Actor(s) (consumers)

Right now the situation is simple: each actor has its own rx and its own tx with a broker (pre Maiko 0.2.5 there was only one Sender for stage 1 - cloned across all actors). For one broker that's manageable number of channels, and the most optimal solution.

So in this Github issue I am debating about any potential form of optimisation (in terms of number of channels involved). Let's treat the Stage 1 and Stage 2 cases separately.

And one important assumption (reminder): in the multi-broker scenario, brokers are split by topics (or topic groups). Two brokers will never be "handling" the same topic. That's a must - as we need to guarantee events ordering within a topic.

Stage 2 can be easily optimised. Each broker knows which topics it deals with, and each actors knows to which topics they subscribe to. We don't need Stage 2 channels between broker that handles i.e. Telemetry topic if an actor has not subscribed into that topic. Simple. The topology of Stage 2 channels, reflects subscriptions topology.

So the optimisation question is purely about stage 1. Any actor can send any type of event. Actors don't "declare" what type of data they will be sending. Subscriptions are for consumers, not for producers. If so, considering that we have Na actors and Nb brokers:

1. Should we have Na*Nb Stage 1 channels? Each actor can send to any broker?
2. Should we have Nb channels - one per broker, shared across actors?
3. Dynamic. Stage 1 connection is created only on the first attempt of sending to a particular broker.
4. Something else?

Main drawbacks
1 - risk of multiple inactive channels
2 - risk of Stage 1 congestions in case of busy producers
3 - delay for first-time event

Development complexity
1 - easy; we expand existing solution
2 - trivial; one sender per broker, cloned across actors
3 - new logic, dynamic channel creation; highest potential for further optimisations

Hope that explains the problem better

[vnermolaev]

Understood.

I intentionally exclude dynamic channel wiring (former Design 3) from this iteration.

Model

Let:

• Na = number of actors
• Nb = number of brokers
• λ = ingress rate to a broker
• μ = processing rate of that broker

If λ $\leq$ μ, the system is stable.
If λ > μ, congestion is unavoidable regardless of channel topology

• Design 1: all send to a single broker, the queue is congested immediately;
• Design 2: single actor will eventually fill up respective; every second the queue of a broker will grow by approx λ - μ events eventually getting congested.

Proposal

Adopt one ingress channel per broker (shared sender cloned across actors), and add per-actor throttling at send time (before enqueue). That is, shared broker ingress + actor-local throttling

Why

• Avoids Na * Nb channel growth.
• Preserves a simple static topology.
• Prevents a single noisy actor from monopolizing broker ingress.
• Improves fairness/isolation without pretending to increase broker throughput.

Impl

#[derive(Clone)]
struct ActorTx<M> {
    tx: mpsc::Sender<M>,
    bucket: Bucket, // token-bucket or leaky-bucket record, depends on our needs, can be generic.
}

impl ActorTx<M> {
fn try_acquire_tx(&self) -> Option<mpsc::Sender<M>> {
  if self.bucket.try_acquire() {
    Some(self.tx.clone())
  } else { 
    None 
  }
}
}

Bucket ultimately decides whether we can access the channel.

Hot path becomes:

if let Some(tx) = ActorTx::try_acquire() {
  tx.send(...)
}

Expected outcome

• Channel count scales with brokers (Nb), not with Na * Nb.
• Congestion control is applied where it is most effective (before enqueue).
• Sustained overload is handled explicitly via throttle policy + existing overflow/drop behavior.

[ddrcode]

I like the direction this is taking. Na*Nb is wasteful, but dynamic wiring adds much complexity. It may be a goal for future development. As such - shared channel per broker feels like a natural progression.

So we have two new concepts here

1. ActortTx - shared sender between multiple actors and a specific broker (Stage 1)
2. EventRouter - as per EventRouter #109. It will hold all ActorTx instances required for Actor->Broker pair.

From actor (producer) perspective - currently the only interface to send events is via Context: ctx.send(my_event).await?;. Please notice there is no try_send alternative atm (see design decisions for explanation). So the issue I see here, we have a blocking interface on producer side, and non-bloking try_acquire below. Whichever way we unify it, it will have design implications.

Overall I like your queuing theory-based approach. We guarantee some form of fairness without multiplying number of channels madly.

---

Just, as alternative, consider for a moment option 3, It could look as follow.

• there is one maintenance (or command) channel between all actors and a broker (MPSC) - similarly to your model,
• when actor decides to send a new event, it sends something like RequestChannel(event, oneshot)
• broker receives the event and pushes it forward, as always, but also:
• it creates a new channel and sends the sender back as a oneshot response
• next time actor decides to send an event from the same topic - it has a dedicated channel in the EventRouter

We gain no lazy-send for the first message. We risk huge number of channels if all actors subscribe to all topics. Simple logic though - no internal throttling. Not crazy difficult to implement.

---

I have zero preference for any solution. Let's discuss.

Perhaps the first question we should ask - how many channels are too many? Channels are cheap. Sender is an Arc with some atomics. In the first comment I suggested that we should aim to optimize the number of channels, but perhaps this is not the most important optimization here? Thoughts?

[ddrcode]

Ok - I must admit my earlier focus on channel optimization was misleading. Let’s step back and look at this from the architectural perspective.

The current solution was designed with multi-broker in mind, treating a single broker as a special case. We are not expanding toward multi-broker; we are moving from the special case to the generic model.

Out of the three options above, only one feels right architecturally:

• Option 1: Na * Nb is topologically complex, noisy, and feels wrong by introducing many channels that would remain unused in practice.
• Option 3: dynamic channels feels like optimization upfront, not architecture. It may become relevant later if the natural solution proves inefficient.
• Option 2, as @vnermolaev pointed out, is the natural next step.

For the first iteration, I would also hold off on throttling. That already belongs in optimization space. Let’s introduce the simplest solution first, measure it, and decide the next step based on evidence.

In practice, the current architecture combined with multiple brokers already gives us strong control over congestion.

Let’s take the weather station example:

We have three classes of events there, which can be naturally grouped into topics:

• data: sensor readings (temperature, pressure). We cannot afford to lose them. Throughput may be high.
• control: e.g. sending alerts. These events cannot be lost, but occur rarely.
• observability: logs, telemetry. These may be dropped under congestion.

Data and control would use OverflowPolicy::Block, while observability would use OverflowPolicy::Drop.

Observability can share a single broker for all monitoring-style topics. That is sufficient.
Control-related topics may use another broker, since control events occur relatively rarely compared to data.
Each busy data stream could potentially have its own topic and its own broker, so temperature readings would not congest pressure readings, etc.

In the example above, without introducing any additional concepts, we are already modeling:

• data flow
• backpressure
• overflow policy
• system behavior under load

We are simply expanding from the degenerated system (single broker) into the architecture it was always meant to have.

Once viewed that way, the question of channel count becomes secondary. And a single shared sender per broker feels sufficient as the starting point.

If, in the design above, we chose a single data broker serving multiple busy sensor streams, then throttling might become necessary. But that would be a consequence of suboptimal topic-to-broker organization, not a fundamental limitation of the model itself.

My call: introduce multi-broker with a single shared Sender per broker in its simplest form. Measure. Optimize later if needed. That feels architecturally right to me. Does that make sense?

[vnermolaev]

Well, that's a lot to digest :)

I think we are very close conceptually, but there is an important distinction:

1. OverflowPolicy is channel-capacity driven and reactive.
2. Throttling is actor-level and proactive.

OverflowPolicy is currently used when sending Broker -> Actor.
There is always a dedicated channel from broker to actor, with actor holding the receiver part.
Importantly, brokers do not typically create congestion by themselves, they mostly relay messages.
So if we want tight control over system pressure, we need to control the opposite direction: Actor -> Broker. That is exactly this discussion.

I agree, a natural first step is to distribute each broker sender to all relevant actors (shared sender per broker), which aligns with the architecture direction.
However, I do not think using OverflowPolicy to control this direction is valid, because it is applied at the level of the full channel (capacity), and one chatty actor with blocking behavior can stall the system significantly. As such, applying OverflowPolicy to Actor -> Broker does not guarantee fairness, which is why I introduced throttling.

Throttling can be adjusted to match your architectural expectations: blocking per bucket, not per full channel, preserving fairness. E.g.:

impl ActorTx<M> {
  async fn send(&self, data: M) -> crate::Result<()> {
    while let BucketAcquire::Wait(duration) = self.bucket.try_acquire() {
      sleep(duration).await;
    }
    self.tx.send(data).await?;
    Ok(())
  }
}

My conclusion

1. Proceed with distributing broker sender parts to all concerned actors.
2. Do not use OverflowPolicy in Actor -> Broker direction based on the channel capacity (as in Broker -> Actor).
3. Apply improved per-actor throttling if required, and optionally map throttling decisions to OverflowPolicy semantics in Actor -> Broker:
   • trigger point: failure to acquire permission from the token bucket,
   • then apply policy-like behavior (Drop / Block / Fail) at that point, instead of waiting for the shared channel to become full.