Fix several issues when undoing transactions with conflict resolutions

neo/client/app.py

@@ -655,7 +655,13 @@ class Application(ThreadedApplication):
         undo_object_tid_dict = {}
         snapshot_tid = p64(u64(self.last_tid) + 1)
         for partition, oid_list in partition_oid_dict.iteritems():
-            cell_list = getCellList(partition, readable=True)
+            cell_list = [cell
+                for cell in getCellList(partition, readable=True)
+                # Exclude nodes that may have missed previous resolved
+                # conflicts. For example, if a network failure happened only
+                # between the client and the storage, the latter would still
+                # be readable until we commit.
+                if txn_context.involved_nodes.get(cell.getUUID(), 0) < 2]
             # We do want to shuffle before getting one with the smallest
             # key, so that all cells with the same (smallest) key has
             # identical chance to be chosen.
@@ -685,7 +691,11 @@ class Application(ThreadedApplication):
                 # object. This is an undo conflict, try to resolve it.
                 try:
                     # Load the latest version we are supposed to see
-                    data = self.load(oid, current_serial)[0]
+                    if current_serial == ttid:
+                        # XXX: see TODO below
+                        data = txn_context.cache_dict[oid]
+                    else:
+                        data = self.load(oid, current_serial)[0]
                     # Load the version we were undoing to
                     undo_data = self.load(oid, undo_serial)[0]
                 except NEOStorageNotFoundError:
@@ -699,8 +709,14 @@ class Application(ThreadedApplication):
                     raise UndoError('Some data were modified by a later ' \
                         'transaction', oid)
                 undo_serial = None
+                # TODO: The situation is similar to deadlock avoidance.
+                #       Reenable the cache size limit to avoid OOM when there's
+                #       a huge amount conflicting data, and get the data back
+                #       from the storage when it's not in cache_dict anymore.
+                txn_context.cache_size = - float('inf')
             self._store(txn_context, oid, current_serial, data, undo_serial)
 
+        self.waitStoreResponses(txn_context)
         return None, txn_oid_list
 
     def _insertMetadata(self, txn_info, extension):

neo/storage/database/manager.py

@@ -14,7 +14,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import threading
+import struct, threading
 from collections import defaultdict
 from contextlib import contextmanager
 from functools import wraps
@@ -569,11 +569,14 @@ class DatabaseManager(object):
                 return current_tid, current_tid
             return current_tid, tid
         if transaction_object:
-            current_tid = current_data_tid = u64(transaction_object[2])
+            try:
+                current_tid = current_data_tid = u64(transaction_object[2])
+            except struct.error:
+                current_tid = current_data_tid = tid
         else:
             current_tid, current_data_tid = getDataTID(before_tid=ltid)
-        if current_tid is None:
-            return (None, None, False)
+            if current_tid is None:
+                return None, None, False
         found_undone_tid, undone_data_tid = getDataTID(tid=undone_tid)
         assert found_undone_tid is not None, (oid, undone_tid)
         is_current = undone_data_tid in (current_data_tid, tid)

neo/tests/threaded/test.py

@@ -156,37 +156,45 @@ class Test(NEOThreadedTest):
                 self.assertEqual((x['tid'], x['size']), expected.pop())
             self.assertFalse(expected)
 
-    @with_cluster()
-    def testUndoConflict(self, cluster, conflict_during_store=False):
+    def _testUndoConflict(self, cluster, *inc):
         def waitResponses(orig, *args):
             orig(*args)
             p.revert()
-            ob.value += 3
-            t.commit()
-        if 1:
-            t, c = cluster.getTransaction()
-            c.root()[0] = ob = PCounterWithResolution()
             t.commit()
-            ob.value += 1
+        t, c = cluster.getTransaction()
+        c.root()[0] = ob = PCounterWithResolution()
+        t.commit()
+        tids = []
+        for x in inc:
+            ob.value += x
             t.commit()
-            undo = TransactionalUndo(cluster.db, (ob._p_serial,))
-            txn = transaction.Transaction()
-            undo.tpc_begin(txn)
-            if conflict_during_store:
-                with Patch(cluster.client, waitResponses=waitResponses) as p:
-                    undo.commit(txn)
-            else:
-                ob.value += 3
-                t.commit()
-                undo.commit(txn)
-            undo.tpc_vote(txn)
-            undo.tpc_finish(txn)
-            t.begin()
-            self.assertEqual(ob.value, 3)
+            tids.append(ob._p_serial)
+        undo = TransactionalUndo(cluster.db, tids)
+        txn = transaction.Transaction()
+        undo.tpc_begin(txn)
+        ob.value += 5
+        with Patch(cluster.client, waitResponses=waitResponses) as p:
+            undo.commit(txn)
+        undo.tpc_vote(txn)
+        undo.tpc_finish(txn)
+        t.begin()
+        self.assertEqual(ob.value, 5)
+        return ob
+
+    @with_cluster()
+    def testUndoConflictSmallCache(self, cluster):
+        big = 'x' * cluster.cache_size
+        def resolve(orig, *args):
+            state = orig(*args)
+            state['x'] = big
+            return state
+        with Patch(PCounterWithResolution, _p_resolveConflict=resolve):
+            self.assertEqual(self._testUndoConflict(cluster, 1, 3).x, big)
 
     @expectedFailure(POSException.ConflictError)
-    def testUndoConflictDuringStore(self):
-        self.testUndoConflict(True)
+    @with_cluster()
+    def testUndoConflictDuringStore(self, cluster):
+        self._testUndoConflict(cluster, 1)
 
     @with_cluster()
     def testStorageDataLock(self, cluster):

neo/tests/zodb/testUndo.py

@@ -28,14 +28,6 @@ class UndoTests(ZODBTestCase, StorageTestBase, TransactionalUndoStorage,
     checkTransactionalUndoAfterPack = expectedFailure()(
         TransactionalUndoStorage.checkTransactionalUndoAfterPack)
 
-for x in ('checkUndoMultipleConflictResolution',
-          'checkUndoMultipleConflictResolutionReversed'):
-    try:
-        setattr(UndoTests, x,
-            expectedFailure(KeyError)(getattr(TransactionalUndoStorage, x)))
-    except AttributeError:
-        pass
-
 if __name__ == "__main__":
     suite = unittest.makeSuite(UndoTests, 'check')
     unittest.main(defaultTest='suite')