Getting Started

Minimal Steps

Install the Bundle:
```
composer require answear/messenger-heartbeat-bundle
```
- Symfony Flex automatically registers the bundle in config/bundles.php.
Enable Heartbeat for Workers: Add the --keepalive flag to your existing messenger:consume and messenger:failed:retry commands:
```
php bin/console messenger:consume async -vv --keepalive
php bin/console messenger:failed:retry all --keepalive
```
Verify Integration:
- Check logs for heartbeat activity (introduced in v3.2.0). Look for messages like:
```
[heartbeat] Sending keepalive signal...
```
- Ensure no message duplicates occur during transport exceptions (fixed in v3.1.0).

First Use Case

Scenario: Your application uses Symfony Messenger with AMQP transport (e.g., RabbitMQ) for processing long-running background jobs (e.g., video encoding, data migration). You notice intermittent connection drops causing message loss or retries.

Solution:

Update your worker command to include --keepalive:

php bin/console messenger:consume workers:video-encoding --keepalive

Monitor logs to confirm heartbeats are active and no silent failures occur.
Test failure scenarios (e.g., network interruptions) to validate resilience.

Implementation Patterns

Usage Patterns

Standard Worker Configuration:

Always include --keepalive for workers handling long-running or critical tasks:

# For async processing
php bin/console messenger:consume async --keepalive --limit=10

# For failed message retries
php bin/console messenger:failed:retry all --keepalive

Environment-Specific Flags:

Use environment variables to toggle heartbeat in CI/CD pipelines or staging:

# In Dockerfile or deployment script
if [ "$ENVIRONMENT" = "production" ]; then
  EXTRA_FLAGS="--keepalive"
fi
php bin/console messenger:consume async $EXTRA_FLAGS

Selective Heartbeat Activation:

Enable heartbeats only for specific queues or workers by creating custom commands:

// In a custom command class
protected function configure(): void
{
    $this->addOption('keepalive', null, InputOption::VALUE_NONE, 'Enable heartbeat for this worker');
}

protected function execute(InputInterface $input, OutputInterface $output): int
{
    if ($input->getOption('keepalive')) {
        // Custom logic to enable heartbeat for this worker only
    }
    return $this->runMessengerCommand($input, $output);
}

Workflows

Heartbeat + Retry Logic:

Combine with Symfony Messenger’s retry logic for robust failure handling:

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    keepalive: true  # Not directly configurable; use CLI flag
                retry_strategy:
                    max_retries: 3
                    delay: 1000
                    multiplier: 2

Monitoring Integration:

Use Symfony’s logging system to track heartbeat activity:

// In a custom listener or subscriber
public function onMessageSent(MessageSentEvent $event)
{
    if (method_exists($event->getBus(), 'getTransport')) {
        $transport = $event->getBus()->getTransport();
        if ($transport instanceof AmqpTransport) {
            $this->logger->info('Heartbeat active for transport', [
                'transport' => get_class($transport),
                'connection' => $transport->getConnection()->getConnectionName(),
            ]);
        }
    }
}

Graceful Shutdown:

Handle worker shutdowns gracefully by combining heartbeat with signals:

// In a custom worker bootstrap script
declare(ticks=1);
pcntl_signal(SIGTERM, function () {
    $this->logger->info('Shutting down worker gracefully...');
    // Allow current message to process before exiting
    sleep(1);
    exit(0);
});

Integration Tips

AMQP Transport Configuration:

Ensure your messenger.yaml is configured for AMQP:

framework:
    messenger:
        transports:
            async:
                dsn: 'amqp://user:pass@rabbitmq:5672/%2f/messages'
                options:
                    keepalive: true  # CLI flag overrides this; kept for clarity

Custom Transport Extensions:

Extend the bundle for custom AMQP configurations:

use Answear\MessengerHeartbeatBundle\DependencyInjection\Configuration;
use Symfony\Component\DependencyInjection\ContainerBuilder;

class CustomHeartbeatExtension extends Configuration
{
    public function load(array $configs, ContainerBuilder $container)
    {
        parent::load($configs, $container);
        // Add custom heartbeat interval or other options
        $container->setParameter('messenger_heartbeat.interval', 30);
    }
}

Docker/Kubernetes Deployments:

Configure liveness probes to complement heartbeats:

# kubernetes/deployment.yaml
livenessProbe:
  exec:
    command:
      - sh
      - -c
      - "php bin/console messenger:consume async --keepalive --limit=1 | grep -q 'Heartbeat active'"
  initialDelaySeconds: 30
  periodSeconds: 10

Gotchas and Tips

Pitfalls

Symfony Version Mismatch:
- Issue: The bundle drops support for Symfony <7.2 (v3.0.0+) and PHP <8.2 (v2.0.0+).
- Fix: Upgrade your Symfony and PHP versions before installation:
```
composer require symfony/*:^7.2 php:^8.2
```
Heartbeat Not Triggering:
- Issue: Heartbeat logs are missing, indicating the --keepalive flag isn’t being recognized.
- Debug:
  - Verify the bundle is registered in config/bundles.php.
  - Check for typos in the CLI flag (must be --keepalive, not --heartbeat).
  - Ensure you’re using the messenger:consume or messenger:failed:retry commands (not custom scripts).
Message Duplication:
- Issue: Messages are duplicated during transport exceptions despite using v3.1.0+.
- Fix: Ensure your AMQP broker (e.g., RabbitMQ) has prefetch_count set appropriately in your transport options:
```
options:
    prefetch_count: 10  # Adjust based on your workload
```
Global Signals Conflicts:
- Issue: Heartbeat global signals (introduced in v3.2.0) interfere with other processes using PCNTL.
- Fix: Isolate workers in separate processes or namespaces, or disable global signals by extending the bundle:
```
// Override the heartbeat service
services:
    Answear\MessengerHeartbeatBundle\Heartbeat\Heartbeat:
        arguments:
            $useGlobalSignals: false
```
Performance Overhead:
- Issue: Heartbeat introduces noticeable CPU overhead in high-throughput systems.
- Mitigation:
  - Monitor CPU usage with tools like htop or Prometheus.
  - Adjust the heartbeat interval (if customizable in future versions) or reduce worker concurrency.

Debugging

Enable Verbose Logging:

Run workers with -vv to see heartbeat activity:

php bin/console messenger:consume async --keepalive -vv

Look for logs like:

[heartbeat] Last activity: 10 seconds ago
[heartbeat] Sending keepalive signal...

Check AMQP Connection:
- Use rabbitmqctl or the RabbitMQ management UI to verify connection status:
```
rabbitmqctl list_connections | grep 'heartbeat'
```

Test Network Interruptions:

Simulate network drops to validate heartbeat resilience:

# Linux (using tc)
sudo tc qdisc add dev eth0 root netem loss 5%  # 5% packet loss
# Re-run worker and check for recovery
php bin/console messenger:consume async --keepalive

Configuration Quirks

CLI Flag Precedence:
- The --keepalive flag overrides any transport-level keepalive options in messenger.yaml. CLI flags take priority.
Bundle Auto-Registration:
- Symfony Flex auto-registers the bundle, but manual

Messenger Heartbeat Bundle Laravel Package