Skip to content

Commit 30a58cd

Browse files
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.
1 parent a00b89f commit 30a58cd

6 files changed

Lines changed: 170 additions & 9 deletions

File tree

‎src/datajoint/adapters/base.py‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1093,6 +1093,28 @@ def current_timestamp_expr(self, precision: int | None = None) -> str:
10931093
"""
10941094
...
10951095

1096+
@abstractmethod
1097+
def statement_timestamp_expr(self, precision: int | None = None) -> str:
1098+
"""
1099+
Expression for the time at which the current statement began.
1100+
1101+
Unlike ``CURRENT_TIMESTAMP`` on PostgreSQL, which is the start of the
1102+
enclosing transaction, this is the start of the statement itself, so it
1103+
advances between statements within one transaction.
1104+
1105+
Parameters
1106+
----------
1107+
precision : int, optional
1108+
Fractional seconds precision (0-6).
1109+
1110+
Returns
1111+
-------
1112+
str
1113+
SQL expression for the statement's start time (e.g., 'NOW(3)' for
1114+
MySQL, 'statement_timestamp()' for PostgreSQL).
1115+
"""
1116+
...
1117+
10961118
@abstractmethod
10971119
def interval_expr(self, value: int, unit: str) -> str:
10981120
"""

‎src/datajoint/adapters/mysql.py‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -822,6 +822,26 @@ def current_timestamp_expr(self, precision: int | None = None) -> str:
822822
return f"CURRENT_TIMESTAMP({precision})"
823823
return "CURRENT_TIMESTAMP"
824824

825+
def statement_timestamp_expr(self, precision: int | None = None) -> str:
826+
"""
827+
Statement start time for MySQL.
828+
829+
MySQL's ``NOW()`` is the time at which the statement began executing.
830+
831+
Parameters
832+
----------
833+
precision : int, optional
834+
Fractional seconds precision (0-6).
835+
836+
Returns
837+
-------
838+
str
839+
NOW() or NOW(n).
840+
"""
841+
if precision is not None:
842+
return f"NOW({precision})"
843+
return "NOW()"
844+
825845
def interval_expr(self, value: int, unit: str) -> str:
826846
"""
827847
INTERVAL expression for MySQL.

‎src/datajoint/adapters/postgres.py‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1051,6 +1051,27 @@ def current_timestamp_expr(self, precision: int | None = None) -> str:
10511051
return f"CURRENT_TIMESTAMP({precision})"
10521052
return "CURRENT_TIMESTAMP"
10531053

1054+
def statement_timestamp_expr(self, precision: int | None = None) -> str:
1055+
"""
1056+
Statement start time for PostgreSQL.
1057+
1058+
PostgreSQL's ``CURRENT_TIMESTAMP`` is the start of the transaction;
1059+
``statement_timestamp()`` is the start of the current statement.
1060+
1061+
Parameters
1062+
----------
1063+
precision : int, optional
1064+
Fractional seconds precision (0-6).
1065+
1066+
Returns
1067+
-------
1068+
str
1069+
statement_timestamp(), cast to timestamptz(n) when a precision is given.
1070+
"""
1071+
if precision is not None:
1072+
return f"CAST(statement_timestamp() AS timestamptz({precision}))"
1073+
return "statement_timestamp()"
1074+
10541075
def interval_expr(self, value: int, unit: str) -> str:
10551076
"""
10561077
INTERVAL expression for PostgreSQL.

‎src/datajoint/autopopulate.py‎

