Akka.Cluster.Sharding.Delivery: pass through options for bypassing Akka.Delivery and talking directly to entity actors

[Aaronontheweb]

So this was an issue raised on a support call with a customer today: if I start using Akka.Cluster.Sharding.Delivery, do I have to use it for 100% of my communication with entity actors? Or is it possible to talk to entity actors sometimes using Akka.Delivery's protocol and sometimes not?

Purpose of this thread is discuss:

• Is it a good idea to have a bypass for Akka.Delivery period? Why or why not?
• If it is a good idea, what's the easiest route for end-users to do so effectively?

Background

When you start a ShardRegion with Akka.Delivery enabled, it's done via code that looks like the following:

[Fact]
public async Task ReliableDelivery_with_Sharding_must_illustrate_Sharding_usage()
{
    await JoinCluster();
    NextId();

    var consumerEndProbe = CreateTestProbe();
    var region = await ClusterSharding.Get(Sys).StartAsync($"TestConsumer-{_idCount}", _ =>
            ShardingConsumerController.Create<Job>(c =>
                    PropsFor(DefaultConsumerDelay, 42, consumerEndProbe.Ref, c),
                ShardingConsumerController.Settings.Create(Sys)), ClusterShardingSettings.Create(Sys),
        HashCodeMessageExtractor.Create(10,
            o => string.Empty, o => o));

    var producerController =
        Sys.ActorOf(
            ShardingProducerController.Create<Job>(ProducerId, region, Option<Props>.None,
                ShardingProducerController.Settings.Create(Sys)), $"shardingController-{_idCount}");
    var producer = Sys.ActorOf(Props.Create(() => new TestShardingProducer(producerController)),
        $"producer-{_idCount}");

    // expecting 3 end messages, one for each entity: "entity-0", "entity-1", "entity-2"
    await consumerEndProbe.ReceiveNAsync(3, TimeSpan.FromSeconds(5)).ToListAsync();
}

This creates an actor hierarchy structure that looks like the following:

The Problem

The issue, in this case, is that all messages sent to the ShardRegion IActorRef for entity Id {x} will be received by the ShardingConsumerController<T>, and at present those messages will not be passed down to the user-defined entity actor, the "consumer" in this case.

Current Work-Arounds

The current work-arounds for this issue mostly rely on using DistributedPubSub and the Put + Send messaging mode, but that only works after the entity has been started by sending an intentionally "undeliverable" message to the ShardingConsumerController<T>. This is cantankerous and a bit of a code smell.

Proposed Solutions

1 - "Won't Fix; Just Use Akka.Delivery"

First solution to this problem is the path of least resistance: "yes, if you're using Akka.Delivery to talk to entity actors then you should just use it consistently." That's a valid suggestion, but I'm thinking there are cases where you really don't want to have to introduce the Akka.Delivery infrastructure where it hasn't been traditionally needed before just because you're using it in some places where it really is necessary.

2 - Introduce an Implicit Bypass inside the ShardingConsumerController<T>

One idea is to adjust the ShardingConsumerController<T> implementation to allow for an implicit bypass of the SequencedMessage<T> protocol, used for enforcing the ordered + guaranteed delivery of messages:

// inside ShardingConsumerControllerImpl.cs
private void Active()
{
    Receive<ConsumerController.SequencedMessage<T>>(seqMsg =>
    {
        if (ConsumerControllers.TryGetValue(seqMsg.ProducerId, out var consumerController))
        {
            consumerController.Tell(seqMsg);
            ProducerControllers = UpdatedProducerControllers(seqMsg.ProducerController, seqMsg.ProducerId);
        }
        else
        {
            _log.Debug("Starting ConsumerController for producerId {0}", seqMsg.ProducerId);
            var cc = Context.ActorOf(
                ConsumerController.Create<T>(Context, seqMsg.ProducerController.AsOption(),
                    Settings.ConsumerControllerSettings), Uri.EscapeDataString($"consumerController-{seqMsg.ProducerId}"));
            Context.Watch(cc);
            cc.Tell(new ConsumerController.Start<T>(_consumer));
            cc.Tell(seqMsg);
            ConsumerControllers = ConsumerControllers.Add(seqMsg.ProducerId, cc);
            ProducerControllers = UpdatedProducerControllers(seqMsg.ProducerController, seqMsg.ProducerId);
        }
    });

    Receive<Terminated>(t => t.ActorRef.Equals(_consumer), _ =>
    {
        _log.Debug("Consumer terminated.");
        Context.Stop(Self);
    });

    Receive<Terminated>(t =>
    {
        if (ProducerControllers.TryGetValue(t.ActorRef, out var producer))
        {
            _log.Debug("ProducerController for producerId [{0}] terminated.", producer);
            ProducerControllers = ProducerControllers.Remove(t.ActorRef);
            if (ConsumerControllers.TryGetValue(producer, out var consumerController))
            {
                consumerController.Tell(ConsumerController.DeliverThenStop<T>.Instance);
            }
        }
        else
        {
            var found = false;
            foreach (var kv in ConsumerControllers.Where(c =>
                         c.Value.Equals(t.ActorRef)))
            {
                found = true;
                _log.Debug("ConsumerController for producerId [{0}] terminated.", kv.Key);
                ConsumerControllers = ConsumerControllers.Remove(kv.Key);
            }

            if (!found)
            {
                _log.Debug("Unknown [{0}] terminated.", t.ActorRef);
            }
        }
    });

    /* NEW CODE */
   ReceiveAny(msg => _consumer.Forward(msg));
}

