Make history database error recovery robust and improve code.

Make more robust:
- Put a lock on HistoryDB() class initialization to prevent re-entrant database creation
- Improved recovery from corrupt databases
- Add some exception handling to potential crash cases

Make code nicer.
- Add method comments
- Replace global variables with class-attributes

sabnzbd/__init__.py

@@ -201,7 +201,7 @@ INIT_LOCK = Lock()
 def connect_db(thread_index=0):
     # Create a connection and store it in the current thread
     if not (hasattr(cherrypy.thread_data, 'history_db') and cherrypy.thread_data.history_db):
-        cherrypy.thread_data.history_db = sabnzbd.database.get_history_handle()
+        cherrypy.thread_data.history_db = sabnzbd.database.HistoryDB()
     return cherrypy.thread_data.history_db
 
 

sabnzbd/api.py

@@ -63,7 +63,7 @@ from sabnzbd.utils.servertests import test_nntp_server_dict
 from sabnzbd.bpsmeter import BPSMeter
 from sabnzbd.rating import Rating
 from sabnzbd.getipaddress import localipv4, publicipv4, ipv6
-from sabnzbd.database import build_history_info, unpack_history_info, get_history_handle
+from sabnzbd.database import build_history_info, unpack_history_info, HistoryDB
 import sabnzbd.notifier
 import sabnzbd.rss
 import sabnzbd.emailer