Lines changed: 18 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -662,7 +662,9 @@ def _populate1(
662662
if jobs is not None and not jobs.reserve(key):
663663
return False
664664

665-
start_time = time.time()
665+
# Elapsed time is measured on this host's monotonic clock. The start time
666+
# itself is read from the server's clock when the job metadata is written.
667+
started = time.monotonic()
666668

667669
# if make is a generator, transaction can be delayed until the final stage
668670
is_generator = inspect.isgeneratorfunction(make)
@@ -730,7 +732,7 @@ def _populate1(
730732
return key, error if return_exception_objects else error_message
731733
else:
732734
self.connection.commit_transaction()
733-
duration = time.time() - start_time
735+
duration = time.monotonic() - started
734736
logger.jobs(f"Success making {key} -> {self.full_table_name}")
735737

736738
# Update hidden job metadata if table has the columns
@@ -739,7 +741,6 @@ def _populate1(
739741

740742
self._update_job_metadata(
741743
key,
742-
start_time=datetime.datetime.fromtimestamp(start_time),
743744
duration=duration,
744745
version=_get_job_version(self.connection._config),
745746
)
@@ -852,25 +853,33 @@ def _has_job_metadata_attrs(self):
852853
all_attrs = self.heading._attributes
853854
return all_attrs is not None and "_job_start_time" in all_attrs
854855

855-
def _update_job_metadata(self, key, start_time, duration, version):
856+
def _update_job_metadata(self, key, duration, version):
856857
"""
857858
Update hidden job metadata for the given key.
858859
860+
``_job_start_time`` is read from the server's clock: the time at which
861+
this statement runs, less ``duration``. Every host that populates the
862+
schema therefore records it on one clock, whatever time zone the host
863+
runs in. ``_job_duration`` is measured on the populating host.
864+
859865
Parameters
860866
----------
861867
key : dict
862868
Primary key identifying the row(s) to update.
863-
start_time : datetime
864-
When computation started.
865869
duration : float
866-
Computation duration in seconds.
870+
Computation duration in seconds, measured on the populating host.
867871
version : str
868872
Code version (truncated to 64 chars).
869873
"""
870874
from .condition import make_condition
871875

876+
adapter = self.connection.adapter
872877
pk_condition = make_condition(self, key, set())
878+
start_time = (
879+
f"{adapter.statement_timestamp_expr(3)} - {adapter.interval_expr(round(duration * 1_000_000), 'microsecond')}"
880+
)
873881
self.connection.query(
874-
f"UPDATE {self.full_table_name} SET _job_start_time=%s, _job_duration=%s, _job_version=%s WHERE {pk_condition}",
875-
args=(start_time, duration, version[:64] if version else ""),
882+
f"UPDATE {self.full_table_name} SET _job_start_time={start_time}, _job_duration=%s, _job_version=%s "
883+
f"WHERE {pk_condition}",
884+
args=(duration, version[:64] if version else ""),
876885
)

‎tests/integration/test_hidden_job_metadata.py‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
"""Tests for hidden job metadata in computed tables."""
22

3+
import datetime
34
import time
45

56
import pytest
@@ -271,3 +272,80 @@ def make(self, key):
271272
finally:
272273
schema.drop()
273274
dj.config.jobs.add_job_metadata = original_setting
275+
276+
277+
@pytest.fixture
278+
def schema_by_backend(connection_by_backend, db_creds_by_backend):
279+
"""A fresh schema with job metadata enabled, on each backend."""
280+
backend = db_creds_by_backend["backend"]
281+
test_id = str(int(time.time() * 1000))[-8:]
282+
schema_name = f"djtest_job_metadata_{backend}_{test_id}"[:64]
283+
quoted = connection_by_backend.adapter.quote_identifier(schema_name)
284+
original_setting = dj.config.jobs.add_job_metadata
285+
dj.config.jobs.add_job_metadata = True
286+
connection_by_backend.query(f"DROP DATABASE IF EXISTS {quoted}")
287+
schema = dj.Schema(schema_name, connection=connection_by_backend)
288+
yield schema
289+
dj.config.jobs.add_job_metadata = original_setting
290+
if connection_by_backend.is_connected:
291+
connection_by_backend.query(f"DROP DATABASE IF EXISTS {quoted}")
292+
293+
294+
@pytest.fixture
295+
def host_time_zone(monkeypatch):
296+
"""Set the time zone of the populating process; the original is restored afterwards."""
297+
298+
def set_zone(zone):
299+
monkeypatch.setenv("TZ", zone)
300+
time.tzset()
301+
302+
yield set_zone
303+
monkeypatch.undo()
304+
time.tzset()
305+
306+
307+
@pytest.mark.skipif(not hasattr(time, "tzset"), reason="changing the process time zone needs time.tzset")
308+
def test_job_start_time_uses_server_clock(schema_by_backend, host_time_zone):
309+
"""
310+
_job_start_time is read from the server's clock, whatever time zone the populating host runs in.
311+
312+
Two populate() calls run as if from hosts on opposite sides of the date line, 26 hours apart.
313+
Each start time must fall between two readings of the server's clock taken around the call (#1584).
314+
"""
315+
316+
@schema_by_backend
317+
class Source(dj.Lookup):
318+
definition = """
319+
source_id : int16
320+
"""
321+
contents = [{"source_id": 1}, {"source_id": 2}]
322+
323+
@schema_by_backend
324+
class Timed(dj.Computed):
325+
definition = """
326+
-> Source
327+
---
328+
result : float32
329+
"""
330+
331+
def make(self, key):
332+
time.sleep(0.2)
333+
self.insert1(dict(key, result=1.0))
334+
335+
connection = schema_by_backend.connection
336+
tolerance = datetime.timedelta(seconds=1)
337+
338+
def server_now():
339+
return connection.query("SELECT LOCALTIMESTAMP(3)").fetchone()[0]
340+
341+
for source_id, zone in ((1, "Etc/GMT+12"), (2, "Etc/GMT-14")):
342+
host_time_zone(zone)
343+
before = server_now()
344+
Timed.populate({"source_id": source_id})
345+
after = server_now()
346+
start_time, duration = connection.query(
347+
f"SELECT _job_start_time, _job_duration FROM {Timed.full_table_name} WHERE source_id = {source_id}"
348+
).fetchone()
349+
assert before - tolerance <= start_time <= after, f"host zone {zone}: {start_time} is not within [{before}, {after}]"
350+
assert duration >= 0.2
351+
assert start_time + datetime.timedelta(seconds=duration) <= after + tolerance

‎tests/unit/test_adapters.py‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,11 @@ def test_current_timestamp_expr(self, adapter):
166166
assert adapter.current_timestamp_expr() == "CURRENT_TIMESTAMP"
167167
assert adapter.current_timestamp_expr(3) == "CURRENT_TIMESTAMP(3)"
168168

169+
def test_statement_timestamp_expr(self, adapter):
170+
"""Test statement start time expression."""
171+
assert adapter.statement_timestamp_expr() == "NOW()"
172+
assert adapter.statement_timestamp_expr(3) == "NOW(3)"
173+
169174
def test_interval_expr(self, adapter):
170175
"""Test INTERVAL expression."""
171176
assert adapter.interval_expr(5, "second") == "INTERVAL 5 SECOND"
@@ -311,6 +316,11 @@ def test_current_timestamp_expr(self, adapter):
311316
assert adapter.current_timestamp_expr() == "CURRENT_TIMESTAMP"
312317
assert adapter.current_timestamp_expr(3) == "CURRENT_TIMESTAMP(3)"
313318

319+
def test_statement_timestamp_expr(self, adapter):
320+
"""Test statement start time expression: not CURRENT_TIMESTAMP, which is the transaction's start."""
321+
assert adapter.statement_timestamp_expr() == "statement_timestamp()"
322+
assert adapter.statement_timestamp_expr(3) == "CAST(statement_timestamp() AS timestamptz(3))"
323+
314324
def test_interval_expr(self, adapter):
315325
"""Test INTERVAL expression with PostgreSQL syntax."""
316326
assert adapter.interval_expr(5, "second") == "INTERVAL '5 seconds'"
@@ -403,6 +413,7 @@ def test_adapter_implements_interface(self, backend):
403413
"commit_sql",
404414
"rollback_sql",
405415
"current_timestamp_expr",
416+
"statement_timestamp_expr",
406417
"interval_expr",
407418
"json_path_expr",
408419
"format_column_definition",

0 commit comments

Comments
 (0)