Delayed implicit streams rewinding

[randa1]

We're using implicit streams subscriptions with our own stream provider implemented on top of PersistentStreamProvider, and it seems there's a bug when an implicit consumer re-subscribes to a certain position in the stream.

This is my test scenario:

• 1 producer, 1 implicit consumer
• Producer produces 5 messages
• Consumer finishes consuming all 5 messages
• Consumer unsubscribes and re-subscribes with a sequence token of the 3rd message

Expected result: Messages 3-5 are re-consumed immediately
Actual result: No messages are consumed until a new message is produced. Only if we produce another message (the 6th), the consumer "wakes up" and re-consumes messages 3-5 and then 6.

In explicit streams it works as expected. Note that I've verified all of the 5 messages are still in the QueueCache.

I've dug in the source code and it seems the problem is that with explicit streams, IStreamPubSub's RegisterConsumer method calls the producer (the PersistentStreamPullingAgent in our case) and refreshes the cursors. In the ImplicitStreamPubSub the RegisterConsumer method does nothing.

How can this be solved?

[jason-bragg]

Understood. This stems from the fact that the pubsub system for implicit subscriptions is stateless, therefore it can't communicate consumer changes (grain resume calls) back to the producer (the agent), unless there is active event delivery.

One solution to this is to introduce a state-full implicit subscription pubsub system. The pubsub system is not a public accessible concept (other than to configure which to use), so this would require contributing changes or branching the Orleans tree. Adding state to the implicit pubsub system introduces runtime overhead, and could introduce vip swap issues if the state was persisted, depending on implementation of course.

Another possible solution would be to have the agent periodically 'ping' inactive streams for state changes, but this would likely prevent any subscribed grain from deactivating on idle, which may or may not be a good thing.

We'll discuss this internally and see if we can come up with more options.

[gabikliot]

Another option is of course to use explicit subscriptions, and leave implicit as is. Implicit means, just like it name says, no explicit support for various subscription operations. It was meant to be simpler and lighter than explicit.

[randa1]

Thanks @jason-bragg and @gabikliot for your quick response.
I was hoping to avoid using explicit subscriptions because of the added work of persisting them. There's also the overhead of verifying that the consumer grains are subscribed to the stream each time I produce a new message, which most of the time won't be needed as the grain will already be subscribed. Since I have 1:1 relation between producer and consumer grains, implicit streams would be perfect for this if the rewinding worked properly.

[jason-bragg]

@randa1 I understand. Implicit subscriptions have some nice advantages. In addition to the reasons you provided, explicit subscription also use a persistent pub/sub system, which is slower, and could introduce issues during service updates.

[gabikliot]

Can we please rename this issue to delayed implicit streams rewinding. Bug doesn't mean much.

[gabikliot]

Thought about an idea to solve it.
We are talking about persistent streams, SMS is not rewindable.
We need consumer to find producers. This does not necessarily need pub sub . One way to do it is upon rewind request consumer extension (stream impl) calculates producer agent via stream to queue mapped and just sends him a rewind request.
This works for static stream to queue mapping.
An alternative idea, that works also for dynamic stream to queue mapping ( mapping that is not dictated by stream queue mapper) is to enhance the handshake. Producer agent passes its grain reference to consumer in all delivery methods, and the consumer now can just call back.
In general, just a general sort of unrelated remark, I like two way handshake protocols alot. On one hand they are not "global coordination /synchronization" so don't have reliability or scalability problem those usually have. On the other hand, if structured correctly, they do allow strong reasoning and "correctness anchor", so they are a way to build robust scalable coordination.

Anyway, there are of course more details to think through, like what happens when agent changes. Also, we said in the past it is consumer responsible to store its desired rewinding token durably, to deal with crashes. So we need to make sure we can always recover and proceed.

[gabikliot]

Any comments on that Jason? Ran?

[jason-bragg]

"We need consumer to find producers"
When using explicit subscriptions, this is the responsibility of the pubsub system. If the implicit subscription system was stateful it could provide the same capabilities. Doing this without state is not impossible, but would require solving a number of other problems, like mapping streamIds to queues, finding which host the queue was on, and learning the id of the system target processing the queue of interest. Again, not impossible, but I think there are easier and more strait forward solutions. I do, however, think this is a good idea to explore, as it could be a superior solution, especially for implicit subscriptions, but it would be a significant change to the PersistentStreamProvider.

"enhance the handshake"
This could improve the protocol, but in the end, I don't think it could solve the problem. Relying on the handshake requires that the agent must talk to the grain at least once. This can't be guaranteed when rewinding a stream. Admittedly, this hole is in an edge case probably so narrow as not to be noticed, but we're in this thread because we missed edge cases that didn't effect our usage.
Additionally, the further modification of the handshake only increases the disparity between the explicit subscription and the implicit subscription code paths, not to mention how brittle that code is already.

My preference is to use a stateful pubsub system for rewindable implicit subscriptions. We can do this by writing a new implementation of the implicit subscription pubsub system, or by correcting the shortcomings in the grain based pubsub system, and using that. The advantages of this approach are:

• Both rewindable implicit and explicit subscriptions can follow almost the same code path, differing only by which pub/sub system they use.
• We can remove the per event delivery handshake logic, making event delivery simpler.
• Since consumers will now be able to talk to producers (agents) for both implicit and explicit subscriptions, subscription errors can be detected at subscribe time, rather than waiting until event delivery detects the error.
• Like explicit subscriptions, any subscriptions that encounters repeated errors can be put into a faulted state, preventing further delivery.
• Working through how to make the implicit subscription pub/sub stateful, we may find solutions to the issues with the grain based pubsub system.

[randa1]

I tend to agree with Jason, bringing stateful pubsub back to implicit streams or creating a new implicit subscription type seems as the most straightforward way, as in the current implementation every issue like this one will create further branching of the code paths.