NATS JetStream source can duplicate ingestion with a single durable consumer cursor

[almostintuitive]

Describe the bug

A NATS JetStream source can ingest duplicate rows when one logical source is expanded into multiple actor-owned synthetic splits even though ACK progress is tracked by a single durable consumer cursor.
This happens when parallelism is >=2 and we have more than 1 views attached to a single source.

Error message/log

JetStream consumer state showed ack_floor stuck at 0 while num_redelivered increased, and RisingWave source tables accumulated duplicate rows for the same message key/time.

To Reproduce

First create a NATS-backed source with a minimal JSON payload containing a single property, for example:

CREATE TABLE nats_values (
  value INT
) APPEND ONLY WITH (
  connector = 'nats',
  server_url = 'nats://127.0.0.1:4222',
  subject = 'values.demo',
  stream = 'values-demo',
  consumer.durable_name = 'values-demo-durable'
) FORMAT PLAIN ENCODE JSON;

Then publish messages like:

{"value":1}

When the source runs with multiple actors for the same logical source, JetStream may repeatedly redeliver the same earliest unacked messages and RisingWave may ingest duplicates of the same { "value": 1 } payload. When parallelism is >1 and there are more than 1 views chained after a single source.

Expected behavior

I expected one stored JetStream message like { "value": 1 } to be ingested once per delivery attempt after checkpoint-safe ACK progression.

Instead, the same message can be redelivered and appended multiple times because multiple actor-owned synthetic splits contend on one durable consumer cursor.

How did you deploy RisingWave?

Generic local/dev deployment against a JetStream-enabled NATS instance.

The version of RisingWave

Observed on current main before the fix in PR #25680.

Additional context

The simplified root cause is a split-model mismatch:

• meta treated NATS as an adaptive-split source and expanded one discovered split per actor
• the NATS connector delegated progress to a single JetStream durable consumer instead of maintaining independent per-split offsets in RisingWave state
• this let multiple actors interfere with ACK progress on one durable consumer and triggered repeated redelivery

[tabVersion]

Thanks for the detailed report. I checked the current NATS source implementation and this diagnosis matches the code, with one important nuance: RisingWave's NATS JetStream source should be treated as at-least-once, not exactly-once.

In the current implementation, RisingWave does not use the JetStream sequence number as a checkpointed replay offset. NatsSplit::update_offset is effectively a no-op and leaves progress to the broker-side durable consumer, while the source message offset is the JetStream reply subject used later for ACK. The ACK is sent only after the RisingWave checkpoint epoch is committed. That avoids data loss, but it still allows redelivery/duplicates if the ACK fails, times out, or otherwise does not advance the durable consumer state after RisingWave has already ingested the rows.

There is also a parallelism/split-model issue here. Before #25680, NATS was advertised as an adaptive-split source: the enumerator returned one template split, then meta expanded it to one synthetic split per source actor. With a single durable consumer cursor, multiple actor-owned readers can contend on the same broker-side progress and amplify duplicate deliveries. #25680 fixes that by no longer enabling adaptive splits for NATS and keeping the discovered NATS split fixed instead of cloning it per actor.

As a mitigation before the fix is available, reducing the NATS source parallelism to 1 is reasonable, for example by setting streaming_parallelism_for_source = 1 before creating the source or using ALTER SOURCE ... SET PARALLELISM = 1 after creation. Reducing downstream MV parallelism to 1 can also reduce downstream concurrency and make behavior easier to reason about, but it does not change the source ACK/redelivery semantics and should not be described as an exactly-once guarantee.

[almostintuitive]

@tabVersion : thanks for the reply!
I'm slightly confused, is there a fix already on main that's not yet released regarding this?
We can confirm that disabling parallelism does restore the desired ACK behavior. Are there any plans to address this and make NATS parallel consumable in the future?
Is #25144 a fix for this problem?
Thanks!

[tabVersion]

Hi @almostintuitive

A bit more context on the current design: for the NATS JetStream source, RisingWave does not maintain its own per-split offset. Instead, we rely on JetStream’s durable consumer to manage worker delivery and consumer progress. RisingWave stores the JetStream reply subject for each delivered message and sends the final +ACK only after the corresponding RisingWave checkpoint has committed. This is intentional: it avoids data loss if RisingWave fails before the data becomes checkpoint-safe.

The tradeoff is that the source has at-least-once semantics. If JetStream considers a message unacked before RisingWave reaches the checkpoint and sends the final ACK, JetStream may redeliver that message. With multiple workers, this can become more visible because the in-flight / pending message window is larger.

Severe duplicates are most likely when one or more of the following happen:

• consumer.ack_wait.sec is shorter than RisingWave’s checkpoint + processing latency.
• The pending window is large, for example due to high source parallelism, large pull batches, or high consumer.max_ack_pending.
• ACK requests fail, time out, or RisingWave crashes after checkpoint commit but before the ACK is sent.
• Multiple RisingWave sources/jobs accidentally reuse the same JetStream durable consumer name.
• consumer.ack_policy = all is used with multiple workers, making cumulative ACK behavior harder to reason about.

Our recommendation is not necessarily to reduce the NATS source to a single worker, since that would hurt throughput. Instead, the configuration should make JetStream’s redelivery window compatible with RisingWave’s checkpoint-after-commit ACK model:

1. Use explicit ACKs:

   consumer.ack_policy = 'explicit'
   

2. Set consumer.ack_wait.sec comfortably above the expected RisingWave checkpoint latency. As a rule of thumb, it should be larger than:

   checkpoint interval + p99 checkpoint commit latency + processing/backpressure margin

   If checkpointing can occasionally take tens of seconds, ack_wait should be configured accordingly rather than left too small.

3. Limit the in-flight pending window with consumer.max_ack_pending, pull batch size, and source parallelism. The goal is to avoid having a large number of messages expire ack_wait before RisingWave has a chance to
   checkpoint and ACK them.

4. Ensure each RisingWave source/job uses a unique durable consumer name:

   consumer.durable_name = ''

5. Monitor ACK failures/timeouts and JetStream redelivery metrics. If redelivery grows while RisingWave checkpoint latency is close to or above ack_wait, duplicates are expected under at-least-once semantics.

So the short answer is: RisingWave’s ACK path is checkpoint-safe, but not exactly-once. For high-throughput multi-worker NATS consumption, the key is to use explicit ACKs, configure ack_wait larger than checkpoint latency, keep the pending window bounded, and avoid sharing a durable consumer across independent RisingWave jobs.