Enhance log processing with scalability improvements

- Added a CircuitBreaker to handle API failures and prevent excessive retries.
- Implemented StateService for persistent state tracking across restarts.
- Introduced FingerprintBackfillService to backfill SHA256-based fingerprints.
- Added LogRetentionService for periodic cleanup of old logs and summaries.
- Replaced offset-based pagination with cursor-based pagination for efficiency.
- Enhanced SignalR with per-connection filtering for real-time log delivery.
- Added bounded memory queue to prevent OOM during sustained load.
- Enabled auto-creation of missing database indexes in the background.
- Introduced SHA256-based log fingerprinting for deduplication and validation.
- Added periodic cleanup of stale entries to prevent memory leaks.
- Moved hard-coded values to appsettings.json for better configurability.
- Wrapped hourly log summary updates in transactions for atomicity.
- Added `/metrics` endpoints for monitoring API and SignalR performance.
- Improved log validation to reject invalid or excessively old logs early.
- Optimized log cache with fingerprint-based deduplication.
- Enhanced logging and debugging for state persistence and cleanup processes.
- Updated `todo.md` with completed tasks and future improvements.

Author: Timothy Dodd

.claude/settings.local.json

@@ -6,8 +6,9 @@
       "Bash(timeout 30 dotnet test src/RoboDodd.OrmLite.Tests/RoboDodd.OrmLite.Tests.csproj --filter \"FullyQualifiedName=RoboDodd.OrmLite.Tests.ServiceStackMigrationIntegrationTests.ErrorHandling_WithInvalidOperations_HandlesGracefully\" --verbosity normal)",
       "Bash(timeout:*)",
       "Bash(npm run build:*)",
-      "Bash(npx ng build:*)"
+      "Bash(npx ng build:*)",
+      "Bash(xargs ls:*)"
     ],
     "deny": []
   }
-}
+}

src/LogMkAgent/Common/LogLineExtensions.cs

@@ -1,4 +1,6 @@
 using System.Collections.Concurrent;
+using System.Security.Cryptography;
+using System.Text;
 using LogMkCommon;
 
 namespace LogMkAgent.Common;
@@ -7,16 +9,21 @@ public static class LogLineExtensions
     // Thread-safe sequence counters per pod
     private static readonly ConcurrentDictionary<string, long> PodSequenceCounters = new();
 
+    // Track last activity time per pod key for cleanup
+    private static readonly ConcurrentDictionary<string, DateTime> PodLastSeen = new();
+
     public static void AssignSequenceNumber(this LogLine logLine)
     {
         var podKey = $"{logLine.DeploymentName}:{logLine.PodName}";
         logLine.SequenceNumber = PodSequenceCounters.AddOrUpdate(podKey, 1, (key, current) => current + 1);
+        PodLastSeen[podKey] = DateTime.UtcNow;
     }
 
     // Alternative method that returns the sequence number
     public static long GetNextSequenceNumber(string deploymentName, string podName)
     {
         var podKey = $"{deploymentName}:{podName}";
+        PodLastSeen[podKey] = DateTime.UtcNow;
         return PodSequenceCounters.AddOrUpdate(podKey, 1, (key, current) => current + 1);
     }
 
@@ -32,6 +39,7 @@ public static void ResetSequenceNumber(string deploymentName, string podName)
     {
         var podKey = $"{deploymentName}:{podName}";
         PodSequenceCounters.TryRemove(podKey, out _);
+        PodLastSeen.TryRemove(podKey, out _);
     }
 
     // Get current sequence without incrementing
@@ -40,6 +48,46 @@ public static long GetCurrentSequenceNumber(string deploymentName, string podNam
         var podKey = $"{deploymentName}:{podName}";
         return PodSequenceCounters.TryGetValue(podKey, out var current) ? current : 0;
     }
-}
 
