Repository: learningequality/kolibri
Issue: Implement integrity check of job_storage database at startup and clear if broken

### Observed behavior
If something like an unclean shutdown has caused the job_storage sqlite file to be in a bad state, then any subsequent queries to the Tasks database may result in errors.

### Expected behavior
We should attempt to read any extant jobs in the tasks database at startup, and if it results in an error and is using SQLite, clear the file.

### User-facing consequences
Unclean shutdowns that leave the job storage database in an unclean state could prevent the usual 'clear at startup' that we do, and then leave 500s coming from queries to the Task API.

Changed files:
kolibri/core/tasks/main.py, kolibri/core/sqlite/utils.py

diff --git a/kolibri/core/sqlite/utils.py b/kolibri/core/sqlite/utils.py
index 2d85c101e5..e9dfc09581 100644
--- a/kolibri/core/sqlite/utils.py
+++ b/kolibri/core/sqlite/utils.py
@@ -8,6 +8,7 @@ from shutil import copyfile
 from django.conf import settings
 from django.core.management import call_command
 from django.db.utils import DatabaseError
+from sqlalchemy import exc
 
 logger = logging.getLogger(__name__)
 
@@ -41,6 +42,33 @@ def regenerate_database(connection):
         call_command("migrate", interactive=False, verbosity=False)
 
 
+def check_sqlite_integrity(connection):
+    """
+    Runs integrity check on sqlite db.
+    Raises ORM specific DatabaseError when db api communication is halted.
+    Raises sqlite3.DatabaseError when integrity check fails.
+    """
+    if settings.DATABASES["default"]["ENGINE"] != "django.db.backends.sqlite3":
+        return
+
+    if hasattr(connection, "name"):
+        # SQLAlchemy ORM db api connection
+        conn = connection.connect()
+    else:
+        # Django ORM db api connection cursor
+        conn = connection.cursor()
+
+    try:
+        result = conn.execute("PRAGMA integrity_check;").fetchall()
+    except (DatabaseError, exc.DatabaseError):
+        raise
+    finally:
+        conn.close()
+
+    if not (len(result) == 1 and result[0] == ("ok",)):
+        raise sqlite3.DatabaseError
+
+
 def repair_sqlite_db(connection):
     from kolibri.core.deviceadmin.utils import KWARGS_IO_WRITE
     from kolibri.core.deviceadmin.utils import default_backup_folder
diff --git a/kolibri/core/tasks/main.py b/kolibri/core/tasks/main.py
index 026d009cad..3e53739793 100644
--- a/kolibri/core/tasks/main.py
+++ b/kolibri/core/tasks/main.py
@@ -1,5 +1,6 @@
 import logging
 import os
+import sqlite3
 
 from django.utils.functional import SimpleLazyObject
 from sqlalchemy import create_engine
@@ -7,6 +8,7 @@ from sqlalchemy import event
 from sqlalchemy import exc
 from sqlalchemy.pool import NullPool
 
+from kolibri.core.sqlite.utils import check_sqlite_integrity
 from kolibri.core.sqlite.utils import repair_sqlite_db
 from kolibri.core.tasks.queue import Queue
 from kolibri.core.tasks.scheduler import Scheduler
@@ -57,9 +59,10 @@ elif conf.OPTIONS["Database"]["DATABASE_ENGINE"] == "postgres":
 def __initialize_connection():
     connection = __create_engine()
 
-    # Add multiprocessing safeguards as recommended by
-    # https://docs.sqlalchemy.org/en/13/core/pooling.html#using-connection-pools-with-multiprocessing
-
+    # Add multiprocessing safeguards as recommended by:
+    # https://docs.sqlalchemy.org/en/13/core/pooling.html#pooling-multiprocessing
+    # Don't make a connection before we've added the multiprocessing guards
+    # as otherwise we will have a connection that doesn't have the 'pid' attribute set.
     @event.listens_for(connection, "connect")
     def connect(dbapi_connection, connection_record):
         connection_record.info["pid"] = os.getpid()
@@ -74,12 +77,10 @@ def __initialize_connection():
                 % (connection_record.info["pid"], pid)
             )
 
-    # Don't make a connection before we've added the multiprocessing guards
-    # as otherwise we will have a connection that doesn't have the 'pid' attribute set.
-    # check if the database is corrupted:
+    # Check if the database is corrupted
     try:
-        connection.execute("SELECT name FROM sqlite_master WHERE type='table';")
-    except (exc.DatabaseError, TypeError):
+        check_sqlite_integrity(connection)
+    except (exc.DatabaseError, sqlite3.DatabaseError):
         repair_sqlite_db(connection)
 
     return connection