diff --git a/src/EventTests/Daemon/ShardStateTrackerTests.cs b/src/EventTests/Daemon/ShardStateTrackerTests.cs
index addf77b..a337e43 100644
--- a/src/EventTests/Daemon/ShardStateTrackerTests.cs
+++ b/src/EventTests/Daemon/ShardStateTrackerTests.cs
@@ -419,4 +419,168 @@ public async Task a_condition_that_blows_up_on_an_unrelated_state_is_still_just_
 
         (await waiter).ShardName.ShouldBe("Trip:All");
     }
+
+    // jasperfx#572 — #568 narrowed the lost-wakeup window but could not close it, because the snapshot the
+    // watcher re-checked was itself written by the very publication walk it was racing: the tracker
+    // subscribed itself first and updated its state map from OnNext on the block's consumer thread. A
+    // watcher could subscribe too late for the walk AND read the snapshot before that walk recorded the
+    // state, so neither path saw it. The tracker now records the state synchronously in PublishAsync, and
+    // subscribe-then-snapshot happens under the same lock as record-then-capture-listeners.
+
+    [Fact]
+    public async Task the_state_map_is_never_behind_a_publication_the_caller_has_already_made()
+    {
+        // This is the invariant the fix rests on: once PublishAsync has returned, no reader can be told
+        // that the tracker has never heard of the shard. Previously the map was only updated when the
+        // consumer thread got around to walking the listeners.
+        await theTracker.PublishAsync(new ShardState("Trip:All", 45) { AgentStatus = "Running" });
+
+        theTracker.CurrentState("Trip:All").ShouldNotBeNull().Sequence.ShouldBe(45);
+        theTracker.CurrentStates().ShouldContain(x => x.ShardName == "Trip:All");
+    }
+
+    [Fact]
+    public async Task the_state_map_does_not_regress_to_an_older_state_still_in_flight()
+    {
+        // The old arrangement had two writers of the map racing: PublishAsync's caller and the consumer
+        // thread replaying an earlier state. A snapshot read could briefly go backwards.
+        await theTracker.PublishAsync(new ShardState("Trip:All", 10));
+        await theTracker.PublishAsync(new ShardState("Trip:All", 45));
+
+        theTracker.CurrentState("Trip:All").ShouldNotBeNull().Sequence.ShouldBe(45);
+    }
+
+    [Fact]
+    public async Task watcher_sees_a_state_that_is_published_but_not_yet_delivered()
+    {
+        // The deterministic form of #572: park the tracker's single consumer thread inside a delivery so
+        // the *next* publication is provably queued and un-walked, then set up a wait for it. Nothing can
+        // ever arrive at OnNext while the gate is shut, so the watcher must find it in the snapshot or
+        // burn its whole timeout on a mark that has already been reached.
+        var gate = new ManualResetEventSlim(false);
+        var blocker = new BlockingObserver(gate);
+        theTracker.Subscribe(blocker);
+
+        var parked = parkTheDeliveryThread(blocker);
+        try
+        {
+            await theTracker.PublishAsync(new ShardState("Trip:All", 45) { AgentStatus = "Running" });
+
+            // Straight at the watcher — WaitForShardState's own pre-check would otherwise mask which
+            // half of the fix is doing the work.
+            var watcher = new ShardStatusWatcher(theTracker, new ShardState("Trip:All", 45), 5.Seconds());
+
+            watcher.Task.IsCompleted.ShouldBeTrue();
+            (await watcher.Task).Sequence.ShouldBe(45);
+        }
+        finally
+        {
+            gate.Set();
+            await parked;
+        }
+    }
+
+    [Fact]
+    public async Task wait_for_shard_condition_sees_a_state_that_is_published_but_not_yet_delivered()
+    {
+        var gate = new ManualResetEventSlim(false);
+        var blocker = new BlockingObserver(gate);
+        theTracker.Subscribe(blocker);
+
+        var parked = parkTheDeliveryThread(blocker);
+        try
+        {
+            await theTracker.PublishAsync(new ShardState("Trip:All", 45) { AgentStatus = "Paused" });
+
+            var waiter = theTracker.WaitForShardCondition(
+                x => x.ShardName == "Trip:All" && x.AgentStatus == "Paused", "Trip:All is paused", 5.Seconds());
+
+            waiter.IsCompleted.ShouldBeTrue();
+            (await waiter).AgentStatus.ShouldBe("Paused");
+        }
+        finally
+        {
+            gate.Set();
+            await parked;
+        }
+    }
+
+    /// <summary>
+    /// Publish the sentinel "Blocker" state from a pool thread and return once its delivery has parked.
+    /// The block's channel allows synchronous continuations, so whichever thread posts an item may be the
+    /// one that runs the delivery — publishing this from the test's own thread would park the test itself.
+    /// Once it is parked, nothing further can be delivered until the gate opens.
+    /// </summary>
+    private Task parkTheDeliveryThread(BlockingObserver blocker)
+    {
+        var parked = Task.Run(async () => await theTracker.PublishAsync(new ShardState("Blocker", 1)));
+        blocker.Entered.Wait(10.Seconds()).ShouldBeTrue("The 'Blocker' state was never delivered");
+        return parked;
+    }
+
+    [Fact]
+    public async Task publish_then_wait_from_another_thread_never_loses_the_wakeup()
+    {
+        // The shape Marten's HighWaterAgentTests hit: a publication in flight on one thread while the wait
+        // is being set up on another, and nothing further is ever published for that shard. Repeat it
+        // enough times to land inside the old window.
+        for (var i = 0; i < 200; i++)
+        {
+            using var tracker = new ShardStateTracker(new NulloLogger());
+            var ready = new ManualResetEventSlim(false);
+
+            var publisher = Task.Run(async () =>
+            {
+                ready.Wait();
+                await tracker.PublishAsync(new ShardState("Trip:All", 45) { AgentStatus = "Running" });
+            });
+
+            ready.Set();
+
+            var state = await tracker.WaitForShardState("Trip:All", 45, 10.Seconds());
+            state.Sequence.ShouldBe(45);
+
+            await publisher;
+        }
+    }
+
+    [Fact]
+    public async Task concurrent_subscribers_are_never_silently_dropped()
+    {
+        // Subscribe was a read-modify-write on a plain field, so two threads adding themselves at once
+        // could leave one of them off the list entirely — a permanently lost wakeup, not a delayed one.
+        var observers = Enumerable.Range(0, 100).Select(_ => new Observer()).ToArray();
+
+        Parallel.ForEach(observers, observer => theTracker.Subscribe(observer));
+
+        await theTracker.PublishAsync(new ShardState("Trip:All", 45));
+        await theTracker.Complete();
+
+        observers.ShouldAllBe(x => x.States.Count == 1);
+    }
+
+    /// <summary>
+    /// Occupies the tracker's consumer thread inside a delivery until released, so a test can prove a
+    /// following publication has not been walked yet.
+    /// </summary>
+    private class BlockingObserver(ManualResetEventSlim gate) : IObserver<ShardState>
+    {
+        public readonly ManualResetEventSlim Entered = new(false);
+
+        public void OnCompleted()
+        {
+        }
+
+        public void OnError(Exception error)
+        {
+        }
+
+        public void OnNext(ShardState value)
+        {
+            if (value.ShardName != "Blocker") return;
+
+            Entered.Set();
+            gate.Wait(30.Seconds());
+        }
+    }
 }
