fix(auditlog): advance the served-rowid mark under the write lock so a purge clamp is never overwritten stale-high (review 2026-10-21 follow-up L1)

docs/requirements/Component-AuditLog.md

@@ -511,7 +511,18 @@ resets it — before that, a healthy site whose pull serves nothing could leave
 `PushAbandoned`/`Rejected` rows owed (and counted on the backlog) indefinitely. It stays
 conservative in the safe direction (retirement can be delayed, never a row lost). The gRPC handler's ordering
 (retire, *then* read) is what keeps the bound from ever vouching for the batch it is
-about to serve.
+about to serve. The mark is advanced under the writer's write lock — the same lock
+the purge's rowid clamp runs under — so a clamp can never be overwritten with a stale-high
+value by a concurrent pull (review 2026-10-21 follow-up L1).
+
+**The bound protects a late-stamped row for one pull cycle, not forever.** A row inserted
+after a batch was served but stamped *below* central's cursor has a rowid above the mark,
+so the next cursor cannot retire it — but the pull never serves it either (the keyset read
+is already past its timestamp), and as soon as a later pull serves any newer row the mark
+rises above it and the following cursor retires it. Its delivery therefore relies on the
+telemetry **push**, which drains every `Pending` row oldest-first regardless of the pull
+cursor and will normally have acked it (`Forwarded`, untouched by the cursor) within that
+cycle; the served-rowid bound only buys the push that window.
 
 **Liveness note (accepted, documented).** A row that was served but never covered by a
 later cursor — central reconciliation stopped for good — stays owed (`Pending` /

src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs

@@ -856,7 +856,7 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi
         // ~0.85 audit rows/s fell permanently behind and latched the stalled signal.
         // A Forwarded row is acked by central; the pull has nothing to repair there.
         IReadOnlyList<AuditEvent> rows;
-        var before = Volatile.Read(ref _maxServedRowId);
+        long batchMaxRowId;
         lock (_readLock)
         {
             ObjectDisposedException.ThrowIf(_disposed, this);
@@ -896,29 +896,31 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi
             }
             cmd.Parameters.AddWithValue("$limit", batchSize);
 
-            rows = ReadServedRows(cmd, batchSize);
+            rows = ReadServedRows(cmd, batchSize, out batchMaxRowId);
         }
 
-        // Persist the (possibly advanced) mark OUTSIDE _readLock so the two locks are never
+        // Advance (and persist) the mark OUTSIDE _readLock so the two locks are never nested in
-        // nested in read→write order. Only when it moved — at most one tiny upsert per pull.
+        // read→write order, and UNDER _writeLock — the lock the purge's rowid clamp runs under.
-        var mark = Volatile.Read(ref _maxServedRowId);
+        // Advancing it under _readLock from a value sampled before the SELECT let a purge clamp
-        if (mark > before)
+        // that landed in between be overwritten with the stale-high value (review 2026-10-21
+        // follow-up L1). Taking max(current, batch max) under _writeLock sees any clamp that
+        // already happened; the served rows themselves are owed and so can never be purged.
+        // Only when it moved — at most one tiny upsert per pull.
+        lock (_writeLock)
         {
-            try
+            if (!_disposed && batchMaxRowId > Volatile.Read(ref _maxServedRowId))
             {
-                lock (_writeLock)
+                Volatile.Write(ref _maxServedRowId, batchMaxRowId);
+                try
                 {
-                    if (!_disposed)
+                    PersistMaxServedRowId(batchMaxRowId);
-                    {
+                }
-                        PersistMaxServedRowId(mark);
+                catch (Exception ex)
-                    }
+                {
+                    _logger.LogWarning(ex,
+                        "SqliteAuditWriter could not persist the served high-water mark {Mark}; a restart re-learns it from the next pull.",
+                        batchMaxRowId);
                 }
-            }
-            catch (Exception ex)
-            {
-                _logger.LogWarning(ex,
-                    "SqliteAuditWriter could not persist the served high-water mark {Mark}; a restart re-learns it from the next pull.",
-                    mark);
             }
         }
 
@@ -1184,10 +1186,11 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi
     /// rows central could possibly have received through the pull path. Bounds
     /// <see cref="MarkReconciledUpToAsync"/> so a row inserted after that point can never be
     /// retired by a cursor that happens to sit above its timestamp. Monotonic (a later, smaller
