Extended progression batches amortize the connection, not the transaction (2.39.2)

Directory.Build.props

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
 <Project>
     <PropertyGroup>
-        <JasperFxVersion>2.39.1</JasperFxVersion>
+        <JasperFxVersion>2.39.2</JasperFxVersion>
         <LangVersion>13</LangVersion>
         <NoWarn>1570;1571;1572;1573;1574;1587;1591;1701;1702;1711;1735;0618</NoWarn>
         <Authors>Jeremy D. Miller;Jaedyn Tonee</Authors>

src/EventTests/Daemon/ExtendedProgressionWriterTests.cs

@@ -134,6 +134,34 @@ public async Task coalesces_heartbeats_into_one_batched_write_per_flush_interval
         writes.Count.ShouldBe(3);
     }
 
+    // marten#5167: the batch is a lock-acquisition order. A store writes it one row per transaction, so
+    // a single writer never holds more than one row lock -- but two writers racing over the same rows
+    // must not be free to take their locks in opposite orders, and the pending dictionary's enumeration
+    // order is an implementation detail, not an agreement between them.
+    [Fact]
+    public async Task a_flushed_batch_is_ordered_by_shard_name()
+    {
+        enablePeriodicHeartbeats();
+
+        // Seed a flush so the interval throttle is active and the rest coalesce into one batch
+        theWriter.OnNext(heartbeat("Zebras:All", sequence: 1));
+        await theDatabase.WaitForWrites(1);
+
+        // Published in deliberately reverse order
+        theWriter.OnNext(heartbeat("Zebras:All", sequence: 2));
+        theWriter.OnNext(heartbeat("Middles:All", sequence: 3));
+        theWriter.OnNext(heartbeat("Alphas:All", sequence: 4));
+
+        theTime.Advance(theWriter.HeartbeatWriteInterval + TimeSpan.FromMilliseconds(1));
+        theWriter.OnNext(heartbeat("Alphas:All", sequence: 5));
+
+        await theDatabase.WaitForWrites(4);
+        var batches = await theDatabase.Batches();
+
+        batches.Count.ShouldBe(2);
+        batches[1].Select(x => x.ShardName).ShouldBe(["Alphas:All", "Middles:All", "Zebras:All"]);
+    }
+
     [Fact]
     public async Task a_transition_flushes_immediately_and_carries_the_pending_heartbeats_along()
     {

src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs

@@ -29,11 +29,15 @@ namespace JasperFx.Events.Daemon;
 /// <c>DaemonSettings.ExtendedProgressionHeartbeatInterval</c>) to a positive value to restore the
 /// old behavior — that is the compatibility hatch, not the recommended shape.</item>
 /// <item>When periodic beats ARE enabled, they are coalesced per shard (latest state wins) and
-/// flushed as ONE batched database write per <see cref="HeartbeatWriteInterval"/> for the whole
-/// database. The write rate is therefore constant per database instead of O(shards): under
-/// per-tenant agent fan-out (agents = projections × tenants) the previous
+/// flushed as ONE batch per <see cref="HeartbeatWriteInterval"/> for the whole database, ordered by
+/// shard name. The connection rent rate is therefore constant per database instead of O(shards):
+/// under per-tenant agent fan-out (agents = projections × tenants) the previous
 /// one-connection-rent-per-shard-per-interval write path drove a sharded multi-tenant deployment
-/// to its database server's connection ceiling (jasperfx#553).</item>
+/// to its database server's connection ceiling (jasperfx#553). What a batch is NOT is one
+/// transaction — see the one-row-per-transaction requirement on
+/// <see cref="IEventDatabase.WriteExtendedProgressionAsync(System.Collections.Generic.IReadOnlyList{ShardState},System.Threading.CancellationToken)"/>,
+/// which is what keeps a slow projection batch on one row from stalling every other shard's
+/// telemetry behind it (marten#5167).</item>
 /// <item>Agent status transitions (<see cref="ShardAction.Started"/>, <see cref="ShardAction.Paused"/>,
 /// <see cref="ShardAction.Stopped"/>) flush immediately — a paused/stopped shard is exactly when the
 /// persisted status matters most, and these writes are rare, so they keep the "durable across a
@@ -135,7 +139,13 @@ private void flush(DateTimeOffset now)
     {
         if (_pending.Count == 0) return;
 
-        var batch = _pending.Values.ToArray();
+        // Ordered by shard name because the batch is a lock-acquisition order (marten#5167). A store
+        // writes these one row per transaction, so the batch never holds more than one row lock at a
+        // time -- but two writers racing over the same rows (the tracker is per-database and shared, and
+        // building a daemon does not go through a cache) would still be free to take their locks in
+        // opposite orders if the batch order were the dictionary's. Sorting removes that hazard for
+        // every store at zero cost; nothing downstream may reorder it.
+        var batch = _pending.Values.OrderBy(x => x.ShardName, StringComparer.Ordinal).ToArray();
         _pending.Clear();
         _lastFlush = now;
 

src/JasperFx.Events/IEventDatabase.cs

@@ -228,22 +228,45 @@ Task WriteExtendedProgressionAsync(ShardState state, CancellationToken token = d
         => Task.CompletedTask;
 
     /// <summary>
-    ///     Persist the extended progression telemetry for a batch of shards in as few database
-    ///     round-trips as the store can manage — ideally one. The <see cref="Daemon.ExtendedProgressionWriter" />
-    ///     coalesces the heartbeat publications of every shard on a database into one batch per flush
-    ///     interval and drives this overload, because the per-shard single-row write does not scale
-    ///     under per-tenant agent fan-out: agents = projections × tenants, and one connection
-    ///     rent + round-trip per agent per heartbeat interval drove a sharded multi-tenant deployment
-    ///     to its database server's connection ceiling (jasperfx#553).
+    ///     Persist the extended progression telemetry for a batch of shards on ONE rented connection.
+    ///     The <see cref="Daemon.ExtendedProgressionWriter" /> coalesces the heartbeat publications of
+    ///     every shard on a database into one batch per flush interval and drives this overload, because
+    ///     renting a connection per shard does not scale under per-tenant agent fan-out:
+    ///     agents = projections × tenants, and one connection rent per agent per heartbeat interval drove
+    ///     a sharded multi-tenant deployment to its database server's connection ceiling (jasperfx#553).
     ///     <para>
     ///     Same contract as the single-state overload: best-effort telemetry, never advance or
     ///     regress progression, missing rows no-op. The batch is at-most-one-state-per-shard
-    ///     (the writer keeps only the latest state per shard between flushes).
+    ///     (the writer keeps only the latest state per shard between flushes), and it arrives ordered by
+    ///     shard name so that two writers racing over the same rows can never take their locks in
+    ///     opposite orders.
     ///     </para>
-    ///     The default implementation degrades to the single-state overload per shard so existing
-    ///     stores keep working unchanged until they implement a true batched write.
+    ///     <para>
+    ///     <b>One row per transaction — required, not an optimization.</b> Implementations MUST NOT fold
+    ///     the batch into a single multi-row statement (an <c>UPDATE … FROM unnest(…)</c>, a
+    ///     <c>VALUES</c> join, or several statements inside one explicit transaction). Such a statement
+    ///     takes a row lock on every shard in the batch and holds all of them until it commits, so one
+    ///     slow projection batch sitting on one row stalls the telemetry write of every OTHER shard on
+    ///     the database behind it — and, transitively, whatever those shards were waiting to do. It was
+    ///     measured (marten#5167): an unrelated shard's progress write, contending with nothing, timed
+    ///     out after 4s queued behind a telemetry statement that had locked its row on the way to a
+    ///     different, genuinely contended one. Rewritten as one autocommit statement per row the same
+    ///     collision clears in ~1ms and only the genuinely contended row waits. Amortize the CONNECTION,
+    ///     not the transaction: N single-row statements on one rented connection cost one rent and N
+    ///     round trips, and answer jasperfx#553 in full.
+    ///     </para>
+    ///     <para>
+    ///     Implementations should also skip rows whose telemetry is unchanged (e.g. a
+    ///     <c>… AND (col IS DISTINCT FROM $n OR …)</c> guard). The progression table is small and hot,
+    ///     and an unconditional <c>SET</c> gives every matched row a new tuple version on every flush;
+    ///     each avoided rewrite is also an avoided row lock.
+    ///     </para>
+    ///     The default implementation degrades to the single-state overload per shard, which already
+    ///     satisfies the one-row-per-transaction rule.
     /// </summary>
-    /// <param name="states">The latest published state per shard, at most one entry per shard name.</param>
+    /// <param name="states">
+    ///     The latest published state per shard, at most one entry per shard name, ordered by shard name.
+    /// </param>
     /// <param name="token"></param>
     async Task WriteExtendedProgressionAsync(IReadOnlyList<ShardState> states, CancellationToken token = default)
     {