Symfony Nats Messenger Laravel Package

Getting Started

1. Install and confirm the runtime constraints first. The package requires PHP 8.2+, Symfony ^7.2/^8.0, and (as of 5.7) psr/log ^3.0 as a direct requirement — it will not install on a Symfony 6.4 LTS app, and will conflict if anything in your tree pins psr/log ^1/^2, so check both before committing.

composer require idct/symfony-nats-messenger

With Symfony Flex, IDCTNatsMessengerBundle is auto-registered; it registers the transport factory and the igbinary serializer, so there is no manual wiring. Without Flex, add the bundle to config/bundles.php.

2. Point one transport at a JetStream stream. The scheme must be nats-jetstream:// (or nats-jetstream+tls://); everything else is options.

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'   # nats-jetstream://nats:4222/events
                options:
                    stream_storage: file
                    stream_replicas: 1
                    stream_max_age: 604800
        routing:
            App\Message\SendWelcomeEmail: async

3. Create/update the stream once per environment. NatsTransport implements SetupableTransportInterface, so this is the supported way to provision:

bin/console messenger:setup-transports async

4. Consume.

bin/console messenger:consume async -vv

First use case: offload side effects (welcome emails, webhooks, report generation) out of the request cycle — dispatch from a controller, return immediately, let a worker drain async. That exercises publish validation, a consumer, and acks with zero extra configuration.

Where to look first (in this order):

