Never forward an agent command to the node it is already on, and split the bool

src/Testing/CoreTests/Runtime/Agents/InvokeOnAgentOrForwardAsyncTests.cs

@@ -133,3 +133,128 @@ public async Task should_return_false_when_agent_not_found_anywhere()
         actionCalled.ShouldBeFalse();
     }
 }
+
+/// <summary>
+/// CritterWatch#1171 — the two questions this routing asks are answered by two different sources, and
+/// nothing handled them disagreeing.
+/// </summary>
+/// <remarks>
+/// <para>
+/// <b>"Run it here?"</b> is answered by the in-process <c>NodeController.Agents</c> dictionary
+/// (<see cref="IAgentRuntime.AllRunningAgentUris"/>). <b>"Forward it where?"</b> is answered by the
+/// durable <c>wolverine_nodes</c> table. During a startup window the table is already populated while
+/// the dictionary is still empty — so on a single-node service the envelope was sent to
+/// <c>node.ControlUri</c>, which is this node, where it took the same branch again. The method
+/// returned <c>true</c> for that, and every caller reads <c>true</c> as "done".
+/// </para>
+/// <para>
+/// Proven downstream by sampling both sources every 250 ms across 10 isolated runs: 2 failed, and one
+/// variable separated them — the agent was running on all 8 passes and on neither failure. The
+/// commands were acked as successful and the daemon never acted, which read as a daemon defect for
+/// weeks because the URI resolution that precedes this goes through the store's shard REGISTRY and
+/// answers happily either way.
+/// </para>
+/// </remarks>
+public class no_self_forward_when_the_node_table_and_the_agent_dictionary_disagree
+{
+    private readonly MockWolverineRuntime _runtime = new();
+    private readonly Uri _agentUri = new("typedfake://alpha");
+
+    /// <summary>The node table claims the agent for THIS node while it is not running here.</summary>
+    private void TableClaimsAgentFor(Guid nodeId, Uri? controlUri)
+    {
+        _runtime.Agents.AllRunningAgentUris().Returns(Array.Empty<Uri>());
+        _runtime.Storage.Nodes.LoadAllNodesAsync(Arg.Any<CancellationToken>())
+            .Returns(new List<WolverineNode>
+            {
+                new() { NodeId = nodeId, ControlUri = controlUri, ActiveAgents = { _agentUri } }
+            });
+    }
+
+    [Fact]
+    public async Task does_not_forward_to_itself_and_says_so()
+    {
+        TableClaimsAgentFor(_runtime.Options.UniqueNodeId, new Uri("dbcontrol://one"));
+
+        var context = new MessageContext(_runtime);
+        var actionCalled = false;
+
+        var outcome = await context.InvokeOnAgentAsync(_agentUri, (_, _) =>
+        {
+            actionCalled = true;
+            return Task.CompletedTask;
+        }, CancellationToken.None);
+
+        outcome.ShouldBe(AgentInvocationOutcome.NotRunningLocally);
+        actionCalled.ShouldBeFalse();
+    }
+
+    [Fact]
+    public async Task the_bool_overload_reports_that_as_a_failure()
+    {
+        // The whole point. It used to answer `true` here — "executed locally or forwarded" — for a
+        // command that did neither, so a caller with a working fallback skipped it and a caller
+        // without one acked a success.
+        TableClaimsAgentFor(_runtime.Options.UniqueNodeId, new Uri("dbcontrol://one"));
+
+        var context = new MessageContext(_runtime);
+
+        var result = await context.InvokeOnAgentOrForwardAsync(
+            _agentUri, (_, _) => Task.CompletedTask, CancellationToken.None);
+
+        result.ShouldBeFalse();
+    }
+
+    [Fact]
+    public async Task a_node_that_owns_the_agent_but_has_no_control_endpoint_is_not_an_owner()
+    {
+        // Guarding the null-forgiving `node!.ControlUri!` this replaced: it would have thrown a
+        // NullReferenceException from inside the routing rather than letting the caller fall back.
+        TableClaimsAgentFor(Guid.NewGuid(), controlUri: null);
+
+        var context = new MessageContext(_runtime);
+
+        var outcome = await context.InvokeOnAgentAsync(
+            _agentUri, (_, _) => Task.CompletedTask, CancellationToken.None);
+
+        outcome.ShouldBe(AgentInvocationOutcome.NoOwner);
+    }
+
+    [Fact]
+    public async Task running_locally_is_still_reported_as_executed_rather_than_forwarded()
+    {
+        // The other half of splitting the bool: "the work is done" and "somebody else will do it" are
+        // now different answers, so a caller can tell whether to wait for anything.
+        _runtime.Agents.AllRunningAgentUris().Returns([_agentUri]);
+
+        var context = new MessageContext(_runtime);
+        var actionCalled = false;
+
+        var outcome = await context.InvokeOnAgentAsync(_agentUri, (_, _) =>
+        {
+            actionCalled = true;
+            return Task.CompletedTask;
+        }, CancellationToken.None);
+
+        outcome.ShouldBe(AgentInvocationOutcome.ExecutedLocally);
+        actionCalled.ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task no_node_at_all_is_distinct_from_this_node_not_running_it()
+    {
+        // Two different failures that a single `false` collapsed together, and they want different
+        // diagnostics: nobody has been assigned the agent, versus we have been assigned it and are
+        // not running it yet.
+        _runtime.Agents.AllRunningAgentUris().Returns(Array.Empty<Uri>());
+        _runtime.Storage.Nodes.LoadAllNodesAsync(Arg.Any<CancellationToken>())
+            .Returns(new List<WolverineNode>());
+
+        var context = new MessageContext(_runtime);
+
+        var outcome = await context.InvokeOnAgentAsync(
+            _agentUri, (_, _) => Task.CompletedTask, CancellationToken.None);
+
+        outcome.ShouldBe(AgentInvocationOutcome.NoOwner);
+    }
+}

src/Wolverine/Runtime/AgentMessagingExtensions.cs

@@ -4,24 +4,53 @@
 
 namespace Wolverine.Runtime;
 
+/// <summary>
+/// What actually happened to an agent command routed by
+/// <see cref="AgentMessagingExtensions.InvokeOnAgentAsync(IMessageContext,Uri,Func{IWolverineRuntime,CancellationToken,Task},CancellationToken)"/>.
+/// </summary>
+/// <remarks>
+/// This exists because a <c>bool</c> cannot carry it. "Executed here" and "handed to another node"
+/// are both successes for the caller, but only the first means the work is done — and the two
+/// failures below are not the same failure and do not want the same fallback.
+/// </remarks>
+public enum AgentInvocationOutcome
+{
+    /// <summary>The action ran on this node. The work is done.</summary>
+    ExecutedLocally,
+
+    /// <summary>Another node owns the agent and the message was sent to it. The work is not done yet.</summary>
+    Forwarded,
+
+    /// <summary>No node in the durable node table claims the agent.</summary>
+    NoOwner,
+
+    /// <summary>
+    /// The node table says THIS node owns the agent, but it is not running here.
+    /// </summary>
+    /// <remarks>
+    /// The routing decision and the execution decision read two different sources — the durable
+    /// <c>wolverine_nodes</c> table and the in-process <c>NodeController.Agents</c> dictionary — and
+    /// during a startup window the table is populated while the dictionary is still empty. There is
+    /// nothing useful to do with the message here: forwarding it would send it to the node it is
+    /// already on, which is a silent no-op. Callers should fall back or fail honestly.
+    /// </remarks>
+    NotRunningLocally
+}
+
 /// <summary>
 /// Extension methods for routing agent commands to the correct node in a Wolverine cluster.
 /// </summary>
 public static class AgentMessagingExtensions
 {
     /// <summary>
-    /// Executes an action locally if the specified agent is running on this node,
-    /// otherwise forwards the current message to the node that owns the agent.
+    /// Executes an action locally if the specified agent is running on this node, otherwise forwards
+    /// the current message to the node that owns the agent, and reports which of those happened.
     /// </summary>
     /// <param name="context">The current message context.</param>
     /// <param name="agentUri">The URI identifying the target agent.</param>
     /// <param name="action">The action to execute if the agent is local to this node.</param>
     /// <param name="cancellationToken">Cancellation token.</param>
-    /// <returns>
-    /// <c>true</c> if the action was executed locally or the message was successfully forwarded;
-    /// <c>false</c> if no node currently owns the specified agent.
-    /// </returns>
-    public static async Task<bool> InvokeOnAgentOrForwardAsync(this IMessageContext context, Uri agentUri,
+    public static async Task<AgentInvocationOutcome> InvokeOnAgentAsync(this IMessageContext context, Uri agentUri,
         Func<IWolverineRuntime, CancellationToken, Task> action, CancellationToken cancellationToken)
     {
         var messageContext = context.As<MessageContext>();
@@ -30,35 +59,56 @@ public static async Task<bool> InvokeOnAgentOrForwardAsync(this IMessageContext
         if (runtime.Agents.AllRunningAgentUris().Contains(agentUri))
         {
             await action(runtime, cancellationToken);
-            return true;
+            return AgentInvocationOutcome.ExecutedLocally;
         }
 
         var all = await runtime.Storage.Nodes.LoadAllNodesAsync(cancellationToken);
         var node = all.FirstOrDefault(x => x.ActiveAgents.Contains(agentUri));
 
-        if (node == null) return false;
+        if (node == null) return AgentInvocationOutcome.NoOwner;
 
-        await messageContext.EndpointFor(node!.ControlUri!).SendAsync(context.Envelope!.Message);
-        return true;
+        // Never forward to ourselves. The check above asked the in-process agent dictionary and this
+        // one asks the durable node table; when they disagree — the table already claims the agent
+        // while the dictionary is still filling during startup — sending the envelope to
+        // node.ControlUri delivers it back to this node, where it takes the same branch again. The
+        // caller was previously told "true" for that, so a command that did nothing at all was
+        // reported as handled. FanOutToAllNodes below has always excluded self for the same reason.
+        if (node.NodeId == runtime.Options.UniqueNodeId)
+        {
+            runtime.Logger.LogWarning(
+                "Node {NodeId} is recorded as the owner of agent {AgentUri} in node storage, but that agent is not running on this node. Not forwarding the message to ourselves.",
+                node.NodeId, agentUri);
+
+            return AgentInvocationOutcome.NotRunningLocally;
+        }
+
+        if (node.ControlUri == null)
+        {
+            runtime.Logger.LogWarning(
+                "Node {NodeId} owns agent {AgentUri} but has no control endpoint, so the message cannot be forwarded to it",
+                node.NodeId, agentUri);
+
+            return AgentInvocationOutcome.NoOwner;
+        }
+
+        await messageContext.EndpointFor(node.ControlUri).SendAsync(context.Envelope!.Message);
+        return AgentInvocationOutcome.Forwarded;
     }
 
     /// <summary>
     /// Executes an action on a specific running agent (cast to type T) if the agent is on this node,
-    /// otherwise forwards the current message to the node that owns the agent.
+    /// otherwise forwards the current message to the node that owns the agent, and reports which of
+    /// those happened.
     /// </summary>
     /// <typeparam name="T">The expected agent type implementing IAgent.</typeparam>
     /// <param name="context">The current message context.</param>
     /// <param name="agentUri">The URI identifying the target agent.</param>
     /// <param name="action">The action to execute on the typed agent if found locally.</param>
     /// <param name="cancellationToken">Cancellation token.</param>
-    /// <returns>
-    /// <c>true</c> if the action was executed locally or the message was successfully forwarded;
-    /// <c>false</c> if no node currently owns the specified agent.
-    /// </returns>
-    public static Task<bool> InvokeOnAgentOrForwardAsync<T>(this IMessageContext context, Uri agentUri,
+    public static Task<AgentInvocationOutcome> InvokeOnAgentAsync<T>(this IMessageContext context, Uri agentUri,
         Func<T, Task> action, CancellationToken cancellationToken) where T : class, IAgent
     {
-        return context.InvokeOnAgentOrForwardAsync(agentUri, async (runtime, ct) =>
+        return context.InvokeOnAgentAsync(agentUri, async (runtime, ct) =>
         {
             if (runtime.Agents.TryFindActiveAgent<T>(agentUri, out var agent))
             {
@@ -73,6 +123,51 @@ public static Task<bool> InvokeOnAgentOrForwardAsync<T>(this IMessageContext con
         }, cancellationToken);
     }
 
+    /// <summary>
+    /// Executes an action locally if the specified agent is running on this node,
+    /// otherwise forwards the current message to the node that owns the agent.
+    /// </summary>
+    /// <param name="context">The current message context.</param>
+    /// <param name="agentUri">The URI identifying the target agent.</param>
+    /// <param name="action">The action to execute if the agent is local to this node.</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>
+    /// <c>true</c> if the action was executed locally or the message was successfully forwarded;
+    /// <c>false</c> if the command could not be delivered to a running agent.
+    /// </returns>
+    /// <remarks>
+    /// ⚠️ A caller that needs to know whether the WORK IS DONE cannot use this — <c>true</c> also
+    /// means "handed to another node, which will do it later". Use
+    /// <see cref="InvokeOnAgentAsync(IMessageContext,Uri,Func{IWolverineRuntime,CancellationToken,Task},CancellationToken)"/>
+    /// and read <see cref="AgentInvocationOutcome"/>.
+    /// </remarks>
+    public static async Task<bool> InvokeOnAgentOrForwardAsync(this IMessageContext context, Uri agentUri,
+        Func<IWolverineRuntime, CancellationToken, Task> action, CancellationToken cancellationToken)
+    {
+        var outcome = await context.InvokeOnAgentAsync(agentUri, action, cancellationToken);
+        return outcome is AgentInvocationOutcome.ExecutedLocally or AgentInvocationOutcome.Forwarded;
+    }
+
+    /// <summary>
+    /// Executes an action on a specific running agent (cast to type T) if the agent is on this node,
+    /// otherwise forwards the current message to the node that owns the agent.
+    /// </summary>
+    /// <typeparam name="T">The expected agent type implementing IAgent.</typeparam>
+    /// <param name="context">The current message context.</param>
+    /// <param name="agentUri">The URI identifying the target agent.</param>
+    /// <param name="action">The action to execute on the typed agent if found locally.</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>
+    /// <c>true</c> if the action was executed locally or the message was successfully forwarded;
+    /// <c>false</c> if the command could not be delivered to a running agent.
+    /// </returns>
+    public static async Task<bool> InvokeOnAgentOrForwardAsync<T>(this IMessageContext context, Uri agentUri,
+        Func<T, Task> action, CancellationToken cancellationToken) where T : class, IAgent
+    {
+        var outcome = await context.InvokeOnAgentAsync(agentUri, action, cancellationToken);
+        return outcome is AgentInvocationOutcome.ExecutedLocally or AgentInvocationOutcome.Forwarded;
+    }
+
     /// <summary>
     /// Publishes a message locally and sends it to every other node in the cluster
     /// via each node's control URI. Node data is loaded fresh from persistence (no caching).
@@ -100,4 +195,4 @@ public static async Task FanOutToAllNodes(this IMessageContext context, object m
             }
         }
     }
-}
+}