Stable node identity + lifecycle for storeless Solo hosts (#3188)

src/Testing/CoreTests/Runtime/Heartbeat/solo_storeless_node_identity.cs

@@ -0,0 +1,87 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using NSubstitute;
+using Shouldly;
+using Wolverine;
+using Wolverine.Runtime;
+using Wolverine.Runtime.Agents;
+using Xunit;
+
+namespace CoreTests.Runtime.Heartbeat;
+
+/// <summary>
+/// #3188 — a Solo host with no durable message store (NullMessageStore) never runs the
+/// NodeAgentController path, so before the fix it kept a random per-process AssignedNodeNumber
+/// and never emitted NodeStarted()/NodeStopped(). The stable identity is now assigned by the
+/// runtime before transports start, and SoloHeartbeatService owns the lifecycle bookends.
+/// </summary>
+public class solo_storeless_node_identity
+{
+    private static IHost BuildHost(Action<WolverineOptions> configure)
+    {
+        return Host.CreateDefaultBuilder()
+            .UseWolverine(configure)
+            .Build();
+    }
+
+    [Fact]
+    public async Task storeless_solo_gets_node_1_and_fires_lifecycle_bookends()
+    {
+        using var host = BuildHost(opts => opts.Durability.Mode = DurabilityMode.Solo);
+
+        var runtime = host.Services.GetRequiredService<IWolverineRuntime>();
+        var observer = Substitute.For<IWolverineObserver>();
+        runtime.Observer = observer;
+
+        await host.StartAsync();
+
+        // Stable identity, set before the messaging transports start.
+        runtime.Options.Durability.AssignedNodeNumber.ShouldBe(1);
+
+        // NodeStarted fires once, while transports are up; NodeStopped not yet.
+        await observer.Received(1).NodeStarted();
+        await observer.DidNotReceive().NodeStopped();
+
+        await host.StopAsync();
+
+        await observer.Received(1).NodeStopped();
+    }
+
+    [Fact]
+    public async Task storeless_solo_node_number_is_stable_across_restart()
+    {
+        async Task<int> bootAndReadNodeNumber()
+        {
+            using var host = BuildHost(opts => opts.Durability.Mode = DurabilityMode.Solo);
+            await host.StartAsync();
+            var number = host.Services.GetRequiredService<IWolverineRuntime>()
+                .Options.Durability.AssignedNodeNumber;
+            await host.StopAsync();
+            return number;
+        }
+
+        var first = await bootAndReadNodeNumber();
+        var second = await bootAndReadNodeNumber();
+
+        first.ShouldBe(1);
+        second.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task does_not_fire_for_a_storeless_non_solo_host()
+    {
+        // Default mode is Balanced; a storeless Balanced host has no NodeAgentController and must
+        // NOT get the Solo lifecycle bookends.
+        using var host = BuildHost(_ => { });
+
+        var runtime = host.Services.GetRequiredService<IWolverineRuntime>();
+        var observer = Substitute.For<IWolverineObserver>();
+        runtime.Observer = observer;
+
+        await host.StartAsync();
+        await host.StopAsync();
+
+        await observer.DidNotReceive().NodeStarted();
+        await observer.DidNotReceive().NodeStopped();
+    }
+}

src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransport.cs

@@ -166,8 +166,17 @@ protected override void tryBuildSystemEndpoints(IWolverineRuntime runtime)
         // and SQS queue names are case-sensitive. Without this, the sender creates
         // "wolverine-response-MyApp-123" but the receiver resolves the reply URI
         // to "wolverine-response-myapp-123" (lowercased by Uri), creating a different queue.
+        // The per-node response queue must be unique to this running node. In Solo mode the assigned
+        // node number is always 1 (#3188), and the service name is not unique per host, so multiple
+        // Solo hosts (e.g. the request/reply compliance sender + receiver, or successive fixtures on
+        // one broker) would share one response queue and cross-deliver each other's replies. Use the
+        // always-unique UniqueNodeId in Solo; Balanced gets a unique AssignedNodeNumber via election.
+        // See #3189.
+        var responseNode = runtime.Options.Durability.Mode == DurabilityMode.Solo
+            ? runtime.Options.UniqueNodeId.ToString("N")
+            : runtime.DurabilitySettings.AssignedNodeNumber.ToString();
         var responseName = SanitizeSqsName(
-            $"wolverine.response.{runtime.Options.ServiceName}.{runtime.DurabilitySettings.AssignedNodeNumber}")
+            $"wolverine.response.{runtime.Options.ServiceName}.{responseNode}")
             .ToLowerInvariant();
 
         var queue = Queues[responseName];

src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransportConfiguration.cs

@@ -240,9 +240,14 @@ public AmazonSqsTransportConfiguration EnableWolverineControlQueues()
     {
         Transport.SystemQueuesEnabled = true;
 
-        // Lowercase to match URI normalization (see tryBuildSystemEndpoints comment)
+        // Lowercase to match URI normalization (see tryBuildSystemEndpoints comment). In Solo mode the
+        // assigned node number is always 1 (#3188), so key the per-node control queue on the unique
+        // node id to keep multiple Solo hosts on one broker from colliding. See #3189.
+        var controlNode = Options.Durability.Mode == DurabilityMode.Solo
+            ? Options.UniqueNodeId.ToString("N")
+            : Options.Durability.AssignedNodeNumber.ToString();
         var queueName = AmazonSqsTransport.SanitizeSqsName(
-            "wolverine.control." + Options.Durability.AssignedNodeNumber)
+            "wolverine.control." + controlNode)
             .ToLowerInvariant();
 
         var queue = Transport.Queues[queueName];

src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusConfiguration.cs

@@ -248,7 +248,12 @@ public AzureServiceBusConfiguration SystemQueuesAreEnabled(bool enabled)
     /// <returns></returns>
     public AzureServiceBusConfiguration EnableWolverineControlQueues()
     {
-        var queueName = "wolverine.control." + Options.Durability.AssignedNodeNumber;
+        // In Solo mode the assigned node number is always 1 (#3188); key the per-node control queue
+        // on the unique node id so multiple Solo hosts on one namespace don't collide. See #3189.
+        var controlNode = Options.Durability.Mode == DurabilityMode.Solo
+            ? Options.UniqueNodeId.ToString("N")
+            : Options.Durability.AssignedNodeNumber.ToString();
+        var queueName = "wolverine.control." + controlNode;
 
         var queue = Transport.Queues[queueName];
 

src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.cs

@@ -199,7 +199,14 @@ protected override void tryBuildSystemEndpoints(IWolverineRuntime runtime)
     {
         if (!SystemQueuesEnabled) return;
 
-        var queueName = $"wolverine.response.{runtime.Options.ServiceName}.{runtime.DurabilitySettings.AssignedNodeNumber}";
+        // In Solo mode the assigned node number is always 1 (#3188) and the service name is not
+        // unique per host, so multiple Solo hosts on one namespace would share a response queue and
+        // cross-deliver replies — key on the always-unique UniqueNodeId instead. Balanced gets a
+        // unique AssignedNodeNumber via election. See #3189.
+        var responseNode = runtime.Options.Durability.Mode == DurabilityMode.Solo
+            ? runtime.Options.UniqueNodeId.ToString("N")
+            : runtime.DurabilitySettings.AssignedNodeNumber.ToString();
+        var queueName = $"wolverine.response.{runtime.Options.ServiceName}.{responseNode}";
 
         var queue = Queues[queueName];
 

src/Transports/GCP/Wolverine.Pubsub/PubsubTransport.cs

@@ -134,7 +134,15 @@ protected override void tryBuildSystemEndpoints(IWolverineRuntime runtime)
             return;
         }
 
-        var responseName = $"{ResponseName}.{Math.Abs(runtime.DurabilitySettings.AssignedNodeNumber)}";
+        // The per-node response endpoint must be unique to this running node. In Solo mode the
+        // assigned node number is always 1 (#3188), so several Solo services sharing a project would
+        // collide on the same response subscription and cross-deliver each other's replies — use the
+        // always unique UniqueNodeId instead. Balanced nodes get a unique AssignedNodeNumber via
+        // election, so they keep the existing node-number name. See #3189.
+        var responseNode = runtime.Options.Durability.Mode == DurabilityMode.Solo
+            ? runtime.Options.UniqueNodeId.ToString("N")
+            : Math.Abs(runtime.DurabilitySettings.AssignedNodeNumber).ToString();
+        var responseName = $"{ResponseName}.{responseNode}";
         var responseTopic = new PubsubEndpoint(responseName, this, EndpointRole.System);
 
         responseTopic.IsListener = responseTopic.IsUsedForReplies = true;

src/Transports/MQTT/Wolverine.MQTT.Tests/buffered_compliance.cs

@@ -69,7 +69,9 @@ public async Task has_response_topic_automatically()
         var transport = options.Transports
             .GetOrCreate<MqttTransport>();
 
-        transport.ResponseTopic.ShouldBe("wolverine/response/" + options.Durability.AssignedNodeNumber);
+        // Solo mode keys the reply topic on the unique node id (not the always-1 assigned node
+        // number) so multiple Solo services on one broker don't collide. See #3189.
+        transport.ResponseTopic.ShouldBe("wolverine/response/" + options.UniqueNodeId.ToString("N"));
 
         transport.ReplyEndpoint().ShouldBeOfType<MqttTopic>().TopicName.ShouldBe(transport.ResponseTopic);
     }

src/Transports/MQTT/Wolverine.MQTT.Tests/inline_compliance.cs

@@ -69,7 +69,9 @@ public void has_response_topic_automatically()
         var transport = options.Transports
             .GetOrCreate<MqttTransport>();
 
-        transport.ResponseTopic.ShouldBe("wolverine/response/" + options.Durability.AssignedNodeNumber);
+        // Solo mode keys the reply topic on the unique node id (not the always-1 assigned node
+        // number) so multiple Solo services on one broker don't collide. See #3189.
+        transport.ResponseTopic.ShouldBe("wolverine/response/" + options.UniqueNodeId.ToString("N"));
 
         transport.ReplyEndpoint().ShouldBeOfType<MqttTopic>().TopicName.ShouldBe(transport.ResponseTopic);
     }

