Re-Engineering a Telemetry Consumer for Resilience — How We Stopped Losing Data Under Load

We Were Losing Telemetry Data and Didn't Know It

The consumer was simple. It read messages from a RabbitMQ queue, called a stored procedure via ADO.NET to insert each record, sent the position to a SignalR hub for live dashboards, forwarded a copy to an integration API, and ACKed the message. One queue. One consumer. One sequential pipeline.

Even at 2,000 devices, the cracks were showing. At peak hours the database would stall, SignalR would freeze, and messages would pile up in RabbitMQ. We treated it as "the database being slow" and moved on. Then I tried to test 10,000 simulated devices — 60,000 messages per minute — and the cracks became a collapse.

The stored procedure was fast for individual calls. But at 1,000 calls per second, the SQL Server connection pool was saturated. Inserts queued. The consumer couldn't ACK fast enough. When a batch of inserts finally failed, the consumer NACKed and requeued the entire message — but some records within that message had already been inserted. On reprocessing, those records were inserted again.

We were losing data in two ways simultaneously: some records were never persisted (dropped during congestion), and others were duplicated (re-inserted on requeue). The SignalR dashboard froze because it was blocked behind the database calls. The integration API stopped receiving data entirely.

Nobody noticed for four hours.

This post covers the complete re-engineering: splitting the monolithic consumer into three independent services using .NET Channels and parallel workers, adding batch throttling over stored procedures using Dapper, and building a deduplication layer that guarantees zero data loss.

Architecture: Before vs After

Bounded Channels — The Backbone of the Design

Every consumer uses a .NET Channel<T> as an internal buffer between RabbitMQ message arrival and actual processing. This is the single most important design decision in the refactor.

Without a channel, every RabbitMQ message is processed synchronously in the consumer callback. If processing is slow, the callback blocks, RabbitMQ stops delivering, and the queue backs up with no visibility. With a bounded channel, the consumer callback writes to the channel and returns immediately. Workers read and process at their own pace.

// DatabaseHandler: Wait mode — backpressure propagates to RabbitMQ
private readonly Channel<(TrackingData Data, int RetryCount)> _channel =
    Channel.CreateBounded<(TrackingData, int)>(
        new BoundedChannelOptions(50_000)
        {
            FullMode = BoundedChannelFullMode.Wait,
            SingleWriter = false,
            SingleReader = false
        });

// SignalRHandler: DropOldest — real-time data is disposable
private readonly Channel<TrackingData> _channel =
    Channel.CreateBounded<TrackingData>(
        new BoundedChannelOptions(10_000)
        {
            FullMode = BoundedChannelFullMode.DropOldest,
            SingleWriter = false,
            SingleReader = false
        });

The DatabaseHandler — Batch Throttling Over Stored Procedures

The stored procedure was never the problem — it was fast, optimized, and battle-tested. The problem was calling it 1,000 times per second individually. The refactor keeps the stored procedure but adds batch throttling at the code level using Dapper.

public class DatabaseHandler : IDisposable
{
    private readonly Channel<(TrackingData Data, int RetryCount)> _channel;
    private readonly Task[] _workers;
    private readonly CancellationTokenSource _cts = new();
    private readonly string _connectionString;
    private readonly ILogger<DatabaseHandler> _logger;

    // Deduplication: DeviceId + PacketNo → insertion timestamp
    private readonly ConcurrentDictionary<string, DateTime> _dedupCache = new();

    private const int WorkerCount = 10;
    private const int ChannelCapacity = 50_000;
    private const int BatchSize = 100;
    private const int MaxRetries = 3;
    private static readonly TimeSpan BatchTimeout = TimeSpan.FromSeconds(2);
    private static readonly TimeSpan DedupWindow = TimeSpan.FromMinutes(5);
    private const int MaxCacheSize = 100_000;

    // Circuit breaker
    private int _consecutiveFailures = 0;
    private DateTime _circuitOpenedAt = DateTime.MinValue;
    private const int CircuitThreshold = 20;
    private static readonly TimeSpan CircuitResetTime = TimeSpan.FromSeconds(30);
    private readonly object _circuitLock = new();

