diff --git a/src/AndreGoepel.Marten.Identity/Users/UserStore.cs b/src/AndreGoepel.Marten.Identity/Users/UserStore.cs index f010ba8..bd6aaad 100644 --- a/src/AndreGoepel.Marten.Identity/Users/UserStore.cs +++ b/src/AndreGoepel.Marten.Identity/Users/UserStore.cs @@ -183,19 +183,27 @@ public async Task UpdateAsync(TUser user, CancellationToken canc { var userId = UserId.Parse(user.Id); - var existingUser = await querySession - .Query() - .FirstOrDefaultAsync(x => x.Id == user.Id, token: cancellationToken); + using var session = documentStore.LightweightSession(); + + // Serialize the whole read-modify-write on the stream. Taking the exclusive + // stream lock *before* reading current state (rather than reading from an + // earlier, unlocked query) closes the window in which a concurrent lockout + // increment could land between the read and this full-state append and be + // silently overwritten — the same accumulation race guarded on the atomic + // lockout path (#22, #70). The re-read below therefore observes the latest + // committed state under the lock. + await session.Events.AppendExclusive(userId.Value); + + var existingUser = await session.LoadAsync(userId.Value, cancellationToken); if (existingUser != null) { // Lockout state (AccessFailedCount / LockoutEnd) is owned exclusively - // by the atomic IUserLockoutStore methods, which serialize writes on - // the stream. The generic update path runs from a request-start - // snapshot with no concurrency control, so it must never carry a stale - // lockout value back into the stream — doing so would resurrect the - // failed-login accumulation race (#22). Treat the persisted values as - // authoritative here. + // by the atomic IUserLockoutStore methods. The generic update path + // carries the caller's request-start snapshot, so it must never write a + // stale lockout value back into the stream — doing so would resurrect the + // failed-login accumulation race (#22). Treat the just-read committed + // values as authoritative here. user.AccessFailedCount = existingUser.AccessFailedCount; user.LockoutEnd = existingUser.LockoutEnd; @@ -206,11 +214,10 @@ public async Task UpdateAsync(TUser user, CancellationToken canc user.Deletable = false; } + // Nothing changed — release the lock without churning the stream. if (existingUser != null && existingUser.AreEqual(user)) return IdentityResult.Success; - using var session = documentStore.LightweightSession(); - session.Events.Append( userId.Value, new UserUpdated(userId) diff --git a/tests/AndreGoepel.Marten.Identity.IntegrationTests/Users/UserStoreTests.cs b/tests/AndreGoepel.Marten.Identity.IntegrationTests/Users/UserStoreTests.cs index 5766dbd..47a2f60 100644 --- a/tests/AndreGoepel.Marten.Identity.IntegrationTests/Users/UserStoreTests.cs +++ b/tests/AndreGoepel.Marten.Identity.IntegrationTests/Users/UserStoreTests.cs @@ -336,6 +336,52 @@ public async Task UpdateAsync_DoesNotRegressConcurrentlyChangedLockoutState() Assert.Equal(3, persisted.AccessFailedCount); } + [Fact] + public async Task UpdateAsync_ConcurrentWithLockoutIncrements_LosesNoIncrements() + { + // #70: the generic update path reads current state then appends a full-state + // event. Without holding the exclusive stream lock across that read-modify-write, + // a lockout increment landing in between would be silently overwritten by the + // stale counter — the #22 race re-entering through UpdateAsync. Both paths now + // serialize on the same stream lock, so no increment is lost even when profile + // updates run concurrently with failed-login counting. + // Arrange + var store = UserStoreTestHelpers.BuildStore(fixture.Store); + var user = UserStoreTestHelpers.NewUser(); + await store.CreateAsync(user, Ct); + + const int increments = 15; + const int updates = 15; + + // Act — interleave failed-login increments with unrelated profile updates, + // each from its own freshly loaded snapshot (as independent requests would). + var incrementTasks = Enumerable + .Range(0, increments) + .Select(async _ => + { + await using var session = fixture.Store.QuerySession(); + var snapshot = await session.LoadAsync(user.UserId.Value, Ct); + await store.IncrementAccessFailedCountAsync(snapshot!, Ct); + }); + + var updateTasks = Enumerable + .Range(0, updates) + .Select(async i => + { + await using var session = fixture.Store.QuerySession(); + var snapshot = await session.LoadAsync(user.UserId.Value, Ct); + snapshot!.PhoneNumber = $"+49 555 {i:0000}"; + await store.UpdateAsync(snapshot, Ct); + }); + + await Task.WhenAll(incrementTasks.Concat(updateTasks)); + + // Assert — every increment accumulated despite the concurrent updates. + await using var verify = fixture.Store.QuerySession(); + var persisted = await verify.LoadAsync(user.UserId.Value, Ct); + Assert.Equal(increments, persisted!.AccessFailedCount); + } + #endregion #region Security stamp (session invalidation)