mysql: fix remaining places where a server disconnection was not catched

neo/lib/exception.py

@@ -26,9 +26,6 @@ class PrimaryFailure(NeoException):
 class StoppedOperation(NeoException):
     pass
 
-class DatabaseFailure(NeoException):
-    pass
-
 class NodeNotReady(NeoException):
     pass
 

neo/lib/interfaces.py

@@ -22,14 +22,14 @@ def check_signature(reference, function):
     a, b, c, d = inspect.getargspec(function)
     x = len(A) - len(a)
     if x < 0: # ignore extra default parameters
-        if x + len(d) < 0:
+        if B or x + len(d) < 0:
             return False
         del a[x:]
         d = d[:x] or None
     elif x: # different signature
         # We have no need yet to support methods with default parameters.
         return a == A[:-x] and (b or a and c) and not (d or D)
-    return a == A and b == B and c == C and d == D
+    return a == A and (b or not B) and (c or not C) and d == D
 
 def implements(obj, ignore=()):
     ignore = set(ignore)

neo/storage/database/__init__.py

@@ -16,8 +16,6 @@
 
 LOG_QUERIES = False
 
-from neo.lib.exception import DatabaseFailure
-
 DATABASE_MANAGER_DICT = {
     'Importer': 'importer.ImporterDatabaseManager',
     'MySQL': 'mysqldb.MySQLDatabaseManager',
@@ -33,3 +31,6 @@ def getAdapterKlass(name):
 
 def buildDatabaseManager(name, args=(), kw={}):
     return getAdapterKlass(name)(*args, **kw)
+
+class DatabaseFailure(Exception):
+    pass

neo/storage/database/importer.py

@@ -23,10 +23,9 @@ from ConfigParser import SafeConfigParser
 from ZODB.config import storageFromString
 from ZODB.POSException import POSKeyError
 
-from . import buildDatabaseManager
+from . import buildDatabaseManager, DatabaseFailure
 from .manager import DatabaseManager
 from neo.lib import compress, logging, patch, util
-from neo.lib.exception import DatabaseFailure
 from neo.lib.interfaces import implements
 from neo.lib.protocol import BackendNotImplemented, MAX_TID
 

neo/storage/database/manager.py

@@ -19,9 +19,9 @@ from collections import defaultdict
 from contextlib import contextmanager
 from functools import wraps
 from neo.lib import logging, util
-from neo.lib.exception import DatabaseFailure
 from neo.lib.interfaces import abstract, requires
 from neo.lib.protocol import CellStates, NonReadableCell, ZERO_TID
+from . import DatabaseFailure
 
 def lazymethod(func):
     def getter(self):

neo/storage/database/mysqldb.py

@@ -16,6 +16,7 @@
 
 from binascii import a2b_hex
 from collections import OrderedDict
+from functools import wraps
 import MySQLdb
 from MySQLdb import DataError, IntegrityError, \
     OperationalError, ProgrammingError
@@ -33,18 +34,55 @@ import struct
 import sys
 import time
 
-from . import LOG_QUERIES
+from . import LOG_QUERIES, DatabaseFailure
 from .manager import DatabaseManager, splitOIDField
 from neo.lib import logging, util
-from neo.lib.exception import DatabaseFailure
 from neo.lib.interfaces import implements
 from neo.lib.protocol import CellStates, ZERO_OID, ZERO_TID, ZERO_HASH
 
 
+class MysqlError(DatabaseFailure):
+
+    def __init__(self, exc, query=None):
+        self.exc = exc
+        self.query = query
+
+    code = property(lambda self: self.exc.args[0])
+
+    def __str__(self):
+        msg = 'MySQL error %s: %s' % self.exc.args
+        return msg if self.query is None else '%s\nQuery: %s' % (
+            msg, getPrintableQuery(self.query[:1000]))
+
+
 def getPrintableQuery(query, max=70):
     return ''.join(c if c in string.printable and c not in '\t\x0b\x0c\r'
         else '\\x%02x' % ord(c) for c in query)
 
+def auto_reconnect(wrapped):
+    def wrapper(self, *args):
+        # Try 3 times at most. When it fails too often for the same
+        # query then the disconnection is likely caused by this query.
+        # We don't want to enter into an infinite loop.
+        retry = 2
+        while 1:
+            try:
+                return wrapped(self, *args)
+            except OperationalError as m:
+                # IDEA: Is it safe to retry in case of DISK_FULL ?
+                # XXX:  However, this would another case of failure that would
+                #       be unnoticed by other nodes (ADMIN & MASTER). When
+                #       there are replicas, it may be preferred to not retry.
+                if (self._active
+                    or SERVER_GONE_ERROR != m.args[0] != SERVER_LOST
+                    or not retry):
+                    raise MysqlError(m, *args)
+                logging.info('the MySQL server is gone; reconnecting')
+                assert not self._deferred
+                self.close()
+                retry -= 1
+    return wraps(wrapped)(wrapper)
+
 
 @implements
 class MySQLDatabaseManager(DatabaseManager):
@@ -65,9 +103,18 @@ class MySQLDatabaseManager(DatabaseManager):
             '(?:([^:]+)(?::(.*))?@)?([^~./]+)(.+)?$', database).groups()
 
     def _close(self):
