Ignore extra serials we might have found for last oid.

This caused incorrect object deletions during replication, when a chunk
lacks rows.

git-svn-id: https://svn.erp5.org/repos/neo/trunk@2410 71dcc9de-d417-0410-9af5-da40c76e7ee4

neo/storage/handlers/replication.py

@@ -133,10 +133,21 @@ class ReplicationHandler(EventHandler):
         ask = conn.ask
         my_object_dict = app.replicator.getObjectHistoryFromResult()
         deleteObject = app.dm.deleteObject
+        max_oid = max(object_dict.iterkeys())
+        max_serial = max(object_dict[max_oid])
         for oid, serial_list in object_dict.iteritems():
             # Check if I have objects, request those which I don't have.
             if oid in my_object_dict:
-                my_serial_set = frozenset(my_object_dict[oid])
+                # We must ignore extra serials we might have locally found for
+                # last received oid, as they can just be present in our list
+                # because we lacked some records (hence, we would have fetched
+                # rows further than other node for the same number of rows).
+                if oid == max_oid:
+                    my_serial_list = (x for x in my_object_dict[oid]
+                        if x <= max_serial)
+                else:
+                    my_serial_list = my_object_dict[oid]
+                my_serial_set = frozenset(my_serial_list)
                 serial_set = frozenset(serial_list)
                 extra_serial_set = my_serial_set - serial_set
                 for serial in extra_serial_set:

neo/tests/storage/testReplicationHandler.py

@@ -97,6 +97,7 @@ class StorageReplicationHandlerTests(NeoUnitTestBase):
             replicator = real_replicator
             dm = Mock({
                 'storeTransaction': None,
+                'deleteObject': None,
             })
         return FakeApp
 
@@ -194,10 +195,12 @@ class StorageReplicationHandlerTests(NeoUnitTestBase):
         oid_1 = self.getOID(1)
         oid_2 = self.getOID(2)
         oid_3 = self.getOID(3)
+        oid_4 = self.getOID(4)
+        tid_list = [self.getNextTID() for x in xrange(7)]
         oid_dict = FakeDict((
-            (oid_1, [self.getNextTID(), self.getNextTID()]),
-            (oid_2, [self.getNextTID()]),
-            (oid_3, [self.getNextTID()]),
+            (oid_1, [tid_list[0], tid_list[1]]),
+            (oid_2, [tid_list[3]]),
+            (oid_3, [tid_list[5]]),
         ))
         flat_oid_list = []
         for oid, serial_list in oid_dict.iteritems():
@@ -208,16 +211,23 @@ class StorageReplicationHandlerTests(NeoUnitTestBase):
         ReplicationHandler(app).answerObjectHistoryFrom(conn, oid_dict)
         self._checkPacketSerialList(conn, flat_oid_list)
         # With some known OID/Serials
+        # For test to be realist, history_result should contain the same
+        # number of serials as oid_dict, otherise it just tests the special
+        # case of the last check in a partition.
         conn = self.getFakeConnection()
         app = self.getApp(conn=conn, history_result={
             oid_1: [oid_dict[oid_1][0], ],
-            oid_3: [oid_dict[oid_3][0], ],
+            oid_3: [tid_list[4], oid_dict[oid_3][0], tid_list[6]],
+            oid_4: [tid_list[2], ],
         })
         ReplicationHandler(app).answerObjectHistoryFrom(conn, oid_dict)
         self._checkPacketSerialList(conn, (
             (oid_1, oid_dict[oid_1][1]),
             (oid_2, oid_dict[oid_2][0]),
         ))
+        calls = app.dm.mockGetNamedCalls('deleteObject')
+        self.assertEqual(len(calls), 1)
+        calls[0].checkArgs(oid_3, tid_list[4])
 
     def test_answerObject(self):
         conn = self.getFakeConnection()