Messenger Heartbeat Bundle Laravel Package

Getting Started

Minimal Steps

1. Installation:

   composer require answear/messenger-heartbeat-bundle
   

   The bundle auto-registers in config/bundles.php via Symfony Flex.

2. Enable Heartbeat: Add --keepalive to your existing messenger:consume and messenger:failed:retry commands:

   php bin/console messenger:consume async --keepalive
   php bin/console messenger:failed:retry --keepalive
   

3. Verify: Check logs for heartbeat activity (enabled by default in v3.2.0+):

   grep -i "heartbeat\|keepalive" var/log/dev.log
   

First Use Case

Scenario: Long-running workers (e.g., processing large files) frequently disconnect due to idle timeouts in RabbitMQ. Solution:

1. Update your supervisor.conf or deployment script to include --keepalive:

   command=php bin/console messenger:consume async --keepalive
   

2. Monitor connection.last_activity logs to confirm heartbeats are active.

Implementation Patterns

Core Workflow

1. Worker Initialization:

   • The bundle hooks into Symfony Messenger’s AMQP transport via a KeepaliveListener.
   • On --keepalive, it registers a global signal handler (e.g., SIGALRM) to send periodic pings.

2. Heartbeat Mechanism:

   • Uses PNCTL (Proactor Network Control) to send keepalive signals without blocking the worker.
   • Default interval: 30 seconds (configurable via AMQP transport settings).

3. Integration Points:

   • Command-Line: Only active when --keepalive is passed to messenger:consume/failed:retry.
   • Transport Layer: Works with symfony/amqp-messenger (RabbitMQ, Qpid).
   • Event Listeners: Extendable via KeepaliveInterface (e.g., custom heartbeat intervals).

Common Patterns

1. Selective Activation:

   # Enable for specific queues only (via environment variables or config)
   if (app()->environment('production')) {
       $this->addOption('--keepalive', null, InputOption::VALUE_NONE, 'Enable heartbeat for production workers');
   }
   

2. Custom Heartbeat Interval: Modify the AMQP transport configuration in config/packages/messenger.yaml:

   framework:
       messenger:
           transports:
               async:
                   dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                   options:
                       heartbeat_interval: 60  # Custom interval in seconds
   

3. Logging and Monitoring: Use Symfony’s logger to track heartbeat status:

   // In a custom listener
   public function onMessageSent(TransportSentEvent $event) {
       if ($event->getTransport() instanceof AmqpTransport) {
           $this->logger->info('Heartbeat active', [
               'connection' => $event->getTransport()->getConnection()->getName(),
               'last_activity' => $event->getTransport()->getConnection()->getLastActivityTime(),
           ]);
       }
   }
   

4. Docker/Kubernetes Deployments: Ensure the --keepalive flag is passed in your container’s command:

   # docker-compose.yml
   services:
       messenger:
           command: php bin/console messenger:consume async --keepalive
   

Anti-Patterns

• Global Enablement Without Testing: Avoid enabling --keepalive across all workers without benchmarking overhead (e.g., CPU/network usage).
• Ignoring Logs: Heartbeat logs (v3.2.0+) are critical for debugging. Silence them only if you have alternative monitoring.
• Mixing with Custom Heartbeat Logic: The bundle uses global signals. Override its behavior only via KeepaliveInterface to avoid conflicts.

Gotchas and Tips

Pitfalls

1. Symfony Version Mismatch:

   • Issue: Bundle drops support for Symfony <7.2 (v3.0.0+) and PHP <8.2 (v2.0.0+).
   • Fix: Upgrade or fork the bundle if stuck on older versions.
   • Check: Run composer why-not symfony/messenger:^7.2 to validate compatibility.

2. AMQP Broker Incompatibility:

   • Issue: Not all AMQP brokers support PNCTL heartbeats (e.g., some RabbitMQ versions or custom brokers).
   • Fix: Test with your broker’s version. Fall back to broker-native heartbeats if needed:

     # config/packages/messenger.yaml
     options:
         heartbeat: 60  # Broker-level heartbeat (RabbitMQ-specific)
     

