akkadotnet / akka.analyzers Goto Github PK

Need more data, but wanted to create an issue for this - looks like this is an SDK / Tooling issue with Roslyn.

Is your feature request related to a problem? Please describe.

From the submission on #55

public sealed class MyActor : ReceiveActor{
   public MyActor(){
      ReceiveAsync<string>(str => Sender.Tell(str)); // should get a warning here
   }
}

Using ReceiveAsync this way is less performant than a normal Receive because this generates an extra ActorTask message plus an additional suspend + resume of the actor's mailbox. That overhead is fine if you're await-ing something, but if the code is essentially synchronous then we should not do so.

Describe the solution you'd like

Issue a WARNING level severity here and a code fix to rewrite the ReceiveAsync to a regular Receive if the ReceiveAsync doesn't actually have any awaits happening inside of it.

Additional context

If the user is invoking a declared method as the ReceiveAsync handler instead of an anonymous delegate, that might be a lot trickier to analyze correctly as the method can get used across multiple callsites. Maybe it's best to skip doing that for now.

Version Information
Version of Akka.NET? Akka.NET v1.5.16, Akka.Analyzers v0.2.2

Describe the bug

This code is currently flagged as an error by AK1002:

ReceiveAsync<StopActor>(async p =>
{
    try
    {
        var actorSelection = Context.System.ActorSelection(p.ActorPath);
       
        var actorRef = await actorSelection.ResolveOne(p.Timeout);

        var stopped = await actorRef.GracefulStop(p.Timeout); // AK1002 false positive occurs here
        if(stopped)
            Sender.Tell(new CommandResponse($"Successfully terminated [{p.ActorPath}] within [{p.Timeout}]"));
        else
            Sender.Tell(new ErroredCommandResponse($"Failed to terminate [{p.ActorPath}] within [{p.Timeout}]"));
    }
    catch (ActorNotFoundException)
    {
        Sender.Tell(new CommandResponse($"No actor found at [{p.ActorPath}] - no termination required"));
    }
    catch (Exception e)
    {
        Sender.Tell(new ErroredCommandResponse($"Could not complete actor termination due to [{e.Message}]"));
    }
});

This is just an actor await-ing on a GracefulStop, not on its own graceful stop.

Expected behavior

Should not be flagged.

Actual behavior

Is flagged.

Using implicit sender inside async functions is dangerous, this has the same severity as AK1001 as ActorCell.Current might not be the same actor ref as the original Self when the async function is finally invoked.

Please describe what you are trying to understand

Need to document all of the error codes on https://getakka.net/ and have the help links point back directly to the site.

The relevant code area / formula for this is here:

https://github.com/Aaronontheweb/akka.analyzers/blob/ff491714e9da2e8a6f4da3f35c86140dea177a32/src/Akka.Analyzers/Utility/RuleDescriptors.cs#L14-L18

I've created a basic rules scheme already - it's just a matter of having that translate 1:1 to https://getakka.net/

Original issue reported here: akkadotnet/akka.net#6131

All messages should be immutable - and this touches on a number of facets:

1. All properties must be readonly (fields in this case), init-only, or get-only
2. All collection properties must be declared as System.Collections.Immutable, IReadOnly<T>, or the new frozen collections introduced in .NET 8

This one will require a lot more work than the others.

Related Akka.NET issue: akkadotnet/akka.net#3376

Is your feature request related to a problem? Please describe.

This is a bit of an edge case, but it should be easy to detect during compilation:

public sealed class MyActor : ReceiveActor{
   public MyActor(){
      ReceiveAsync<string>(async str => {
         await Context.Self.GracefulStop(); // THIS WILL DEADLOCK
      }):
   }
}

Describe the solution you'd like

Detect if an actor is trying to await itself shutting down and mark it as a compilation error - also, provide a codefix that simply removes the await in this case. It's the await-ing that is the problem.

Describe alternatives you've considered

Removing the await call might cause problems if we're inside an async method (i.e. there's no Task that gets returned), but I wouldn't go nuts trying to handle every possible edge case with the code fix like this. That's the end-user's problem and they can clean it up. We just need to make the analysis rule that's been flagged go away.

Akin to what we did over on https://github.com/akkadotnet/akkadotnet-templates

Sender, Self, Context.Sender and Context.Self must be closed when accessed inside a lambda expression argument body, because it can go out of context.

This should help move users onto the newer APIs that prevent leaking of background tasks over time. We can also pair this with a "fixit" suggestion.

Is your feature request related to a problem? Please describe.

