From 15c61b2ae629d00ded4c9978185343d7c59d626a Mon Sep 17 00:00:00 2001 From: Dimitri Yatsenko Date: Sat, 17 Oct 2026 17:01:34 -0500 Subject: [PATCH] fix(jobs): record completed_time at millisecond precision complete() and error() took completed_time from SELECT CURRENT_TIMESTAMP, which on MySQL has no fractional seconds, while reserve() writes reserved_time as CURRENT_TIMESTAMP(3) into the same datetime(3) columns. A job that finished within the second it was reserved recorded a completed_time earlier than its reserved_time. Error records carry no duration, so that difference is the only record of how long a failed job ran, and it came out negative or up to a second short. Both now read SELECT CURRENT_TIMESTAMP(3), as reserve() does. populate() calls them after the make() transaction has been committed or cancelled, and the PostgreSQL connection runs in autocommit mode, so there the value is the time of the statement, not the start of an enclosing transaction. The tests populate three keys through a make() that sleeps 0.15 s, once succeeding with keep_completed=True and once raising, and check that every job's completed_time is at least 0.1 s after its reserved_time, on MySQL and PostgreSQL. Both fail on MySQL without the change. Closes #1600. (cherry picked from commit d47ca3b3ac4de50e54726848e9b1e133f85050a8) --- src/datajoint/jobs.py | 8 ++-- tests/integration/test_jobs.py | 79 ++++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 4 deletions(-) diff --git a/src/datajoint/jobs.py b/src/datajoint/jobs.py index 920d4eab0..90665666e 100644 --- a/src/datajoint/jobs.py +++ b/src/datajoint/jobs.py @@ -495,8 +495,8 @@ def complete(self, key: dict, duration: float | None = None) -> None: - If False: deletes the job entry """ if self.connection._config.jobs.keep_completed: - # Use server time for completed_time - server_now = self.connection.query("SELECT CURRENT_TIMESTAMP").fetchone()[0] + # Server time at the column's datetime(3) precision, as reserve() writes reserved_time + server_now = self.connection.query("SELECT CURRENT_TIMESTAMP(3)").fetchone()[0] pk = self._get_pk(key) update_row = { **pk, @@ -525,8 +525,8 @@ def error(self, key: dict, error_message: str, error_stack: str | None = None) - if len(error_message) > ERROR_MESSAGE_LENGTH: error_message = error_message[: ERROR_MESSAGE_LENGTH - len(TRUNCATION_APPENDIX)] + TRUNCATION_APPENDIX - # Use server time for completed_time - server_now = self.connection.query("SELECT CURRENT_TIMESTAMP").fetchone()[0] + # Server time at the column's datetime(3) precision, as reserve() writes reserved_time + server_now = self.connection.query("SELECT CURRENT_TIMESTAMP(3)").fetchone()[0] pk = self._get_pk(key) update_row = { diff --git a/tests/integration/test_jobs.py b/tests/integration/test_jobs.py index 60701bec8..1dbfb4588 100644 --- a/tests/integration/test_jobs.py +++ b/tests/integration/test_jobs.py @@ -1,7 +1,11 @@ """Tests for per-table Job management (AutoPopulate 2.0).""" +import datetime import random import string +import time + +import pytest import datajoint as dj from datajoint.jobs import ERROR_MESSAGE_LENGTH, TRUNCATION_APPENDIX @@ -230,3 +234,78 @@ def test_jobs_table_uuid_fk_derived_pk(schema_uuid): pk = BasicComputed.jobs.heading.primary_key assert "item" in pk assert BasicComputed.jobs.heading["item"].original_type == "uuid" + + +# Each make() below sleeps this long between reserve() and complete()/error(). +MAKE_SECONDS = 0.15 + + +@pytest.fixture(scope="function") +def schema_by_backend(connection_by_backend, db_creds_by_backend): + """Create a fresh schema per test, parameterized across backends.""" + backend = db_creds_by_backend["backend"] + test_id = str(int(time.time() * 1000))[-8:] + schema = dj.Schema(f"djtest_jobs_{backend}_{test_id}"[:64], connection=connection_by_backend) + yield schema + schema.drop(prompt=False) + + +def test_completed_time_precision(schema_by_backend): + """completed_time keeps datetime(3) precision on success, so it never precedes reserved_time (#1600).""" + + @schema_by_backend + class Item(dj.Manual): + definition = """ + item : int32 + """ + + @schema_by_backend + class Result(dj.Computed): + definition = """ + -> Item + --- + value : int32 + """ + + def make(self, key): + time.sleep(MAKE_SECONDS) + self.insert1(dict(key, value=key["item"])) + + Item.insert({"item": i} for i in range(3)) + with dj.config.override(jobs__keep_completed=True): + Result.populate(reserve_jobs=True) + jobs = Result.jobs.completed.to_dicts() + assert len(jobs) == 3 + for job in jobs: + # Truncated to the second, completed_time falls before reserved_time or short of make()'s sleep + assert job["completed_time"] - job["reserved_time"] >= datetime.timedelta(seconds=0.1), job + + +def test_error_time_precision(schema_by_backend): + """completed_time keeps datetime(3) precision on error, so it never precedes reserved_time (#1600).""" + + @schema_by_backend + class Item(dj.Manual): + definition = """ + item : int32 + """ + + @schema_by_backend + class Result(dj.Computed): + definition = """ + -> Item + --- + value : int32 + """ + + def make(self, key): + time.sleep(MAKE_SECONDS) + raise RuntimeError("not ready") + + Item.insert({"item": i} for i in range(3)) + Result.populate(reserve_jobs=True, suppress_errors=True) + jobs = Result.jobs.errors.to_dicts() + assert len(jobs) == 3 + for job in jobs: + # Error records carry no duration, so this difference is the only record of how long make() ran + assert job["completed_time"] - job["reserved_time"] >= datetime.timedelta(seconds=0.1), job