The extended-progression batch write no longer builds a lock convoy (#5167)

docs/events/projections/async-daemon.md

@@ -745,8 +745,18 @@ opts.Projections.ExtendedProgressionHeartbeatInterval = TimeSpan.FromSeconds(30)
 ```
 
 Budget it as `databases-on-this-node / interval` connection acquisitions per second. All shard states
-that arrive within one interval are coalesced into a single batched `UPDATE`, so the cost scales with
-the number of databases rather than the number of shards.
+that arrive within one interval are coalesced into a single flush on one rented connection, so the
+connection cost scales with the number of databases rather than the number of shards.
+
+Within that flush each shard's row is written by its **own single-row statement, in its own implicit
+transaction**. That is deliberate and load-bearing: a batch is there to amortize the *connection*, not
+the transaction. Marten 9.22.4 and earlier folded the flush into one `UPDATE ... FROM unnest(...)`,
+which takes a row lock on every shard in the batch and holds all of them until it commits — so a single
+slow projection batch sitting on one progression row stalled the telemetry write of every *other* shard
+on that database, and whatever was queued behind those. If you are diagnosing lock waits on
+`mt_event_progression` on an older version, note that `pg_blocking_pids()` reports only *direct*
+blockers: the telemetry `UPDATE` genuinely is the blocker, and the projection transaction that is the
+real root is one hop further down the chain. Walk it recursively.
 
 ::: tip
 Status transitions are always written immediately regardless of this setting, so `agent_status`,

src/DaemonTests/extended_progression_batch_write.cs

@@ -6,6 +6,8 @@
 using JasperFx.Events.Projections;
 using Marten.Events.Aggregation;
 using Marten.Storage;
+using Marten.Testing.Harness;
+using Npgsql;
 using Shouldly;
 using Weasel.Core;
 using Xunit;
@@ -33,9 +35,14 @@ public void Apply(BatchTelemetryEvent @event, BatchTelemetryStream projection) {
 
 // jasperfx#553 — the batched extended-progression write. The JasperFx.Events ExtendedProgressionWriter
 // coalesces every shard's heartbeat on a database into one batch per flush interval; Marten's overload
-// must land the whole batch in ONE round trip with the exact semantics of
+// must land the whole batch on ONE rented connection with the exact semantics of
 // mt_mark_event_progression_extended: update-only telemetry decoration of existing progression rows,
 // never INSERT, never touch last_seq_id / last_updated.
+//
+// #5167 — and it must land as one single-row statement per shard, each its own implicit transaction.
+// The batch amortizes the CONNECTION, not the transaction: a multi-row statement holds a row lock on
+// every shard in the batch until it commits, so one slow projection batch on one row stalls every other
+// shard's telemetry (and whatever queues behind that) on the whole database.
 public class extended_progression_batch_write: DaemonContext
 {
     public extended_progression_batch_write(ITestOutputHelper output): base(output)
@@ -161,6 +168,110 @@ await database.WriteExtendedProgressionAsync([
         missing.status.ShouldBeNull();
     }
 
+    // #5167 — the lock-convoy regression. A row that is locked by an in-flight projection batch is the
+    // normal case, not an exotic one, and the batch write is going to wait on it either way. What must
+    // NOT happen is the batch dragging every OTHER shard's row into that wait: measured against
+    // PostgreSQL, an unrelated shard's progress write timed out after 4s queued behind a telemetry
+    // statement that had locked its row on the way to a different, genuinely contended one.
+    //
+    // The rows are written in shard-name order, so "BatchTelemetryStream:All" goes first and the write
+    // then parks on the deliberately-locked "OtherBatchTelemetry:All". Seeing the first row's telemetry
+    // from ANOTHER connection while the batch is still parked is proof that its write committed on its
+    // own — under a single multi-row statement nothing would be visible until the whole batch committed,
+    // and the row would still be locked.
+    [Fact]
+    public async Task a_contended_row_does_not_hold_the_locks_of_the_rows_already_written()
+    {
+        await seedProgressionRowsAsync();
+
+        var database = (MartenDatabase)theStore.Storage.Database;
+        var schema = theStore.Events.DatabaseSchemaName;
+
+        await using var blocker = new NpgsqlConnection(ConnectionSource.ConnectionString);
+        await blocker.OpenAsync(TestContext.Current.CancellationToken);
+        await using var blocking = await blocker.BeginTransactionAsync(TestContext.Current.CancellationToken);
+
+        // Stands in for a projection batch transaction sitting on its own progression row
+        await blocker
+            .CreateCommand(
+                $"update {schema}.mt_event_progression set last_seq_id = last_seq_id where name = 'OtherBatchTelemetry:All'")
+            .ExecuteNonQueryAsync(TestContext.Current.CancellationToken);
+
+        var write = database.WriteExtendedProgressionAsync([
+            telemetry("BatchTelemetryStream:All", "Running", node: 3),
+            telemetry("OtherBatchTelemetry:All", "Paused", "boom")
+        ], TestContext.Current.CancellationToken);
+
+        // The first row's write commits on its own while the batch is parked on the contended one
+        var deadline = DateTimeOffset.UtcNow.AddSeconds(3);
+        while ((await readRowAsync("BatchTelemetryStream:All")).status == null)
+        {
+            DateTimeOffset.UtcNow.ShouldBeLessThan(deadline,
+                "The first row's telemetry never became visible -- the batch is holding its lock");
+            await Task.Delay(50, TestContext.Current.CancellationToken);
+        }
+
+        // ...and it is genuinely still parked, so that visibility was not just "the batch finished"
+        write.IsCompleted.ShouldBeFalse();
+
+        // The counterfactual from the report: an unrelated writer touching the already-written row must
+        // not queue behind the batch. lock_timeout makes "it waited" a failure instead of a hang.
+        await using (var unrelated = new NpgsqlConnection(ConnectionSource.ConnectionString))
+        {
+            await unrelated.OpenAsync(TestContext.Current.CancellationToken);
+            await unrelated.CreateCommand($"""
+                set lock_timeout = '2s';
+                update {schema}.mt_event_progression set last_seq_id = last_seq_id
+                where name = 'BatchTelemetryStream:All';
+                """).ExecuteNonQueryAsync(TestContext.Current.CancellationToken);
+        }
+
+        await blocking.RollbackAsync(TestContext.Current.CancellationToken);
+
+        await write;
+        (await readRowAsync("OtherBatchTelemetry:All")).status.ShouldBe("Paused");
+    }
+
+    // #5167 Finding 3 — the SET list used to be unconditional, so every flush gave every matched row a
+    // new tuple version whether anything had changed or not, on a small hot table. xmin is the
+    // inserting transaction of the live tuple, so an unchanged xmin is exactly "this row was not
+    // rewritten".
+    [Fact]
+    public async Task replaying_identical_telemetry_does_not_rewrite_the_row()
+    {
+        await seedProgressionRowsAsync();
+
+        var database = (MartenDatabase)theStore.Storage.Database;
+        var state = telemetry("BatchTelemetryStream:All", "Running", node: 3);
+
+        await database.WriteExtendedProgressionAsync([state], TestContext.Current.CancellationToken);
+        var written = await readTupleVersionAsync("BatchTelemetryStream:All");
+
+        // Byte-identical replay: nothing to change, so nothing is written
+        await database.WriteExtendedProgressionAsync([state], TestContext.Current.CancellationToken);
+        (await readTupleVersionAsync("BatchTelemetryStream:All")).ShouldBe(written);
+
+        // ...but a real change still lands
+        await database.WriteExtendedProgressionAsync([
+            telemetry("BatchTelemetryStream:All", "Paused", "boom", node: 3)
+        ], TestContext.Current.CancellationToken);
+
+        (await readTupleVersionAsync("BatchTelemetryStream:All")).ShouldNotBe(written);
+        (await readRowAsync("BatchTelemetryStream:All")).status.ShouldBe("Paused");
+    }
+
+    private async Task<string> readTupleVersionAsync(string shard)
+    {
+        await using var session = theStore.QuerySession();
+        var raw = await session.Connection
+            .CreateCommand(
+                $"select xmin::text from {theStore.Events.DatabaseSchemaName}.mt_event_progression where name = :name")
+            .With("name", shard)
+            .ExecuteScalarAsync();
+
+        return (string)raw!;
+    }
+
     [Fact]
     public async Task an_empty_batch_is_a_no_op_and_a_single_state_batch_delegates()
     {