-        self.conn.close()
+        try:
+            conn = self.__dict__.pop('conn')
+        except KeyError:
+            return
+        conn.close()
 
-    def _connect(self):
+    def __getattr__(self, attr):
+        if attr == 'conn':
+            self._tryConnect()
+        return DatabaseManager.__getattr__(self, attr)
+
+    def _tryConnect(self):
         kwd = {'db' : self.db, 'user' : self.user}
         if self.passwd is not None:
             kwd['passwd'] = self.passwd
@@ -118,22 +165,19 @@ class MySQLDatabaseManager(DatabaseManager):
                 % (name, self._max_allowed_packet // 1024))
         self._max_allowed_packet = int(value)
 
+    _connect = auto_reconnect(_tryConnect)
+
     def _commit(self):
         self.conn.commit()
         self._active = 0
 
+    @auto_reconnect
     def query(self, query):
         """Query data from a database."""
         if LOG_QUERIES:
             logging.debug('querying %s...',
                 getPrintableQuery(query.split('\n', 1)[0][:70]))
-        # Try 3 times at most. When it fails too often for the same
-        # query then the disconnection is likely caused by this query.
-        # We don't want to enter into an infinite loop.
-        retry = 2
-        while 1:
         conn = self.conn
-            try:
         conn.query(query)
         if query.startswith("SELECT "):
             r = conn.store_result()
@@ -141,20 +185,6 @@ class MySQLDatabaseManager(DatabaseManager):
                 tuple([d.tostring() if isinstance(d, array) else d
                       for d in row])
                 for row in r.fetch_row(r.num_rows())])
-                break
-            except OperationalError as m:
-                code, m = m.args
-                # IDEA: Is it safe to retry in case of DISK_FULL ?
-                # XXX:  However, this would another case of failure that would
-                #       be unnoticed by other nodes (ADMIN & MASTER). When
-                #       there are replicas, it may be preferred to not retry.
-                if self._active or SERVER_GONE_ERROR != code != SERVER_LOST \
-                   or not retry:
-                    raise DatabaseFailure('MySQL error %d: %s\nQuery: %s'
-                        % (code, m, getPrintableQuery(query[:1000])))
-                logging.info('the MySQL server is gone; reconnecting')
-                self._connect()
-                retry -= 1
         r = query.split(None, 1)[0]
         if r in ("INSERT", "REPLACE", "DELETE", "UPDATE"):
             self._active = 1
