Add result to jobs

README.md

@@ -10,7 +10,12 @@
 [![Contributor Covenant](https://img.shields.io/badge/Contributor%20Covenant-v1.4%20adopted-ff69b4.svg)](https://github.com/procrastinate-org/procrastinate/blob/main/CODE_OF_CONDUCT.md)
 [![Discord](https://img.shields.io/discord/1197292025725329549?logo=discord&logoColor=white&label=Discord&color=%237289da)](https://discord.gg/JWZeNq6P6Z)
 
-**Procrastinate is looking for** [additional maintainers!](https://github.com/procrastinate-org/procrastinate/discussions/748)
+### Fork changes
+
+1. Added `result` column to `jobs` table so failed job can store it's failure reason
+
+**Procrastinate is looking for
+** [additional maintainers!](https://github.com/procrastinate-org/procrastinate/discussions/748)
 
 Procrastinate is an open-source Python 3.10+ distributed task processing
 library, leveraging PostgreSQL 13+ to store task definitions, manage locks and
@@ -31,12 +36,14 @@ import procrastinate
 # Make an app in your code
 app = procrastinate.App(connector=procrastinate.SyncPsycopgConnector())
 
+
 # Then define tasks
 @app.task(queue="sums")
 def sum(a, b):
     with open("myfile", "w") as f:
         f.write(str(a + b))
 
+
 with app.open():
     # Launch a job
     sum.defer(a=3, b=5)
@@ -72,11 +79,13 @@ import procrastinate
 # Make an app in your code
 app = procrastinate.App(connector=procrastinate.PsycopgConnector())
 
+
 # Define tasks using coroutine functions
 @app.task(queue="sums")
 async def sum(a, b):
     await asyncio.sleep(a + b)
 
+
 async with app.open_async():
     # Launch a job
     await sum.defer_async(a=3, b=5)
@@ -106,7 +115,11 @@ If you encounter a bug, or want to get in touch, you're always welcome to open a
 [ticket].
 
 [docs]: https://procrastinate.readthedocs.io/
+
 [procrastinate]: https://en.wikipedia.org/wiki/Procrastination
+
 [django]: https://procrastinate.readthedocs.io/en/stable/howto/django/configuration.html
+
 [quickstart]: https://procrastinate.readthedocs.io/en/stable/quickstart.html
+
 [ticket]: https://github.com/procrastinate-org/procrastinate/issues/new

procrastinate/contrib/django/migrations/0042_pre_result_column.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.08.00_01_pre_result_to_job_procedure.sql"
+        ),
+    ]
+    name = "0042_pre_result_column"
+    dependencies = [
+        ("procrastinate", "0041_post_retry_failed_job"),
+    ]

procrastinate/manager.py

@@ -291,6 +291,7 @@ async def finish_job(
         job: jobs_module.Job,
         status: jobs_module.Status,
         delete_job: bool,
+        exc_info: bool | BaseException = False,
     ) -> None:
         """
         Set a job to its final state (``succeeded``, ``failed`` or ``aborted``).
@@ -303,20 +304,22 @@ async def finish_job(
         """
         assert job.id  # TODO remove this
         await self.finish_job_by_id_async(
-            job_id=job.id, status=status, delete_job=delete_job
+            job_id=job.id, status=status, delete_job=delete_job, exc_info=exc_info
         )
 
     async def finish_job_by_id_async(
         self,
         job_id: int,
         status: jobs_module.Status,
         delete_job: bool,
+        exc_info: bool | BaseException = False,
     ) -> None:
         await self.connector.execute_query_async(
             query=sql.queries["finish_job"],
             job_id=job_id,
             status=status.value,
             delete_job=delete_job,
+            exc_info=str(exc_info) if exc_info else None,
         )
 
     def cancel_job_by_id(

procrastinate/sql/migrations/03.08.00_01_pre_result_to_job_procedure.sql

@@ -0,0 +1,35 @@
+ALTER TABLE procrastinate_jobs ADD COLUMN result TEXT;
+
+drop function if exists "public"."procrastinate_finish_job_v1"(job_id bigint, end_status procrastinate_job_status, delete_job boolean);
+
+
+CREATE FUNCTION procrastinate_finish_job_v1(job_id bigint, end_status procrastinate_job_status, delete_job boolean, job_result text default null)
+    RETURNS void
+    LANGUAGE plpgsql
+AS $$
+DECLARE
+    _job_id bigint;
+BEGIN
+    IF end_status NOT IN ('succeeded', 'failed', 'aborted') THEN
+        RAISE 'End status should be either "succeeded", "failed" or "aborted" (job id: %)', job_id;
+    END IF;
+    IF delete_job THEN
+        DELETE FROM procrastinate_jobs
+        WHERE id = job_id AND status IN ('todo', 'doing')
+        RETURNING id INTO _job_id;
+    ELSE
+        UPDATE procrastinate_jobs
+        SET status = end_status,
+            abort_requested = false,
+            result = job_result,
+            attempts = CASE status
+                WHEN 'doing' THEN attempts + 1 ELSE attempts
+            END
+        WHERE id = job_id AND status IN ('todo', 'doing')
+        RETURNING id INTO _job_id;
+    END IF;
+    IF _job_id IS NULL THEN
+        RAISE 'Job was not found or not in "doing" or "todo" status (job id: %)', job_id;
+    END IF;
+END;
+$$

procrastinate/sql/schema.sql

@@ -59,6 +59,7 @@ CREATE TABLE procrastinate_jobs (
     attempts integer DEFAULT 0 NOT NULL,
     abort_requested boolean DEFAULT false NOT NULL,
     worker_id bigint REFERENCES procrastinate_workers(id) ON DELETE SET NULL,
+    result text,
     CONSTRAINT check_not_todo_abort_requested CHECK (NOT (status = 'todo' AND abort_requested = true))
 );
 
@@ -245,7 +246,7 @@ BEGIN
 END;
 $$;
 
-CREATE FUNCTION procrastinate_finish_job_v1(job_id bigint, end_status procrastinate_job_status, delete_job boolean)
+CREATE FUNCTION procrastinate_finish_job_v1(job_id bigint, end_status procrastinate_job_status, delete_job boolean, job_result text default null)
     RETURNS void
     LANGUAGE plpgsql
 AS $$
@@ -263,6 +264,7 @@ BEGIN
         UPDATE procrastinate_jobs
         SET status = end_status,
             abort_requested = false,
+            result = job_result,
             attempts = CASE status
                 WHEN 'doing' THEN attempts + 1 ELSE attempts
             END

procrastinate/testing.py

@@ -290,7 +290,13 @@ async def fetch_job_one(
         self.events[job["id"]].append({"type": "started", "at": utils.utcnow()})
         return job
 
-    async def finish_job_run(self, job_id: int, status: str, delete_job: bool) -> None:
+    async def finish_job_run(
+        self,
+        job_id: int,
+        status: str,
+        delete_job: bool,
+        exc_info: bool | BaseException = False,
+    ) -> None:
         if delete_job:
             self.jobs.pop(job_id)
             return

procrastinate/worker.py

@@ -140,6 +140,7 @@ async def _persist_job_status(
         retry_decision: retry.RetryDecision | None,
         context: job_context.JobContext,
         job_result: job_context.JobResult | None,
+        exc_info: bool | BaseException = False,
     ):
         if retry_decision:
             await self.app.job_manager.retry_job(
@@ -156,7 +157,7 @@ async def _persist_job_status(
                 jobs.DeleteJobCondition.SUCCESSFUL: status == jobs.Status.SUCCEEDED,
             }[self.delete_jobs]
             await self.app.job_manager.finish_job(
-                job=job, status=status, delete_job=delete_job
+                job=job, status=status, delete_job=delete_job, exc_info=exc_info
             )
 
         assert job.id
@@ -314,6 +315,7 @@ async def ensure_async() -> Callable[..., Awaitable[Any]]:
                     retry_decision=retry_decision,
                     context=context,
                     job_result=job_result,
+                    exc_info=exc_info,
                 )
             )
             try:

tests/unit/test_manager.py

@@ -291,7 +291,7 @@ async def test_finish_job(job_manager, job_factory, connector):
     )
     assert connector.queries[-1] == (
         "finish_job",
-        {"job_id": 1, "status": "succeeded", "delete_job": False},
+        {"job_id": 1, "status": "succeeded", "delete_job": False, "exc_info": None},
     )
 
 
@@ -302,7 +302,7 @@ async def test_finish_job_with_deletion(job_manager, job_factory, connector):
     await job_manager.finish_job(job=job, status=jobs.Status.SUCCEEDED, delete_job=True)
     assert connector.queries[-1] == (
         "finish_job",
-        {"job_id": 1, "status": "succeeded", "delete_job": True},
+        {"job_id": 1, "status": "succeeded", "delete_job": True, "exc_info": None},
     )
     assert 1 not in connector.jobs