fix(site-event-log): queries on own read connection off the actor thread, bounded keyword window; cap trim counts/vacuums rarely (review 2026-10-21 02/P1, 02/P4)

2026-10-22 08:30:35 -04:00
parent f7c4adbaed
commit 3169429c3f
9 changed files with 194 additions and 35 deletions
@@ -194,9 +194,9 @@ Each event entry contains:
  - Time range
  - Instance ID
  - Severity
-  - **Keyword search**: Free-text search on message and source fields (SQLite LIKE query). Useful for finding events by script name, alarm name, or error message across all instances.
+  - **Keyword search**: Free-text search on message and source fields (SQLite LIKE query). Useful for finding events by script name, alarm name, or error message across all instances. The `LIKE` is a leading-wildcard full scan, so a keyword search is **bounded to a time window**: a query with no `From` (or a range wider than `ScadaBridge:SiteEventLog:KeywordSearchMaxWindow`, default 7 days) is clamped server-side to `(To ?? now) - window`, not rejected (review 2026-10-21 02/P1).
 - Results are **paginated** with a configurable page size (default: 500 events). Each response includes a continuation token for fetching additional pages. This prevents broad queries from overwhelming the communication channel.
- The site processes the query locally against SQLite and returns matching results to central.
+- The site processes the query locally against SQLite and returns matching results to central. The query runs off the actor thread (`Task.Run` + `PipeTo`) on its own short-lived read connection — WAL readers do not block the single writer, so a slow scan can no longer hold the logger's write lock and overflow the bounded write queue (review 2026-10-21 02/P1). The storage-cap trim tracks its row count locally (re-read from the database after the first batch and every 10th batch) and runs `PRAGMA incremental_vacuum` once per tick rather than per batch (02/P4).

 ## Dependencies

@@ -20,10 +20,14 @@ public class EventLogHandlerActor : ReceiveActor
    {
        _queryService = queryService;

+        // The query is synchronous SQLite I/O (a keyword search can scan millions of
+        // rows), so run it on the thread pool and pipe the reply back rather than pinning
+        // an Akka dispatcher thread (review 2026-10-21 02/P1). ExecuteQuery never throws —
+        // it returns Success:false — so no failure mapper is needed beyond the safety net.
        Receive<EventLogQueryRequest>(msg =>
        {
-            var response = _queryService.ExecuteQuery(msg);
-            Sender.Tell(response);
+            var service = _queryService;
+            Task.Run(() => service.ExecuteQuery(msg)).PipeTo(Sender);
        });
    }
 }
@@ -58,6 +58,9 @@ public class EventLogPurgeService : BackgroundService
    /// <summary>Number of events deleted per cap-purge batch.</summary>
    private const int CapPurgeBatchSize = 1000;

+    /// <summary>Cap-trim batches between database row-count re-reads (see <see cref="PurgeByStorageCap"/>).</summary>
+    private const int CapTrimCountRefreshBatches = 10;
+
    /// <summary>Number of events deleted per retention-purge batch — same size as <see cref="CapPurgeBatchSize"/>.</summary>
    private const int RetentionPurgeBatchSize = 1000;

@@ -504,6 +507,8 @@ public class EventLogPurgeService : BackgroundService
        // deleted nothing or because concurrent recorder writes are outrunning the trim.
        var totalDeleted = 0;
        var currentSizeBytes = estimate.EstimatedBytes;
