diff --git a/src/Testing/CoreTests/Runtime/Agents/failed_rebuild_resumes_continuous_execution.cs b/src/Testing/CoreTests/Runtime/Agents/failed_rebuild_resumes_continuous_execution.cs new file mode 100644 index 000000000..e54bb1047 --- /dev/null +++ b/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; + +/// +/// 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. +/// +public class failed_rebuild_resumes_continuous_execution +{ + public class through_the_live_agent + { + private readonly IProjectionDaemon _daemon = Substitute.For(); + private readonly ShardName _shardName = new("Incident"); + private readonly EventSubscriptionAgent _agent; + + public through_the_live_agent() + { + _daemon.StartAgentAsync(Arg.Any(), Arg.Any()) + .Returns(Substitute.For()); + _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(), Arg.Any(), Arg.Any()) + .ThrowsAsync(new InvalidOperationException("the connection pool has been exhausted")); + + var ex = await Should.ThrowAsync(() => _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()); + _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(), Arg.Any(), Arg.Any()) + .ThrowsAsync(new InvalidOperationException("rebuild failed")); + _daemon.StartAgentAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(new TimeoutException("resume failed")); + + var ex = await Should.ThrowAsync(() => _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(), Arg.Any(), Arg.Any()) + .Returns(async _ => + { + await cancellation.CancelAsync(); + throw new OperationCanceledException(cancellation.Token); + }); + + await Should.ThrowAsync(() => _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()); + } + + [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(); + private readonly EventStoreAgents theAgents; + + public through_the_registered_projection_rebuild() + { + theDaemon.Tracker.Returns(new ShardStateTracker(NullLogger.Instance)); + theDaemon.StartAgentAsync(Arg.Any(), Arg.Any()) + .Returns(Substitute.For()); + + 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(); + store.Identity.Returns(new EventStoreIdentity("main", "marten")); + store.TryCreateUsage(Arg.Any()).Returns(Task.FromResult(usage)); + store.BuildProjectionDaemonAsync(Arg.Any()) + .Returns(new ValueTask(theDaemon)); + + theAgents = new EventStoreAgents(store, []); + } + + private async Task 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(x => x.Name == projectionName), Arg.Any()); + + [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(), Arg.Any(), Arg.Any()) + .ThrowsAsync(new InvalidOperationException("the connection pool has been exhausted")); + + await Should.ThrowAsync(() => + 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(), Arg.Any()); + } + } +} diff --git a/src/Testing/CoreTests/Runtime/Agents/paused_shard_failure_surfacing.cs b/src/Testing/CoreTests/Runtime/Agents/paused_shard_failure_surfacing.cs index 814b2c66f..86d3175f4 100644 --- a/src/Testing/CoreTests/Runtime/Agents/paused_shard_failure_surfacing.cs +++ b/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(), Arg.Any()); + } + + [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++; diff --git a/src/Wolverine/Runtime/Agents/EventStoreAgents.cs b/src/Wolverine/Runtime/Agents/EventStoreAgents.cs index 00570f265..32510d656 100644 --- a/src/Wolverine/Runtime/Agents/EventStoreAgents.cs +++ b/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[] _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 _builtAgents = new(); private ImHashMap _daemons = ImHashMap.Empty; // The observer subscriptions taken out against each database's daemon Tracker, so they can be @@ -291,11 +297,14 @@ public async Task 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; } /// @@ -381,10 +390,43 @@ public async Task 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) diff --git a/src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs b/src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs index 1a21a95c2..c73c510de 100644 --- a/src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs +++ b/src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs @@ -132,7 +132,21 @@ public async Task RebuildAsync(CancellationToken cancellationToken) // Pass the shard's tenant so a per-tenant agent rebuilds ONLY its tenant's partition under // single-database per-tenant partitioning. _shardName.TenantId is null for store-global / // database-per-tenant shards, where the tenant-less behavior is correct. - await _daemon.RebuildProjectionAsync(_shardName.Name, _shardName.TenantId, cancellationToken); + try + { + await _daemon.RebuildProjectionAsync(_shardName.Name, _shardName.TenantId, cancellationToken); + } + catch (Exception e) when (!cancellationToken.IsCancellationRequested) + { + // GH-5003: a rebuild that throws has already hard-stopped the continuous agent and torn down + // the projection's state, so not resuming here left the shard stopped -- for good when the + // pre-rebuild agent was carrying an Other failure, which the sweep only reports. Resume anyway: + // the progress is already reset, so the continuous agent effectively rebuilds from the start. + // The rebuild's own exception still propagates; a resume failure is only logged under it. + _logger.LogError(e, "Rebuild of {ShardName} failed; resuming its continuous execution", _shardName.Identity); + await TryResumeAfterRebuildAsync(); + throw; + } // GH-3520: the daemon-level rebuild stops the continuous agent and never restarts it. Under // Wolverine-managed distribution there is no store coordinator to resurrect it, and this wrapper @@ -142,6 +156,37 @@ public async Task RebuildAsync(CancellationToken cancellationToken) await resumeContinuousAsync(cancellationToken); } + /// + public bool IsRebuilding => _daemon.IsRebuilding(_shardName); + + internal ShardName ShardName => _shardName; + + internal IProjectionDaemon Daemon => _daemon; + + /// + /// GH-5003: resume continuous execution after a rebuild that ran outside this wrapper -- the + /// registered-projection rebuild in + /// hard-stops this node's agents for the projection and never restarts them. A wrapper that was stopped + /// on purpose (or never started) is left alone. Never throws: the rebuild's outcome is what the caller + /// reports, and a failed resume is still visible to NodeAgentController's sweep. + /// + internal async Task TryResumeAfterRebuildAsync() + { + if (_status == AgentStatus.Stopped) + { + return; + } + + try + { + await resumeContinuousAsync(CancellationToken.None); + } + catch (Exception e) + { + _logger.LogError(e, "Could not resume continuous execution of {ShardName} after a rebuild", _shardName.Identity); + } + } + public async Task RewindAsync(long? sequenceFloor, DateTimeOffset? timestamp, CancellationToken cancellationToken) { // The per-tenant overload delegates to the store-global path when TenantId is null, so this diff --git a/src/Wolverine/Runtime/Agents/IEventSubscriptionAgent.cs b/src/Wolverine/Runtime/Agents/IEventSubscriptionAgent.cs index bf1f2552f..364e787d1 100644 --- a/src/Wolverine/Runtime/Agents/IEventSubscriptionAgent.cs +++ b/src/Wolverine/Runtime/Agents/IEventSubscriptionAgent.cs @@ -44,6 +44,15 @@ Task RewindAsync(long? sequenceFloor, DateTimeOffset? timestamp, CancellationTok /// ShardFailure? Failure => null; + /// + /// GH-5003: whether this agent's projection is being rebuilt right now. A rebuild hard-stops the + /// continuous agent and leaves its old failure in place, so for the whole replay the agent reads as a + /// stopped shard -- which is exactly what + /// restarts. It checks this first and leaves the shard to the rebuild. Default false so existing + /// implementations are unaffected. + /// + bool IsRebuilding => false; + /// /// GH-3888: whether this agent has burned through its budget of node-local auto-restarts without /// its sequence advancing. The auto-restart in is node-local diff --git a/src/Wolverine/Runtime/Agents/NodeAgentController.cs b/src/Wolverine/Runtime/Agents/NodeAgentController.cs index da8edc069..98be44123 100644 --- a/src/Wolverine/Runtime/Agents/NodeAgentController.cs +++ b/src/Wolverine/Runtime/Agents/NodeAgentController.cs @@ -766,6 +766,15 @@ internal async Task ReportFailedLocalAgentsAsync() continue; } + // GH-5003: a shard whose projection is being rebuilt reads Stopped -- the rebuild hard-stopped it + // and left its old failure in place -- for the whole replay. Restarting it here would put a + // continuous agent on the same progression row as the rebuild. The rebuild hands continuous + // execution back when it finishes, so leave it alone until then. + if (subscription.IsRebuilding) + { + continue; + } + // GH-3888: an agent that burned through its local auto-restart budget without ever // advancing is asking to be placed somewhere else. Checked BEFORE the Running // short-circuit below — a stalled shard still reads Running.