Fix: SQLite lock contention causing db read timeout.

ThingConnect.Pulse.Server/Infrastructure/SqliteWalInterceptor.cs

@@ -0,0 +1,51 @@
+using Microsoft.EntityFrameworkCore.Diagnostics;
+using System.Data.Common;
+
+namespace ThingConnect.Pulse.Server.Infrastructure;
+
+/// <summary>
+/// Enables WAL mode and busy_timeout on every SQLite connection to allow concurrent reads
+/// during high-frequency writes from the monitoring background service.
+///
+/// Without WAL, SQLite's default DELETE journal mode holds an exclusive lock for the
+/// duration of each write, causing read timeouts when probes are writing continuously.
+/// WAL mode decouples readers from writers entirely.
+/// </summary>
+public sealed class SqliteWalInterceptor : DbConnectionInterceptor
+{
+    public override void ConnectionOpened(DbConnection connection, ConnectionEndEventData eventData)
+    {
+        ApplyPragmas(connection);
+    }
+
+    public override async Task ConnectionOpenedAsync(
+        DbConnection connection,
+        ConnectionEndEventData eventData,
+        CancellationToken cancellationToken = default)
+    {
+        await ApplyPragmasAsync(connection, cancellationToken);
+    }
+
+    [System.Diagnostics.CodeAnalysis.SuppressMessage("Security", "CA2100", Justification = "Pragma SQL is a compile-time constant, not user input.")]
+    private static void ApplyPragmas(DbConnection connection)
+    {
+        using DbCommand cmd = connection.CreateCommand();
+        cmd.CommandText = BuildPragmaSql();
+        cmd.ExecuteNonQuery();
+    }
+
+    [System.Diagnostics.CodeAnalysis.SuppressMessage("Security", "CA2100", Justification = "Pragma SQL is a compile-time constant, not user input.")]
+    private static async Task ApplyPragmasAsync(DbConnection connection, CancellationToken ct)
+    {
+        await using DbCommand cmd = connection.CreateCommand();
+        cmd.CommandText = BuildPragmaSql();
+        await cmd.ExecuteNonQueryAsync(ct);
+    }
+
+    private static string BuildPragmaSql() => """
+        PRAGMA journal_mode=WAL;
+        PRAGMA busy_timeout=5000;
+        PRAGMA synchronous=NORMAL;
+        PRAGMA cache_size=-8000;
+        """;
+}

ThingConnect.Pulse.Server/Program.cs

