`DistributedPubSub`: add a `wait-for-subscribers` value to `Publish` commands to delay publishing / discarding until subscribers are available

[Aaronontheweb]

Is your feature request related to a problem? Please describe.

A common problem that occurs with DistributedPubSub is when both the publisher and the subscriber start at essentially the same time, but on different nodes, and it takes a moment for that subscription to replicate. You end up with an immediate DeadLetter per #7626 and the publisher is never notified that this happened unless the end-user implements their own explicit ACK-reply system from all subscribers.

This makes DistributedPubSub kind of unwieldy to use in cases that require a higher degree of consistency.

Describe the solution you'd like

I think it would be worth doing two things:

1. Create a new PublishWithAck message that sends a reply back to the publisher, letting them know that the message was sent to N nodes with subscribers. That way the publisher can get feedback on whether or not their message was actually delivered to someone without having to write infrastructure to do this themselves in the subscriber actors.
2. Create a new WaitForSubscribers value - we shouldn't let the user do stuff like specifying the number of subscribers because users will 100% interpret that to mean "the number of subscribers" and not the "number of nodes with subscribers," the latter being what we actually can measure. This should also have a timeout property to prevent memory build up and the DistributedPubSub should have a max-buffer-size setting for the number of "waiting" messages it will allow per-topic.

Describe alternatives you've considered

The alternatives are forcing the user to do all of this themselves via the queries we added to support #3663 - which is very high-boilerplate, janky, and generally not good.

[jasonshave]

@Aaronontheweb is this the ack-reply 'pattern' that would work in the meantime?

• Subscribe in the constructor (or lifecycle startup of the actor)
• Create a message handler to receive the SubscribeAck and put a behavior switch in the handler delegate (i.e. Become(Ready))
• Stash all messages until ack is processed and behavior is switched

This basically puts the actor into a pseudo-waiting state on the subscription confirmation before starting to process messages. I suppose it could have an impact on the performance of the actor as a consequence of waiting for confirmation + stashing depending on how long it takes to get the ack.

[Aaronontheweb]

@jasonshave you'd also want to make sure that you send an ACK message to the Sender each time you get a message that you believe was sent via DistributedPubSub - it's that portion you'd really need to make this workable at publish-time.

[Arkatufus]

Ok, from the discussion with @Aaronontheweb, this is how this feature supposed to be implemented.

this feature only gets used if the user uses the PublishWithAck wrapper envelope to send their message. When this happens:

• The mediator will try to send the message
• If this fails, the mediator will
  • Try and create a topic actor (may or may not available)
  • Send the message to the topic actor
• The topic actor will buffer the message until a Subscribe message comes in
• The topic actor then forwards all pending buffered messages to the new subscriber

[Aaronontheweb]

@Arkatufus can you show what the PublishWithAck message might look like with all of its properties and CTOR defined? Also - what do the ACK / NACK messages look like?

From a flow perspective, how do we deal with a publisher pushing several messages in a row for the same topic that all require ACKs? Do we time them out individually or as a whole? And should there be a limit on the number of buffered messages that can be retained?

[Arkatufus]

Problems needs to be addressed with this design:

1. Topic actor pruning has to be modified so that the actor will not be pruned so long as it still has a message inside its buffer.
2. Each topic actor will have a timer that checks for message timeouts and will have their own special DeadLetter handling for these timed out messages.
3. This design only works on locally subscribed topics, because remote subscription happened through the registry bucket and inter-cluster gossip, not the topic actor
   1. A mechanism has to be made to re-create and kill remote topic actors locally
   2. We need to track which actors are remote and which are local, which one supposed to be cleaned up when a remote subscriber went missing and it doesn't have any local listener

[Aaronontheweb]

This design only works on locally subscribed topics, because remote subscription happened through the registry bucket and inter-cluster gossip, not the topic actor

Must work with remote subscribers - that's the entire purpose of this feature.

[Aaronontheweb]

That's also why I mentioned we don't want to fixate on the number of subscribers, because that's unknowable. Any design that does this is doomed. It should be a binary value: do we have subscribers or not?

[Arkatufus]

Basically, in the actual mechanism, there is no way for the local topic actor to know if a topic is subscribed remotely, we will need to connect the registry bucket with how the local topic actors work

[Aaronontheweb]

Ok, so in this case maybe #7652 has the right idea then - this functionality has to live inside the mediator in order for it to make sense. We don't need to know how many subscribers there are locally in the Topic, just whether or not any exist. We can dump all of the messages in the buffer as soon as we get a Subscribe or a Delta containing a transition from 0 to 1 subscriber.

We are absolutely not going down the road of letting users specify how many subscribers we need to wait for. They can setup a RabbitMQ instance if it's that important.