Redis ListQueueBroker not triggering task in basic FastAPI setup

[ishaan-os]

Hi, I'm very interested in getting Taskiq setup for my FastAPI project but am having trouble setting it up for even a basic test. If anyone could take a look to tell me what i'm doing wrong that would be very helpful.

I'm trying to using Redis for both queuing tasks and as the results backend. It's working successfully as a results backend when I use the InMemoryBroker to kiq, but the task doesn't get triggered when I use ListQueueBroker to kiq. Not sure what I'm doing wrong.

Here's the relevant code in my main.py:

# doesnt work with this broker
broker = ListQueueBroker(settings.redis_url).with_result_backend(
    RedisAsyncResultBackend(settings.redis_url)
)

# works with this broker
# # broker = InMemoryBroker().with_result_backend(
#     RedisAsyncResultBackend(settings.redis_url)
# )

redis_client = Redis.from_url(settings.redis_url)

app = FastAPI()

taskiq_fastapi.init(broker, "app.main:app")

@broker.task()
async def add(x: int, y: int) -> int:
    logging.info("IN TASK") # logged when using in memory broker, not when using list queue broker
    await asyncio.sleep(5)
    return x + y

@app.on_event("startup")
async def startup_event():
    await redis.flushall() # clearing queue
    if not broker.is_worker_process:
        await broker.startup()

    task = await add.kiq(1, 2)
    current_keys = await redis_client.keys(*)
    logging.info(f"Current keys: {current_keys}") # empty when inmemorybroker, but message when list broker not picked up

    results = await task.wait_result() # completes when InMemoryBroker, gets hung when using list queue broker
    updated_keys = await redis_client.keys(*)
    print(f"Updated keys: {updated_keys}") # shows results message for completed task when using in memory broker

@app.on_event("shutdown")
async def shutdown_event():
    if broker.is_worker_process:
        await broker.shutdown()

Here's my docker-compose for reference as well:

services:
  dev:
    &app
    build:
      context: .
      target: development
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
    volumes:
      - ./app:/app
    ports:
      - 8000:8000
    environment:
      - APP_ENV=development
      - APP_SECRET=secret
      - APP_VERSION=dev123
      - REDIS_URL=redis://redis/
      - DB_URL=postgresql+asyncpg://postgres:postgres@db:5432/postgres
    env_file:
      - .env
    depends_on:
      - db
      - redis

  db:
    image: postgres:15.4
    expose:
      - 5432
    ports:
      - 5432:5432
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=postgres

  redis:
    image: bitnami/redis:6.2.5
    hostname: redis
    restart: always
    environment:
      ALLOW_EMPTY_PASSWORD: "yes"
    healthcheck:
      test: redis-cli ping
      interval: 1s
      timeout: 3s
      retries: 50