Help. How to add/remove cron jobs dynamically

[superjcvd]

I want to dynamically (from an API) be able to schedule new cron jobs.
But also remove those if needed.

I am able to create new cron jobs like this (FastAPI):

@router.post("/create/")
async def create_job(
    body: Annotated[dict, Body],
    cron: Annotated[str, Query(title="CRON Schedule")] = None,
):
    await Queue().declare()
    if cron:
        task = Job("my_job_here", args=body, result_ttl=timedelta(minutes=5), cron=cron)
    else:
        task = Job("my_job_here", args=body, result_ttl=timedelta(minutes=5))

    routing_key, _, _ = await task.enqueue()

    # result = await task.result
    return {
        "message": "Task enqueued",
        "result_id": task.result_id,
        "routing_key_id": routing_key.id_,
    }

But now how can I stop and remove the job ?

[aleksul]

Hi, unfortunately due to a way Repid schedules messages, and the fact that currently there is a unified ABC for all message brokers, there is no native way to delete/stop a job.
You can make it work by either

• adding some logic to your actor to prevent next scheduling
• or by directly interacting with your broker (e.g. redis) and removing the message from there

I wanted to look into it in #108 but I hadn't had the time and as far as lib development goes I'm mostly focusing on AsyncAPI integration.

[superjcvd]

I understand.
I will try to find a workaround or wait for an improvement :)

Here is what I tried so far:

• I created a few cron repid Job
• Then I went to my Redis instance
• I can see the following entries

127.0.0.1:6379> KEYS *
1) "q:default:5:dead"
2) "q:default:5:d"
3) "m:default:5:test:07f5631aba47426facd49c0f4ef33294"
4) "4c7dadf51859444288accf9dfd4ad077"

It's maybe a stupid question, but which entry should I delete to remove the scheduled task ?
I tried to remove the HASH value "m:default:5:test:07f5631aba47426facd49c0f4ef33294"
But it seems to go the the dead letter queue and comeback by itself 😆

[superjcvd]

Ok I will answer the question mysself if this can help 😄
I need to delete keys: 3 and 4 of my example.
If I delete only the key number 3 it will comeback at the next scheduled time

[aleksul]

• q:default:5:d - this is your delayed queue, it is using a sorted set - https://redis.io/docs/latest/develop/data-types/sorted-sets

Repid periodically checks whether something there is past execution time. So primarily you need to remove your message's id (in the format of {topic}:{id_}) from there, i.e. in Redis you need to execute

ZREM q:default:5:d "test:07f5631aba47426facd49c0f4ef33294"

• m:default:5:test:07f5631aba47426facd49c0f4ef33294 - this is where details of your message are stored in a form of hashmap - https://redis.io/docs/latest/develop/data-types/hashes , you also need to delete it, but before that you can read payload to get your arguments bucket id. The following HGET should return something like {"__repid_payload_id": "{args_id}"}. You can also get your args id from a Job or supply your own.

HGET m:default:5:test:07f5631aba47426facd49c0f4ef33294 payload

DEL m:default:5:test:07f5631aba47426facd49c0f4ef33294

• 4c7dadf51859444288accf9dfd4ad077 - supposably this is your arguments bucket, you can also remove it:

DEL 4c7dadf51859444288accf9dfd4ad077

[superjcvd]

Thanks a lot for your help :)

⚠️ (just in case you didn't know)
Croniter library is going to be abandonned soon 😢
https://github.com/corpusops/croniter