Improve batching, logging, and duplicate handling

Enhanced `BatchingService` and `LogController` to improve scalability, robustness, and clarity:
- Added throttling for drop warnings to reduce log spam.
- Optimized queue size enforcement by dropping 10% of the queue at capacity.
- Handled HTTP 206 responses as successful in `SendBatchWithRetryAsync`.
- Differentiated between client (4xx) and server (5xx) errors; client errors no longer trigger retries.
- Increased default batch size, queue size, and batch rate limits in `BatchingOptions` and `appsettings.json`.
- Improved duplicate log handling in `Create`:
  - Counted duplicates separately and excluded them from validation errors.
  - Adjusted batch status and HTTP responses to treat duplicates as expected behavior.
- Enhanced logging and metrics:
  - Logged duplicate counts at debug level.
  - Summarized validation errors and limited detailed logs to the first five.
  - Updated metrics to exclude duplicates from validation error counts.

These changes optimize performance, reduce unnecessary retries, and provide clearer insights into system behavior.

Author: Timothy Dodd

src/LogMkAgent/Services/BatchService.cs

@@ -50,6 +50,10 @@ public class BatchingService : IDisposable
     private readonly TimeSpan _maxWaitTime = TimeSpan.FromSeconds(2); // Force send after 2 seconds max
     private DateTime _firstLogAddedTime = DateTime.MinValue;
 
+    // Throttle drop warnings to avoid log spam
+    private DateTime _lastDropWarning = DateTime.MinValue;
+    private long _dropsSinceLastWarning;
+
     private volatile bool _disposed;
 
     public BatchingService(
@@ -107,16 +111,26 @@ public void AddData(LogLine data)
             return;
         }
 
-        // Enforce queue size limit - drop oldest items if at capacity
+        // Enforce queue size limit - drop 10% of queue to make room and avoid per-item churn
         if (_options.MaxQueueSize > 0 && _batchData.Count >= _options.MaxQueueSize)
         {
+            var toDrop = Math.Max(1, _options.MaxQueueSize / 10);
             var dropped = 0;
-            while (_batchData.Count >= _options.MaxQueueSize && _batchData.TryDequeue(out _))
+            while (dropped < toDrop && _batchData.TryDequeue(out _))
             {
                 dropped++;
             }
             Interlocked.Add(ref _totalDropped, dropped);
-            _logger.LogWarning("Queue at capacity ({MaxSize}), dropped {Dropped} oldest items", _options.MaxQueueSize, dropped);
+            Interlocked.Add(ref _dropsSinceLastWarning, dropped);
+
+            // Throttle warning to at most once per 10 seconds
+            var now = DateTime.UtcNow;
+            if ((now - _lastDropWarning).TotalSeconds >= 10)
+            {
+                _logger.LogWarning("Queue at capacity ({MaxSize}), dropped {Dropped} oldest items (total since last warning: {TotalDrops})",
+                    _options.MaxQueueSize, dropped, Interlocked.Exchange(ref _dropsSinceLastWarning, 0));
+                _lastDropWarning = now;
+            }
         }
 
         _batchData.Enqueue(data);