diff --git a/src/JasperFx.Events/Daemon/ShardStateTracker.cs b/src/JasperFx.Events/Daemon/ShardStateTracker.cs
index ba477d9..00a84de 100644
--- a/src/JasperFx.Events/Daemon/ShardStateTracker.cs
+++ b/src/JasperFx.Events/Daemon/ShardStateTracker.cs
@@ -15,9 +15,17 @@ public class ShardStateTracker: IObservable<ShardState>, IObserver<ShardState>,
 {
     private readonly Block<ShardState> _block;
     private readonly ILogger _logger;
-    private readonly IDisposable _subscription;
+
+    // jasperfx#572: _listeners and _states are mutated from arbitrary publisher threads, from watcher
+    // threads subscribing, and from the block's consumer thread. One lock over both of them is what makes
+    // "subscribe, then read the snapshot" atomic with respect to "record the state, then hand it to the
+    // listeners that existed at that moment" -- see the comment on SubscribeAndCaptureCurrentStates.
+    // They were also both read-modify-write on plain fields before, so two concurrent Subscribe calls
+    // could silently lose one of the subscribers outright.
+    private readonly object _lock = new();
     private ImmutableList<IObserver<ShardState>> _listeners = ImmutableList<IObserver<ShardState>>.Empty;
     private ImHashMap<string, ShardState> _states = ImHashMap<string, ShardState>.Empty;
+    private long _highWaterMark;
 
     public ShardStateTracker(ILogger logger)
     {
@@ -25,15 +33,17 @@ public ShardStateTracker(ILogger logger)
         _block = new Block<ShardState>(publish);
         _block.OnError = (state, ex) =>
             _logger.LogError(ex, "Failure while publishing shard state {State}", state);
-
-        _subscription = Subscribe(this);
     }
 
     /// <summary>
     ///     Currently known "high water mark" denoting the highest complete sequence
     ///     of the event storage
     /// </summary>
-    public long HighWaterMark { get; private set; }
+    public long HighWaterMark
+    {
+        get => Volatile.Read(ref _highWaterMark);
+        private set => Volatile.Write(ref _highWaterMark, value);
+    }
 
     /// <summary>
     ///     <see cref="IEventDatabase.Identifier" /> of the database this tracker belongs to. Stamped onto every
@@ -57,7 +67,6 @@ public ShardStateTracker(ILogger logger)
 
     void IDisposable.Dispose()
     {
-        _subscription.Dispose();
         _block.Complete();
     }
 
@@ -69,14 +78,51 @@ void IDisposable.Dispose()
     /// <returns></returns>
     public IDisposable Subscribe(IObserver<ShardState> observer)
     {
-        if (!_listeners.Contains(observer))
+        lock (_lock)
         {
-            _listeners = _listeners.Add(observer);
+            addListener(observer);
         }
 
         return new Unsubscriber(this, observer);
     }
 
+    /// <summary>
+    ///     jasperfx#572: subscribe an observer AND read the current state snapshot as one atomic step with
+    ///     respect to publication. A watcher has two ways to learn that its shard reached a sequence — an
+    ///     <see cref="IObserver{T}.OnNext" /> from the publication walk, or the snapshot it reads right
+    ///     after subscribing — and it must never fall between them. Doing the two separately left exactly
+    ///     that hole: the walk could capture the listener list (missing the watcher) and the watcher could
+    ///     then read the snapshot before that state was recorded, so neither path saw it. If nothing
+    ///     further was ever published for that shard — a high water agent that has reached the head has
+    ///     nothing left to detect — the wait could only end in a timeout, however generous.
+    ///     Taking the same lock <see cref="PublishAsync" /> uses to record the state and <see cref="publish" />
+    ///     uses to capture its listener list totally orders the two sides: either this call wins the lock,
+    ///     and the walk that follows sees the new listener, or the publication wins and the state is in the
+    ///     snapshot returned here.
+    /// </summary>
+    internal IDisposable SubscribeAndCaptureCurrentStates(IObserver<ShardState> observer,
+        out IReadOnlyList<ShardState> currentStates)
+    {
+        lock (_lock)
+        {
+            addListener(observer);
+            currentStates = snapshot();
+        }
+
+        return new Unsubscriber(this, observer);
+    }
+
+    private void addListener(IObserver<ShardState> observer)
+    {
+        if (!_listeners.Contains(observer))
+        {
+            _listeners = _listeners.Add(observer);
+        }
+    }
+
+    private IReadOnlyList<ShardState> snapshot()
+        => _states.Enumerate().Select(x => x.Value).ToList();
+
     void IObserver<ShardState>.OnCompleted()
     {
     }
@@ -85,9 +131,23 @@ void IObserver<ShardState>.OnError(Exception error)
     {
     }
 
+    /// <summary>
+    ///     Only reachable by wiring this tracker up as an observer of some other observable. The tracker no
+    ///     longer subscribes to itself to maintain its own state map — <see cref="PublishAsync" /> records
+    ///     the state synchronously instead, so the map is never behind a publication the caller has already
+    ///     made (jasperfx#572).
+    /// </summary>
     void IObserver<ShardState>.OnNext(ShardState value)
     {
-        _states = _states.AddOrUpdate(value.ShardName, value);
+        recordState(value);
+    }
+
+    private void recordState(ShardState state)
+    {
+        lock (_lock)
+        {
+            _states = _states.AddOrUpdate(state.ShardName, state);
+        }
     }
 
     public ValueTask PublishAsync(ShardState state)
@@ -109,6 +169,13 @@ public ValueTask PublishAsync(ShardState state)
             state.AssignedNodeNumber = AssignedNodeNumber;
         }
 
