kiq jobs with unique names, so they won't be created twice

[gegnew]

I can't seem to find anything about this in the docs. I'd like to assign a job (not a task) a unique name, so that if I were to try to kiq a job with the same parameters twice, the broker would ignore the duplicate job. Something like:

@broker.task
async def foo(param):
    print(param)
    await asyncio.sleep(1)

if __name__ == __main__:
    params = ['a', 'a', 'b', 'c']  # the second "a" should be skipped
    for p in params:
        await foo.kiq(p, job_id=f"job_{p}")

>>> "a"
>>> "b"
>>> "c"

Is this possible?

[gegnew]

Clearly the AsyncBroker can take an alternative task_id_generator, but it doesn't seem possible to pass a custom task_id anywhere, nor is there uniqueness handling, I think?

[s3rius]

Hi, @gegnew and thanks for your interest in the project. You're right. We don't handle uniqueness of ids. Ids meant to be unique only during task execution to not save calculation result twice.

But here's what you can do. You can create a middleware that tracks ids so they are unique among tasks with specific name.

from taskiq.abc.broker import AsyncBroker
from taskiq.message import TaskiqMessage
from taskiq_redis import ListQueueBroker
from taskiq import TaskiqMiddleware
from redis.asyncio import Redis


class UniqueIdMiddleware(TaskiqMiddleware):
    def __init__(self, redis_url: str) -> None:
        self.pool = Redis.from_url(redis_url)

    def set_broker(self, broker: AsyncBroker) -> None:
        return super().set_broker(broker)

    async def pre_execute(self, message: TaskiqMessage) -> TaskiqMessage:
        if await self.pool.get(f"unique:{message.task_name}:{message.task_id}"):
            raise ValueError("Task is already running")

        await self.pool.set(f"unique:{message.task_name}:{message.task_id}", 1, ex=20)
        return message


broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
    UniqueIdMiddleware("redis://localhost:6379")
)


@broker.task
async def my_task():
    print("I'm a task")


async def main():
    await my_task.kicker().with_task_id("2").kiq()
    await my_task.kicker().with_task_id("2").kiq()


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

I won't implement this functionality on broker's behalf, because it doesn't meant to be like this. If you need it, you can easily implement it on your own.

[osttra-o-rotel]

Hej,

When I tried this method, every time the ValueError gets thrown, it crashes the task runner, and it doesn't acnkowledge the message in the queue. So when the task worker restarts, I get the same task message again.

What would be the proper way to drop the message from the queue?

[s3rius]

Okay, I will take a look at it a bit later. Will try to come up with the fix or a new way of calling middlewares.

[osttra-o-rotel]

Hej s3rius!

Any news on how to skip tasks from running from the middleware? I'm thinking changing/adding methods to the middleware tha tallows messages to be skipped, or raising a specific Skip Exception that get's handled differently where the middleware is called?

[tuokri]

@broker.task(schedule=[{"cron": "*/1 * * * *"}])
async def arbitrarily_long_task() -> None:
    # Depending on external factors, this task could last from a few minutes up to to ~30.
    # May start an arbitrary number of other tasks. 
    if await task_already_running():
        return
    
    # Long running work...

I have a long running task that frequently checks whether certain external conditions are met and potentially launches a series of other long running tasks.
The task should only run if another instance of the task is not already running.

Currently I am setting a Redis key when the task starts and checking the existence of said key every time with task_already_running to prevent multiple instances of the task running in parallel.

A built-in way of doing this would be useful.

[magnus-kr]

it would be very useful to have some sort of idempotency when queueing tasks, and it'd even help with scheduled tasks, by having multiple instances of the scheduler running we don't have a single point of failure and don't risk the same task being executed more than once

[epicwhale]

it would be very useful to have some sort of idempotency when queueing tasks, and it'd even help with scheduled tasks, by having multiple instances of the scheduler running we don't have a single point of failure and don't risk the same task being executed more than once

cannot agree more! this would be a great addition to the core library to simplify the developer experience of using it.

[d3vyce]

Having had the same problem, I created Taskiq Deduplication. It's a Redis-backed deduplication middleware for Taskiq that prevents duplicate tasks from being queued or executed concurrently.
Give it a try, and don't hesitate to open an issue if it doesn't solve your problem or if a feature is missing!