GH-5003 Resume continuous execution after a projection rebuild, even a failed one

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

@@ -0,0 +1,203 @@
+using JasperFx;
+using JasperFx.Descriptors;
+using JasperFx.Events;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Descriptors;
+using JasperFx.Events.Projections;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using NSubstitute.ExceptionExtensions;
+using Shouldly;
+using Wolverine.Runtime.Agents;
+using Xunit;
+using JasperFxSubscriptionAgent = JasperFx.Events.Daemon.ISubscriptionAgent;
+
+namespace CoreTests.Runtime.Agents;
+
+/// <summary>
+/// GH-5003. A projection rebuild hard-stops the continuous agent and tears down the projection's state
+/// before it replays. When the rebuild threw, nothing restarted the shard: the live-agent path skipped
+/// its resume, and the registered-projection path (GH-3163) never resumed at all, leaving it to
+/// NodeAgentController's sweep -- which only reports a shard whose pre-rebuild agent carried an Other
+/// failure, so that one stayed stopped until the process restarted.
+/// </summary>
+public class failed_rebuild_resumes_continuous_execution
+{
+    public class through_the_live_agent
+    {
+        private readonly IProjectionDaemon _daemon = Substitute.For<IProjectionDaemon>();
+        private readonly ShardName _shardName = new("Incident");
+        private readonly EventSubscriptionAgent _agent;
+
+        public through_the_live_agent()
+        {
+            _daemon.StartAgentAsync(Arg.Any<ShardName>(), Arg.Any<CancellationToken>())
+                .Returns(Substitute.For<JasperFxSubscriptionAgent>());
+            _agent = new EventSubscriptionAgent(
+                new Uri("event-subscriptions://marten/incident/all"), _shardName, _daemon);
+        }
+
+        [Fact]
+        public async Task a_rebuild_that_throws_still_resumes_continuous_execution_and_still_throws()
+        {
+            await _agent.StartAsync(CancellationToken.None);
+            _daemon.RebuildProjectionAsync(Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+                .ThrowsAsync(new InvalidOperationException("the connection pool has been exhausted"));
+
+            var ex = await Should.ThrowAsync<InvalidOperationException>(() => _agent.RebuildAsync(CancellationToken.None));
+            ex.Message.ShouldBe("the connection pool has been exhausted");
+
+            // Once for the original start, once for the resume after the failed rebuild
+            await _daemon.Received(2).StartAgentAsync(_shardName, Arg.Any<CancellationToken>());
+            _agent.Status.ShouldBe(AgentStatus.Running);
+        }
+
+        [Fact]
+        public async Task a_failed_resume_does_not_hide_why_the_rebuild_failed()
+        {
+            await _agent.StartAsync(CancellationToken.None);
+            _daemon.RebuildProjectionAsync(Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+                .ThrowsAsync(new InvalidOperationException("rebuild failed"));
+            _daemon.StartAgentAsync(Arg.Any<ShardName>(), Arg.Any<CancellationToken>())
+                .ThrowsAsync(new TimeoutException("resume failed"));
+
+            var ex = await Should.ThrowAsync<InvalidOperationException>(() => _agent.RebuildAsync(CancellationToken.None));
+            ex.Message.ShouldBe("rebuild failed");
+        }
+
+        [Fact]
+        public async Task a_cancelled_rebuild_is_not_resumed()
+        {
+            await _agent.StartAsync(CancellationToken.None);
+            using var cancellation = new CancellationTokenSource();
+            _daemon.RebuildProjectionAsync(Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+                .Returns(async _ =>
+                {
+                    await cancellation.CancelAsync();
+                    throw new OperationCanceledException(cancellation.Token);
+                });
+
+            await Should.ThrowAsync<OperationCanceledException>(() => _agent.RebuildAsync(cancellation.Token));
+
+            // Only the original start: the host is shutting down or somebody called the rebuild off
+            await _daemon.Received(1).StartAgentAsync(_shardName, Arg.Any<CancellationToken>());
+        }
+
+        [Fact]
+        public void is_rebuilding_comes_from_the_daemon()
+        {
+            _agent.IsRebuilding.ShouldBeFalse();
+
+            _daemon.IsRebuilding(_shardName).Returns(true);
+
+            _agent.IsRebuilding.ShouldBeTrue();
+        }
+    }
+
+    public class through_the_registered_projection_rebuild
+    {
+        private readonly DatabaseId theDatabaseId = new("localhost", "claims1");
+        private readonly IProjectionDaemon theDaemon = Substitute.For<IProjectionDaemon>();
+        private readonly EventStoreAgents theAgents;
+
+        public through_the_registered_projection_rebuild()
+        {
+            theDaemon.Tracker.Returns(new ShardStateTracker(NullLogger.Instance));
+            theDaemon.StartAgentAsync(Arg.Any<ShardName>(), Arg.Any<CancellationToken>())
+                .Returns(Substitute.For<JasperFxSubscriptionAgent>());
+
+            var usage = new EventStoreUsage
+            {
+                Database = new DatabaseUsage
+                {
+                    MainDatabase = new DatabaseDescriptor { ServerName = "localhost", DatabaseName = "claims1" }
+                },
+                Subscriptions =
+                {
+                    new SubscriptionDescriptor(SubscriptionType.SingleStreamProjection)
+                    {
+                        Name = "trips",
+                        Lifecycle = ProjectionLifecycle.Async,
+                        ShardNames = [ShardName.Compose("trips")]
+                    },
+                    new SubscriptionDescriptor(SubscriptionType.SingleStreamProjection)
+                    {
+                        Name = "invoices",
+                        Lifecycle = ProjectionLifecycle.Async,
+                        ShardNames = [ShardName.Compose("invoices")]
+                    }
+                }
+            };
+
+            var store = Substitute.For<IEventStore>();
+            store.Identity.Returns(new EventStoreIdentity("main", "marten"));
+            store.TryCreateUsage(Arg.Any<CancellationToken>()).Returns(Task.FromResult<EventStoreUsage?>(usage));
+            store.BuildProjectionDaemonAsync(Arg.Any<DatabaseId>())
+                .Returns(new ValueTask<IProjectionDaemon>(theDaemon));
+
+            theAgents = new EventStoreAgents(store, []);
+        }
+
+        private async Task<EventSubscriptionAgent> startAgentAsync(string projectionName)
+        {
+            var shardPath = ShardName.Compose(projectionName).RelativeUrl;
+            var uri = new Uri($"event://marten/main/localhost/claims1/{shardPath}");
+            var agent = await theAgents.BuildAgentAsync(uri, theDatabaseId, shardPath);
+            await agent.StartAsync(CancellationToken.None);
+            return agent;
+        }
+
+        private Task startsOf(string projectionName, int expected)
+            => theDaemon.Received(expected).StartAgentAsync(
+                Arg.Is<ShardName>(x => x.Name == projectionName), Arg.Any<CancellationToken>());
+
+        [Fact]
+        public async Task resumes_this_nodes_agents_for_the_rebuilt_projection_only()
+        {
+            await startAgentAsync("trips");
+            await startAgentAsync("invoices");
+
+            (await theAgents.TryRebuildRegisteredProjectionAsync("trips:All", null, CancellationToken.None))
+                .ShouldBeTrue();
+
+            await startsOf("trips", 2);
+            await startsOf("invoices", 1);
+        }
+
+        [Fact]
+        public async Task resumes_them_when_the_rebuild_throws_and_still_throws()
+        {
+            var trips = await startAgentAsync("trips");
+            theDaemon.RebuildProjectionAsync(Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+                .ThrowsAsync(new InvalidOperationException("the connection pool has been exhausted"));
+
+            await Should.ThrowAsync<InvalidOperationException>(() =>
+                theAgents.TryRebuildRegisteredProjectionAsync("trips:All", null, CancellationToken.None));
+
+            await startsOf("trips", 2);
+            trips.Status.ShouldBe(AgentStatus.Running);
+        }
+
+        [Fact]
+        public async Task does_not_resume_an_agent_that_was_stopped_on_purpose()
+        {
+            var trips = await startAgentAsync("trips");
+            await trips.StopAsync(CancellationToken.None);
+
+            (await theAgents.TryRebuildRegisteredProjectionAsync("trips:All", null, CancellationToken.None))
+                .ShouldBeTrue();
+
+            await startsOf("trips", 1);
+            trips.Status.ShouldBe(AgentStatus.Stopped);
+        }
+
+        [Fact]
+        public async Task starts_nothing_for_a_projection_with_no_agent_on_this_node()
+        {
+            (await theAgents.TryRebuildRegisteredProjectionAsync("trips:All", null, CancellationToken.None))
+                .ShouldBeTrue();
+
+            await theDaemon.DidNotReceive().StartAgentAsync(Arg.Any<ShardName>(), Arg.Any<CancellationToken>());
+        }
+    }
+}

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

@@ -415,6 +415,56 @@ public async Task restarts_a_locally_owned_shard_that_stopped_with_nothing_to_re
             agent.StartCount.ShouldBe(2);
         }
 
+        // GH-5003. A rebuild hard-stops the continuous agent and leaves its old failure in place, so for
+        // the whole replay the shard reads exactly like the stopped shards the sweep restarts. Restarting
+        // it would put a continuous agent on the same progression row as the rebuild.
+
+        [Theory]
+        [InlineData(false)]
+        [InlineData(true)]
+        public async Task leaves_a_shard_alone_while_its_projection_is_being_rebuilt(bool hadTransientFailure)
+        {
+            var uri = new Uri("event-subscriptions://marten/incident/all");
+            var agent = new FakeSubscriptionAgent(uri);
+            var controller = controllerFor(agent);
+
+            await controller.StartAgentAsync(uri);
+
+            agent.IsRebuilding = true;
+            agent.SimulateFailure(AgentStatus.Stopped, hadTransientFailure
+                ? FailureOf(ShardFailureCategory.Other, sequence: null) with
+                {
+                    OccurredAt = DateTimeOffset.UtcNow.Subtract(1.Minutes())
+                }
+                : null);
+
+            await controller.ReportFailedLocalAgentsAsync();
+
+            agent.StartCount.ShouldBe(1);
+            agent.StopCount.ShouldBe(0);
+            await _observer.DidNotReceive().AgentPaused(Arg.Any<Uri>(), Arg.Any<ShardFailure?>());
+        }
+
+        [Fact]
+        public async Task picks_a_stopped_shard_back_up_once_its_rebuild_is_over()
+        {
+            var uri = new Uri("event-subscriptions://marten/incident/all");
+            var agent = new FakeSubscriptionAgent(uri);
+            var controller = controllerFor(agent);
+
+            await controller.StartAgentAsync(uri);
+
+            agent.IsRebuilding = true;
+            agent.SimulateFailure(AgentStatus.Stopped, null);
+            await controller.ReportFailedLocalAgentsAsync();
+            agent.StartCount.ShouldBe(1);
+
+            // The rebuild ended without handing the shard back; the ordinary GH-4193 recovery applies again
+            agent.IsRebuilding = false;
+            await controller.ReportFailedLocalAgentsAsync();
+            agent.StartCount.ShouldBe(2);
+        }
+
         [Fact]
         public async Task restarts_it_once_per_transition_not_once_per_tick()
         {
@@ -524,6 +574,9 @@ public void SimulateRecovered()
         // the very next sweep sees Running, clears the ledger, and re-arms.
         public bool StaysStoppedAfterStart { get; set; }
 
+        // GH-5003: the daemon is replaying this shard's projection right now
+        public bool IsRebuilding { get; set; }
+
         public Task StartAsync(CancellationToken cancellationToken)
         {
             StartCount++;

src/Wolverine/Runtime/Agents/EventStoreAgents.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using ImTools;
 using JasperFx.Core;
 using JasperFx.Descriptors;
@@ -13,6 +14,11 @@ public class EventStoreAgents : IAsyncDisposable
     private readonly IEventStore _store;
     private readonly IObserver<ShardState>[] _observers;
     private readonly SemaphoreSlim _daemonLock = new(1);
+
+    // GH-5003: every wrapper this store has built, so a rebuild that runs outside a wrapper can hand
+    // continuous execution back to the ones it stopped. A wrapper that has since been stopped stays in
+    // here until its URI is built again, but reads Stopped and is skipped.
+    private readonly ConcurrentDictionary<Uri, EventSubscriptionAgent> _builtAgents = new();
     private ImHashMap<DatabaseId, IProjectionDaemon> _daemons = ImHashMap<DatabaseId, IProjectionDaemon>.Empty;
 
     // The observer subscriptions taken out against each database's daemon Tracker, so they can be
@@ -291,11 +297,14 @@ public async Task<EventSubscriptionAgent> BuildAgentAsync(Uri uri, DatabaseId da
 
         var daemon = await FindDaemonAsync(databaseId);
 
-        return new EventSubscriptionAgent(uri, shardName, daemon)
+        var agent = new EventSubscriptionAgent(uri, shardName, daemon)
         {
             OnStarted = () => agentStartedAsync(databaseId),
             OnStopped = () => agentStoppedAsync(databaseId)
         };
+
+        _builtAgents[uri] = agent;
+        return agent;
     }
 
     /// <summary>
@@ -381,10 +390,43 @@ public async Task<bool> TryRebuildRegisteredProjectionAsync(string shardIdentity
         // tenantId flows to the daemon's tenant overload to rebuild just that partition, leaving the other
         // tenants untouched. For a store-global projection the tenant is a no-op. RebuildProjectionAsync
         // builds a rebuild-mode agent, replays to the high-water mark, then stops it.
-        await daemon.RebuildProjectionAsync(match.Subscription.Name, tenantId, cancellation);
+        try
+        {
+            await daemon.RebuildProjectionAsync(match.Subscription.Name, tenantId, cancellation);
+        }
+        finally
+        {
+            // GH-5003: the daemon's rebuild hard-stops this node's continuous agents for the projection and
+            // never restarts them, whether it succeeds or throws. Hand them back here rather than leaving
+            // NodeAgentController's sweep to notice -- it never does for an agent whose pre-rebuild failure
+            // was an Other one, and it can only restart the rest by stopping and starting them again.
+            if (!cancellation.IsCancellationRequested)
+            {
+                await resumeAgentsStoppedByRebuildAsync(daemon, match.Subscription.Name, tenantId);
+            }
+        }
+
         return true;
     }
 
+    private async Task resumeAgentsStoppedByRebuildAsync(IProjectionDaemon daemon, string projectionName,
+        string? tenantId)
+    {
+        // Only this daemon's agents were stopped, and a per-tenant rebuild only stops that tenant's
+        // partition -- or the whole projection, when the projection isn't partitioned per tenant and the
+        // tenant argument is a no-op, which is what a shard with no tenant of its own means.
+        var stopped = _builtAgents.Values.Where(agent =>
+            ReferenceEquals(agent.Daemon, daemon)
+            && agent.ShardName.Name == projectionName
+            && (tenantId == null || agent.ShardName.TenantId == null
+                || string.Equals(agent.ShardName.TenantId, tenantId, StringComparison.OrdinalIgnoreCase)));
+
+        foreach (var agent in stopped)
+        {
+            await agent.TryResumeAfterRebuildAsync();
+        }
+    }
+
     private static DatabaseId? ResolveDefaultDatabaseId(EventStoreUsage usage)
     {
         if (usage.Database.MainDatabase != null)