Fix broken job lease extension

The `extend_job_leases` function was checking for `RunState.started` in a
`while` loop, however the coroutine is first executed by the (already running)
job executors when the scheduler is still `RunState.starting`. This caused the
loop condition to immediately evaluate to `False` and the function to return.
This commit fixes that by checking for both `RunState.starting` and
`RunState.started` in `extend_job_leases`, similar to the `cleanup` task.

With the `MemoryDataStore`, if a job ran longer than the lease *and* past the
next scheduled execution, then the exact same job would be executed again (along
with a job for the next schedule). The first execution cleans up when it finally
finishes, but then the second execution fails in cleanup.

The issue primarily affects the `MemoryDataStore` because its `acquire_jobs`
method seems to acquire even expired jobs (which the scheduler re-executes) more
aggressively than the other data stores, which have different logic for
discarding jobs / handling task job slots.

This commit partially fixes agronholm#803, agronholm#834, agronholm#952, agronholm#967, and maybe others. I suspect
there are more bugs lurking that I haven't pinned down yet, likely related to
the acquiring, releasing, and perhaps cleanup logic.

Author: JacobHayes

docs/versionhistory.rst

@@ -64,6 +64,8 @@ APScheduler, see the :doc:`migration section <migration>`.
   schema if it's missing (PR by @zhu0629)
 - Fixed an issue with ``CronTrigger`` infinitely looping to get next date when DST ends
   (`#980 <https://github.com/agronholm/apscheduler/issues/980>`_; PR by @hlobit)
+- Fix broken job lease extension (`#1030
+  <https://github.com/agronholm/apscheduler/issues/1030>`_; PR by @JacobHayes)
 
 **4.0.0a5**
 

src/apscheduler/_schedulers/async_.py

@@ -1106,7 +1106,7 @@ async def check_queue_capacity(event: Event) -> None:
                 wakeup_event.set()
 
         async def extend_job_leases() -> None:
-            while self._state is RunState.started:
+            while self._state in (RunState.starting, RunState.started):
                 await sleep(self.lease_duration.total_seconds() / 2)
                 job_ids = {job.id for job in self._running_jobs}
                 await self.data_store.extend_acquired_job_leases(