feat: add NATS pub/sub backend - #4550
Conversation
Selectable per deployment with SERVER_MSGQUEUE_PUBSUB_KIND=nats and SERVER_MSGQUEUE_PUBSUB_NATS_URL. NATS implements only the best-effort pub/sub side of the split introduced in #4480; durable queues stay on RabbitMQ and Postgres. Core NATS, no JetStream, so delivery is at-most-once. Connect is synchronous and validates the server's max_payload against the 16MiB the durable RabbitMQ path allows, because task stream events routinely exceed the NATS default of 1MiB and a startup failure is easier to diagnose than a later oversized publish. ReconnectBufSize(-1) makes publishes fail fast while disconnected rather than silently buffering messages a best-effort subscriber will never wait for. The other backends namespace installations at the connection level (a RabbitMQ vhost, a Postgres database); NATS namespaces by subject, so SERVER_MSGQUEUE_PUBSUB_NATS_SUBJECT_PREFIX is the parity knob for installations that share a NATS server. Username and password are passed as connect options rather than embedded in the URL so that reconnects to gossip-discovered cluster peers authenticate too.
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
github-actions
Bot
added
documentation
|
|
|
|
| // requiredMaxPayload is the minimum NATS server max_payload we accept. | ||
| // The durable RabbitMQ path allows 16MiB, so the NATS server must match. | ||
| const requiredMaxPayload = 16 * 1024 * 1024 |
There was a problem hiding this comment.
this makes me a bit nervous since I think this can e.g. change across rabbit versions (pretty sure we ran into this)
There was a problem hiding this comment.
Good catch. Our rabbitmq backend hardcodes its own client-side maxPayloadSize = 16MiB and never reads the broker's limit. Fixed in 830c4d6 by adding a single shared msgqueue.MaxMessageSize used by rabbit and NATS. If the contract ever changes, at least this will happen in one place.
| natsgo.MaxReconnects(-1), // reconnect indefinitely | ||
| natsgo.ReconnectBufSize(-1), // do not buffer pubs during disconnect |
There was a problem hiding this comment.
should these be configurable?
There was a problem hiding this comment.
I don't think either makes sense as a tunable config:
- MaxReconnects: any finite value permanently closes the connection once exhausted, leaving the engine without pub/sub until a process restart (so on a NATS outage, if the broker recovers after the MaxReconnects we'd never know). The rabbitmq backend also retries forever.
- ReconnectBufSize(-1): with a buffer, publishes during a disconnect get queued and flushed on reconnect. But these signals are latency optimizations with polling paths as fallbacks; by the time a buffered publish flushed, polling would have likely covered it.
Documented the rationale inline in 830c4d6 as comments so it doesn't surprise the readers.
| if opts.username != "" || opts.password != "" { | ||
| connectOpts = append(connectOpts, natsgo.UserInfo(opts.username, opts.password)) | ||
| } |
There was a problem hiding this comment.
do these need to both be non-empty? or just one? I guess maybe it doesn't matter since the connect after this will fail if they're wrong
There was a problem hiding this comment.
Thanks for raising. NATS marshals this with omitempty, so UserInfo("", "") is the same as not setting it at all. Empty credentials never reach the broker.
In 830c4d6 I deleted the conditional and pass UserInfo unconditionally. Whether a lone username (or password) is acceptable is the server's call, and it fails loudly at the synchronous Connect on startup, as you said.
- Hoist the 16MiB limit into a shared msgqueue.MaxMessageSize: it is Hatchet's own message-size contract (never read from a broker), so one constant now moves the rabbitmq chunking, the NATS startup check, and the error messages together instead of drifting apart. - Pass nats.UserInfo unconditionally: empty credentials are omitted from the CONNECT payload, so the either-set conditional was a no-op guard. - Document why MaxReconnects(-1) and ReconnectBufSize(-1) are fixed rather than configurable: finite reconnects permanently kill pub/sub, and buffered publishes flushed after reconnect add nothing over the consumers' polling paths.
Benchmark resultsCompared against |
|
|
2 participants
Description
Note
This is an experimental prototype. We are not launching this as a feature yet; we need to gain more experience running this internally first.
Adds NATS as an opt-in pub/sub backend, set:
SERVER_MSGQUEUE_PUBSUB_KIND=natsSERVER_MSGQUEUE_PUBSUB_NATS_URL=...It covers only the best-effort pub/sub side of the #4480 split: core NATS, at-most-once. Durable queues stay on RabbitMQ/Postgres.
The defaults are unchanged, NATS remain optional, until the config is set.
Type of change
What's Changed
internal/msgqueue/natsplus config, docs, and a compose service.Technicalities:
max_payloadis under the 16MiB. 16MiB is what RabbitMQ allows, so as a drop in replacement, we want to keep this.SERVER_MSGQUEUE_PUBSUB_NATS_SUBJECT_PREFIX(defaulthatchet.pubsub) for isolating hatchet installations sharing a NATS server (not unlike like a RabbitMQ vhost or Postgres database).Checklist
Changes have been:
Testing
🤖 AI Disclosure
I acknowledge that an LLM was used in the creation of this Pull Request, in accordance with Hatchet's AI_POLICY.md.
Details: Cursor agent implemented the backend, config/docs/compose wiring, and tests from a reviewed plan.