Any of the messages that were previously handled manually prior to akkadotnet/akka.net#6717 should be removed from the end-user's shard message extractor, given that they're now automatically handled by Akka.Cluster itself.

Describe the solution you'd like

An Analysis rule with a DiagnositicServerity.Warning label along with a CodeFix that just removes the offending lines.

Additional context

This rule will only apply to Akka.NET versions 1.5.15 and higher.

          Thanks for the explanation @Aaronontheweb. But I've one follow up question regarding this topic.

Also @wesselkranenborg, you code is fine - minus the bit I said about the perf hit using ReceiveAsync instead of Receive when you're not performing any awaits.

Basically we are now writing non-blocking code if I hear you correctly. And we might thus break the actor model?
I've three examples on how we could write this:

In this example we have the LeaveClusterCore async method which is awaiting code. We want this method to be executed and freeing up the actor once this logic is performed. We are now using the first method, the second method is to make it with the Receive<> but if I hear you correctly we should use the last way of writing it. Am I correct on that?

Originally posted by @wesselkranenborg in #55 (comment)

Need this to support things like DI et al:

private class TestProducer : IIndirectActorProducer
        {
            TestLatch latchActor;

            public TestProducer(TestLatch lp, TestLatch la)
            {
                latchActor = la;
                lp.Reset();
                lp.CountDown();
            }

            public ActorBase Produce()
            {
                latchActor.CountDown();
                return new PropsTestActor();
            }

            public Type ActorType
            {
                get { return typeof(PropsTestActor); }
            }


            public void Release(ActorBase actor)
            {
                actor = null;
            }
        }

This should not trigger AK1000.

Version Information
Version of Akka.NET? 1.5.16
Which Akka.NET Modules? Akka.Analyzers

Describe the bug
The documentation for rule AK1001 says that in order to pass sender to PipeTo method, local variable should be used, because PipeTo will be executed later. I have serious doubts that this is the case.

Here is the code that is provided in the documentation:

public sealed class MyActor : UntypedActor{

    protected override void OnReceive(object message){
        async Task<int> LocalFunction(){
            await Task.Delay(10);
            return message.ToString().Length;
        }

        // potentially unsafe use of Context.Sender
        LocalFunction().PipeTo(Sender); 
    }
}

AFAIK, here is what happens in the line LocalFunction().PipeTo(Sender);:

1. We execute local function. It is marker as async. This means that state machine will be used.
   So, what will happen: the control will be passed to the function, it will create new instance of state machine class, start it and simply return task which can be used to get information about the work being done. I'll stress here: the code will NOT wait until the job is done, it will just start the job and return instance of Task class for caller to be able to track the job. All this will be done synchronously, in the very same thread that we started.
2. Then this instance of Task class will be passed to PipeTo method as the first argument (as PipeTo is an extension method) and the second argument will be value returned by Sender getter method. I'll repeat: the getter method will be executed right here, BEFORE calling PipeTo, because in order to call PipeTo we must know what arguments we provide to the method. And all this happens synchronously, in the same thread that we started in. We're still synchronously processing current message and Akka will not change sender as we are not done yet. Sender getter will return address of sender object (as it is a reference type) and this address will be used inside of PipeTo method. So even when later Sender getter will be returning another object, PipeTo method will still use the address we provided to it.
3. PipeTo is async method, so again, instance of state machine will be created, started and task will be returned. And we will continue
   execution in the same thread as before. If after this line we will get value of this.Sender (or Context.Sender) again, we will get the same value as the one we've passed to PipeTo.

To Reproduce

The unit test I use to verify the behavior is following:

using Akka.Actor;
using Akka.TestKit;
using Akka.TestKit.Xunit2;

namespace UnitTests;

public class UnitTest1 : TestKit
{
    [Fact]
    public void Test1()
    {
        IActorRef sut = this.Sys.ActorOf(Props.Create(() => new MyActor()));
        TestProbe testProbe1 = this.CreateTestProbe();
        TestProbe testProbe2 = this.CreateTestProbe();

        sut.Tell("first", testProbe1.Ref);
        sut.Tell("second", testProbe2.Ref);

        testProbe1.ExpectMsg(5);
        testProbe2.ExpectMsg(6);
    }

    public sealed class MyActor : UntypedActor
    {
        protected override void OnReceive(object message)
        {
            async Task<int> LocalFunction()
            {
                Console.WriteLine($"Begin {nameof(LocalFunction)} for message \"{message}\"");
                await Task.Delay(1000);
                Console.WriteLine($"End {nameof(LocalFunction)} for message \"{message}\"");
                return message.ToString().Length;
            }

            Console.WriteLine($"Received \"{message}\", this.Sender={this.Sender}, Context.Sender={Context.Sender}");
#pragma warning disable AK1001
            LocalFunction().PipeTo(this.Sender);
#pragma warning restore AK1001
        }
    }
}