-    /// batch never lowers it) except for the purge clamp. Written under <c>_readLock</c> (the
+    /// batch never lowers it) except for the purge clamp. Every WRITE happens under
-    /// read path) and read under <c>_writeLock</c> (the flip/purge paths) — two different
+    /// <c>_writeLock</c> (the pull's advance and the purge's clamp alike — review 2026-10-21
-    /// locks, so the accesses go through <see cref="Volatile"/> for visibility. A stale read
+    /// follow-up L1), so a clamp can never be overwritten with a stale-high value; reads from
-    /// can only make the bound smaller, i.e. more conservative, never unsafe.
+    /// other paths go through <see cref="Volatile"/> for visibility. A stale read can only make
+    /// the bound smaller, i.e. more conservative, never unsafe.
     /// </summary>
     private long _maxServedRowId;
 
@@ -1236,14 +1239,15 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi
 
     /// <summary>
     /// Executes a reconciliation-pull read whose projection carries <c>fs.rowid</c> as an
-    /// 11th column, materialising the canonical rows while advancing
+    /// 11th column, materialising the canonical rows and reporting the highest rowid served.
-    /// <see cref="_maxServedRowId"/>. Serving a row is what makes it eligible for
+    /// The caller folds that into <see cref="_maxServedRowId"/> under <c>_writeLock</c> before
-    /// cursor-proved retirement, so the two must happen in the same step.
+    /// returning the rows — serving a row is what makes it eligible for cursor-proved
+    /// retirement, so the mark must cover a batch before central can see it.
     /// </summary>
-    private IReadOnlyList<AuditEvent> ReadServedRows(SqliteCommand cmd, int capacityHint)
+    private IReadOnlyList<AuditEvent> ReadServedRows(SqliteCommand cmd, int capacityHint, out long maxRowId)
     {
         var rows = new List<AuditEvent>(Math.Min(capacityHint, 256));
-        var maxRowId = Volatile.Read(ref _maxServedRowId);
+        maxRowId = 0;
         using (var reader = cmd.ExecuteReader())
         {
             while (reader.Read())
@@ -1254,7 +1258,6 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi
             }
         }
 
-        Volatile.Write(ref _maxServedRowId, maxRowId);
         return rows;
     }
 

tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterForwardStateTests.cs

@@ -165,4 +165,38 @@ public class SqliteAuditWriterForwardStateTests : IDisposable
             Assert.Equal(0, (await restarted.GetBacklogStatsAsync()).PendingCount);
         }
     }
+
+    /// <summary>
+    /// Follow-up L1: the served high-water mark is only ever MUTATED under the write lock — the
+    /// same lock the purge's clamp runs under. Before the fix the read path wrote the mark under
+    /// its own read lock from a value it sampled before the SELECT, so a purge that clamped the
+    /// mark in between was overwritten with the stale-high value, re-vouching for reusable rowids.
+    /// Holding the write lock across a pull must therefore leave the mark untouched until released.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ServedHighWaterMark_IsOnlyAdvancedUnderTheWriteLock()
+    {
+        await using var writer = CreateWriter();
+        await WriteAsync(writer, DateTime.UtcNow.AddMinutes(-5));
+        var before = writer.MaxServedRowIdForTests;
+
+        Task<IReadOnlyList<AuditEvent>> pull;
+        System.Threading.Monitor.Enter(writer.WriteLockForTests);
+        try
+        {
+            pull = Task.Run(() => writer.ReadPendingSinceAsync(DateTime.MinValue, 100));
+            // The SELECT itself needs only the read lock; give it ample time to run. A
+            // synchronous sleep (not an await) so Monitor.Exit runs on the owning thread.
+            Thread.Sleep(500);
+            Assert.Equal(before, writer.MaxServedRowIdForTests);
+        }
+        finally
+        {
+            System.Threading.Monitor.Exit(writer.WriteLockForTests);
+        }
+
+        Assert.Single(await pull.WaitAsync(TimeSpan.FromSeconds(10)));
+        Assert.True(writer.MaxServedRowIdForTests > before);
+    }
 }