@@ -48,7 +48,8 @@ public static async Task Main(string[] args)
 
             // Add services to the container.
             builder.Services.AddDbContext<PulseDbContext>(options =>
-                options.UseSqlite(builder.Configuration.GetConnectionString("DefaultConnection")));
+                options.UseSqlite(builder.Configuration.GetConnectionString("DefaultConnection"))
+                       .AddInterceptors(new SqliteWalInterceptor()));
 
             // Configure Identity and Authentication
             builder.Services.AddIdentity<ApplicationUser, IdentityRole>(options =>
@@ -86,7 +87,7 @@ public static async Task Main(string[] args)
                 options.Cookie.HttpOnly = true;
                 options.Cookie.SecurePolicy = builder.Environment.IsDevelopment()
                     ? CookieSecurePolicy.SameAsRequest
-                    : CookieSecurePolicy.Always;
+                    : CookieSecurePolicy.None;
                 options.Cookie.SameSite = SameSiteMode.Lax;
                 options.Cookie.Name = "ThingConnect.Pulse.Auth";
                 options.Events.OnRedirectToLogin = context =>
@@ -150,6 +151,9 @@ public static async Task Main(string[] args)
             builder.Services.AddSingleton<ISettingsService, SettingsService>();
 
             // Add monitoring services
+            builder.Services.AddSingleton<CheckResultWriteQueue>();
+            builder.Services.AddSingleton<ICheckResultWriteQueue>(p => p.GetRequiredService<CheckResultWriteQueue>());
+            builder.Services.AddHostedService<CheckResultWriteQueue>(p => p.GetRequiredService<CheckResultWriteQueue>());
             builder.Services.AddScoped<IProbeService, ProbeService>();
             builder.Services.AddSingleton<IOutageDetectionService, OutageDetectionService>();
             builder.Services.AddSingleton<IDiscoveryService, DiscoveryService>();
@@ -164,6 +168,7 @@ public static async Task Main(string[] args)
 
             // Add prune services
             builder.Services.AddScoped<IPruneService, PruneService>();
+            builder.Services.AddHostedService<PruneBackgroundService>();
 
             // Add log cleanup service
             builder.Services.AddHostedService<LogCleanupBackgroundService>();

ThingConnect.Pulse.Server/Services/Monitoring/CheckResultWriteQueue.cs

@@ -0,0 +1,122 @@
+using System.Threading.Channels;
+using Microsoft.EntityFrameworkCore;
+using ThingConnect.Pulse.Server.Data;
+
+namespace ThingConnect.Pulse.Server.Services.Monitoring;
+
+public interface ICheckResultWriteQueue
+{
+    void Enqueue(WriteQueueItem item);
+}
+
+public sealed record WriteQueueItem(
+    CheckResultRaw RawResult,
+    Guid? EndpointIdForRttUpdate,
+    double? RttMs);
+
+/// <summary>
+/// Serializes all check_result_raw inserts through a single background writer, eliminating
+/// the concurrent SaveChangesAsync flood that causes SQLite read timeouts under high probe rates.
+/// </summary>
+public sealed class CheckResultWriteQueue : BackgroundService, ICheckResultWriteQueue
+{
+    private readonly Channel<WriteQueueItem> _channel = Channel.CreateBounded<WriteQueueItem>(
+        new BoundedChannelOptions(10_000) { FullMode = BoundedChannelFullMode.DropOldest });
+
+    private readonly IServiceProvider _serviceProvider;
+    private readonly ILogger<CheckResultWriteQueue> _logger;
+
+    public CheckResultWriteQueue(IServiceProvider serviceProvider, ILogger<CheckResultWriteQueue> logger)
+    {
+        _serviceProvider = serviceProvider;
+        _logger = logger;
+    }
+
+    public void Enqueue(WriteQueueItem item) => _channel.Writer.TryWrite(item);
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        _logger.LogInformation("Check result write queue started");
+
+        var batch = new List<WriteQueueItem>(100);
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            batch.Clear();
+
+            try
+            {
+                // Block until at least one item is available
+                if (!await _channel.Reader.WaitToReadAsync(stoppingToken))
+                {
+                    break;
+                }
+
+                // Drain all ready items up to batch limit (no extra waiting)
+                while (batch.Count < 100 && _channel.Reader.TryRead(out WriteQueueItem? item))
+                {
+                    batch.Add(item);
+                }
+
+                await FlushBatchAsync(batch, stoppingToken);
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error flushing check result batch of {Count} items", batch.Count);
+                // Brief pause to avoid tight error loop, then continue
+                await Task.Delay(1000, stoppingToken).ConfigureAwait(false);
+            }
+        }
+
+        // Drain remaining items on shutdown
+        batch.Clear();
+        while (_channel.Reader.TryRead(out WriteQueueItem? item))
+        {
+            batch.Add(item);
+        }
+
+        if (batch.Count > 0)
+        {
+            try
+            {
+                await FlushBatchAsync(batch, CancellationToken.None);
+                _logger.LogInformation("Flushed {Count} remaining check results on shutdown", batch.Count);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Failed to flush remaining {Count} check results on shutdown", batch.Count);
+            }
+        }
+
+        _logger.LogInformation("Check result write queue stopped");
+    }
+
+    private async Task FlushBatchAsync(List<WriteQueueItem> batch, CancellationToken ct)
+    {
+        using IServiceScope scope = _serviceProvider.CreateScope();
+        PulseDbContext context = scope.ServiceProvider.GetRequiredService<PulseDbContext>();
+
+        // Bulk insert all check results in one round-trip
+        context.CheckResultsRaw.AddRange(batch.Select(i => i.RawResult));
+        await context.SaveChangesAsync(ct);
+
+        // Update LastRttMs with direct UPDATE statements — no entity load needed
+        var rttUpdates = batch
+            .Where(i => i.EndpointIdForRttUpdate.HasValue && i.RttMs.HasValue)
+            .GroupBy(i => i.EndpointIdForRttUpdate!.Value)
+            .Select(g => (EndpointId: g.Key, RttMs: g.Last().RttMs!.Value));
+
+        foreach ((Guid endpointId, double rttMs) in rttUpdates)
+        {
+            await context.Endpoints
+                .Where(e => e.Id == endpointId)
+                .ExecuteUpdateAsync(s => s.SetProperty(e => e.LastRttMs, rttMs), ct);
+        }
+
+        _logger.LogDebug("Flushed batch of {Count} check results", batch.Count);
+    }
+}

