README

A Symfony Messenger transport integration for NATS JetStream, enabling reliable asynchronous messaging with persistent message streaming.

Features

🚀 High-Performance Messaging - Leverage NATS JetStream for fast, reliable message delivery
📦 Symfony Integration - Implements Messenger's TransportInterface, MessageCountAwareInterface, SetupableTransportInterface, KeepaliveReceiverInterface, and CloseableTransportInterface
⚙️ Configurable Consumers - Support for multiple consumer strategies
🔄 Flexible Batching - Adjustable message batch sizes and timeouts
🔐 Authentication Support - Built-in support for NATS authentication
📊 Stream Configuration - Configurable retention policies and replication
📍 Stream Placement - Pin a stream to a JetStream cluster and to tagged servers; an option you set changes only its own half of an existing placement (see Stream Placement)
🔁 Automatic Reconnect - Optional mode in which the NATS client recovers a lost connection and a worker polls through an outage instead of exiting (see Automatic Reconnect)
🌐 NATS Clusters - One server in the DSN, the others learned from the cluster; failover, replicated streams and TLS on a cluster (see Using a NATS Cluster)
🧪 Thoroughly Tested - 673 unit tests, ~99.8% coverage, mutation-tested (96.9% MSI)

🚀 This project looks for funding. Love my work? Support it! 💖

☕ Buy me a coffee: https://buymeacoffee.com/idct
💝 Sponsor: https://github.com/sponsors/ideaconnect

Requirements

System Requirements

PHP: ^8.2
Symfony: ^7.2 || ^8
NATS Server: ^2.9 with JetStream enabled, ^2.12 for scheduled messages support and for ! exclusion tags in stream_placement_tags.

Installation

composer require idct/symfony-nats-messenger

The package ships IDCTNatsMessengerBundle, which registers the transport factory and the igbinary serializer for you - no per-project services.yaml entries are required. With Symfony Flex the bundle is enabled automatically when the package is installed or updated. Without Flex, enable it manually:

// config/bundles.php
return [
    // ...
    IDCT\NatsMessenger\IDCTNatsMessengerBundle::class => ['all' => true],
];

If you leave the bundle out, register the transport factory yourself. Messenger finds it by its tag, and nothing else registers it, so without it every nats-jetstream:// DSN fails with "No transport supports Messenger DSN":

# config/services.yaml (only without the bundle)
services:
    IDCT\NatsMessenger\NatsTransportFactory:
        tags: [messenger.transport_factory]

Tested by: transportFactoryIsRegisteredAndTaggedForMessengerDiscovery, withoutTheBundleNoTransportFactorySupportsANatsDsn, theReadmesIgbinaryExampleGivesTheTransportTheIgbinarySerializer, theReadmesSetupWithoutTheBundleGivesAWorkingTransport - the functional Behat suite also runs entirely on the bundle's service registration.

Upgrading from 5.4 or earlier

With Symfony Flex, updating the package adds the bundle to config/bundles.php, as installing it does. Commit that change with the update.
The services.yaml entries you added for the transport keep working next to the bundle, and you can delete them, with one exception: the bundle does not register the igbinary_serializer id earlier versions of this README suggested. If a transport has serializer: 'igbinary_serializer', change it to serializer: 'IDCT\NatsMessenger\Serializer\IgbinarySerializer' before you delete that entry, or keep the entry. Otherwise the container cannot be built, because the service the transport names is gone.
Going back to 5.4 or earlier does not take the bundle out of config/bundles.php. The application then stops with Class "IDCT\NatsMessenger\IDCTNatsMessengerBundle" not found, and so does every composer install or composer update that clears the cache. Delete the line when you downgrade, and add it back yourself if you update again later: Flex configures a package once, and the package's entry in symfony.lock stays after the downgrade.

Tested by: theServiceEntriesUsedBeforeTheBundleKeepWorking, theReadmesIgbinaryExampleGivesTheTransportTheIgbinarySerializer

Upgrading from 5.6

Nothing to change. The reconnect option is off by default, and with it off the transport recovers a lost connection as before. Two differences apply in either mode:

A pull the server ended for its own shutdown (409 Server Shutdown) or for a leadership change (409 Leadership Change) is an empty batch now, not a missing consumer. With auto_setup the transport no longer runs setup() against a server that is going away, which failed every graceful stop of the server with "Failed to setup NATS stream" and ended the worker. Every other 409 keeps its meaning.
If the application has a logger service, the bundle hands it to the NATS client, which writes its connection lifecycle at info level and its errors at error level, under Monolog's messenger channel (see Logging). Those lines went nowhere before.

Tested by: testAPullTheServerEndedForItsShutdownIsEmptyAndTheNextOperationChecksTheConnection, testA409ReportingAConfigurationErrorStaysVisible, theBundleHandsTheApplicationsLoggerToTheFactory, the example reconnect.php

Upgrading from 5.7.0

Nothing to change. 5.7.1 requires idct/php-nats-jetstream-client 2.27.1 or later, which composer update installs. With reconnect: true, an ack() or reject() whose write is the first to notice a lost connection now returns at once, its frame buffered for the connection the client's recovery makes, where with client 2.27.0 it waited for that recovery (see Automatic Reconnect). With the option off nothing changes.

Tested by: testAnAcknowledgementWhoseWriteFindsTheConnectionLostReturnsAtOnceAndTheRecoverySendsIt

Logging

The NATS client reports what it does with the connection through a PSR-3 logger: the connection's lifecycle (connected, reconnected, closed) at info level, a reconnect attempt that failed at warning level, and its errors at error level. Among the errors is the only trace of acknowledgements lost to an outage with reconnect: true: Reconnect exhausted: N bytes of buffered publishes were discarded. There is no debug-level chatter, so a wired logger costs a few lines per connection.

With the bundle, the application's logger service is handed to the factory, tagged for Monolog's messenger channel, so the client's lines land next to the worker's. Without that service, or without the bundle, the client gets no logger, as before 5.7.0. To wire one yourself, give it to the factory, which hands it to every transport it creates, or to a transport constructed directly, as the fourth constructor argument: new NatsTransport($dsn, $options, $serializer, $logger).

# config/services.yaml (without the bundle, with a logger)
services:
    IDCT\NatsMessenger\NatsTransportFactory:
        arguments: ['@?logger']
        tags: [messenger.transport_factory]

Tested by: testBuildPassesTheLoggerToTheClientAndNoneByDefault, createTransport_HandsTheFactorysLoggerToTheClient, theBundleHandsTheApplicationsLoggerToTheFactory, theFactoryIsTaggedForMonologsMessengerChannel, testAReconnectThatGivesUpWithABufferedAcknowledgementIsReportedToTheLoggerAndTheNextOperationFails, the examples reconnect.php and cluster-failover.php

Development Setup

For contributors and development:

# Install dependencies
composer install

# Run static analysis and the default unit test suite after every modification
composer test

# Start NATS server for testing
composer nats:start

# Run unit tests with coverage
composer test:unit

# Set up functional tests
composer test:functional:setup

# Run functional tests
composer test:functional

# Stop NATS server
composer nats:stop

Quick Start

1. Configure NATS Server

Ensure your NATS server has JetStream enabled:

nats-server -js

2. Set Up Transport in Symfony

Add the NATS transport to your Symfony Messenger configuration:

# config/packages/messenger.yaml
framework:
  messenger:
    transports:
      nats_transport:
        dsn: 'nats-jetstream://localhost:4222/my-stream/my-topic'
        options:
          consumer: 'my-consumer'
          batching: 5
          max_batch_timeout: 1.0

    routing:
      'App\Message\MyAsyncMessage': nats_transport

Tested by: testReadmeDsnExamplesParseSuccessfully[README: quick-start transport], testReadmeConfigurationOptionsAreAccepted, testReadmeBatchingExamplesAreAccepted, testReadmeTimeoutExamplesAreAccepted

3. Configure Custom Serializers (Optional)

This transport ships with a high-performance IgbinarySerializer, but under the Symfony framework it is not selected automatically. Symfony Messenger always resolves a serializer and passes it to the transport factory - its framework default is the native PhpSerializer - so to use igbinary you must set the transport's serializer: key explicitly (shown below). The transport's own igbinary auto-selection (and the PhpSerializer fallback, with an E_USER_WARNING, when ext-igbinary is unavailable) only applies when you construct NatsTransport directly without passing a serializer, e.g. outside the framework.

Using IgbinarySerializer (recommended)

# config/packages/messenger.yaml
framework:
  messenger:
    transports:
      nats_transport:
        dsn: 'nats-jetstream://localhost:4222/my-stream/my-topic'
        serializer: 'IDCT\NatsMessenger\Serializer\IgbinarySerializer'
        options:
          consumer: 'my-consumer'

The IDCT\NatsMessenger\Serializer\IgbinarySerializer service the serializer: key refers to is registered by IDCTNatsMessengerBundle (see Installation), so the configuration above works as-is. Only when the bundle is not enabled do you need to register it yourself, next to the transport factory:

# config/services.yaml (only without the bundle)
services:
    IDCT\NatsMessenger\Serializer\IgbinarySerializer: ~

Like Symfony's own serializers, it leaves out the stamps Symfony marks as not to be sent (NonSendableStampInterface), such as the AckStamp a worker adds to the message it handles, so a message Symfony re-sends for a retry or to the failure transport is serialized without them. Any serializer extending AbstractEnveloperSerializer does the same.

Tested by: createTransport_UsesProvidedSerializer, serialize_WithValidEnvelope_ReturnsSerializedString, decode_WithValidEncodedEnvelope_ReturnsEnvelope, testConstructorWithoutIgbinaryDoesNotCrash, theReadmesIgbinaryExampleGivesTheTransportTheIgbinarySerializer, theReadmesSetupWithoutTheBundleGivesAWorkingTransport, aSignedMessageSurvivesTheSerializerUnderEveryIdTheBundleGivesIt, encode_EnvelopeAWorkerIsHandling_EncodesWithoutItsNonSendableStamps, encode_WithNonSendableStamps_LeavesThemOutAndKeepsTheRest, Behat scenario Failed message routed to Symfony failure transport via TERM with the igbinary serializer

Creating Custom Serializers

