Dedupe follow-up: ProjectionCoordinatorBase (#4516) + HiloSequenceBase (#4527) (#4532)

* #4516: reduce ProjectionCoordinator to a ProjectionCoordinatorBase subclass

#326 lifted the projection-coordinator execute loop + agent-start
resilience + lifecycle (Start/Pause/Resume/Stop) into
JasperFx.Events.Daemon.ProjectionCoordinatorBase. Reduce Marten's coordinator to
a subclass:

- Supply via the base ctor: BuildDistributor(store), the store's
  ResiliencePipeline, Events.TimeProvider, and the three settings
  (LeadershipPollingTime as a TimeSpan, AgentPauseTime, HealthCheckPollingTime).
- Implement the seams: ResolveDaemon (keeps Marten's ImHashMap +
  double-checked-lock daemon cache) and ResolvedDaemons; keep the existing
  DaemonForMainDatabase / DaemonForDatabase / AllDaemonsAsync.
- Delete the lifted-into-base members: executeAsync, start/stopAgentsIfNecessary,
  tryStartAgent, the Start/Pause/Resume/Stop bodies, pauseDistributor, and the
  DaemonShardName record. Behavior is unchanged (the base adopted Marten's
  resilient agent-start + single-cancellation lifecycle verbatim).

DaemonTests 185/0; HotCold leadership-election tests (detect_high_water_mark,
end_to_end_with_events_already_published) pass.

Closes #4516.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* #4527: derive HiloSequence from Weasel.Core.Sequences.HiloSequenceBase

Completes #4527 (the serialization-enum subset landed in #4529). weasel#287
lifted the dialect-agnostic Hi-Lo pieces into Weasel.Core.Sequences:
HiloSequenceBase (state + arithmetic), ISequence, IReadOnlyHiloSettings,
HiloSettings, and HiloSequenceAdvanceToNextHiAttemptsExceededException.

- Marten's HiloSequence now derives from HiloSequenceBase, keeping only the
  PostgreSQL I/O (mt_get_next_hi stored function in AdvanceToNextHi/Sync, the
  mt_hilo floor update in SetFloor). The client-side arithmetic
  (AdvanceValue/ShouldAdvanceHi/NextInt/NextLong/TrySetCurrentHi/hi-lo state) is
  inherited verbatim — id allocation is unchanged.
- Delete Marten's ISequence / IReadOnlyHiloSettings / HiloSettings /
  HiloSequenceAdvanceToNextHiAttemptsExceededException copies and alias the old
  names via the shared dedupe file (shapes identical; the exception's base
  changed MartenException -> Exception but nothing catches it as MartenException).
- AdvanceToNextHiSync is protected on the base; the one test that drove it
  directly now invokes it reflectively to keep exercising the sync I/O path.

Hi-Lo + sequence-id tests pass (32/0).

Closes #4527.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jeremydmiller

src/CoreTests/Storage/Identification/closed_shape_id_strategies_tests.cs

@@ -80,7 +80,7 @@ public async Task identity_key_assigns_alias_slash_sequence()
         {
             opts.Schema.For<IdentityKeyDoc>().IdStrategy(
                 new Marten.Schema.Identity.Sequences.IdentityKeyGeneration(
-                    (DocumentMapping)null!, new Marten.Schema.Identity.Sequences.HiloSettings()));
+                    (DocumentMapping)null!, new HiloSettings()));
         });
 
         var doc = new IdentityKeyDoc { Name = "v1" };

src/DocumentDbTests/Writing/Identity/Sequences/hilo_tests.cs

@@ -2,6 +2,7 @@
 using System.Collections;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reflection;
 using System.Threading.Tasks;
 using JasperFx.Core;
 using JasperFx.Core.Reflection;