ThingConnect.Pulse.Server/Services/Monitoring/DiscoveryService.cs

@@ -1,3 +1,4 @@
+using System.Linq;
 using System.Net;
 using System.Net.Sockets;
 using System.Text.RegularExpressions;
@@ -47,7 +48,7 @@ public IEnumerable<string> ExpandCidr(string cidr)
         }
 
         byte[] addressBytes = ipAddress.GetAddressBytes();
-        uint addressInt = BitConverter.ToUInt32(addressBytes.Reverse().ToArray(), 0);
+        uint addressInt = BitConverter.ToUInt32(addressBytes, 0);
 
         int hostBits = 32 - prefixLength;
         uint hostCount = (uint)(1 << hostBits);
@@ -59,7 +60,8 @@ public IEnumerable<string> ExpandCidr(string cidr)
 
         for (uint address = startAddress; address < endAddress && address > networkAddress; address++)
         {
-            byte[] bytes = BitConverter.GetBytes(address).Reverse().ToArray();
+            byte[] bytes = BitConverter.GetBytes(address);
+            System.Array.Reverse(bytes);
             var ip = new IPAddress(bytes);
             yield return ip.ToString();
         }

ThingConnect.Pulse.Server/Services/Monitoring/MonitoringBackgroundService.cs

@@ -15,6 +15,7 @@ public sealed class MonitoringBackgroundService : BackgroundService
     private readonly SemaphoreSlim _concurrencySemaphore;
     private readonly ConcurrentDictionary<Guid, Timer> _endpointTimers = new();
     private readonly ConcurrentDictionary<Guid, bool> _probeExecuting = new();
+    private readonly ConcurrentDictionary<Guid, Data.Endpoint> _endpointCache = new();
     private readonly int _maxConcurrentProbes;
 
     public MonitoringBackgroundService(IServiceProvider serviceProvider,
@@ -134,6 +135,12 @@ private async Task RefreshEndpointsAsync(CancellationToken cancellationToken)
             .Where(e => e.Enabled)
             .ToListAsync(cancellationToken);
 
+        // Refresh in-memory cache so probes don't need to re-read endpoint data from DB
+        foreach (Data.Endpoint endpoint in endpoints)
+        {
+            _endpointCache[endpoint.Id] = endpoint;
+        }
+
         var currentEndpointIds = endpoints.Select(e => e.Id).ToHashSet();
         var existingEndpointIds = _endpointTimers.Keys.ToHashSet();
 
@@ -144,7 +151,8 @@ private async Task RefreshEndpointsAsync(CancellationToken cancellationToken)
             if (_endpointTimers.TryRemove(endpointId, out Timer? timer))
             {
                 await StopTimerGracefullyAsync(timer, endpointId);
-                _probeExecuting.TryRemove(endpointId, out _); // Clean up execution tracking
+                _probeExecuting.TryRemove(endpointId, out _);
+                _endpointCache.TryRemove(endpointId, out _);
                 _logger.LogInformation("Stopped monitoring endpoint: {EndpointId}", endpointId);
             }
         }
