Unstarted jobs should always be cleaned up (#4604)

supervisor/jobs/__init__.py

@@ -86,7 +86,7 @@ class SupervisorJob:
         }
 
     @contextmanager
-    def start(self, *, on_done: Callable[["SupervisorJob"], None] | None = None):
+    def start(self):
         """Start the job in the current task.
 
         This can only be called if the parent ID matches the job running in the current task.
@@ -107,8 +107,6 @@ class SupervisorJob:
             self.done = True
             if token:
                 _CURRENT_JOB.reset(token)
-            if on_done:
-                on_done(self)
 
 
 class JobManager(FileConfiguration, CoreSysAttributes):

supervisor/jobs/decorator.py

@@ -201,6 +201,7 @@ class Job(CoreSysAttributes):
                 internal=self._internal,
             )
 
+            try:
                 # Handle condition
                 if self.conditions:
                     try:
@@ -211,14 +212,19 @@ class Job(CoreSysAttributes):
                         return self._handle_job_condition_exception(err)
 
                 # Handle exection limits
-            if self.limit in (JobExecutionLimit.SINGLE_WAIT, JobExecutionLimit.ONCE):
+                if self.limit in (
+                    JobExecutionLimit.SINGLE_WAIT,
+                    JobExecutionLimit.ONCE,
+                ):
                     await self._acquire_exection_limit()
                 elif self.limit in (
                     JobExecutionLimit.GROUP_ONCE,
                     JobExecutionLimit.GROUP_WAIT,
                 ):
                     try:
-                    await obj.acquire(job, self.limit == JobExecutionLimit.GROUP_WAIT)
+                        await obj.acquire(
+                            job, self.limit == JobExecutionLimit.GROUP_WAIT
+                        )
                     except JobGroupExecutionLimitExceeded as err:
                         if self.on_condition:
                             raise self.on_condition(str(err)) from err
@@ -244,26 +250,35 @@ class Job(CoreSysAttributes):
                     JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT,
                 ):
                     # Only reprocess array when necessary (at limit)
-                if len(self.rate_limited_calls(group_name)) >= self.throttle_max_calls:
+                    if (
+                        len(self.rate_limited_calls(group_name))
+                        >= self.throttle_max_calls
+                    ):
                         self.set_rate_limited_calls(
                             [
                                 call
                                 for call in self.rate_limited_calls(group_name)
-                            if call > datetime.now() - self.throttle_period(group_name)
+                                if call
+                                > datetime.now() - self.throttle_period(group_name)
                             ],
                             group_name,
                         )
 
-                if len(self.rate_limited_calls(group_name)) >= self.throttle_max_calls:
+                    if (
+                        len(self.rate_limited_calls(group_name))
+                        >= self.throttle_max_calls
+                    ):
                         on_condition = (
-                        JobException if self.on_condition is None else self.on_condition
+                            JobException
+                            if self.on_condition is None
+                            else self.on_condition
                         )
                         raise on_condition(
                             f"Rate limit exceeded, more then {self.throttle_max_calls} calls in {self.throttle_period(group_name)}",
                         )
 
                 # Execute Job
-            with job.start(on_done=self.sys_jobs.remove_job if self.cleanup else None):
+                with job.start():
                     try:
                         self.set_last_call(datetime.now(), group_name)
                         if self.rate_limited_calls(group_name) is not None:
@@ -291,6 +306,11 @@ class Job(CoreSysAttributes):
                         ):
                             obj.release()
 
+            # Jobs that weren't started are always cleaned up. Also clean up done jobs if required
+            finally:
+                if job.done is None or self.cleanup:
+                    self.sys_jobs.remove_job(job)
+
         return wrapper
 
     @staticmethod

tests/jobs/test_job_decorator.py

@@ -1044,3 +1044,61 @@ async def test_job_starting_separate_task(coresys: CoreSys):
     await test.job_await()
     await test.job_release()
     await task
+
+
+async def test_job_always_removed_on_check_failure(coresys: CoreSys):
+    """Test that the job instance is always removed if the condition or limit check fails."""
+
+    class TestClass:
+        """Test class."""
+
+        event = asyncio.Event()
+        limit_job: Job | None = None
+
+        def __init__(self, coresys: CoreSys) -> None:
+            """Initialize object."""
+            self.coresys = coresys
+
+        @Job(
+            name="test_job_always_removed_on_check_failure_condition",
+            conditions=[JobCondition.HAOS],
+            on_condition=JobException,
+            cleanup=False,
+        )
+        async def condition_check(self):
+            """Job that will fail a condition check."""
+            raise AssertionError("should not run")
+
+        @Job(
+            name="test_job_always_removed_on_check_failure_limit",
+            limit=JobExecutionLimit.ONCE,
+            cleanup=False,
+        )
+        async def limit_check(self):
+            """Job that can fail a limit check."""
+            self.limit_job = self.coresys.jobs.current
+            await self.event.wait()
+
+        def release_limit_check(self):
+            """Release the limit check job."""
+            self.event.set()
+
+    test = TestClass(coresys)
+
+    with pytest.raises(JobException):
+        await test.condition_check()
+    assert coresys.jobs.jobs == []
+
+    task = coresys.create_task(test.limit_check())
+    await asyncio.sleep(0)
+    assert (job := test.limit_job)
+
+    with pytest.raises(JobException):
+        await test.limit_check()
+    assert test.limit_job == job
+    assert coresys.jobs.jobs == [job]
+
+    test.release_limit_check()
+    await task
+    assert job.done
+    assert coresys.jobs.jobs == [job]