    // Metrics
    private long _inserted, _failed, _deduplicated, _retried;
    public long Inserted => Interlocked.Read(ref _inserted);
    public long Failed => Interlocked.Read(ref _failed);
    public long Deduplicated => Interlocked.Read(ref _deduplicated);
    public long Retried => Interlocked.Read(ref _retried);
    public int QueueLength => _channel.Reader.Count;
    public int CacheSize => _dedupCache.Count;
    public bool IsCircuitOpen() => _consecutiveFailures >= CircuitThreshold
        && DateTime.UtcNow - _circuitOpenedAt < CircuitResetTime;

    public Task Enqueue(TrackingData data)
    {
        var key = $"{data.DeviceId}|{data.PacketNo}";
        if (_dedupCache.TryGetValue(key, out var ts)
            && DateTime.UtcNow - ts < DedupWindow)
        {
            Interlocked.Increment(ref _deduplicated);
            return Task.CompletedTask;
        }

        if (IsCircuitOpen())
            throw new InvalidOperationException(
                "Database circuit breaker is open");

        if (!_channel.Writer.TryWrite((data, 0)))
            throw new InvalidOperationException(
                "Database channel full — backpressure");

        return Task.CompletedTask;
    }

    private async Task WorkerLoop(int workerId)
    {
        var batch = new List<TrackingData>(BatchSize);

        await foreach (var (data, retryCount) in
            _channel.Reader.ReadAllAsync(_cts.Token))
        {
            var key = $"{data.DeviceId}|{data.PacketNo}";
            if (_dedupCache.ContainsKey(key))
            {
                Interlocked.Increment(ref _deduplicated);
                continue;
            }

            if (IsCircuitOpen())
            {
                await Task.Delay(1_000, _cts.Token);
                await _channel.Writer.WriteAsync(
                    (data, retryCount), _cts.Token);
                continue;
            }

            batch.Add(data);

            if (batch.Count >= BatchSize)
            {
                await FlushBatchWithRetry(batch, workerId);
                batch.Clear();
            }
        }

        if (batch.Count > 0)
            await FlushBatchWithRetry(batch, workerId);
    }

    private async Task FlushBatchWithRetry(
        List<TrackingData> batch, int workerId)
    {
        for (int attempt = 0; attempt <= MaxRetries; attempt++)
        {
            try
            {
                using var conn = new SqlConnection(_connectionString);
                await conn.OpenAsync();
                using var tx = conn.BeginTransaction();

                await conn.ExecuteAsync(
                    "usp_InsertTelemetryRecord",
                    batch.Select(r => new
                    {
                        r.DeviceId, r.PacketNo,
                        r.Latitude, r.Longitude,
                        r.Speed, r.Heading, r.Altitude,
                        r.EventCode, r.Timestamp,
                        ReceivedAt = DateTime.UtcNow
                    }),
                    tx,
                    commandType: CommandType.StoredProcedure);

                await tx.CommitAsync();

                // Mark ALL records AFTER successful insert
                var now = DateTime.UtcNow;
                foreach (var r in batch)
                    _dedupCache[$"{r.DeviceId}|{r.PacketNo}"] = now;

                Interlocked.Add(ref _inserted, batch.Count);
                ResetCircuit();
                return;
            }
            catch (Exception ex)
            {
                Interlocked.Increment(ref _retried);
                IncrementCircuit();

                if (attempt == MaxRetries)
                {
                    Interlocked.Add(ref _failed, batch.Count);
                    _logger.LogError(ex,
                        "[Worker {Id}] Batch of {N} failed after {Max} retries",
                        workerId, batch.Count, MaxRetries);
                }
                else
                {
                    await Task.Delay(
                        TimeSpan.FromSeconds(Math.Pow(2, attempt)));
                }
            }
        }
    }
}

Why Dapper + Stored Procedure in Batch

When you pass an IEnumerable of parameters to Dapper's ExecuteAsync with CommandType.StoredProcedure, Dapper calls the stored procedure for each parameter set — but on the same open connection within the same transaction. Instead of 100 separate connections for 100 records, you get 1 connection handling 100 procedure calls. The database can optimize internally, and the connection pool breathes.

The Dual-Trigger Batch Collector

Batches flush on whichever comes first: batch hits 100 records OR 2 seconds elapse. During peak traffic, batches fill quickly — efficient writes. During low traffic (nighttime, trucks parked), the timer ensures records don't wait minutes for 99 more records to arrive.

Why Dedup Marks AFTER Insert

The SignalR Handler — Fire-and-Forget with Parallel Senders

public class SignalRHandler : IDisposable
{
    private readonly Channel<TrackingData> _channel;
    private readonly Task[] _senders;
    private readonly IHubContext<FleetHub> _hubContext;

