Add persistent pause/resume queue

docs/howto/advanced.md

@@ -9,6 +9,7 @@ advanced/locks
 advanced/schedule
 advanced/priorities
 advanced/cancellation
+advanced/pause_queue
 advanced/queueing_locks
 advanced/cron
 advanced/retry

docs/howto/advanced/pause_queue.md

@@ -0,0 +1,91 @@
+# Pause a queue
+
+We can pause a queue so that workers stop fetching its jobs, then resume it
+later. This is useful to temporarily hold a queue's work, for example during a
+maintenance window or while an external dependency is unavailable, without
+stopping the worker process.
+
+Pausing only stops the *fetching* of new jobs: jobs already being processed run
+to completion, and pending jobs stay in the `todo` state until the queue is
+resumed. Workers keep fetching jobs from the other queues.
+
+The pause is stored in the database, so it is shared by every worker consuming
+the queue and it survives a worker restart.
+
+## Pause a queue
+
+```python
+# by using the sync method
+app.job_manager.pause_queue("some_queue")
+# or by using the async method
+await app.job_manager.pause_queue_async("some_queue")
+```
+
+## Resume a queue
+
+```python
+# by using the sync method
+app.job_manager.resume_queue("some_queue")
+# or by using the async method
+await app.job_manager.resume_queue_async("some_queue")
+```
+
+Resuming a queue that is not paused does nothing.
+
+## Pause keys: several independent holders
+
+Every pause is held under a *pause key* (`"default"` when not specified). A
+queue is paused as long as it holds at least one key, and each key must be
+resumed for the queue to start working again. This lets independent processes
+pause the same queue without stepping on each other: if a deploy script and a
+maintenance task both pause a queue and the deploy finishes first, its resume
+only releases its own key — the queue stays paused until the maintenance task
+resumes too.
+
+```python
+# deploy script
+await app.job_manager.pause_queue_async("some_queue", pause_key="deploy")
+...
+await app.job_manager.resume_queue_async("some_queue", pause_key="deploy")
+
+# maintenance task, meanwhile
+await app.job_manager.pause_queue_async("some_queue", pause_key="maintenance")
+...
+await app.job_manager.resume_queue_async("some_queue", pause_key="maintenance")
+```
+
+## Resume all keys at once
+
+A pause whose holder crashed before resuming (or was never going to resume)
+stays in place until it is resumed under the same key. As an escape hatch,
+`all_keys=True` removes every pause key from a queue, regardless of who holds
+them:
+
+```python
+app.job_manager.resume_queue("some_queue", all_keys=True)
+```
+
+## List paused queues
+
+```python
+# by using the sync method
+app.job_manager.list_paused_queues()
+# or by using the async method
+await app.job_manager.list_paused_queues_async()
+```
+
+This returns one `dict` per held pause key, with `queue_name`, `pause_key` and
+`paused_at` keys — a queue paused by several holders appears once per key:
+
+```python
+[
+    {"queue_name": "some_queue", "pause_key": "deploy", "paused_at": ...},
+    {"queue_name": "some_queue", "pause_key": "maintenance", "paused_at": ...},
+]
+```
+
+Both methods accept optional `queue` and `pause_key` filters:
+
+```python
+await app.job_manager.list_paused_queues_async(queue="some_queue")
+```

docs/howto/django/models.md

@@ -1,6 +1,6 @@
 # Interact with Procrastinate tables as Django models
 
-Procrastinate exposes 3 of its internal tables as Django models. You can use
+Procrastinate exposes some of its internal tables as Django models. You can use
 them to query the state of your jobs. They're also exposed in the Django admin.
 
 :::{note}
@@ -15,6 +15,7 @@ from procrastinate.contrib.django.models import (
     ProcrastinateEvent,
     ProcrastinatePeriodicDefer,
     ProcrastinateWorker,
+    ProcrastinatePausedQueue,
 )
 
 ProcrastinateJob.objects.filter(task_name="mytask").count()