+        long rowCount = estimate.RowCount;
+        var batchesRun = 0;

        while (currentSizeBytes > capBytes)
        {
@@ -533,7 +538,47 @@ public class EventLogPurgeService : BackgroundService
                    """;
                var rows = cmd.ExecuteNonQuery();

-                // Reclaim free pages so page_count/freelist measurement reflects the delete.
+                return rows;
+            });
+
+            if (deleted == 0)
+                break;
+
+            totalDeleted += deleted;
+            budget.Record(deleted);
+            batchesRun++;
+
+            // Row count is tracked locally (decremented by what each batch deleted) and only
+            // re-read from the database after the first batch and then every
+            // CapTrimCountRefreshBatches batches — COUNT(*) on a multi-million-row table is a
+            // full index walk under the writer lock, and doing it per 1 000-row batch held that
+            // lock for most of a ceiling-sized tick (review 2026-10-21 02/P4). The re-read keeps
+            // concurrent recorder writes visible; row width stays fixed for the tick.
+            var rereadCount = batchesRun == 1 || batchesRun % CapTrimCountRefreshBatches == 0;
+            if (rereadCount)
+                rowCount = GetSiteEventsRowCount();
+            else
+                rowCount = Math.Max(0, rowCount - deleted);
+            currentSizeBytes = (long)(rowCount * estimate.BytesPerRow);
+
+            // The progress guard only has teeth on a re-read (a local decrement always shrinks).
+            if (rereadCount && currentSizeBytes >= previousSizeBytes)
+            {
+                _logger.LogWarning(
+                    "Estimated site_events size did not decrease after a cap-purge batch " +
+                    "({Size:F1} MB); stopping to avoid emptying the log. Deletes are landing, so " +
+                    "this means new events are arriving at least as fast as the trim removes them.",
+                    currentSizeBytes / (1024.0 * 1024.0));
+                break;
+            }
+        }
+
+        if (totalDeleted > 0)
+        {
+            _eventLogger.WithConnection(connection =>
+            {
+                // Run once per tick, not per batch (review 2026-10-21 02/P4), after the loop.
+                // Reclaims free pages so page_count/freelist measurement reflects the deletes.
                // This is effective ONLY on a consolidated
                // LocalDb file that was FRESH when SiteLocalDbSetup.EnsureAutoVacuumOnFreshFile
                // ran (Host/SiteLocalDbSetup.cs) — auto_vacuum can only be enabled before any
@@ -546,29 +591,7 @@ public class EventLogPurgeService : BackgroundService
                using var vacuumCmd = connection.CreateCommand();
                vacuumCmd.CommandText = "PRAGMA incremental_vacuum";
                vacuumCmd.ExecuteNonQuery();
-
-                return rows;
            });
-
-            if (deleted == 0)
-                break;
-
-            totalDeleted += deleted;
-            budget.Record(deleted);
-
-            // Row count re-read from the database, not decremented locally, so concurrent
-            // recorder writes are visible; row width stays fixed for the tick (see remarks).
-            currentSizeBytes = (long)(GetSiteEventsRowCount() * estimate.BytesPerRow);
-
-            if (currentSizeBytes >= previousSizeBytes)
-            {
-                _logger.LogWarning(
-                    "Estimated site_events size did not decrease after a cap-purge batch " +
-                    "({Size:F1} MB); stopping to avoid emptying the log. Deletes are landing, so " +
-                    "this means new events are arriving at least as fast as the trim removes them.",
-                    currentSizeBytes / (1024.0 * 1024.0));
-                break;
-            }
        }
    }

@@ -101,8 +101,7 @@ public class EventLogQueryService : IEventLogQueryService
        {
            // Clamp caller-supplied PageSize to a hard upper
            // bound so a central client sending int.MaxValue can't force the query
-            // to materialise the entire log into a single list while holding the
-            // shared write lock. Silent clamp — misconfigured clients still get a
+            // to materialise the entire log into a single list. Silent clamp — misconfigured clients still get a
            // usable response.
            var requestedSize = request.PageSize > 0 ? request.PageSize : _options.QueryPageSize;
            var pageSize = Math.Min(requestedSize, _options.MaxQueryPageSize);
@@ -129,7 +128,19 @@ public class EventLogQueryService : IEventLogQueryService
                parameters.Add(new SqliteParameter("$afterId", afterId));
            }

-            if (request.From.HasValue)
+            // A keyword search full-scans its window (leading-wildcard LIKE), so bound the
+            // window server-side: clamp From up to (To ?? now) - KeywordSearchMaxWindow.
+            // Clamped rather than rejected so callers that send no window keep working.
+            var effectiveFrom = request.From;
+            if (!string.IsNullOrWhiteSpace(request.KeywordFilter))
+            {
+                var upper = request.To ?? DateTimeOffset.UtcNow;
+                var floor = upper - _options.KeywordSearchMaxWindow;
+                if (effectiveFrom is null || effectiveFrom.Value < floor)
+                    effectiveFrom = floor;
+            }
+
+            if (effectiveFrom.HasValue)
            {
                // Timestamps are stored as ISO
                // 8601 "o" UTC strings (always +00:00). The store compares them
@@ -139,7 +150,7 @@ public class EventLogQueryService : IEventLogQueryService
                // string that sorts wrongly against the +00:00 stored values and the
                // range filter silently includes/excludes the wrong rows.
                whereClauses.Add("timestamp >= $from");
-                parameters.Add(new SqliteParameter("$from", request.From.Value.ToUniversalTime().ToString("o")));
+                parameters.Add(new SqliteParameter("$from", effectiveFrom.Value.ToUniversalTime().ToString("o")));
            }

            if (request.To.HasValue)
@@ -184,10 +195,9 @@ public class EventLogQueryService : IEventLogQueryService
                ? "WHERE " + string.Join(" AND ", whereClauses)
                : "";

-            // Run the read against the shared connection under the logger's write
-            // lock — the connection is not thread-safe and is also used by the
-            // recorder and the purge service on other threads.
-            var entries = _eventLogger.WithConnection(connection =>
+            // Run the read on its own read connection (WAL readers don't block the
+            // writer) so a slow scan never holds the logger's write lock.
+            var entries = _eventLogger.WithReadConnection(connection =>
            {
                using var cmd = connection.CreateCommand();

@@ -98,4 +98,14 @@ public class SiteEventLogOptions
    /// so the drop is observable.
    /// </summary>
    public int WriteQueueCapacity { get; set; } = 10_000;
+
+    /// <summary>
+    /// Widest time window a <em>keyword</em> search may scan; default 7 days. The keyword filter
+    /// is a leading-wildcard <c>LIKE</c> over <c>message</c>/<c>source</c> that full-scans the
+    /// window, so an unbounded search against a log at its storage cap walks millions of rows.
+    /// A keyword query with no <c>From</c> (or a wider range) is clamped server-side to
+    /// <c>[To ?? now - window, To ?? now]</c> rather than rejected, so existing callers keep
+    /// working (review 2026-10-21 02/P1).
+    /// </summary>
+    public TimeSpan KeywordSearchMaxWindow { get; set; } = TimeSpan.FromDays(7);
 }
@@ -33,6 +33,10 @@ public sealed class SiteEventLogOptionsValidator : OptionsValidatorBase<SiteEven
            $"ScadaBridge:SiteEventLog:QueryPageSize must be greater than 0 " +
            $"(was {options.QueryPageSize}).");

+        builder.RequireThat(options.KeywordSearchMaxWindow > TimeSpan.Zero,
+            $"ScadaBridge:SiteEventLog:KeywordSearchMaxWindow must be a positive duration " +
+            $"(was {options.KeywordSearchMaxWindow}); it bounds the time range a keyword search scans.");
+
        builder.RequireThat(options.WriteQueueCapacity > 0,
            $"ScadaBridge:SiteEventLog:WriteQueueCapacity must be greater than 0 " +
            $"(was {options.WriteQueueCapacity}); it bounds the background write channel.");
@@ -45,6 +45,7 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable
    internal const int WriteBatchSize = 256;

    private readonly SqliteConnection _connection;
+    private readonly ILocalDb _localDb;
    private readonly ILogger<SiteEventLogger> _logger;
    private readonly object _writeLock = new();
    private readonly Channel<PendingEvent> _writeQueue;
@@ -80,6 +81,7 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable
        ArgumentNullException.ThrowIfNull(localDb);

        _logger = logger;
+        _localDb = localDb;

        // Already open — CreateConnection returns a live, pragma-configured,
        // UDF-registered connection. Calling Open() on it again would throw.
@@ -141,6 +143,25 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable
        }
    }

+    /// <summary>
+    /// Runs <paramref name="func"/> on a short-lived, dedicated READ connection that is
+    /// NOT serialised on the write lock. <c>site_events</c> lives in a WAL database, so a
+    /// reader never blocks the single writer (and vice versa): a slow central event-log
+    /// query (full-scan keyword search at the storage cap) can no longer stall the writer
+    /// loop's <c>FlushBatch</c> and overflow the bounded write queue (review 2026-10-21
+    /// 02/P1). The same pattern <c>OperationTrackingStore.GetStatusAsync</c> uses.
+    /// </summary>
+    /// <typeparam name="T">The return type of the function.</typeparam>
+    /// <param name="func">The function to run against the read connection.</param>
+    /// <returns>The value returned by <paramref name="func"/>.</returns>
+    internal T WithReadConnection<T>(Func<SqliteConnection, T> func)
+    {
+        ArgumentNullException.ThrowIfNull(func);
+        ObjectDisposedException.ThrowIf(_disposed, this);
+        using var readConnection = _localDb.CreateConnection();
+        return func(readConnection);
+    }
+
    /// <summary>
    /// Runs <paramref name="func"/> against the shared connection while holding the
    /// write lock. Throws <see cref="ObjectDisposedException"/> if the logger has
@@ -124,6 +124,61 @@ public class EventLogQueryServiceTests : IDisposable
        Assert.All(response.Entries, e => Assert.Equal("inst-1", e.InstanceId));
    }

+    private void InsertRawEvent(DateTimeOffset ts, string message) =>
+        _eventLogger.WithConnection(c =>
+        {
+            using var cmd = c.CreateCommand();
+            cmd.CommandText = @"INSERT INTO site_events (id, timestamp, event_type, severity, instance_id, source, message, details)
+                                VALUES ($id, $ts, 'script', 'Info', NULL, 'Src', $msg, NULL)";
+            cmd.Parameters.AddWithValue("$id", Guid.NewGuid().ToString("N"));
+            cmd.Parameters.AddWithValue("$ts", ts.ToUniversalTime().ToString("o"));
+            cmd.Parameters.AddWithValue("$msg", message);
+            cmd.ExecuteNonQuery();
+        });
+
+    /// <summary>
+    /// Review 2026-10-21 02/P1: a keyword search with no window is clamped to
+    /// <c>KeywordSearchMaxWindow</c> (default 7 days) instead of full-scanning the log.
+    /// </summary>
+    [Fact]
+    public void Query_KeywordSearch_WithoutWindow_IsClampedToMaxWindow()
+    {
+        InsertRawEvent(DateTimeOffset.UtcNow.AddDays(-30), "ancient needle");
+        InsertRawEvent(DateTimeOffset.UtcNow.AddDays(-1), "recent needle");
+
+        var keyword = _queryService.ExecuteQuery(MakeRequest(keyword: "needle"));
+        Assert.True(keyword.Success);
+        Assert.Single(keyword.Entries);
+        Assert.Equal("recent needle", keyword.Entries[0].Message);
+
+        // Non-keyword queries are unaffected.
+        var all = _queryService.ExecuteQuery(MakeRequest());
+        Assert.Equal(2, all.Entries.Count);
+    }
+
+    /// <summary>Review 2026-10-21 02/P1: a query must complete while another thread holds the writer lock.</summary>
+    [Fact]
+    public async Task Query_DoesNotBlockOnTheWriteLock()
+    {
+        await SeedEvents();
+        var held = new ManualResetEventSlim();
+        var release = new ManualResetEventSlim();
+        var holder = Task.Run(() => _eventLogger.WithConnection(_ =>
+        {
+            held.Set();
+            release.Wait(TimeSpan.FromSeconds(30)); // blocks the holder thread only
+        }));
+        try
+        {
+            Assert.True(await Task.Run(() => held.Wait(TimeSpan.FromSeconds(5))));
+            var q = Task.Run(() => _queryService.ExecuteQuery(MakeRequest()));
+            var done = await Task.WhenAny(q, Task.Delay(TimeSpan.FromSeconds(10)));
+            Assert.Same(q, done);
+            Assert.Equal(5, (await q).Entries.Count);
+        }
+        finally { release.Set(); await holder; }
+    }
+
    /// <summary>Verifies that a keyword search matches the event message.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
@@ -363,6 +363,38 @@ public class EventLogStorageCapScopeTests : IDisposable
            e.Level == LogLevel.Warning && e.Message.Contains("did not decrease", StringComparison.Ordinal));
    }

+    private sealed class CountingRowCountPurgeService(
+        SiteEventLogger eventLogger,
+        IOptions<SiteEventLogOptions> options,
+        ILogger<EventLogPurgeService> logger)
+        : EventLogPurgeService(eventLogger, options, logger)
+    {
+        /// <summary>Gets how many times the row count was read from the database.</summary>
+        public int RowCountReads { get; private set; }
+
+        /// <inheritdoc />
+        internal override long GetSiteEventsRowCount()
+        {
+            RowCountReads++;
+            return base.GetSiteEventsRowCount();
+        }
+    }
+
+    /// <summary>Review 2026-10-21 02/P4: the cap trim must not COUNT(*) after every batch.</summary>
+    [Fact]
+    public void PurgeByStorageCap_RereadsRowCountRarely_NotPerBatch()
+    {
+        SeedEvents(12_000); // 12 batches of 1 000 at cap 0 MB
+
+        var purge = new CountingRowCountPurgeService(
+            _eventLogger, Options.Create(CapOptions(0)), new CapturingLogger());
+
+        purge.RunPurge();
+
+        Assert.Equal(0, EventCount());
+        Assert.True(purge.RowCountReads <= 3, $"Expected ≤3 COUNT(*) reads for 12 batches, saw {purge.RowCountReads}.");
+    }
+
    private sealed record LogEntry(LogLevel Level, string Message);

    private sealed class CapturingLogger : ILogger<EventLogPurgeService>