@@ -22,7 +23,11 @@ public class AdvanceToNextHi : IEnumerable<object[]>
 {
     private static readonly Func<HiloSequence, Task> AsyncNext = sequence => sequence.AdvanceToNextHi();
     private static readonly Func<HiloSequence, Task> SyncNext = sequence =>  {
-        sequence.AdvanceToNextHiSync();
+        // AdvanceToNextHiSync is protected on the lifted Weasel.Core HiloSequenceBase (#4527),
+        // so invoke it reflectively to keep exercising the synchronous hi-advance I/O path.
+        typeof(HiloSequence)
+            .GetMethod("AdvanceToNextHiSync", BindingFlags.Instance | BindingFlags.NonPublic)!
+            .Invoke(sequence, null);
         return Task.CompletedTask;
     };
     public IEnumerator<object[]> GetEnumerator()

src/Marten/Events/Daemon/Coordination/ProjectionCoordinator.cs

@@ -1,10 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-using System.Threading;
 using System.Threading.Tasks;
 using ImTools;
-using JasperFx;
 using JasperFx.Core;
 using JasperFx.Events.Daemon;
 using JasperFx.Events.Projections;
@@ -23,35 +21,42 @@ public ProjectionCoordinator(T documentStore, ILogger<ProjectionCoordinator> log
     }
 }
 
-public class ProjectionCoordinator: IProjectionCoordinator
+// 9.0 (#4516 dedupe): the leadership-election + agent-lifecycle loop now lives in
+// JasperFx.Events.Daemon.ProjectionCoordinatorBase. Marten supplies the distributor +
+// settings via the base ctor and implements the daemon-resolution seams, keeping its
+// own ImHashMap + double-checked-lock daemon cache.
+public class ProjectionCoordinator: ProjectionCoordinatorBase, IProjectionCoordinator
 {
     private readonly System.Threading.Lock _daemonLock = new();
-    private readonly ILogger<ProjectionCoordinator> _logger;
-    private readonly StoreOptions _options;
-
-    private readonly ResiliencePipeline _resilience;
-    private readonly TimeProvider _timeProvider;
-    private CancellationTokenSource? _cancellation;
+    private readonly ILogger _logger;
 
     private ImHashMap<string, IProjectionDaemon> _daemons = ImHashMap<string, IProjectionDaemon>.Empty;
-    private Task? _runner;
 
     public ProjectionCoordinator(IDocumentStore documentStore, ILogger<ProjectionCoordinator> logger)
+        : this((DocumentStore)documentStore, logger)
     {
-        var store = (DocumentStore)documentStore;
+    }
 
+    private ProjectionCoordinator(DocumentStore store, ILogger<ProjectionCoordinator> logger)
+        : base(
+            BuildDistributor(store),
+            logger,
+            store.Options.ResiliencePipeline,
+            store.Options.Events.TimeProvider,
+            store.Options.Projections.LeadershipPollingTime.Milliseconds(),
+            store.Options.Projections.AgentPauseTime,
+            store.Options.Projections.HealthCheckPollingTime)
+    {
         Mode = store.Options.Projections.AsyncMode;
-
-        Distributor = BuildDistributor(store);
-
-        _options = store.Options;
-        _logger = logger;
-        _resilience = store.Options.ResiliencePipeline;
-        _timeProvider = _options.Events.TimeProvider;
         Store = store;
+        _logger = logger;
     }
 
-    // 9.0 (#4349 dedupe): the Solo / SingleTenant / MultiTenanted distributors now live in
+    public DaemonMode Mode { get; }
+
+    public DocumentStore Store { get; }
+
+    // 9.0 (#4349 dedupe): the Solo / SingleTenant / MultiTenanted distributors live in
     // JasperFx.Events. Marten wires them with closures over its own tenancy, shard, and lock
     // surfaces. ProjectionSet (Marten-side) remains the IProjectionSet implementation, and the
     // Postgres lock factory hands back Weasel's AdvisoryLock — which implements
@@ -108,110 +113,36 @@ private static Func<IProjectionDatabase, IAdvisoryLock> buildLockFactory(Documen
             });
     }
 
-    public DaemonMode Mode { get; }
-
-    public DocumentStore Store { get; }
+    protected override IProjectionDaemon ResolveDaemon(IProjectionSet set)
+    {
+        return findDaemonForDatabase((MartenDatabase)set.Database);
+    }
 
-    public IProjectionDistributor Distributor { get; }
+    protected override IReadOnlyList<IProjectionDaemon> ResolvedDaemons()
+    {
+        return _daemons.Enumerate().Select(x => x.Value).ToList();
+    }
 
-    public IProjectionDaemon DaemonForMainDatabase()
+    public override IProjectionDaemon DaemonForMainDatabase()
     {
         var database = (MartenDatabase)Store.Tenancy.Default.Database;
 
         return findDaemonForDatabase(database);
     }
 
