The 3 AM RabbitMQ Crisis That Taught Me Everything About Message Durability

// This is the pattern I use in every production system now
// After losing messages twice, I never skip these flags

const queueOptions = {
  durable: true,        // Queue survives broker restart
  exclusive: false,     // Queue can be accessed by multiple connections  
  autoDelete: false,    // Queue won't be deleted when consumer disconnects
  arguments: {
    'x-queue-mode': 'lazy',  // Forces messages to disk immediately
    'x-max-priority': 10     // Enables priority queuing if needed
  }
};

await channel.assertQueue('payment-processing', queueOptions);

// Exchange durability - don't make my mistake of forgetting this
await channel.assertExchange('payment-events', 'topic', {
  durable: true,        // Exchange survives restart
  autoDelete: false,    // Exchange won't be auto-deleted
  internal: false       // Exchange can receive messages from publishers
});

// The message publishing pattern I use for critical data
// Every option here prevents a specific type of message loss

const publishOptions = {
  persistent: true,           // Message survives broker restart
  mandatory: true,            // Ensures message reaches a queue
  deliveryMode: 2,           // Explicitly marks message as persistent
  timestamp: Date.now(),     // Helps with message tracking
  messageId: generateUUID(), // Unique identifier for deduplication
  expiration: '3600000'      // 1 hour TTL to prevent queue bloat
};

const success = channel.publish(
  'payment-events',           // Exchange name
  'payment.processed',        // Routing key
  Buffer.from(JSON.stringify(messageData)),
  publishOptions
);

if (!success) {
  // Channel is blocked - implement backpressure handling
  await new Promise(resolve => channel.once('drain', resolve));
}

// Enable publisher confirms - this is non-negotiable for production
await channel.confirmSelect();

async function publishWithConfirmation(exchange, routingKey, message, options) {
  return new Promise((resolve, reject) => {
    const success = channel.publish(
      exchange, 
      routingKey, 
      Buffer.from(JSON.stringify(message)), 
      options,
      (err, ok) => {
        if (err) {
          reject(new Error(`Failed to publish message: ${err.message}`));
        } else if (ok) {
          resolve(ok);
        } else {
          reject(new Error('Message was not confirmed by broker'));
        }
      }
    );
    
    if (!success) {
      reject(new Error('Channel write buffer full'));
    }
  });
}

// Usage with proper error handling
try {
  await publishWithConfirmation('payment-events', 'payment.processed', paymentData, publishOptions);
  console.log('Message safely persisted to disk');
} catch (error) {
  // Implement retry logic or dead letter handling
  await handlePublishFailure(paymentData, error);
}

// Never use auto-ack in production - I learned this lesson painfully
await channel.consume('payment-processing', async (msg) => {
  if (!msg) return;
  
  try {
    const paymentData = JSON.parse(msg.content.toString());
    
    // Process the message (database writes, API calls, etc.)
    await processPayment(paymentData);
    
    // Only acknowledge AFTER successful processing
    channel.ack(msg);
    
  } catch (error) {
    console.error('Payment processing failed:', error);
    
    // Check if this is a retryable error
    if (isRetryableError(error)) {
      // Reject and requeue for retry
      channel.nack(msg, false, true);
    } else {
      // Send to dead letter queue for manual investigation
      channel.nack(msg, false, false);
    }
  }
}, {
  noAck: false  // Manual acknowledgment mode
});

async function processWithRetry(msg, maxRetries = 3) {
  const retryCount = (msg.properties.headers['x-retry-count'] || 0);
  
  if (retryCount >= maxRetries) {
    // Send to dead letter queue
    await channel.publish('dlx-exchange', '', msg.content, {
      persistent: true,
      headers: {
        'original-queue': 'payment-processing',
        'failure-reason': 'max-retries-exceeded',
        'original-routing-key': msg.fields.routingKey
      }
    });
    channel.ack(msg);
    return;
  }
  
  try {
    await processPayment(JSON.parse(msg.content.toString()));
    channel.ack(msg);
  } catch (error) {
    // Increment retry count and requeue with delay
    setTimeout(() => {
      channel.publish(msg.fields.exchange, msg.fields.routingKey, msg.content, {
        persistent: true,
        headers: {
          'x-retry-count': retryCount + 1
        }
      });
      channel.ack(msg);
    }, Math.pow(2, retryCount) * 1000); // Exponential backoff
  }
}

// Dead letter exchange setup - your safety net for failed messages
await channel.assertExchange('dlx-exchange', 'direct', { durable: true });
await channel.assertQueue('dead-letter-queue', {
  durable: true,
  arguments: {
    'x-message-ttl': 86400000,  // 24 hours before cleanup
    'x-queue-mode': 'lazy'      // Force to disk immediately
  }
});
await channel.bindQueue('dead-letter-queue', 'dlx-exchange', '');

// Main queue with dead letter routing
await channel.assertQueue('payment-processing', {
  durable: true,
  arguments: {
    'x-dead-letter-exchange': 'dlx-exchange',
    'x-dead-letter-routing-key': 'failed',
    'x-queue-mode': 'lazy'
  }
});

// In-memory deduplication cache (use Redis in production)
const processedMessages = new Set();

async function processWithDeduplication(msg) {
  const messageId = msg.properties.messageId;
  
  if (processedMessages.has(messageId)) {
    console.log(`Duplicate message detected: ${messageId}`);
    channel.ack(msg);
    return;
  }
  
  try {
    await processPayment(JSON.parse(msg.content.toString()));
    processedMessages.add(messageId);
    channel.ack(msg);
  } catch (error) {
    channel.nack(msg, false, true);
  }
}

// Health check endpoint that monitors queue durability
app.get('/health/rabbitmq', async (req, res) => {
  try {
    const queueInfo = await channel.checkQueue('payment-processing');
    
    const healthMetrics = {
      queueExists: true,
      messageCount: queueInfo.messageCount,
      consumerCount: queueInfo.consumerCount,
      isDurable: queueInfo.durable,
      status: 'healthy'
    };
    
    // Alert if queue becomes non-durable (shouldn't happen but does)
    if (!queueInfo.durable) {
      healthMetrics.status = 'critical';
      healthMetrics.error = 'Queue durability lost';
    }
    
    // Alert if messages are piling up
    if (queueInfo.messageCount > 10000) {
      healthMetrics.status = 'warning';
      healthMetrics.warning = 'High message backlog';
    }
    
    res.json(healthMetrics);
  } catch (error) {
    res.status(500).json({
      status: 'critical',
      error: error.message
    });
  }
});

// Batch publishing to reduce overhead
async function publishBatch(messages) {
  const promises = messages.map(msg => 
    publishWithConfirmation(msg.exchange, msg.routingKey, msg.data, msg.options)
  );
  
  return Promise.all(promises);
}

// Connection pooling for high-throughput scenarios
class RabbitMQPool {
  constructor(connectionString, poolSize = 5) {
    this.connections = [];
    this.channels = [];
    this.poolSize = poolSize;
    this.connectionString = connectionString;
  }
  
  async initialize() {
    for (let i = 0; i < this.poolSize; i++) {
      const connection = await amqp.connect(this.connectionString);
      const channel = await connection.createConfirmChannel();
      
      this.connections.push(connection);
      this.channels.push(channel);
    }
  }
  
  getChannel() {
    // Round-robin channel selection
    return this.channels[Math.floor(Math.random() * this.channels.length)];
  }
}