fix(jobs): record _job_start_time on the server's clock

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:
         """

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.

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.

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 ""),
         )

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

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",