You can create your own serializer by extending AbstractEnveloperSerializer:

use IDCT\NatsMessenger\Serializer\AbstractEnveloperSerializer;
use Symfony\Component\Messenger\Envelope;

class MyCustomSerializer extends AbstractEnveloperSerializer
{
    protected function serialize(Envelope $envelope): string
    {
        // Your custom serialization logic
        return serialize($envelope);
    }

    protected function deserialize(string $data): mixed
    {
        // Your custom deserialization logic
        return unserialize($data);
    }
}

Tested by: readmeCustomSerializerExample_EncodeDecode_RoundTrips, readmeCustomSerializerExample_DecodeInvalidBody_ThrowsException - the exact code above is compiled and exercised via ReadmeExampleSerializer in the unit tests.

For reference implementations, see:

src/Serializer/IgbinarySerializer.php - Binary serialization
src/Serializer/AbstractEnveloperSerializer.php - Base class

4. Send Messages

use App\Message\MyAsyncMessage;
use Symfony\Component\Messenger\MessageBusInterface;

class MyController
{
    public function __construct(private MessageBusInterface $bus) {}

    public function send(): void
    {
        $this->bus->dispatch(new MyAsyncMessage('Hello NATS!'));
    }
}

Tested by: testSendPublishesEncodedBodyWithoutHeaders, testSendUsesPublishWithHeadersWhenHeadersArePresent, Behat scenario Complete message flow - send, check stats, consume, verify

5. Handle Messages

use App\Message\MyAsyncMessage;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
class MyAsyncMessageHandler
{
    public function __invoke(MyAsyncMessage $message): void
    {
        echo "Processing: " . $message->getText();
    }
}

Tested by: Behat scenarios Complete message flow - send, check stats, consume, verify, Send and consume messages with a custom consumer name, and High-volume message processing with file output verification - handlers are exercised through real messenger:consume runs.

6. Consume Messages

symfony console messenger:consume nats_transport

Tested by: Behat scenarios Complete message flow - send, check stats, consume, verify, Send and consume messages with a custom consumer name, and Partial message consumption with multiple consumers - the Behat context runs messenger:consume as a Symfony CLI process.

Examples

examples/ holds runnable scripts, one per behaviour, that use the transport directly against the test server and print OK when what they show held:

Example	Shows
`send-and-consume.php`	`setup()`, sending, the queue depth, receiving and acknowledging
`duplicate-protection.php`	`DeduplicationIdStamp` and `deduplicate: true`: a message sent again is stored once
`request-timeout.php`	`request_timeout`, and sending again, with the same deduplication id, a message whose send timed out
`keepalive.php`	`keepalive()` from a SIGALRM handler, as `messenger:consume --keepalive` calls it
`connection-checks.php`	`ping_after_idle`, and a new dial once the connection is gone
`reconnect.php`	`reconnect: true`: a worker polls through a restart of its server, and `auto_setup` recreates the memory stream the restart lost
`cluster-failover.php`	`reconnect: true` on a cluster: a worker carries on through another server when the one it is connected to stops
`stream-placement.php`	`stream_placement_tags` and `stream_placement_cluster` on a cluster, and staging a move with a replica change

composer nats:start runs a three-server JetStream cluster for the last two next to the single test servers; examples/README.md describes it, and Using a NATS Cluster what they mean for an application.

composer nats:start
composer examples                   # or one at a time: php examples/keepalive.php

Tested by: CI runs composer examples after the functional suite, on the latest and the oldest supported NATS.

Configuration Guide

DSN Format

nats-jetstream://[user:password@]host:port/stream-name/topic-name

Tested by: testBuildWithValidDsnReturnsConfiguration, testBuildWithoutPathThrowsException, testBuildWithoutTopicThrowsException, createTransport_WithValidDsn_ReturnsNatsTransportInstance

Examples:

# Default port (4222)
nats-jetstream://localhost/my-stream/my-topic

# Custom port
nats-jetstream://localhost:5000/my-stream/my-topic

# With authentication
nats-jetstream://user:password@localhost:4222/my-stream/my-topic

# With query parameters
nats-jetstream://localhost/my-stream/my-topic?consumer=worker&batching=10

# TLS transport scheme
nats-jetstream+tls://localhost:4222/my-stream/my-topic

Tested by: testReadmeDsnExamplesParseSuccessfully - each DSN above is parsed through the configuration builder via a dedicated data provider case.

Configuration Options

framework:
  messenger:
    transports:
      nats_transport:
        dsn: 'nats-jetstream://localhost:4222/my-stream/my-topic'
        options:
          # Consumer Configuration
          consumer: 'my-consumer'           # Consumer group name (default: 'client')

          # Performance Tuning
          batching: 5                       # Messages per batch (default: 1)
          max_batch_timeout: 1.0            # Timeout in seconds for batch fetching (default: 1)
          connection_timeout: 1.0           # Connection (dial) timeout in seconds (default: 1)
          ping_after_idle: 30               # Seconds the connection may go unused before the next
                                            # operation checks it with a PING and dials again if the
                                            # server does not answer (default: 30; 0 turns it off).
                                            # See "Losing the Connection".
          request_timeout: 10               # Seconds to wait for the server's reply to a publish,
                                            # an ack_sync ACK, setup() or a message count (default: 10)

          # Connection Resilience
          reconnect: false                  # Who recovers a lost connection (default: false)
                                            # false => the operation in flight fails and the transport dials
                                            #          again on the next one; a worker exits and is restarted
                                            # true  => the NATS client re-dials in the background and the
                                            #          operations wait for it (see Automatic Reconnect)
          max_reconnect_attempts: null      # Attempts per recovery before the client gives up
                                            # (null = the client's own default of 10). Positive integer.

          # Consumer Flow Control & Lifecycle
          max_ack_pending: 1000             # Max delivered-but-unacked messages outstanding
                                            # (null = server default). Primary flow-control lever.
          inactive_threshold: 300           # Seconds of no pull activity before NATS removes the
                                            # durable consumer (null = server default).
          replay_policy: 'instant'          # instant|original (default: null = server default 'instant')
                                            # ⚠️ Immutable once the durable consumer exists: NATS rejects
                                            # the change, so the transport skips it and the option only
                                            # takes effect on a freshly created consumer.

          # Stream Retention Policies
          stream_max_age: 86400             # Max message age in seconds (0 = unlimited, default: 0)
          stream_max_bytes: 1073741824      # Max storage size in bytes (null = unlimited)
          stream_max_messages: 1000000      # Max number of messages in the stream (null = unlimited)
          stream_max_messages_per_subject: 1000 # Max number of messages retained per subject (null = unlimited)
          stream_max_message_size: 1048576  # Max size of a single message in bytes.
                                            # Must be a positive integer, at most 2147483647.
                                            # Unset means "leave it to the server": a new stream gets
                                            # JetStream's unlimited default, and an existing stream
                                            # keeps whatever limit it already has. To lift a limit on
                                            # an existing stream, change it in NATS directly.
          stream_max_consumers: 10          # Max consumers allowed on the stream (null = unlimited)
                                            # ⚠️ NATS up to 2.11 refuses to change this on an existing
                                            # stream. When left unset the transport keeps whatever the
                                            # stream already has, so it never breaks an existing setup.

          # Stream Retention & De-duplication Behavior
          stream_retention: 'limits'        # limits|interest|workqueue (default: null = server default 'limits').
                                            # ⚠️ Immutable once the stream exists: NATS rejects changing it,
                                            # so a changed value is ignored on an existing stream - recreate
                                            # the stream to change retention (see note below).
          stream_discard: 'old'             # old|new - what to drop when a limit is hit (default: null = 'old')
          stream_duplicate_window: 120      # De-duplication window in seconds (null = server default).
                                            # Must be a positive integer; 0 is rejected because NATS
                                            # would substitute its own 2-minute default.
                                            # ⚠️ NATS forbids a window larger than a finite max age, so
                                            # lowering stream_max_age below the stream's current window
                                            # rewrites the window down to match (the server does the
                                            # same at creation time). Raising stream_max_age again does
                                            # not restore it - set this option explicitly to do that.
                                            # Must not exceed stream_max_age when that is set.

          # Storage Backend
          stream_storage: 'file'            # Storage type: 'file' or 'memory' (default: 'file')
                                            # ⚠️ Immutable once the stream exists (like stream_retention).
          stream_compression: 'none'        # none|s2 (default: null = server default 'none')
          stream_description: 'my stream'   # Human-readable stream description (null = unset).
                                            # At most 4096 characters.

          # Stream Access Policy (null = leave the server default untouched)
          stream_deny_delete: false         # Deny message deletion from the stream.
                                            # ⚠️ One-way: NATS can turn this on but never off, so once
                                            # a stream denies deletes, setting false here is ignored.
          stream_deny_purge: false          # Deny stream purge (one-way, exactly like deny_delete)
          stream_allow_direct: false        # Allow direct get access
          stream_allow_rollup_headers: false # Allow Nats-Rollup headers

          # High Availability
          stream_replicas: 1                # Number of replicas of a new stream (default: 1). Unset,
                                            # an existing stream keeps its count. More than 1 needs a
                                            # JetStream cluster (see "Using a NATS Cluster").
          stream_placement_cluster: null    # Pin the stream to a named JetStream cluster
                                            # (null = leave the stream's placement untouched)
          stream_placement_tags: null       # Only place the stream on servers carrying ALL of these
                                            # tags, e.g. ['ssd', 'eu-west'] (or 'ssd,eu-west' in a DSN)
                                            # (null = leave the stream's placement untouched)

          # Failure Handling Strategy
          retry_handler: 'symfony'          # symfony|nats (default: symfony)
                                            # symfony => TERM on failed/rejected message
                                            # nats    => NAK on failed/rejected message

          # NATS-native Redelivery Tuning (mainly relevant with retry_handler: nats)
          nak_delay: 0                      # Seconds to wait before NATS redelivers a NAK'd
                                            # message (default: 0 = immediate). Use to back off
                                            # instead of hot-looping a failing message.
          ack_wait: null                    # Seconds JetStream waits for an ACK before redelivering
                                            # (default: null = server default, ~30s). Raise it for
                                            # slow handlers to avoid premature redelivery.
          max_deliver: null                 # Max redelivery attempts before NATS gives up
                                            # (default: null = unlimited). Set it to stop a poison
                                            # message redelivering forever under retry_handler: nats.
          backoff: null                     # List of per-attempt delays in seconds, e.g. [1, 5, 30].
                                            # Pairs with max_deliver. (default: null)

          # Duplicate Protection
          deduplicate: false                # Give every message an id JetStream deduplicates it by
                                            # within stream_duplicate_window (default: false)

          # Acknowledgement Mode
          ack_sync: false                   # Wait for server confirmation of each ACK (default: false)
                                            # false => fire-and-forget ACK (lower latency)
                                            # true  => JetStream double-ack; a dropped ACK cannot
                                            #          silently cause redelivery, at a latency cost

          # Scheduled / Delayed Messages (requires NATS >= 2.12)
          scheduled_messages: false         # Enable scheduled message support (default: false)
                                            # When enabled, Symfony DelayStamp triggers NATS
                                            # scheduled message delivery via Nats-Schedule headers

          # Provisioning
          auto_setup: false                 # Provision the stream/consumer on first send/get (default: false).
                                            # When false (the default), run `messenger:setup-transports`
                                            # explicitly. When true, setup() runs once, lazily, on first use.

          # TLS Configuration
          tls_required: false               # Force TLS for NATS connection (default: false)
          tls_handshake_first: false        # Use TLS-first handshake mode (default: false)
          tls_ca_file: null                 # Path to CA certificate file
          tls_cert_file: null               # Path to client certificate file
          tls_key_file: null                # Path to client private key
          tls_key_passphrase: null          # Passphrase for encrypted private key
          tls_peer_name: null               # Override TLS peer name for certificate validation
          tls_verify_peer: true             # Verify TLS peer certificate (default: true). Only
                                            # false, 0, no or off turns it off; any other value,
                                            # an empty one included, keeps it on

          # Additional Authentication
          token: null                       # NATS token authentication
          username: null                    # Overrides DSN username if provided
          password: null                    # Overrides DSN password if provided
          jwt: null                         # JWT authentication value
          nkey: null                        # NKey public value

