feat: add NATS JetStream provider connector#140
Draft
miotte wants to merge 78 commits into
Draft
Conversation
miotte
added
enhancement
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>
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>
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>
Ephemeral consumers were left to expire via InactiveThreshold (5m). Each one counts against the server's per-stream consumer limit until then, so clients churning transient subscriptions faster than the threshold would accumulate dead consumers on the server. Delete the consumer eagerly when the subscription tears down; the threshold remains as the janitor for unclean exits. The delete is best-effort on a detached, bounded context — the subscription's own context is already cancelled when it runs. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The amqp091 connector fills SourceStats publish_rate/deliver_rate from the RabbitMQ management API; JetStream exposes only absolute counters, so those fields were left at zero. Derive them proxy-side instead: a per-connection rateTracker differences the stream's LastSeq (publish) and the durable consumer's delivered count (deliver) between successive SourceStats calls. The first observation baselines at zero, a counter that moves backwards (recreated stream or consumer) re-baselines, and a non-advancing clock reports zero rather than dividing by zero. Caveats, also documented: the publish rate covers the whole address stream rather than a single binding, and sources without a durable consumer have no consumer identity to sample, so their deliver rate stays zero. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
MessageTTL and Expires stay in SupportedSourceOptions so sources written for the amqp091 connector validate unchanged, but natsjs retention is stream-wide and the options are not applied. That was documented but otherwise silent — and silent divergence in data retention is the one place a client should not have to read a design doc to notice. Subscribe now logs a warning naming the source and the ignored options. Rejecting such sources outright was considered and dropped: it would fail every existing client that legitimately sets a TTL the moment it is pointed at natsjs, turning a retention caveat into an outage. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Header filters cannot narrow server-side delivery on NATS, so the connector receives and discards non-matching messages. State plainly that the cost scales with subject-matched traffic, not header-matched traffic, and when to remodel header routing onto subjects. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
miotte
marked this pull request as draft
July 11, 2026 16:35
DeadLetter published to the DLQ best-effort and then Term'd the original unconditionally: a failed ensure of the DLQ stream was skipped silently and a failed DLQ publish was only logged at debug, yet the message was still removed from redelivery — gone from both the source and the DLQ. RabbitMQ has no such window because the DLX move happens broker-side. Order the two steps so the data survives: Term only after the DLQ publish succeeds. On failure, warn and return an error while leaving the message resolvable and ack-pending; the server then falls back to a nack, so the message is redelivered and dead-lettering is retried. The DLQ copy now carries a Nats-Msg-Id derived from the original's stream sequence so a retried dead-letter cannot duplicate it within the dedup window (the header map is copied to keep the option from mutating the original message). Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
AMQP prefetch 0 means unlimited, and the amqp091 connector honors that by only calling SetPrefetch for positive values, leaving the channel default (unlimited). natsjs clamped prefetch <= 0 to MaxAckPending 1 — the opposite extreme: one in-flight message per consumer, a silent throughput cliff for any client relying on the AMQP convention. Map <= 0 to MaxAckPending -1, JetStream's explicit unlimited. The pinned-offset test now sets prefetch 1 explicitly, since it depends on one-at-a-time delivery to keep its backlog undelivered across a re-subscribe. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The attach-existing-durable fallback fired on any CreateOrUpdateConsumer error whenever the durable already existed. It was aimed at JetStream's refusal to move a pinned start position (err 10012, deliver policy / start sequence), but as written it also masked unrelated update failures: a consumer whose update was rejected for any other reason would be resumed unchanged with only a warning — silently consuming with a configuration (ack policy, filter subjects) the source never asked for. Classify the error instead: fall back only when the API error is the consumer-create code carrying one of the server's start-position refusals (deliver policy / start sequence / start time can not be updated); every other error stays fatal. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
…exits The eager delete of an ephemeral consumer ran only on the normal subscription teardown. A declare-only subscribe returned before registering that cleanup, and a Consume() failure returned past it, so in both cases the consumer lingered on the server for the full inactivity threshold. The declare-only case can never be consumed at all — its server-generated name is not surfaced — so delete it before returning, and do the same when consuming fails to start. Also log (at debug) when the ack of a header-filtered message fails: the redelivery is harmless, but it inflates NumDelivered and thus the synthesized x-retry-count of a message the client never saw, which is worth a trace when debugging retry counts. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
A subscription's delivered-but-unacked messages could never be resolved after its consume stream closed (acks travel on that stream), yet their entries stayed in the active-message map for the life of the connection - a leak that also inflated the active-message stat - and the messages themselves stayed ack-pending on the server until AckWait expired. RabbitMQ requeues a closed channel's unacked deliveries immediately, so redelivery latency regressed from ~0 to up to AckWait on every consumer restart. Subscribe teardown now waits (bounded) for the consume loop to finish unsubscribing, flushes the connection so the server has processed the unsubscribe - a nak'd message could otherwise be redelivered straight into the dying subscription's still-registered pull request and strand until AckWait a second time - and then releases every in-flight message belonging to this consumer: entries are dropped from the map, and durable consumers' messages are nak'd for prompt redelivery. Ephemeral consumers skip the nak since the consumer is deleted on the same path. A delivery that loses the race against subscription shutdown is nak'd in the delivery handler instead of only being dropped from the map. This also deflakes TestDurableOffsetPinnedAtCreation: with prefetch 1, acking m0 let the server deliver m1 to the first subscription right before it was cancelled, leaving m1 ack-pending; the re-attach then timed out waiting for a redelivery that only AckWait would trigger (observed roughly 1 in 4 runs, now clean over repeated -count=25). Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The 30s AckWait was hardcoded. It sets one dial between two failure modes - a crashed client's in-flight messages are stuck for the full ack wait before another consumer gets them, while a healthy consumer that holds a message longer than the ack wait (including time queued in the client-side pull buffer behind a large prefetch) gets a duplicate redelivery - and the right value is deployment-specific: RabbitMQ's equivalent consumer timeout defaults to 30 minutes, so consumers doing tens of seconds of work per message are in-contract on RabbitMQ and silently double-processed here. Keep the failover-friendly 30s default and expose NATSJS_ACK_WAIT (Go duration), documented alongside the other stream/consumer knobs. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
SingleActiveConsumer sources only got a durable consumer, which is the opposite of single-active: every subscriber of the durable pulls from it, so instances compete and per-key ordering - the reason a client asks for SAC - silently breaks the moment a service scales past one replica. The parity table nevertheless claimed Native support. Map SAC onto JetStream's pinned-client priority group (nats-server 2.11+): the consumer is created with PriorityPolicyPinned and a single group, the server delivers only to the pinned client, standbys' pulls wait (still receiving idle heartbeats, so the liveness warning does not trip), and when the pinned client stops pulling for the configured TTL the pin moves to a standby. Failover time is bounded by NATSJS_SAC_PINNED_TTL (default 1m; it must comfortably exceed the ~30s pull re-issue cadence or the pin flaps between healthy standbys). The pull request must carry the group exactly when the consumer has one, so Consume keys off the effective (server-reported) config rather than the requested one: a server that predates priority groups drops the fields silently, and an existing durable the subscribe merely attached to may lack them - in both cases consumers compete as before, and a warning now says so instead of nothing. Priority config is update-mutable, so durables created before their source set SingleActiveConsumer upgrade in place, keeping name and ack position (covered by TestSingleActiveConsumerUpgradesExistingDurable; live pin-and-failover covered by TestSingleActiveConsumerPinsOneClient). Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The address-to-stream, source-to-consumer mapping puts message storage one level above where AMQP puts it, which is why per-source MessageTTL and max-length are not honored and acked messages are not deleted on ack. The obvious alternative -- a stream per source, sourcing from a per-address origin -- is more attractive in outline than in fact, and had been re-litigated from first principles more than once. Record the decision and the evidence against the alternative: it does not deliver expire-into-the-dead-letter-exchange, which JetStream cannot express at any topology (no advisory fires on limit eviction, and subject delete markers are a key-value feature -- no body, last message on a subject only, and stamped with a rollup that purges the subject); it weakens publish confirmation to the origin stream; it costs a stored copy, a replication group and a synchronous write per message per source; WorkQueuePolicy is narrower than a queue (errs 10099/10100/10101); and origin retention becomes a silent-loss dependency. Also record why InterestPolicy is not a cheaper route to delete-on-ack, and the conditions under which to revisit. Record a fourth accepted consequence: a transient source becomes an ephemeral consumer, and no start position for it is faithful. A queue's existence defines what is retained for that queue, while a stream retains what the address received regardless of listeners, so DeliverNew skips messages a queue that ought to have existed would hold, and DeliverAll replays history a fresh transient queue never had. DeliverNew is the narrower error, but it is a trade rather than a mapping. Also rewrap an over-length line left behind in the source-stats limitation. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Expires (AMQP x-expires: delete the queue after this many milliseconds without consumers) was accepted, warned about, and silently ignored - filed alongside MessageTTL as blocked by the shared per-address stream. That classification was wrong: MessageTTL governs message retention and is genuinely stream-wide here, but Expires governs the source's lifetime, and consumers are per-source, so it maps cleanly onto the consumer's InactiveThreshold - which the connector already set, just hardcoded to 5 minutes for ephemerals and never for durables. A client overriding Expires was honored by amqp091 and silently diverged here. Now the value is parsed and applied to both kinds of consumer (the amqp091 connector likewise sets x-expires on any queue; nats-server supports the threshold on durables since 2.9 and allows updating it). Unset keeps the previous defaults: transients expire after 5 minutes - the same default amqp091 applies to auto-delete/exclusive queues without x-expires - and durables never expire. A non-integer value is rejected with amqp091's exact error string; a non-positive value, which the AMQP broker itself refuses, is rejected too rather than silently meaning never (durable) or the server default (ephemeral). Only MessageTTL remains in the accepted-but-unapplied warn. The design doc's parity table, options table, and Known limitations no longer lump the two options together. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
A Disconnect with a live subscription ended that Subscribe, and the
server ends the caller's whole consume stream when Subscribe returns -
the client saw end-of-stream. The amqp091 connector leaves that stream
open: closing the AMQP connection leaves its subscribe loop blocked in
a select none of whose channels can fire again (the close notification
arrives once as a zero-value error while the connection state still
reads connected, and the shim's delivery-forwarding goroutine exits
without closing its output channel), so the loop just goes quiet and
the integration suite pins the observable result: a client that
disconnects and then acks its in-flight messages gets each ack answered
with a per-ack 'could not retrieve broker details' failure rather than
end-of-stream (TestSubscribeAckInvalidIDNoConnect).
Subscribe now parks until the stream's own context ends when the
subscription was torn down by this client's Disconnect. Delivery cannot
resume either way - the disconnect already stopped the consume
machinery and drained the connection. The case is discriminated by a
clientDisconnect flag (mirroring amqp091's) set before the consume
contexts are stopped, not by the CLOSED connection state alone, so a
connection the NATS library closes terminally still ends the
subscription as before.
Also aligns the no-broker-details error string with amqp091 ('for this
connection'), which that test asserts as a prefix.
Integration standing moves to 42 passing (was 41); the remaining 8
failures are the known, triaged set.
Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
PublishOne to a STREAM address auto-created the backing stream, where amqp091 routes unary stream publishes through the RabbitMQ Streams client, which refuses a stream nobody declared (declaring is the reader's job - streamSubscribe's DeclareStream is what creates it). The divergence is a silent-success footgun: a publisher with a typo'd or never-declared stream address got success, and its messages accumulated in a junk stream no reader will ever see, instead of the immediate error RabbitMQ gives. The unary path now looks the stream up instead of ensuring it: a miss answers "stream ... does not exist" and creates nothing, and the no-stream-response self-heal re-checks existence rather than resurrecting a stream deleted out from under the connection (a RabbitMQ stream publisher errors after deletion too). A hit shares the per-connection memo with the ensure path. The streaming Publish path deliberately keeps auto-creating: amqp091 sends every address type there, STREAM included, over an auto-declared exchange and reports no error, so refusing it would diverge in the other direction - and storing the message beats dropping it into an unbound exchange. Message-contract refusals (dedup on a queue address, PublishId without PublisherName) now precede the stream assertion, as they do on amqp091, so a publish that is wrong twice over reports the contract error rather than the missing stream, and a refused publish creates no topology. The design doc's topology-mapping section now states when streams come into being and this exception. Integration TestProduceFailsIfNoStream and TestProduceStreamWithDeduplicationFailWithOutProducerName both pass against JetStream. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
The advisory-driven DLQ (a stream capturing $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES / MSG_TERMINATED) is the common NATS idiom for this, so document why the connector does not use it: an advisory names the origin stream_seq rather than carrying the message, so a dead-letter consumer would need a second fetch to see the body, and the record dangles once the primary stream's MaxAge evicts the original. Republishing keeps the DLQ an ordinary address an unmodified AMQP client can consume. Also document that redelivery is unbounded. RabbitMQ quorum queues carry a delivery limit (20 by default on 4.x); this connector sets no MaxDeliver deliberately, because RabbitMQ's delayed-retry idiom republishes through a TTL queue and a DLX and so resets the delivery count, while Retry here is a NakWithDelay that increments the same NumDelivered a limit would cap. Matching the number would cut off retries RabbitMQ allows without bound. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
CreateOrUpdateStream replaces a stream's source set wholesale, but ensureStreamFor only populated it when the current call had bindings of its own to declare. Any assertion that reached the create path without them therefore unbound every address-to-address binding another subscriber had declared on that stream. The call that does this is the most ordinary one there is: a publisher naming a bound address directly carries no ParentAddress, so it arrives with no bindings. The per-connection knownStreams memo hides it from the connection that declared the binding, which is why the existing coverage missed it -- the regression only appears from a second connection, where routing from the parent silently stops with no error anywhere. Read the live source set on every assertion and write it back, so an assertion that declares nothing preserves what is already there. Both paths now do a read-modify-write, so both are serialized under bindMu; the parent's own stream is asserted before that lock is taken, since the nested assertion runs through the same lock. boundSources loses its I/O and becomes the pure merge it always was, and the stream info it reads is the one Stream() already fetched. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
A dead-letter that fails leaves the message in flight so the server's fallback nack retries it rather than losing it. That nack was immediate, and the retry runs the same dead-letter attempt against the same configuration -- so a failure that never clears spins the message between server and client as fast as the client can nack it, with nothing to stop it: MaxDeliver is deliberately left unset so a slow consumer is never silently cut off. The cheapest way in is an empty DeadLetterAddress. The gRPC server enters DeadLetter whenever the option key is present, whatever its value, so an empty one reaches the connector, fails every time, and loops. Measured on an embedded server: over 300 deliveries a second of a single message, indefinitely. A DeadLetterAddress that cannot be resolved into a stream behaves the same way. Delay the redelivery instead (NATSJS_DEADLETTER_RETRY_DELAY, default 5s). A transient failure still recovers promptly; a permanent one costs one redelivery per interval instead of hundreds a second. Keeping the message rather than dropping it is unchanged -- terminating it would lose it while it is also absent from any DLQ. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Resolving a message (take, then delete) and marking one for redelivery (read, then update) are both read-modify-writes over activeMessages. The map is individually concurrency-safe, but nothing serialized the pairs, so a mark landing between a concurrent resolve's read and its delete reinserted an entry nothing could ever resolve -- leaked for the life of the connection and counted in the active-message stat. Exercising the two concurrently over 300 messages left 138 orphans. Serialize the pairs under a dedicated mutex; the network-bound naks in releaseInFlight are moved outside it. Two smaller items found in the same sweep: An unrecognized Offset was rejected only after ensureStreamFor had already created the address's stream, leaving topology behind for an address the client only ever named by mistake. Validate it with the other contract refusals, before any topology is touched -- the ordering Expires and the single-active ConsumerGroup check already follow. The durable header-filter conflict check is per connection, so it cannot see two clients on separate connections attaching to one durable. That is the topology it is nominally protecting, so record what it does and does not guarantee rather than leaving it to read as complete. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Deduplication is a STREAM-address feature on both brokers. amqp091's PublishOne routes only Address_STREAM to publishOneStream — the RabbitMQ Streams client, the one thing that can deduplicate — and sends every other type to publishOneQueue, which refuses a publish id outright. The connector only refused Address_QUEUE, so a TOPIC or FILTER address silently accepted the id and deduplicated on it. TOPIC is the case that matters: it is the protobuf zero value, so an address that never sets a type took the accepting path. A client developing against natsjs would get deduplication there and then a hard error the moment the same code ran against RabbitMQ — the exact cross-broker divergence the refusal exists to prevent. Scope the refusal to the unary path while widening it by address type, because amqp091 splits the same way: its streaming Publish hands every non-STREAM address to a plain channel publish that neither refuses the id nor deduplicates on it. So the streaming path now ignores the id rather than honouring it — refusing would be stricter than RabbitMQ and break a publisher that works there today, while honouring it hands the client a guarantee that disappears on cutover. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
A headers exchange routes on header arguments alone. RabbitMQ matches every one of an exchange's bindings on those arguments and never reads the routing key, so the binding keys amqp091 passes to QueueBind and ExchangeBind for an Address_FILTER decide nothing at all. The connector turned those keys into NATS consumer filter subjects, so a message published to a headers address under a key no binding named was dropped — delivering strictly less than RabbitMQ would, silently. The no-subjects case already had this right (a headers address with filters selects the whole address); it was only the with-subjects case that narrowed. Select the whole address for a headers address either way and let evaluateFilters decide, which is what it is there for. The keys still travel to the parent in an address-to-address binding, where the parent's type decides how they are matched — so a topic parent goes on matching them as routing keys, and a headers parent ignores them in turn. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Because JetStream replaces a stream's source set wholesale on update, every assertion reads the current set and writes it back — a read-modify-write serialized by p.bindMu. That lock is process-local, and Arke runs more than one replica: two processes interleaving a read and a write on one stream can still drop a binding the other just added, which is the failure bindMu was introduced to prevent within a process. Compare the live configuration against the one built and return early when the stream already satisfies it. That takes the overwhelmingly common case out of the race entirely — a bare assertion against a steady-state stream is every publisher's first touch and every reconnect, and it now reads and stops without writing. What remains is callers that genuinely change the configuration, which for bindings are symmetric: both are adding. A stream created before a limit existed is still retrofitted, since its configuration differs. Two details the comparison has to get right. The config being built needs its own copy of the source set: mergeBoundSources appends to a StreamSource's transforms in place, so sharing them with the snapshot would mutate the very thing being compared, the merge would equal itself, and a newly declared binding would be skipped instead of written. Only a second binding onto one parent exposes that — a first appends a whole element and changes the length. And JetStream normalizes an unset size limit to its -1 sentinel and echoes that back, so a config asking for 0 describes the same stream; comparing the two literally made every assertion look like a change and defeated the skip entirely. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
amqp091 reads PublishId and PublisherName in exactly one place:
publishOneStream (amqp091.go:1544/1638), which only PublishOne reaches, and
only for an Address_STREAM. The streaming Publish calls prepareAndSend
directly for every address type -- the plain channel publish, which reads
neither field.
The connector keyed both halves of the contract on the address type alone, so
the streaming path inherited rules that belong to the unary one:
- a STREAM address was deduplicated on the streaming path, silently
discarding a message RabbitMQ stores. A publisher reusing an id there is
doing something RabbitMQ told it was free.
- a missing PublisherName was refused on every address type, so a streaming
publish that works on RabbitMQ errored here.
Gate both on the unary path; the unary behaviour is unchanged.
TestStreamingPublishIgnoresPublishID now covers STREAM alongside TOPIC and
QUEUE -- its old name asserted the wrong scope -- and
TestStreamingPublishAcceptsPublishIDWithoutPublisherName pins the refusal to
PublishOne.
Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
…m a parent An address-to-address binding sources the parent's messages into the child's stream keeping the subject they were published under, and the child's consumer selects exactly those subjects -- so a message the source consumes can be rooted at the PARENT address while the source names the child. routingKeyFromSubject recovered the key by stripping the consuming address's own prefix, which a parent-rooted subject never matches, so it reported an empty routing key for every such message. Dead-lettering one then published it to the bare DLA subject instead of under its original key, where a DLQ consumer bound by routing key would never see it. RabbitMQ preserves the key through both the exchange-to-exchange route and the DLX move. Locate the reserved delimiter instead of stripping a particular address's prefix. That is unambiguous: escapeToken never emits a bare "~" token (an escaped token is "~e", or '~' followed by two-character codes, so at least three characters) and subjectPrefix inserts exactly one, so the first token equal to the delimiter is always the one the connector put there. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Signed-off-by: Richard Sugg <richardsugg@gmail.com>
nats.go serializes headers with net/http's Header.Write (Msg.headerBytes).
That writer is lossy twice, and both losses are silent -- no error ever
reaches the publisher:
- a name failing httpguts.ValidHeaderFieldName is dropped outright. The
stdlib says so in a comment: "we have no good way to provide the error
back ... so just drop invalid headers instead".
- values get CR/LF replaced with spaces, then leading and trailing space
and tab trimmed (textproto.TrimString).
AMQP field-table keys are length-prefixed binary, so RabbitMQ round-trips a
header name containing spaces, colons or non-ASCII exactly. Measured against a
live broker of each kind through one proxy: of 38 header names built from
punctuation and non-ASCII, amqp091 delivered all 38 and natsjs delivered 15.
Values with surrounding whitespace arrived trimmed on natsjs and intact on
amqp091 -- that half needs no exotic name at all, so it can corrupt an
entirely conventional header.
Escape any entry NATS cannot carry verbatim, the way escapeToken already does
for subjects: the mapping is injective and conventional names pass through
untouched, so ordinary traffic stays natively readable to a non-Arke consumer
and only the unrepresentable entries are encoded. Encoding name and value
together keeps one rule and one decode path. Unpadded base64url is used
because its alphabet is entirely valid header-name characters. A name already
carrying the escape prefix is escaped too, which is what makes the mapping
injective: an unescaped name on the wire provably never starts with the
prefix, so decoding only ever fires on entries this connector encoded.
The header round trip had no test coverage at all. It does now, including one
that drives the escaped form through the actual http.Header.Write rather than
a restatement of its rule, so stdlib drift fails the test rather than silently
resurrecting the bug.
Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Takes the timestamp_in_ms fix and keeps its substance, which is right and
corrects a long-standing misclassification: this was parked in the per-broker
skip list as a capability gap ("NATS has no concept of it") when in fact
JetStream carries a per-message timestamp in its metadata. The reasoning that
parked it -- the header comes from a RabbitMQ server feature
(message_interceptors.incoming.set_header_timestamp), NATS has no such server
feature, therefore it is impossible -- skipped a step: Arke is a proxy, so it
has to reproduce the broker's observable result, not its mechanism. The
connector already makes exactly that move for x-retry-count, which it
synthesizes from NumDelivered because x-death has no NATS equivalent.
Three adjustments to the merged version:
- The synthesis moves into handleDelivery, which already reads Metadata()
three lines further down for x-retry-count. That drops a second metadata
read per delivery and, more importantly, lets natsToPbHeader keep taking a
nats.Header. That signature is load-bearing now: it is also the decode half
of the header escaping added in the previous commit, and it is a pure
function only because it takes headers rather than a jetstream.Msg -- which
is what makes the escaping unit-testable without a live broker.
- Dropped the unused HeaderNatsTimeStamp constant ("Nats-Time-Stamp" is not a
real NATS header and nothing referenced it) and the exported DumpMsg debug
helper, which looked like leftover scaffolding. Say the word and either
comes back.
- UnixMilli() instead of UnixNano()/int64(time.Millisecond).
The header is now set unconditionally, matching the overwrite = true the
RabbitMQ interceptor conf uses, so a publisher-supplied value is replaced
rather than kept -- covered by a test, along with the value being the broker
store time rather than anything the publisher sent.
Reported to resolve TestHeaders_Consume in tests/integration. Not re-verified
here: that suite needs the rig plus a configured RabbitMQ, and it still owes a
run for the preceding three fix sweeps.
Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
miotte
force-pushed
the
139-natsjs-jetstream-connector
branch
from
July 21, 2026 20:27
44744c6 to
1eb2750
Compare
Signed-off-by: Richard Sugg <richardsugg@gmail.com>
A durable that drops SingleActiveConsumer has its priority-group config cleared immediately by CreateOrUpdateConsumer (confirmed live: PriorityPolicy=0, PriorityGroups=[] right after the update), but the server's own pin is runtime state independent of that config. A plain (non-priority-group) pull against the downgraded consumer received nothing -- NumWaiting=1, NumPending=1, zero deliveries -- until the previous pin's TTL elapsed (NATSJS_SAC_PINNED_TTL, default 1m), with no error or warning anywhere. The upgrade direction (plain -> SAC) was already tested and works; the reverse had no test in either direction. Worse: once the clearing update has already run, explicitly unpinning the group (Stream.UnpinConsumer, exported since nats-server 2.11) itself fails with "priority group does not exist for this consumer" -- verified against the embedded server before writing any fix. So the release has to happen before the update, not after. releaseStalePins reads the durable's live config first (a plain lookup; ErrConsumerNotFound on an ordinary first-ever subscribe costs nothing) and unpins whatever priority group it still carries, called from Subscribe only when a durable subscribe does not want SingleActiveConsumer. TestSingleActiveConsumerDowngradeReleasesPinPromptly runs at the connector's real default TTL rather than a shortened test-only one, so it proves the wait is gone rather than just faster; proven to fail (5s timeout, no delivery) with the fix's call site stubbed out. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
amqp091's streamSubscribe stamps x-current-offset on every STREAM-source delivery from the RabbitMQ Streams consumer's own offset (amqp091.go:1279); natsjs never carried it, a plain parity gap. The JetStream analogue of that per-reader position is the message's own stream sequence, converted through the existing offsetOf helper -- the same Offset vocabulary SourceStats already reports and accepts, so a value read here and handed back as a source's Offset resumes at the same message. Guarded to STREAM sources only, matching amqp091 exactly: queueSubscribe never sets this header, so a QUEUE/TOPIC source must not carry it either. Both directions covered by TestStreamDeliveryStampsCurrentOffset and TestQueueDeliveryOmitsCurrentOffset; the STREAM test proven to fail (empty header) with the fix's guard disabled. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
amqp091's streamSubscribe decompresses a gzip-tagged body and strips the header (amqp091.go:1286-1292); natsjs passed the compressed bytes straight through -- body corruption, not just a missing header, for whatever consumer read it. Root-caused before porting the fix over blindly: amqp091's OWN STREAM publish path transparently gzips a body that exceeds the RabbitMQ-Streams client library's ~1MiB framing limit (compression.go), a workaround for that library's own ceiling. JetStream's max_payload has no equivalent (deploy-configured to 8MB here), so natsjs correctly never compresses on publish -- it was never going to hit the RabbitMQ problem this exists to solve. But the decompression check is a plain Transfer-Encoding: gzip value match, not tied to who set it, so a publisher that gzips its own body and sets the header itself needs the same treatment on both brokers -- and that publisher-driven case is the one natsjs actually needs to handle. decompressBody mirrors amqp091's own implementation. Guarded to STREAM sources only, matching amqp091 exactly: QUEUE/TOPIC sources pass the header and body through untouched on both connectors, covered by TestQueueGzipBodyPassesThroughUntouched. TestStreamGzipBodyIsDecompressed proven to fail (raw gzip bytes, Transfer-Encoding header still present) with the fix's branch disabled. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
amqp091's queueSubscribe starts (or continues, if the publisher already set traceparent/tracestate) a span for every delivered message and, when tracing is enabled, writes that span's W3C trace context back into the headers the consumer receives -- so a distributed trace always records arke's own "received from broker" hop, even from a publisher that set no trace headers at all. natsjs had zero tracing instrumentation, so that hop was invisible on this connector. Deliberately narrow in scope: amqp091 has roughly a dozen other tracing calls (spans around publish/ack/nack/subscribe-setup), which are arke's own observability instrumentation, not something a downstream client observes or a broker-parity question. Porting all of them would be a much larger, more invasive OTEL-instrumentation change for gaps nobody has flagged; this ports only the one consumer-visible span amqp091 actually adds on delivery. TestDeliveryInjectsTraceHeadersWhenTracingEnabled brackets the process-global tracing.Enabled() flag carefully (flips it via InitTracerProvider, restores the disabled state via defer) since other tests in this package assert exact header sets and would break under a leaked global; proven to fail (empty traceparent) with tracing.Enabled() short-circuited to false at the injection site. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Found by diffing this branch against upstream main: a sibling commit there (retry procedure hardening) added a fallback -- when the delayed- retry path (Retry) returns an error, the server now nacks the same uuid instead of leaving it (server.go). That fallback only works if the uuid is still resolvable when it fires. natsjs's Retry used takeMessage, which deletes the activeMessages entry before attempting NakWithDelay -- the same shape of bug DeadLetter had before its own fix (a46a8e2): if the nak fails, the uuid is already gone, so the new fallback nack finds nothing and errors again, and the message sits ack-pending until AckWait expires on its own rather than redelivering promptly. Retry now peeks the entry, marks it redeliverable (so the fallback nack means "put it back", not "reject" -- a freshly delivered message's redeliverOnNack defaults to false, and a failed retry is a request to try again), and only deletes it once NakWithDelay actually succeeds. Ports the matching 3-line addition to server.go's Ack/Nack dispatch, provider-agnostic and additive. TestRetryFailureKeepsMessageResolvable forces the failure the same way TestDeadLetterFailedTermKeepsMessageResolvable does (ack the underlying message directly, so NakWithDelay returns ErrMsgAlreadyAckd), and is proven to fail against the old takeMessage-based Retry before this fix. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
…edge branches Add regression tests for paths the existing suite left uncovered, each at the intersection of features added in separate sweeps or on a branch no prior test drove: - TestHeaderFilterMatchesWireEscapedHeader: a headers (FILTER) source whose match key is not an RFC 7230 token and whose value would be trimmed by http.Header.Write must still match exactly the messages a RabbitMQ headers exchange would. The filter runs on the decoded header map, so the wire escaping has to round-trip transparently before the match; proven to fail (delivery times out) with escaping disabled. The escaping tests and the header-filter tests never covered the composition together. - TestStreamGzipDecompressFailureKeepsRawBody: a STREAM body mislabelled Transfer-Encoding: gzip that is not valid gzip is forwarded raw with the header kept, matching amqp091's streamSubscribe. Only the success path was covered before. - TestDecompressBody: drives decompressBody's three outcomes directly — valid round-trip, non-gzip (reader-header error), and a truncated payload (the io.ReadAll error branch neither delivery test reached). - TestEvaluateFilters: add an x-match=any-all-miss case and a no-matches filter (the len(matches)==0 short-circuit), taking filterMatches to 100%. - TestFilterSubjectsForStreamSourceNoSubjects: a subjectless STREAM source reads the whole log (boundNothingSubjects' STREAM branch). - TestNatsToPbHeaderSkipsEmptyValueSlice: a raw NATS header key with no values is skipped, not indexed out of range. filterMatches / boundNothingSubjects / natsToPbHeader now 100%; decompressBody 88.9% (residual is the r.Close() error line, not deterministically triggerable); package 94.3%. No production code change. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
…lake
TestDurableOffsetPinnedAtCreation could fail intermittently with "timed out
waiting for a message" on its re-subscribe. It needed two things to coincide:
1. The first subscription reads ahead and pulls m1 into flight before the
test cancels it. PrefetchCount 1 makes this uncommon but not impossible
-- once m0's handler returns, the puller is free to fetch m1.
2. The teardown nak that would make m1 promptly redeliverable does not take
effect. releaseInFlight naks a durable's abandoned deliveries for exactly
this reason, but it is best-effort: as its own comment notes, a nak that
loses its race with the unsubscribe is redelivered into the dying pull
request and sits unclaimed until AckWait.
AckWait then defaults to 30s, which outlasts recv's 10s timeout, so the
re-subscribe gave up before m1 came back. Reproduced deterministically by
holding the first subscription open long enough to fetch m1 while suppressing
the teardown nak; with those two conditions the assertion never completed, and
pinning AckWait below recv's timeout makes it pass in ~0.5s.
The subject of this test is the pinned start position, not redelivery latency,
so it should not depend on that timer at all. TestTeardownReleasesInFlightMessages
deliberately keeps the real 30s AckWait and uses the 10s-vs-30s gap as its
assertion that the nak works, so the nak path stays covered and a genuine
regression there is still caught.
Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
Signed-off-by: Richard Sugg <richardsugg@gmail.com>
Signed-off-by: Richard Sugg <richardsugg@gmail.com>
A consumer that attaches to a source's dead-letter queue after messages have been dead-lettered received none of them. Dead-lettering itself worked -- the copies were stored in the dead-letter stream -- but the reader could not see them. A dead-letter reader is transient by convention, which makes it an ephemeral consumer starting at DeliverNew: the stream tail, behind the very messages it exists to read. The amqp091 connector does not have this problem, and the reason is not a broker feature. Its setupDeadLetter eagerly declares a queue named <source>.dlq bound to the dead-letter exchange as soon as a source declares a DeadLetterAddress, before anything can be dead-lettered, so that queue accumulates from declare time and a reader attaching later finds the messages waiting. Clients attach by that conventional name. So the faithful start position is neither the tail at attach time (what this connector used, which misses everything) nor DeliverAll (which would replay history a transient queue never held). It is the tail as of declare time. Mirror the other connector: pre-declare a durable consumer for the dead-letter queue when a source declares its DeadLetterAddress, and let a later transient reader of that name adopt it and inherit the position it has been holding. Best-effort, as it is there: a failure to pre-declare warns and leaves dead-lettering itself working. Two deliberate choices: Adoption is gated on marker metadata, not on the name alone, so a transient source can never hijack an unrelated durable that happens to share its name and take over its pending messages and ack position. The marker is carried forward when attaching, because attaching re-asserts the whole consumer config and one omitting Metadata erases it -- after which the queue is unadoptable and the next reader silently falls back to the tail. The pre-declared consumer expires after NATSJS_DLQ_DECLARE_TTL (default 1h) where the amqp091 queue never expires. Exact parity would leak: clients commonly name transient consumers with a fresh UUID, so a transient source with a DeadLetterAddress would strand one permanent durable per connection. Set NATSJS_DLQ_DECLARE_TTL=0 for exact parity where source names are stable. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
… one Adoption asked the server for a consumer on every transient subscribe, and for all but a dead-letter reader the answer is always "consumer not found". That is a management round trip per subscribe, and because the answer is an error response it also inflates the JetStream API error counter that deployments watch as a health signal -- worst during the reconnect storm after a restart, when every transient consumer re-subscribes at once and that signal matters most. Check the source name first. This is exact rather than a heuristic: the only durables pre-declared here are named streamNameFor(<source>.dlq) and streamNameFor is injective, so a match requires the reader's own name to carry the suffix. No lookup is skipped that could have hit. Signed-off-by: Michael Otteni <MichaelGOtteni@gmail.com>
…efore The pre-declared dead-letter consumer expired after a fixed hour with nothing reading it, while its stream retains messages for MaxAge (72h by default). That left a window where the dead letters were still stored but the queue holding them was gone -- and a reader attaching in that window falls back to an ephemeral at the stream tail and silently sees nothing, which is the exact failure pre-declaring it is meant to prevent. The window was not an edge case: a dead-letter reader attaches AFTER the failures, because finding out what broke is why it is attaching. An hour is far shorter than the gap between "it broke overnight" and "someone looked", so the fixed default missed the case that motivates the feature. Default to the stream's own retention instead, so the queue never expires while there is still something in it to read. This closes the window by construction rather than by guessing a duration, and it tracks whatever retention a deployment actually configures. A stream set to retain forever now yields a queue that never expires, consistently. Bounding it at all still matters -- an unbounded threshold strands one permanent durable per connection for clients that name transient consumers with a fresh UUID -- so this trades only the parity past the point where the messages themselves are gone. 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
3 participants
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
#→>,*→*)QUEUEaddress)ParentAddress)Offset(first/continue/last/next/numeric)DeliverPolicy+OptStartSeqExpires(x-expires)InactiveThresholdTerm()— a rejection, exactly as amqp091 nacks withrequeue=falserequeue_delayNakWithDelayTerm()x-retry-count/x-deathNumDeliveredNats-Msg-Id+ stream duplicates windowSourceStats)SourceStats)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. Address, routing-key, and stream/durable name
mapping is injective end to end: tokens with reserved characters take an
escaped form that decodes unambiguously, so distinct AMQP names can never
merge onto one subject, stream, or durable. Conventional dotted names pass
through unchanged. The resulting subject layout is part of the connector's
persistence contract — see the design doc.
Publish contract
Streams are asserted on use: a subscribe (or declare-only call) creates the
address's stream, and so does a publish — except a unary publish to a
STREAMaddress, which requires the stream to already exist, as the amqp091connector's stream publisher does (declaring a stream is its readers' job).
A typo'd stream address therefore errors instead of silently minting a junk
stream. The streaming publish path refuses
Confirm(amqp091 parity) andreplies exactly once per message;
PublishIdwithoutPublisherName, anddedup on a
QUEUEaddress, are refused with amqp091's wording.Consumer start position
The
Offsetsource option accepts the same vocabulary as the amqp091connector's Streams support:
firstmaps to deliver-all,lastto the finalmessage,
next(or unset) to new messages only, and an absolute number tothat position.
continueresumes from the stored position — the durable'sack floor, including for a group-less stream source, which gets a durable
named after itself so its position survives, matching RabbitMQ's server-side
offset tracking. Unrecognized values fail the subscribe instead of silently
starting the consumer at a different position. Numeric offsets count from 0
exactly as amqp091 reports them, and a
SourceStatsoffset round-trips backthrough the
Offsetoption without skipping a message. JetStream fixes adurable'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).
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 (scoped to endpoint + credential
identity), 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.
Topology can change out from under a live NATS connection in ways an AMQP
connection would not survive: a stream deleted server-side is re-asserted and
the publish retried once (never for the declared-stream unary path above); a
consumer deleted or expired server-side ends the subscription with a
non-fatal error so the client's re-subscribe rebuilds it, instead of the
consume machinery stopping silently. Subscription teardown releases only its
own in-flight messages, so competing consumers sharing a durable do not
redeliver each other's messages on shutdown; competing subscribers whose
proxy-side header filters disagree are refused rather than silently dropping
each other's messages. A client-initiated
Disconnectleaves the consumestream open (amqp091 parity): stragglers' acks are answered per-ack with the
same error string, not end-of-stream.
Connection lifecycle
Connection state is tracked via the NATS lifecycle callbacks
(connect / reconnect / disconnect / closed) so
WaitForConnectreflects thereal link instead of optimistically reporting connected. Concurrent
Connectcalls for one client are serialized; a redundant dial is closed instead of
leaking a reconnecting connection.
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/reject semantics, dead-letter (includingfailure-path resolvability), dedup, header filtering, durable-backlog resume,
retained-log offset replay (
first/last/next/ numeric, withindependent per-consumer positions), single-active failover, stream/consumer
loss recovery, and coexistence of prefix-related and escaped addresses.
Helper unit tests cover the subject/wildcard mapping, durable-name selection,
and filter evaluation. Package coverage is >90%. The
tests/integrationsuite also runs against a NATS broker unchanged (
ARKE_BROKER_TYPE=natsjs);remaining failures there are RabbitMQ-specific assertions (management-API
wording, broker-injected headers) and the documented TTL→DLX and
start-position differences below.
Known limitations / follow-ups
MessageTTLis accepted for interface compatibility but notapplied — retention comes from the stream-wide
NATSJS_STREAM_MAX_AGE/NATSJS_STREAM_MAX_BYTESconfiguration (one stream per address root). NoJetStream topology can expire a message into a DLX (there is no expiry
advisory or routing hook), so RabbitMQ's TTL→dead-letter idiom has no
equivalent; the design doc records the decision and rejected alternatives.
consumer attaches should be durable: an ephemeral consumer starts at the
stream's tail, while a RabbitMQ queue stores from declaration and a late
consumer reads from the head. No start position is faithful in both
directions; the design doc records why.
ephemeral consumer (fan-out), where RabbitMQ auto-delete consumers compete
and an exclusive queue refuses the second consumer.
follow-up.
embedded server).