Queue Backlog Incident

Symptoms

Indicators:
├── Queue depth increasing
├── Messages piling up
├── Consumer lag growing
├── Processing time increasing
├── Downstream effects
└── Time-sensitive jobs delayed

Impact:
├── Order confirmations delayed
├── Email notifications late
├── Search index stale
├── Inventory sync delayed
└── Analytics data outdated

Detection

# Check queue depth
rabbitmqctl list_queues name messages consumers

# Check consumer status
rabbitmqctl list_connections name state

# Check message rates
rabbitmqctl list_queues name messages message_stats.publish

# Monitor with Prometheus
queue_depth{queue="order.export"} 1500
queue_depth{queue="product.sync"} 800

Common Causes

1. Consumer crashed/stopped
2. Consumer too slow
3. Message spike
4. Network issues
5. Resource exhaustion
6. Poison messages
7. Configuration issues
8. Dependency failures

Monitor Consumer Health

// Consumer health check
class ConsumerMonitor
{
    public function checkHealth()
    {
        $queues = $this->queueService->getQueues();
        
        foreach ($queues as $queue) {
            $depth = $queue->getDepth();
            $consumers = $queue->getConsumerCount();
            $rate = $queue->getConsumptionRate();
            
            // Check for backlog
            if ($depth > 1000) {
                $this->alert('Queue backlog: ' . $queue->getName(), 'warning');
            }
            
            // Check for stuck consumers
            if ($consumers === 0 && $depth > 0) {
                $this->alert('No consumers for: ' . $queue->getName(), 'critical');
            }
            
            // Check consumption rate
            if ($rate < 10 && $depth > 100) {
                $this->alert('Slow consumption: ' . $queue->getName(), 'warning');
            }
        }
    }
}

Consumer Metrics

// Track consumer metrics
$metrics = [
    'queue_depth' => $queue->getDepth(),
    'consumer_count' => $queue->getConsumerCount(),
    'consumption_rate' => $queue->getConsumptionRate(),
    'processing_time' => $queue->getAvgProcessingTime(),
    'error_rate' => $queue->getErrorRate()
];

// Alert thresholds
$thresholds = [
    'queue_depth' => ['warning' => 500, 'critical' => 1000],
    'consumer_count' => ['warning' => 1, 'critical' => 0],
    'processing_time' => ['warning' => 5000, 'critical' => 10000],
    'error_rate' => ['warning' => 5, 'critical' => 10]
];

Immediate Actions

// 1. Scale up consumers
public function scaleConsumers($queueName, $count)
{
    for ($i = 0; $i < $count; $i++) {
        $this->consumerService->start($queueName);
    }
}

// 2. Prioritize critical messages
public function prioritizeMessages($queueName)
{
    $messages = $this->queueService->getMessages($queueName);
    
    foreach ($messages as $message) {
        if ($this->isCritical($message)) {
            $this->queueService->prioritize($message);
        }
    }
}

// 3. Purge non-critical messages
public function purgeNonCritical($queueName)
{
    $messages = $this->queueService->getMessages($queueName);
    
    foreach ($messages as $message) {
        if (!$this->isCritical($message)) {
            $this->queueService->delete($message);
        }
    }
}

// 4. Process batch
public function processBatch($queueName, $batchSize = 100)
{
    $messages = $this->queueService->getMessages($queueName, $batchSize);
    
    $this->consumerService->processBatch($messages);
}

Horizontal Scaling

// Auto-scaling based on queue depth
public function autoScale($queueName)
{
    $depth = $this->queueService->getDepth($queueName);
    $consumers = $this->consumerService->getCount($queueName);
    
    // Scale up if depth is high
    if ($depth > 500 && $consumers < 10) {
        $this->scaleConsumers($queueName, $consumers + 2);
    }
    
    // Scale down if depth is low
    if ($depth < 10 && $consumers > 2) {
        $this->scaleConsumers($queueName, $consumers - 1);
    }
}

Resource Management

// Monitor resource usage
$resources = [
    'cpu' => $this->getCPUUsage(),
    'memory' => $this->getMemoryUsage(),
    'connections' => $this->getConnectionCount()
];

// Scale based on resources
if ($resources['cpu'] > 80) {
    $this->alert('High CPU usage', 'warning');
}

if ($resources['memory'] > 80) {
    $this->alert('High memory usage', 'warning');
}

if ($resources['connections'] > 100) {
    $this->alert('High connection count', 'warning');
}

Queue Configuration

// Optimize queue configuration
$config = [
    'prefetch_count' => 10,  // Messages per consumer
    'acknowledgement' => true,
    'dead_letter_queue' => true,
    'message_ttl' => 3600,
    'max_retries' => 3
];

// Apply configuration
foreach ($config as $key => $value) {
    $this->queueService->setConfig($queueName, $key, $value);
}