Bugfix: don't drop scheduled runs when the clock outruns them

docs/config.rst

@@ -26,6 +26,7 @@ config file.
       since: null               # now, or YYYY-MM-DDTHH:MM:SSZ
       max_age: null             # seconds
       horizon: 86400            # seconds
+      clock_margin: 1800        # duration
 
     waiting:
       max_time: null            # duration
@@ -94,6 +95,18 @@ than this.  This prevents very old runs from running spuriously.
 `schedule.horizon` specifies how far forward in time, in seconds, to schedule new
 runs.
 
+`schedule.clock_margin` specifies how far before the stored schedule time to
+start scheduling runs on startup.  Apsis records the time through which it has
+started scheduled runs, but it does so before those runs are stored, so if Apsis
+stops abruptly the recorded time can be later than the last run that was
+actually stored.  Without a margin, runs in that gap are never started.
+
+Apsis does not create a run for a schedule time if a run for the same job, args,
+and schedule time already exists, so the margin does not cause runs to be
+started twice.  Note, however, that a job added while Apsis was down has no
+existing runs, so its runs are created back to the margin; keep the margin small
+enough that this is acceptable.  Set to 0 to disable.
+
 
 Waiting
 -------

python/apsis/apsis.py

@@ -16,6 +16,7 @@
 from .jobs import Jobs, load_jobs_dir, diff_jobs_dirs
 from .lib.api import run_to_summary_jso
 from .lib.asyn import TaskGroup, Publisher, KeyPublisher
+from .lib.parse import parse_duration
 from .lib.py import more_gc_stats
 from .lib.sys import to_signal
 from .output import OutputStore
@@ -110,12 +111,32 @@ def __init__(self, cfg, jobs, db):
         # Continue scheduling from the last time we handled scheduled jobs.
         # FIXME: Rename: schedule horizon?
         stop_time = db.clock_db.get_time()
+
+        # The clock is advanced before the runs it covers are persisted, so
+        # after a crash it can be ahead of the runs that actually reached the
+        # database, and those runs would never be scheduled again.  Start a
+        # little earlier than the clock to cover that skew, and rely on the
+        # schedule-time check in the scheduler to avoid creating a run that
+        # already exists.
+        #
+        # The skew is bounded by how long the scheduled loop went between
+        # advancing the clock and persisting the runs it had popped, so a modest
+        # margin is enough.  Keep it modest: the margin is also how far back a
+        # job added while Apsis was down is scheduled, and those runs have no
+        # existing counterpart to suppress them.
+        margin = parse_duration(cfg.get("schedule", {}).get("clock_margin", 1800))
+        assert margin >= 0
+        if margin > 0:
+            log.info(f"scheduling from {stop_time - margin}: clock {stop_time} - {margin} s")
+            stop_time -= margin
+
         self.scheduler = Scheduler(
             cfg,
             self.jobs,
             # All runs scheduled by the scheduler are expected.
             partial(self.schedule, expected=True),
             stop_time,
+            get_schedule_times=self.__get_schedule_times,
         )
 
         self.scheduled = ScheduledRuns(db.clock_db, self.scheduler.get_scheduler_time, self._wait)
@@ -213,6 +234,20 @@ def start_loops(self):
         # We're running now.
         self.running_flag.set()
 
+    def __get_schedule_times(self):
+        """
+        Returns nominal schedule times of runs that already exist.
+
+        Used by the scheduler to avoid creating a second run for a schedule
+        time that has already been handled.  See `RunStore.get_schedule_times`.
+        """
+        # Bound the query by the earliest schedule time the scheduler can
+        # revisit.  A run that has started has a timestamp at or after its
+        # schedule time, so any started run with a schedule time in the
+        # scheduling window also has a timestamp in it, and is found.  Runs that
+        # haven't started are in memory, which this bound doesn't apply to.
+        return self.run_store.get_schedule_times(min_timestamp=self.scheduler.get_scheduler_time())
+
     def _wait(self, run):
         """
         Starts waiting for `run`.

python/apsis/runs.py

@@ -1,4 +1,4 @@
-from collections import namedtuple
+from collections import Counter, namedtuple
 import itertools
 import jinja2
 import logging
@@ -634,6 +634,51 @@ def count_runs(
             min_timestamp=min_ts,
         )
 
+    def get_schedule_times(self, *, min_timestamp):
+        """
+        Counts runs by nominal schedule time.
+
+        Used to avoid recreating a run that already exists for a given nominal
+        schedule time.  See `Apsis.schedule`.
+
+        Covers in-memory runs (expected and active) as well as persisted runs,
+        since a scheduled expected run is not persisted until it starts.
+
+        Counts rather than merely noting presence, since a job may have several
+        schedules that produce the same schedule time and args, and each is a
+        run in its own right.
+
+        :param min_timestamp:
+          Ignore runs older than this.  Bounds the work; runs older than this
+          are not candidates for rescheduling anyway.
+        :return:
+          Mapping from `(job_id, canonical args JSON)` to a mapping from nominal
+          schedule time to the number of runs with that schedule time, in any
+          state.
+        """
+        from .sqlite import canonical_args_json
+
+        res = {}
+
+        def add(job_id, args_json, time):
+            if time is None:
+                # Nothing to key on.
+                return
+            times = res.setdefault((job_id, args_json), Counter())
+            times[time] += 1
+
+        # In-memory runs: expected (scheduled, not yet persisted) and active.
+        for run in itertools.chain(self.__expected_runs.values(), self.__active_runs.values()):
+            add(run.inst.job_id, canonical_args_json(run.inst.args), run.times.get("schedule"))
+
+        # Persisted runs, including finished ones.
+        for job_id, args_json, time in self.__run_db.query_schedule_times(
+            min_timestamp=min_timestamp
+        ):
+            add(job_id, args_json, time)
+
+        return res
+
     def get_stats(self):
         # Note: deliberately no DB count here.  Counting runs in the lookback window
         # requires a query over the runs table, which has no index on timestamp and

python/apsis/scheduler.py

@@ -4,6 +4,7 @@
 from ora import Time, now
 
 from .runs import Instance
+from .sqlite import canonical_args_json
 from apsis.lib.parse import parse_duration
 
 log = logging.getLogger(__name__)
@@ -39,12 +40,17 @@ class Scheduler:
     Does not own any runs.
     """
 