@@ -200,18 +208,16 @@ private async Task ProbeEndpointAsync(Guid endpointId)
 
         try
         {
+            // Use cached endpoint data — RefreshEndpointsAsync keeps this up to date every 15s
+            if (!_endpointCache.TryGetValue(endpointId, out Data.Endpoint? endpoint) || !endpoint.Enabled)
+            {
+                return;
+            }
+
             using IServiceScope scope = _serviceProvider.CreateScope();
-            PulseDbContext context = scope.ServiceProvider.GetRequiredService<PulseDbContext>();
             IProbeService probeService = scope.ServiceProvider.GetRequiredService<IProbeService>();
             IOutageDetectionService outageService = scope.ServiceProvider.GetRequiredService<IOutageDetectionService>();
 
-            // Get endpoint details
-            Data.Endpoint? endpoint = await context.Endpoints.FindAsync(endpointId);
-            if (endpoint == null || !endpoint.Enabled)
-            {
-                return; // Endpoint was deleted or disabled
-            }
-
             // Perform the probe
             Models.CheckResult result = await probeService.ProbeAsync(endpoint);
 

ThingConnect.Pulse.Server/Services/Monitoring/OutageDetectionService.cs

@@ -14,12 +14,14 @@ public sealed class OutageDetectionService : IOutageDetectionService
 {
     private readonly IServiceProvider _serviceProvider;
     private readonly ILogger<OutageDetectionService> _logger;
+    private readonly ICheckResultWriteQueue _writeQueue;
     private readonly ConcurrentDictionary<Guid, MonitorState> _states = new();
 
-    public OutageDetectionService(IServiceProvider serviceProvider, ILogger<OutageDetectionService> logger)
+    public OutageDetectionService(IServiceProvider serviceProvider, ILogger<OutageDetectionService> logger, ICheckResultWriteQueue writeQueue)
     {
         _serviceProvider = serviceProvider;
         _logger = logger;
+        _writeQueue = writeQueue;
     }
 
     public async Task<bool> ProcessCheckResultAsync(CheckResult result, CancellationToken cancellationToken = default)
@@ -69,8 +71,8 @@ public async Task<bool> ProcessCheckResultAsync(CheckResult result, Cancellation
                     result.EndpointId, state.SuccessStreak);
             }
 
-            // Save the raw check result to database (independent of transitions)
-            await SaveCheckResultAsync(result, cancellationToken);
+            // Enqueue raw check result — written by the background write queue (non-blocking)
+            SaveCheckResult(result);
 
             return stateChanged;
         }
@@ -515,11 +517,8 @@ private async Task UpdateEndpointStatusAsync(PulseDbContext context, Guid endpoi
         return (endpointStatus, openOutageId, false);
     }
 
-    private async Task SaveCheckResultAsync(CheckResult result, CancellationToken cancellationToken)
+    private void SaveCheckResult(CheckResult result)
     {
-        using IServiceScope scope = _serviceProvider.CreateScope();
-        PulseDbContext context = scope.ServiceProvider.GetRequiredService<PulseDbContext>();
-
         CheckResultRaw rawResult = new CheckResultRaw
         {
             EndpointId = result.EndpointId,
@@ -529,21 +528,10 @@ private async Task SaveCheckResultAsync(CheckResult result, CancellationToken ca
             Error = result.Error
         };
 
-        context.CheckResultsRaw.Add(rawResult);
+        Guid? endpointIdForRtt = result.Status == UpDown.up && result.RttMs.HasValue
+            ? result.EndpointId
+            : null;
 
-        // Update LastRttMs for successful probes
-        if (result.Status == UpDown.up && result.RttMs.HasValue)
-        {
-            Data.Endpoint? endpoint = await context.Endpoints.FindAsync([result.EndpointId], cancellationToken);
-            if (endpoint != null)
-            {
-                endpoint.LastRttMs = result.RttMs.Value;
-                _logger.LogInformation(
-                    "Updated LastRttMs for endpoint {EndpointId} to {RttMs}ms",
-                    result.EndpointId, result.RttMs.Value);
-            }
-        }
-
-        await context.SaveChangesAsync(cancellationToken);
+        _writeQueue.Enqueue(new WriteQueueItem(rawResult, endpointIdForRtt, result.RttMs));
     }
 }