-    public async ValueTask<IProjectionDaemon> DaemonForDatabase(string databaseIdentifier)
+    public override async ValueTask<IProjectionDaemon> DaemonForDatabase(string databaseIdentifier)
     {
         var database =
             (MartenDatabase)await Store.Storage.FindOrCreateDatabase(databaseIdentifier).ConfigureAwait(false);
         return findDaemonForDatabase(database);
     }
 
-    public async ValueTask<IReadOnlyList<IProjectionDaemon>> AllDaemonsAsync()
+    public override async ValueTask<IReadOnlyList<IProjectionDaemon>> AllDaemonsAsync()
     {
         var all = await Store.Storage.AllDatabases().ConfigureAwait(false);
         return all.OfType<MartenDatabase>().Select(findDaemonForDatabase).ToList();
     }
 
-    public Task StartAsync(CancellationToken cancellationToken)
-    {
-        _cancellation?.SafeDispose();
-
-        _cancellation = new CancellationTokenSource();
-        _runner = Task.Run(() => executeAsync(_cancellation.Token), _cancellation.Token);
-
-        return Task.CompletedTask;
-    }
-
-    public async Task PauseAsync()
-    {
-        _logger.LogInformation("Pausing ProjectionCoordinator");
-        if (_cancellation != null)
-        {
-            await _cancellation.CancelAsync().ConfigureAwait(false);
-        }
-
-        await pauseDistributor().ConfigureAwait(false);
-
-        foreach (var pair in _daemons.Enumerate())
-        {
-            try
-            {
-                await pair.Value.StopAllAsync().ConfigureAwait(false);
-            }
-            catch (Exception exception)
-            {
-                _logger.LogError(exception, "Error while trying to stop daemon agents in database {Name}", pair.Key);
-            }
-        }
-    }
-
-    private async Task pauseDistributor()
-    {
-        if (_runner == null) return;
-
-        try
-        {
-#pragma warning disable VSTHRD003
-            await _runner.ConfigureAwait(false);
-#pragma warning restore VSTHRD003
-        }
-        catch (TaskCanceledException)
-        {
-            // Nothing, just from shutting down
-        }
-        catch (OperationCanceledException)
-        {
-            // Nothing, just from shutting down
-        }
-        catch (Exception e)
-        {
-            _logger.LogError(e, "Error while trying to stop the ProjectionCoordinator");
-        }
-    }
-
-    public Task ResumeAsync()
-    {
-        return StartAsync(default);
-    }
-
-    public async Task StopAsync(CancellationToken cancellationToken)
-    {
-        await PauseAsync().ConfigureAwait(false);
-
-        foreach (var daemon in _daemons.Enumerate()) daemon.Value.SafeDispose();
-
-        try
-        {
-            await Distributor.ReleaseAllLocks().ConfigureAwait(false);
-        }
-        catch (Exception e)
-        {
-            _logger.LogError(e, "Error trying to release subscription agent locks");
-        }
-    }
-
     private IProjectionDaemon findDaemonForDatabase(MartenDatabase database)
     {
         if (_daemons.TryFind(database.Identifier, out var daemon))
@@ -232,153 +163,4 @@ private IProjectionDaemon findDaemonForDatabase(MartenDatabase database)
 
         return daemon;
     }