The output is following:

Received "first", this.Sender=[akka://test/system/testActor2#247961004], Context.Sender=[akka://test/system/testActor2#247961004]
Begin LocalFunction for message "first"
Received "second", this.Sender=[akka://test/system/testActor3#1017927033], Context.Sender=[akka://test/system/testActor3#1017927033]
Begin LocalFunction for message "second"
End LocalFunction for message "first"
End LocalFunction for message "second"

By the end of the local function called for the "first" the "second" message will already be processed, so the sender should have been changed to testProbe2, but the first message is correctly delivered to testProbe1 - test passes.

I tried running another variant of this test, making sure that local function for "first" end in the middle of processing "second":

public class UnitTest1 : TestKit
{
    [Fact]
    public void Test2()
    {
        IActorRef sut = this.Sys.ActorOf(Props.Create(() => new MyActor2()));
        TestProbe testProbe1 = this.CreateTestProbe();
        TestProbe testProbe2 = this.CreateTestProbe();

        sut.Tell("first", testProbe1.Ref);
        sut.Tell("second", testProbe2.Ref);

        testProbe1.ExpectMsg(5);
        testProbe2.ExpectMsg(6);
    }

    public sealed class MyActor2 : UntypedActor
    {
        protected override void OnReceive(object message)
        {
            async Task<int> LocalFunction()
            {
                Console.WriteLine($"Begin {nameof(LocalFunction)} for message \"{message}\"");
                await Task.Delay(100);
                Console.WriteLine($"End {nameof(LocalFunction)} for message \"{message}\"");
                return message.ToString().Length;
            }

            // wait some time to make local function for "first" end while
            // "second" is being processed
            Console.WriteLine($"Received \"{message}\", this.Sender={this.Sender}, Context.Sender={Context.Sender}, waiting");
            Thread.Sleep(100);
            Console.WriteLine($"Continuing processing \"{message}\", this.Sender={this.Sender}, Context.Sender={Context.Sender}");
#pragma warning disable AK1001
            LocalFunction().PipeTo(this.Sender);
#pragma warning restore AK1001
        }
    }
}

The output for the second test:

Received "first", this.Sender=[akka://test/system/testActor2#719986033], Context.Sender=[akka://test/system/testActor2#719986033], waiting
Continuing processing "first", this.Sender=[akka://test/system/testActor2#719986033], Context.Sender=[akka://test/system/testActor2#719986033]
Begin LocalFunction for message "first"
Received "second", this.Sender=[akka://test/system/testActor3#1932466648], Context.Sender=[akka://test/system/testActor3#1932466648], waiting
End LocalFunction for message "first"
Continuing processing "second", this.Sender=[akka://test/system/testActor3#1932466648], Context.Sender=[akka://test/system/testActor3#1932466648]
Begin LocalFunction for message "second"
End LocalFunction for message "second"

Even in this case test passes and first message is successfully delivered to testProbe1.

I might miss something here, but I don't really see what.

Expected behavior
It would be great if someone could provide the code that will demonstrate scenario when code violating AK1001 does not work as expected.

Actual behavior
The sample code works fine.

Environment
Linux, .NET 6.0.417.

using Akka.Actor;
using Akka.TestKit;
using Akka.TestKit.Xunit2;

namespace UnitTests;

public class UnitTest1 : TestKit
{
    [Fact]
    public void Test1()
    {
        IActorRef sut = this.Sys.ActorOf(Props.Create(() => new MyActor()));
        TestProbe testProbe1 = this.CreateTestProbe();
        TestProbe testProbe2 = this.CreateTestProbe();

        sut.Tell("first", testProbe1.Ref);
        sut.Tell("second", testProbe2.Ref);

        testProbe1.ExpectMsg(5);
        testProbe2.ExpectMsg(6);
    }

    [Fact]
    public void Test2()
    {
        IActorRef sut = this.Sys.ActorOf(Props.Create(() => new MyActor2()));
        TestProbe testProbe1 = this.CreateTestProbe();
        TestProbe testProbe2 = this.CreateTestProbe();

        sut.Tell("first", testProbe1.Ref);
        sut.Tell("second", testProbe2.Ref);

        testProbe1.ExpectMsg(5);
        testProbe2.ExpectMsg(6);
    }

    public sealed class MyActor : UntypedActor
    {
        protected override void OnReceive(object message)
        {
            async Task<int> LocalFunction()
            {
                Console.WriteLine($"Begin {nameof(LocalFunction)} for message \"{message}\"");
                await Task.Delay(1000);
                Console.WriteLine($"End {nameof(LocalFunction)} for message \"{message}\"");
                return message.ToString().Length;
            }

            Console.WriteLine($"Received \"{message}\", this.Sender={this.Sender}, Context.Sender={Context.Sender}");
#pragma warning disable AK1001
            LocalFunction().PipeTo(this.Sender);
#pragma warning restore AK1001
        }
    }

    public sealed class MyActor2 : UntypedActor
    {
        protected override void OnReceive(object message)
        {
            async Task<int> LocalFunction()
            {
                Console.WriteLine($"Begin {nameof(LocalFunction)} for message \"{message}\"");
                await Task.Delay(100);
                Console.WriteLine($"End {nameof(LocalFunction)} for message \"{message}\"");
                return message.ToString().Length;
            }

            // wait some time to make local function for "first" end while
            // "second" is being processed
            Console.WriteLine($"Received \"{message}\", this.Sender={this.Sender}, Context.Sender={Context.Sender}, waiting");
            Thread.Sleep(100);
            Console.WriteLine($"Continuing processing \"{message}\", this.Sender={this.Sender}, Context.Sender={Context.Sender}");
#pragma warning disable AK1001
            LocalFunction().PipeTo(this.Sender);
#pragma warning restore AK1001
        }
    }
}

Version Information
Version of Akka.NET? v0.2.0
Which Akka.NET Modules? Akka.Analyzers

Describe the bug

using Akka.Actor;
using System.Threading.Tasks;

public sealed class MyActor : UntypedActor{

   protected override void OnReceive(object message){
       async Task<int> LocalFunction(){
           await Task.Delay(10);
           return message.ToString().Length;
       }

       // incorrect use of closures
       LocalFunction().PipeTo(Context.Sender);
   }
}

Context.Sender should get flagged by AK1001 but currently does not, even though it's equally unsafe.

Calling any of the IWithTimers.ScheduleXXX methods will not work inside PreRestart because those timers will immediately be cancelled when the actor restarts, by design. If an actor wants to schedule a message to itself between restarts it should do that using the older methods.

Version Information
Version of Akka.NET? v0.1.1
Which Akka.NET Modules? Akka.Analyzer

Describe the bug

We have the following original code:

https://github.com/akkadotnet/akka.net/blob/2633f914122b42aae87de807cee1561ab9749f27/src/core/Akka.Docs.Tests/Streams/StreamRefsDocTests.cs#L56-L61

It should be refactored to:

 // create a source
var sender = Sender;
StreamLogs(request.StreamId)
    // materialize it using stream refs
    .RunWith(StreamRefs.SourceRef<string>(), Context.System.Materializer())
    // and send to sender
    .PipeTo(sender, success: sourceRef => new LogsOffer(request.StreamId, sourceRef));

But is INSTEAD refactored to

 // create a source
var sender = Sender;
StreamLogs(request.StreamId)
    // materialize it using stream refs
    .RunWith(StreamRefs.SourceRef<string>(), Context.System.Materializer())
    // and send to sender
    .PipeTo(sender));

The success: sourceRef => new LogsOffer(request.StreamId, sourceRef) gets deleted by accident here.

Expected behavior

Only the replyTo argument on the PipeTo call should be changed, not 100% of the arguments.

Actual behavior

The success: sourceRef => new LogsOffer(request.StreamId, sourceRef) gets deleted by accident here.

Additional context

If there are any other optional parameters for PipeTo, we should add cases to cover those inside the code fix test suite for it as well. Do this exhaustively so we ensure that there are no problems.

The design of the test suite, especially, is lifted from how xUnit designed theirs. Need to provide some attribution in the file headers on those files.

Version Information
Version of Akka.NET? 1.5.15
Which Akka.NET Modules? Akka.Analyzers

Describe the bug
When using Akka.Analyzers I now see the following warning (which in our solution turns into an error because we threat warnings as errors).

The code fix however suggests the following 'fix'. But outside the ReceiveAsync (in the constructor of my actor in this case) I don't have a valid Sender. So this code fix is wrong IMO (please explain if I'm wrong).

Expected behavior
I expect that this is the correct way to fix it:

Additional question is (if I need to create a separate issue for this, please let me know), why is this wrong?

As I've been using this over the past years on a lot of places and it always was working fine. We however always await tasks in our actors and never do async work in our actors which we don't await.