Add timeout to process_visit.

This change protects against extremely long (possibly hours) delays
when performing network I/O.

This involves reinstating the WORKER_TIMEOUT environment variable,
which was inadvertantly removed in the migration from Gunicorn to Keda.
Unlike the original worker-based implementation, this one lets the
process try to clean up (through normal exception handling) instead of
a hard kill.

Author: kfindeisen

python/activator/activator.py

@@ -60,6 +60,8 @@
 instrument_name = os.environ["RUBIN_INSTRUMENT"]
 # The skymap to use in the central repo
 skymap = os.environ["SKYMAP"]
+# The maximum time to spend on a single visit.
+global_timeout = os.environ.get("WORKER_TIMEOUT", None)
 # URI to the main repository to contain processing results
 write_repo = os.environ["CENTRAL_REPO"]
 # URI to the main repository containing calibs and templates
@@ -535,7 +537,14 @@ async def _manage_process_visit(expected_visit: FannedOutVisit):
     This function encapsulates the use of asyncio by process_visit, while
     delegating all actual processing to `_process_visit_or_cancel`.
     """
-    await _process_visit_or_cancel(expected_visit)
+    try:
+        async with asyncio.timeout(global_timeout):  # No timeout if None
+            await _process_visit_or_cancel(expected_visit)
+    except TimeoutError:
+        _log.error(f"Processing of group {expected_visit.group} detector {expected_visit.detector} "
+                   f"failed to complete in {global_timeout} s. "
+                   "Outputs have *NOT* been written.")
+        raise
 
 
 async def _process_visit_or_cancel(expected_visit: FannedOutVisit):