-
-    private async Task executeAsync(CancellationToken stoppingToken)
-    {
-        await Distributor.RandomWait(stoppingToken).ConfigureAwait(false);
-
-        while (!stoppingToken.IsCancellationRequested)
-        {
-            try
-            {
-                var sets = await Distributor
-                    .BuildDistributionAsync().ConfigureAwait(false);
-
-                foreach (var set in sets)
-                {
-                    // Is it already running here?
-                    if (Distributor.HasLock(set))
-                    {
-                        var daemon = resolveDaemon(set);
-
-                        // check if it's still running
-                        await startAgentsIfNecessaryAsync(set, daemon, stoppingToken).ConfigureAwait(false);
-                        continue;
-                    }
-
-                    try
-                    {
-                        if (await Distributor.TryAttainLockAsync(set, stoppingToken).ConfigureAwait(false))
-                        {
-                            var daemon = resolveDaemon(set);
-
-                            // check if it's still running
-                            await startAgentsIfNecessaryAsync(set, daemon, stoppingToken).ConfigureAwait(false);
-                        }
-                        else
-                        {
-                            // We don't hold the lock, so we might've lost it due to a postgres outage. We should make sure any agents are no longer running on this node.
-                            var daemon = resolveDaemon(set);
-
-                            await stopAgentsIfNecessaryAsync(set, daemon).ConfigureAwait(false);
-                        }
-                    }
-                    catch (Exception e)
-                    {
-                        _logger.LogError(e, "Error trying to attain a lock for set {Name} and lock id {LockId}. Will retry later", set.Names.Select(x => x.Identity).Join(", "), set.LockId);
-                        await Task.Delay(_options.Projections.LeadershipPollingTime.Milliseconds(), stoppingToken)
-                            .ConfigureAwait(false);
-                    }
-                }
-            }
-            catch (Exception e)
-            {
-                if (stoppingToken.IsCancellationRequested)
-                {
-                    return;
-                }
-
-                // Only really expect any errors if there are dynamic tenants in place
-                _logger.LogError(e, "Error trying to resolve projection distributions");
-            }
-
-            if (stoppingToken.IsCancellationRequested)
-            {
-                return;
-            }
-
-            try
-            {
-                if (_daemons.Enumerate().Any(x => x.Value.HasAnyPaused()))
-                {
-                    await Task.Delay(_options.Projections.AgentPauseTime, stoppingToken).ConfigureAwait(false);
-                }
-                else
-                {
-                    await Task.Delay(_options.Projections.LeadershipPollingTime.Milliseconds(), stoppingToken)
-                        .ConfigureAwait(false);
-                }
-            }
-            catch (TaskCanceledException)
-            {
-                // just get out of here, this signals a graceful shutdown attempt
-            }
-            catch (OperationCanceledException)
-            {
-                // Nothing, just from shutting down
-            }
-        }
-    }
-
-
-    private async Task startAgentsIfNecessaryAsync(IProjectionSet set,
-        IProjectionDaemon daemon, CancellationToken stoppingToken)
-    {
-        foreach (var name in set.Names)
-        {
-            var agent = daemon.CurrentAgents().FirstOrDefault(x => x.Name.Equals(name));
-            if (agent == null)
-            {
-                await tryStartAgent(stoppingToken, daemon, name, set).ConfigureAwait(false);
-            }
-            else if (agent is { Status: AgentStatus.Paused, PausedTime: not null } &&
-                     _timeProvider.GetUtcNow().Subtract(agent.PausedTime.Value) >
-                     _options.Projections.HealthCheckPollingTime)
-            {
-                await tryStartAgent(stoppingToken, daemon, name, set).ConfigureAwait(false);
-            }
-        }
-    }
-
-    private async Task stopAgentsIfNecessaryAsync(IProjectionSet set, IProjectionDaemon daemon)
-    {
-        foreach (var shardName in set.Names)
-        {
-            var status = daemon.StatusFor(shardName.Identity);
-            if (status == AgentStatus.Running)
-            {
-                await daemon.StopAgentAsync(shardName.Identity).ConfigureAwait(false);
-            }
-
-        }
-    }
-
-    private IProjectionDaemon resolveDaemon(IProjectionSet set)
-    {
-        return findDaemonForDatabase((MartenDatabase)set.Database);
-    }
-
-    private async Task tryStartAgent(CancellationToken stoppingToken, IProjectionDaemon daemon, ShardName name,
-        IProjectionSet set)
-    {
-        try
-        {
-            await _resilience.ExecuteAsync(
-                static (x, t) => new ValueTask(x.Daemon.StartAgentAsync(x.Name.Identity, t)),
-                new DaemonShardName(daemon, name), stoppingToken).ConfigureAwait(false);
-        }
-        catch (Exception e)
-        {
-            _logger.LogError(e, "Error trying to start subscription {Name} on database {Database}", name.Identity,
-                set.Database.Identifier);
-            if (daemon.StatusFor(name.Identity) == AgentStatus.Paused)
-            {
-                daemon.EjectPausedShard(name.Identity);
-            }
-
-            await Distributor.ReleaseLockAsync(set).ConfigureAwait(false);
-        }
-    }
-
-    internal record DaemonShardName(IProjectionDaemon Daemon, ShardName Name);
 }