src/Transports/MQTT/Wolverine.MQTT/Internals/MqttTransport.cs

@@ -49,7 +49,15 @@ protected override MqttTopic findEndpointByUri(Uri uri)
 
     public override async ValueTask InitializeAsync(IWolverineRuntime runtime)
     {
-        ResponseTopic = "wolverine/response/" + runtime.Options.Durability.AssignedNodeNumber;
+        // The per-node reply topic must be unique to this running node. In Solo mode the assigned
+        // node number is always 1 (#3188), so several Solo services on one broker would collide on
+        // the same topic and cross-deliver each other's replies — use the always unique
+        // UniqueNodeId instead. Balanced nodes get a unique AssignedNodeNumber via election, so they
+        // keep the existing, more readable topic. See #3189.
+        var responseNode = runtime.Options.Durability.Mode == DurabilityMode.Solo
+            ? runtime.Options.UniqueNodeId.ToString("N")
+            : runtime.Options.Durability.AssignedNodeNumber.ToString();
+        ResponseTopic = "wolverine/response/" + responseNode;
         var mqttTopic = Topics[ResponseTopic];
         mqttTopic.IsUsedForReplies = true;
         mqttTopic.IsListener = true;

src/Transports/NATS/Wolverine.Nats/Internal/NatsTransport.cs

@@ -77,7 +77,15 @@ public override async ValueTask ConnectAsync(IWolverineRuntime runtime)
     {
         _logger = runtime.LoggerFactory.CreateLogger<NatsTransport>();
 
-        ResponseSubject = $"wolverine.response.{runtime.Options.Durability.AssignedNodeNumber}";
+        // The per-node reply subject must be unique to this running node. In Solo mode the
+        // assigned node number is always 1 (#3188), so several Solo services on one broker would
+        // collide on the same subject and cross-deliver each other's replies — use the always
+        // unique UniqueNodeId instead. Balanced nodes get a unique AssignedNodeNumber via election,
+        // so they keep the existing, more readable subject. See #3189.
+        var responseNode = runtime.Options.Durability.Mode == DurabilityMode.Solo
+            ? runtime.Options.UniqueNodeId.ToString("N")
+            : runtime.Options.Durability.AssignedNodeNumber.ToString();
+        ResponseSubject = $"wolverine.response.{responseNode}";
         var responseEndpoint = _endpoints[ResponseSubject];
         responseEndpoint.IsUsedForReplies = true;
         responseEndpoint.IsListener = true;

src/Transports/Redis/Wolverine.Redis/Internal/RedisTransport.cs

@@ -412,8 +412,15 @@ protected override void tryBuildSystemEndpoints(IWolverineRuntime runtime)
     {
         if (!SystemQueuesEnabled) return;
 
-        // Create a per-node reply stream endpoint similar to other transports (using database 0)
-        var replyStreamKey = $"wolverine.response.{runtime.Options.ServiceName}.{runtime.DurabilitySettings.AssignedNodeNumber}".ToLowerInvariant();
+        // Create a per-node reply stream endpoint similar to other transports (using database 0).
+        // In Solo mode the assigned node number is always 1 (#3188) and the service name is not
+        // unique per host, so multiple Solo hosts on one Redis would share a reply stream and
+        // cross-deliver replies — key on the always-unique UniqueNodeId instead. Balanced gets a
+        // unique AssignedNodeNumber via election. See #3189.
+        var replyNode = runtime.Options.Durability.Mode == DurabilityMode.Solo
+            ? runtime.Options.UniqueNodeId.ToString("N")
+            : runtime.DurabilitySettings.AssignedNodeNumber.ToString();
+        var replyStreamKey = $"wolverine.response.{runtime.Options.ServiceName}.{replyNode}".ToLowerInvariant();
         var cacheKey = $"{ReplyDatabaseId}:{replyStreamKey}";
         var replyEndpoint = new RedisStreamEndpoint(
             new Uri($"{ProtocolName}://stream/{ReplyDatabaseId}/{replyStreamKey}?consumerGroup=wolverine-replies"),

src/Wolverine/HostBuilderExtensions.cs

@@ -27,6 +27,7 @@
 using Wolverine.Runtime;
 using Wolverine.Runtime.Agents;
 using Wolverine.Runtime.Handlers;
+using Wolverine.Runtime.Heartbeat;
 
 namespace Wolverine;
 
@@ -204,6 +205,13 @@ internal static IServiceCollection AddWolverine(this IServiceCollection services
         // The runtime is also a hosted service
         services.AddSingleton(s => (IHostedService)s.GetRequiredService<IWolverineRuntime>());
 
+        // Lightweight Solo liveness bookends for a storeless Solo host (#3188). Registered AFTER
+        // the runtime hosted service so NodeStarted() fires once transports are up, and (hosted
+        // services stop in reverse) NodeStopped() fires before the runtime tears them down.
+        // No-ops unless the host is Solo with a NullMessageStore.
+        services.AddSingleton<SoloHeartbeatService>();
+        services.AddSingleton(s => (IHostedService)s.GetRequiredService<SoloHeartbeatService>());
+
         services.MessagingRootService(x => x.MessageTracking);
 
         services.AddSingleton<InMemorySagaPersistor>();

src/Wolverine/Runtime/Agents/IWolverineObserver.cs

@@ -130,12 +130,27 @@ await _runtime.Storage.Nodes.LogRecordsAsync(NodeRecord.For(_runtime.Options,
 
     public async Task NodeStarted()
     {
-        await _runtime.Storage.Nodes.LogRecordsAsync(NodeRecord.For(_runtime.Options, NodeRecordType.NodeStarted));
+        try
+        {
+            await _runtime.Storage.Nodes.LogRecordsAsync(NodeRecord.For(_runtime.Options, NodeRecordType.NodeStarted));
+        }
+        catch (NotSupportedException)
+        {
+            // NullMessageStore does not support node persistence; a storeless Solo node still
+            // fires this bookend (via SoloHeartbeatService) but has nowhere to log it. See #3188.
+        }
     }
 
     public async Task NodeStopped()
     {
-        await _runtime.Storage.Nodes.LogRecordsAsync(NodeRecord.For(_runtime.Options, NodeRecordType.NodeStopped));
+        try
+        {
+            await _runtime.Storage.Nodes.LogRecordsAsync(NodeRecord.For(_runtime.Options, NodeRecordType.NodeStopped));
+        }
+        catch (NotSupportedException)
+        {
+            // NullMessageStore does not support node persistence; nothing to log.
+        }
     }
 
     public async Task AgentStarted(Uri agentUri)

src/Wolverine/Runtime/Heartbeat/SoloHeartbeatService.cs

@@ -0,0 +1,62 @@
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Wolverine.Persistence.Durability;
+
+namespace Wolverine.Runtime.Heartbeat;
+
+/// <summary>
+/// Lightweight, store-independent liveness for a Solo host that has no durable message store.
+/// A storeless Solo node never runs the heavyweight <c>NodeAgentController</c> path
+/// (<c>StartSoloModeAsync</c> sits behind the <see cref="NullMessageStore"/> early return), so
+/// without this it would never emit the <c>NodeStarted()</c>/<c>NodeStopped()</c> lifecycle
+/// bookends and would heartbeat under an unstable, per-process node id. See #3188.
+///
+/// The stable node number itself is assigned by the runtime <em>before</em> the messaging
+/// transports start (envelope ownership and the heartbeat both read it); this service only owns
+/// the lifecycle bookends. It is registered <em>after</em> the runtime hosted service so
+/// <c>NodeStarted()</c> fires once transports are up, and — because hosted services stop in
+/// reverse order — <c>NodeStopped()</c> fires before the runtime tears the transports down, so a
+/// publish-based observer (e.g. CritterWatch) can still emit the final record.
+/// </summary>
+internal class SoloHeartbeatService : IHostedService
+{
+    private readonly IWolverineRuntime _runtime;
+    private readonly ILogger<SoloHeartbeatService> _logger;
+    private bool _started;
+
+    public SoloHeartbeatService(IWolverineRuntime runtime, ILogger<SoloHeartbeatService> logger)
+    {
+        _runtime = runtime;
+        _logger = logger;
+    }
+
+    // Only a Solo host with no durable store needs this — every other shape already gets its
+    // node identity and lifecycle through the NodeAgentController path.
+    private bool ShouldRun => _runtime.Options.Durability.Mode == DurabilityMode.Solo
+                              && _runtime.Storage is NullMessageStore;
+
+    public async Task StartAsync(CancellationToken cancellationToken)
+    {
+        if (!ShouldRun)
+        {
+            return;
+        }
+
+        _started = true;
+        await _runtime.Observer.NodeStarted();
+
+        _logger.LogInformation("Solo node {NodeNumber} started without a durable message store",
+            _runtime.Options.Durability.AssignedNodeNumber);
+    }
+
+    public async Task StopAsync(CancellationToken cancellationToken)
+    {
+        if (!_started)
+        {
+            return;
+        }
+
+        _started = false;
+        await _runtime.Observer.NodeStopped();
+    }
+}

src/Wolverine/Runtime/WolverineRuntime.Agents.cs

@@ -149,6 +149,18 @@ private async Task startAgentsAsync()
     {
         if (Storage is NullMessageStore)
         {
+            // A Solo host is always logically node 1, even with no durable store to coordinate
+            // through. StartSoloModeAsync() — the only place a stored Solo node gets its number —
+            // sits past this early return, so without this a storeless Solo host keeps its random
+            // per-process default and leaks a churning id into heartbeats / envelope ownership.
+            // Identity must be set here, before the messaging transports start; the matching
+            // NodeStarted()/NodeStopped() lifecycle bookends are owned by SoloHeartbeatService so
+            // they fire while transports are up. See #3188.
+            if (Options.Durability.Mode == DurabilityMode.Solo)
+            {
+                Options.Durability.AssignedNodeNumber = 1;
+            }
+
             return;
         }