Linked to #6

When a job is re-tried, its started_at, scheduled_at (and one day ended_at?) are overwritten. We should make sure not to lose this data, for example by storing it in a "cabbage_events" table.
This table could be populated when a job is created / started / ended. As we see it, it should contain the following columns:

• A foreign key to a job
• A type of event: schedule, started, ended (more?)
• A datetime with timezone field, containing what is today in cabbage_job.scheduled_at / cabbage_job.started_at
• The current job.attempts (make sure it makes sense for scheduled, started and ended events)

started_at and ended_at (if it exists) may be removed from the job table. scheduled_at is still relevant on the job table. Although it may be renamed into "next_attempt_scheduled_at" (for example).

ping @mgu @ewjoachim

If tasks are to be executed in order (especially when they share the same lock) then we need to always select the first one, pk wise.

Also, add the relevant badges to the README and the doc

We need to be able to retry a task on demand

• SQL side: we need to know the "retry" number on a job

• Python side: define a RetryStategy on a task (exponential/fibonnacci/whatever, cutoff, etc, to implement in a dedicated object), and autoretry or "on_raise=retry".

Add a column with the scheduled date on the job table, and when selecting a job, only consider those whose scheduled date is in the past.

Add a app.defer() method that allows deferring a job even if you don't have the Task object (for example if the task deferring and the task running are done in 2 different codebases)

Signature would be something like:

def defer(self, task_name:str, [all the arguments from task.defer]):

Get a list of Job()s that are:

• in a given queue or not
• with a given name or not
• in the doing state for more than X secondes

For now, it's all fun and games, no one uses Procrastinate yet, and we haven't really thought about migrations, but it's probably time to setup something.

• People might hop in and install procrastinate at any version (starting at 1.0, say)
• Starting from there, people will need to upgrade procrastinate (including the schema)
• Ideally, in a simple manner
• Ideally, without impact and downtime
• Optionally, using https://github.com/peopledoc/septentrion/ given we're developing this tool too.

One way of doing it could be:

• Adapt the migration files to septentrion. Remove procrastinate migrate in favor of septentrion migrate.
• Create a first migration for something trivial, just to test it.
• Document the hell out of it, so that it's easy to understand. Don't forget Septentrion is by far not as properly documented 1. as it should be and 2. as procrastinate is.
• (make it easy for people who want to read migrations instead of executing them)

