Waiting in producer for task completion or cancelation

[Qolum]

I'm currently working on a producer-consumer scenario in which my producer runs asynchronous tasks that can either be completed normally or canceled. I want the producer to wait until the task is completed or canceled before moving on to the next piece of code.
How can I do this?

[wlipski]

I'm new to pgqueuer

For now I don't see an option which gives you exact answer and it's rather something you need to build by yourself.

A few possibilities from me:

• Once you schedule a task, enqueue function gives you the ids of the jobs which you could check in pgqueuer table.
  If ids are present - the job is running.
• You could report back to the producer via something (db table, api, etc.)

But better to wait for the answer from someone who is more familiar with pgqueuer

[Qolum]

I do something like this 😅

[janbjorge]

At the moment, PGQueuer does not support bidirectional communication between the producer and consumers. The best option available right now is to poll the log table using the job ID and wait for a status update that matches your criteria. While polling isn't ideal, it's the only viable approach currently.

That said, this sounds like an interesting feature. Out of curiosity, could you share more details about your use case? Why do you need the producer to wait for the job to be processed before continuing? If the goal is to enforce sequential execution or dependency handling, there might be alternative ways to structure this.

Would love to hear more about your specific scenario!

[Qolum]

It's a rare case to wait for an asynchronous task but in my case, it’s a critical part of the process. Here’s some abstract code that demonstrates my use case:

async def get_user_profile(user_id):
    if data :=  await repository.get_profile_info(user_id=user_id):
        return data
    status = await scrap_user_profile_task(user_id=user_id)
    if status == "COMPLETED":
        return await repository.get_profile_info(user_id=user_id)
    elif status == "CANCELED":
        return f"Unable to construct a profile for user:{user_id}"
    else:
        return f"Can't find profile info for user:{user_id}"

In this example, I first check the worker’s result table for the user profile data. If no data is found, I trigger a long-running task to scrape the user’s information and wait for it to complete. Within the task function, I enqueue a task to construct the user profile and continuously check the status in the queue table. If the status is neither “COMPLETED” nor “CANCELED”, I wait briefly before checking again; once the task is completed or canceled, I return the status.

Additionally, it would be beneficial to enhance the enqueue function by introducing parameters such as:

• wait_until_complete=True: This would automatically wait until the task finishes.
• Returning a comprehensive task object that includes all fields represented in the queue table.
  These enhancements could simplify task management and provide more detailed insight into the task's execution state.

[janbjorge]

I see that you're using the queue to ensure the profile is available before proceeding. Typically, a queue is used to offload work asynchronously so the main process doesn't block. Since you're waiting for the task to complete, I'm curious—why not call the scraping function directly instead of enqueuing it?

If you absolutely need to use the queue for this, then polling the log table for status updates is the best workaround in PGQueuer today. That said, adding an option like wait_until_complete=True could be interesting, though it might introduce unintended complexity. Would a direct function call or a synchronous job execution mode (if implemented) work better for your use case?