The trick here though, is that you'd want the messaging to go directly to the ShardRegion and not try to send this through the ShardingProducerController<T> side of Akka.Delivery. This means that ultimately there are going to be two distinct ways of communicating with entity actors present:

1. Using the Producer and SharidngProducerController<T> (the Akka.Delivery way) and
2. Going directly to the entity via the ShardRegion.

3 - Introduce an Explicit Bypass inside the ShardingProducerController<T> AND the ShardingConsumerController<T>

Rather than implicitly forward any messages not decorated with the SequencedMessage<T> protocol, we can retool the Akka.Delivery infrastructure to support an explicit bypass of the ordered delivery protocol, wherein messages not requiring acknowledgement can "jump the line" ahead of other ordered delivery messages:

// inside ShardingConsumerControllerImpl.cs
private void Active()
{
    Receive<ConsumerController.SequencedMessage<T>>(seqMsg =>
    {
        if (ConsumerControllers.TryGetValue(seqMsg.ProducerId, out var consumerController))
        {
            consumerController.Tell(seqMsg);
            ProducerControllers = UpdatedProducerControllers(seqMsg.ProducerController, seqMsg.ProducerId);
        }
        else
        {
            _log.Debug("Starting ConsumerController for producerId {0}", seqMsg.ProducerId);
            var cc = Context.ActorOf(
                ConsumerController.Create<T>(Context, seqMsg.ProducerController.AsOption(),
                    Settings.ConsumerControllerSettings), Uri.EscapeDataString($"consumerController-{seqMsg.ProducerId}"));
            Context.Watch(cc);
            cc.Tell(new ConsumerController.Start<T>(_consumer));
            cc.Tell(seqMsg);
            ConsumerControllers = ConsumerControllers.Add(seqMsg.ProducerId, cc);
            ProducerControllers = UpdatedProducerControllers(seqMsg.ProducerController, seqMsg.ProducerId);
        }
    });

    Receive<Terminated>(t => t.ActorRef.Equals(_consumer), _ =>
    {
        _log.Debug("Consumer terminated.");
        Context.Stop(Self);
    });

    Receive<Terminated>(t =>
    {
        if (ProducerControllers.TryGetValue(t.ActorRef, out var producer))
        {
            _log.Debug("ProducerController for producerId [{0}] terminated.", producer);
            ProducerControllers = ProducerControllers.Remove(t.ActorRef);
            if (ConsumerControllers.TryGetValue(producer, out var consumerController))
            {
                consumerController.Tell(ConsumerController.DeliverThenStop<T>.Instance);
            }
        }
        else
        {
            var found = false;
            foreach (var kv in ConsumerControllers.Where(c =>
                         c.Value.Equals(t.ActorRef)))
            {
                found = true;
                _log.Debug("ConsumerController for producerId [{0}] terminated.", kv.Key);
                ConsumerControllers = ConsumerControllers.Remove(kv.Key);
            }

            if (!found)
            {
                _log.Debug("Unknown [{0}] terminated.", t.ActorRef);
            }
        }
    });

    /* NEW CODE */
   Receive<ConsumerController.UnsequencedMessage>(msg => _consumer.Forward(msg.Payload));
}

Where this gets a bit tricky is if we added this bypass to the ProducerController<T> designs, as it would alter the backpressure requirements that the Producer (user-defined actor) currently has to adhere to.

Are there any other suggestions I'm currently missing?

[Aaronontheweb]

Resolved via #7106