diff --git a/Directory.Build.props b/Directory.Build.props index 5a5d709..5bf6ad2 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -1,7 +1,7 @@ - 2.39.1 + 2.39.2 13 1570;1571;1572;1573;1574;1587;1591;1701;1702;1711;1735;0618 Jeremy D. Miller;Jaedyn Tonee diff --git a/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs b/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs index 08886a4..76a8978 100644 --- a/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs +++ b/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() { diff --git a/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs b/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs index 7f97b51..6087367 100644 --- a/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs +++ b/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs @@ -29,11 +29,15 @@ namespace JasperFx.Events.Daemon; /// DaemonSettings.ExtendedProgressionHeartbeatInterval) to a positive value to restore the /// old behavior — that is the compatibility hatch, not the recommended shape. /// When periodic beats ARE enabled, they are coalesced per shard (latest state wins) and -/// flushed as ONE batched database write per 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 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). +/// 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 +/// , +/// which is what keeps a slow projection batch on one row from stalling every other shard's +/// telemetry behind it (marten#5167). /// Agent status transitions (, , /// ) 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; diff --git a/src/JasperFx.Events/IEventDatabase.cs b/src/JasperFx.Events/IEventDatabase.cs index 4c58f57..e10e447 100644 --- a/src/JasperFx.Events/IEventDatabase.cs +++ b/src/JasperFx.Events/IEventDatabase.cs @@ -228,22 +228,45 @@ Task WriteExtendedProgressionAsync(ShardState state, CancellationToken token = d => Task.CompletedTask; /// - /// Persist the extended progression telemetry for a batch of shards in as few database - /// round-trips as the store can manage — ideally one. The - /// 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 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). /// /// 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. /// - /// The default implementation degrades to the single-state overload per shard so existing - /// stores keep working unchanged until they implement a true batched write. + /// + /// One row per transaction — required, not an optimization. Implementations MUST NOT fold + /// the batch into a single multi-row statement (an UPDATE … FROM unnest(…), a + /// VALUES 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. + /// + /// + /// Implementations should also skip rows whose telemetry is unchanged (e.g. a + /// … AND (col IS DISTINCT FROM $n OR …) guard). The progression table is small and hot, + /// and an unconditional SET gives every matched row a new tuple version on every flush; + /// each avoided rewrite is also an avoided row lock. + /// + /// The default implementation degrades to the single-state overload per shard, which already + /// satisfies the one-row-per-transaction rule. /// - /// The latest published state per shard, at most one entry per shard name. + /// + /// The latest published state per shard, at most one entry per shard name, ordered by shard name. + /// /// async Task WriteExtendedProgressionAsync(IReadOnlyList states, CancellationToken token = default) {