From e0f0fa338e45365058d73745f300a396bd5e3c2f Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Mon, 10 May 2021 20:45:42 +0200
Subject: [PATCH] Add periodic history audit prune task

that runs when not using celery.
---
 lib/galaxy/app.py           | 11 ++++++++
 lib/galaxy/util/task.py     | 51 +++++++++++++++++++++++++++++++++++++
 test/unit/util/test_task.py | 28 ++++++++++++++++++++
 3 files changed, 90 insertions(+)
 create mode 100644 lib/galaxy/util/task.py
 create mode 100644 test/unit/util/test_task.py

diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py
index fa6b7e6b62a..dffa996a080 100644
--- a/lib/galaxy/app.py
+++ b/lib/galaxy/app.py
@@ -61,6 +61,7 @@ from galaxy.util import (
     heartbeat,
     StructuredExecutionTimer,
 )
+from galaxy.util.task import IntervalTask
 from galaxy.visualization.data_providers.registry import DataProviderRegistry
 from galaxy.visualization.genomes import Genomes
 from galaxy.visualization.plugins.registry import VisualizationsRegistry
@@ -304,6 +305,16 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication):
             self.authnz_manager = managers.AuthnzManager(self,
                                                          self.config.oidc_config_file,
                                                          self.config.oidc_backends_config_file)
+
+        if not self.config.enable_celery_tasks and self.config.history_audit_table_prune_interval > 0:
+            self.prune_history_audit_task = IntervalTask(
+                func=lambda: galaxy.model.HistoryAudit.prune(self.model.session),
+                name="HistoryAuditTablePruneTask",
+                interval=self.config.history_audit_table_prune_interval,
+                immediate_start=False,
+                time_execution=True)
+            self.application_stack.register_postfork_function(self.prune_history_audit_task.start)
+            self.haltables.append(("HistoryAuditTablePruneTask", self.prune_history_audit_task.shutdown))
         # Start the job manager
         self.application_stack.register_postfork_function(self.job_manager.start)
         self.proxy_manager = ProxyManager(self.config)
diff --git a/lib/galaxy/util/task.py b/lib/galaxy/util/task.py
new file mode 100644
index 00000000000..9da0e347d08
--- /dev/null
+++ b/lib/galaxy/util/task.py
@@ -0,0 +1,51 @@
+import logging
+from threading import (
+    Event,
+    Thread,
+)
+
+from galaxy.util import ExecutionTimer
+
+log = logging.getLogger(__name__)
+
+
+class IntervalTask:
+
+    def __init__(self, func, name="Periodic task", interval=3600, immediate_start=False, time_execution=False):
+        """
+        Run an arbitrary function `func` every `interval` seconds.
+
+        Set `immediate_start` to True to run `func` when task is started.
+        """
+        self.func = func
+        self.name = name
+        self.interval = interval
+        self.time_execution = time_execution
+        self.immediate_start = immediate_start
+        self.event = Event()
+        self.thread = Thread(target=self.run, name=self.name, daemon=True)
+        self.running = False
+
+    def start(self):
+        self.running = True
+        self.thread.start()
+
+    def _exec(self):
+        if self.time_execution:
+            timer = ExecutionTimer()
+        self.func()
+        if self.time_execution:
+            log.debug(f"Executed periodic task {self.name} {timer}")
+
+    def run(self):
+        if self.immediate_start:
+            self._exec()
+        while not self.event.isSet():
+            self.event.wait(self.interval)
+            if self.running:
+                self._exec()
+
+    def shutdown(self):
+        self.running = False
+        self.event.set()
+        self.thread.join(5)
diff --git a/test/unit/util/test_task.py b/test/unit/util/test_task.py
new file mode 100644
index 00000000000..a1b7b317634
--- /dev/null
+++ b/test/unit/util/test_task.py
@@ -0,0 +1,28 @@
+import time
+
+from galaxy.util.task import IntervalTask
+
+
+def test_interval_task_immediate_start():
+    results = []
+    task = IntervalTask(lambda: results.append(1), name="test_task", interval=0.2, immediate_start=True)
+    task.start()
+    task.shutdown()
+    assert len(results) == 1
+
+
+def test_interval_task_delayed_start():
+    results = []
+    task = IntervalTask(lambda: results.append(1), name="test_task", interval=0.2, immediate_start=False)
+    task.start()
+    task.shutdown()
+    assert len(results) == 0
+
+
+def test_interval_task_delayed_start_run_once():
+    results = []
+    task = IntervalTask(lambda: results.append(1), name="test_task", interval=0.2, immediate_start=False)
+    task.start()
+    time.sleep(0.25)
+    task.shutdown()
+    assert len(results) == 1