NATS WorkQueue unexpected "filtered consumer not unique on workqueue stream"

[theobouwman]

My WorkQueue Pull handlers work fine. But, when I start multiple instances for horizontal scaling we observe a nats.js.errors.BadRequestError: nats: BadRequestError: code=400 err_code=10100 description='filtered consumer not unique on workqueue stream' error. But to my understanding our subjects do not overlap...

MOMO_WORKER_DEFAULT = JStream(
    name="momo-worker-default",
    retention=RetentionPolicy.WORK_QUEUE,
    max_age=60*60*1*3, # 3 hours
    declare=True,
    storage=StorageType.MEMORY,
    allow_direct=False,
)

# 
# DI container
# 
container, _, nats_router = make_container()

# 
# FastStream setup
# 
setup_faststream_ioc(
    container,
    nats_router,
    finalize_container=False,
)

# 
# FastAPI setup
# 
_fastapi_app = FastAPI(
    title="momo-worker",
    lifespan=nats_router.lifespan_context,
    docs_url=get_config().swagger_url(),
    redoc_url=get_config().redoc_url(),
    debug=get_config().DEBUG,
    default_response_class=ORJSONResponse,
)

setup_fastapi_ioc(container, _fastapi_app)
init_observability("momo-worker", _fastapi_app)
init_exception_handlers(_fastapi_app)
init_middleware(_fastapi_app)
firebase_initialization()

# NOTE: subject: ENTITY.DOMAIN.ACTION
# e.g. user.email.summary.send
# e.g. organisation.feed.reset
# e.g. organisation.event.created
# e.g. organisation.event.deleted
# e.g. organisation.event.started
# e.g. organisation.event.ended
# e.g. organisation.member.added

# 
# Handlers
#
 
@nats_router.subscriber(
    "scheduled.bimonthly.user.email.summary.send",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=1,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10),
)
@inject
async def handler(user_service: Annotated[UserService, FromDishka()]):
    users = await user_service.get_all_users()

    tasks = []

    # for user in users:
        # tasks.append(asyncio.create_task(nats_broker.publish(UserEmailSummarySendPayload(user_id=user.uid), "user.email.summary.send")))

    await asyncio.gather(*tasks)


@nats_router.subscriber(
    "user.email.summary.send",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=1,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10)
)
@inject
async def handler(msg: UserEmailSummarySendPayload, user_service: Annotated[UserService, FromDishka()]):
    print(f"Sending email to {msg.user_id}")
    return msg


# 
# Add routers
# 
_fastapi_app.include_router(nats_router)

app = _fastapi_app

[theobouwman]

I think i found the correct setup with setting the durable parameter:

@nats_router.subscriber(
    "user.email.summary.send",
    durable="user-email-summary-send",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=5,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10)
)
@inject
async def handler(msg: UserEmailSummarySendPayload, user_service: Annotated[UserService, FromDishka()]):
    print(f"Sending email to {msg.user_id}")
    return msg

Is this the recommended way to setup a work queue with deliver once?

[Lancetnik]

@theobouwman durable tells NATS server to save your consumer reading position. It works for Pull-based subscription, so you did it correctly (but I am not sure, that DeliverPolicy has any effect in this case).
Btw, durable can't be used with multiple Push-based subscribers in the same time, so u should use queue instead