Error handling and retries

[adrienbrault]

Hey,

How should I handle retries with this library?

Let's say I have a worker/consumer that connects to a remote API, and it is down.

[janbjorge]

Hi,

For retries, you can decide how you want them to behave based on your requirements. If you want to implement exponential backoff, it's possible to do so directly in your entrypoint function by adding delays between retries and re-queuing jobs when needed.

The entrypoint function will be executed in this dispatach function.

pgqueuer/pgqueuer/qm.py

Line 408 in 5b54585

async def _dispatch(

The dispatch function is getting a bit busy, but focus on the try/execpt block of the function.

For producers, its a plain sql-insert. Unsure if there is anything i can help with there?

[janbjorge]

EDIT:

Another approach is to write an EntrypointExecutor that contains the retry logic you want. This way you can reuse it on per entrypoint.

[adrienbrault]

Thank you! I will explore these ideas.

Is there a way to publish a job to be pick up later, say 1h?

[janbjorge]

As of now, no, but it's a feature I've been thinking of. Any suggestions? How would you prefer the queue interface to look? I think there are two options: an execute at (datetime) option, or an execute after (interval). I think execute after would be easier due to clock synchronization?

Suggested update to function signature.

from typing import overload
import datetime

class QueueManager:
    @overload
    async def enqueue(
        self,
        entrypoint: str,
        payload: bytes | None,
        priority: int = 0,
        execute_after: datetime.timedelta | None = None,
    ) -> list[models.JobId]: ...

    @overload
    async def enqueue(
        self,
        entrypoint: list[str],
        payload: list[bytes | None],
        priority: list[int],
        execute_after: list[datetime.timedelta | None],
    ) -> list[models.JobId]: ...

    async def enqueue(
        self,
        entrypoint: str | list[str],
        payload: bytes | None | list[bytes | None],
        priority: int | list[int] = 0,
        execute_after: datetime.timedelta | None | list[datetime.timedelta | None] = None,
    ) -> list[models.JobId]:
        """
        Insert new jobs into the queue with optional scheduling.

        Adds one or more jobs to the queue with specified entrypoints, payloads,
        and priorities. Supports inserting multiple jobs in a single operation
        by accepting lists for the parameters. Jobs can be scheduled to execute
        after a delay.

        Args:
            entrypoint (str | list[str]): The entrypoint(s) associated with the job(s).
            payload (bytes | None | list[bytes | None]): The payload(s) for the job(s).
            priority (int | list[int]): The priority level(s) for the job(s).
            execute_after (datetime.timedelta | None | list[datetime.timedelta | None]):
                The delay after which to execute the job(s).

        Returns:
            list[models.JobId]: A list of JobId instances representing the IDs of the enqueued jobs.

        Raises:
            ValueError: If the lengths of the lists provided do not match when using multiple jobs.
        """

        # If they are not lists, create single-item lists for uniform processing
        normed_entrypoint = entrypoint if isinstance(entrypoint, list) else [entrypoint]
        normed_payload = payload if isinstance(payload, list) else [payload]
        normed_priority = priority if isinstance(priority, list) else [priority]
        normed_execute_after = execute_after if isinstance(execute_after, list) or execute_after is None else [execute_after]

        return [
            models.JobId(row["id"])
            for row in await self.driver.fetch(
                self.qb.create_enqueue_query(),
                normed_priority,
                normed_entrypoint,
                normed_payload,
                normed_execute_after,
            )
        ]

[janbjorge]

@adrienbrault Ive got this PR, could you take a look if looks promising for your use-case?

#195

[adrienbrault]

This seems to be exactly what I need

[janbjorge]

https://github.com/janbjorge/pgqueuer/releases/tag/v0.16.0

[adrienbrault]

Thank you!