From 75ddea0145e34abcfabb594e3b428e9d8b82c807 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 23 Oct 2026 16:21:36 +0200 Subject: [PATCH 1/3] test: reproduce initialization retry and concurrency failures (#85) On unchanged product code: 5 failing regressions, 1 passing idempotence test. Cover index/checkpoint setup, shared concurrent outcomes, and recursive entry. Co-authored by Alessandro Giorgetti and Codex (OpenAI). --- .../AcceptanceTests/Issues/Issue85Tests.cs | 303 ++++++++++++++++++ 1 file changed, 303 insertions(+) create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue85Tests.cs diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue85Tests.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue85Tests.cs new file mode 100644 index 0000000..37e5e57 --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue85Tests.cs @@ -0,0 +1,303 @@ +using FluentAssertions; +using global::MongoDB.Bson; +using global::MongoDB.Driver; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Serialization; + +#if MSTEST +using Microsoft.VisualStudio.TestTools.UnitTesting; +#endif + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + public abstract class Issue85InitializationConcern : SpecificationBase + { + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(20); + protected IMongoDatabase _database = null!; + protected InitializationProbe _engine = null!; + + protected override void Context() + { + var options = new MongoPersistenceOptions(); + var builder = new MongoUrlBuilder(AcceptanceTestMongoPersistenceFactory.GetConnectionString()) + { + DatabaseName = $"issue85-{Guid.NewGuid():N}" + }; + _database = options.ConnectToDatabase(builder.ToString()); + _engine = new InitializationProbe(_database, options); + } + + protected ConcurrentResult RunConcurrentAttempt(bool fail, bool recurse) + { + using var entered = new ManualResetEventSlim(); + using var release = new ManualResetEventSlim(); + var result = new ConcurrentResult(); + _engine.BeforeInitialize = () => + { + // Only the original attempt is gated. An accidental second attempt is observable. + if (_engine.Attempts != 1) + { + return; + } + if (recurse) + { + result.RecursiveError = Catch.Exception(_engine.Initialize); + } + entered.Set(); + if (!release.Wait(Timeout)) + { + throw new TimeoutException("The initialization test did not release its gate."); + } + if (fail) + { + throw new InvalidOperationException("Injected initialization failure."); + } + }; + var first = new Thread(() => result.FirstError = Catch.Exception(_engine.Initialize)) { IsBackground = true }; + var second = new Thread(() => result.SecondError = Catch.Exception(_engine.Initialize)) { IsBackground = true }; + first.Start(); + try + { + entered.Wait(Timeout).Should().BeTrue("the first caller must enter initialization"); + second.Start(); + SpinWait.SpinUntil(() => + (second.ThreadState & (ThreadState.WaitSleepJoin | ThreadState.Stopped)) != 0, + Timeout).Should().BeTrue("the second caller must either wait or return"); + result.WaiterWasBlocked = (second.ThreadState & ThreadState.WaitSleepJoin) != 0; + } + finally + { + release.Set(); + first.Join(Timeout).Should().BeTrue("the original attempt must finish"); + if ((second.ThreadState & ThreadState.Unstarted) == 0) + { + second.Join(Timeout).Should().BeTrue("the waiting caller must finish"); + } + } + return result; + } + + protected void AssertAllIndexesExist() + { + ReadIndexCounts().Should().Equal(6, 2, 3); + } + + protected int[] ReadIndexCounts() + { + return new[] + { + _database.GetCollection("Commits").Indexes.List().ToList().Count, + _database.GetCollection("Snapshots").Indexes.List().ToList().Count, + _database.GetCollection("Streams").Indexes.List().ToList().Count + }; + } + + protected override void Cleanup() + { + _engine?.Dispose(); + if (_database != null) + { + _database.Client.DropDatabase(_database.DatabaseNamespace.DatabaseName); + } + } + + protected sealed class ConcurrentResult + { + public bool WaiterWasBlocked { get; set; } + public Exception? FirstError { get; set; } + public Exception? SecondError { get; set; } + public Exception? RecursiveError { get; set; } + } + + protected sealed class InitializationProbe(IMongoDatabase database, MongoPersistenceOptions options) + : MongoPersistenceEngine(database, new DocumentObjectSerializer(), options) + { + private int _attempts; + public int Attempts => Volatile.Read(ref _attempts); + public Action? BeforeInitialize { get; set; } + + protected override void TryMongo(Action callback) + { + Interlocked.Increment(ref _attempts); + BeforeInitialize?.Invoke(); + base.TryMongo(callback); + } + } + } + +#if MSTEST + [TestClass] +#endif + public class Issue_85_index_creation_retry : Issue85InitializationConcern + { + private Exception? _firstError; + + protected override void Context() + { + base.Context(); + var commits = _database.GetCollection("Commits"); + commits.Indexes.CreateOne(new CreateIndexModel( + Builders.IndexKeys.Ascending("conflictingField"), + new CreateIndexOptions { Name = MongoCommitIndexes.GetFromCheckpoint })); + _firstError = Catch.Exception(_engine.Initialize); + commits.Indexes.DropOne(MongoCommitIndexes.GetFromCheckpoint); + } + + protected override void Because() + { + _engine.Initialize(); + } + + [Fact] + public void A_failed_index_creation_can_be_retried_after_the_conflict_is_removed() + { + _firstError.Should().BeOfType(); + _engine.Attempts.Should().Be(2, "a failed attempt must not mark storage initialized"); + AssertAllIndexesExist(); + } + } + +#if MSTEST + [TestClass] +#endif + public class Issue_85_checkpoint_generator_retry : Issue85InitializationConcern + { + private Exception? _firstError; + private int[] _indexesAfterFailure = null!; + + protected override void Context() + { + base.Context(); + var commits = _database.GetCollection("Commits"); + commits.InsertOne(new BsonDocument(MongoCommitFields.CheckpointNumber, "not a checkpoint")); + _firstError = Catch.Exception(_engine.Initialize); + _indexesAfterFailure = ReadIndexCounts(); + commits.DeleteMany(Builders.Filter.Empty); + } + + protected override void Because() + { + _engine.Initialize(); + } + + [Fact] + public void A_failed_checkpoint_generator_construction_can_be_retried_after_indexes_succeed() + { + _firstError.Should().BeOfType(); + _indexesAfterFailure.Should().Equal(6, 2, 3); + _engine.Attempts.Should().Be(2, "index creation alone does not finish initialization"); + AssertAllIndexesExist(); + } + } + +#if MSTEST + [TestClass] +#endif + public class Issue_85_successful_initialization : Issue85InitializationConcern + { + protected override void Context() + { + base.Context(); + _engine.Initialize(); + } + + protected override void Because() + { + _engine.Initialize(); + _engine.Initialize(); + } + + [Fact] + public void Successful_initialization_is_idempotent() + { + _engine.Attempts.Should().Be(1); + AssertAllIndexesExist(); + } + } + +#if MSTEST + [TestClass] +#endif + public class Issue_85_concurrent_success : Issue85InitializationConcern + { + private ConcurrentResult _result = null!; + + protected override void Because() + { + _result = RunConcurrentAttempt(fail: false, recurse: false); + } + + [Fact] + public void Concurrent_callers_wait_for_the_same_successful_attempt() + { + _result.WaiterWasBlocked.Should().BeTrue("storage is not ready while initialization is still running"); + _result.FirstError.Should().BeNull(); + _result.SecondError.Should().BeNull(); + _engine.Attempts.Should().Be(1); + AssertAllIndexesExist(); + } + } + +#if MSTEST + [TestClass] +#endif + public class Issue_85_concurrent_failure : Issue85InitializationConcern + { + private ConcurrentResult _result = null!; + private int _attemptsAfterFailure; + + protected override void Context() + { + base.Context(); + _result = RunConcurrentAttempt(fail: true, recurse: false); + _attemptsAfterFailure = _engine.Attempts; + _engine.BeforeInitialize = null; + } + + protected override void Because() + { + _engine.Initialize(); + } + + [Fact] + public void Concurrent_callers_observe_the_same_failure_and_a_later_call_retries() + { + _result.WaiterWasBlocked.Should().BeTrue(); + _result.FirstError.Should().BeOfType(); + _result.SecondError.Should().BeSameAs(_result.FirstError, "waiters must not silently retry a failed attempt"); + _attemptsAfterFailure.Should().Be(1); + _engine.Attempts.Should().Be(2); + AssertAllIndexesExist(); + } + } + +#if MSTEST + [TestClass] +#endif + public class Issue_85_recursive_initialization : Issue85InitializationConcern + { + private ConcurrentResult _result = null!; + + protected override void Context() + { + base.Context(); + _result = RunConcurrentAttempt(fail: false, recurse: true); + } + + protected override void Because() + { + _engine.Initialize(); + } + + [Fact] + public void A_caught_recursive_call_does_not_replace_the_in_progress_attempt() + { + _result.RecursiveError.Should().BeOfType("recursive initialization must fail rather than deadlock or report premature success"); + _result.WaiterWasBlocked.Should().BeTrue(); + _result.FirstError.Should().BeNull(); + _result.SecondError.Should().BeNull(); + _engine.Attempts.Should().Be(1, "a recursive caller must not discard the outer caller's attempt"); + } + } +} From 3fb5929c15c2be5ba7031fd00a3e876dc84aba8d Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 23 Oct 2026 16:21:42 +0200 Subject: [PATCH 2/3] fix: share initialization outcomes and retry failed setup (#85) Publish success after setup, let concurrent callers join the same attempt, and reject recursive initialization without poisoning a running attempt. Co-authored by Alessandro Giorgetti and Codex (OpenAI). --- .../MongoPersistenceEngine.cs | 44 +++++++++++++++++-- 1 file changed, 41 insertions(+), 3 deletions(-) diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs index c59e2c2..6645d1e 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs @@ -19,7 +19,9 @@ public partial class MongoPersistenceEngine : IPersistStreams private const string ConcurrencyException = "E1100"; private static readonly ILogger Logger = LogFactory.BuildLogger(typeof(MongoPersistenceEngine)); private readonly IDocumentSerializer _serializer; - private int _initialized; + private readonly object _initializationLock = new(); + private Lazy? _initialization; + private int _initializingThreadId; private readonly MongoPersistenceOptions _options; private readonly string _systemBucketName; private ICheckpointGenerator? _checkpointGenerator; @@ -79,11 +81,47 @@ public void Dispose() /// public virtual void Initialize() { - if (Interlocked.Increment(ref _initialized) > 1) + Lazy initialization; + lock (_initializationLock) { - return; + if (_initializingThreadId == Environment.CurrentManagedThreadId) + { + throw new InvalidOperationException("Recursive storage initialization is not supported."); + } + initialization = _initialization ??= new Lazy(() => + { + lock (_initializationLock) + { + _initializingThreadId = Environment.CurrentManagedThreadId; + } + try + { + InitializeStorage(); + lock (_initializationLock) + { + _initializingThreadId = 0; + } + return true; + } + catch + { + // Only the executing attempt may enable a retry; + // its waiters retain the failed attempt. + lock (_initializationLock) + { + _initializingThreadId = 0; + _initialization = null; + } + throw; + } + }, LazyThreadSafetyMode.ExecutionAndPublication); } + _ = initialization.Value; + } + + private void InitializeStorage() + { MongoPersistenceEngineLogMessages.InitializingStorage(Logger); TryMongo(() => From def00730b4f359b6977aacca4f193c5686fbb03c Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Mon, 26 Oct 2026 11:20:19 +0100 Subject: [PATCH 3/3] docs: update changelog for #85 --- Changelog.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/Changelog.md b/Changelog.md index 80a4cca..50aeb45 100644 --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,9 @@ # NEventStore.Persistence.MongoDB +## Unreleased + +- Fix initialization races: concurrent callers share a single initialization attempt and observe its success or failure; a later call can retry a failed attempt. Same-thread recursive initialization now fails explicitly. [#85](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/85) + ## 12.1.0 - Added explicit support for net8.0, net9.0, net10.0.