(Note that in PeopleDoc, we actually don't need this part at all, because our fantastic DBA team will take care of the migrations for us, so this is strictly an effort to make procrastinate more usable by other people)

To be able to quickly setup a development environment, we could wrap the procrastinate demo app in its own container in the docker-compose.yml. This would avoid the use all the exports in the CONTRIBUTING.rst and maybe some of the installs of the README.rst.

The new image should have the necessary variable to connect to the database and use the demo app, and its entrypoint should be the procrastinate command. Of course, it should bind the volume to the procrastinate directory and use a develop install, so that changes of code can be instantly checked on the demo app.

For maximum logging extensibility and effectiveness, here's what we could do:

• Every message is bundled with all the context we can think of as extras, including a unique code for this log message (e.g. "action": "start_job")
• All logs have a simple associated message, that's "good enough" for the job, without being too much
• All loggers defined directly as logging.getLogger("cabbage")
• And we should document how one could implement a log filter, and plug it on cabbage to:
  • Change the log level if they want
  • Change the associated message if they want, including using template variables (loaded from extras)
  • Delete a specific message
  • Implement a specific handler so that these logs are actually not sent to normal logging but, say, written to a file (and then disable bubbling)
  • ...

This way, cabbage logs may be adapted to anyone's need.

Click already has an builtin function to implement the version flag for us.

https://click.palletsprojects.com/en/7.x/api/#click.version_option

Currently, database errors are raised from psycopg2 or aiopg. However, we want to use our own errors, so that we have less dependancy with those librairies.

We need to catch all psyopg2/aiopg errors at the source so that we use our own custom-defined errors, with a from argument to not lose the base error.

This can probably be done inside a decorator that we apply on all aiopg_connector functions.

This issue suggests improving the docs with respect to task names. How are task names used? Should people use specific task names? Or should they rely on default names? We should probably add a section to the "Discussions" chapter to discuss task names.

The test_wait_for_jobs_timeout integration test produces an error with Python 3.8:

    async def wait_for_jobs(connection: aiopg.Connection, socket_timeout: float):
        try:
            await asyncio.wait_for(connection.notifies.get(), timeout=socket_timeout)
>       except asyncio.futures.TimeoutError:
E       AttributeError: module 'asyncio.futures' has no attribute 'TimeoutError'

TimeoutError is no longer in asyncio.futures. And the docs say to use asyncio.TimeoutError anyway, which should work with at least Python 3.5, 3.6, 3.7 and 3.8.

PR to come.

[python -m] cabbage [--app dotted.path.to.app] [-v[v[v]]] worker [QUEUE_NAME[ ...]]
[python -m] cabbage [--app dotted.path.to.app] [-v[v[v]]] migrate
CABBAGE_APP=dotted.path.to.app cabbage ...

This will help integration.

To facilitate edition following the black format, we could add an .editorconfig to the repository, so that code editors will be able to help with formatting from the start.

https://editorconfig.org

While running the tests on Python 3.8 we get a DeprecationWarning:

.../procrastinate/lib/python3.8/site-packages/aiopg/connection.py:90: DeprecationWarning: The loop argument is deprecated since Python 3.8, and scheduled for removal
 in Python 3.10.

We get this warning several times.

In this ticket, we'll make the first part: async defering tasks

The rest will be done in #105 and #106

Errors in quickstart:

• Depending on whether the script is lanched or loaded by procrastinate, main module is named __main__ or tutorial, and procrastinate is lost
• Also, when using procrastinate CLI, tutorial.py is not in the pythonpath
• Missing from time import sleep

Write all the missing parts.

Split how-tos in several pages (maybe join them in the final build, but it would be easier to work with smaller files)

Let users define how they want job payloads to be encoded/decoded, especially if they want to instantiate / format special types.

Ping @hsmett who was up for contributing :)

Python's standard lib has https://docs.python.org/3/library/cmd.html , which will help us implement easily an interactive prompt.

We could probably use the following features:

• List queues (and number of queued & processing jobs per queue, with stats like processing rate, average wait time etc.)
• List tasks (and ... same stats)
• List jobs & search for a specific job (by task, queue, lock, id, maybe args)
• List stalled jobs and error jobs

(it would be nice if those pages could stay on and autorefresh)
(also, use colors ?)

