Timeout problem in start up in long running tasks

[alpersergeny]

Update!
Hi,
i'll try and give a little context to what we expirienced:

Setup uses Redis as Broker with streams and a task that takes longer then 3 sec to complete.

We have two Scenarios that should behave the same but they don't:
1st:
We we start the subscriber, then we a a task.
Result: Tast gets completed, faststream waits for more tasks.

2nd:
Subscriber is stopped. We add a Task, then start the Subcriber.
Result: Tasts gets completed but faststream stops with the Timeout-Error.

Reason for the timeout is in the screenshot above: the start()-function throws it, bc. it consumes a tast that takes longer then 3 seconds.

Thx for any assistance and help :)

Details :
``Hey,
I have a question about the long running tasks. Firstly I have a small environment for reproducing the issue. I am providing the code here (Code for the publisher):

Code for the sub:

Code files for the reproducing will be added here too

To reproduce the issue publisher needs to work on fastapi and sub needs to have a long running task (longer than 3.0 sec).
Steps to reproduce :

1. Run the fastapi
2. Run Redis
3. Post a task
4. Run the subscriber.
   If the steps followed end result should be something like this :
   
   As I tracked down the issue comes down to this part of the library :
   
   A small modification to anyio.fail_after(3.0) solves the issue for the long running tasks however, I didnt understand the whole part of the project therefore not sure should I do a change on there. Thank you for all the effort and I am attaching an example case to reproduce the error. Please create the envs for it. Publisher needs
   fastapi[standard]
   redis
   faststream[cli]
   Subs:
   redis
   faststream[cli]

test_pub

from faststream.redis.fastapi import RedisRouter
from fastapi import FastAPI
router = RedisRouter(
    url="redis://localhost",
    port=6379,
)

    
description = "Test"
app = FastAPI(
    title="Test",
    description=description,
    openapi_tags=[
        {
            "name": "Test",
            "description": "Test ",
        },
        {
            "name": "Test",
            "description": "Test",
        },
    ],
)
@router.post("/run_test")
async def test_pub(str:str):
    await router.broker.publish(f"Hello World_{str}", stream="test",rpc_timeout=100)
    return {"status": "ok"}
app.include_router(router)

Test_sub:

from faststream.redis import RedisRouter, StreamSub
from faststream.redis import RedisBroker
from faststream import FastStream
import time
broker = RedisBroker(
    url="redis://localhost",
    port=6379,
)
version = "0.1.0"
title = "My FastStream service"
description = "Description of my FastStream service"
app = FastStream(broker, title=title, version=version, description=description)

router = RedisRouter(prefix="")
@router.subscriber(
    stream=StreamSub(
        "test",group="test",consumer=1, max_records=1))
def test_sub(msg:str):
    print(msg)
    time.sleep(3.5)

broker.include_router(router)

[guy-dono]

i have the exact same issue, for now im gonna fork and remove the timeout altogether until it's fixed :
did you find a workaround?

[alpersergeny]

Hey firstly, I found a work around. My main consideration the solution I couldnt understand the timeout part I changed it. Basically I patched it without forking the library so it is a bit easier on container level to work with. I will try to follow the tread and share my solution with you. Hopefully it helps.

from faststream.redis.subscriber.usecase import LogicSubscriber as RedisStreamSubscriber

original_start = RedisStreamSubscriber.start

async def patched_start(
    self,
    *args: Any,
) -> None:
    if self.task:
        return

    await super(RedisStreamSubscriber, self).start()

    start_signal = anyio.Event()

    if self.calls:
        self.task = asyncio.create_task(self._consume(*args, start_signal=start_signal))

        # Modified timeout value here
        with anyio.fail_after(600.0):  # Changed from 3.0 to 10.0
            await start_signal.wait()

    else:
        start_signal.set()

and add this to your main.py (or wherever you start your function)

RedisStreamSubscriber.start = patched_start

Just a headsup with the naming convention from faststream.redis.subscriber.usecase import LogicSubscriber as RedisStreamSubscriber
Edit : I am not sure of the short comings of this solution so beaware 😄