From 530a096111fdf63e6e9bc57f7b50a1d7343be935 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 09:44:38 -0400 Subject: [PATCH] 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 | 13 +++- .../Site/SqliteAuditWriter.cs | 59 ++++++++++--------- .../SqliteAuditWriterForwardStateTests.cs | 34 +++++++++++ 3 files changed, 77 insertions(+), 29 deletions(-) diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index 19ff1e8b..eea019aa 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/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` / diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs index 9cdac6d8..da18166f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs +++ b/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 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 - // nested in read→write order. Only when it moved — at most one tiny upsert per pull. - var mark = Volatile.Read(ref _maxServedRowId); - if (mark > before) + // Advance (and persist) the mark OUTSIDE _readLock so the two locks are never nested in + // read→write order, and UNDER _writeLock — the lock the purge's rowid clamp runs under. + // Advancing it under _readLock from a value sampled before the SELECT let a purge clamp + // 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(mark); - } + PersistMaxServedRowId(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.", + 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 /// 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 _readLock (the - /// read path) and read under _writeLock (the flip/purge paths) — two different - /// locks, so the accesses go through for visibility. A stale read - /// can only make the bound smaller, i.e. more conservative, never unsafe. + /// batch never lowers it) except for the purge clamp. Every WRITE happens under + /// _writeLock (the pull's advance and the purge's clamp alike — review 2026-10-21 + /// follow-up L1), so a clamp can never be overwritten with a stale-high value; reads from + /// other paths go through for visibility. A stale read can only make + /// the bound smaller, i.e. more conservative, never unsafe. /// private long _maxServedRowId; @@ -1236,14 +1239,15 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi /// /// Executes a reconciliation-pull read whose projection carries fs.rowid as an - /// 11th column, materialising the canonical rows while advancing - /// . Serving a row is what makes it eligible for - /// cursor-proved retirement, so the two must happen in the same step. + /// 11th column, materialising the canonical rows and reporting the highest rowid served. + /// The caller folds that into under _writeLock before + /// 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. /// - private IReadOnlyList ReadServedRows(SqliteCommand cmd, int capacityHint) + private IReadOnlyList ReadServedRows(SqliteCommand cmd, int capacityHint, out long maxRowId) { var rows = new List(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; } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterForwardStateTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterForwardStateTests.cs index bbeaa2c1..e07a4852 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterForwardStateTests.cs +++ b/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); } } + + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ServedHighWaterMark_IsOnlyAdvancedUnderTheWriteLock() + { + await using var writer = CreateWriter(); + await WriteAsync(writer, DateTime.UtcNow.AddMinutes(-5)); + var before = writer.MaxServedRowIdForTests; + + Task> 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); + } }