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.

Author: dimitri-yatsenko

src/datajoint/jobs.py

@@ -338,11 +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 the
-            age-based check; the cleanup of success and error records whose
-            keys have left both key_source and the target runs regardless.
+            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.
@@ -363,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)
         """
@@ -433,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:

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

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