Tested by: testReadmeConfigurationOptionsAreAccepted (all options above), testReadmeBatchingExamplesAreAccepted, testReadmeTimeoutExamplesAreAccepted, testReadmeStreamRetentionExamplesAreAccepted, testBuildAcceptsAndNormalizesNewStreamAndConsumerOptions, testSetupPassesNewStreamPolicyOptions, testSetupPassesNewConsumerOptions, testBuildWithTlsAndAuthOptionsPropagatesToNatsOptions, testTlsVerifyPeerStaysOnUnlessExplicitlyDisabled, testTlsVerifyPeerInTheDsnStaysOnUnlessExplicitlyDisabled, testBuildAcceptsStreamPlacementOptions, testBuildWithReconnectOptionsPropagatesToNatsOptions

Retry Handler Behavior

retry_handler: symfony (default) sends TERM when a message fails during transport decoding or is rejected. Symfony's retry/failure transport then handles redelivery.
retry_handler: nats sends NAK when a message fails during transport decoding or is rejected, so NATS redelivers the message itself.

Symfony's retry delays need scheduled_messages: true. Symfony's retry strategy waits between attempts (1, 2 and 4 seconds with the framework's default strategy) by re-sending the failed message with a DelayStamp. The transport applies a DelayStamp only when scheduled_messages is enabled, which needs NATS 2.12 or later, and otherwise publishes the message at once. So with the defaults, each retry runs right after the failure, all of them are used up within milliseconds, and nothing reports that the delays were dropped. To keep the delays:

on NATS 2.12 or later, enable scheduled_messages (see Delayed / Scheduled Messages);
on an older server, let NATS space out the attempts instead: retry_handler: nats with nak_delay or backoff, bounded by max_deliver, and with max_retries: 0 (see below).

Tested by: testSendWithDelayStampButScheduledMessagesDisabledPublishesNormally, testSendWithDelayStampPublishesToDelayedSubjectWithScheduleHeaders

When NATS manages redelivery (retry_handler: nats), tune it with nak_delay, ack_wait, max_deliver, and backoff:

nak_delay delays each NAK so a failing message backs off instead of redelivering immediately (a hot loop).
max_deliver caps redeliveries. ⚠️ Without it, retry_handler: nats redelivers a permanently-failing ("poison") message forever - set max_deliver in production.
backoff sets an escalating per-attempt delay schedule (e.g. [1, 5, 30] seconds); pair it with max_deliver greater than the number of backoff steps.
ack_wait is how long JetStream waits for an ACK before considering a delivery failed and redelivering - raise it for handlers that legitimately take a while.

Symfony's retry strategy is ignored with retry_handler: nats. Symfony's retry listener runs for every transport, and FrameworkBundle gives each one a retry strategy (max_retries: 3 by default). In nats mode the transport does not publish the copy that retry sends, because NATS redelivers the original after the NAK, so max_retries, delay and multiplier have no effect and max_deliver bounds the attempts. Set max_retries: 0 on the transport anyway: with retries left on, Symfony still logs each failure as a retry it never makes ("Sending for retry #1") and never treats the message as failed for good.

A failure transport does not combine well with nats mode. Symfony sends a message there each time it treats a failure as final: with max_retries: 0 that is every failed delivery, so a message that keeps failing lands there up to max_deliver times, and with retries left on it never lands there. For a transport in nats mode, rely on NATS's own handling of a message that keeps failing instead (max_deliver, and the JetStream MAX_DELIVERIES advisory).

framework:
  messenger:
    transports:
      nats_transport:
        dsn: 'nats-jetstream://localhost:4222/my-stream/my-topic?retry_handler=nats&max_deliver=5&nak_delay=10'
        retry_strategy:
          max_retries: 0

Tested by: testRejectUsesTermByDefault, testRejectUsesNakWhenRetryHandlerIsNats, testHandleFailedDeliveryUsesNakWithDelayWhenConfigured, testSetupAppliesConsumerRetryTuning, testBuildAcceptsNatsRetryTuningOptions, testBuildUsesRetryHandlerFromQuery, testNatsModeDoesNotPublishTheCopySymfonysRetrySends, testWorkerRetryInNatsModeNaksTheOriginalWithoutPublishingACopy, Behat scenarios nats_nak.feature, nats_term.feature and Symfony's retry strategy is ignored with the NATS retry handler

Long-Running Handlers (`--keepalive`)

NATS redelivers a message that is not acknowledged within the consumer's ack_wait (30 seconds unless set), possibly to another worker, even while a handler is still working on it. messenger:consume --keepalive prevents that: every few seconds (5 by default, --keepalive=N to change it) Symfony has the transport tell NATS that the message is still in progress, which restarts its ack_wait.

symfony console messenger:consume nats_transport --keepalive=10

Symfony sends the keepalive from a SIGALRM signal handler, which needs the pcntl extension and can interrupt the handler at any point. The NATS client is asynchronous, and PHP does not let it wait inside a signal handler, so the transport queues the in-progress acknowledgement there, and it goes out the next time the event loop runs:

A handler that waits on asynchronous work - any call to NATS through this library, such as dispatching a message to a NATS transport, or amphp-based I/O - lets it out at once, so the keepalive works as intended.
A handler that only blocks - plain PHP work, or blocking calls such as PDO queries or curl requests - lets it out only once it returns, too late to restart the timer. For such a handler, set ack_wait longer than its longest run instead.

Keep the keepalive interval well below ack_wait, so a keepalive that goes out a little late still arrives in time.

Tested by: testKeepaliveSendsInProgressForReplyToken, testKeepaliveFromASignalHandlerQueuesTheAcknowledgementInsteadOfWaiting, testKeepaliveDoesNotWaitForTheAcknowledgement, testKeepaliveWithoutAConnectionSendsNothingAndDoesNotDial, testKeepaliveWhoseAcknowledgementFailsRaisesNoUnhandledError, Behat scenario A message handled for longer than ack_wait is not redelivered while keepalive runs

Important: Consumer Strategies

This is critical to understand before setting up multiple transport instances:

⚠️ Strategy A: Same Consumer, Batching = 1

Use when: Multiple instances should cooperate on the same consumer

# All instances use the same consumer with batching=1
transports:
  nats_worker_1:
    dsn: 'nats-jetstream://localhost/my-stream/my-topic'
    options:
      consumer: 'shared-consumer'  # Same consumer name
      batching: 1                  # MUST be 1 for shared consumers

  nats_worker_2:
    dsn: 'nats-jetstream://localhost/my-stream/my-topic'
    options:
      consumer: 'shared-consumer'  # Same consumer name
      batching: 1                  # MUST be 1 for shared consumers

Why batching must be 1:

With explicit acknowledge (ACK) mode, only messages that are explicitly acknowledged are considered processed
Multiple instances sharing the same consumer need to ACK individually
Batching > 1 with multiple instances causes delivery conflicts
Each instance should fetch and ACK one message at a time

Benefits:

Automatic load balancing across instances
NATS handles message distribution
Guaranteed single processing per message

Tested by: testReadmeBatchingExamplesAreAccepted (batching=1), Behat scenario Partial message consumption with multiple consumers

✅ Strategy B: Different Consumers, Any Batching

Use when: Each instance needs independent message processing (duplicates allowed)

# Each instance uses a different consumer
transports:
  nats_worker_1:
    dsn: 'nats-jetstream://localhost/my-stream/my-topic'
    options:
      consumer: 'worker-1-consumer'   # Unique consumer per instance
      batching: 10                    # Can use any batching

  nats_worker_2:
    dsn: 'nats-jetstream://localhost/my-stream/my-topic'
    options:
      consumer: 'worker-2-consumer'   # Unique consumer per instance
      batching: 10                    # Can use any batching

Why this works:

Each consumer maintains its own state
All messages are delivered to all consumers independently
Each instance can use higher batching for better throughput
Duplicate processing is expected (fan-out pattern)

Use cases:

Event broadcasting to multiple systems
Multiple independent processors
Audit logging / event replay

Tested by: testReadmeBatchingExamplesAreAccepted (batching=10), Behat scenario Partial message consumption with multiple consumers

Batching & Timeouts

Batching Explained

Higher batching: Better throughput, slightly higher latency
Lower batching: Lower latency, slightly reduced throughput
Optimal batching: Depends on message size and processing time

options:
  batching: 1        # Fetch 1 message at a time (low latency)
  batching: 5        # Fetch 5 messages (balanced)
  batching: 20       # Fetch 20 messages (high throughput)

Tested by: testReadmeBatchingExamplesAreAccepted - values 1, 5, 10, 20, 50 are all verified.

Batch Timeout

Controls how long to wait for a batch to fill:

options:
  batching: 10
  max_batch_timeout: 0.5  # Wait max 0.5s for batch to fill
                          # Returns early if timeout reached

Tested by: testReadmeTimeoutExamplesAreAccepted - values 0.5, 1.0, 2.0 are verified. Behat scenarios nats_batching.feature.

Example scenarios:

If you set batching: 10 and max_batch_timeout: 0.5
If 10 messages arrive quickly, all are fetched immediately
If only 3 messages arrive in 0.5s, return those 3

Connection Timeout

Controls the timeout for establishing the NATS connection: the first dial, and each new dial after the connection was lost (see Losing the Connection):

options:
  connection_timeout: 2.0  # Connection (dial) timeout in seconds

Tested by: testReadmeTimeoutExamplesAreAccepted (1.0, 2.0, 3.0), testBuildWithConnectionTimeoutPropagatesMs

Purpose:

Sets the timeout for the initial TCP/TLS dial and handshake when connecting to NATS
Also bounds the PING that checks a connection before it is used again, after it sat idle or after an operation failed on it (ping_after_idle), unless request_timeout is shorter: a server that does not answer within it is treated as gone
Does not bound how long an operation waits for the server's reply, which is request_timeout; the batch fetch is bounded separately by max_batch_timeout
Bounds each dial attempt: a dial that fails is tried three times, with pauses of 2 and 4 seconds in between
Lower values fail faster only when an attempt hangs, as with an unreachable host (about 6 seconds plus three times the value in all); a refused connection fails after about 6 seconds whatever the value
Higher values tolerate slower connection establishment

When to adjust:

Increase for high-latency networks or geographically distant NATS servers
Decrease for faster failure detection in local environments
Default of 1 second works well for most local/regional deployments

Request Timeout

Controls how long an operation waits for the server's reply: the publish acknowledgement of send(), the confirmation of an ACK with ack_sync, and the JetStream API calls of setup() and getMessageCount():

options:
  request_timeout: 30  # seconds (default: 10)

Tested by: testRequestTimeoutDefaultsToTenSeconds, testBuildWithRequestTimeoutPropagatesMs, testBuildWithInvalidRequestTimeoutThrowsException, testReadmeTimeoutExamplesAreAccepted (10, 30)

A reply that does not arrive in time fails the operation with the client's TimeoutException. For send() that does not prove the message was not stored: the server may have stored it and only its acknowledgement was late, so dispatching the message again can store it twice, unless it carries a deduplication id (see Duplicate Protection). Raise the value for a slow or distant server, or for a stream whose writes can take long, such as a replicated one under load; lower it to fail faster. The wait of a pull in get() is bounded by max_batch_timeout instead (the client gives up a second later on a pull the server does not answer at all), and the dial by connection_timeout.

Duplicate Protection

A send() that timed out may still have stored the message (see Request Timeout), so dispatching it again can store it twice. JetStream drops a message whose Nats-Msg-Id it already stored within the stream's duplicate window (stream_duplicate_window, 2 minutes unless set), and the transport fills that header from a DeduplicationIdStamp:

use IDCT\NatsMessenger\Stamp\DeduplicationIdStamp;

// Dispatched again with the same id after a send() that failed, the message is stored once.
$bus->dispatch(new OrderPlaced($orderId), [new DeduplicationIdStamp('order-placed-' . $orderId)]);

With deduplicate: true the transport gives every message an id of its own on its first send. The id is sent with the message, so a copy sent again keeps it, which covers what Symfony sends again itself: a worker that stopped after a retry or failure-transport send that timed out, the server having stored the copy, gets the original redelivered by NATS after ack_wait and sends the same copy again, which JetStream drops. For that, stream_duplicate_window has to be longer than ack_wait plus the handler's run. Each of Symfony's retries, and the copy for a failure transport, gets an id of its own, so none of them is taken for the original. An id you add yourself, as above, is used with the option off as well, and covers a message your application dispatches again as a new envelope, which would otherwise get a new id.

The header holds the id together with the transport's subject and the retry count, and on a failure-transport copy the transport the message failed on. JetStream deduplicates per stream whatever the subject, and Symfony hands each transport a message is routed to the envelope the one before returned, id included, so transports that share a stream (see Multi-Subject Streams) each keep their copy.

options:
  deduplicate: true   # default: false

Tested by: testSendWithDeduplicationStampsTheEnvelopeAndPublishesItsMessageId, testSendKeepsTheDeduplicationIdTheEnvelopeCarries, testSendGivesEachRetryAndTheFailureTransportCopyAMessageIdOfItsOwn, testSendWithoutDeduplicationSendsNoMessageIdAndAddsNoStamp, testSendUsesAnApplicationDeduplicationIdWithTheOptionOff, testSendDelayedMessageWithDeduplicationPublishesItsMessageId, testDeduplicateIsOffByDefaultAndCanBeEnabled, testSendGivesTheCopiesForTwoTransportsOnOneStreamMessageIdsOfTheirOwn, testSendGivesTheFailureCopiesOfTwoTransportsMessageIdsOfTheirOwn, Behat scenarios A message dispatched again with the same deduplication id is stored once, Distinct messages are all stored with the deduplicate option and A message routed to two deduplicating transports on one stream is stored for each

Losing the Connection

Unless reconnect is on, the transport does not reconnect in the background, and nothing in Symfony calls close(). When the connection has been lost, the next operation dials again instead:

A connection is lost when the NATS server restarts, when the network drops it, or when the server closes it because the client stopped answering its pings. A PHP process answers them only while it is inside a transport call, so a server with default settings drops a connection that has been idle for about 6 to 8 minutes; a load balancer or NAT gateway may drop an idle connection sooner.
The operation that runs into the lost connection fails. For send() that means the outcome is unknown: the server may or may not have stored the message. Once the client reports the connection closed, the next operation opens a new one, on a cluster possibly to another server (see Using a NATS Cluster). A consumer worker exits on the failed get() and its supervisor starts a new one; a process that sends for longer than one request (a web app in worker mode such as FrankenPHP or RoadRunner, a daemon, a handler that dispatches) keeps working from the next operation on.
With auto_setup, the new connection is verified in the same call that dials it, and a dial that fails surfaces as the connection error itself.

A connection that was dropped while it sat idle would make the next operation the one that fails. To avoid that, an operation that finds the connection unused for longer than ping_after_idle (30 seconds by default) first checks it with one PING. If the server does not answer within connection_timeout (or request_timeout, when that is shorter), the transport closes that connection and runs the operation on a new one, so the first message after a quiet period goes out instead of failing. The PING costs one round trip, and only after a quiet period. A lower value also catches a server restart between two operations further apart than it, at the cost of that round trip more often. The same check follows an operation that failed on the connection - a JetStream error reply does not count, since the server answered it - because the client can keep reporting a dead connection open, and a send to a connection that silently stopped delivering only times out. A pull from such a connection ends empty, which reads as no messages rather than a failure, but the server never answered it: a server that is there ends every pull itself, after max_batch_timeout at the latest. So the check follows such a pull too, and the pull after it runs on a new connection. ping_after_idle: 0 turns both checks off. keepalive() never checks: Symfony calls it from a signal handler.

options:
  ping_after_idle: 30   # seconds; 0 turns the check off

Tested by: testOperationAfterTheClientClosedDialsAgain, testAutoSetupVerifiesTheNewConnectionBeforeTheSameCallPublishes, testFailedDialAfterTheClientClosedSurfacesAsTheConnectionErrorAndIsRetried, testIdleConnectionIsCheckedWithAPingAndKeptWhenTheServerAnswers, testIdleConnectionThatDoesNotAnswerThePingIsReplacedBeforeTheOperation, testPingUnansweredWithinTheConnectionTimeoutReplacesTheConnection, testRecentlyUsedConnectionIsNotPinged, testPingAfterIdleZeroTurnsTheCheckOff, testKeepaliveNeitherPingsNorDials, testOperationThatFailedOnTheConnectionMakesTheNextOneCheckItFirst, testJetStreamReplyDoesNotMakeTheNextOperationCheckTheConnection, testPullTheServerDidNotAnswerMakesTheNextOperationCheckTheConnection, testOnlyAPullTheServerDidNotAnswerMakesTheNextOneCheckTheConnection, testIdleTimeCountsFromTheEndOfAnEmptyPull, testBuildWithInvalidPingAfterIdleThrowsException, Behat scenarios A message sent after the server dropped the idle connection goes out and Without the check the message sent after the server dropped the connection fails

Subscription Limits

The transport holds up to two subscriptions on its connection: the reply inbox of its requests, which the first send(), ack_sync acknowledgement, setup() or getMessageCount() on a connection subscribes, and the reply inbox of the pull in progress. NATS caps the subscriptions a connection may hold at the smallest of the server's max_subscriptions, the account's limit and the user's. With a cap of one, the server refuses the pull's inbox with maximum subscriptions exceeded, and every pull after the first request on that connection fails. get() then throws Symfony's TransportException, with the client's SubscriptionLimitException as its previous exception:

Cannot pull from NATS stream "orders", consumer "worker": the subscription limit was exceeded. This transport requires capacity for at least two subscriptions per connection. Check server, account and user limits.

Like any exception from get(), it ends a worker's run. Raise the limit to at least two. The operation after the failure checks the connection first, as after any other failure on it (see Losing the Connection): the server sends the same error when an account's limit is lowered below what the connection holds, and closes the connection right behind it.

Tested by: testAtALimitOfOneThePullFailsWithATransportExceptionNamingTheLimit, testAtALimitOfOneAPullWhoseRefusalTheHeartbeatReadsFailsTheSameWay, testAtALimitOfTwoPullsDeliverAndAcknowledge, testALimitErrorTheServerClosesTheConnectionBehindKeepsTheCheckBeforeTheNextOperation, testALimitErrorTheServerClosesTheConnectionBehindLeavesAClosedClientThatIsReplacedWithoutACheck, testPullRefusedAtTheSubscriptionLimitFailsWithATransportExceptionAndTheNextOperationChecksTheConnection, testAutoSetupRetriedPullRefusedAtTheSubscriptionLimitFailsWithATransportException, testPullThatFailedOnTheConnectionOtherwisePropagatesAsItIs

Automatic Reconnect

Who recovers a lost connection is a choice between two modes.

reconnect: false (the default): Symfony owns recovery. The NATS client runs with reconnect off. The operation that runs into the lost connection fails at once, the next one dials again, a consumer worker exits on the failed get() and its supervisor starts a new one, and Symfony's retries and failure transport take care of the message. See Losing the Connection; nothing below applies.

reconnect: true: the client owns recovery. The NATS client re-dials in the background as soon as it notices the loss, with exponential backoff (100 ms doubling up to 10 s, with jitter) and up to max_reconnect_attempts times per recovery (the client's default is 10), replays its subscriptions and flushes the frames it buffered meanwhile. The transport steps back from recovering the connection itself:

options:
  reconnect: true             # the client recovers a lost connection (default: false)
  max_reconnect_attempts: 20  # attempts per recovery (default: null = the client's default of 10)

In a DSN: ?reconnect=true&max_reconnect_attempts=20.

A client in the middle of its recovery is left alone: no PING, no dial of the transport's own. The operation waits for the recovery within its own budget: a request (send(), an ack_sync ACK, setup(), getMessageCount()) within request_timeout, a pull within max_batch_timeout plus one second, after which it ends empty. The budget is per request, not per method: setup() makes three or more requests, a publish retries a missing responder three times on its own, and the first dial of a process is not bounded by it at all.
The PING check of Losing the Connection still runs, and when the PING fails, what the client made of the connection decides. A client that is recovering, or that already recovered (its reconnect count moved, so the PONG belonged to the lost connection), keeps its connection. A client that still reports the same connection open is replaced, as without the option: nothing in the client recovers a connection it does not know to be lost before its heartbeat has gone unanswered three times, about 90 s at the client's defaults, and the heartbeat only runs while an operation holds the event loop. A client that gave up is Closed and dialled again. A PHP process runs the event loop only inside a transport call, so the client notices a lost connection when an operation writes to it or when this check runs: a ping_after_idle shorter than the quiet periods between operations makes the first operation after one check first and go out on the recovered connection, where without the check it writes into the dead socket, waits request_timeout and fails, and only the next operation goes out.
With auto_setup, provisioning is verified again once the client reconnected, because the stream may have gone with the server (a memory stream over a restart, a failover to a server without it). A send begun while the client was recovering can still land on a server without the stream; when its publish fails with the client's 503 "No JetStream responder", setup() runs once and the same frame is published once more, then a second failure propagates. Nothing answered the first publish, so the second cannot duplicate it.
When the recovery gives up, the client is Closed and the acknowledgements it buffered during the outage are discarded; the messages come back after ack_wait, counting one delivery against max_deliver. The client reports the discard at error level through the logger the bundle gives it (Reconnect exhausted: N bytes of buffered publishes were discarded, under Monolog's messenger channel). The next operation dials again, and with the option on that dial runs the client's attempt loop inside the operation. Exhaustion propagates as the client's ConnectionException ("Reconnect attempts exhausted") rather than being flattened into an empty queue, so a worker ends and its supervisor restarts it. The same loop runs for a failed first dial, so messenger:setup-transports against a server that is down fails only after it.
What a worker gets: it polls through an outage shorter than one recovery cycle without exiting, every pull ending empty after max_batch_timeout plus a second. max_reconnect_attempts sizes the cycle in attempts, not in seconds: the client's default of 10 gives about 109 s against a server that refuses connections, where a refused dial costs about 6 s. An ack_sync acknowledgement that runs into the outage still fails after request_timeout, which ends the worker's run as any exception from ack() does: synchronous acknowledgement keeps its meaning.
An ACK, NAK or TERM written during a recovery is buffered by the client, and ack() or reject() returns at once. That holds too when its own write is the first to notice the lost connection while the client still reports it open: the write starts the recovery, and the frame goes out on the connection the recovery makes. Before client 2.27.1, which the transport requires since 5.7.1, such a write ran the recovery inline, so one ack() or reject() could wait for a whole attempt loop (client issue #244). keepalive() only queues its frame, as it must inside a signal handler, and never waits, PINGs or dials.
A server that stops gracefully ends every pull in flight with 409 Server Shutdown, and a consumer whose leader moved ends them with 409 Leadership Change. Both are empty batches, in either mode, and no longer read as a missing consumer; after a shutdown the next operation checks the connection first.

Measured on nats-server 2.14.2 with client 2.27.0 and PHP 8.5, request_timeout 10, connection_timeout 1, max_batch_timeout 1, auto_setup with a memory stream, the server stopped with docker stop for about 53 s and the same transport object throughout. A refused dial on localhost costs about 6 s:

Operation while the server is down	`max_reconnect_attempts` 10 (the client's default)	`max_reconnect_attempts` 2
`send()` in flight at the stop	10.0 s, `TimeoutException`	10.0 s, `TimeoutException`
next `send()`	10.0 s, `TimeoutException` (the client is still recovering)	2.4 s, `Reconnect attempts exhausted` (the recovery gave up inside the wait)
`get()`	2.0 s, empty	18.4 s, `Reconnect attempts exhausted`
plain `ack()`	0.00 s, buffered	18.4 s, `Reconnect attempts exhausted`
`getMessageCount()`	20.0 s (two requests), 0	18.4 s, 0
`send()` after those	10.0 s, `TimeoutException`	18.4 s, `Reconnect attempts exhausted`
first `send()` once the server is back	2.8 s, stored; the memory stream recreated	0.13 s, stored; the memory stream recreated

With 10 attempts the recovery cycle outlasted the outage, so every operation waited within its own budget and none dialled; once the server was back, the first send waited for the client's next attempt, at the 10 s cap of the backoff by then. With 2 attempts the client gave up about 12 s into the outage, and from then on every operation dialled inline, one dial and two attempts, and failed with the client's exception. Before this mode existed in the transport, the option as first proposed ran the whole attempt loop inside every operation during an outage, 109 s each at the defaults. The figures are measurements, not bounds: the option bounds attempts per recovery, not time, and a process can run any number of recovery cycles over its life.

When to enable: long-running workers where a restart is expensive, against a NATS cluster whose nodes restart or fail over (see Losing a Server). Not for web requests that dispatch: a request handler stalls for a whole attempt loop when it runs into an outage once the client gave up. Leave it off to rely on the process supervisor and Symfony's retries, and to have messenger:setup-transports fail fast when NATS is unreachable.

Tested by: testBuildTurnsClientReconnectAndPedanticModeOff, testBuildWithReconnectOptionsPropagatesToNatsOptions, testBuildWithReconnectFromDsnQueryString, testBuildKeepsTheClientReconnectAttemptDefaultWhenOnlyReconnectIsEnabled, testBuildWithInvalidMaxReconnectAttemptsThrowsException, testBuildPassesTheLoggerToTheClientAndNoneByDefault, testReconnectModeLeavesAClientInTheMiddleOfItsRecoveryAlone, testReconnectModeReadsAFailedPingAgainstTheClientsRecovery, testReconnectModeTakesARecoveredConnectionAsChecked, testWithoutReconnectAFailedPingReplacesTheConnectionAndTheReconnectCountIsNeverRead, testReconnectModeVerifiesProvisioningAgainAfterTheClientReconnected, testReconnectModeReprovisionsOnceWhenTheFirstPublishAfterTheReconnectFindsNoResponder, testReconnectModeDoesNotRetryAPublishOutsideTheNoResponderCase, testAPullTheServerEndedForItsShutdownIsEmptyAndTheNextOperationChecksTheConnection, testAPullEndedByALeadershipChangeIsEmptyAndTheNextOneDoesNotCheckTheConnection, testA409ReportingAConfigurationErrorStaysVisible, theBundleHandsTheApplicationsLoggerToTheFactory, testAnAcknowledgementWhoseWriteFindsTheConnectionLostReturnsAtOnceAndTheRecoverySendsIt, the real-client tests in tests/unit/ReconnectModeTest.php, and the Behat scenarios A worker keeps consuming across a server restart and A message sent after a server restart goes out on the reconnected connection and the memory stream is recreated; the examples reconnect.php and cluster-failover.php (a failover in a three-server cluster)

Stream Configuration

Retention Policies

Control how long messages are kept in the stream:

options:
  # By age (24 hours)
  stream_max_age: 86400

  # By total size (1GB)
  stream_max_bytes: 1073741824

  # By total message count across the entire stream (NATS: max_msgs)
  stream_max_messages: 1000000

  # By message count per individual subject (NATS: max_msgs_per_subject)
  stream_max_messages_per_subject: 1000

  # Unlimited (default)
  stream_max_age: 0
  stream_max_bytes: null
  stream_max_messages: null
  stream_max_messages_per_subject: null

Tested by: testReadmeStreamRetentionExamplesAreAccepted - all retention options above are verified. Behat scenarios nats_stream_limits.feature.

Note: stream_max_messages limits the total number of messages stored in the stream (maps to NATS max_msgs), while stream_max_messages_per_subject limits messages retained per individual subject (maps to NATS max_msgs_per_subject). The per-subject limit is especially useful with multi-subject streams to prevent one high-volume subject from dominating retention.

High Availability

options:
  # Single replica (no redundancy)
  stream_replicas: 1

  # 3 replicas (recommended for production)
  stream_replicas: 3

More than one replica needs a JetStream cluster: a standalone server refuses to create such a stream. Left unset, stream_replicas gives a new stream one replica and leaves the count of an existing stream alone, so a stream scaled up in NATS is not scaled back down by the next setup(). See Using a NATS Cluster.

Tested by: testReadmeStreamRetentionExamplesAreAccepted (replicas 1 and 3), testSetupPassesConfiguredStreamOptions, testSetupPreservesExistingReplicaCountWhenStreamReplicasNotConfigured, testSetupOverridesExistingReplicaCountWhenStreamReplicasExplicitlyConfigured

Stream Placement

In a JetStream cluster you can pin a stream to a named cluster and/or to the servers that carry specific tags (server_tags in the NATS server configuration). A stream is only placed on servers carrying all of the listed tags:

options:
  # Only on servers tagged both "ssd" and "eu-west"
  stream_placement_tags: ['ssd', 'eu-west']

  # In the "east" cluster of a super-cluster
  stream_placement_cluster: 'east'

  # Both at once
  stream_placement_cluster: 'east'
  stream_placement_tags: ['ssd']

In a DSN the tags are a comma-separated list: ?stream_placement_tags=ssd,eu-west.

Tested by: testBuildAcceptsStreamPlacementOptions, testBuildSplitsCommaSeparatedStreamPlacementTags, testBuildParsesStreamPlacementFromDsnQueryString, testSetupPassesStreamPlacementOnCreate, testSetupUpdateAppliesConfiguredStreamPlacementToAnExistingStream, testSetupUpdatePreservesServerStreamPlacementWhenNotConfigured, testSetupUpdateMergesConfiguredPlacementIntoTheLivePlacement, testSetupExplainsHowToStageAMoveAndScaleRejection, testAutoSetupRepeatsARejectedMoveAndScaleUpdateOnEverySend, Behat scenarios in nats_stream_placement.feature, and the example stream-placement.php on a three-server cluster

Both options default to null, which leaves the stream's placement untouched: a new stream gets the server's default placement and an existing stream keeps whatever placement it already has. This matters because a JetStream update that omits placement clears it, so the transport echoes the live value back whenever the options are unset.
On an existing stream each option replaces its own field of the live placement and nothing else: stream_placement_tags replaces the tag list and keeps a cluster pin set in NATS, and stream_placement_cluster replaces the cluster and keeps the live tags. Neither option can clear the other field, so to remove a placement, or one half of it, change it in NATS directly.
nats-server compares the stored placement with the new one field by field and treats any difference as a request to place the replicas again, even when the servers they are on already satisfy the new constraints, and even when only the order of the tags changed. The transport therefore keeps the live tag list as the server stores it whenever the configured tags name the same set, and compares the tags exactly: SSD and ssd match the same servers, but the server's unique_tag bypass (below) is case-sensitive, so a tag that differs only in case is sent as a change. The server accepts a new placement at once and setup() returns; the replica migration runs in the background (nats stream info shows the replicas as not current until it finishes).
In a super-cluster a placement that names tags but no cluster is resolved in the cluster the client running setup() is connected to, then in the clusters that one lists as alternates, on create and on a move alike. A tags-only change can therefore relocate the stream to another cluster. Set stream_placement_cluster to keep the stream where it is: the transport does not fill the cluster in from the live stream, because a pin it made up would show in nats stream info as configuration nobody asked for, and the options could never clear it again.
NATS refuses to change the placement and stream_replicas in the same update (err_code 10123), and setup() fails with the server's message and the staging steps. Stage the two changes: run messenger:setup-transports with only one of them (for example the placement options as they were and the new stream_replicas), wait until nats stream info shows every replica as current, then run it again with the other change. Running the same combined configuration twice is not staging. With auto_setup: true the transport runs setup() before the first send() or get() of a process, so every one of them repeats the rejected update and fails until the configuration is staged; nothing is published or consumed meanwhile. The transport does not split the update itself, because a setup() that applied one half and returned would count as done and the other half would never be applied.
Exclusion tags and the unique_tag bypass (NATS 2.12+). A tag prefixed with ! excludes the servers that carry it, for example !disk:hdd. The server strips the ! before it looks at the cluster's unique_tag setting, and any placement tag that starts with that configured prefix switches the uniqueness constraint off for the stream (nats-server's own comment: "disable uniqueness check if explicitly listed in tags"). So with unique_tag: "zone:", the tag !zone:ignore excludes nobody and only lifts the requirement that replicas sit in distinct zones, which lets a 3-replica stream be created on a cluster with fewer than 3 zones. A positive tag that every server carries and that starts with the prefix, such as zone:any added to each server's server_tags, has the same effect on any server version. On 2.11 and older the ! is not stripped, so !zone:ignore is a mandatory tag nothing matches and creation fails with tags not matched.
On a clustered server, tags that no server carries make stream creation fail with a JetStream error such as no suitable peers for placement. A standalone (non-clustered) server accepts and stores any placement without acting on it.

Taking an existing single-replica stream to three replicas on ssd servers, in two stages:

# Stage 1: the replica count alone. The placement options stay as they were (here unset, so the live
# placement is echoed back). Run messenger:setup-transports, then wait until `nats stream info` shows
# every replica as current.
options:
  stream_replicas: 3

# Stage 2: the placement. Run messenger:setup-transports again.
options:
  stream_replicas: 3
  stream_placement_tags: ['ssd']

Using a NATS Cluster

The transport runs against a NATS cluster as it runs against a single server. A cluster changes how a process finds the servers, what happens when the one it is connected to goes away, and how many servers a stream lives on. Two examples show it on the three-server JetStream cluster composer nats:start runs next to the test servers (examples/README.md describes it): cluster-failover.php stops the server a worker is connected to, and the worker carries on through another one; stream-placement.php places a stream's replicas on tagged servers and stages a move.

framework:
  messenger:
    transports:
      nats_transport:
        # One server of the cluster, or a name that resolves to several; the client learns the others from it.
        dsn: 'nats-jetstream+tls://nats.internal:4222/orders/orders.created'
        options:
          stream_replicas: 3            # the stream outlives the loss of one server
          reconnect: true               # a worker fails over to another server instead of exiting
          max_reconnect_attempts: 20
          tls_required: true            # TLS for the servers learned from the cluster as well
          tls_ca_file: '/etc/nats/ca.pem'

Tested by: testReadmeDsnExamplesParseSuccessfully[README: cluster transport], testReadmeClusterOptionsAreAccepted, the examples cluster-failover.php and stream-placement.php (CI runs both on the test cluster)

Finding the Servers

The DSN names one server. Once connected, the NATS client learns the other servers of the cluster from it (the connect_urls the server sends) and keeps them for its later dials. What it learns is the address each server advertises, so behind NAT, in Docker or in Kubernetes set client_advertise in every server's configuration to an address the application reaches that server on, as the test cluster does. A cluster that advertises nothing (no_advertise) leaves the client with the DSN's server alone.

Until it has connected once, a process knows only the server its DSN names, and so does every PHP-FPM request that dispatches, since each one builds its transport anew. One that starts while that server is down cannot connect, whichever servers of the cluster are up: it fails after one dial, about 6 s against a refused port, or with reconnect: true after the client's whole attempt loop over that one server. Point the DSN at a name that resolves to several servers of the cluster (the dial tries each address it resolves to in turn), or at a load balancer or a Kubernetes Service in front of the cluster.

Losing a Server

What a process does when the server it is connected to goes away depends on reconnect:

reconnect: false (the default): the operation that runs into the lost connection fails, and the next one dials the next server in the client's list, learned ones included; a server there that is down as well costs that operation a failed dial, and the one after it tries the next. A process that keeps running, such as a long-lived sender, carries on through another server. A consumer worker exits on the failed get(), though, and the worker its supervisor starts knows only the DSN's server again (see above).
reconnect: true: the client recovers the connection in the background on a server it learned, and a worker polls on. cluster-failover.php stops the server a worker is connected to halfway through its messages: the worker receives the rest through another server, without an exception.

Measured on the test cluster (nats-server 2.15.1, client 2.27.2, PHP 8.5), a stream with three replicas, the DSN on the server that leads it, that server stopped with docker stop, request_timeout 10 and max_reconnect_attempts 2 (for the pause, ping_after_idle 1 and 2 s), one transport object throughout:

	`reconnect: false`	`reconnect: true`
first `send()`, right after the stop	fails at once, `Reconnect is disabled`	10.0 s, `TimeoutException`; the client is on another server by then
first `send()`, after a pause longer than `ping_after_idle`	stored at once, through another server	stored at once, through another server
next `send()`	stored at once, through another server	stored at once, through another server
a new transport whose DSN names the stopped server	6.0 s, connection refused	18.4 s, `Reconnect attempts exhausted`

Right after the stop, the first send ran into the lost connection: with the option off the client reported the loss at once, with it on the send went into the dead connection before the client noticed and waited for a reply that never came. After a pause longer than ping_after_idle the check that runs first finds the connection gone, in either mode, and the send goes out through another server (see Losing the Connection). Whether a send that failed was stored is unknown either way (see Duplicate Protection).

Streams on a Cluster

Replicas. A stream with one replica lives on one server and is unavailable while that server is down. With stream_replicas: 3 it outlives the loss of one server, with 5 the loss of two. The durable consumer the transport creates takes the stream's replica count. Left unset, the option gives a new stream one replica and leaves an existing stream's count alone (see High Availability).
Placement. stream_placement_tags and stream_placement_cluster choose the servers, and in a super-cluster the cluster, that hold the replicas. A change of the replica count and of the placement has to be staged in two runs of messenger:setup-transports. See Stream Placement.
Provisioning. messenger:setup-transports can run through any server of the cluster. In a super-cluster a stream without stream_placement_cluster is created in the cluster that server belongs to, or in one of its alternates.
Leader changes. A pull that ends because the consumer's leader moved (409 Leadership Change) is an empty batch in either mode, and the worker pulls on.

TLS on a Cluster

The nats-jetstream+tls:// scheme turns TLS on for the server the DSN names. The servers the client learns from the cluster are dialled with TLS only when tls_required: true is set as well. Without it, a server that requires TLS refuses the client (TLS upgrade requested but no TLS context was configured at connect time), so no failover succeeds, in either mode, and a server that also accepts plain connections would receive the credentials unencrypted. Set both, as the example above does.

Each dial checks the server's certificate against the host it dialled: the DSN's host for the first dial, the address a learned server advertises (client_advertise) for the others. Issue each server's certificate for those names, or set tls_peer_name to a name every server's certificate carries.

Checked by hand on a two-server TLS cluster (nats-server 2.15.1, client 2.27.2): with the +tls scheme alone, every dial of the second server failed once the first one stopped, with reconnect off and on; with tls_required: true the next dial reached it. A server advertising a name its certificate does not carry was refused (Peer certificate CN=... did not match expected CN=...) until tls_peer_name named one it carries.

Testing

Unit Tests

# Install dependencies
composer install

# Run static analysis and the fast unit suite after every modification
composer test

# Run NATS
composer nats:start

# Run all unit tests with coverage (recommended)
composer test:unit

# Or run tests manually
./vendor/bin/phpunit

The target is to have at least 90% of code coverage.

What's tested:

DSN parsing and validation
Configuration option handling
Authentication support
Port configuration
Error handling
Interface compliance

Mutation Tests

The unit suite is mutation-tested with Infection to ensure the tests actually detect behavioral changes (not just execute lines):

# Requires a coverage driver (xdebug or pcov)
composer test:mutation:canary     # mutants that change nothing must all escape
composer test:mutation
composer test:mutation:check-log  # no mutant may be killed by the test harness instead of a test

Configuration lives in infection.json5. It enforces a minimum MSI of 90% and a minimum covered MSI of 95%; the suite scores 96.9% (1008 of 1040 mutants killed, with 100% mutation code coverage). docs/TESTS.md lists the 32 mutants that survive, with the reason for each: 27 are equivalent, three make no difference with the real client, and two differ about once in a million runs. CI runs all three on the PHP 8.5 job and every day. Scores published before, 100% among them, were not real: in Infection's mutant processes the client's final classes could not be doubled, so every mutant their tests covered counted as killed.

Functional Tests

Functional tests require a running NATS server with JetStream enabled:

# Set up functional test dependencies
composer test:functional:setup

# Start the NATS test servers in Docker, with the three-server cluster the examples use (generates the test
# TLS certificates first, which needs openssl)
composer nats:start

# Run functional tests
composer test:functional

# Stop NATS server
composer nats:stop

Manual approach:

# Generate the test TLS certificates the TLS and mTLS servers use (never committed)
bash tests/nats/certs/generate.sh

# Set up NATS in Docker (optional)
cd tests/nats
docker-compose up -d

# Run functional tests
cd ../functional
./vendor/bin/behat features/

# Stop NATS
cd ../nats
docker-compose down

What's tested:

Message publishing
Message consumption
Message acknowledgment
Consumer setup
Stream persistence

See also: tests/functional/README.md

Advanced Usage

Multiple Transports

Set up multiple independent transports for different use cases:

framework:
  messenger:
    transports:
      # High-priority, low-latency messages
      nats_fast:
        dsn: 'nats-jetstream://localhost/fast-stream/fast-topic'
        options:
          consumer: 'fast-consumer'
          batching: 1

      # Bulk processing, high throughput
      nats_bulk:
        dsn: 'nats-jetstream://localhost/bulk-stream/bulk-topic'
        options:
          consumer: 'bulk-consumer'
          batching: 50

      # Audit logging
      nats_audit:
        dsn: 'nats-jetstream://localhost/audit-stream/audit-topic'
        options:
          consumer: 'audit-consumer'
          stream_max_age: 2592000  # 30 days
          stream_replicas: 3

Tested by: testReadmeDsnExamplesParseSuccessfully[README: fast transport], testReadmeDsnExamplesParseSuccessfully[README: bulk transport], testReadmeDsnExamplesParseSuccessfully[README: audit transport], testReadmeAuditTransportOptionsAreAccepted, testReadmeBatchingExamplesAreAccepted

Multi-Subject Streams

Multiple transports can share the same NATS stream with different subjects. When messenger:setup-transports runs, each transport adds its subject to the existing stream rather than overwriting it:

framework:
  messenger:
    transports:
      # Both transports share the "events" stream
      nats_orders:
        dsn: 'nats-jetstream://localhost/events/orders'
        options:
          consumer: 'order-consumer'
          batching: 1
          stream_max_age: 300

      nats_payments:
        dsn: 'nats-jetstream://localhost/events/payments'
        options:
          consumer: 'payment-consumer'
          batching: 2

The events stream will have both orders and payments as subjects.

Tested by: testReadmeDsnExamplesParseSuccessfully[README: multi-subject orders], testReadmeDsnExamplesParseSuccessfully[README: multi-subject payments], testReadmeMultiSubjectOptionsAreAccepted, testSetupUpdatesExistingStreamMergesSubjectsAndPreservesServerConfig, Behat scenario Setup command merges subjects for transports sharing one stream

Note: When a stream already exists, setup reads the current JetStream configuration, merges in any new subjects, and then overlays the stream settings managed by this transport. Existing subjects are preserved, duplicate subjects are not added, and the existing storage backend is kept for already-created streams.

Provisioning the Stream & Consumer

The stream and its durable pull consumer must exist before messages flow. There are two ways to provision them.

Explicit (default, recommended for production). Provision once via the Symfony command:

framework:
  messenger:
    transports:
      nats_transport:
        dsn: 'nats-jetstream://localhost/my-stream/my-topic'
        options:
          consumer: 'my-consumer'

symfony console messenger:setup-transports nats_transport

This will:

Create the stream with the configured settings, or update an existing one (merging subjects and keeping the server's own fields)
Create the consumer with an explicit ACK policy
Verify that the consumer matches the configuration

Automatic (auto_setup). Set auto_setup=true to have the transport provision the stream and consumer itself, lazily, on the first send()/get(). Setup then runs once per transport instance, and again only if JetStream reports the stream or consumer as missing during a pull (for example after NATS removed an idle consumer that hit inactive_threshold), after close(), or after the transport dialled again because the connection was lost (see Losing the Connection). It defaults to false, so unless you opt in, provisioning stays explicit (no hidden stream/consumer creation on the hot path).

⚠️ Re-provisioning replays the stream. A durable consumer holds the acknowledgement state, so when one is lost that state is gone with it. The replacement is created with deliver_policy=all and will therefore redeliver every message the stream still retains, including messages that were already acknowledged. Under the default limits retention that means the whole stream, so make your handlers idempotent before combining auto_setup with a short inactive_threshold. workqueue retention does not have this problem, because acknowledged messages are removed from the stream. This is inherent to losing a durable consumer, not to auto_setup: a restarted worker rebuilds the consumer the same way.

framework:
  messenger:
    transports:
      nats_transport:
        dsn: 'nats-jetstream://localhost/my-stream/my-topic?auto_setup=true'

Note on retention & storage: stream_retention and stream_storage are fixed when the stream is created - NATS rejects changing either on an existing stream. On the update path the transport preserves the live server values, so changing these options on an already-created stream is a no-op. To change retention or storage, recreate the stream: delete it, then re-run messenger:setup-transports. With auto_setup=true a running worker recreates it on its next pull: deleting the stream deletes its consumers with it, so the pull fails with status 503 (nothing is left to answer it), which is one of the statuses that trigger re-provisioning.

Note on a server that stops: a pull the server ended for its own shutdown (409 Server Shutdown) or for a leadership change (409 Leadership Change) is an empty batch, in either mode, and not one of the statuses that trigger re-provisioning: with auto_setup the transport does not run setup() against a server that is going away. After a shutdown the next operation checks the connection first. Every other 409, Consumer Deleted among them, still triggers one re-provisioning attempt, and the 409s that report configuration (Consumer is push based, Exceeded MaxRequestBatch) still propagate.

Note on replay_policy: the replay policy of a durable consumer is fixed when the consumer is created. NATS rejects an update that changes it, in both directions: once a consumer exists as original, removing the option again does not restore instant either, because the server reads the omitted field as a change back to its default. The transport therefore always writes the existing consumer's own replay policy and applies the configured one only to a consumer that does not exist yet. Changing it on a running deployment is a no-op until you delete the consumer and let setup recreate it.

Note on the deny flags: stream_deny_delete and stream_deny_purge can be switched on but never off. NATS rejects an update that cancels either one, so on a stream that already denies deletes or purges the transport keeps the server's value and a false in your configuration is ignored. Recreate the stream to lift a deny. stream_allow_direct and stream_allow_rollup_headers are freely mutable in both directions.

Note on stream_max_consumers: NATS servers up to and including 2.11 also refuse to change max_consumers on an existing stream. The transport therefore writes it only when you set the option explicitly; when you leave it unset, the stream keeps whatever value it already has. Setting it on an already-created stream works on NATS 2.12 and newer, and is rejected by the server on older versions - recreate the stream to change it there.

Note on placement: stream_placement_cluster and stream_placement_tags follow the same rule as stream_max_consumers: they are written only when you set them, and unset options leave the stream's existing placement in place. A JetStream update that omits placement would clear it, so the transport echoes the live value back whenever the options are unset, and a set option replaces only its own field of the live placement. NATS rejects a placement change combined with a stream_replicas change on an existing stream, and with auto_setup every send() and get() repeats that rejected update until the two are staged. See Stream Placement for the steps.

Tested by: testSetupCreatesStreamAndConsumer, testSetupPassesConfiguredStreamOptions, testSetupPassesNewStreamPolicyOptions, testSetupPassesNewConsumerOptions, testSetupPassesStreamPlacementOnCreate, testSetupUpdateAppliesConfiguredStreamPlacementToAnExistingStream, testSetupUpdatePreservesServerStreamPlacementWhenNotConfigured, testAutoSetupProvisionsOnFirstSendOnce, testAutoSetupProvisionsOnFirstGet, testAutoSetupDisabledByDefaultDoesNotProvisionOnSend, testSetupUpdatesExistingStreamMergesSubjectsAndPreservesServerConfig, Behat scenarios Setup NATS stream with max age configuration, Setup command handles existing streams gracefully, and Custom consumer name is registered in JetStream

Delayed / Scheduled Messages

Requires NATS Server >= 2.12 with JetStream enabled. If scheduled_messages is enabled against an older server, messenger:setup-transports fails with a clear, actionable error telling you to upgrade NATS or disable the option.

Enable scheduled_messages in the DSN to use Symfony's DelayStamp for deferred delivery:

framework:
  messenger:
    transports:
      nats_transport:
        dsn: 'nats-jetstream://localhost/my-stream/my-topic?scheduled_messages=true'

Then dispatch messages with a delay:

use Symfony\Component\Messenger\Stamp\DelayStamp;

// Deliver after 30 seconds
$bus->dispatch(new MyMessage(), [new DelayStamp(30000)]);

Tested by: testSendWithDelayStampPublishesToDelayedSubjectWithScheduleHeaders, testSendDelayedMessageSchedulesAtRequestedDelay, testSendDelayedMessageNeverSchedulesBeforeRequestedDelay, testReadmeScheduledMessagesDsnEnablesFeature, Behat scenarios Delayed messages are delivered after the scheduled time and Delayed messages are not available to the consumer before the scheduled time

When scheduled_messages is enabled and a DelayStamp is present:

The message is published to a {topic}.delayed.{uuid} subject with Nats-Schedule and Nats-Schedule-Target headers
The stream is created with an additional {topic}.delayed.> subject and allow_msg_schedules enabled
NATS holds the message and delivers it to the original topic at the scheduled time
The consumer processes it like any other message

The DelayStamp delay (milliseconds) maps onto a NATS @at schedule, which has whole-second resolution. The delay is rounded up to the next whole second, so a message is never delivered before the requested delay elapses (it may arrive up to ~1 second later); a sub-second delay therefore schedules at the next whole second rather than firing immediately.

When scheduled_messages is disabled (the default), any DelayStamp on the envelope is silently ignored and messages are published immediately. That includes the DelayStamp Symfony's retry adds, so retries run back to back (see Retry Handler Behavior).

Stream Monitoring

View stream and consumer information:

# List streams
nats stream list

# View stream info
nats stream info my-stream

# List consumers
nats consumer list my-stream

# View consumer info
nats consumer info my-stream my-consumer

# View message count
nats consumer info my-stream my-consumer --json | jq '.state.num_pending'

Manual Message Operations

use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Transport\TransportInterface;

// Get message count
$count = $transport->getMessageCount();

// Check if messages are pending
if ($count > 0) {
    echo "Pending messages: $count";
}

getMessageCount(), which messenger:stats shows, returns 0 when NATS cannot be reached, the same as for an empty queue, so during an outage the queue reads as empty there. It gives up after a single connection attempt (about 6 seconds for a refused connection, see Connection Timeout). Monitor the connection itself, or the stream with the nats CLI, rather than reading a 0 as "nothing to do".

Tested by: testGetMessageCountReturnsConsumerPendingMessages, testGetMessageCountFallsBackToStreamState, testGetMessageCountReturnsZeroWhenLookupsFail, testGetMessageCountSumsAckPendingAndPending, testMessageCountThatCannotConnectDialsOnceAndReturnsZero

Troubleshooting

Connection Issues

Error: "Connection refused"

# Check NATS is running
nats-server --js

# Verify host and port
nats-jetstream://localhost:4222/stream/topic

Tested by: Behat scenario Setup command fails gracefully when NATS is unavailable

Error: "Stream not found"

# Run setup command to create stream
symfony console messenger:setup-transports nats_transport

Tested by: testSetupCreatesStreamAndConsumer, testSetupUpdatesStreamWhenItAlreadyExists, Behat scenarios Setup NATS stream with max age configuration and Setup command handles existing streams gracefully

Error: "Cannot pull from NATS stream ...: the subscription limit was exceeded"

The connection may hold only one subscription, and the transport needs two. Raise the server's max_subscriptions, or the account's or user's subscription limit, to at least two (see Subscription Limits).

Tested by: testAtALimitOfOneThePullFailsWithATransportExceptionNamingTheLimit, testAtALimitOfTwoPullsDeliverAndAcknowledge

Error: "Reconnect attempts exhausted"

With reconnect: true the client's recovery of a lost connection gave up after max_reconnect_attempts, and the operation that reports it ran the client's attempt loop inside the call. The outage outlasted one recovery cycle: raise max_reconnect_attempts for longer outages (the client's default of 10 covers about 109 s against a server that refuses connections), or leave the option off to fail fast and let the process supervisor restart the worker. The logger shows the attempts at warning level and, when acknowledgements were buffered during the outage, the discard at error level (see Logging).

Tested by: testAReconnectThatGivesUpWithABufferedAcknowledgementIsReportedToTheLoggerAndTheNextOperationFails

Error: "TLS upgrade requested but no TLS context was configured at connect time"

On a cluster that requires TLS, the client dialled a server it learned from the cluster without TLS, because only the nats-jetstream+tls:// scheme asked for it. Set tls_required: true as well (see TLS on a Cluster).

Error: "replicas > 1 not supported in non-clustered mode"

stream_replicas above 1 needs a JetStream cluster; on a standalone server leave it at 1 (see High Availability).

Message Processing Issues

Messages not being consumed

# Check consumer exists
nats consumer list my-stream

# View consumer status
nats consumer info my-stream my-consumer

# Check for errors in consumer
nats consumer info my-stream my-consumer --json | jq '.state'

Messages stuck in pending

# Check handler is not throwing exceptions
# Verify handler implementation
# Check application logs for errors

Architecture

The bridge consists of two main components, wired into the framework by a thin bundle:

IDCTNatsMessengerBundle

Registers NatsTransportFactory and the IgbinarySerializer service in the application container
Hands the application's logger service to the factory, under Monolog's messenger channel
Auto-enabled by Symfony Flex; no semantic configuration

NatsTransportFactory

Handles DSN scheme detection (nats-jetstream://)
Creates NatsTransport instances, handing them its optional PSR-3 logger for the NATS client
Validates configuration

NatsTransport

Implements Symfony's TransportInterface, MessageCountAwareInterface, SetupableTransportInterface, KeepaliveReceiverInterface, and CloseableTransportInterface
Manages stream and consumer connections: dials again itself after a lost connection, or leaves the recovery to the NATS client with reconnect: true
Handles message serialization via a pluggable SerializerInterface (igbinary when constructed directly without one)
Supports batching and explicit acknowledgment

Performance Tips

Choose appropriate batching
- Start with batching: 5 for balanced performance
- Increase to 20+ for high throughput workloads
- Use 1 for strict low-latency requirements
Set reasonable timeouts
- max_batch_timeout: 0.5 for responsive systems
- max_batch_timeout: 2.0 for background jobs
- connection_timeout: 1.0 for local/regional deployments
- connection_timeout: 3.0+ for cross-region or high-latency networks
Use appropriate replicas
- stream_replicas: 1 for development
- stream_replicas: 3 for production, on a JetStream cluster (see Using a NATS Cluster)
Monitor performance
- Use getMessageCount() to track queue depth
- Monitor handler execution time
- Watch for stuck messages

Security Considerations

⚠️ Deserialization of Untrusted Data

The default IgbinarySerializer (and any serializer extending AbstractEnveloperSerializer) deserializes raw message payloads from NATS into PHP objects. PHP object unserialization is a well-known attack vector - a crafted payload can trigger arbitrary code execution via magic methods (__wakeup, __destruct, etc.).

⚠️ PhpSerializer fallback: when no serializer is configured and ext-igbinary is not installed, the transport automatically falls back to Symfony's PhpSerializer, which uses native unserialize() - the same untrusted-deserialization (object injection) risk as igbinary, not a safer alternative. The transport emits an E_USER_WARNING when this happens. Do not rely on the fallback in production: either install ext-igbinary or explicitly configure a serializer (ideally a safe-format one, per below).

If your NATS topics are not fully trusted (e.g. shared infrastructure, external publishers), you should:

Implement a custom serializer that uses a safe format (JSON, Protobuf) instead of PHP object serialization
Add message-level authentication (e.g. HMAC signatures) to verify publisher identity before deserializing
Restrict NATS topic publish permissions via ACLs so only trusted services can publish

The type check (instanceof Envelope) happens after deserialization, which is too late to prevent exploitation.

Stream-Exists Detection During Setup

During setup(), if createStream fails the transport detects a pre-existing stream deterministically by querying JetStream stream info: a 404 means the stream is absent (so the creation error was genuine and is rethrown), while a successful lookup means the stream exists (so it is updated, reusing the fetched configuration). This relies on the JetStream stream-info API rather than matching server-specific conflict strings ("already in use" / "already exists"), whose wording varies across NATS versions.

If you experience unexpected behavior during stream setup, confirm the stream can be queried via JetStream stream-info APIs and review the exact error returned by your NATS server version.

Publish Response Validation

On send(), the transport awaits the JetStream publish acknowledgement returned by the client's publish() call. The client validates that acknowledgement and raises a JetStreamException if JetStream reports an error or returns an empty/malformed response, so a proxy or protocol misconfiguration fails closed instead of silently accepting an invalid publish acknowledgement.

General Recommendations

Authentication
- Prefer environment variables or explicit options for credentials over hard-coded DSNs
- If you use credentials in a DSN, avoid logging the full DSN because it may expose secrets
- Store credentials in environment variables
- Never commit credentials to version control
Message Encryption
- Encrypt sensitive data before dispatching
- NATS can be configured with TLS for transit encryption
- Implement application-level encryption for sensitive payloads
Access Control
- Restrict stream/consumer creation to authorized users
- Use NATS access control lists (ACLs) for fine-grained permissions
- Audit stream operations

Contributing

Contributions are welcome! Please ensure:

Every modification runs the relevant verification commands before it is considered done
Minimum verification for PHP changes: composer test
All tests pass: composer test:unit
Code coverage remains above 90%
New features include corresponding tests
Documentation is updated
Functional tests pass: composer test:functional (if applicable)
docs/TESTS.md is kept up to date when tests are added, removed, or renamed
Each release has an entry in docs/CHANGELOG.md following Keep a Changelog format
When a PR is merged or its features are adapted, a description is added to docs/PRs/

Quick Development Workflow

# 1. Run static analysis and the default unit suite after each modification
composer test

# 2. Set up functional tests (first time only)
composer test:functional:setup

# 3. Start NATS for functional tests
composer nats:start

# 4. Run functional tests
composer test:functional

# 5. Clean up
composer nats:stop

License

MIT License - see LICENSE file for details

Support

For issues, questions, or suggestions:

Check the troubleshooting section
Check existing issues on GitHub
Create a new issue with detailed information

idct / symfony-nats-messenger

Package info

Security

Fund package maintenance!

Statistics