-// Up
+    /// <summary>
+    /// Removes sequence counter entries for pods not seen within the given timespan.
+    /// Call periodically to prevent unbounded dictionary growth.
+    /// </summary>
+    public static int CleanupStaleEntries(TimeSpan maxAge)
+    {
+        var cutoff = DateTime.UtcNow - maxAge;
+        var removed = 0;
+
+        foreach (var kvp in PodLastSeen)
+        {
+            if (kvp.Value < cutoff)
+            {
+                if (PodLastSeen.TryRemove(kvp.Key, out _))
+                {
+                    PodSequenceCounters.TryRemove(kvp.Key, out _);
+                    removed++;
+                }
+            }
+        }
+
+        return removed;
+    }
+
+    public static int TrackedPodCount => PodSequenceCounters.Count;
+
+    /// <summary>
+    /// Computes a content-based SHA256 fingerprint of the log line content.
+    /// Truncated to 16 hex characters for compactness.
+    /// </summary>
+    public static void AssignFingerprint(this LogLine logLine)
+    {
+        if (string.IsNullOrEmpty(logLine.Line))
+        {
+            logLine.Fingerprint = string.Empty;
+            return;
+        }
+
+        var hashBytes = SHA256.HashData(Encoding.UTF8.GetBytes(logLine.Line));
+        logLine.Fingerprint = Convert.ToHexString(hashBytes, 0, 8).ToLowerInvariant();
+    }
+}

src/LogMkAgent/Program.cs

@@ -29,8 +29,10 @@ public static async Task Main(string[] args)
                 // Configure strongly typed settings
                 services.Configure<LogWatcherOptions>(configuration.GetSection("LogWatcherOptions"));
                 services.Configure<ApiSettings>(configuration.GetSection("LoggingApi"));
+                services.Configure<BatchingOptions>(configuration.GetSection("BatchingOptions"));
 
                 // Register services
+                services.AddSingleton<StateService>();
                 services.AddSingleton<BatchingService>();
                 services.AddSingleton<HttpLogger>();
                 services.AddSingleton<SettingsService>();

src/LogMkAgent/Services/BatchService.cs

@@ -28,11 +28,21 @@ public class BatchingService : IDisposable
         TimeSpan.FromSeconds(15)
     };
 
+    // Circuit breaker
+    private readonly CircuitBreaker _circuitBreaker;
+
+    // Rate limiting
+    private readonly int _maxBatchesPerMinute;
+    private readonly Queue<DateTime> _batchSendTimestamps = new();
+    private readonly object _rateLimitLock = new();
+
     // Metrics for monitoring
     private long _totalItemsProcessed;
     private long _totalBatchesSent;
     private long _totalFailures;
     private long _totalValidationFailures;
+    private long _totalRateLimitDelays;
+    private long _totalDropped;
     private DateTime _lastSuccessfulSend = DateTime.UtcNow;
 
     // Debounce configuration
@@ -53,6 +63,11 @@ public BatchingService(
         _options = batchingOptions?.Value ?? throw new ArgumentNullException(nameof(batchingOptions));
         _settingsService = settingsService ?? throw new ArgumentNullException(nameof(settingsService));
 
+        _maxBatchesPerMinute = _options.MaxBatchesPerMinute;
+        _circuitBreaker = new CircuitBreaker(logger,
+            failureThreshold: _options.CircuitBreakerFailureThreshold,
+            cooldownPeriod: _options.CircuitBreakerCooldown);
+
         ValidateOptions();
 
         // Configure debounce delay from options if available, otherwise use defaults
@@ -62,8 +77,8 @@ public BatchingService(
             _maxWaitTime = _options.BatchInterval;
         }
 
-        _logger.LogInformation("BatchingService initialized with debounce: {DebounceDelay}ms, max wait: {MaxWait}s, max size: {MaxSize}, timeout: {Timeout}",
-            _debounceDelay.TotalMilliseconds, _maxWaitTime.TotalSeconds, _options.MaxBatchSize, _options.SendTimeout);
+        _logger.LogInformation("BatchingService initialized with debounce: {DebounceDelay}ms, max wait: {MaxWait}s, max size: {MaxSize}, timeout: {Timeout}, rate limit: {RateLimit}/min, max queue: {MaxQueue}, max retry queue: {MaxRetryQueue}",
+            _debounceDelay.TotalMilliseconds, _maxWaitTime.TotalSeconds, _options.MaxBatchSize, _options.SendTimeout, _maxBatchesPerMinute, _options.MaxQueueSize, _options.MaxRetryQueueSize);
     }
 
     private void ValidateOptions()
