Add wait-for-subscribers feature #7652

Arkatufus · 2025-05-15T21:09:03Z

Fixes #7627

DistributedPubSub wait-for-subscriber feature flag

HOCON settings added

akka.cluster.pub-sub.buffered-messages.max-per-topic = 1000
akka.cluster.pub-sub.buffered-messages.timeout-check-interval = 1s

Settings added to DistributedPubSubSettings

MaxBufferedMessagePerTopic
BufferedMessageTimeoutCheckInterval

Buffer

If the mediator received a PublishWithAck message and able to deliver it immediately, it will send back a PublishSucceeded message back to the original sender as a signal.
If the mediator could not deliver a PublishWithAck message, it will store the message as a BufferedMessage inside the buffer.

Buffer timeout

The PublishWithAck message contains the timeout value for each messages
Every timeout-check-interval time, the mediator will scan through all of the buffered message to check and see if it has been sitting in the buffer for more than timeout period.
Timed out buffered message will be removed from the buffer and either silently discarded or sent to the dead letter.
A PublishFailure message will also be sent to the original sender to signal publish failures.

Buffer delivery

We're now tracking every new key insertion into the bucket registry.
When keys are being inserted, the mediator send a NewBucketKeysAdded message to itself.
When a NewBucketKeysAdded message is received, the mediator checks the buffer and re-send all waiting messages that matches the keys to itself.
For each delivered message, the mediator will send a PublishSucceeded message to the original sender as a signal.

Buffer overflow

When a new message is inserted and the topic buffer count exceeds buffered-messages.max-per-topic, the mediator will discard the oldest message in the buffer and send a PublishFailed to the original sender of that message.

Checklist

For significant changes, please ensure that the following have been completed (delete if not relevant):

This change follows the Akka.NET API Compatibility Guidelines.
This change follows the Akka.NET Wire Compatibility Guidelines.
Changes in public API reviewed, if any.
I have added website documentation for this feature.

…bers-feature # Conflicts: # src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/reference.conf

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/OverflowStrategy.cs

…us/akka.net into Add-wait-for-subscribers-feature

Arkatufus

Self-review

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedMessages.cs

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

Arkatufus · 2025-05-19T14:42:00Z

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

-                    .ToImmutableDictionary(kv => kv.Key, kv => kv.Value);
-            }
-        }
+            => _registry


No logic change, just modernization

Arkatufus · 2025-05-19T14:56:06Z

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubSettings.cs