@@ -444,9 +474,9 @@ class MySQLDatabaseManager(DatabaseManager):
                     PARTITION p%u VALUES IN (%u))""" % (offset, offset)
                 for table in 'trans', 'obj':
                     try:
-                        self.conn.query(add % table)
-                    except OperationalError as e:
-                        if e.args[0] != SAME_NAME_PARTITION:
+                        self.query(add % table)
+                    except MysqlError as e:
+                        if e.code != SAME_NAME_PARTITION:
                             raise
 
     def dropPartitions(self, offset_list):
@@ -468,9 +498,9 @@ class MySQLDatabaseManager(DatabaseManager):
                 ','.join(' p%u' % i for i in offset_list)
             for table in 'trans', 'obj':
                 try:
-                    self.conn.query(drop % table)
-                except OperationalError as e:
-                    if e.args[0] != DROP_LAST_PARTITION:
+                    self.query(drop % table)
+                except MysqlError as e:
+                    if e.code != DROP_LAST_PARTITION:
                         raise
 
     def _getUnfinishedDataIdList(self):

neo/tests/storage/testStorageMySQL.py

@@ -15,14 +15,14 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import unittest
-from MySQLdb import NotSupportedError, OperationalError
+from MySQLdb import NotSupportedError, OperationalError, ProgrammingError
 from MySQLdb.constants.ER import UNKNOWN_STORAGE_ENGINE
 from ..mock import Mock
-from neo.lib.exception import DatabaseFailure
 from neo.lib.protocol import ZERO_OID
 from neo.lib.util import p64
-from .. import DB_PREFIX, DB_SOCKET, DB_USER
+from .. import DB_PREFIX, DB_SOCKET, DB_USER, Patch
 from .testStorageDBTests import StorageDBTests
+from neo.storage.database import DatabaseFailure
 from neo.storage.database.mysqldb import MySQLDatabaseManager
 
 
@@ -72,18 +72,11 @@ class StorageMySQLdbTests(StorageDBTests):
         from MySQLdb.constants.CR import SERVER_GONE_ERROR
         class FakeConn(object):
             def query(*args):
+                p.revert()
                 raise OperationalError(SERVER_GONE_ERROR, 'this is a test')
-        self.db.conn = FakeConn()
-        self.connect_called = False
-        def connect_hook():
-            # mock object, break raise/connect loop
-            self.db.conn = Mock()
-            self.connect_called = True
-        self.db._connect = connect_hook
-        # make a query, exception will be raised then connect() will be
-        # called and the second query will use the mock object
-        self.db.query('INSERT')
-        self.assertTrue(self.connect_called)
+        with Patch(self.db, conn=FakeConn()) as p:
+            self.assertRaises(ProgrammingError, self.db.query, 'QUERY')
+            self.assertFalse(p.applied)
 
     def test_query3(self):
         # OperationalError > raise DatabaseFailure exception

neo/tests/threaded/test.py

@@ -30,7 +30,7 @@ from ZODB.DB import TransactionalUndo
 from neo.storage.transactions import TransactionManager, ConflictError
 from neo.lib.connection import ConnectionClosed, \
     ServerConnection, MTClientConnection
-from neo.lib.exception import DatabaseFailure, StoppedOperation
+from neo.lib.exception import StoppedOperation
 from neo.lib.handler import DelayEvent, EventHandler
 from neo.lib import logging
 from neo.lib.protocol import (CellStates, ClusterStates, NodeStates, NodeTypes,
@@ -42,6 +42,7 @@ from neo.lib.util import add64, makeChecksum, p64, u64
 from neo.client.exception import NEOPrimaryMasterLost, NEOStorageError
 from neo.client.transactions import Transaction
 from neo.master.handlers.client import ClientServiceHandler
+from neo.storage.database import DatabaseFailure
 from neo.storage.handlers.client import ClientOperationHandler
 from neo.storage.handlers.identification import IdentificationHandler
 from neo.storage.handlers.initialization import InitializationHandler