diff --git a/src/DaemonTests/Aggregations/side_effects_in_aggregations.cs b/src/DaemonTests/Aggregations/side_effects_in_aggregations.cs index baa803aa23..ba989e1a0b 100644 --- a/src/DaemonTests/Aggregations/side_effects_in_aggregations.cs +++ b/src/DaemonTests/Aggregations/side_effects_in_aggregations.cs @@ -464,12 +464,30 @@ public class SideEffects2: IRevisioned public class RecordingMessageOutbox: IMessageOutbox { - public readonly List Batches = new(); + private readonly List _batches = new(); + + // The daemon raises side effects for the slices in a batch concurrently, so both of the + // collections in this harness have to be synchronized -- an unguarded List.Add silently drops + // batches and messages, which is what made blue_green_side_effect_gate look flaky (#5065). + // The sibling copy in Composites/composite_rebuild_suppresses_side_effects.cs already locks. + public IReadOnlyList Batches + { + get + { + lock (_batches) + { + return _batches.ToList(); + } + } + } public ValueTask CreateBatch(DocumentSessionBase session) { var batch = new RecordingMessageBatch(); - Batches.Add(batch); + lock (_batches) + { + _batches.Add(batch); + } return new ValueTask(batch); } @@ -479,7 +497,18 @@ public record TenantMessage(string tenantId, object message); public class RecordingMessageBatch: IMessageBatch { - public readonly List Messages = new(); + private readonly List _messages = new(); + + public IReadOnlyList Messages + { + get + { + lock (_messages) + { + return _messages.ToList(); + } + } + } public Task AfterCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token) { @@ -498,7 +527,11 @@ public Task BeforeCommitAsync(IDocumentSession session, IChangeSet commit, Cance public bool BeforeCommitWasCalled { get; set; } public ValueTask PublishAsync(T message, string tenantId) { - Messages.Add(new TenantMessage(tenantId, message)); + lock (_messages) + { + _messages.Add(new TenantMessage(tenantId, message)); + } + return new ValueTask(); } } diff --git a/src/Marten/Events/Aggregation/IMessageBatch.cs b/src/Marten/Events/Aggregation/IMessageBatch.cs index c7e58a016b..23f964b928 100644 --- a/src/Marten/Events/Aggregation/IMessageBatch.cs +++ b/src/Marten/Events/Aggregation/IMessageBatch.cs @@ -3,6 +3,23 @@ namespace Marten.Events.Aggregation; +/// +/// A batch of messages published by projection side effects within a single projection update. +/// +/// +/// +/// Implementations must be thread safe. The async daemon raises side effects for the event +/// slices in one batch concurrently, so is called from +/// several threads against the same instance -- measured at up to 8 +/// simultaneous callers across 10 threads for a single-stream projection catching up over 20 +/// streams. An implementation that appends to an unsynchronized collection silently loses +/// messages under load; see #5065, where exactly that made a daemon test look flaky. +/// +/// +/// The same applies to any collection an uses to track the batches +/// it hands out, since is reached from those threads too. +/// +/// public interface IMessageBatch: IMessageSink, IChangeListener {