fix: serialize UserStore.UpdateAsync under the exclusive stream lock (#70)

src/AndreGoepel.Marten.Identity/Users/UserStore.cs

@@ -183,19 +183,27 @@ public async Task<IdentityResult> UpdateAsync(TUser user, CancellationToken canc
         {
             var userId = UserId.Parse(user.Id);
 
-            var existingUser = await querySession
-                .Query<TUser>()
-                .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<User>(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<IdentityResult> 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)

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>(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>(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>(user.UserId.Value, Ct);
+        Assert.Equal(increments, persisted!.AccessFailedCount);
+    }
+
     #endregion
 
     #region Security stamp (session invalidation)