-    def __init__(self, cfg, jobs, schedule, stop):
+    def __init__(self, cfg, jobs, schedule, stop, *, get_schedule_times=None):
         """
         :param jobs:
           Jobs object.
         :param schedule:
           Function of `time, run` that schedules a run.
+        :param get_schedule_times:
+          Function of no args returning a mapping from `(job_id, args JSON)` to
+          the set of nominal schedule times for which a run already exists.
+          Used to avoid recreating runs after a restart.  If none, no such
+          check is made.
         """
         cfg = cfg.get("schedule", {})
 
@@ -68,6 +74,7 @@ def __init__(self, cfg, jobs, schedule, stop):
         self.__schedule = schedule
         self.__horizon = horizon
         self.__max_age = max_age
+        self.__get_schedule_times = get_schedule_times
 
     def set_jobs(self, jobs):
         """
@@ -90,10 +97,35 @@ async def schedule(self, stop):
             return
 
         log.debug(f"scheduling runs until {stop}")
+
+        # Counts of runs that already exist, by job, args, and nominal schedule
+        # time, so that we don't create a second run for a schedule time we
+        # already handled.  Only needed when scheduling into the past, i.e.
+        # catching up after downtime; in steady state the window is in the
+        # future, where no run exists yet.
+        if self.__get_schedule_times is not None and self.__stop < now():
+            existing = self.__get_schedule_times()
+            log.info(f"scheduling from {self.__stop}: {len(existing)} existing run keys")
+        else:
+            existing = None
+
         n = 0
+        skipped = 0
         for job in self.__jobs.get_jobs():
             items = get_insts_to_schedule(job, self.__stop, stop)
             for sched_time, stop_time, inst in items:
+                if existing is not None:
+                    # Account for each existing run against one schedule that
+                    # would produce it.  A job may have several schedules that
+                    # produce the same schedule time and args, in which case
+                    # each is a run of its own, so match them up one for one
+                    # rather than skipping every schedule that collides.
+                    times = existing.get((inst.job_id, canonical_args_json(inst.args)))
+                    if times is not None and times.get(sched_time, 0) > 0:
+                        times[sched_time] -= 1
+                        skipped += 1
+                        continue
+
                 await self.__schedule(sched_time, inst, stop_time=stop_time)
                 # using modulo instead of batching a generator because reducing allocations actually matters here for
                 # because of GC pressure
@@ -105,6 +137,9 @@ async def schedule(self, stop):
                     # after measuring startup times.
                     await asyncio.sleep(0)
 
+        if skipped > 0:
+            log.info(f"skipped {skipped} runs that already exist")
+
         self.__stop = stop
 
     async def loop(self):

python/apsis/sqlite.py

@@ -575,6 +575,29 @@ def fmt_params(**kwargs):
         )
         return runs
 
+    def query_schedule_times(self, *, min_timestamp=None):
+        """
+        Returns `(job_id, args, schedule_time)` for runs, without building
+        `Run` objects.
+
+        Only the nominal schedule time is extracted from the `times` column,
+        which makes this much cheaper than `query()` for the whole table: no
+        program, conds, or actions are deserialized.
+
+        :return:
+          Iterator of `(job_id, args_json, schedule_time)`, where `args_json`
+          is the canonical args JSON and `schedule_time` may be `None` if the
+          run has no schedule time.
+        """
+        query = sa.select([TBL_RUNS.c.job_id, TBL_RUNS.c.args, TBL_RUNS.c.times])
+        if min_timestamp is not None:
+            query = query.where(TBL_RUNS.c.timestamp >= dump_time(min_timestamp))
+
+        with self.__engine.connect() as conn:
+            for job_id, args_json, times_json in conn.execute(query):
+                time = ujson.loads(times_json).get("schedule")
+                yield job_id, args_json, None if time is None else ora.Time(time)
+
     def count_runs(
         self,
         *,