GH-3520: EventSubscriptionAgent restores continuous execution after Rebuild/Rewind

Directory.Packages.props

@@ -35,13 +35,13 @@
     <PackageVersion Include="Grpc.StatusProto" Version="2.76.0" />
     <PackageVersion Include="Grpc.Tools" Version="2.76.0" />
     <PackageVersion Include="HtmlTags" Version="9.0.0" />
-    <PackageVersion Include="JasperFx" Version="2.30.1" />
-    <PackageVersion Include="JasperFx.Events" Version="2.30.1" />
-    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.30.1" />
+    <PackageVersion Include="JasperFx" Version="2.30.2" />
+    <PackageVersion Include="JasperFx.Events" Version="2.30.2" />
+    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.30.2" />
     <!-- RuntimeCompiler is on its own 5.x line (the Roslyn compiler package) — not the 2.1.x
          family; it stays at 5.0.0. -->
     <PackageVersion Include="JasperFx.RuntimeCompiler" Version="5.0.0" />
-    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.30.1" />
+    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.30.2" />
     <PackageVersion Include="Lamar.Microsoft.DependencyInjection" Version="16.0.0" />
     <PackageVersion Include="Marten" Version="9.16.1" />
     <PackageVersion Include="Microsoft.Data.SqlClient" Version="6.1.3" />

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

@@ -0,0 +1,55 @@
+using JasperFx;
+using JasperFx.Events.Projections;
+using NSubstitute;
+using Shouldly;
+using Wolverine.Runtime.Agents;
+using Xunit;
+using IProjectionDaemon = JasperFx.Events.Daemon.IProjectionDaemon;
+using JasperFxSubscriptionAgent = JasperFx.Events.Daemon.ISubscriptionAgent;
+
+namespace CoreTests.Runtime.Agents;
+
+/// <summary>
+/// Regression coverage for GH-3520. Under Wolverine-managed event-subscription distribution there is no
+/// store coordinator to resurrect a shard the daemon stopped. RebuildProjectionAsync stops the continuous
+/// agent and never restarts it; RewindSubscriptionAsync restarts it daemon-side but the wrapper still
+/// pointed at the stopped pre-rewind agent and reported Running. Either way the shard froze at
+/// RegisteredIdle while its high-water climbed. EventSubscriptionAgent.RebuildAsync/RewindAsync must now
+/// restore continuous execution themselves through the registered daemon start path and refresh their view.
+/// </summary>
+public class event_subscription_agent_rebuild_rewind_resume
+{
+    private readonly IProjectionDaemon _daemon = Substitute.For<IProjectionDaemon>();
+
+    // Store-global shard ("Incident:All", TenantId null) - the shape in the GH-3519/3520 report.
+    private readonly ShardName _shardName = new("Incident");
+    private readonly JasperFxSubscriptionAgent _restarted = Substitute.For<JasperFxSubscriptionAgent>();
+    private readonly EventSubscriptionAgent _agent;
+
+    public event_subscription_agent_rebuild_rewind_resume()
+    {
+        _daemon.StartAgentAsync(Arg.Any<ShardName>(), Arg.Any<CancellationToken>()).Returns(_restarted);
+        _agent = new EventSubscriptionAgent(
+            new Uri("event-subscriptions://marten/incident/all"), _shardName, _daemon);
+    }
+
+    [Fact]
+    public async Task rebuild_restores_continuous_execution_through_the_registered_start_path()
+    {
+        await _agent.RebuildAsync(CancellationToken.None);
+
+        // Before the fix RebuildAsync returned without restarting anything: no start call, and the
+        // wrapper kept its stale status. Now it resumes through the registered daemon start path.
+        await _daemon.Received(1).StartAgentAsync(Arg.Any<ShardName>(), Arg.Any<CancellationToken>());
+        _agent.Status.ShouldBe(AgentStatus.Running);
+    }
+
+    [Fact]
+    public async Task rewind_restores_continuous_execution_through_the_registered_start_path()
+    {
+        await _agent.RewindAsync(0, null, CancellationToken.None);
+
+        await _daemon.Received(1).StartAgentAsync(Arg.Any<ShardName>(), Arg.Any<CancellationToken>());
+        _agent.Status.ShouldBe(AgentStatus.Running);
+    }
+}

src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs

@@ -59,8 +59,7 @@ public EventSubscriptionAgent(Uri uri, ShardName shardName, IProjectionDaemon da
 
     public async Task StartAsync(CancellationToken cancellationToken)
     {
-        _innerAgent = await _daemon.StartAgentAsync(_shardName, cancellationToken);
-        Status = AgentStatus.Running;
+        await resumeContinuousAsync(cancellationToken);
 
         // Only count an agent that actually started - a throw above leaves the count untouched
         if (OnStarted != null)
@@ -69,6 +68,17 @@ public async Task StartAsync(CancellationToken cancellationToken)
         }
     }
 
+    // Start (or re-adopt) the continuous inner agent through the registered daemon path and refresh the
+    // wrapper's view of it. Deliberately does NOT invoke OnStarted: the running-agent bookkeeping in
+    // EventStoreAgents counts one logical agent per node for the observer-subscription lifecycle, and a
+    // rebuild/rewind is transparent to that count (the wrapper never observed a matching stop), so
+    // re-counting here would leak the database's tracker subscriptions. See GH-3520.
+    private async Task resumeContinuousAsync(CancellationToken cancellationToken)
+    {
+        _innerAgent = await _daemon.StartAgentAsync(_shardName, cancellationToken);
+        Status = AgentStatus.Running;
+    }
+
     public async Task StopAsync(CancellationToken cancellationToken)
     {
         await _daemon.StopAgentAsync(_shardName);
@@ -86,6 +96,13 @@ public async Task RebuildAsync(CancellationToken cancellationToken)
         // 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);
+
+        // 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
+        // would otherwise keep reporting Running against the now-stopped agent, so NodeAgentController
+        // sees nothing to fix and the shard freezes at RegisteredIdle while its high-water climbs.
+        // Restore continuous execution ourselves through the registered daemon start path.
+        await resumeContinuousAsync(cancellationToken);
     }
 
     public async Task RewindAsync(long? sequenceFloor, DateTimeOffset? timestamp, CancellationToken cancellationToken)
@@ -95,6 +112,14 @@ public async Task RewindAsync(long? sequenceFloor, DateTimeOffset? timestamp, Ca
         // CritterWatch needs because DaemonForDatabase() throws under Wolverine-managed distribution.
         await _daemon.RewindSubscriptionAsync(_shardName.Name, _shardName.TenantId, cancellationToken,
             sequenceFloor, timestamp);
+
+        // GH-3520: same freeze as RebuildAsync. RewindSubscriptionAsync restarts continuous agents
+        // daemon-side, but the wrapper's _innerAgent still points at the stopped pre-rewind agent and
+        // Status still reads Running. Re-adopt the live agent through the registered start path so the
+        // wrapper reflects reality. This requires JasperFx#536 (rewind registers its restarted agent) so
+        // the registered start resolves to that same running agent idempotently rather than spinning up a
+        // duplicate on the same progression row - this ships in lockstep with that JasperFx bump.
+        await resumeContinuousAsync(cancellationToken);
     }
 
     public Uri Uri { get; }