@@ -30,7 +31,7 @@ or events through the ORM.
 
 ```{eval-rst}
 .. automodule:: procrastinate.contrib.django.models
-    :members: ProcrastinateJob, ProcrastinateEvent, ProcrastinatePeriodicDefer, ProcrastinateWorker
+    :members: ProcrastinateJob, ProcrastinateEvent, ProcrastinatePeriodicDefer, ProcrastinateWorker, ProcrastinatePausedQueue
 ```
 
 

procrastinate/contrib/django/migrations/0042_pre_add_queue_pause.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from django.db import migrations
+
+from .. import migrations_utils
+
+
+class Migration(migrations.Migration):
+    operations = [
+        migrations_utils.RunProcrastinateSQL(
+            name="03.10.00_01_pre_add_queue_pause.sql"
+        ),
+    ]
+    name = "0042_pre_add_queue_pause"
+    dependencies = [
+        ("procrastinate", "0041_post_retry_failed_job"),
+    ]

procrastinate/contrib/django/migrations/0043_post_add_queue_pause.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from django.db import migrations
+
+from .. import migrations_utils
+
+
+class Migration(migrations.Migration):
+    operations = [
+        migrations_utils.RunProcrastinateSQL(
+            name="03.10.00_50_post_add_queue_pause.sql"
+        ),
+    ]
+    name = "0043_post_add_queue_pause"
+    dependencies = [
+        ("procrastinate", "0042_pre_add_queue_pause"),
+    ]

procrastinate/contrib/django/migrations/0044_add_paused_queue_model.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from django.db import migrations, models
+
+import procrastinate.contrib.django.models
+
+
+class Migration(migrations.Migration):
+    operations = [
+        migrations.CreateModel(
+            name="ProcrastinatePausedQueue",
+            fields=[
+                ("id", models.BigAutoField(primary_key=True, serialize=False)),
+                ("queue_name", models.CharField(max_length=128)),
+                ("pause_key", models.CharField(max_length=128)),
+                ("paused_at", models.DateTimeField()),
+            ],
+            options={
+                "db_table": "procrastinate_paused_queues",
+                "managed": False,
+            },
+            bases=(
+                procrastinate.contrib.django.models.ProcrastinateReadOnlyModelMixin,
+                models.Model,
+            ),
+        ),
+    ]
+    name = "0044_add_paused_queue_model"
+    dependencies = [
+        ("procrastinate", "0043_post_add_queue_pause"),
+    ]

procrastinate/contrib/django/models.py

@@ -165,3 +165,23 @@ class Meta:  # type: ignore
         managed = False
         db_table = "procrastinate_periodic_defers"
         unique_together = [("task_name", "periodic_id", "defer_timestamp")]
+
+
+class ProcrastinatePausedQueue(ProcrastinateReadOnlyModelMixin, models.Model):
+    id = models.BigAutoField(primary_key=True)
+    queue_name = models.CharField(max_length=128)
+    pause_key = models.CharField(max_length=128)
+    paused_at = models.DateTimeField()
+
+    objects = ProcrastinateReadOnlyManager()
+
+    class Meta:  # type: ignore
+        managed = False
+        db_table = "procrastinate_paused_queues"
+        unique_together = [("queue_name", "pause_key")]
+
+    def __str__(self) -> str:
+        return (
+            f"Queue {self.queue_name} - "
+            f"Paused with key {self.pause_key} at {self.paused_at}"
+        )

procrastinate/jobs.py

@@ -32,7 +32,12 @@ class AbortJobRequested(TypedDict):
     job_id: int
 
 
-Notification: TypeAlias = JobInserted | AbortJobRequested
+class QueueResumed(TypedDict):
+    type: Literal["queue_resumed"]
+    queue_name: str
+
+
+Notification: TypeAlias = JobInserted | AbortJobRequested | QueueResumed
 
 
 def check_aware(