+        // jasperfx#572: record the state BEFORE handing it to the block. Delivery to the listeners stays
+        // asynchronous, but "what the tracker knows" is now synchronous with the caller, so a snapshot read
+        // can never lag a publication that has already happened. This used to be done by the tracker's own
+        // OnNext on the consumer thread, which put the map update *after* the point at which a watcher
+        // could subscribe too late for the walk.
+        recordState(state);
+
         return _block.PostAsync(state);
     }
 
@@ -155,7 +222,12 @@ public ValueTask MarkSkippingAsync(long lastKnownGoodHighWaterMark, long newHigh
     /// </summary>
     /// <param name="shardName">The raw <see cref="ShardName.Identity" />, or <see cref="ShardState.HighWaterMark" />.</param>
     public ShardState? CurrentState(string shardName)
-        => _states.TryFind(shardName, out var state) ? state : null;
+    {
+        lock (_lock)
+        {
+            return _states.TryFind(shardName, out var state) ? state : null;
+        }
+    }
 
     /// <summary>
     ///     <see cref="CurrentState(string)" /> for a strongly typed shard name.
@@ -177,7 +249,12 @@ public bool TryGetCurrentState(string shardName, [NotNullWhen(true)] out ShardSt
     ///     underlying map is immutable, so a concurrent publication can't disturb the returned list.
     /// </summary>
     public IReadOnlyList<ShardState> CurrentStates()
-        => _states.Enumerate().Select(x => x.Value).ToList();
+    {
+        lock (_lock)
+        {
+            return snapshot();
+        }
+    }
 
     /// <summary>
     ///     Use to "wait" for an expected projection shard state. Safe to call after the state has already
@@ -189,12 +266,10 @@ public IReadOnlyList<ShardState> CurrentStates()
     /// <returns></returns>
     public Task<ShardState> WaitForShardState(ShardState expected, TimeSpan? timeout = null)
     {
-        if (_states.TryFind(expected.ShardName, out var state))
+        var state = CurrentState(expected.ShardName);
+        if (state != null && (state.Equals(expected) || state.Sequence >= expected.Sequence))
         {
-            if (state.Equals(expected) || state.Sequence >= expected.Sequence)
-            {
-                return Task.FromResult(state);
-            }
+            return Task.FromResult(state);
         }
 
         timeout ??= 1.Minutes();
@@ -223,12 +298,10 @@ public Task<ShardState> WaitForShardState(string shardName, long sequence, TimeS
     /// <returns></returns>
     public Task<ShardState> WaitForShardState(ShardName name, long sequence, TimeSpan? timeout = null)
     {
-        if (_states.TryFind(name.Identity, out var state))
+        var state = CurrentState(name.Identity);
+        if (state != null && state.Sequence >= sequence)
         {
-            if (state.Sequence >= sequence)
-            {
-                return Task.FromResult(state);
-            }
+            return Task.FromResult(state);
         }
 
         return WaitForShardState(new ShardState(name.Identity, sequence), timeout);
@@ -277,7 +350,17 @@ private void publish(ShardState state)
             _logger.LogDebug("Received {ShardState}", state);
         }
 
-        foreach (var observer in _listeners)
+        // The listener list is captured under the same lock that records state and that a subscribing
+        // watcher takes, so this walk and "subscribe then read the snapshot" are totally ordered against
+        // each other (jasperfx#572). Notification itself stays outside the lock -- an observer is arbitrary
+        // user code and must never be able to deadlock the tracker by publishing or subscribing from OnNext.
+        ImmutableList<IObserver<ShardState>> listeners;
+        lock (_lock)
+        {
+            listeners = _listeners;
+        }
+
+        foreach (var observer in listeners)
         {
             try
             {
@@ -294,7 +377,13 @@ private void publish(ShardState state)
 
     public void Finish()
     {
-        foreach (var observer in _listeners)
+        ImmutableList<IObserver<ShardState>> listeners;
+        lock (_lock)
+        {
+            listeners = _listeners;
+        }
+
+        foreach (var observer in listeners)
         {
             try
             {
@@ -309,7 +398,10 @@ public void Finish()
 
     public void MarkAsRestarted(ShardName name)
     {
-        _states = _states.Remove(name.Identity);
+        lock (_lock)
+        {
+            _states = _states.Remove(name.Identity);
+        }
     }
 
     private class Unsubscriber: IDisposable
@@ -327,7 +419,10 @@ public void Dispose()
         {
             if (_observer != null)
             {
-                _tracker._listeners = _tracker._listeners.Remove(_observer);
+                lock (_tracker._lock)
+                {
+                    _tracker._listeners = _tracker._listeners.Remove(_observer);
+                }
             }
         }
     }
diff --git a/src/JasperFx.Events/Daemon/ShardStatusWatcher.cs b/src/JasperFx.Events/Daemon/ShardStatusWatcher.cs
index e5a9c51..4a6c51d 100644
--- a/src/JasperFx.Events/Daemon/ShardStatusWatcher.cs
+++ b/src/JasperFx.Events/Daemon/ShardStatusWatcher.cs
@@ -43,8 +43,6 @@ private ShardStatusWatcher(ShardStateTracker tracker, Func<ShardState, bool> con
             }
         });
 
-        _unsubscribe = tracker.Subscribe(this);
-
         // jasperfx#568: publication is asynchronous — PublishAsync posts to a Block and returns, and the
         // consumer thread walks the listener list that existed when it started. A state can therefore be
         // delivered (and recorded in the tracker's state map) in the window between a caller checking the
@@ -52,7 +50,15 @@ private ShardStatusWatcher(ShardStateTracker tracker, Func<ShardState, bool> con
         // OnNext only ever sees states delivered AFTER this point, re-check what the tracker already knows
         // now that we're subscribed. TrySetResult makes a double hit harmless — whichever path wins first,
         // the other is a no-op. This is also the only current-state check WaitForShardCondition gets.
-        foreach (var state in tracker.CurrentStates())
+        //
+        // jasperfx#572: subscribing and then reading the snapshot as two separate steps still left a hole
+        // for a watcher to fall through — the publication walk could capture the listener list without this
+        // watcher, and the snapshot could still be read before that state was recorded, so NEITHER path saw
+        // it. Taking both in one atomic step against the tracker's publication lock closes it rather than
+        // narrowing it.
+        _unsubscribe = tracker.SubscribeAndCaptureCurrentStates(this, out var currentStates);
+
+        foreach (var state in currentStates)
         {
             try
             {