• Display a job, and provide specific actions: discard it, retry it (now or later), manually set it to success or failure (well it's all about changing the state of the job)
• Batch action ?

And finally, the cmd prompt should just be an interface, and the real actions should be implemented independently in an admin module, because we'll probably have exactly the same needs for the API and/or the web admin panel.

It's up to the user to run the task periodically, but we could provide a task to ease this. Something along the way of:

procrastinate defer procrastinate.builtin_tasks.remove_old_jobs max_hours=72 remove_error=false

It's kinda explicit what it does: delete all finished jobs and optionally also error jobs. When removing error jobs, it produces a log that contains all its informations, so that log exploitation could maybe recover them if need be.

I'm not sure whether it's more relevant to use days or hours but hours feels like it's easier for both use cases.

Note: I've used a syntax for cli defer that has not been implemented yet but I'm considering.

What I'd really like is:

• PR merged should automatically add a line in a draft release in GitHub
• Tags created in GitHub should trigger the corresponding version to be released on pypi

(in this schema, both changelog and version releasing are expected not to require any code change)

Ok, we have, I think, our first real bug.

TL;DR

If I launch procrastinate_fetch_job quickly twice, I can get 2 jobs that share the same lock.

Proof

$ psql -c 'SELECT * FROM procrastinate_jobs;'
 id | queue_name |           task_name            | lock |   args    | status | scheduled_at | started_at | attempts
----+------------+--------------------------------+------+-----------+--------+--------------+------------+----------
  1 | sleep      | procrastinate_demo.tasks.sleep | yay  | {"i": 20} | todo   |              |            |        0
  2 | sleep      | procrastinate_demo.tasks.sleep | yay  | {"i": 20} | todo   |              |            |        0
(2 rows)

$ psql -c 'SELECT procrastinate_fetch_job(NULL);' & psql -c 'SELECT procrastinate_fetch_job(NULL);'&
[1] 22792
[2] 22793
                                       procrastinate_fetch_job
-----------------------------------------------------------------------------------------------------
 (1,sleep,procrastinate_demo.tasks.sleep,yay,"{""i"": 20}",doing,,"2019-10-27 20:14:12.230384+00",0)
(1 row)

[2]  + 22793 done       psql -c 'SELECT procrastinate_fetch_job(NULL);'
                                       procrastinate_fetch_job
-----------------------------------------------------------------------------------------------------
 (2,sleep,procrastinate_demo.tasks.sleep,yay,"{""i"": 20}",doing,,"2019-10-27 20:14:12.233998+00",0)
(1 row)

[1]  + 22792 done       psql -c 'SELECT procrastinate_fetch_job(NULL);'

The second call should have returned (,,,,,,,,).

For the sake of it, if I put the 2 calls in the same transaction, I DO have the expected result:

$ psql -c 'SELECT * FROM procrastinate_jobs;'
 id | queue_name |           task_name            | lock |   args    | status | scheduled_at | started_at | attempts
----+------------+--------------------------------+------+-----------+--------+--------------+------------+----------
  1 | sleep      | procrastinate_demo.tasks.sleep | yay  | {"i": 20} | todo   |              |            |        0
  2 | sleep      | procrastinate_demo.tasks.sleep | yay  | {"i": 20} | todo   |              |            |        0
(2 rows)

$ psql -c 'SELECT procrastinate_fetch_job(NULL), procrastinate_fetch_job(NULL);'
                                      procrastinate_fetch_job                                       | procrastinate_fetch_job
----------------------------------------------------------------------------------------------------+-------------------------
 (1,sleep,procrastinate_demo.tasks.sleep,yay,"{""i"": 20}",doing,,"2019-10-27 20:17:46.28228+00",0) | (,,,,,,,,)
(1 row)

Leads

• When we call FOR UPDATE OF procrastinate_jobs, should we somehow also lock procrastinate_job_locks? But there's no row to lock yet...
• Should there be a UNIQUE constraint on procrastinate_job_locks.object? I guess it would help but if we add just that, we'll get a crash, and still not the expected behaviour. Or should we also add a ON CONFLICT DO NOTHING statement? The procedure would probably need quite a refactor, because in this case it needs to not do `UPDATE procrastinate_jobs SET status = 'doing' and return nothing.
• Is it because the procrastinate_job_locks is UNLOGGED?

See #93 (comment)

1. The retry strategy should be provided the exception that triggers the retry
2. A builtin mechanism should allow to retry only if the exception is of a given type or among a few types.

https://procrastinate.readthedocs.io/en/latest/howto/cron.html

Last code example is not displayed

• For devs
• For sysadmins / support teams

Ideally following Daniele Procida's 4 part doc template.

For Cabbage to be used in a real codebase, we'll need a few tools that will make testability easier:

• A pluggable in-memory list of created tasks
• A way to launch these task synchronously
• A way to launch a single task

I'm thinking something along the lines of:

from cabbage.testing import TestTaskManager
tm = TestTaskManager()

something = []

@tm.task(queue="yay")
def t(task_run, a):
    something.append(a)

t.defer(a=1)

# A pluggable in-memory list of created tasks
assert tm.scheduled == [{"name": "t", "args": {"a": 1}]

# A way to launch these task synchronously
tm.run_scheduled()

assert something == [1]

# A way to launch a single task
t.run(a=2)

assert something == [1, 2]

(For now, no idea how we'd substitute the TaskManager with the TestTaskManager in a real case, but we can think of something)

This should:

• Connect to the database and launch a simple query (SELECT TRUE)
• Check the existence (and version?) of the table procrastinate_jobs
• Count the number of jobs in each status

Return a dict describing all this.

def app.monitor(self) -> Dict[str, Any]:

This is the third and last step in asynchronous compatibility.

• #9 is for defering jobs asynchronously
• #105 is for accepting a task to be a coroutine
• This ticket is for executing asynchronous tasks in parallel

For now, our tasks can be asynchronous, but there's still only one task executed at a time. We need a way to launch a pool of workers in the same event loop.
We need to check precisely where the right place is to place our pool. I don't know yet if we want 1 DB connection per process or per worker, and the potential impact on transactions. We'll probably need to setup a connection pool etc.

Note: we'll need some kind of ContextVar for the log context in Worker, because if we start parallelizing tasks, contexts will be ALL OVER THE PLACE. But ContextVar is py>=3.7 and we're supporting 3.6.

What if a worker that is setup to run parallel async tasks recieves a non-async task ? Should it run (and block all the other tasks running at the same time) or not run (but this breaks our model of never taking a task in the DB if we can't run it) ? To be determined before implementation.

Also we'll need to switch to a pool of pg connections, because one single connection cannot hold concurrent queries (we could have a specific parameter for concurrent pg connections vs concurrent workers)

The idea is to register a pipe, then in the select call in postgres.py, listen to the pipe too, and in the signal handler, write in this pipe. This way, we'd leave immediately without waiting for the timeout.

I know we can't do everything in a signal handler, but if we can write to a pipe, then it may help us to stop faster (and make the "timeout" argument really much less important).

https://stackoverflow.com/a/4661284

This is a 2-fold ticket:

• Without a py.typed, our type annotations cannot be used by projects using procrastinate, and it's quite sad. Let's add a py.typed (see https://www.python.org/dev/peps/pep-0561/ )
• Mypy doesn't like our metaprogramming trick @async.add_sync_api and frankly, I understand it makes its job harder. Let's help it by implementing a mypy plugin. We can take direct inspiration from https://github.com/strawberry-graphql/strawberry/blob/4d11e2d26cf3f8f1453c95106606acede85125ac/strawberry/ext/mypy_plugin.py and https://github.com/strawberry-graphql/strawberry/tree/4d11e2d26cf3f8f1453c95106606acede85125ac#type-checking (doc: https://mypy.readthedocs.io/en/latest/extending_mypy.html )

If this ticket is succesful, then from a new python project using mypy, installing procrastinate from PyPI, on the following lines:

app = procrastinate.App(...)
app.run_worker(["hello"])  # should pass mypy
app.run_worker("hello")  # should fail mypy

See #68 (comment)

The execution of the unit tests currently leads to two RuntimeWarning: coroutine 'xxxxxx' was never awaited:

====================================================================================== warnings summary =======================================================================================
tests/unit/test_testing.py::test_listen_for_jobs_run                                                                                                                                           
  /home/elemoine/src/procrastinate/tests/unit/test_testing.py:259: RuntimeWarning: coroutine 'BaseJobStore.listen_for_jobs' was never awaited                                                  
    job_store.listen_for_jobs(queues=["a", "b"])                                                                                                                                               
                                                                                                                                                                                               
tests/unit/test_testing.py::test_wait_for_jobs                                                                                                                                                 
  /home/elemoine/src/procrastinate/tests/unit/test_testing.py:264: RuntimeWarning: coroutine 'InMemoryJobStore.wait_for_jobs' was never awaited                                                
    job_store.wait_for_jobs()

It would be good to fix that.

PR to come.

It would be nice to provide a simple way to disable the warning that triggers everytime you define a task with a name that is not it's python import path. Of course, using a logging filter, people can already do that, but it lacks a bit of user friendliness. I'm thinking about a boolean parameter in App.__init__() ?

https://github.com/peopledoc/procrastinate/blob/81fe265a7a1865db77f93302cd82e8dcf9b22421/procrastinate/tasks.py#L83-L93

It could be interesting to accept that task can return anything, and the return value will be added to the logs.

This way, the logs would naturally be enriched without having to add log to the tasks. Of course, a task could always return None if there's nothing smart to say.

I can imagine this being a nice help for ops more often than not.

On https://procrastinate.readthedocs.io/en/latest/ the title of the lib is kinda butchered by the new line. We should either modify the theme one way or another, or find another nice looking theme.

Coverage is reported under 100%, but when launched locally, coverage is 100% Something is wrong in codecov.

When we launch a worker, it will receive jobs from the database. We have to link them to tasks, but 100% of the codebase may not have been imported, which means some tasks may not have been registered yet.

It seems there are 3 solutions:

• Provide the worker with a list of imports to do before starting, àla Celery
• Start the worker on a given module and only gather the tasks from this module. If we want tasks implemented elsewhere, we have to import them in this module (àla Dramatiq, if I'm not mistaken). This module may or may not contain the task manager itself, and this may or may not replace the "registration" par of the task decorator. I think this can only work if the task manager lives in cabbage instead of in the user's module, which I'm not a huge fan of, or with metaclasses which I'd rather avoid.
• Don't register tasks, but have a task name be a python importable path, load tasks lazily when we receive them. This means we can't do a list of tasks, which might appear limitating in the long run.

Ping @mgu @Evelf @sdispater

For now, the only retry strategy we've implemented is the constant backoff, but an exponential one makes sense too.

Current syntax is

procrastinate defer dotted.path.to.task '{"json": "arguments"}'

I'm thinking:

procrastinate defer dotted.path.to.task json=arguments

Task argument and defer parameters would be recognized through the face defer parameters are flags with --, - not being a valid python identifier (though it could be passed as **kwargs but...).

But then how to do typing ?
I'm not sure the solutions we used for the similar problem in https://github.com/peopledoc/vault-cli apply here. And I'd like to avoid yaml here, because it's nowhere else in the project.
Interestingly, ansible has the same problem with -e and it's a mess.

• I'd like explicit strings, like json="arguments" (so all values right of equal would be passed to json.loads() but double quotes are going to be eaten by bash.
• But we could document that it's necessary to wrap double quotes in simple quotes.
• We could use the task type annotations but it really feels dangerous.
  We could type variables (scrapy/scrapy#356 (comment)) (read the rest of the ticket)
• Or we could stay with json.
• Finally, we could have multiple ways, as long as it's not complexifying the code too much. A json for complex cases, simple keypairs for the rest (but like, I can live without lists, but bools and ints should be 1st class citizens)

Interesing reads:

• AWS: https://docs.aws.amazon.com/fr_fr/cli/latest/userguide/cli-usage-parameters.html#cli-usage-parameters-json
• Everything from https://www.google.com/search?q=cli+arguments+json

Any idea ?

• Link the connection object to the TaskManager,
• Make the TaskManager extensible (inheritance or, preferably, composition) so that someone could implement their own method of connecting to the DB (maybe from Django), with a default implementation that would take no argument and read from env vars, like currently
• Standardize whether functions receive the task manager, the connection or the cursor as parameter. Find the right one and use it consistently where applicable.
• Use consistent connection options and cursor classes.

This is the second step in asynchronous compatibility.

• #9 is for defering jobs asynchronously
• This ticket is for accepting a task to be a coroutine
• #106 is for executing asynchronous tasks in parallel

In #4, we implemented "simple" message logs. They're so simple that they actually lack precision .The details are in the structured part, but it's not printed by defaut. Someone analyzing logs in a structured way will have everything they need, but someone just using the default log formatter will be in the dark.

I think it would be nice if the default logger could log the extra informations too
OR
The default message could be made dynamic (though f-strings) and include the extra info.

curs -> cursor
conn -> connection
task_worker.py -> worker.py
worker() -> run_worker() (or something else ?)
task run -> job

I'm being told the project would be easier to use if we provided a simple cron-like process. It might be interesting to look.

@EliotBerriot, you mentionned you were maybe interested to try a contribution ?

I plan to place the project under the Contributor Covenant. Anyone against it ?