3. Signal Handler Conflicts:

   • Issue: Global signals (e.g., SIGALRM) may interfere with other processes in shared environments (e.g., Docker containers with multiple workers).
   • Fix: Isolate workers or use process managers like Supervisor with numprocs=1.

4. Message Duplication:

   • Issue: Transport exceptions (e.g., network blips) can still cause duplicates (fixed in v3.1.0).
   • Fix: Use idempotent message handlers or enable skip_message_on_failure in listeners (v4.1.0+):

     // config/packages/messenger.yaml
     failure_transport: failed
     transports:
         failed:
             dsn: '%env(FAILED_TRANSPORT_DSN)%'
             options:
                 skip_message_on_failure: true
     

5. Log Overload:

   • Issue: Verbose heartbeat logs may clutter production logs.
   • Fix: Adjust log level in config/packages/monolog.yaml:

     handlers:
         main:
             level: warning  # Suppress INFO-level heartbeat logs
     

Debugging Tips

1. Verify Heartbeat Activity:

   # Check if keepalive is active
   php bin/console debug:container | grep KeepaliveListener
   
   # Tail logs for heartbeat events
   tail -f var/log/dev.log | grep -i "heartbeat\|keepalive"
   

2. Connection Health Checks: Use Symfony’s TransportConnection events to monitor status:

   // src/EventListener/ConnectionListener.php
   use Symfony\Component\Messenger\Transport\Connection;
   
   public function onConnectionEstablished(ConnectionEvent $event) {
       if ($event->getConnection() instanceof Connection) {
           $this->logger->info('Connection established', [
               'last_activity' => $event->getConnection()->getLastActivityTime(),
           ]);
       }
   }
   

3. AMQP-Specific Issues:

   • RabbitMQ: Check broker logs for connection drops:

     rabbitmqctl list_connections | grep -i "heartbeat\|idle"
     

   • Qpid: Verify heartbeat settings in the broker’s configuration file.

4. Performance Profiling:

   • Measure CPU/network overhead with --keepalive:

     # Compare with/without heartbeat
     hyperfine --warmup 5 'php bin/console messenger:consume async --keepalive'
     

Extension Points

1. Custom Heartbeat Interval: Implement Answear\MessengerHeartbeatBundle\Keepalive\KeepaliveInterface:

   use Answear\MessengerHeartbeatBundle\Keepalive\KeepaliveInterface;
   
   class CustomKeepalive implements KeepaliveInterface {
       public function getInterval(): int {
           return 45; // Custom interval in seconds
       }
   }
   

   Register it in services.yaml:

   services:
       Answear\MessengerHeartbeatBundle\Keepalive\KeepaliveInterface: '@App\CustomKeepalive'
   

2. Preventing Signal Conflicts: Override the signal handler for specific environments:

   // src/Keepalive/EnvironmentAwareKeepalive.php
   use Answear\MessengerHeartbeatBundle\Keepalive\KeepaliveInterface;
   
   class EnvironmentAwareKeepalive implements KeepaliveInterface {
       public function __construct(private string $environment) {}
   
       public function getInterval(): int {
           return $this->environment === 'production' ? 60 : 30;
       }
   }
   

3. Integration with Monitoring: Export heartbeat metrics to Prometheus:

   use Symfony\Component\Messenger\Transport\Connection;
   use Prometheus\CollectorRegistry;
   
   public function onMessageSent(TransportSentEvent $event) {
       if ($event->getTransport() instanceof AmqpTransport) {
           $registry = new CollectorRegistry();
           $registry->getOrRegisterCounter('messenger_heartbeat_active', 'Heartbeat status')
                    ->inc($event->getTransport()->getConnection()->isHeartbeatActive() ? 1 : 0);