Stream Listener with a generic interface failed to subscribe to stream again

[centur]

Looks like pub-sub message can't be pushed to a Grain which implements generic parametrized interface. I've explain the problem using pretty plain example.

I have this listener class :

[ImplicitStreamSubscription(nameof(GrainCommandType.PermissionGrainDelete))]
public class PermissionCommandsListener: BaseProxyGrain, IPermissionCommandsListener
{...}

where interface is defined like :

public interface IPermissionCommandsListener: IStreamListener<GrainCommandData>, IGrainWithGuidKey 
{...}

and it works perfectly. Pubsub stream message being delivered and processed and so on.

Now I'm doing small optimization and instead of implementing IPermissionCommandsListener I want to implement generic interface directly on the grain. Presumably it doesn't changes anything. So now I have :

[ImplicitStreamSubscription(nameof(GrainCommandType.PermissionGrainDelete))]
public class PermissionCommandsListener: BaseProxyGrain,IStreamListener<GrainCommandData>, IGrainWithGuidKey

Everything is the same as it was before, even to make this case cleaner - I've removed Orleans.Streams.PubSubRendezvousGrain State from Azure table

Doing same sample run I'm receiving this:

 Orleans.Runtime.OrleansException: Unexpected: Cannot find an implementation class for grain interface -1256675927
   at Orleans.Runtime.GrainTypeManager.GetTypeInfo(Int32 typeCode, String& grainClass, PlacementStrategy& placement, String genericArguments)
   at Orleans.Runtime.Catalog.GetGrainTypeInfo(Int32 typeCode, String& grainClass, PlacementStrategy& placement, String genericArguments)
   at Orleans.Runtime.Placement.PlacementContextExtensions.GetGrainPlacementStrategy(IPlacementContext this, Int32 typeCode, String genericArguments)
   at Orleans.Runtime.Placement.PlacementContextExtensions.GetGrainPlacementStrategy(IPlacementContext this, GrainId grainId, String genericArguments)
   at Orleans.Runtime.Dispatcher.<AddressMessage>d__17.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.GetResult()
   at Orleans.Runtime.Dispatcher.<AsyncSendMessage>d__14.MoveNext()

Interesting side note - Orleans writes new Orleans.Streams.PubSubRendezvousGrain State information into the table, but then fails to push the message to subscriber. If I revert it back to the IPermissionCommandsListener implementation - everything gets back to normal. I don't even need to delete PubSubRendezvousGrain state.

So there is an issue but I'm not 100% sure where it is.

[gabikliot]

Nothing is written to the table for ImplicitStreamSubscriptions. There is some kind of mismatch here. To debug, start by configuring the pub sub in the stream provider to be implicit only. But even without that, the table may be created but should remain empty, unless you of course also use other, unrelated, explicit streams.

If things still don't work, please create a PR reproducing this issue. If there is indeed an issue/bug, it may involve code gen as well as streaming bug, which means it will spread the debugging efforts between our code gen masters and streaming experts. Without a clear repro I am doubtful we will be able to resolve it.

[gabikliot]

@centur , do we still have this as an open issue? Was it resolved, together maybe with https://github.com/dotnet/orleans/issues/1272?

[centur]

Yes, this is open, I got back to it this morning after all project nugets were upgraded and it's still there. I'll try to send a PR with repro before end of this week.

[centur]

I can't figure out how to repro it in Orleans Solution and which unit tests and silos I should use. It's a bit complicated for a simple task I need. I've created a sample set of 2 console apps and going to use it as a repro environment. Going to push it to GH once it'll be working with repro

[gabikliot]

@centur , do you have any follow up on this issue? If not, can it be closed?

[gabikliot]

Closing due to the lack of additional data. Feel free to re-open.

[centur]

Well. You won't believe this... I was pretty unexperienced when this issue was open, and then never got back to it. But during my demo preparation I stepped on the same problem and now I have an easy to run repro :)

https://github.com/centur/altnet-streams-demo

on buggy-stream branch if you'll run the app and send any message (type anything in client console) - it will give you that above error.

On fixed-stream - it's fixed.

Key difference between 2 branches - an empty interface.

public interface IChatMessageListener : IAsyncObserver<ChatMsg>, IGrainWithGuidKey { }

[centur]

I can't reopen this issue so I either need to summon @gabikliot or @sergeybykov

[xiazen]

Hi @centur I think this issue has been fixed in 2.0 . I upgraded your buggy-stream sample to use Orleans 2.0 rc and played with join channel and send messages, and I didn't see any exception like you posted before.

The upgrade version of the buggy-stream is in https://github.com/xiazen/altnet-streams-demo/tree/update-to-2.0 feel free to pull it down and double check if the issue has been fixed.

[xiazen]

And sorry for the late response. The team has been real busy with 2.0 .

[xiazen]

close this since it seems 2.0 fixed this bug. Feel free to reopen if you think the bug still is there.

[worldspawn]

3 years later :D Hi! I'm working on the same code base as centur and we are uncomfortably stuck with 1.5.4 for the foreseeable future.

I have even less understanding of the problem and was just trying to reduce the surface area of our listeners. I've found if I make a new listener that looks like this:

[ImplicitStreamSubscription("Requests")]
    public class RequestListener : Grain, IStreamSubscriptionObserver<IBaseRequest>

I get the error mentioned above when putting an item on the stream.

But if I define and add an interface (IStreamListener<IBaseRequest>) (this was lying around in our code) it just magically starts working.

public interface IStreamListener<TMessageType>: IGrainWithGuidKey
    {
        Task ReceiveCommand(TMessageType item, StreamSequenceToken token = null);
        Task OnCommandCompleted();
        Task OnCommandError(Exception ex);
    }

We dont seem to have any dependencies elsewhere on this interface. No idea why this matters. I didn't want the methods so I made my own interface:

public interface IStreamListener<TMessageType> {}

This didn't work. I had to make the interface implement IGrainWithGuidKey. Dropping IStreamListener from the my listener class and just having it implement IGrainWithGuidKey directly doesn't work either.

So there seems to be a requirement to have some rando-arbitrary interface defined and implemented on the listener class and it must in turn implement IGrainWithGuidKey. Didn't bother checking if the interface had to be generic.

This is right up there in my top 5 omgwtfbbq coding moments :)

[ReubenBond]

Hi Sam,

I think I understand the problem (could be totally wrong).

Essentially, the system is trying to send a message to your grain but your grain was never registered with the system because it doesn't implement any grain interfaces. Looking at the code (in the old version of GrainTypeManager.cs from v1.5.4), it seems like a quirk/bug in how grain classes were registered during startup.

So try adding an empty interface such as:

public interface IRequestListenerGrain : IGrainWithGuidKey { }

Then implementing that in your grain class:

public class RequestListener : Grain, IRequestListenerGrain, IStreamSubscriptionObserver<IBaseRequest>

That may solve the issue for you.

[worldspawn]

Thanks @ReubenBond. Thats what I ended up doing except I had left a generic argument on the interface which I've now removed and its still working. Cheers 👍