From 30a58cd3381a77e753d9514f22c8a2f5f0fbb22c Mon Sep 17 00:00:00 2001 From: Dimitri Yatsenko Date: Sat, 17 Oct 2026 18:32:30 -0500 Subject: [PATCH] fix(jobs): record _job_start_time on the server's clock populate() wrote _job_start_time as the populating host's local time, with no zone: time.time() at the start of the job, converted by datetime.fromtimestamp(). Two hosts in different time zones recorded different values for the same instant, and nothing in the row said which zone applied. The value could not be placed beside the job table's reserved_time and completed_time, which come from the server. The start time is now computed in the UPDATE that already writes the hidden columns, as the server's current time less _job_duration, so no query is added. The expression is the start of the statement, not of the transaction: NOW(3) on MySQL, and statement_timestamp() on PostgreSQL, where CURRENT_TIMESTAMP is the start of the transaction. It comes from a new adapter method, statement_timestamp_expr(). _job_duration is still measured on the populating host, now with time.monotonic(), which neither the host's zone nor an adjustment to its clock affects. The UPDATE runs just after the commit that the duration includes, so the derived start is late by the time that one statement takes to reach the server. Rows written before this change keep the populating host's local time. They cannot be corrected, because the zone was never recorded. The test populates two keys as if from hosts in UTC-12 and UTC+14 and checks, on MySQL and PostgreSQL, that each _job_start_time falls between two readings of the server's clock taken around the call. Without the change it fails on both, 12 hours off. Closes #1584. --- src/datajoint/adapters/base.py | 22 ++++++ src/datajoint/adapters/mysql.py | 20 +++++ src/datajoint/adapters/postgres.py | 21 +++++ src/datajoint/autopopulate.py | 27 ++++--- tests/integration/test_hidden_job_metadata.py | 78 +++++++++++++++++++ tests/unit/test_adapters.py | 11 +++ 6 files changed, 170 insertions(+), 9 deletions(-) diff --git a/src/datajoint/adapters/base.py b/src/datajoint/adapters/base.py index 5b4a7ab72..1aa12f52f 100644 --- a/src/datajoint/adapters/base.py +++ b/src/datajoint/adapters/base.py @@ -1093,6 +1093,28 @@ def current_timestamp_expr(self, precision: int | None = None) -> str: """ ... + @abstractmethod + def statement_timestamp_expr(self, precision: int | None = None) -> str: + """ + Expression for the time at which the current statement began. + + Unlike ``CURRENT_TIMESTAMP`` on PostgreSQL, which is the start of the + enclosing transaction, this is the start of the statement itself, so it + advances between statements within one transaction. + + Parameters + ---------- + precision : int, optional + Fractional seconds precision (0-6). + + Returns + ------- + str + SQL expression for the statement's start time (e.g., 'NOW(3)' for + MySQL, 'statement_timestamp()' for PostgreSQL). + """ + ... + @abstractmethod def interval_expr(self, value: int, unit: str) -> str: """ diff --git a/src/datajoint/adapters/mysql.py b/src/datajoint/adapters/mysql.py index 2ef9baf84..5f361204a 100644 --- a/src/datajoint/adapters/mysql.py +++ b/src/datajoint/adapters/mysql.py @@ -822,6 +822,26 @@ def current_timestamp_expr(self, precision: int | None = None) -> str: return f"CURRENT_TIMESTAMP({precision})" return "CURRENT_TIMESTAMP" + def statement_timestamp_expr(self, precision: int | None = None) -> str: + """ + Statement start time for MySQL. + + MySQL's ``NOW()`` is the time at which the statement began executing. + + Parameters + ---------- + precision : int, optional + Fractional seconds precision (0-6). + + Returns + ------- + str + NOW() or NOW(n). + """ + if precision is not None: + return f"NOW({precision})" + return "NOW()" + def interval_expr(self, value: int, unit: str) -> str: """ INTERVAL expression for MySQL. diff --git a/src/datajoint/adapters/postgres.py b/src/datajoint/adapters/postgres.py index 3a55d8eb8..c23d40a98 100644 --- a/src/datajoint/adapters/postgres.py +++ b/src/datajoint/adapters/postgres.py @@ -1051,6 +1051,27 @@ def current_timestamp_expr(self, precision: int | None = None) -> str: return f"CURRENT_TIMESTAMP({precision})" return "CURRENT_TIMESTAMP" + def statement_timestamp_expr(self, precision: int | None = None) -> str: + """ + Statement start time for PostgreSQL. + + PostgreSQL's ``CURRENT_TIMESTAMP`` is the start of the transaction; + ``statement_timestamp()`` is the start of the current statement. + + Parameters + ---------- + precision : int, optional + Fractional seconds precision (0-6). + + Returns + ------- + str + statement_timestamp(), cast to timestamptz(n) when a precision is given. + """ + if precision is not None: + return f"CAST(statement_timestamp() AS timestamptz({precision}))" + return "statement_timestamp()" + def interval_expr(self, value: int, unit: str) -> str: """ INTERVAL expression for PostgreSQL. diff --git a/src/datajoint/autopopulate.py b/src/datajoint/autopopulate.py index 61b36c5c9..1738b52e5 100644 --- a/src/datajoint/autopopulate.py +++ b/src/datajoint/autopopulate.py @@ -662,7 +662,9 @@ def _populate1( if jobs is not None and not jobs.reserve(key): return False - start_time = time.time() + # Elapsed time is measured on this host's monotonic clock. The start time + # itself is read from the server's clock when the job metadata is written. + started = time.monotonic() # if make is a generator, transaction can be delayed until the final stage is_generator = inspect.isgeneratorfunction(make) @@ -730,7 +732,7 @@ def _populate1( return key, error if return_exception_objects else error_message else: self.connection.commit_transaction() - duration = time.time() - start_time + duration = time.monotonic() - started logger.jobs(f"Success making {key} -> {self.full_table_name}") # Update hidden job metadata if table has the columns @@ -739,7 +741,6 @@ def _populate1( self._update_job_metadata( key, - start_time=datetime.datetime.fromtimestamp(start_time), duration=duration, version=_get_job_version(self.connection._config), ) @@ -852,25 +853,33 @@ def _has_job_metadata_attrs(self): all_attrs = self.heading._attributes return all_attrs is not None and "_job_start_time" in all_attrs - def _update_job_metadata(self, key, start_time, duration, version): + def _update_job_metadata(self, key, duration, version): """ Update hidden job metadata for the given key. + ``_job_start_time`` is read from the server's clock: the time at which + this statement runs, less ``duration``. Every host that populates the + schema therefore records it on one clock, whatever time zone the host + runs in. ``_job_duration`` is measured on the populating host. + Parameters ---------- key : dict Primary key identifying the row(s) to update. - start_time : datetime - When computation started. duration : float - Computation duration in seconds. + Computation duration in seconds, measured on the populating host. version : str Code version (truncated to 64 chars). """ from .condition import make_condition + adapter = self.connection.adapter pk_condition = make_condition(self, key, set()) + start_time = ( + f"{adapter.statement_timestamp_expr(3)} - {adapter.interval_expr(round(duration * 1_000_000), 'microsecond')}" + ) self.connection.query( - f"UPDATE {self.full_table_name} SET _job_start_time=%s, _job_duration=%s, _job_version=%s WHERE {pk_condition}", - args=(start_time, duration, version[:64] if version else ""), + f"UPDATE {self.full_table_name} SET _job_start_time={start_time}, _job_duration=%s, _job_version=%s " + f"WHERE {pk_condition}", + args=(duration, version[:64] if version else ""), ) diff --git a/tests/integration/test_hidden_job_metadata.py b/tests/integration/test_hidden_job_metadata.py index bcf56dde7..9171b8255 100644 --- a/tests/integration/test_hidden_job_metadata.py +++ b/tests/integration/test_hidden_job_metadata.py @@ -1,5 +1,6 @@ """Tests for hidden job metadata in computed tables.""" +import datetime import time import pytest @@ -271,3 +272,80 @@ def make(self, key): finally: schema.drop() dj.config.jobs.add_job_metadata = original_setting + + +@pytest.fixture +def schema_by_backend(connection_by_backend, db_creds_by_backend): + """A fresh schema with job metadata enabled, on each backend.""" + backend = db_creds_by_backend["backend"] + test_id = str(int(time.time() * 1000))[-8:] + schema_name = f"djtest_job_metadata_{backend}_{test_id}"[:64] + quoted = connection_by_backend.adapter.quote_identifier(schema_name) + original_setting = dj.config.jobs.add_job_metadata + dj.config.jobs.add_job_metadata = True + connection_by_backend.query(f"DROP DATABASE IF EXISTS {quoted}") + schema = dj.Schema(schema_name, connection=connection_by_backend) + yield schema + dj.config.jobs.add_job_metadata = original_setting + if connection_by_backend.is_connected: + connection_by_backend.query(f"DROP DATABASE IF EXISTS {quoted}") + + +@pytest.fixture +def host_time_zone(monkeypatch): + """Set the time zone of the populating process; the original is restored afterwards.""" + + def set_zone(zone): + monkeypatch.setenv("TZ", zone) + time.tzset() + + yield set_zone + monkeypatch.undo() + time.tzset() + + +@pytest.mark.skipif(not hasattr(time, "tzset"), reason="changing the process time zone needs time.tzset") +def test_job_start_time_uses_server_clock(schema_by_backend, host_time_zone): + """ + _job_start_time is read from the server's clock, whatever time zone the populating host runs in. + + Two populate() calls run as if from hosts on opposite sides of the date line, 26 hours apart. + Each start time must fall between two readings of the server's clock taken around the call (#1584). + """ + + @schema_by_backend + class Source(dj.Lookup): + definition = """ + source_id : int16 + """ + contents = [{"source_id": 1}, {"source_id": 2}] + + @schema_by_backend + class Timed(dj.Computed): + definition = """ + -> Source + --- + result : float32 + """ + + def make(self, key): + time.sleep(0.2) + self.insert1(dict(key, result=1.0)) + + connection = schema_by_backend.connection + tolerance = datetime.timedelta(seconds=1) + + def server_now(): + return connection.query("SELECT LOCALTIMESTAMP(3)").fetchone()[0] + + for source_id, zone in ((1, "Etc/GMT+12"), (2, "Etc/GMT-14")): + host_time_zone(zone) + before = server_now() + Timed.populate({"source_id": source_id}) + after = server_now() + start_time, duration = connection.query( + f"SELECT _job_start_time, _job_duration FROM {Timed.full_table_name} WHERE source_id = {source_id}" + ).fetchone() + assert before - tolerance <= start_time <= after, f"host zone {zone}: {start_time} is not within [{before}, {after}]" + assert duration >= 0.2 + assert start_time + datetime.timedelta(seconds=duration) <= after + tolerance diff --git a/tests/unit/test_adapters.py b/tests/unit/test_adapters.py index bd0afa9e0..0e6e65441 100644 --- a/tests/unit/test_adapters.py +++ b/tests/unit/test_adapters.py @@ -166,6 +166,11 @@ def test_current_timestamp_expr(self, adapter): assert adapter.current_timestamp_expr() == "CURRENT_TIMESTAMP" assert adapter.current_timestamp_expr(3) == "CURRENT_TIMESTAMP(3)" + def test_statement_timestamp_expr(self, adapter): + """Test statement start time expression.""" + assert adapter.statement_timestamp_expr() == "NOW()" + assert adapter.statement_timestamp_expr(3) == "NOW(3)" + def test_interval_expr(self, adapter): """Test INTERVAL expression.""" assert adapter.interval_expr(5, "second") == "INTERVAL 5 SECOND" @@ -311,6 +316,11 @@ def test_current_timestamp_expr(self, adapter): assert adapter.current_timestamp_expr() == "CURRENT_TIMESTAMP" assert adapter.current_timestamp_expr(3) == "CURRENT_TIMESTAMP(3)" + def test_statement_timestamp_expr(self, adapter): + """Test statement start time expression: not CURRENT_TIMESTAMP, which is the transaction's start.""" + assert adapter.statement_timestamp_expr() == "statement_timestamp()" + assert adapter.statement_timestamp_expr(3) == "CAST(statement_timestamp() AS timestamptz(3))" + def test_interval_expr(self, adapter): """Test INTERVAL expression with PostgreSQL syntax.""" assert adapter.interval_expr(5, "second") == "INTERVAL '5 seconds'" @@ -403,6 +413,7 @@ def test_adapter_implements_interface(self, backend): "commit_sql", "rollback_sql", "current_timestamp_expr", + "statement_timestamp_expr", "interval_expr", "json_path_expr", "format_column_definition",