Fix 2

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/HistoryService.cs

@@ -84,14 +84,9 @@ private async Task<List<RawCheckDto>> GetRawDataAsync(Guid endpointId, DateTimeO
         long fromUnix = UnixTimestamp.ToUnixSeconds(from);
         long toUnix = UnixTimestamp.ToUnixSeconds(to);
 
-        // SQLite limitation: fetch all data and filter in memory
-        var rawData = await _context.CheckResultsRaw
-            .Where(c => c.EndpointId == endpointId)
-            .Select(c => new { c.Ts, c.Status, c.RttMs, c.Error })
-            .ToListAsync();
-
-        return rawData
-            .Where(c => c.Ts >= fromUnix && c.Ts <= toUnix)
+        return await _context.CheckResultsRaw
+            .Where(c => c.EndpointId == endpointId && c.Ts >= fromUnix && c.Ts <= toUnix)
+            .AsNoTracking()
             .OrderBy(c => c.Ts)
             .Select(c => new RawCheckDto
             {
@@ -100,22 +95,17 @@ private async Task<List<RawCheckDto>> GetRawDataAsync(Guid endpointId, DateTimeO
                 RttMs = c.RttMs,
                 Error = c.Error
             })
-            .ToList();
+            .ToListAsync();
     }
 
     private async Task<List<RollupBucketDto>> GetRollup15mDataAsync(Guid endpointId, DateTimeOffset from, DateTimeOffset to)
     {
         long fromUnix = UnixTimestamp.ToUnixSeconds(from);
         long toUnix = UnixTimestamp.ToUnixSeconds(to);
 
-        // SQLite limitation: fetch all data and filter in memory
-        var rollupData = await _context.Rollups15m
-            .Where(r => r.EndpointId == endpointId)
-            .Select(r => new { r.BucketTs, r.UpPct, r.AvgRttMs, r.DownEvents })
-            .ToListAsync();
-
-        return rollupData
-            .Where(r => r.BucketTs >= fromUnix && r.BucketTs <= toUnix)
+        return await _context.Rollups15m
+            .Where(r => r.EndpointId == endpointId && r.BucketTs >= fromUnix && r.BucketTs <= toUnix)
+            .AsNoTracking()
             .OrderBy(r => r.BucketTs)
             .Select(r => new RollupBucketDto
             {
@@ -124,7 +114,7 @@ private async Task<List<RollupBucketDto>> GetRollup15mDataAsync(Guid endpointId,
                 AvgRttMs = r.AvgRttMs,
                 DownEvents = r.DownEvents
             })
-            .ToList();
+            .ToListAsync();
     }
 
     private async Task<List<DailyBucketDto>> GetRollupDailyDataAsync(Guid endpointId, DateTimeOffset from, DateTimeOffset to)
@@ -133,14 +123,9 @@ private async Task<List<DailyBucketDto>> GetRollupDailyDataAsync(Guid endpointId
         long fromUnix = UnixTimestamp.ToUnixDate(DateOnly.FromDateTime(from.Date));
         long toUnix = UnixTimestamp.ToUnixDate(DateOnly.FromDateTime(to.Date));
 
-        // SQLite limitation: fetch all data and filter in memory
-        var dailyData = await _context.RollupsDaily
-            .Where(r => r.EndpointId == endpointId)
-            .Select(r => new { r.BucketDate, r.UpPct, r.AvgRttMs, r.DownEvents })
-            .ToListAsync();
-
-        return dailyData
-            .Where(r => r.BucketDate >= fromUnix && r.BucketDate <= toUnix)
+        return await _context.RollupsDaily
+            .Where(r => r.EndpointId == endpointId && r.BucketDate >= fromUnix && r.BucketDate <= toUnix)
+            .AsNoTracking()
             .OrderBy(r => r.BucketDate)
             .Select(r => new DailyBucketDto
             {
@@ -149,22 +134,17 @@ private async Task<List<DailyBucketDto>> GetRollupDailyDataAsync(Guid endpointId
                 AvgRttMs = r.AvgRttMs,
                 DownEvents = r.DownEvents
             })
-            .ToList();
+            .ToListAsync();
     }
 
     private async Task<List<OutageDto>> GetOutagesAsync(Guid endpointId, DateTimeOffset from, DateTimeOffset to)
     {
         long fromUnix = UnixTimestamp.ToUnixSeconds(from);
         long toUnix = UnixTimestamp.ToUnixSeconds(to);
 
-        // SQLite limitation: fetch all data and filter in memory
-        var outageData = await _context.Outages
-            .Where(o => o.EndpointId == endpointId)
-            .Select(o => new { o.StartedTs, o.EndedTs, o.DurationSeconds, o.LastError })
-            .ToListAsync();
-
-        return outageData
-            .Where(o => o.StartedTs <= toUnix && (o.EndedTs == null || o.EndedTs >= fromUnix))
+        return await _context.Outages
+            .Where(o => o.EndpointId == endpointId && o.StartedTs <= toUnix && (o.EndedTs == null || o.EndedTs >= fromUnix))
+            .AsNoTracking()
             .OrderBy(o => o.StartedTs)
             .Select(o => new OutageDto
             {
@@ -173,7 +153,7 @@ private async Task<List<OutageDto>> GetOutagesAsync(Guid endpointId, DateTimeOff
                 DurationS = o.DurationSeconds,
                 LastError = o.LastError
             })
-            .ToList();
+            .ToListAsync();
     }
 
     private EndpointDto MapToEndpointDto(Data.Endpoint endpoint)

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();
         }