[ImplicitStreamSubscription] Stream ID Should NOT Be Coupled With Grain ID

[ThiagoT1]

TL;DR: There's value in having the same Grain Activation/Reference implicitly subscribe to
multiple streams with different GUID's, specially when said ID's represent something other than an Actor identity.

Current State

1. A grain with ImplicitStreamSubscriptionAttribue must implement IGrainWithGuidCompoundKey
2. If it's not activated yet, one instance (reference) will be activated for every GUID + NameSpapce,
   provided it's IStreamNamespacePredicate says it's a match
   • When it's IStreamNamespacePredicate says it's NOT a match, it WON'T be activated
3. Despite being activated, no stream message will be deliverd to said grain, unless:
   • It runs SubscribeAsync with the EXACT same <T> used by the OnNextAsync caller

Observations

4. Disregaring mistakes, the DE FACTO Stream ID is composed of GUID + NameSpace + StreamType,
   with StreamType being the generic parameter used both by the producer and the consumer.
   Any message sent with this trio not being explicitly listened to will bite the dust.
   • Also, this makes it very hard to reason about the cost of organizing Streams and Grains.
   • Should I have lots of Types with the "same" Stream?
   • Should I have lots of NameSpaces for the same Type?
5. As such, the only main difference between ImplictSubscription and Normal Grain communication is the fact that
   the producer does not need to know the consumer Interface Type. It ONLY has to know the [GU]ID of the consumer, and the Type`s it likes.
6. By having multiple ImplicitStreamSubscription attributes, a Grain Type can capture multiple namespaces (if not all),
   but will only be activated by Stream Messages sent to his GU[ID].
7. There's no way the same Grain Activation/Reference can by activated by messages from Streams with different [GU]ID's,
   even if the IStreamNamespacePredicate says it's a match.
8. By all means, Stream ID's are coupled with Grain ID's, while just being decoupled of its Behavior Type.

Issue

9. Any other way of representing Stream ID's, like when a Stream [GU]ID is instead coupled to
   the Type flowing in it (with maybe the NameSpace being used to represent the
   Type Content's Identity), is impossible.
10. There's no way to make the same OnNextAsync call impact 2 grains with different GU[ID]

Use Case 1 (quick fix?) => Two Namespaces, Two Streams, Same Grain

11. It would be valuable to have multiple namespaces arriving on the same grain.
    This could be done with a NEW attribute derived from ImplicitStreamSubscriptionAttribute, or some other implict way.

Use Case 2 (much more powerful?) => Completelly decouple Stream ID from Grain ID

12. It would be valuable to be able to declare a Grain that implements other IGrain than IGrainWithGuidCompoundKey and implicitly subscribes to
    Streams with [GU]ID's like G, A, and F.
13. That would make it possible to have two types of Grains which subscribe to the same stream:

[ImplicitStreamSubscription(BeerGUID)] //internally uses AllStreamNamespacesPredicate
public class AllBeersSubscriptionGrain : Grain

and

[ImplicitStreamSubscription(BeerGUID, "PaleAle")] //internally uses ExactMatchStreamNamespacePredicate
public class PaleAleSubscriptionGrain : Grain

14. This would probably give rise to the need of a Func<TID, TContext> statically associated with the
    Grain Type that would use TContext to produce a desired TID and still keep it possible
    for a unknown Grain to be activated by a Stream Message that only knew a GUID, a NameSpace and the Message Type.

Related Issues

#4198

---

Could this be done?
Would you guys accept a PR doing that?

One's alternative would be to write Stateless grains (or other GrainPlacement witchery) to act as a proper PubSub layer and make all this decoupling work.

[KevinCathcart]

While I cannot speak for the Orleans team, most of what you have said looks correct to me, but I'll start off with a some clarifications:

A grain with ImplicitStreamSubscriptionAttribue must implement IGrainWithGuidCompoundKey

Technically, if the grain only needs to support a fixed set of namespaces, it can use IGrainWithGuidKey. In that case, all namespaces allowed by the predicates get delivered to the same grain instance. I think this technically covers your #11 case. Obviously this does still require a GUID match, and the grain needs to resume the subscription for each namespace it supports, which is why this can only support a fixed set of namespaces.

(And even that is not strictly true, since implementing IStreamSubscriptionObserver can be used to resume a subscription just-in-time.)

Disregaring mistakes, the DE FACTO Stream ID is composed of GUID + NameSpace + StreamType,
with StreamType being the generic parameter used both by the producer and the consumer.

That is mostly true, subject to the fact that a grain can use a supertype for the "StreamType" when subscribing, and not just an exact match. However it is actually a bit worse than that, since the effective StreamID is actually the tuple of (StreamProvider, Guid, NameSpace, Type).

If you have multiple stream providers, explicit PubSub treats them all as completely separate. But oddly, with implicit streams, you can't specify a specific provider, and will get activated if messages from any of them with Implicit streams enabled, but message from providers the grain does not resume subscriptions for will get ignored, much like a type mismatch.

---

Insofar as your proposed changes, the way I see it, implicit streams fundamentally need to know two things: Does this grain Type want to be implicitly subscribed to some specific Stream? If so, what is the Key for the grain?

The information available to work with when answering these questions is: Metadata about the Grain Type, the provider name, the stream GUID, and the stream namespace.

Answering the first question needs to be fast, since it will be done for every single message. Hence why the current implementation caches the list of types whose predicate returns true for any given stream namespace. Caching along more than one axis might be possible, but would cause the implicit subscription table to bloat even more. Storing a list for every GUID and namespace combination seen would use a lot of memory, and would probably necessitate the use of something like an LRU cache.

The second part (determining what Grain Primary Key to use) is obviously much simpler. The ability to specify some custom StreamId to Grain PrimaryKey mapper here is obviously possible, and seems reasonable. The fast path for GrainKey==StreamId could still exist for maximal performance, with the plugable mechanism available for those that can stand to lose a bit of performance in favor of much greater flexibility.

---

At this point, it is worth pointing out that Orleans 4.0 will get rid of the existing multiple types of primary key system in favor just of having a string Key for Grains. For backwards compatibility there will be come conventions used for the legacy grain types The work for this is already underway.

Obviously this is going to necessitate changes to implicit streams, but the exact impact of this remains up in the air. I suspect the Stream GUIDs will become a string to match. But many details are potentially subject to change.

It would also be wise to fix the problem with implicit streams not working right with heterogeneous silos at the same time. Orleans 4.0 will have a new grain metadata system (see Reuben Bond's pull request #6503). The predicate type and parameters can get put into that system, so that the information is available in silos even when the the grain implementation is not available. Something similar is being done with placement strategy attributes, including their custom parameters. See the handling of stateless workers in that pull request, for an example of how one can handle attributes that take parameters.

[ThiagoT1]

Thanks a lot for the time!

About this:

It would also be wise to fix the problem with implicit streams not working right with heterogeneous silos at the same time.

Would you elaborate?

Also, could you roughly describe the communication model on Streams, regarding a heterogenous cluster?
Are we already smart enough to send a message only once per silo, even if lots of subscribers live there?

[KevinCathcart]

The current code assumes that all producers (pulling agents for persistent streams) will have access to the grain implementation classes of all grains that use an ImplicitStreamSubscription attribute (so that they can inspect the attribute). Obviously that does not work right if only some silos have access to that grain implementation class.

It obviously would be better if all the silos only needed the attribute available, and would get the name of the attribute and its parameters as metadata from other silos, and merely used that.

Are we already smart enough to send a message only once per silo, even if lots of subscribers live there?

Orleans cannot easily do that with persistent streams, because each subscriber might not be fully caught up. The pulling agent sends messages to grains in order, and handles the fact that not every grain will process them at the same speed. Getting ahead of the grain processing makes it harder to guarantee at-least-once delivery. The actual mechanism used to send these messages is ordinary grain call.

I'm not familiar enough with SimpleMessageStreamProvider, to speak about that scenario. I think in theory such an optimization would be applicable there, although it might not be as simple to implement as it sounds.

[ThiagoT1]

@KevinCathcart,

Could we extend this topic a bit?

What is the current state of Orleans (and it's vision for 4.0) regarding sequential logs?

I mean, like one could use RabbitMQ or Kafka for, and consume it as a Stream of Events?

[KevinCathcart]

I'm not familiar with RabbitMq being used for sequential logs. My understanding is that its most common use cases are competing consumer queues, and pub sub style messaging, although a whole bunch of other topologies are possible.

For kafka-like event sequential logs, two concepts apply.

The first is the JournaledGrain and ILogConsistencyProvider system. In practice, this is really just a way to allow multiple activations of a grain to usefully co-exist, primarily for the benefit of geo-replication. The StateStorage provider does not store the events. The LogStorage does, but will load all events from the beginning and reconstruct the grain state from scratch on each activation, making it typically not production suitable. And if you use the CustomStorage provider would allow a hybrid, but then you are already interfacing with whatever underlying storage system directly.

The second concept is rewindable persistent streams. For those, you can provide a StreamSequenceToken when you subscribe, that represents some point in time from which you want to receive events. There are two types of rewindable providers. Partially rewindable providers, that allow you to rewind back as far as the queue cache goes, which could be say a few hours, and gigs worth of data. This is the type that the EventHubs stream provider is.

Secondly there can be fully rewindable stream providers that can go back as far as the underlying data source goes. I'm not sure if any of these actually exist. These would require a unique implementation of the queue cache, that would begin pulling older data if any cursors were created for it.

I am not personally aware of any specific plans for changes here. I suspect @jason-bragg might be able to shed more light.

[ThiagoT1]

@KevinCathcart,

Regarding extreme thruput expectations, is it safe to say that implicit subscriptions (that had ID's decoupled from the subscriber's ID) are a big no-no, as:

1. If we optimize for single message to a silo (so it locally distributes to subscribers) every message will cause a check if any actor should be activated.
2. If we forgo such optimization, then each homogeneous group in an heterogeneous cluster has to receive ALL Stream messages from EVERY Stream?

?

We are marking this issue as stale due to the lack of activity in the past six months. If there is no further activity within two weeks, this issue will be closed. You can always create a new issue based on the guidelines provided in our pinned announcement.

This issue has been marked stale for the past 30 and is being closed due to lack of activity.