From 76bc1eeeb0dfc70f65edcac63952eefe21db7e72 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Tue, 4 Aug 2026 10:22:29 -0500 Subject: [PATCH] The extended-progression batch write no longer builds a lock convoy (#5167) The batched extended-progression write was one `UPDATE ... FROM unnest(...)` covering every shard in the flush. A multi-row statement takes a row lock on EVERY row it matches and holds all of them until it commits, so one slow projection batch sitting on one progression row stalled the telemetry write of every OTHER shard on that database -- and, transitively, whatever those shards had queued behind it. Reproduced against PostgreSQL: an unrelated projection committing progress for a row the slow batch never touched timed out after 4s, queued behind the telemetry statement, which had locked that row on its way to the genuinely contended one and then stalled there. Rewritten as one single-row statement per shard, each its own implicit transaction, the same collision clears in ~1ms and only the genuinely contended row waits. Blast radius goes from "every shard on the database" to "the shard this write is about". The batch is still one rented connection -- that is what jasperfx#553 was about, and N single-row statements on one connection cost one rent. What it is no longer is one transaction. The writes go in shard-name order so two writers racing over the same rows cannot take their locks in opposite orders. Also adds an `is distinct from` guard so replaying unchanged telemetry is an UPDATE 0 instead of a fresh tuple version. The SET list was unconditional, so every flush rewrote every matched row whether anything had changed or not, on a small hot table; each avoided rewrite is also an avoided row lock. Both new tests fail against the shape they guard: the lock test against the same loop wrapped in an explicit transaction, the replay test against the neutralized guard. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017CTtw2kVRSZKp1p5RTxgAy --- docs/events/projections/async-daemon.md | 14 +- .../extended_progression_batch_write.cs | 113 +++++++++++- .../Storage/MartenDatabase.EventStorage.cs | 163 +++++++++++------- 3 files changed, 220 insertions(+), 70 deletions(-) diff --git a/docs/events/projections/async-daemon.md b/docs/events/projections/async-daemon.md index 96ba88b1f1..329215ed7a 100644 --- a/docs/events/projections/async-daemon.md +++ b/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`, diff --git a/src/DaemonTests/extended_progression_batch_write.cs b/src/DaemonTests/extended_progression_batch_write.cs index 9e9d5932a6..f92141fd73 100644 --- a/src/DaemonTests/extended_progression_batch_write.cs +++ b/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 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() { diff --git a/src/Marten/Storage/MartenDatabase.EventStorage.cs b/src/Marten/Storage/MartenDatabase.EventStorage.cs index 34eb12eaa3..34da60dc57 100644 --- a/src/Marten/Storage/MartenDatabase.EventStorage.cs +++ b/src/Marten/Storage/MartenDatabase.EventStorage.cs @@ -19,6 +19,7 @@ using Marten.Services; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; +using Npgsql; using NpgsqlTypes; using Weasel.Postgresql; @@ -60,17 +61,39 @@ public Task WriteExtendedProgressionAsync(ShardState state, CancellationToken to } /// - /// Persist the extended progression telemetry for a whole batch of shards in ONE round trip on - /// ONE rented connection. The JasperFx.Events ExtendedProgressionWriter coalesces every - /// shard's heartbeat 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 — jasperfx#553). Deliberately a plain UPDATE ... FROM unnest - /// join instead of a database function, so no schema object is added and deployments running - /// AutoCreate.None pick it up without a migration. Semantics match the - /// mt_mark_event_progression_extended function this replaced (the function is still - /// installed for anything calling it directly): update-only telemetry decoration of existing - /// progression rows — never INSERT, never touch last_seq_id / last_updated, shards - /// without a progression row yet are skipped silently. + /// Persist the extended progression telemetry for a whole batch of shards on ONE rented + /// connection, as ONE SINGLE-ROW STATEMENT PER SHARD. The JasperFx.Events + /// ExtendedProgressionWriter coalesces every shard's heartbeat 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 — jasperfx#553). + /// Deliberately plain UPDATE statements instead of a database function, so no schema object is + /// added and deployments running AutoCreate.None pick it up without a migration. Semantics + /// match the mt_mark_event_progression_extended function this replaced (the function is + /// still installed for anything calling it directly): update-only telemetry decoration of + /// existing progression rows — never INSERT, never touch last_seq_id / last_updated, + /// shards without a progression row yet are skipped silently. + /// + /// #5167: this used to be ONE UPDATE … FROM unnest(…) covering the whole batch, and that + /// is a lock convoy. A multi-row 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 — and, transitively, the progress writes + /// queued behind those. Measured against PostgreSQL: 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 this way the same + /// collision clears in ~1ms and only the genuinely contended row waits. The load-bearing property + /// is ONE ROW PER TRANSACTION, so these must stay separate round trips in autocommit — several + /// statements batched into one Npgsql command would share an implicit transaction and reproduce + /// the convoy exactly. What is amortized is the CONNECTION, which is what jasperfx#553 was about. + /// Writes are applied in shard-name order so two writers racing over the same rows cannot take + /// their locks in opposite orders. Being separate transactions, a failure partway through leaves + /// the earlier rows written — correct for best-effort telemetry, where an all-or-nothing batch + /// buys nothing. + /// + /// + /// The is distinct from guard makes a replay of unchanged telemetry an UPDATE 0 + /// rather than a new tuple version. mt_event_progression is small and hot, and every + /// avoided rewrite is also an avoided row lock. + /// /// /// #5048 / jasperfx#565: the four failure_* columns follow a different rule from the rest. /// They are written when the state carries a , CLEARED when a @@ -89,66 +112,47 @@ public async Task WriteExtendedProgressionAsync(IReadOnlyList states await EnsureStorageExistsAsync(typeof(IEvent), token).ConfigureAwait(false); - var names = new string[states.Count]; - var heartbeats = new DateTimeOffset?[states.Count]; - var statuses = new string?[states.Count]; - var reasons = new string?[states.Count]; - var nodes = new int?[states.Count]; - var touchFailures = new bool[states.Count]; - var failureCategories = new string?[states.Count]; - var failureSequences = new long?[states.Count]; - var failureEventTypes = new string?[states.Count]; - var failureTenantIds = new string?[states.Count]; - - for (var i = 0; i < states.Count; i++) - { - var state = states[i]; - - names[i] = state.ShardName; - heartbeats[i] = state.LastHeartbeat; - statuses[i] = state.AgentStatus; - reasons[i] = state.PauseReason; - nodes[i] = state.RunningOnNode; - - var failure = state.Failure; - touchFailures[i] = failure != null || state.Action == ShardAction.Started; - failureCategories[i] = failure?.Category.ToString(); - failureSequences[i] = failure?.Event?.Sequence; - failureEventTypes[i] = failure?.Event?.EventTypeName; - failureTenantIds[i] = failure?.Event?.TenantId; - } - await using var conn = CreateConnection(); try { await conn.OpenAsync(token).ConfigureAwait(false); - await conn.CreateCommand($""" - update {Options.EventGraph.DatabaseSchemaName}.mt_event_progression as p - set heartbeat = t.heartbeat, - agent_status = t.agent_status, - pause_reason = t.pause_reason, - running_on_node = t.running_on_node, - failure_category = case when t.touch_failure then t.failure_category else p.failure_category end, - failure_event_sequence = case when t.touch_failure then t.failure_event_sequence else p.failure_event_sequence end, - failure_event_type = case when t.touch_failure then t.failure_event_type else p.failure_event_type end, - failure_event_tenant_id = case when t.touch_failure then t.failure_event_tenant_id else p.failure_event_tenant_id end - from unnest(:names, :heartbeats, :statuses, :reasons, :nodes, :touch_failures, - :failure_categories, :failure_sequences, :failure_event_types, :failure_tenant_ids) - as t(name, heartbeat, agent_status, pause_reason, running_on_node, touch_failure, - failure_category, failure_event_sequence, failure_event_type, failure_event_tenant_id) - where p.name = t.name - """) - .With("names", names, NpgsqlDbType.Array | NpgsqlDbType.Varchar) - .With("heartbeats", heartbeats, NpgsqlDbType.Array | NpgsqlDbType.TimestampTz) - .With("statuses", statuses, NpgsqlDbType.Array | NpgsqlDbType.Varchar) - .With("reasons", reasons, NpgsqlDbType.Array | NpgsqlDbType.Text) - .With("nodes", nodes, NpgsqlDbType.Array | NpgsqlDbType.Integer) - .With("touch_failures", touchFailures, NpgsqlDbType.Array | NpgsqlDbType.Boolean) - .With("failure_categories", failureCategories, NpgsqlDbType.Array | NpgsqlDbType.Varchar) - .With("failure_sequences", failureSequences, NpgsqlDbType.Array | NpgsqlDbType.Bigint) - .With("failure_event_types", failureEventTypes, NpgsqlDbType.Array | NpgsqlDbType.Varchar) - .With("failure_tenant_ids", failureTenantIds, NpgsqlDbType.Array | NpgsqlDbType.Varchar) - .ExecuteNonQueryAsync(token).ConfigureAwait(false); + + using var command = conn.CreateCommand(extendedProgressionSql()); + + var name = command.Parameters.Add(new NpgsqlParameter("name", NpgsqlDbType.Varchar)); + var heartbeat = command.Parameters.Add(new NpgsqlParameter("heartbeat", NpgsqlDbType.TimestampTz)); + var status = command.Parameters.Add(new NpgsqlParameter("agent_status", NpgsqlDbType.Varchar)); + var reason = command.Parameters.Add(new NpgsqlParameter("pause_reason", NpgsqlDbType.Text)); + var node = command.Parameters.Add(new NpgsqlParameter("running_on_node", NpgsqlDbType.Integer)); + var touchFailure = command.Parameters.Add(new NpgsqlParameter("touch_failure", NpgsqlDbType.Boolean)); + var failureCategory = command.Parameters.Add(new NpgsqlParameter("failure_category", NpgsqlDbType.Varchar)); + var failureSequence = command.Parameters.Add(new NpgsqlParameter("failure_sequence", NpgsqlDbType.Bigint)); + var failureEventType = + command.Parameters.Add(new NpgsqlParameter("failure_event_type", NpgsqlDbType.Varchar)); + var failureTenantId = + command.Parameters.Add(new NpgsqlParameter("failure_tenant_id", NpgsqlDbType.Varchar)); + + // The writer already hands these over sorted, but a direct caller need not, and the ordering + // is what keeps two racing writers from deadlocking against each other. + foreach (var state in states.OrderBy(x => x.ShardName, StringComparer.Ordinal)) + { + var failure = state.Failure; + + name.Value = state.ShardName; + heartbeat.Value = (object?)state.LastHeartbeat ?? DBNull.Value; + status.Value = (object?)state.AgentStatus ?? DBNull.Value; + reason.Value = (object?)state.PauseReason ?? DBNull.Value; + node.Value = (object?)state.RunningOnNode ?? DBNull.Value; + touchFailure.Value = failure != null || state.Action == ShardAction.Started; + failureCategory.Value = (object?)failure?.Category.ToString() ?? DBNull.Value; + failureSequence.Value = (object?)failure?.Event?.Sequence ?? DBNull.Value; + failureEventType.Value = (object?)failure?.Event?.EventTypeName ?? DBNull.Value; + failureTenantId.Value = (object?)failure?.Event?.TenantId ?? DBNull.Value; + + // One ExecuteNonQueryAsync per shard, deliberately: each is its own implicit transaction, + // so this loop never holds more than a single row lock at a time. + await command.ExecuteNonQueryAsync(token).ConfigureAwait(false); + } } finally { @@ -156,6 +160,31 @@ as t(name, heartbeat, agent_status, pause_reason, running_on_node, touch_failure } } + private string extendedProgressionSql() + { + return $""" + update {Options.EventGraph.DatabaseSchemaName}.mt_event_progression as p + set heartbeat = :heartbeat, + agent_status = :agent_status, + pause_reason = :pause_reason, + running_on_node = :running_on_node, + failure_category = case when :touch_failure then :failure_category else p.failure_category end, + failure_event_sequence = case when :touch_failure then :failure_sequence else p.failure_event_sequence end, + failure_event_type = case when :touch_failure then :failure_event_type else p.failure_event_type end, + failure_event_tenant_id = case when :touch_failure then :failure_tenant_id else p.failure_event_tenant_id end + where p.name = :name + and (p.heartbeat is distinct from :heartbeat + or p.agent_status is distinct from :agent_status + or p.pause_reason is distinct from :pause_reason + or p.running_on_node is distinct from :running_on_node + or (:touch_failure + and (p.failure_category is distinct from :failure_category + or p.failure_event_sequence is distinct from :failure_sequence + or p.failure_event_type is distinct from :failure_event_type + or p.failure_event_tenant_id is distinct from :failure_tenant_id))) + """; + } + public async Task FindEventStoreFloorAtTimeAsync(DateTimeOffset timestamp, CancellationToken token) { var sql =