@@ -1838,7 +1838,7 @@ def build_history(start=None, limit=None, verbose=False, verbose_list=None, sear
         close_db = False
     except:
         # Required for repairs at startup because Cherrypy isn't active yet
-        history_db = get_history_handle()
+        history_db = HistoryDB()
         close_db = True
 
     # Fetch history items
@@ -2125,7 +2125,7 @@ def del_from_section(kwargs):
 def history_remove_failed():
     """ Remove all failed jobs from history, including files """
     logging.info('Scheduled removal of all failed jobs')
-    history_db = get_history_handle()
+    history_db = HistoryDB()
     del_job_files(history_db.get_failed_paths())
     history_db.remove_failed()
     history_db.close()
@@ -2135,7 +2135,7 @@ def history_remove_failed():
 def history_remove_completed():
     """ Remove all completed jobs from history """
     logging.info('Scheduled removal of all completed jobs')
-    history_db = get_history_handle()
+    history_db = HistoryDB()
     history_db.remove_completed()
     history_db.close()
     del history_db

sabnzbd/bpsmeter.py

@@ -157,7 +157,7 @@ class BPSMeter(object):
         """ Get the latest data from the database and assign to a fake server """
         logging.debug('Setting default BPS meter values')
         import sabnzbd.database
-        history_db = sabnzbd.database.get_history_handle()
+        history_db = sabnzbd.database.HistoryDB()
         grand, month, week = history_db.get_history_size()
         history_db.close()
         self.grand_total = {}

sabnzbd/database.py

@@ -33,6 +33,7 @@ import datetime
 import zlib
 import logging
 import sys
+import threading
 
 import sabnzbd
 import sabnzbd.cfg
@@ -40,17 +41,9 @@ from sabnzbd.constants import DB_HISTORY_NAME, STAGES
 from sabnzbd.encoding import unicoder
 from sabnzbd.bpsmeter import this_week, this_month
 from sabnzbd.misc import format_source_url
+from sabnzbd.decorators import synchronized
 
-_HISTORY_DB = None        # Will contain full path to history database
-_DONE_CLEANING = False    # Ensure we only do one Vacuum per session
-
-
-def get_history_handle():
-    """ Get an instance of the history db hanlder """
-    global _HISTORY_DB
-    if not _HISTORY_DB:
-        _HISTORY_DB = os.path.join(sabnzbd.cfg.admin_dir.get_path(), DB_HISTORY_NAME)
-    return HistoryDB(_HISTORY_DB)
+DB_LOCK = threading.RLock()
 
 
 def convert_search(search):
@@ -74,39 +67,52 @@ def convert_search(search):
     return search
 
 
-# TODO: Add support for execute return values
 class HistoryDB(object):
-
-    def __init__(self, db_path):
-        self.db_path = db_path
+    """ Class to access the History database
+        Each class-instance will create an access channel that
+        can be used in one thread.
+        Each thread needs its own class-instance!
+    """
+    # These class attributes will be accessed directly because
+    # they need to be shared by all instances
+    db_path = None        # Will contain full path to history database
+    done_cleaning = False # Ensure we only do one Vacuum per session
+
+    @synchronized(DB_LOCK)
+    def __init__(self):
+        """ Determine databse path and create connection """
         self.con = self.c = None
+        if not HistoryDB.db_path:
+            HistoryDB.db_path = os.path.join(sabnzbd.cfg.admin_dir.get_path(), DB_HISTORY_NAME)
         self.connect()
+        
 
     def connect(self):
-        global _DONE_CLEANING
-        if not os.path.exists(self.db_path):
-            create_table = True
-        else:
-            create_table = False
-        self.con = sqlite3.connect(self.db_path)
+        """ Create a connection to the database """
+        create_table = not os.path.exists(HistoryDB.db_path)
+        self.con = sqlite3.connect(HistoryDB.db_path)
         self.con.row_factory = dict_factory
         self.c = self.con.cursor()
         if create_table:
             self.create_history_db()
-        elif not _DONE_CLEANING:
+        elif not HistoryDB.done_cleaning:
             # Run VACUUM on sqlite
             # When an object (table, index, or trigger) is dropped from the database, it leaves behind empty space
             # http://www.sqlite.org/lang_vacuum.html
-            _DONE_CLEANING = True
+            HistoryDB.done_cleaning = True
             self.execute('VACUUM')
 
         self.execute('PRAGMA user_version;')
-        version = self.c.fetchone()['user_version']
+        try:
+            version = self.c.fetchone()['user_version']
+        except TypeError:
+            version = 0
         if version < 1:
             # Add any missing columns added since first DB version
-            self.execute('PRAGMA user_version = 1;')
-            self.execute('ALTER TABLE "history" ADD COLUMN series TEXT;')
-            self.execute('ALTER TABLE "history" ADD COLUMN md5sum TEXT;')
+            # Use "and" to stop when database has been reset due to corruption
+            self.execute('PRAGMA user_version = 1;') and \
+                self.execute('ALTER TABLE "history" ADD COLUMN series TEXT;') and \
+                self.execute('ALTER TABLE "history" ADD COLUMN md5sum TEXT;')
 
     def execute(self, command, args=(), save=False):
         ''' Wrapper for executing SQL commands '''
@@ -124,15 +130,18 @@ class HistoryDB(object):
                 logging.error(T('Cannot write to History database, check access rights!'))
                 # Report back success, because there's no recovery possible
                 return True
-            elif 'not a database' in error or 'malformed' in error:
+            elif 'not a database' in error or 'malformed' in error or 'duplicate column name' in error:
                 logging.error(T('Damaged History database, created empty replacement'))
                 logging.info("Traceback: ", exc_info=True)
                 self.close()
                 try:
-                    os.remove(self.db_path)
+                    os.remove(HistoryDB.db_path)
                 except:
                     pass
                 self.connect()
+                # Return False in case of "duplicate column" error
+                # because the column addition in connect() must be terminated
+                return 'duplicate column name' not in error
             else:
                 logging.error(T('SQL Command Failed, see log'))
                 logging.debug("SQL: %s", command)
@@ -141,9 +150,10 @@ class HistoryDB(object):
                     self.con.rollback()
                 except:
                     logging.debug("Rollback Failed:", exc_info=True)
-            return False
+            return True
 
     def create_history_db(self):
+        """ Create a new (empty) database file """
         self.execute("""
         CREATE TABLE "history" (
             "id" INTEGER PRIMARY KEY,
@@ -177,6 +187,7 @@ class HistoryDB(object):
         self.execute('PRAGMA user_version = 1;')
 
     def save(self):
+        """ Save database to disk """
         try:
             self.con.commit()
         except:
@@ -184,6 +195,7 @@ class HistoryDB(object):
             logging.info("Traceback: ", exc_info=True)
 
     def close(self):
+        """ Close database connection """
         try:
             self.c.close()
             self.con.close()
@@ -192,6 +204,7 @@ class HistoryDB(object):
             logging.info("Traceback: ", exc_info=True)
 
     def remove_completed(self, search=None):
+        """ Remove all completed jobs from the database, optional with `search` pattern """
         search = convert_search(search)
         return self.execute("""DELETE FROM history WHERE name LIKE ? AND status = 'Completed'""", (search,), save=True)
 
@@ -205,10 +218,12 @@ class HistoryDB(object):
             return []
 
     def remove_failed(self, search=None):
+        """ Remove all failed jobs from the database, optional with `search` pattern """
         search = convert_search(search)
         return self.execute("""DELETE FROM history WHERE name LIKE ? AND status = 'Failed'""", (search,), save=True)
 
     def remove_history(self, jobs=None):
+        """ Remove all jobs in the list `jobs`, empty list will remove all completed jobs """
         if jobs is None:
             self.remove_completed()
         else:
@@ -221,7 +236,7 @@ class HistoryDB(object):
         self.save()
 
     def add_history_db(self, nzo, storage, path, postproc_time, script_output, script_line):
-
+        """ Add a new job entry to the database """
         t = build_history_info(nzo, storage, path, postproc_time, script_output, script_line)
 
         if self.execute("""INSERT INTO history (completed, name, nzb_name, category, pp, script, report,
@@ -231,7 +246,7 @@ class HistoryDB(object):
             self.save()
 
     def fetch_history(self, start=None, limit=None, search=None, failed_only=0, categories=None):
-
+        """ Return records for specified jobs """
         search = convert_search(search)
 
         post = ''
@@ -336,6 +351,7 @@ class HistoryDB(object):
         return (total, month, week)
 
     def get_script_log(self, nzo_id):
+        """ Return decompressed log file """
         data = ''
         t = (nzo_id,)
         if self.execute('SELECT script_log FROM history WHERE nzo_id=?', t):
@@ -346,6 +362,7 @@ class HistoryDB(object):
         return data
 
     def get_name(self, nzo_id):
+        """ Return name of the job `nzo_id` """
         t = (nzo_id,)
         name = ''
         if self.execute('SELECT name FROM history WHERE nzo_id=?', t):
@@ -356,6 +373,7 @@ class HistoryDB(object):
         return name
 
     def get_path(self, nzo_id):
+        """ Return the `incomplete` path of the job `nzo_id` """
         t = (nzo_id,)
         path = ''
         if self.execute('SELECT path FROM history WHERE nzo_id=?', t):
@@ -366,6 +384,7 @@ class HistoryDB(object):
         return path
 
     def get_other(self, nzo_id):
+        """ Return additional data for job `nzo_id` """
         t = (nzo_id,)
         if self.execute('SELECT * FROM history WHERE nzo_id=?', t):
             try:
@@ -381,6 +400,7 @@ class HistoryDB(object):
 
 
 def dict_factory(cursor, row):
+    """ Return a dictionary for the current database position """
     d = {}
     for idx, col in enumerate(cursor.description):
         d[col[0]] = row[idx]

sabnzbd/nzbqueue.py

@@ -444,7 +444,7 @@ class NzbQueue(TryList):
 
             if add_to_history:
                 # Create the history DB instance
-                history_db = database.get_history_handle()
+                history_db = database.HistoryDB()
                 # Add the nzo to the database. Only the path, script and time taken is passed
                 # Other information is obtained from the nzo
                 history_db.add_history_db(nzo, '', '', 0, '', '')

sabnzbd/nzbstuff.py

@@ -54,7 +54,7 @@ import sabnzbd.config as config
 import sabnzbd.cfg as cfg
 from sabnzbd.trylist import TryList
 from sabnzbd.encoding import unicoder, platform_encode, name_fixer
-from sabnzbd.database import get_history_handle
+from sabnzbd.database import HistoryDB
 from sabnzbd.rating import Rating
 
 __all__ = ['Article', 'NzbFile', 'NzbObject']
@@ -1556,7 +1556,7 @@ class NzbObject(TryList):
             return False, False
 
         res = False
-        history_db = get_history_handle()
+        history_db = HistoryDB()
 
         # dupe check off nzb contents
         if no_dupes:

sabnzbd/osxmenu.py

@@ -51,7 +51,7 @@ import sabnzbd.scheduler as scheduler
 import sabnzbd.downloader
 import sabnzbd.dirscanner as dirscanner
 from sabnzbd.bpsmeter import BPSMeter
-from sabnzbd.database import get_history_handle
+from sabnzbd.database import HistoryDB
 from sabnzbd.encoding import unicoder
 
 status_icons = {'idle': '../Resources/sab_idle.tiff', 'pause': '../Resources/sab_pause.tiff', 'clicked': '../Resources/sab_clicked.tiff'}
@@ -395,7 +395,7 @@ class SABnzbdDelegate(NSObject):
         try:
             # Fetch history items
             if not self.history_db:
-                self.history_db = sabnzbd.database.get_history_handle()
+                self.history_db = sabnzbd.database.HistoryDB()
             items, fetched_items, _total_items = self.history_db.fetch_history(0, 10, None)
 
             self.menu_history = NSMenu.alloc().init()
@@ -695,7 +695,7 @@ class SABnzbdDelegate(NSObject):
             NzbQueue.do.remove_all()
         elif mode == "history":
             if not self.history_db:
-                self.history_db = sabnzbd.database.get_history_handle()
+                self.history_db = sabnzbd.database.HistoryDB()
             self.history_db.remove_history()
 
     def pauseAction_(self, sender):

sabnzbd/postproc.py

@@ -543,7 +543,7 @@ def process_job(nzo):
     postproc_time = int(time.time() - start)
 
     # Create the history DB instance
-    history_db = database.get_history_handle()
+    history_db = database.HistoryDB()
     # Add the nzo to the database. Only the path, script and time taken is passed
     # Other information is obtained from the nzo
     history_db.add_history_db(nzo, clip_path(workdir_complete), nzo.downpath, postproc_time, script_log, script_line)