@@ -423,20 +437,34 @@ private async Task<bool> SendBatchWithRetryAsync(List<LogLine> batch, int attemp
 
             var response = await _httpClient.SendDataAsync("api/log", batch, linkedCts.Token).ConfigureAwait(false);
 
-            if (response.IsSuccessStatusCode)
+            var statusCode = (int)response.StatusCode;
+
+            if (response.IsSuccessStatusCode || statusCode == 206)
             {
+                // 2xx or 206 Partial — API received and processed the batch
                 _circuitBreaker.RecordSuccess();
                 Interlocked.Add(ref _totalItemsProcessed, batch.Count);
                 Interlocked.Increment(ref _totalBatchesSent);
                 _lastSuccessfulSend = DateTime.UtcNow;
 
-                _logger.LogDebug("Successfully sent batch of {Count} log lines", batch.Count);
+                _logger.LogDebug("Successfully sent batch of {Count} log lines (status {StatusCode})", batch.Count, statusCode);
                 return true;
             }
+            else if (statusCode >= 400 && statusCode < 500)
+            {
+                // 4xx client error — API received the request but rejected it (validation, duplicates, etc.)
+                // Do NOT retry: the same data will get the same response
+                _circuitBreaker.RecordSuccess(); // API is reachable, not a connectivity issue
+                _logger.LogWarning("Batch rejected by API ({StatusCode} {ReasonPhrase}), not retrying — {Count} logs dropped",
+                    response.StatusCode, response.ReasonPhrase, batch.Count);
+                Interlocked.Add(ref _totalFailures, batch.Count);
+                return true; // Return true to prevent retry
+            }
             else
             {
+                // 5xx server error — API is having issues, worth retrying
                 _circuitBreaker.RecordFailure();
-                _logger.LogWarning("HTTP error sending batch: {StatusCode} {ReasonPhrase}",
+                _logger.LogWarning("Server error sending batch: {StatusCode} {ReasonPhrase}",
                     response.StatusCode, response.ReasonPhrase);
                 return false;
             }
@@ -575,9 +603,9 @@ private class BatchItem
 public class BatchingOptions
 {
     public TimeSpan BatchInterval { get; set; } = TimeSpan.FromSeconds(2); // Now acts as max wait time
-    public int MaxBatchSize { get; set; } = 100;
+    public int MaxBatchSize { get; set; } = 500;
     public TimeSpan SendTimeout { get; set; } = TimeSpan.FromSeconds(30);
-    public int MaxBatchesPerMinute { get; set; } = 60;
+    public int MaxBatchesPerMinute { get; set; } = 120;
     public int MaxQueueSize { get; set; } = 10000;
     public int MaxRetryQueueSize { get; set; } = 5000;
     public int CircuitBreakerFailureThreshold { get; set; } = 5;

src/LogMkAgent/appsettings.json

@@ -29,8 +29,9 @@
         "MaxDaysOld": 30
     },
     "BatchingOptions": {
-        "MaxBatchesPerMinute": 60,
-        "MaxQueueSize": 10000,
-        "MaxRetryQueueSize": 5000
+        "MaxBatchSize": 500,
+        "MaxBatchesPerMinute": 120,
+        "MaxQueueSize": 50000,
+        "MaxRetryQueueSize": 10000
     }
 }

src/LogMkApi/Controllers/LogController.cs

@@ -94,6 +94,7 @@ public async Task<ActionResult<LogResponse>> Create(
         var validLogs = new List<Log>();
         var errors = new List<LogValidationError>();
         var skippedCount = 0;
+        var duplicateCount = 0;
 
         _logger.LogDebug("Processing batch {BatchId} with {Count} log lines", batchId, logLines.Count);
         _metrics.IncrementLogsReceived(logLines.Count);
@@ -111,6 +112,14 @@ public async Task<ActionResult<LogResponse>> Create(
 
                 if (!validationResult.IsValid)
                 {
+                    // If the ONLY error is duplicate detection, count it separately (not as an error)
+                    if (validationResult.Errors.Length == 1 && validationResult.Errors[0] == "Duplicate log entry detected")
+                    {
+                        duplicateCount++;
+                        skippedCount++;
+                        continue;
+                    }
+
                     errors.Add(new LogValidationError
                     {
                         Index = i,
@@ -214,7 +223,13 @@ public async Task<ActionResult<LogResponse>> Create(
             }
         }
 
-        // Log validation errors for monitoring
+        // Log duplicate count at debug level (duplicates are normal, not errors)
+        if (duplicateCount > 0)
+        {
+            _logger.LogDebug("Batch {BatchId}: {DuplicateCount} duplicate logs skipped", batchId, duplicateCount);
+        }
+
+        // Log real validation errors for monitoring
         if (errors.Any())
         {
             // Group errors by type for better insights
@@ -225,7 +240,7 @@ public async Task<ActionResult<LogResponse>> Create(
 
             _logger.LogWarning("Batch {BatchId}: {ErrorCount} validation errors, {SkippedCount} logs skipped. " +
                 "Error breakdown: {ErrorSummary}",
-                batchId, errors.Count, skippedCount,
+                batchId, errors.Count, skippedCount - duplicateCount,
                 string.Join(", ", errorSummary.Select(kvp => $"{kvp.Key}: {kvp.Value}")));
 
             // Log individual rejected lines for debugging (limit to first 5)
@@ -235,7 +250,7 @@ public async Task<ActionResult<LogResponse>> Create(
                     batchId, error.Index, string.Join(", ", error.Errors), error.LogLine);
             }
 
-            _metrics.IncrementErrors("validation", skippedCount);
+            _metrics.IncrementErrors("validation", errors.Count);
         }
 
         // Log insert errors
@@ -255,7 +270,7 @@ public async Task<ActionResult<LogResponse>> Create(
             ProcessedCount = insertedCount,
             SkippedCount = skippedCount,
             ReceivedAt = receivedAt,
-            Status = GetBatchStatus(logLines.Count, insertedCount, skippedCount, insertErrors.Any())
+            Status = GetBatchStatus(logLines.Count, insertedCount, skippedCount, insertErrors.Any(), duplicateCount)
         };
 
         // Include validation errors if any (but limit them)
@@ -270,19 +285,22 @@ public async Task<ActionResult<LogResponse>> Create(
         }
 
         // Return appropriate status code
-        if (insertedCount == 0 && logLines.Count > 0)
+        // Duplicates are NOT errors — they're expected in normal operation (agent retries, etc.)
+        var realErrorCount = errors.Count; // Excludes duplicates (already filtered above)
+
+        if (insertedCount == 0 && realErrorCount > 0)
         {
-            // No logs were inserted
+            // No logs inserted AND there were real validation/insert errors
             return BadRequest(response);
         }
-        else if (skippedCount > 0 || insertErrors.Any())
+        else if (insertErrors.Any() || realErrorCount > 0)
         {
-            // Partial success
-            return StatusCode(206, response); // 206 Partial Content
+            // Partial success with real errors (not just duplicates)
+            return StatusCode(206, response);
         }
         else
         {
-            // Complete success
+            // Complete success, or all skipped were just duplicates
             return Ok(response);
         }
     }
@@ -405,14 +423,15 @@ private string SanitizeForLogging(LogLine logLine)
         }
     }
 
-    private string GetBatchStatus(int received, int processed, int skipped, bool hasInsertErrors)
+    private string GetBatchStatus(int received, int processed, int skipped, bool hasInsertErrors, int duplicates = 0)
     {
-        if (processed == 0 && received > 0)
+        var realErrors = skipped - duplicates;
+        if (processed == 0 && realErrors > 0)
             return "Failed";
-        else if (processed == received && !hasInsertErrors)
-            return "Success";
-        else
+        else if (hasInsertErrors || realErrors > 0)
             return "Partial";
+        else
+            return "Success";
     }
 
     [HttpGet("stats")]