feat: add NATS JetStream provider connector#140
Draft
miotte wants to merge 11 commits into
Draft
Conversation
Add a natsjs connector implementing the provider.Provider contract so a NATS JetStream server can be used as an Arke backend, selectable per connection via ConnectionConfiguration.provider = "natsjs" alongside the existing amqp091 connector. The connector maps AMQP/RabbitMQ idioms onto JetStream primitives: - topic exchange + routing key -> subject root + token wildcards (# -> >, * -> *), with one stream per address root and per-source subject filters; - durable consumers for non-transient QUEUE / consumer-group / single-active sources (reconnect resumes the backlog) and ephemeral consumers for auto-delete / exclusive / temporary sources; - ack/nack/requeue/dead-letter onto Ack/Nak/NakWithDelay/Term, synthesizing the x-retry-count header from JetStream's delivery count; - publish dedup via Nats-Msg-Id + the stream Duplicates window; - header-filter exchanges evaluated proxy-side; - stream MaxAge/MaxBytes bounds (configurable) to keep the JetStream log from growing without limit, and a configurable replication factor for HA. Connection state is tracked via the NATS connection lifecycle callbacks so WaitForConnect reflects the real link. Behavioral tests run against an in-process JetStream server (publish/subscribe/ack/retry/dead-letter/dedup/ header-filter/durable-backlog); helper unit tests cover the subject mapping, durable-name selection, and filter evaluation. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Document the natsjs connector: subject/topology mapping, durable vs ephemeral consumer selection, the ack/retry/dead-letter mapping table, the retry-count header synthesis, deduplication, the work-queue vs append-log retention model and its configuration knobs, a RabbitMQ<->NATS feature-parity matrix, supported source options, and known limitations. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
miotte
added
enhancement
Bound JetStream management API calls (stream and consumer creation) with an explicit deadline (NATSJS_API_TIMEOUT, default 30s) instead of the client library's 5s default, which can expire while a replicated stream is still forming its raft group on cold or network-attached storage. Collapse concurrent ensureStream calls for the same stream across all client connections into a single CreateOrUpdateStream, so a mass reconnect no longer issues one redundant creation call per connection against the JetStream metadata leader. Failures are not cached and success is still memoized per connection. Run consumers with an explicit 5s idle heartbeat and a consume error handler: a stalled delivery path is now logged at warn level and the pull request is re-issued after ~2x the heartbeat, instead of failing silently on the library defaults. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Adapts the natsjs connector to the removal of client-provided CA certificates (#133): the TLS path now relies on GetTls() and the system trust store, matching amqp091. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Each address gets its own stream, and JetStream requires stream subject spaces to be disjoint — but address names are dotted, so two addresses can sit in a prefix relationship (events.orders and events.orders.filtered). Their capture wildcards then overlap: whichever stream forms first wins and every publish/subscribe on the other address fails forever with 'subjects overlap with an existing stream' (10065), retried by clients in a hot loop. Fix by reserving a delimiter token '~' between the address root and the routing-key tokens, and stripping '~' from address tokens, so any two distinct roots map to disjoint subject spaces by construction. This also stops cross-address leakage (routing key 'filtered.x' on events.orders was previously indistinguishable from 'x' on events.orders.filtered), gives empty routing keys a concrete publishable subject (<root>.~ instead of the wildcard <root>.>), replaces the empty address's capture-everything '>' subject, and sanitizes wildcard characters out of published routing keys (literal in AMQP publishes, forbidden by NATS). Binding patterns with a trailing '#' now also match the zero-word case, as AMQP does. Existing data stored under the old subject scheme no longer matches consumer filters after the stream's subjects move forward; see the migration note in doc/design/natsjs-connector.md. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The stream retains acked messages under its retention limits (LimitsPolicy), so its message count keeps growing after consumers are caught up. Anything reading SourceStats message_count as queue length — consumer autoscaling in particular — would see a permanently inflated backlog. Report the durable consumer's undelivered + unacked count instead (the amqp091 connector's ready+unacked equivalent), populate current_offset from the consumer's ack floor, and fall back to the stream view for sources without a durable consumer. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The design doc framed the subject encoding as something with an upgrade path from an older scheme. No older scheme has ever shipped — it only existed in unmerged history on this branch — so drop the upgrade instructions and instead state the durable fact: the subject layout is part of the connector's persistence contract, any future encoding change is breaking for deployed data, and the scheme (including the '~' delimiter) is canonical for anything reading JetStream directly. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The connector advertises the `Offset` source option in SupportedSourceOptions, and the amqp091 connector's toStreamOffset accepts first/continue/last/next plus an absolute numeric offset. natsjs previously mapped only `first` to DeliverAll and silently degraded everything else -- including `last` and numeric offsets -- to DeliverNew, so a consumer asking to replay from the last message or from a given offset received only new messages, with no error. deliverPolicyFor now mirrors toStreamOffset: first/continue -> DeliverAll, last -> DeliverLast, next/"" -> DeliverNew, and an absolute number -> DeliverByStartSequence with OptStartSeq. Matching is case-insensitive, and offset 0 maps to DeliverAll because JetStream sequences are 1-based. Numeric offsets are JetStream stream sequence numbers (as surfaced by SourceStats) and are not portable across brokers; this is now stated in the code and the design doc. Adds behavioral tests for retained-log replay, independent per-consumer offsets, and the last/next/numeric start positions. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Two gaps remained after natsjs adopted the full RabbitMQ Streams offset vocabulary. An unrecognized Offset still fell back silently to DeliverNew. The amqp091 connector's offset parsing rejects such values, and starting a consumer at a silently different position than the one it asked for loses or replays data, so deliverPolicyFor now returns an error and the subscribe fails with it. JetStream fixes a durable consumer's start position at creation and rejects DeliverPolicy/OptStartSeq updates (err 10012), so re-subscribing an existing durable with a different Offset failed the subscribe -- permanently, since retrying cannot heal a config conflict. The documented contract is that the offset applies on first creation only and a reconnecting durable resumes from its stored ack position, so Subscribe now implements exactly that: when CreateOrUpdateConsumer fails and the durable already exists, it attaches to the existing consumer unchanged and logs a warning naming the conflict. Repositioning requires a new durable (source or ConsumerGroup) name. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The source-options table claimed both options were mapped to the stream MaxAge. Neither is read by the connector: retention comes solely from the stream-wide NATSJS_STREAM_MAX_AGE / NATSJS_STREAM_MAX_BYTES configuration, because a per-source value cannot be mapped onto the shared per-address-root stream without one source's value flapping another's retention. The options stay in SupportedSourceOptions so existing amqp091 client sources validate unchanged; the doc now states what actually happens instead of implying per-source fidelity. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The 2.11 line reached end of life with 2.11.17; patch releases now ship on the 2.12.x and 2.14.x lines. The connector's behavioral suite runs against the embedded server, so testing against a current release matches what new deployments run. No connector changes were needed -- it uses only the nats.go jetstream package, and nats.go stays at 1.52.0. Transitive bumps: jwt v2.8.2, nkeys v0.4.16, x/crypto v0.53.0, x/sys v0.46.0, x/term v0.44.0. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #139.
Adds a
natsjsconnector implementingprovider.Providerso a NATS JetStreamserver can be used as an Arke backend, selectable per connection via
ConnectionConfiguration.provider = "natsjs"alongside the existingamqp091connector. One Arke instance can serve both broker types at once.
What's included
internal/provider/connectors/natsjs/— the connector (natsjs.go,helpers.go) + tests, registered inconnectors.go.doc/design/natsjs-connector.md— design notes following theconnector-interface contract: mapping, consumer model, ack/retry/DLQ table,
retention model, configuration, operational resilience, the RabbitMQ↔NATS
feature-parity matrix, supported source options, and known limitations.
How idioms map
#→>,*→*)Offset(first/continue/last/next/numeric)DeliverPolicy+OptStartSeqrequeue_delayNakWithDelayTerm()x-retry-count/x-deathNumDeliveredNats-Msg-Id+ stream duplicates windowSourceStats)NATSJS_STREAM_REPLICAS)JetStream requires stream subject spaces to be disjoint, but address names are
dotted and can sit in a prefix relationship (
events.ordersvs.events.orders.filtered). Subjects therefore reserve a~delimiter tokenbetween the address root and the routing-key tokens, making any two distinct
roots disjoint by construction. The resulting subject layout is part of the
connector's persistence contract — see the design doc.
Consumer start position
The
Offsetsource option accepts the same vocabulary as the amqp091connector's Streams support:
first/continuemap to deliver-all,lasttothe final message,
next(or unset) to new messages only, and an absolutenumber to that stream sequence. Unrecognized values fail the subscribe instead
of silently starting the consumer at a different position. JetStream fixes a
durable's start position at creation, so a re-subscribe that requests a
different offset logs a warning and resumes from the durable's stored ack
position (repositioning requires a new durable name). Numeric offsets are
JetStream sequence numbers (as surfaced by
SourceStats) and are not portableacross brokers.
Operational resilience
JetStream management calls carry an explicit deadline (
NATSJS_API_TIMEOUT,default 30s), and concurrent stream assertion collapses into a single inflight
call per stream via a provider-wide registry, so topology creation under a
connection stampede neither times out spuriously nor hammers the JetStream
API. Consumers run with pull heartbeats, and consume errors are logged at warn
instead of being dropped.
Connection lifecycle
Connection state is tracked via the NATS lifecycle callbacks
(connect / reconnect / disconnect / closed) so
WaitForConnectreflects thereal link instead of optimistically reporting connected.
Testing
Behavioral tests run against an in-process JetStream server (
nats-server/v2v2.14.3, test-only dependency), exercising publish/subscribe/ack, delayed
retry +
x-retry-count, nack redelivery, dead-letter, dedup, headerfiltering, durable-backlog resume, retained-log offset replay (
first/last/next/ numeric, with independent per-consumer positions), andcoexistence of prefix-related addresses. Helper
unit tests cover the subject/wildcard mapping, durable-name selection, and
filter evaluation. Package coverage is ~90%.
Known limitations / follow-ups
MessageTTL/Expiresare accepted for interface compatibilitybut not applied — retention comes from the stream-wide
NATSJS_STREAM_MAX_AGE/NATSJS_STREAM_MAX_BYTESconfiguration (onestream per address root); true per-source TTL needs a per-source stream
topology.
SourceStatspublish/deliver rates are not exposed by JetStream and are leftat zero.
follow-up.
embedded server).