    private const int SenderCount = 4;
    private const int ChannelCapacity = 10_000;

    private long _sent, _failed, _dropped;
    public long Sent => Interlocked.Read(ref _sent);
    public long Failed => Interlocked.Read(ref _failed);
    public long Dropped => Interlocked.Read(ref _dropped);
    public int QueueLength => _channel.Reader.Count;

    public SignalRHandler(
        IHubContext<FleetHub> hubContext,
        ILogger<SignalRHandler> logger)
    {
        _hubContext = hubContext;

        _channel = Channel.CreateBounded<TrackingData>(
            new BoundedChannelOptions(ChannelCapacity)
            {
                FullMode = BoundedChannelFullMode.DropOldest,
                SingleWriter = false,
                SingleReader = false
            });

        _senders = Enumerable.Range(0, SenderCount)
            .Select(id => Task.Run(() => SenderLoop(id)))
            .ToArray();
    }

    public void Enqueue(TrackingData data)
    {
        if (!_channel.Writer.TryWrite(data))
            Interlocked.Increment(ref _dropped);
    }

    private async Task SenderLoop(int senderId)
    {
        await foreach (var data in
            _channel.Reader.ReadAllAsync())
        {
            try
            {
                await _hubContext.Clients
                    .Group($"device_{data.DeviceId}")
                    .SendAsync("PositionUpdate", new
                    {
                        data.DeviceId,
                        data.Latitude, data.Longitude,
                        data.Speed, data.Heading,
                        data.Timestamp
                    });

                Interlocked.Increment(ref _sent);
            }
            catch
            {
                Interlocked.Increment(ref _failed);
                // No retry — next position in 10 seconds
            }
        }
    }
}

The Worker — Fault-Isolated Consumer Management

public class Worker : BackgroundService
{
    private readonly TelemetryConsumer _telemetry;
    private readonly RealTimeConsumer _realtime;
    private readonly IntegrationConsumer _integration;
    private readonly ILogger<Worker> _logger;

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        var tasks = new[]
        {
            SafeStart(() => _telemetry.StartAsync(ct), "Telemetry"),
            SafeStart(() => _realtime.StartAsync(ct), "RealTime"),
            SafeStart(() => _integration.StartAsync(ct), "Integration"),
        };

        await Task.WhenAll(tasks);
    }

    private async Task SafeStart(Func<Task> start, string name)
    {
        try
        {
            _logger.LogInformation("[Worker] Starting {Name}...", name);
            await start();
        }
        catch (Exception ex)
        {
            _logger.LogError(ex,
                "[Worker] {Name} failed to start — others continue",
                name);
            // Do NOT rethrow — other consumers must keep running
        }
    }
}

If the IntegrationConsumer fails to start (queue doesn't exist yet), TelemetryConsumer and RealTimeConsumer keep running. If TelemetryConsumer's circuit breaker opens during a database outage, the real-time dashboard keeps updating.

Health Monitor Signals — Why They Matter

The health endpoint is not a status page. It's an operational control surface that drives automated responses and gives the on-call engineer actionable information within seconds.

{
  "status": "Degraded",
  "telemetry": {
    "status": "Unhealthy",
    "inserted": 243500,
    "failed": 12,
    "deduplicated": 1488,
    "retried": 36,
    "queue_length": 4200,
    "dedup_cache_size": 45000,
    "circuit_breaker": "OPEN"
  },
  "realtime": {
    "status": "Healthy",
    "signalr_sent": 244800,
    "signalr_failed": 0,
    "signalr_dropped": 200,
    "queue_length": 12
  },
  "integration": {
    "status": "Healthy",
    "processed": 245000,
    "failed": 0
  },
  "summary": {
    "total_processed": 735000,
    "total_failed": 12,
    "total_deduplicated": 1488
  }
}

This response tells you exactly what's happening: the database is down (circuit breaker OPEN, telemetry queue backing up at 4,200), but the fleet map is live (SignalR healthy, only 200 positions dropped) and the integration API is flowing. Without per-consumer health signals, all you'd see is "Unhealthy" — with no idea what to fix or what's still working.

The 6 Signals and Why Each Matters

Benchmark Tables

The Production Outcome

The three patterns that made this work: bounded channels for backpressure, parallel workers for throughput, and batch throttling to stop hammering the database. None of them are exotic. All of them are boring. And boring is what keeps you sleeping at night.