From 56c20d0018456df24ed8bfb6052d065521e945a5 Mon Sep 17 00:00:00 2001 From: Dimitri Yatsenko Date: Sat, 17 Oct 2026 18:38:11 -0500 Subject: [PATCH] fix(jobs): stale_timeout=0 means no grace period refresh() removes a job whose key has left key_source once the job is older than stale_timeout, a grace period for a key_source that is still being filled. stale_timeout=0 skipped that check instead, so there was no way to ask for the cleanup now: refresh(stale_timeout=0) left every such job in place, and so did populate() with config.jobs.stale_timeout = 0. 0 now means no grace period: on that refresh, every job whose key is no longer in key_source is removed, whatever its age, unless it is marked ignore. That includes reserved jobs, and success records of rows that remain after their key left key_source, which a nonzero stale_timeout keeps for that many seconds. stale_timeout=None still means the configured value (config.jobs.stale_timeout, 3600 seconds by default), so a refresh that does not pass it keeps the grace period. The setting's description, "Seconds before pending job is checked for staleness", now describes the grace period and what 0 does. Skipping the check is no longer expressible as 0; a long stale_timeout keeps such jobs for as long as it says. The tests delete upstream rows and check, on MySQL and PostgreSQL, that a default refresh keeps the pending jobs of the deleted keys, that refresh(stale_timeout=0) removes them, and that populate() with config.jobs.stale_timeout = 0 removes them instead of trying to compute them. Each fails without the change. They replace the test that held 0 to its old meaning. Closes #1577. --- src/datajoint/jobs.py | 19 ++++++++++------ src/datajoint/settings.py | 7 +++++- tests/integration/test_jobs_refresh.py | 31 ++++++++++++++++++++------ 3 files changed, 42 insertions(+), 15 deletions(-) diff --git a/src/datajoint/jobs.py b/src/datajoint/jobs.py index 0668db904..e5ed838cf 100644 --- a/src/datajoint/jobs.py +++ b/src/datajoint/jobs.py @@ -338,9 +338,10 @@ def refresh( Priority for new jobs (lower = more urgent). Default from ``config.jobs.default_priority``. stale_timeout : float, optional - Seconds after which jobs are checked for staleness. - Jobs older than this are removed if key not in key_source. - Default from ``config.jobs.stale_timeout``. Set to 0 to skip. + Grace period in seconds for jobs whose keys are no longer in + key_source: such jobs are removed once older than this. 0 means no + grace period, and removes them on this refresh. Default None uses + ``config.jobs.stale_timeout``. orphan_timeout : float, optional Seconds after which reserved jobs are considered orphaned. Reserved jobs older than this are deleted and re-added as pending. @@ -361,7 +362,8 @@ def refresh( 3. Remove vanished jobs, judged against the unrestricted key_source: a. success and error records whose keys are in neither key_source nor target, at any age - b. stale jobs (except ``ignore``): older than stale_timeout, keys not in key_source + b. stale jobs (except ``ignore``): keys not in key_source, older than stale_timeout + (at any age when stale_timeout is 0) 4. Remove orphaned jobs: reserved jobs older than orphan_timeout (if specified) """ @@ -431,11 +433,14 @@ def refresh( .delete_quick(get_count=True) ) - # 3b. Remove stale jobs (not ignore status) - use server CURRENT_TIMESTAMP for consistent timing + # 3b. Remove stale jobs (not ignore status) once older than stale_timeout, a grace period + # for a key_source that is still being filled. stale_timeout=0 means no grace period. + # Age is judged on the server's CURRENT_TIMESTAMP for consistent timing. + stale_jobs = vanished & "status != 'ignore'" if stale_timeout > 0: stale_interval = self.adapter.interval_expr(stale_timeout, "second") - stale_jobs = vanished & f"created_time < CURRENT_TIMESTAMP - {stale_interval}" & "status != 'ignore'" - result["removed"] += stale_jobs.delete_quick(get_count=True) + stale_jobs = stale_jobs & f"created_time < CURRENT_TIMESTAMP - {stale_interval}" + result["removed"] += stale_jobs.delete_quick(get_count=True) # 4. Handle orphaned reserved jobs - use server CURRENT_TIMESTAMP for consistent timing if orphan_timeout is not None and orphan_timeout > 0: diff --git a/src/datajoint/settings.py b/src/datajoint/settings.py index 38339a1d9..231a38bdd 100644 --- a/src/datajoint/settings.py +++ b/src/datajoint/settings.py @@ -294,7 +294,12 @@ class JobsSettings(BaseSettings): auto_refresh: bool = Field(default=True, description="Auto-refresh jobs queue on populate") keep_completed: bool = Field(default=False, description="Keep success records in jobs table") - stale_timeout: int = Field(default=3600, ge=0, description="Seconds before pending job is checked for staleness") + stale_timeout: int = Field( + default=3600, + ge=0, + description="Grace period in seconds before refresh() removes a job whose key is no longer in key_source. " + "0 means no grace period: such jobs are removed on the next refresh.", + ) default_priority: int = Field(default=5, ge=0, le=255, description="Default priority for new jobs (lower = more urgent)") version_method: Literal["git", "none"] | None = Field( default=None, description="Method to obtain version: 'git' (commit hash), 'none' (empty), or None (disabled)" diff --git a/tests/integration/test_jobs_refresh.py b/tests/integration/test_jobs_refresh.py index 9fe554db1..9c8576996 100644 --- a/tests/integration/test_jobs_refresh.py +++ b/tests/integration/test_jobs_refresh.py @@ -6,6 +6,8 @@ restriction -- reserved by other workers, or recording errors -- are not stale. ``success`` and ``error`` records whose key has left both ``key_source`` and the target are removed on refresh, in one statement rather than one per key. +``stale_timeout=0`` means no grace period for other jobs whose key has left +``key_source``: they are removed on that refresh, not kept indefinitely. """ import time @@ -166,19 +168,34 @@ def make(self, key): assert _statuses(Gated.jobs) == {0: "success", 1: "success"} -def test_stale_timeout_zero_still_skips_stale_check(pipeline): - """``stale_timeout=0`` keeps skipping the age-based check for pending jobs.""" +def test_stale_timeout_zero_means_no_grace_period(pipeline): + """``stale_timeout=0`` removes vanished pending jobs at once; the default keeps them for its grace period.""" + Raw, Proc = pipeline + Raw.insert([dict(raw_id=i, v=float(i)) for i in range(3)]) + Proc.jobs.refresh() + (Raw & "raw_id in (1, 2)").delete(prompt=False) + + result = Proc.jobs.refresh() # stale_timeout=None: the configured grace period applies + assert _statuses(Proc.jobs) == {0: "pending", 1: "pending", 2: "pending"} + assert result["removed"] == 0 + + result = Proc.jobs.refresh(stale_timeout=0) + assert _statuses(Proc.jobs) == {0: "pending"} + assert result["removed"] == 2 + + +def test_configured_stale_timeout_zero_means_no_grace_period(pipeline): + """``config.jobs.stale_timeout = 0`` makes the refresh in ``populate()`` remove vanished jobs at once.""" Raw, Proc = pipeline Raw.insert([dict(raw_id=i, v=float(i)) for i in range(2)]) Proc.jobs.refresh() + assert Proc.jobs.reserve({"raw_id": 0}) # keeps populate() from computing raw_id=0 (Raw & {"raw_id": 1}).delete(prompt=False) - _backdate(Proc.jobs) - Proc.jobs.refresh(stale_timeout=0) - assert _statuses(Proc.jobs) == {0: "pending", 1: "pending"} + with dj.config.override(jobs={"stale_timeout": 0}): + Proc.populate(reserve_jobs=True, suppress_errors=True) - Proc.jobs.refresh() - assert _statuses(Proc.jobs) == {0: "pending"} + assert _statuses(Proc.jobs) == {0: "reserved"} def test_refresh_cleanup_is_set_based(pipeline, monkeypatch):