@@ -92,6 +107,18 @@ public void AddData(LogLine data)
             return;
         }
 
+        // Enforce queue size limit - drop oldest items if at capacity
+        if (_options.MaxQueueSize > 0 && _batchData.Count >= _options.MaxQueueSize)
+        {
+            var dropped = 0;
+            while (_batchData.Count >= _options.MaxQueueSize && _batchData.TryDequeue(out _))
+            {
+                dropped++;
+            }
+            Interlocked.Add(ref _totalDropped, dropped);
+            _logger.LogWarning("Queue at capacity ({MaxSize}), dropped {Dropped} oldest items", _options.MaxQueueSize, dropped);
+        }
+
         _batchData.Enqueue(data);
 
         // Track when first log was added if this is the first in a batch
@@ -232,11 +259,48 @@ private async Task ProcessRetryQueueAsync()
         }
     }
 
+    private bool IsRateLimited()
+    {
+        lock (_rateLimitLock)
+        {
+            var now = DateTime.UtcNow;
+            var windowStart = now.AddMinutes(-1);
+
+            // Remove timestamps outside the sliding window
+            while (_batchSendTimestamps.Count > 0 && _batchSendTimestamps.Peek() < windowStart)
+            {
+                _batchSendTimestamps.Dequeue();
+            }
+
+            if (_batchSendTimestamps.Count >= _maxBatchesPerMinute)
+            {
+                return true;
+            }
+
+            _batchSendTimestamps.Enqueue(now);
+            return false;
+        }
+    }
+
     private async Task ProcessNewBatchAsync()
     {
         if (_batchData.IsEmpty)
             return;
 
+        // Check rate limit before extracting the batch
+        if (IsRateLimited())
+        {
+            Interlocked.Increment(ref _totalRateLimitDelays);
+            _logger.LogWarning("Rate limit reached ({MaxBatchesPerMinute}/min), delaying batch send", _maxBatchesPerMinute);
+            // Re-trigger after a short delay instead of dropping
+            _ = Task.Run(async () =>
+            {
+                await Task.Delay(TimeSpan.FromSeconds(2), _cancellationTokenSource.Token).ConfigureAwait(false);
+                ResetDebounceTimer();
+            }, _cancellationTokenSource.Token);
+            return;
+        }
+
         var currentBatch = ExtractBatch();
         if (currentBatch.Count == 0)
             return;
@@ -259,9 +323,22 @@ private async Task ProcessNewBatchAsync()
 
         if (!success)
         {
-            // Add to retry queue
+            // Add to retry queue with size enforcement
             lock (_retryLock)
             {
+                if (_options.MaxRetryQueueSize > 0)
+                {
+                    var retryItemCount = _retryQueue.Sum(r => r.Data.Count);
+                    if (retryItemCount + currentBatch.Count > _options.MaxRetryQueueSize)
+                    {
+                        var dropped = currentBatch.Count;
+                        Interlocked.Add(ref _totalDropped, dropped);
+                        _logger.LogWarning("Retry queue at capacity ({MaxSize} items), dropping batch of {Count} logs",
+                            _options.MaxRetryQueueSize, dropped);
+                        return;
+                    }
+                }
+
                 var retryItem = new BatchItem
                 {
                     Data = currentBatch,
@@ -329,6 +406,13 @@ private async Task<bool> SendBatchWithRetryAsync(List<LogLine> batch, int attemp
         if (batch.Count == 0)
             return true;
 
+        // Check circuit breaker before attempting to send
+        if (!_circuitBreaker.AllowRequest())
+        {
+            _logger.LogDebug("Circuit breaker is open, skipping batch send of {Count} items", batch.Count);
+            return false;
+        }
+
         try
         {
             _logger.LogDebug("Sending batch of {Count} log lines (attempt {Attempt})", batch.Count, attemptNumber);
@@ -341,6 +425,7 @@ private async Task<bool> SendBatchWithRetryAsync(List<LogLine> batch, int attemp
 
             if (response.IsSuccessStatusCode)
             {
+                _circuitBreaker.RecordSuccess();
                 Interlocked.Add(ref _totalItemsProcessed, batch.Count);
                 Interlocked.Increment(ref _totalBatchesSent);
                 _lastSuccessfulSend = DateTime.UtcNow;
@@ -350,13 +435,15 @@ private async Task<bool> SendBatchWithRetryAsync(List<LogLine> batch, int attemp
             }
             else
             {
+                _circuitBreaker.RecordFailure();
                 _logger.LogWarning("HTTP error sending batch: {StatusCode} {ReasonPhrase}",
                     response.StatusCode, response.ReasonPhrase);
                 return false;
             }
         }
         catch (TaskCanceledException ex)
         {
+            _circuitBreaker.RecordFailure();
             _logger.LogWarning(ex, "Request cancelled sending batch (attempt {Attempt})", attemptNumber);
             return false;
         }
@@ -367,17 +454,19 @@ private async Task<bool> SendBatchWithRetryAsync(List<LogLine> batch, int attemp
         }
         catch (OperationCanceledException)
         {
+            _circuitBreaker.RecordFailure();
             _logger.LogWarning("Batch send timed out after {Timeout}", _options.SendTimeout);
             return false;
         }
         catch (HttpRequestException ex)
         {
+            _circuitBreaker.RecordFailure();
             _logger.LogWarning(ex, "Network error sending batch (attempt {Attempt})", attemptNumber);
             return false;
         }
-
         catch (Exception ex)
         {
+            _circuitBreaker.RecordFailure();
             _logger.LogError(ex, "Unexpected error sending batch (attempt {Attempt})", attemptNumber);
             return false;
         }
@@ -432,8 +521,12 @@ public BatchingServiceStats GetStats()
                 TotalBatchesSent = _totalBatchesSent,
                 TotalFailures = _totalFailures,
                 TotalValidationFailures = _totalValidationFailures,
+                TotalRateLimitDelays = _totalRateLimitDelays,
+                TotalDropped = _totalDropped,
                 PendingRetries = _retryQueue.Count,
-                LastSuccessfulSend = _lastSuccessfulSend
+                LastSuccessfulSend = _lastSuccessfulSend,
+                CircuitBreakerState = _circuitBreaker.State.ToString(),
+                CircuitBreakerFailures = _circuitBreaker.ConsecutiveFailures
             };
         }
     }
@@ -484,6 +577,11 @@ public class BatchingOptions
     public TimeSpan BatchInterval { get; set; } = TimeSpan.FromSeconds(2); // Now acts as max wait time
     public int MaxBatchSize { get; set; } = 100;
     public TimeSpan SendTimeout { get; set; } = TimeSpan.FromSeconds(30);
+    public int MaxBatchesPerMinute { get; set; } = 60;
+    public int MaxQueueSize { get; set; } = 10000;
+    public int MaxRetryQueueSize { get; set; } = 5000;
+    public int CircuitBreakerFailureThreshold { get; set; } = 5;
+    public TimeSpan CircuitBreakerCooldown { get; set; } = TimeSpan.FromSeconds(30);
 }
 
 public class BatchingServiceStats
@@ -493,8 +591,12 @@ public class BatchingServiceStats
     public long TotalBatchesSent { get; set; }
     public long TotalFailures { get; set; }
     public long TotalValidationFailures { get; set; }
+    public long TotalRateLimitDelays { get; set; }
+    public long TotalDropped { get; set; }
     public int PendingRetries { get; set; }
     public DateTime LastSuccessfulSend { get; set; }
+    public string CircuitBreakerState { get; set; } = string.Empty;
+    public int CircuitBreakerFailures { get; set; }
 }
 
 public class ApiSettings