• Options/NatsTransportConfigurationBuilder.php — the authoritative DSN + option parsing/validation. Trust it over any README snippet.
• Options/TransportOption.php and Options/RetryHandler.php — the complete, small option surface (RetryHandler has exactly two values; reconnect adds one boolean plus max_reconnect_attempts).
• tests/functional/features/*.feature — these read as executable documentation for batching, NAK/TERM, delayed messages, consumer names, and TLS/mTLS.
• README → Automatic Reconnect (per-operation time bounds) and tests/unit/ReconnectModeTest.php — the only honest description of what happens to a worker during a broker outage.

Implementation Patterns

Topology: one transport per stream, split by delivery guarantees, not by message class. Retention, storage, and replication are per-stream properties, so group messages that need the same lifetime and durability.

transports:
    critical:
        dsn: 'nats-jetstream://nats:4222/critical'
        options: { stream_storage: file, stream_replicas: 3, stream_max_age: 1209600 }
    bulk:
        dsn: 'nats-jetstream://nats:4222/bulk'
        options: { stream_storage: memory, stream_max_age: 3600, batching: 50, max_batch_timeout: 1 }

Retry policy: choose one side, deliberately.

• retry_handler: symfony (default) — the transport TERMs a failed message, so Messenger's own retry_strategy (max_retries, delay, multiplier) and failure_transport govern behavior. This is the right default for PHP-only apps.
• retry_handler: nats — the transport NAKs, and JetStream redelivers according to its own ack_wait/delivery limits. Use when NATS (or another language's consumer) should own redelivery.

async:
    dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
    retry_strategy: { max_retries: 3, delay: 1000, multiplier: 2 }
    failure_transport: failed

Surviving broker outages: reconnect mode (5.7+). Off by default; turn it on for long-running workers that should outlive a NATS restart or failover rather than exit and be restarted by the supervisor.

options:
    reconnect: true
    max_reconnect_attempts: 20   # default: null → the client's own default of 10

Semantics worth internalising:

• A client already recovering is left alone; the operation waits inside its own budget — request_timeout for a request, max_batch_timeout + 1s for a pull, which then returns an empty batch rather than throwing.
• A PING failure is triaged: recovering/already-recovered ⇒ keep the connection; still-open with no reconnect behind it ⇒ replace, as before; gave up ⇒ dial again (attempt loop runs inside the operation and propagates its exception — this is the only case that blocks the caller).
• With auto_setup, provisioning is re-verified after a recovery, and a publish that lands on a server missing the stream gets one setup() plus one re-publish of the same frame.
• Default budget is ~109 s of tolerated outage (10 attempts against a refusing server). Size max_reconnect_attempts against your worst deploy/failover, and remember the fallback: once the client gives up, the next operation dials inline and stalls that operation for the length of the attempt loop.

Keep supervisor restart policies anyway — reconnect covers lost connections, not a wedged handler or OOM.

Give the NATS client your logger (5.7+). If the container has a logger service, the bundle injects it automatically; the client writes connection lifecycle at info and errors at error, under Monolog's messenger channel. Budget for the volume (a flapping broker at info level is noisy) — either raise the channel's level to notice, or set up a dedicated handler for messenger so transport chatter doesn't drown your app logs. No config needed to opt in; nothing to do to opt out beyond not having a logger service.

Throughput via batching — tune together, they trade latency against fullness:

options: { batching: 50, max_batch_timeout: 1 }

Use batching > 1 for aggregation/analytics work; keep it at 1 when each message calls an external API with per-item side effects.

Delayed messages need the flag and the stamp:

options: { scheduled_messages: true }

$bus->dispatch(new SendReminder($orderId), [new DelayStamp(60_000)]);

Deduplication for publish retries (outbox double-dispatch, HTTP retry, redeploy replays):

options: { deduplicate: true }

$bus->dispatch($event, [new DeduplicationIdStamp($orderId)]);

This is sent as Nats-Msg-Id and deduped inside JetStream's duplicate window — a short server-side horizon, not a durable idempotency store.

Consumer strategy. The default shared consumer distributes work across all workers on the transport (horizontal scaling for free). Opt into independent consumers, with explicit consumer names, when you need per-worker/per-tenant isolation or priority lanes — and make the name unique per lane, or workers will contend for the same consumer.

Interop or debuggability override. igbinary is PHP-only; if non-PHP services read the stream, or you want to eyeball payloads, delegate to Symfony's serializer on the transport:

async:
    dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
    serializer: messenger.transport.symfony_serializer

For a bespoke wire format that still preserves envelope semantics (stamps/headers), extend AbstractEnveloperSerializer rather than reimplementing encode/decode.

Deployment wiring. Keep the DSN in an env var per environment, run messenger:setup-transports in the deploy script (it is idempotent and also updates existing streams), and supervise workers with --time-limit plus --keepalive for long-running handlers. Treat MessageCountAwareInterface queue depth as a dashboard signal alongside messenger:failed:show.

Gotchas and Tips

• The scheme is the whole integration. Only nats-jetstream:// and nats-jetstream+tls:// resolve a factory. A plain nats:// DSN fails transport creation with an unhelpful "no factory" style error. TLS/mTLS is the scheme plus options, including client certificates.
• setup-transports is not optional in production. No stream means publish acknowledgement validation fails at dispatch time, not at boot — you'll discover it under load. Run it after every routing change.
• Stream updates merge, they never remove. Adding subjects updates the stream and preserves unrelated config, but deleted routing leaves stale subjects behind forever. Prune with the nats CLI.
• Retention silently deletes. stream_max_age / stream_max_messages* discard messages still waiting for a slow consumer. A backlog longer than the window is lost, not delayed — size limits against your worst-case consumer outage. Reconnect mode does not suspend retention: a 3-minute recovery against a 60-second stream_max_age still loses the backlog.
• ack_wait vs handler duration. A handler slower than the ack window causes redelivery and duplicated side effects. Use --keepalive (it queues the in-progress ack from the signal handler), keep handlers short, and make them idempotent regardless — keepalive won't save work killed hard.
• Reconnect does not cover in-flight acks. A reconnect that gives up discards its buffered acknowledgements and logs it at error level; those messages come back after ack_wait. An ack_sync acknowledgement that hits an outage still fails after request_timeout. So: idempotent handlers, and expect duplicates after any outage, not just after failures.
• Reconnect can stall the loop when it gives up. While the client is recovering, an operation waits at most request_timeout/max_batch_timeout + 1s; once the client has given up, the next operation dials inline and blocks for the whole attempt loop (the docs measure 18 s with 2 attempts). With the default 10 attempts that is a long stall — set max_reconnect_attempts deliberately and keep request_timeout sane.
• 5.7 changes pull semantics even with reconnect off. A 409 Server Shutdown or 409 Leadership Change is now an empty batch, and with auto_setup the transport no longer runs setup() against a server that is going away. Don't alert on "idle pulls" turning into "no messages"; do check anything that previously matched on that exception.
• Batching changes ack granularity. With batching > 1, acknowledgement happens per pulled batch, so one poison message can redeliver its siblings. Keep batches modest, keep handlers idempotent, and validate against nats_batching.feature rather than assuming per-message acks.
• Don't run both retry mechanisms. With retry_handler: nats, Symfony's retry_strategy no longer controls backoff, and the failure-transport handoff happens at a different point. Pick one side, configure only that side, and document it.
• Signature changes if you extend the internals. 5.7 adds an optional PSR-3 logger to NatsTransportFactory and a fourth constructor argument to NatsTransport. If you decorate, subclass, or instantiate either by hand, update the constructors on upgrade.
• One service id per class name is intentional. Symfony wraps a transport serializer for signing once per id; aliasing or subclassing the serializer to "share" it produces a double wrap and signature failures. Configure the serializer option instead of touching container wiring.
• igbinary is opaque. The default IgbinarySerializer falls back to PhpSerializer; convenient for PHP↔PHP, unreadable from other languages and unpleasant in nats stream view. Switch serializers early if interop is even plausible.
• Three different timeout knobs. connection_timeout (dial), request_timeout (server replies and the wait budget during a recovery — raise it behind proxies or on high-latency links), max_batch_timeout (pull latency; too high looks like idle workers, too low wastes round trips). Reconnect mode makes request_timeout load-bearing for outage tolerance, not just dialing.
• The client is amphp/coroutine based. sleep(), large synchronous dumps, or blocking HTTP clients stall the loop and can trip keepalive/request timeouts. Use the transport's retry delay rather than sleep(), and offload heavy work.
• Maturity risk is real, though improving. Small star count/few dependents and a fast-moving release cadence mean you are an early adopter: pin the exact version, read docs/CHANGELOG.md and docs/TESTS.md, and budget time for upstream fixes or a fork. The suite is genuinely fast and NATS-free (composer test, 669 unit tests at 99.78% statement coverage, 96.9% mutation score as of 5.7), so running it locally on upgrade is cheap insurance.
• Debugging stack: messenger:failed:show and messenger:consume -vvv on the Symfony side, the messenger log channel for client connection lifecycle (5.7+), then nats stream report, nats consumer info <stream> <consumer>, and nats stream view <stream> for the JetStream side. Publish-ack validation means broker-side problems surface as dispatch exceptions — read them before suspecting your handler.