Global Serverside interceptors throws KeyNotFound exception

[Dewaldf]

Hi, I have implemented the following global serverside interceptor.

public class ServerSideInterceptor : IBootstrapProvider
    {
        public Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            providerRuntime.SetInvokeInterceptor((method, request, grain, invoker) =>
            {
                Debug.WriteLine("Doing some magic here");
                return invoker.Invoke(grain, request);

            });
            return Task.FromResult(0);
        }

        public Task Close()
        {
            return Task.FromResult(0);
        }

        public string Name { get; }
    }

This works well when you have grains calling other grains, but as soon as you have a grain that subscribes to a stream then a KeyNotFoundException gets thrown. If you just call the grain normally without using streams everything works 100%.

Below is an example of the grain subscribing to the stream that is causing the exception to be thrown

[ImplicitStreamSubscription("TestStream")]
    public class SubGrain : Grain, ISubGrain , IAsyncObserver<Guid> 
    {
        private StreamSubscriptionHandle<Guid> _subscriptionHandle;

        public override async Task OnActivateAsync()
        {
            var streamProvider = GetStreamProvider("SMSProvider");
            var incomingStream = streamProvider.GetStream<Guid>(this.GetPrimaryKey(), "TestStream");
            _subscriptionHandle = await incomingStream.SubscribeAsync((IAsyncObserver<Guid>)this);
            await base.OnActivateAsync();
        }

        public Task SayHallo()
        {
            return TaskDone.Done;
        }

         public async Task OnNextAsync(Guid item, StreamSequenceToken token = null)
        {
            Console.WriteLine($"Hallo from Grain : {this.GetPrimaryKey()}");
            var t = GrainFactory.GetGrain<TestHost.Grains.IFooGrain>(item);
            await t.KeepAlive();
        }

        public Task OnCompletedAsync()
        {
            return TaskDone.Done;
        }

        public Task OnErrorAsync(Exception ex)
        {
            return TaskDone.Done;
        }

        public Task<FilterRow[]> GetFilters()
        {
            return Task.FromResult(new FilterRow[] {});
        }
    }

[sergeybykov]

Must be a bug related to grain extensions not being handled correctly.

[gabikliot]

Who is throwing KeyNotFound? Global Serverside interceptor? Can you provide the stack trace?

And what if you just remove the interceptor, but still use streams? Does that work?

Finally, if that does work with streams without interceptor, the fault may be more related to interceptor rather than to streams. What is your interceptor doing?

[Dewaldf]

The streams works without the interceptor. Currently the interceptor only does what is shown in the code snippet above. I initially thought that it might be something that I am doing that is causing the issue, so i removed all the code and only added a Debug.WriteLine()

Stack trace

   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Orleans.Runtime.GrainReference.<InvokeMethodAsync>d__42`1.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 Orleans.Providers.Streams.SimpleMessageStream.SimpleMessageStreamProducerExtension.StreamConsumerExtensionCollection.<DeliverToRemote>d__7.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 Orleans.Providers.Streams.SimpleMessageStream.SimpleMessageStreamProducer`1.<OnNextAsync>d__19.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 TestHost.Grains.FirstGrain.<AddGrains>d__3.MoveNext() in C:\orleansapp\TestHost\Grains\FirstGrain.cs:line 45
--- 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 TestHost.Grains.FirstGrain.<KeepAlive>d__2.MoveNext() in C:\orleansapp\TestHost\Grains\FirstGrain.cs:line 27

[gabikliot]

Ohh, I see.
So with the interceptor you are making the grain call via the invoker that is passed to the interceptor, which is very different from the regular case.
And of course the invoker is part of how the grain extensions are installed.
So probably the invoker that is passed to the interceptor is not doing the same thing as the invoker that is passed used in the regular case.

Basicaly, what @sergeybykov wrote: "Must be a bug related to grain extensions not being handled correctly." More exactly: A bug related to interceptor not using the invoker correctly in combination with extensions.

[gabikliot]

@ReubenBond , why are we passing implementationInvoker in line https://github.com/dotnet/orleans/blob/master/src/OrleansRuntime/Core/InsideRuntimeClient.cs#L382
and not invoker like in the regular non-global interceptor and non-interceptor cases?

Maybe the fix is as simple as that? I can't really figure out the logic in InterceptedMethodInvoker.

[Dewaldf]

@gabikliot I tried the fix you suggested on a local copy of Orleans, I am still getting the error.

[ReubenBond]

@Dewaldf is that the entire/correct stack trace? I don't see the exception there. I'll try to create a repro test case from your code and have a look.

@gabikliot that code is fairly ugly in order to preserve as much efficiency and ensure consumers only 'pay for what you use' as much as possible. I'll submit a PR to clean up some naming and comments to make it more apparent what's going on.

To answer your question, if the grain implements IGrainInvokeInterceptor then we need to pass an implementation of IGrainMethodInvoker which calls the grain's Invoke method. In my working copy I've changed that code to this:

// If the target has a grain-level interceptor or there is a silo-level interceptor, intercept the call.
var shouldCallSiloWideInterceptor = SiloProviderRuntime.Instance.GetInvokeInterceptor() != null && target is IGrain;
var intercepted = target as IGrainInvokeInterceptor;
var grainHasInterceptor = intercepted != null;
if (grainHasInterceptor || shouldCallSiloWideInterceptor)
{
    // Get an invoker which delegates to the grain's IGrainInvocationInterceptor implementation.
    // If the grain does not implement IGrainInvocationInterceptor, then the invoker simply
    // delegates calls to the provided invoker.
    var interceptedInvoker =
        this.interceptedMethodInvokers.GetOrCreate(
            target.GetType(),
            request.InterfaceId,
            invoker);
    var methodInfo = interceptedInvoker.GetMethodInfo(request.MethodId);
    if (shouldCallSiloWideInterceptor)
    {
        // There is a silo-level interceptor and possibly a grain-level interceptor.
        // As a minor optimization, only pass the intercepted invoker if there is a grain-level
        // interceptor.
        resultTask = SiloProviderRuntime.Instance.CallInvokeInterceptor(
            methodInfo,
            request,
            target,
            grainHasInterceptor ? interceptedInvoker : invoker);
    }
    else
    {
        // The grain has an interceptor, but there is no silo-wide interceptor.
        resultTask = intercepted.Invoke(methodInfo, request, invoker);
    }
}
else
{
    // The call is not intercepted at either the silo or the grain level, so call the invoker
    // directly.
    resultTask = invoker.Invoke(target, request);
}

resultObject = await resultTask;

Does that make it clearer? I changed the call to CallInvokeInterceptor to pass grainHasInterceptor ? interceptedInvoker : invoker instead of just interceptedInvoker. It's benign, but passing invoker directly in that case skips a hop into a no-op InterceptedMethodInvoker for every call.