NATS Work Queue handles messages synchronous

[theobouwman]

I am trying to setup multiple tasks handlers with NATS which will hook into messages like user.signup, user.signin, user.payment etc. So I created a test.

In this task user.signup I will do normal work etc. This task should be executed exactly once, therefor the RetentionPolicy.WORK_QUEUE policy. This works all fine as expected. Tasks are handled once a worker is connected.

But these messages are handled synchronously... I expected the handler function to be called multiple times without waiting for earlier tasks to finish. In the output you can see that the handler waits for each task to be finished before handling new tasks.

Here is my code, I probably misconfigured something I guess. Help really appreciated. I am currently learning NATS :)

import asyncio
import random
from typing import Annotated
from faststream import FastStream, Logger
from faststream.nats import JStream, NatsBroker, RetentionPolicy, DeliverPolicy

from dishka.integrations.faststream import FromDishka, inject, setup_dishka
from common.services.user_service import UserService

from di import container

from common.config import get_config
from common.schemas.worker import NewUserSignupPayload

broker = NatsBroker(get_config().NATS_SERVER_URL)
app = FastStream(broker)

setup_dishka(container, app)

stream = JStream(
    name="momo", 
    retention=RetentionPolicy.WORK_QUEUE,
    declare=False
) # TODO: declare=False to prevent stream creation???


@broker.subscriber(
    "user.signup",
    queue="worker", # needed to be able to have multiple workers for the same message
    stream=stream,
    deliver_policy=DeliverPolicy.ALL,
)
@inject
async def handler(msg: NewUserSignupPayload, logger: Logger, user_service: Annotated[UserService, FromDishka()]):
    logger.info(msg)
    try :
        random_seconds_between_1_and_5 = random.randint(1, 5)
        await asyncio.sleep(random_seconds_between_1_and_5)
        await user_service.get_user(msg.user_id)
    except Exception as e:
        logger.error(e)
    

And the output:

2024-03-14 22:17:33,453 ERROR    - momo | worker | user.signup | ceb8473d-1 - 
2024-03-14 22:17:33,456 INFO     - momo | worker | user.signup | ceb8473d-1 - Processed
2024-03-14 22:17:33,457 INFO     - momo | worker | user.signup | 86d861fb-5 - Received
2024-03-14 22:17:33,459 INFO     - momo | worker | user.signup | 86d861fb-5 - user_id='uid-1'
2024-03-14 22:17:34,476 ERROR    - momo | worker | user.signup | 86d861fb-5 - 
2024-03-14 22:17:34,480 INFO     - momo | worker | user.signup | 86d861fb-5 - Processed
2024-03-14 22:17:34,481 INFO     - momo | worker | user.signup | e3d49acc-e - Received
2024-03-14 22:17:34,483 INFO     - momo | worker | user.signup | e3d49acc-e - user_id='uid-2'
2024-03-14 22:17:38,496 ERROR    - momo | worker | user.signup | e3d49acc-e - 
2024-03-14 22:17:38,499 INFO     - momo | worker | user.signup | e3d49acc-e - Processed
2024-03-14 22:17:38,500 INFO     - momo | worker | user.signup | f0139ec0-1 - Received
2024-03-14 22:17:38,503 INFO     - momo | worker | user.signup | f0139ec0-1 - user_id='uid-3'
2024-03-14 22:17:43,517 ERROR    - momo | worker | user.signup | f0139ec0-1 - 
2024-03-14 22:17:43,521 INFO     - momo | worker | user.signup | f0139ec0-1 - Processed
2024-03-14 22:17:43,521 INFO     - momo | worker | user.signup | 2c20af38-d - Received
2024-03-14 22:17:43,523 INFO     - momo | worker | user.signup | 2c20af38-d - user_id='uid-4'
2024-03-14 22:17:44,536 ERROR    - momo | worker | user.signup | 2c20af38-d - 
2024-03-14 22:17:44,540 INFO     - momo | worker | user.signup | 2c20af38-d - Processed
2024-03-14 22:17:44,541 INFO     - momo | worker | user.signup | 9a79c42a-1 - Received
2024-03-14 22:17:44,542 INFO     - momo | worker | user.signup | 9a79c42a-1 - user_id='uid-5'
2024-03-14 22:17:49,559 ERROR    - momo | worker | user.signup | 9a79c42a-1 - 
2024-03-14 22:17:49,563 INFO     - momo | worker | user.signup | 9a79c42a-1 - Processed
2024-03-14 22:17:49,565 INFO     - momo | worker | user.signup | ce21cf83-9 - Received
2024-03-14 22:17:49,569 INFO     - momo | worker | user.signup | ce21cf83-9 - user_id='uid-6'
2024-03-14 22:17:51,585 ERROR    - momo | worker | user.signup | ce21cf83-9 - 
2024-03-14 22:17:51,589 INFO     - momo | worker | user.signup | ce21cf83-9 - Processed
2024-03-14 22:17:51,589 INFO     - momo | worker | user.signup | 0cdb44e8-b - Received
2024-03-14 22:17:51,592 INFO     - momo | worker | user.signup | 0cdb44e8-b - user_id='uid-7'
2024-03-14 22:17:52,607 ERROR    - momo | worker | user.signup | 0cdb44e8-b - 
2024-03-14 22:17:52,611 INFO     - momo | worker | user.signup | 0cdb44e8-b - Processed
2024-03-14 22:17:52,612 INFO     - momo | worker | user.signup | faf34528-2 - Received
2024-03-14 22:17:52,614 INFO     - momo | worker | user.signup | faf34528-2 - user_id='uid-8'
2024-03-14 22:17:55,626 ERROR    - momo | worker | user.signup | faf34528-2 - 
2024-03-14 22:17:55,630 INFO     - momo | worker | user.signup | faf34528-2 - Processed
2024-03-14 22:17:55,631 INFO     - momo | worker | user.signup | 56f84e51-9 - Received
2024-03-14 22:17:55,633 INFO     - momo | worker | user.signup | 56f84e51-9 - user_id='uid-9'
2024-03-14 22:17:56,654 ERROR    - momo | worker | user.signup | 56f84e51-9 - 
2024-03-14 22:17:56,660 INFO     - momo | worker | user.signup | 56f84e51-9 - Processed

[theobouwman]

Okay I read the docs better (https://faststream.airt.ai/latest/nats/examples/direct/?h=scaling). Indeed by setting max_workers=5 solves the issue. But shouldn't this be handled automatically by the FastStream instance based on CPU/RAM load?

[Lancetnik]

I don't like any kind of magic, so I think, user should setup this option manually. Sequential messages consuming is an expected behavior by default for me.

[theobouwman]

@Lancetnik I get that. How would you, or do you have any articles where I can read about it, set the max_workers? My service is going to handle multiple different topics. One more important to be handled first than others, so curious on how to tackle this.

[Lancetnik]

In my opinion max_workers is a bad way to scale your processing. I prefer to scale processing services horizontally. But, if you want to use 100% of your service hardware, you can set workers as much as your CPU allows you (priority subjects should has more workers than others)

[theobouwman]

Thanks!