diff --git a/src/Testing/CoreTests/Tracking/envelope_history_is_thread_safe_3968.cs b/src/Testing/CoreTests/Tracking/envelope_history_is_thread_safe_3968.cs new file mode 100644 index 0000000000..fcf811bc70 --- /dev/null +++ b/src/Testing/CoreTests/Tracking/envelope_history_is_thread_safe_3968.cs @@ -0,0 +1,138 @@ +using Wolverine.Tracking; +using Wolverine.Transports; +using Xunit; + +namespace CoreTests.Tracking; + +/// +/// GH-3968, backporting GH-3069/GH-3070 to the 5.x line. kept its +/// s in a bare List<T> with no synchronisation, while records arrive +/// concurrently. A resize racing a read threw NullReferenceException out of IsComplete(), which +/// surfaced as an AggregateException from TrackedSession.AssertNoExceptionsWereThrown() with no +/// hint of the real cause. +/// +/// +/// GH-3069 reported it through RecordCrossApplication with RabbitMQ. GH-3968 confirmed the same +/// race independently on the purely LOCAL path with no external transport, reached through +/// WolverineRuntime.MessageSucceeded — so it races on handler-completion threads too, not only +/// transport-listener threads. Their trigger was MultipleHandlerBehavior.Separated: one +/// ExecuteAndWaitAsync fanning out to many envelopes completing at once. Reproduced roughly 1 run in 7 +/// of a ~1850-test suite; ≈3% over a longer history. +/// +/// A race that shows up 3% of the time in a full suite is no use as a regression test, so these drive +/// the contention directly and hard. All three fail on the unfixed 5.x class, on every run. +/// +public class envelope_history_is_thread_safe_3968 +{ + private static EnvelopeRecord RecordOf(MessageEventType eventType) + { + var envelope = new Envelope + { + Message = new object(), + Destination = new Uri($"{TransportConstants.Local}://one"), + MessageType = "thing" + }; + + return new EnvelopeRecord(eventType, envelope, 0, null); + } + + /// + /// Writers racing readers, which is the reported shape: the resize inside _records.Add is what the + /// reader trips over. + /// + [Fact] + public async Task recording_while_reading_does_not_corrupt_the_history() + { + var history = new EnvelopeHistory(Guid.NewGuid()); + using var start = new Barrier(participantCount: 8); + + var tasks = Enumerable.Range(0, 8).Select(i => Task.Run(() => + { + start.SignalAndWait(); + + for (var j = 0; j < 2_000; j++) + { + if (i % 2 == 0) + { + history.RecordLocally(RecordOf(MessageEventType.Sent)); + } + else + { + // Deliberately the ENUMERATING readers. IsComplete() and Has() walk the list with an + // indexed for loop, and List answers Count without enumerating, so all three stay + // quiet under concurrent mutation -- they can read a torn value but they do not throw, + // and a test built on them passes on the unfixed class. Message and MessageFor run + // LINQ over the list, which is what actually trips "Collection was modified". + _ = history.Message; + _ = history.MessageFor(MessageEventType.Sent); + _ = history.Records.ToArray(); + + // Kept for coverage of the indexed readers, which still need the lock to avoid + // reading a half-written slot even though they cannot throw + history.IsComplete(); + history.Has(MessageEventType.Sent); + } + } + })).ToArray(); + + // No assertion beyond "nothing threw": an unsynchronised List resizing under a concurrent + // enumeration throws NullReferenceException or InvalidOperationException, which is the bug. + await Task.WhenAll(tasks); + } + + /// + /// The completion path specifically. MessageSucceeded walks every record marking it complete while + /// other threads are still adding, and GH-3968's stack traces land in exactly that loop. + /// + [Fact] + public async Task concurrent_completions_do_not_corrupt_the_history() + { + var history = new EnvelopeHistory(Guid.NewGuid()); + using var start = new Barrier(participantCount: 8); + + var tasks = Enumerable.Range(0, 8).Select(i => Task.Run(() => + { + start.SignalAndWait(); + + for (var j = 0; j < 1_000; j++) + { + history.RecordLocally(RecordOf(MessageEventType.Sent)); + + // Two completions in the same instant is what the report saw -- two identical inner + // exceptions from one ExecuteAndWaitAsync + history.RecordLocally(RecordOf(MessageEventType.MessageSucceeded)); + history.IsComplete(); + } + })).ToArray(); + + await Task.WhenAll(tasks); + + // Every record was added, none lost to a torn write + history.Records.Count().ShouldBe(8 * 1_000 * 2); + } + + /// + /// Records must hand out a SNAPSHOT. Returning the live list lets a caller enumerate it while a + /// recording thread is adding, which is the same race one level removed -- and on 5.x it was + /// public IEnumerable<EnvelopeRecord> Records => _records;. + /// + [Fact] + public async Task Records_hands_out_a_snapshot_rather_than_the_live_list() + { + var history = new EnvelopeHistory(Guid.NewGuid()); + history.RecordLocally(RecordOf(MessageEventType.Sent)); + + var snapshot = history.Records; + var countAtCapture = snapshot.Count(); + + await Task.Run(() => + { + for (var i = 0; i < 500; i++) history.RecordLocally(RecordOf(MessageEventType.Sent)); + }); + + // The captured sequence is unchanged by everything added after it -- and enumerating it again here + // would throw "Collection was modified" if it were the live list + snapshot.Count().ShouldBe(countAtCapture); + history.Records.Count().ShouldBe(countAtCapture + 500); + } +} diff --git a/src/Wolverine/Tracking/EnvelopeHistory.cs b/src/Wolverine/Tracking/EnvelopeHistory.cs index 90527f7432..0f9cdb644e 100644 --- a/src/Wolverine/Tracking/EnvelopeHistory.cs +++ b/src/Wolverine/Tracking/EnvelopeHistory.cs @@ -5,6 +5,13 @@ namespace Wolverine.Tracking; internal class EnvelopeHistory { + // GH-3069/GH-3070, backported from 6.x. Records arrive concurrently -- from parallel transport + // listener threads when external transports are tracked, and equally from HANDLER-COMPLETION + // threads on the purely local path via WolverineRuntime.MessageSucceeded -- so every access to + // _records has to be synchronized. Unsynchronized, a List resize racing a read threw + // NullReferenceException out of IsComplete(), surfacing as an AggregateException from + // TrackedSession.AssertNoExceptionsWereThrown() with no hint of the real cause. + private readonly object _lock = new(); private readonly List _records = new(); public EnvelopeHistory(Guid envelopeId) @@ -18,12 +25,26 @@ public object? Message { get { - return _records - .FirstOrDefault(x => x.Envelope!.Message != null)?.Envelope!.Message; + lock (_lock) + { + return _records + .FirstOrDefault(x => x.Envelope!.Message != null)?.Envelope!.Message; + } } } - public IEnumerable Records => _records; + public IEnumerable Records + { + get + { + // A snapshot, not the live list: handing _records out directly lets a caller enumerate it + // while a recording thread is adding to it. + lock (_lock) + { + return _records.ToArray(); + } + } + } private EnvelopeRecord? lastOf(MessageEventType eventType) { @@ -51,141 +72,147 @@ private void markLastCompleted(MessageEventType eventType) // ReSharper disable once CyclomaticComplexity public void RecordLocally(EnvelopeRecord record) { - switch (record.MessageEventType) + lock (_lock) { - case MessageEventType.Sent: - // Not tracking anything outgoing - // when it's testing locally - if (record.Envelope!.Destination?.Scheme != TransportConstants.Local || - record.Envelope.MessageType == TransportConstants.ScheduledEnvelope) - { + switch (record.MessageEventType) + { + case MessageEventType.Sent: + // Not tracking anything outgoing + // when it's testing locally + if (record.Envelope!.Destination?.Scheme != TransportConstants.Local || + record.Envelope.MessageType == TransportConstants.ScheduledEnvelope) + { + record.IsComplete = true; + record.WasScheduled = true; + } + + if (record.Envelope.Status == EnvelopeStatus.Scheduled) + { + record.WasScheduled = true; + record.IsComplete = true; + } + + break; + + case MessageEventType.Received: + if (record.Envelope!.Destination?.Scheme == TransportConstants.Local) + { + markLastCompleted(MessageEventType.Sent); + } + + break; + + case MessageEventType.ExecutionStarted: + // Nothing special here + break; + + + case MessageEventType.ExecutionFinished: + markLastCompleted(MessageEventType.ExecutionStarted); record.IsComplete = true; - record.WasScheduled = true; - } + break; - if (record.Envelope.Status == EnvelopeStatus.Scheduled) - { - record.WasScheduled = true; - record.IsComplete = true; - } - - break; - - case MessageEventType.Received: - if (record.Envelope!.Destination?.Scheme == TransportConstants.Local) - { - markLastCompleted(MessageEventType.Sent); - } + case MessageEventType.NoHandlers: + case MessageEventType.NoRoutes: + case MessageEventType.MessageFailed: + case MessageEventType.MessageSucceeded: + case MessageEventType.Discarded: + case MessageEventType.MovedToErrorQueue: + // The message is complete + foreach (var envelopeRecord in _records) envelopeRecord.IsComplete = true; - break; - - case MessageEventType.ExecutionStarted: - // Nothing special here - break; - - - case MessageEventType.ExecutionFinished: - markLastCompleted(MessageEventType.ExecutionStarted); - record.IsComplete = true; - break; - - case MessageEventType.NoHandlers: - case MessageEventType.NoRoutes: - case MessageEventType.MessageFailed: - case MessageEventType.MessageSucceeded: - case MessageEventType.Discarded: - case MessageEventType.MovedToErrorQueue: - // The message is complete - foreach (var envelopeRecord in _records) envelopeRecord.IsComplete = true; + record.IsComplete = true; - record.IsComplete = true; + break; - break; + case MessageEventType.Requeued: + // Do nothing, just informative + break; - case MessageEventType.Requeued: - // Do nothing, just informative - break; + case MessageEventType.AutoFaultPublished: + // Informational marker derived from the outgoing Sent envelope; never blocks completion. + record.IsComplete = true; + break; - case MessageEventType.AutoFaultPublished: - // Informational marker derived from the outgoing Sent envelope; never blocks completion. - record.IsComplete = true; - break; + default: + throw new ArgumentOutOfRangeException(nameof(record.MessageEventType), record.MessageEventType, null); + } - default: - throw new ArgumentOutOfRangeException(nameof(record.MessageEventType), record.MessageEventType, null); + _records.Add(record); } - - _records.Add(record); } public void RecordCrossApplication(EnvelopeRecord record) { - switch (record.MessageEventType) + lock (_lock) { - case MessageEventType.Sent: - if (record.Envelope!.Status == EnvelopeStatus.Scheduled) - { - record.WasScheduled = true; - record.IsComplete = true; + switch (record.MessageEventType) + { + case MessageEventType.Sent: + if (record.Envelope!.Status == EnvelopeStatus.Scheduled) + { + record.WasScheduled = true; + record.IsComplete = true; + + record.TryUseInnerFromScheduledEnvelope(); + } + + // This can be out of order with Rabbit MQ *somehow*, so: + var received = _records.LastOrDefault(x => x.MessageEventType == MessageEventType.Received); + if (received != null) + { + record.IsComplete = true; + } + + break; + + case MessageEventType.ExecutionStarted: + break; + + case MessageEventType.Received: + markLastCompleted(MessageEventType.Sent); + break; - record.TryUseInnerFromScheduledEnvelope(); - } - // This can be out of order with Rabbit MQ *somehow*, so: - var received = _records.LastOrDefault(x => x.MessageEventType == MessageEventType.Received); - if (received != null) - { + case MessageEventType.ExecutionFinished: + markLastCompleted(MessageEventType.ExecutionStarted, record.UniqueNodeId); record.IsComplete = true; - } - - break; - - case MessageEventType.ExecutionStarted: - break; - - case MessageEventType.Received: - markLastCompleted(MessageEventType.Sent); - break; + break; + + case MessageEventType.MovedToErrorQueue: + case MessageEventType.MessageFailed: + case MessageEventType.Discarded: + case MessageEventType.MessageSucceeded: + // The message is complete + for (var i = 0; i < _records.Count; i++) + { + if (_records[i].UniqueNodeId == record.UniqueNodeId) + _records[i].IsComplete = true; + } + record.IsComplete = true; - case MessageEventType.ExecutionFinished: - markLastCompleted(MessageEventType.ExecutionStarted, record.UniqueNodeId); - record.IsComplete = true; - break; - - case MessageEventType.MovedToErrorQueue: - case MessageEventType.MessageFailed: - case MessageEventType.Discarded: - case MessageEventType.MessageSucceeded: - // The message is complete - for (var i = 0; i < _records.Count; i++) - { - if (_records[i].UniqueNodeId == record.UniqueNodeId) - _records[i].IsComplete = true; - } - - record.IsComplete = true; + break; - break; + case MessageEventType.NoHandlers: + case MessageEventType.NoRoutes: + record.IsComplete = true; + break; - case MessageEventType.NoHandlers: - case MessageEventType.NoRoutes: - record.IsComplete = true; - break; + case MessageEventType.Requeued: + break; - case MessageEventType.Requeued: - break; + case MessageEventType.AutoFaultPublished: + // Informational marker derived from the outgoing Sent envelope; never blocks completion. + record.IsComplete = true; + break; - case MessageEventType.AutoFaultPublished: - // Informational marker derived from the outgoing Sent envelope; never blocks completion. - record.IsComplete = true; - break; + default: + throw new ArgumentOutOfRangeException(nameof(record.MessageEventType), record.MessageEventType, null); + } - default: - throw new ArgumentOutOfRangeException(nameof(record.MessageEventType), record.MessageEventType, null); + _records.Add(record); } - - _records.Add(record); } private void markLastCompleted(MessageEventType eventType, Guid uniqueNodeId) @@ -199,28 +226,37 @@ private void markLastCompleted(MessageEventType eventType, Guid uniqueNodeId) public bool IsComplete() { - for (var i = 0; i < _records.Count; i++) + lock (_lock) { - if (!_records[i].IsComplete) return false; - } + for (var i = 0; i < _records.Count; i++) + { + if (!_records[i].IsComplete) return false; + } - return true; + return true; + } } public bool Has(MessageEventType eventType) { - for (var i = 0; i < _records.Count; i++) + lock (_lock) { - if (_records[i].MessageEventType == eventType) return true; - } + for (var i = 0; i < _records.Count; i++) + { + if (_records[i].MessageEventType == eventType) return true; + } - return false; + return false; + } } public object? MessageFor(MessageEventType eventType) { - return _records.Where(x => x.MessageEventType == eventType) - .LastOrDefault(x => x.Envelope!.Message != null)?.Envelope!.Message; + lock (_lock) + { + return _records.Where(x => x.MessageEventType == eventType) + .LastOrDefault(x => x.Envelope!.Message != null)?.Envelope!.Message; + } } public override string ToString()