+        /// <param name="maxBufferedMessagePerTopic">Maximum message buffer size for each topic</param>
+        /// <param name="bufferedMessageTimeoutCheckInterval">Buffered message timeout condition check interval</param>
+        /// <exception cref="ArgumentException">Thrown if a user tries to use a <see cref="ConsistentHashingRoutingLogic"/> with routingLogic.</exception>
+        public DistributedPubSubSettings(


New backward compatible .ctor

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubSettings.cs

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/Internal/TopicMessages.cs

Aaronontheweb

Made some suggestions

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

Aaronontheweb · 2025-05-19T15:57:46Z

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubSettings.cs

        /// <param name="maxDeltaElements">The maximum number of delta elements that can be propagated in a single gossip tick.</param>
        /// <param name="sendToDeadLettersWhenNoSubscribers">When a message is published to a topic with no subscribers send it to the dead letters.</param>
        /// <exception cref="ArgumentException">Thrown if a user tries to use a <see cref="ConsistentHashingRoutingLogic"/> with routingLogic.</exception>
+        [Obsolete("Use .ctor that supports WaitForSubscribers instead. Since 1.4.42")]


A thing we did on #7637 that we should do here:

Don't even make the CTORs on these settings classes public. Force users to use the static SettingsClass.Create method. Eliminates a ton of binary compatibility issues long-term.

Make all of these settings classes into record types - this eliminates the .WithValueName( method spam.

Maybe we do that across the board in v1.6 - I'm doing it in v1.5 for the TcpSettings type on #7637. I'm fine doing that for DistributedPubSub too.

Aaronontheweb · 2025-05-19T16:01:42Z

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubSettings.cs

        /// <param name="maxDeltaElements">The maximum number of delta elements that can be propagated in a single gossip tick.</param>
        /// <param name="sendToDeadLettersWhenNoSubscribers">When a message is published to a topic with no subscribers send it to the dead letters.</param>
        /// <exception cref="ArgumentException">Thrown if a user tries to use a <see cref="ConsistentHashingRoutingLogic"/> with routingLogic.</exception>
+        [Obsolete("Use .ctor that supports WaitForSubscribers instead. Since 1.4.42")]


Implication of this design change is that users have to call DistributedPubSubSettings.Create(ActorSystem) and then they can manipulate it from there.

Aaronontheweb · 2025-05-19T16:02:10Z

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubSettings.cs

        /// <param name="maxDeltaElements">The maximum number of delta elements that can be propagated in a single gossip tick.</param>
        /// <param name="sendToDeadLettersWhenNoSubscribers">When a message is published to a topic with no subscribers send it to the dead letters.</param>
        /// <exception cref="ArgumentException">Thrown if a user tries to use a <see cref="ConsistentHashingRoutingLogic"/> with routingLogic.</exception>
+        [Obsolete("Use .ctor that supports WaitForSubscribers instead. Since 1.4.42")]


Changing this on v1.5 will probably blow a bunch of stuff up immediately, so we should probably save that for v1.6

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/Internal/TopicMessages.cs

Aaronontheweb

Left you a suggestion on the pruning check

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

Aaronontheweb

More comments

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/Internal/TopicMessages.cs

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

Aaronontheweb

Left some more suggestions but I think overall the design looks good - what's the testing plan?

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedMessages.cs

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

Aaronontheweb · 2025-05-19T17:41:34Z

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

        {
+            if (message is PublishWithAck needAck)
+            {
+                sender.Tell(new PublishFailed(needAck, PublishFailReason.Timeout));


Do we want to do an early return here and skip logging the dead letter or do we want to do both? The way we have it does both and I actually think that's ok - since the DeadLetter monitoring will usually get universal attention.

I don't think its wise to change old behavior

Aaronontheweb · 2025-05-19T17:41:58Z

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

        }

-        private void PublishToEachGroup(string path, Publish publish)
+        private void PublishToEachGroup(string path, IWrappedMessage publish)


IMHO, do a common interface for both of the Publish message types and handle that instead.

This can't be done because PublishMessage is also used to send SendAll

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubSettings.cs

Arkatufus added 6 commits May 16, 2025 04:08

Add wait-for-subscribers feature

acae403

Merge remote-tracking branch 'upstream/dev' into Add-wait-for-subscri…

e72cbd4

…bers-feature # Conflicts: # src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs

Fix missing code

8f05840

Update API approval list

0177811

Fix HOCON errors

d30b8a2

Merge branch 'dev' into Add-wait-for-subscribers-feature

001e056

Aaronontheweb reviewed May 16, 2025

View reviewed changes

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/reference.conf Outdated Show resolved Hide resolved

Aaronontheweb closed this May 16, 2025

Aaronontheweb mentioned this pull request May 16, 2025

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

Closed

Aaronontheweb reopened this May 16, 2025

Aaronontheweb reviewed May 16, 2025

View reviewed changes

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/OverflowStrategy.cs Outdated Show resolved Hide resolved

Arkatufus added 8 commits May 16, 2025 22:14

Merge branch 'dev' into Add-wait-for-subscribers-feature

98a6999

Merge branch 'dev' into Add-wait-for-subscribers-feature

8083046

Trim features to fit

c0af9fe

Merge branch 'Add-wait-for-subscribers-feature' of github.com:Arkatuf…

89ecf1d

…us/akka.net into Add-wait-for-subscribers-feature

Simplify logic

b4b1c5e

Code cleanup

00761f5

Code cleanup

6cf2e00

Update API Approval list

4c0f026

Arkatufus commented May 19, 2025

View reviewed changes

Aaronontheweb requested changes May 19, 2025

View reviewed changes

Fix code

6792915

Aaronontheweb requested changes May 19, 2025

View reviewed changes

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs Show resolved Hide resolved

Aaronontheweb reviewed May 19, 2025

View reviewed changes

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/Internal/TopicMessages.cs Outdated Show resolved Hide resolved

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedPubSubMediator.cs Show resolved Hide resolved

Arkatufus added 4 commits May 20, 2025 00:18

Fix codes

cba4f30

Fix codes

13d05ca

Update code

6f26d41

Update API Approval list

ef05c31

Aaronontheweb requested changes May 19, 2025

View reviewed changes

Fix PublishWithAck

1dbdd89

Aaronontheweb merged commit 3bc4457 into akkadotnet:dev May 19, 2025
11 checks passed

Arkatufus mentioned this pull request May 21, 2025

Update RELEASE_NOTES.md for 1.5.42 release #7668

Merged

Arkatufus added this to the 1.5.42 milestone May 21, 2025

This was referenced Jul 25, 2025

Bump Akka.Streams and 15 others Aaronontheweb/Alpakka#873

Closed

Bump Akka and 14 others petabridge/TurboMqtt#282

Closed

Bump Akka.Streams and 17 others Aaronontheweb/Alpakka#874

Closed

This was referenced Aug 15, 2025

Bump Akka.Streams from 1.5.40 to 1.5.47 Aaronontheweb/Alpakka#880

Closed

Bump Akka.Streams.TestKit from 1.5.40 to 1.5.47 Aaronontheweb/Alpakka#881

Closed

This was referenced Aug 25, 2025

Bump Akka.Streams from 1.5.40 to 1.5.48 Aaronontheweb/Alpakka#882

Closed

Bump Akka.Streams.TestKit from 1.5.40 to 1.5.48 Aaronontheweb/Alpakka#883

Closed

This was referenced Sep 23, 2025

Bump Akka.Streams from 1.5.40 to 1.5.50 Aaronontheweb/Alpakka#887

Open

Bump Akka.Streams.TestKit from 1.5.40 to 1.5.50 Aaronontheweb/Alpakka#888

Open

Add wait-for-subscribers feature #7652

Add wait-for-subscribers feature #7652

Uh oh!

Conversation

Arkatufus commented May 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

DistributedPubSub wait-for-subscriber feature flag

HOCON settings added

Settings added to DistributedPubSubSettings

Buffer

Buffer timeout

Buffer delivery

Buffer overflow

Checklist

Uh oh!

Uh oh!

Uh oh!

Arkatufus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Aaronontheweb left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Aaronontheweb left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Aaronontheweb left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Aaronontheweb left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Arkatufus commented May 15, 2025 •

edited

Loading