Add retry for check_vectors if data is not yet available in the caches

Fixes #614 (closed) and #623 (closed).

check_vectors raises a ValueError when it tries to look for state vector data that's not yet available. This MR changes this so that it waits 5 seconds then tries again (up to four times), which should typically be enough for data to arrive at CIT.

Cases like this for this superevent should be handled by the retry. In this case, the early warning event with merger time 53.2 was uploaded at 53.0. Then, the check_vectors task is called at 54.306, asking for data to be checked up to 54.2.

Since the requested data end time (54.2) is before the time of the function call (54.306), the check built into check_vectors to skip for data after now passes. However, the 54.2 data hasn't arrived onto the cluster yet at 54.306, so the task runs into this ValueError.

This retry will now wait 5 seconds and try again, up to 4 times for a total of 20 seconds. This should be a long enough latency that any data will have arrived at CIT. If, after these 20 seconds the data is still not there, a ValueError will be raised and block the pipeline. This is the desired behaviour, since we want to make sure that we can check the data quality before an alert goes out.

gwcelery/tasks/detchar.py

@@ -374,8 +374,8 @@ def check_vector(cache, channel, start, end, bits, logic_type='all'):
             None for key in bits if key is not None}
 
 
-@app.task(shared=False)
-def check_vectors(event, graceid, start, end):
+@app.task(shared=False, bind=True, default_retry_delay=5)
+def check_vectors(self, event, graceid, start, end):
     """Perform data quality checks for an event and labels/logs results to
     GraceDB.
 
@@ -452,8 +452,14 @@ def check_vectors(event, graceid, start, end):
         analysis_channels = {k: v for k, v in analysis_channels
                              if k[3:] != 'DMT-DQ_VECTOR'}.items()
     for channel, bits in analysis_channels:
-        states.update(check_vector(caches[channel.split(':')[0]], channel,
-                                   start, end, bit_defs[bits]))
+        try:
+            states.update(check_vector(
+                caches[channel.split(':')[0]], channel,
+                start, end, bit_defs[bits]))
+        except ValueError as exc:
+            # check_vector likely failed to find the requested data
+            # in the cache because it has yet to arrive
+            raise self.retry(exc=exc, max_retries=4)
     # Pick out DQ and injection states, then filter for active detectors
     dq_states = {key: value for key, value in states.items()
                  if key.split('_')[-1] != 'INJ'}