Skip to content

Bug:NATS PullSubscriber works not clear with max_workers > 1 #2563

Description

@diskream

Describe the bug
Max_workers with value > 1 consumes more messages than value. The correct number of messages is processed at the same time, but the part that exceeds max_workers is recognized by NATS as being sent to the client and then redelivered, resulting in duplicate messages

How to reproduce
MRE:

import uuid

import asyncio
from nats.js.api import RetentionPolicy

from typing import AsyncGenerator

from fast_depends import Depends
from faststream.nats import JStream, NatsBroker, NatsMessage, PullSub

stream = JStream(
    "test",
    subjects=["test.>"],
    retention=RetentionPolicy.WORK_QUEUE,
)

broker = NatsBroker()


async def progress_sender(message: NatsMessage) -> AsyncGenerator[None]:  # noqa: RUF029, RUF100
    async def in_progress_task() -> None:
        while True:
            await asyncio.sleep(5.0)
            await message.in_progress()

    task = asyncio.create_task(in_progress_task())
    yield
    task.cancel()


@broker.subscriber(
    "test.processing",
    durable="test-worker",
    pull_sub=PullSub(batch_size=2, batch=True),
    stream=stream,
    dependencies=[Depends(progress_sender)],
    max_workers=2,
)
async def test(msg: str) -> None:
    print(f"Received message {msg}")
    await asyncio.sleep(30)


async def main() -> None:
    await broker.start()
    await asyncio.gather(
        *[broker.publish(f"{uuid.uuid4()}", "test.processing") for _ in range(4)]
    )

    while True:
        await asyncio.sleep(1)


if __name__ == "__main__":
    asyncio.run(main())

when I specify max_workers=1 then everything works as expected

Expected behavior
Only max_workers amount of massages can be retrieved from NATS

Observed behavior
More than max_workers is being retrieved from NATS resulting redelivery and message duplication

Environment
Running FastStream 0.6.0rc4 with CPython 3.13.5 on Darwin

Additional context
Provide any other relevant context or information about the problem here.

Activity

  1. added
    CoreIssues related to core FastStream functionality and affects to all brokers
    on Oct 9, 2025
  2. Lancetnik commented on Oct 9, 2025

    @Lancetnik
    Member

    I think we should try to find a workaround to move the lock to reading phase instead of processing.
    Also looks related to #2531

  3. szkytk commented on Oct 14, 2025

    @szkytk

    @Lancetnik

    I think we should try to find a workaround to move the lock to reading phase instead of processing.
    Also looks related to #2531

    I think this is a bug in the core functionality. Do you have an estimated timeline for a fix?

    On a separate note, I've also observed a similar issue with Redis Stream, where tasks are canceled before they can finish processing a message during shutdown, similar to what's described in #2462.

  4. Lancetnik commented on Oct 14, 2025

    @Lancetnik
    Member

    @szkytk I am not sure about timeline - I am busy a lot of AG now and also have a bugs with higher priority. But, I hope to fix the problem right after the hottest things fix.

  5. added theissue type on Aug 2, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    CoreIssues related to core FastStream functionality and affects to all brokersbugSomething isn't working

    Type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions