Bump protocol version

neo/client/Storage.py

@@ -14,8 +14,9 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from ZODB import BaseStorage, ConflictResolution, POSException
-from ZODB.POSException import ConflictError, UndoError
+from ZODB import BaseStorage, ConflictResolution
+from ZODB.POSException import (
+    ConflictError, POSKeyError, ReadOnlyError, UndoError)
 from zope.interface import implementer
 import ZODB.interfaces
 
@@ -25,7 +26,7 @@ from .app import Application
 from .exception import NEOStorageNotFoundError, NEOStorageDoesNotExistError
 
 def raiseReadOnlyError(*args, **kw):
-    raise POSException.ReadOnlyError()
+    raise ReadOnlyError
 
 @implementer(
         ZODB.interfaces.IStorage,
@@ -39,7 +40,7 @@ class Storage(BaseStorage.BaseStorage,
               ConflictResolution.ConflictResolvingStorage):
     """Wrapper class for neoclient."""
 
-    def __init__(self, master_nodes, name, read_only=False,
+    def __init__(self, master_nodes, name,
             compress=None, logfile=None, _app=None, **kw):
         """
         Do not pass those parameters (used internally):
@@ -50,30 +51,28 @@ class Storage(BaseStorage.BaseStorage,
         if logfile:
             logging.setup(logfile)
         BaseStorage.BaseStorage.__init__(self, 'NEOStorage(%s)' % (name, ))
-        # Warning: _is_read_only is used in BaseStorage, do not rename it.
-        self._is_read_only = read_only
-        if read_only:
-            for method_id in (
-                        'new_oid',
-                        'tpc_begin',
-                        'tpc_vote',
-                        'tpc_abort',
-                        'store',
-                        'deleteObject',
-                        'undo',
-                        'undoLog',
-                    ):
-                setattr(self, method_id, raiseReadOnlyError)
         if _app is None:
             ssl = [kw.pop(x, None) for x in ('ca', 'cert', 'key')]
             _app = Application(master_nodes, name, compress=compress,
                                ssl=ssl if any(ssl) else None, **kw)
         self.app = _app
+        if __debug__ and self._is_read_only:
+            # For ZODB checkWriteMethods:
+            self.store = self.undo = raiseReadOnlyError
+            # For tpc_begin, it's checked in Application because it's used
+            # internally (e.g. pack) and the caller does not want to clean up
+            # with tpc_abort.
+            # For other methods, either the master rejects with
+            # READ_ONLY_ACCESS or the call is outside of a transaction.
 
     @property
     def _cache(self):
         return self.app._cache
 
+    @property
+    def _is_read_only(self): # used in BaseStorage, do not rename it
+        return self.app.read_only
+
     def load(self, oid, version=''):
         # XXX: interface definition states that version parameter is
         # mandatory, while some ZODB tests do not provide it. For now, make
@@ -82,7 +81,7 @@ class Storage(BaseStorage.BaseStorage,
         try:
             return self.app.load(oid)[:2]
         except NEOStorageNotFoundError:
-            raise POSException.POSKeyError(oid)
+            raise POSKeyError(oid)
         except Exception:
             logging.exception('oid=%r', oid)
             raise
@@ -151,7 +150,7 @@ class Storage(BaseStorage.BaseStorage,
         try:
             return self.app.load(oid, serial)[0]
         except NEOStorageNotFoundError:
-            raise POSException.POSKeyError(oid)
+            raise POSKeyError(oid)
         except Exception:
             logging.exception('oid=%r, serial=%r', oid, serial)
             raise
@@ -160,7 +159,7 @@ class Storage(BaseStorage.BaseStorage,
         try:
             return self.app.load(oid, None, tid)
         except NEOStorageDoesNotExistError:
-            raise POSException.POSKeyError(oid)
+            raise POSKeyError(oid)
         except NEOStorageNotFoundError:
             return None
         except Exception:
@@ -195,7 +194,7 @@ class Storage(BaseStorage.BaseStorage,
         try:
             data, serial, _ = self.app.load(oid)
         except NEOStorageNotFoundError:
-            raise POSException.POSKeyError(oid)
+            raise POSKeyError(oid)
         except Exception:
             logging.exception('oid=%r', oid)
             raise
@@ -215,7 +214,7 @@ class Storage(BaseStorage.BaseStorage,
         try:
             return self.app.history(oid, *args, **kw)
         except NEOStorageNotFoundError:
-            raise POSException.POSKeyError(oid)
+            raise POSKeyError(oid)
         except Exception:
             logging.exception('oid=%r', oid)
             raise

neo/client/app.py

@@ -25,7 +25,8 @@ try:
 except ImportError:
     from cPickle import dumps, loads
     _protocol = 1
-from ZODB.POSException import UndoError, ConflictError, ReadConflictError
+from ZODB.POSException import (
+    ConflictError, ReadConflictError, ReadOnlyError, UndoError)
 
 from neo.lib import logging
 from neo.lib.compress import decompress_list, getCompress
@@ -72,7 +73,7 @@ class Application(ThreadedApplication):
     wait_for_pack = False
 
     def __init__(self, master_nodes, name, compress=True, cache_size=None,
-                 ignore_wrong_checksum=False, **kw):
+                 read_only=False, ignore_wrong_checksum=False, **kw):
         super(Application, self).__init__(parseMasterList(master_nodes),
                                           name, **kw)
         # Internal Attributes common to all thread
@@ -108,6 +109,7 @@ class Application(ThreadedApplication):
         self._connecting_to_storage_node = Lock()
         self._node_failure_dict = {}
         self.compress = getCompress(compress)
+        self.read_only = read_only
         self.ignore_wrong_checksum = ignore_wrong_checksum
 
     def __getattr__(self, attr):
@@ -200,6 +202,8 @@ class Application(ThreadedApplication):
         fail_count = 0
         ask = self._ask
         handler = self.primary_bootstrap_handler
+        conn = None
+        try:
             while 1:
                 self.ignore_invalidations = True
                 # Get network connection to primary master
@@ -213,10 +217,10 @@ class Application(ThreadedApplication):
                         # Otherwise, check one by one.
                         master_list = self.nm.getMasterList()
                         if not master_list:
-                        # XXX: On shutdown, it already happened that this list
-                        #      is empty, leading to ZeroDivisionError. This
-                        #      looks a minor issue so let's wait to have more
-                        #      information.
+                            # XXX: On shutdown, it already happened that this
+                            #      list is empty, leading to ZeroDivisionError.
+                            #      This looks a minor issue so let's wait to
+                            #      have more information.
                             logging.error('%r', self.__dict__)
                         index = (index + 1) % len(master_list)
                         node = master_list[index]
@@ -226,10 +230,12 @@ class Application(ThreadedApplication):
                         node=node,
                         dispatcher=self.dispatcher)
                     p = Packets.RequestIdentification(NodeTypes.CLIENT,
-                    self.uuid, None, self.name, None, {})
+                        self.uuid, None, self.name, None,
+                        {'read_only': True} if self.read_only else {})
                     try:
                         ask(conn, p, handler=handler)
                     except ConnectionClosed:
+                        conn = None
                         fail_count += 1
                     else:
                         self.primary_master_node = node
@@ -247,9 +253,14 @@ class Application(ThreadedApplication):
                     if self.pt.operational():
                         break
                 except ConnectionClosed:
-                logging.error('Connection to %s lost', self.trying_master_node)
-                self.primary_master_node = None
+                    conn = self.primary_master_node = None
+                    logging.error('Connection to %s lost',
+                                  self.trying_master_node)
                 fail_count += 1
+        except:
+            if conn is not None:
+                conn.close()
+            raise
         logging.info("Connected and ready")
         return conn
 
@@ -497,6 +508,8 @@ class Application(ThreadedApplication):
 
     def tpc_begin(self, storage, transaction, tid=None, status=' '):
         """Begin a new transaction."""
+        if self.read_only:
+            raise ReadOnlyError
         # First get a transaction, only one is allowed at a time
         txn_context = self._txn_container.new(transaction)
         # use the given TID or request a new one to the master
@@ -523,6 +536,9 @@ class Application(ThreadedApplication):
             compressed_data = ''
             compression = 0
             checksum = ZERO_HASH
+            if data_serial is None:
+                assert oid not in txn_context.resolved_dict, oid
+                txn_context.delete_list.append(oid)
         else:
             size, compression, compressed_data = self.compress(data)
             checksum = makeChecksum(compressed_data)
@@ -573,6 +589,7 @@ class Application(ThreadedApplication):
                 'Conflict resolution succeeded for %s@%s with %s',
                 dump(oid), dump(old_serial), dump(serial))
             # Mark this conflict as resolved
+            assert oid not in txn_context.delete_list, oid
             resolved_dict[oid] = serial
             # Try to store again
             self._store(txn_context, oid, serial, data)
@@ -725,13 +742,22 @@ class Application(ThreadedApplication):
             self.tpc_vote(transaction)
         txn_context = txn_container.pop(transaction)
         cache_dict = txn_context.cache_dict
-        checked_list = [oid for oid, data  in cache_dict.iteritems()
-                            if data is CHECKED_SERIAL]
-        for oid in checked_list:
+        getPartition = self.pt.getPartition
+        checked = set()
+        for oid, data in cache_dict.items():
+            if data is CHECKED_SERIAL:
                 del cache_dict[oid]
+                checked.add(getPartition(oid))
+        deleted = txn_context.delete_list
+        if deleted:
+            oids = set(cache_dict)
+            oids.difference_update(deleted)
+            deleted = map(getPartition, deleted)
+        else:
+            oids = list(cache_dict)
         ttid = txn_context.ttid
-        p = Packets.AskFinishTransaction(ttid, list(cache_dict),
-                                         checked_list, txn_context.pack)
+        p = Packets.AskFinishTransaction(ttid, oids, deleted, checked,
+                                         txn_context.pack)
         try:
             tid = self._askPrimary(p, cache_dict=cache_dict, callback=f)
             assert tid

neo/client/component.xml

@@ -26,9 +26,9 @@
     </key>
     <key name="read-only" datatype="boolean">
       <description>
-        If true, only reads may be executed against the storage.  Note
-        that the "pack" operation is not considered a write operation
-        and is still allowed on a read-only neostorage.
+        If true, only reads may be executed against the storage.
+        If false when cluster is backing up, POSException.ReadOnlyError
+        is raised.
       </description>
     </key>
     <key name="logfile" datatype="existing-dirpath">

neo/client/transactions.py

@@ -38,6 +38,7 @@ class Transaction(object):
         self.txn = txn
         # data being stored
         self.data_dict = {}                 # {oid: (value, serial, [node_id])}
+        self.delete_list = []               # [oid]
         # data stored: this will go to the cache on tpc_finish
         self.cache_dict = {}                # {oid: value}
         # conflicts to resolve

neo/lib/protocol.py

@@ -26,7 +26,7 @@ except ImportError:
 
 # The protocol version must be increased whenever upgrading a node may require
 # to upgrade other nodes.
-PROTOCOL_VERSION = 4
+PROTOCOL_VERSION = 5
 # By encoding the handshake packet with msgpack, the whole NEO stream can be
 # decoded with msgpack. The first byte is 0x92, which is different from TLS
 # Handshake (0x16).
@@ -497,7 +497,14 @@ class Packets(dict):
 
     InvalidateObjects = notify("""
         Notify about a new transaction modifying objects,
-        invalidating client caches.
+        invalidating client caches. Deleted objects are excluded.
+
+        :nodes: M -> C
+        """)
+
+    InvalidatePartitions = notify("""
+        Notify about a new transaction, listing partitions
+        with modified or deleted objects.
 
         :nodes: M -> C
         """)

neo/master/app.py

@@ -150,7 +150,10 @@ class Application(BaseApplication):
         self.election_handler = master.ElectionHandler(self)
         self.secondary_handler = master.SecondaryHandler(self)
         self.client_service_handler = client.ClientServiceHandler(self)
-        self.client_ro_service_handler = client.ClientReadOnlyServiceHandler(self)
+        self.client_ro_service_handler = client.ClientReadOnlyServiceHandler(
+            self)
+        self.client_backup_service_handler = client.ClientBackupServiceHandler(
+            self)
         self.storage_service_handler = storage.StorageServiceHandler(self)
 
         registerLiveDebugger(on_log=self.log)
@@ -559,23 +562,26 @@ class Application(BaseApplication):
         # I have received all the lock answers now:
         # - send a Notify Transaction Finished to the initiated client node
         # - Invalidate Objects to the other client nodes
-        ttid = txn.getTTID()
-        tid = txn.getTID()
-        transaction_node = txn.getNode()
-        invalidate_objects = Packets.InvalidateObjects(tid, txn.getOIDList())
+        ttid = txn.ttid
+        tid = txn.tid
+        transaction_node = txn.node
+        invalidate_objects = Packets.InvalidateObjects(tid, txn.oid_list)
+        invalidate_partitions = Packets.InvalidatePartitions(
+            tid, txn.partition_list)
         client_list = self.nm.getClientList(only_identified=True)
         for client_node in client_list:
             if client_node is transaction_node:
                 client_node.send(Packets.AnswerTransactionFinished(ttid, tid),
-                                 msg_id=txn.getMessageId())
+                                 msg_id=txn.msg_id)
             else:
-                client_node.send(invalidate_objects)
+                client_node.send(invalidate_partitions
+                    if client_node.extra.get('backup') else
+                    invalidate_objects)
 
         # Unlock Information to relevant storage nodes.
         notify_unlock = Packets.NotifyUnlockInformation(ttid)
         getByUUID = self.nm.getByUUID
-        txn_storage_list = txn.getUUIDList()
-        for storage_uuid in txn_storage_list:
+        for storage_uuid in txn.involved:
             getByUUID(storage_uuid).send(notify_unlock)
 
         # Notify storage nodes about new pack order if any.

neo/master/handlers/__init__.py

@@ -41,7 +41,7 @@ class MasterHandler(EventHandler):
 
     def askLastIDs(self, conn):
         tm = self.app.tm
-        conn.answer(Packets.AnswerLastIDs(tm.getLastTID(), tm.getLastOID()))
+        conn.answer(Packets.AnswerLastIDs(tm.getLastTID(), tm.getLastOID(), tm.getFirstTID()))
 
     def askLastTransaction(self, conn):
         conn.answer(Packets.AnswerLastTransaction(

neo/master/handlers/administration.py

@@ -239,6 +239,9 @@ class AdministrationHandler(MasterHandler):
         app = self.app
         if app.getLastTransaction() <= tid:
             raise AnswerDenied("Truncating after last transaction does nothing")
+        if app.tm.getFirstTID() > tid:
+            raise AnswerDenied("Truncating before first transaction is"
+                               " probably not what you intended to do")
         if app.pm.getApprovedRejected(add64(tid, 1))[0]:
             # TODO: The protocol must be extended to support safe cases
             #       (e.g. no started pack whose id is after truncation tid).

neo/master/handlers/backup.py

@@ -63,13 +63,12 @@ class BackupHandler(EventHandler):
             raise RuntimeError("upstream DB truncated")
         app.ignore_invalidations = False
 
-    def invalidateObjects(self, conn, tid, oid_list):
+    def invalidatePartitions(self, conn, tid, partition_list):
         app = self.app
         if app.ignore_invalidations:
             return
-        getPartition = app.app.pt.getPartition
-        partition_set = set(map(getPartition, oid_list))
-        partition_set.add(getPartition(tid))
+        partition_set = set(partition_list)
+        partition_set.add(app.app.pt.getPartition(tid))
         prev_tid = app.app.getLastTransaction()
         app.invalidatePartitions(tid, prev_tid, partition_set)
 

neo/master/handlers/client.py

@@ -64,7 +64,8 @@ class ClientServiceHandler(MasterHandler):
         conn.answer((Errors.Ack if app.tm.vote(app, *args) else
                      Errors.IncompleteTransaction)())
 
-    def askFinishTransaction(self, conn, ttid, oid_list, checked_list, pack):
+    def askFinishTransaction(self, conn, ttid, oid_list,
+                             deleted, checked, pack):
         app = self.app
         if pack:
             tid = pack[1]
@@ -74,7 +75,8 @@ class ClientServiceHandler(MasterHandler):
             app,
             ttid,
             oid_list,
-            checked_list,
+            deleted,
+            checked,
             conn.getPeerId(),
         )
         if tid:
@@ -131,12 +133,13 @@ class ClientServiceHandler(MasterHandler):
         else:
             pack.waitForPack(conn.delayedAnswer(Packets.WaitedForPack))
 
-# like ClientServiceHandler but read-only & only for tid <= backup_tid
+
 class ClientReadOnlyServiceHandler(ClientServiceHandler):
 
+    _read_only_message = 'read-only access as requested by the client'
+
     def _readOnly(self, conn, *args, **kw):
-        conn.answer(Errors.ReadOnlyAccess(
-            'read-only access because cluster is in backuping mode'))
+        conn.answer(Errors.ReadOnlyAccess(self._read_only_message))
 
     askBeginTransaction     = _readOnly
     askNewOIDs              = _readOnly
@@ -145,9 +148,15 @@ class ClientReadOnlyServiceHandler(ClientServiceHandler):
     askPack                 = _readOnly
     abortTransaction        = _readOnly
 
+
+# like ClientReadOnlyServiceHandler but only for tid <= backup_tid
+class ClientBackupServiceHandler(ClientReadOnlyServiceHandler):
+
+    _read_only_message = 'read-only access because cluster is in backuping mode'
+
     # XXX LastIDs is not used by client at all, and it requires work to determine
     # last_oid up to backup_tid, so just make it non-functional for client.
-    askLastIDs              = _readOnly
+    askLastIDs = ClientReadOnlyServiceHandler._readOnly.__func__ # Py3
 
     # like in MasterHandler but returns backup_tid instead of last_tid
     def askLastTransaction(self, conn):

neo/master/handlers/identification.py

@@ -17,7 +17,7 @@
 from neo.lib import logging
 from neo.lib.exception import NotReadyError, PrimaryElected, ProtocolError
 from neo.lib.handler import EventHandler
-from neo.lib.protocol import CellStates, ClusterStates, NodeStates, \
+from neo.lib.protocol import CellStates, ClusterStates, Errors, NodeStates, \
     NodeTypes, Packets, uuid_str
 from ..app import monotonic_time
 
@@ -63,10 +63,17 @@ class IdentificationHandler(EventHandler):
         new_nid = extra.pop('new_nid', None)
         state = NodeStates.RUNNING
         if node_type == NodeTypes.CLIENT:
+            read_only = extra.pop('read_only', 'backup' in extra)
             if app.cluster_state == ClusterStates.RUNNING:
-                handler = app.client_service_handler
+                handler = (app.client_ro_service_handler if read_only else
+                           app.client_service_handler)
             elif app.cluster_state == ClusterStates.BACKINGUP:
-                handler = app.client_ro_service_handler
+                if not read_only:
+                    conn.answer(Errors.ReadOnlyAccess(
+                        "read-write access requested"
+                        " but cluster is backing up"))
+                    return
+                handler = app.client_backup_service_handler
             else:
                 raise NotReadyError
             human_readable_node_type = ' client '

neo/master/transactions.py

@@ -20,87 +20,40 @@ from struct import pack, unpack
 from neo.lib import logging
 from neo.lib.exception import ProtocolError
 from neo.lib.handler import DelayEvent, EventQueue
-from neo.lib.protocol import uuid_str, ZERO_OID, ZERO_TID
+from neo.lib.protocol import uuid_str, ZERO_OID, ZERO_TID, MAX_TID
 from neo.lib.util import dump, u64, addTID, tidFromTime
 
 class Transaction(object):
     """
         A pending transaction
     """
-    _tid = None
-    _msg_id = None
-    _oid_list = None
-    _failed = frozenset()
-    _prepared = False
-    # uuid dict hold flag to known who has locked the transaction
-    _uuid_set = None
-    _lock_wait_uuid_set = None
+    tid = None
+    oid_list = ()
+    failed = \
+    involved = frozenset()
 
     def __init__(self, node, storage_readiness, ttid):
         """
             Prepare the transaction, set OIDs and UUIDs related to it
         """
-        self._node = node
-        self._storage_readiness = storage_readiness
-        self._ttid = ttid
+        self.node = node
+        self.storage_readiness = storage_readiness
+        self.ttid = ttid
         self._birth = time()
         # store storage uuids that must be notified at commit
         self._notification_set = set()
 
     def __repr__(self):
-        return "<%s(client=%r, tid=%r, oids=%r, storages=%r, age=%.2fs) at %x>" % (
+        return "<%s(client=%r, tid=%r, invalidated=%r, storages=%r, age=%.2fs) at %x>" % (
                 self.__class__.__name__,
-                self._node,
-                dump(self._tid),
-                map(dump, self._oid_list or ()),
-                map(uuid_str, self._uuid_set or ()),
+                self.node,
+                dump(self.tid),
+                map(dump, self.oid_list),
+                map(uuid_str, self.involved),
                 time() - self._birth,
                 id(self),
         )
 
-    def getNode(self):
-        """
-            Return the node that had began the transaction
-        """
-        return self._node
-
-    def getTTID(self):
-        """
-            Return the temporary transaction ID.
-        """
-        return self._ttid
-
-    def getTID(self):
-        """
-            Return the transaction ID
-        """
-        return self._tid
-
-    def getMessageId(self):
-        """
-            Returns the packet ID to use in the answer
-        """
-        return self._msg_id
-
-    def getUUIDList(self):
-        """
-            Returns the list of node's UUID that lock the transaction
-        """
-        return list(self._uuid_set)
-
-    def getOIDList(self):
-        """
-            Returns the list of OIDs used in the transaction
-        """
-
-        return list(self._oid_list)
-
-    def isPrepared(self):
-        """
-            Returns True if the commit has been requested by the client
-        """
-        return self._prepared
-
     def registerForNotification(self, uuid):
         """
             Register a node that requires a notification at commit
@@ -114,14 +67,13 @@ class Transaction(object):
         """
         return list(self._notification_set)
 
-    def prepare(self, tid, oid_list, uuid_set, msg_id):
-
-        self._tid = tid
-        self._oid_list = oid_list
-        self._msg_id = msg_id
-        self._uuid_set = uuid_set
-        self._lock_wait_uuid_set = uuid_set.copy()
-        self._prepared = True
+    def prepare(self, tid, oid_list, partition_list, involved, msg_id):
+        self.tid = tid
+        self.oid_list = oid_list
+        self.partition_list = partition_list
+        self.msg_id = msg_id
+        self.involved = involved
+        self.locking = involved.copy()
 
     def storageLost(self, uuid):
         """
@@ -133,16 +85,17 @@ class Transaction(object):
         # XXX: We might lose information that a storage successfully locked
         # data but was later found to be disconnected. This loss has no impact
         # on current code, but it might be disturbing to reader or future code.
-        if self._prepared:
-            self._lock_wait_uuid_set.discard(uuid)
-            self._uuid_set.discard(uuid)
-            return self.locked()
+        if self.tid:
+            locking = self.locking
+            locking.discard(uuid)
+            self.involved.discard(uuid)
+            return not locking
         return False
 
     def clientLost(self, node):
-        if self._node is node:
-            if self._prepared:
-                self._node = None # orphan
+        if self.node is node:
+            if self.tid:
+                self.node = None # orphan
             else:
                 return True # abort
         else:
@@ -154,14 +107,9 @@ class Transaction(object):
             Define that a node has locked the transaction
             Returns true if all nodes are locked
         """
-        self._lock_wait_uuid_set.remove(uuid)
-        return self.locked()
-
-    def locked(self):
-        """
-            Returns true if all nodes are locked
-        """
-        return not self._lock_wait_uuid_set
+        locking = self.locking
+        locking.remove(uuid)
+        return not locking
 
 
 class TransactionManager(EventQueue):
@@ -179,6 +127,9 @@ class TransactionManager(EventQueue):
         self._ttid_dict = {}
         self._last_oid = ZERO_OID
         self._last_tid = ZERO_TID
+        self._first_tid = MAX_TID
+        # avoid the overhead of min_tid on every _unlockPending
+        self._unlockPending = self._firstUnlockPending
         # queue filled with ttids pointing to transactions with increasing tids
         self._queue = deque()
 
@@ -212,6 +163,13 @@ class TransactionManager(EventQueue):
         self._last_oid = oid_list[-1]
         return oid_list
 
+    def setFirstTID(self, tid):
+        if self._first_tid > tid:
+            self._first_tid = tid
+
+    def getFirstTID(self):
+        return self._first_tid
+
     def setLastOID(self, oid):
         if self._last_oid < oid:
             self._last_oid = oid
@@ -302,7 +260,7 @@ class TransactionManager(EventQueue):
         txn = self[ttid]
         # The client does not know which nodes are not expected to have
         # transactions in full. Let's filter out them.
-        failed = app.getStorageReadySet(txn._storage_readiness)
+        failed = app.getStorageReadySet(txn.storage_readiness)
         failed.intersection_update(uuid_list)
         if failed:
             operational = app.pt.operational
@@ -312,41 +270,42 @@ class TransactionManager(EventQueue):
                 return False
             all_failed = failed.copy()
             for t in self._ttid_dict.itervalues():
-                all_failed |= t._failed
+                all_failed |= t.failed
             if not operational(all_failed):
                 # Other transactions were voted and unless they're aborted,
                 # we won't be able to finish this one, because that would make
-                # the cluster non-operational. Let's tell the caller to retry
-                # later.
+                # the cluster non-operational. Let's retry later.
                 raise DelayEvent
             # Allow the client to finish the transaction,
             # even if this will disconnect storage nodes.
-            txn._failed = failed
+            txn.failed = failed
         return True
 
-    def prepare(self, app, ttid, oid_list, checked_list, msg_id):
+    def prepare(self, app, ttid, oid_list, deleted, checked, msg_id):
         """
             Prepare a transaction to be finished
         """
         txn = self[ttid]
         pt = app.pt
 
-        failed = txn._failed
+        failed = txn.failed
         if failed and not pt.operational(failed):
             return None, None
-        ready = app.getStorageReadySet(txn._storage_readiness)
+        ready = app.getStorageReadySet(txn.storage_readiness)
         getPartition = pt.getPartition
         partition_set = set(map(getPartition, oid_list))
-        partition_set.update(map(getPartition, checked_list))
+        partition_set.update(deleted)
+        partition_list = list(partition_set)
         partition_set.add(getPartition(ttid))
+        partition_set.update(checked)
         node_list = []
-        uuid_set = set()
+        involved = set()
         for partition in partition_set:
             for cell in pt.getCellList(partition):
                 node = cell.getNode()
                 if node.isIdentified():
                     uuid = node.getUUID()
-                    if uuid in uuid_set:
+                    if uuid in involved:
                         continue
                     if uuid in failed:
                         # This will commit a new PT with outdated cells before
@@ -354,7 +313,7 @@ class TransactionManager(EventQueue):
                         # the verification phase.
                         node.getConnection().close()
                     elif uuid in ready:
-                        uuid_set.add(uuid)
+                        involved.add(uuid)
                         node_list.append(node)
         # A node that was not ready at the beginning of the transaction
         # can't have readable cells. And if we're still operational without
@@ -369,8 +328,8 @@ class TransactionManager(EventQueue):
             tid = self._nextTID(ttid, pt.getPartitions())
             self._queue.append(ttid)
         logging.debug('Finish TXN %s for %s (was %s)',
-                      dump(tid), txn.getNode(), dump(ttid))
-        txn.prepare(tid, oid_list, uuid_set, msg_id)
+                      dump(tid), txn.node, dump(ttid))
+        txn.prepare(tid, oid_list, partition_list, involved, msg_id)
         # check if greater and foreign OID was stored
         if oid_list:
             self.setLastOID(max(oid_list))
@@ -382,7 +341,7 @@ class TransactionManager(EventQueue):
         """
         logging.debug('Abort TXN %s for %s', dump(ttid), uuid_str(uuid))
         txn = self[ttid]
-        if txn.isPrepared():
+        if txn.tid:
             raise ProtocolError("commit already requested for ttid %s"
                                 % dump(ttid))
         del self[ttid]
@@ -412,6 +371,16 @@ class TransactionManager(EventQueue):
         if unlock:
             self._unlockPending()
 
+    def _firstUnlockPending(self):
+        """Set first TID when the first transaction is committed
+
+        Masks _unlockPending on reset.
+        Unmasks and call it when called.
+        """
+        self.setFirstTID(self._ttid_dict[self._queue[0]].tid)
+        del self._unlockPending
+        self._unlockPending()
+
     def _unlockPending(self):
         """Serialize transaction unlocks
 
@@ -424,7 +393,7 @@ class TransactionManager(EventQueue):
         while queue:
             ttid = queue[0]
             txn = self._ttid_dict[ttid]
-            if not txn.locked():
+            if txn.locking:
                 break
             del queue[0], self._ttid_dict[ttid]
             self._on_commit(txn)
@@ -433,7 +402,7 @@ class TransactionManager(EventQueue):
     def clientLost(self, node):
         for txn in self._ttid_dict.values():
             if txn.clientLost(node):
-                tid = txn.getTTID()
+                tid = txn.ttid
                 del self[tid]
                 yield tid, txn.getNotificationUUIDList()
 

neo/master/verification.py

@@ -139,11 +139,12 @@ class VerificationManager(BaseServiceHandler):
     def notifyPackCompleted(self, conn, pack_id):
         self.app.nm.getByUUID(conn.getUUID()).completed_pack_id = pack_id
 
-    def answerLastIDs(self, conn, ltid, loid):
+    def answerLastIDs(self, conn, ltid, loid, ftid):
         self._uuid_set.remove(conn.getUUID())
         tm = self.app.tm
         tm.setLastTID(ltid)
         tm.setLastOID(loid)
+        tm.setFirstTID(ftid)
 
     def answerPackOrders(self, conn, pack_list):
         self._uuid_set.remove(conn.getUUID())

neo/neoctl/app.py

+# -*- coding: utf-8 -*-
 #
 # Copyright (C) 2006-2019  Nexedi SA
 #
@@ -70,9 +71,9 @@ class TerminalNeoCTL(object):
         return getattr(ClusterStates, value.upper())
 
     def asTID(self, value):
-        if '.' in value:
+        if value.lower().startswith('tid:'):
+            return p64(int(value[4:], 0))
         return tidFromTime(float(value))
-        return p64(int(value, 0))
 
     asNode = staticmethod(uuid_int)
 
@@ -386,7 +387,8 @@ class Application(object):
 
     def usage(self):
         output_list = ('Available commands:', self._usage(action_dict),
-            "TID arguments can be either integers or timestamps as floats,"
-            " e.g. '257684787499560686', '0x3937af2eeeeeeee' or '1325421296.'"
-            " for 2012-01-01 12:34:56 UTC")
+            "The syntax of « TID » arguments is either tid:<integer>"
+            " (case insensitive) for a TID or <float> for a UNIX timestamp,"
+            " e.g. 'tid:257684787499560686', 'tid:0x3937af2eeeeeeee' or"
+            " '1325421296' for 2012-01-01 12:34:56 UTC.")
         return '\n'.join(output_list)

neo/neoctl/neoctl.py

@@ -137,7 +137,7 @@ class NeoCTL(BaseApplication):
         response = self.__ask(Packets.AskLastIDs())
         if response[0] != Packets.AnswerLastIDs:
             raise RuntimeError(response)
-        return response[1:]
+        return response[1:3]
 
     def getLastTransaction(self):
         response = self.__ask(Packets.AskLastTransaction())

neo/storage/database/importer.py

@@ -601,6 +601,9 @@ class ImporterDatabaseManager(DatabaseManager):
         zodb = self.zodb[bisect(self.zodb_index, oid) - 1]
         return zodb, oid - zodb.shift_oid
 
+    def getFirstTID(self):
+        return min(next(zodb.iterator()).tid for zodb in self.zodb)
+
     def getLastIDs(self):
         tid, oid = self.db.getLastIDs()
         return (max(tid, util.p64(self.zodb_ltid)),

neo/storage/database/manager.py

@@ -758,6 +758,19 @@ class DatabaseManager(object):
     # XXX: Consider splitting getLastIDs/_getLastIDs because
     #      sometimes the last oid is not wanted.
 
+    def _getFirstTID(self, partition):
+        """Return tid of first transaction in given 'partition'
+
+        tids are in unpacked format.
+        """
+
+    @requires(_getFirstTID)
+    def getFirstTID(self):
+        """Return tid of first transaction
+        """
+        x = self._readable_set
+        return util.p64(min(map(self._getFirstTID, x))) if x else MAX_TID
+
     def _getLastTID(self, partition, max_tid=None):
         """Return tid of last transaction <= 'max_tid' in given 'partition'
 

neo/storage/database/mysql.py

@@ -53,7 +53,7 @@ from .manager import MVCCDatabaseManager, splitOIDField
 from neo.lib import logging, util
 from neo.lib.exception import NonReadableCell, UndoPackError
 from neo.lib.interfaces import implements
-from neo.lib.protocol import CellStates, ZERO_OID, ZERO_TID, ZERO_HASH
+from neo.lib.protocol import CellStates, ZERO_OID, ZERO_TID, ZERO_HASH, MAX_TID
 
 
 class MysqlError(DatabaseFailure):
@@ -457,6 +457,12 @@ class MySQLDatabaseManager(MVCCDatabaseManager):
     def _getPartitionTable(self):
         return self.query("SELECT * FROM pt")
 
+    def _getFirstTID(self, partition):
+        (tid,), = self.query(
+            "SELECT MIN(tid) as t FROM trans FORCE INDEX (PRIMARY)"
+            " WHERE `partition`=%s" % partition)
+        return util.u64(MAX_TID) if tid is None else tid
+
     def _getLastTID(self, partition, max_tid=None):
         sql = ("SELECT MAX(tid) as t FROM trans FORCE INDEX (PRIMARY)"
                " WHERE `partition`=%s") % partition

neo/storage/database/sqlite.py

@@ -28,7 +28,7 @@ from .manager import DatabaseManager, splitOIDField
 from neo.lib import logging, util
 from neo.lib.exception import NonReadableCell, UndoPackError
 from neo.lib.interfaces import implements
-from neo.lib.protocol import CellStates, ZERO_OID, ZERO_TID, ZERO_HASH
+from neo.lib.protocol import CellStates, ZERO_OID, ZERO_TID, ZERO_HASH, MAX_TID
 
 def unique_constraint_message(table, *columns):
     c = sqlite3.connect(":memory:")
@@ -343,6 +343,11 @@ class SQLiteDatabaseManager(DatabaseManager):
     def _getPartitionTable(self):
         return self.query("SELECT * FROM pt")
 
+    def _getFirstTID(self, partition):
+        tid = self.query("SELECT MIN(tid) FROM trans WHERE partition=?",
+                         (partition,)).fetchone()[0]
+        return util.u64(MAX_TID) if tid is None else tid
+
     def _getLastTID(self, partition, max_tid=None):
         x = self.query
         if max_tid is None:

neo/storage/handlers/initialization.py

@@ -55,7 +55,8 @@ class InitializationHandler(BaseMasterHandler):
             if packed:
                 self.app.completed_pack_id = pack_id = min(packed.itervalues())
                 conn.send(Packets.NotifyPackCompleted(pack_id))
-        conn.answer(Packets.AnswerLastIDs(*dm.getLastIDs()))
+        last_tid, last_oid = dm.getLastIDs() # PY3
+        conn.answer(Packets.AnswerLastIDs(last_tid, last_oid, dm.getFirstTID()))
 
     def askPartitionTable(self, conn):
         pt = self.app.pt

neo/tests/protocol

@@ -13,7 +13,7 @@ AnswerFetchObjects(?p64,?p64,{:})
 AnswerFetchTransactions(?p64,[],?p64)
 AnswerFinalTID(p64)
 AnswerInformationLocked(p64)
-AnswerLastIDs(?p64,?p64)
+AnswerLastIDs(?p64,?p64,p64)
 AnswerLastTransaction(p64)
 AnswerLockedTransactions({p64:?p64})
 AnswerMonitorInformation([?bin],[?bin],bin)
@@ -46,7 +46,7 @@ AskClusterState()
 AskFetchObjects(int,int,p64,p64,p64,{p64:[p64]})
 AskFetchTransactions(int,int,p64,p64,[p64],bool)
 AskFinalTID(p64)
-AskFinishTransaction(p64,[p64],[p64],?(?[p64],p64))
+AskFinishTransaction(p64,[p64],[int],[int],?(?[p64],p64))
 AskLastIDs()
 AskLastTransaction()
 AskLockInformation(p64,p64,bool)
@@ -76,6 +76,7 @@ CheckReplicas({int:?int},p64,?)
 Error(int,bin)
 FailedVote(p64,[int])
 InvalidateObjects(p64,[p64])
+InvalidatePartitions(p64,[int])
 NotPrimaryMaster(?int,[(bin,int)])
 NotifyClusterInformation(ClusterStates)
 NotifyDeadlock(p64,p64)

neo/tests/storage/testStorageDBTests.py

@@ -183,6 +183,7 @@ class StorageDBTests(NeoUnitTestBase):
         txn1, objs1 = self.getTransaction([oid1])
         txn2, objs2 = self.getTransaction([oid2])
         # nothing in database
+        self.assertEqual(self.db.getFirstTID(), MAX_TID)
         self.assertEqual(self.db.getLastIDs(), (None, None))
         self.assertEqual(self.db.getUnfinishedTIDDict(), {})
         self.assertEqual(self.db.getObject(oid1), None)
@@ -199,6 +200,7 @@ class StorageDBTests(NeoUnitTestBase):
                 ([oid2], 'user', 'desc', 'ext', False, p64(2), None))
             self.assertEqual(self.db.getTransaction(tid1, False), None)
             self.assertEqual(self.db.getTransaction(tid2, False), None)
+        self.assertEqual(self.db.getFirstTID(), tid1)
         self.assertEqual(self.db.getTransaction(tid1, True),
             ([oid1], 'user', 'desc', 'ext', False, p64(1), None))
         self.assertEqual(self.db.getTransaction(tid2, True),

neo/tests/stress.py

@@ -200,7 +200,7 @@ class StressApplication(AdminApplication):
         if conn:
             conn.ask(Packets.AskLastIDs())
 
-    def answerLastIDs(self, ltid, loid):
+    def answerLastIDs(self, ltid, loid, ftid):
         self.loid = loid
         self.ltid = ltid
         self.em.setTimeout(int(time.time() + 1), self.askLastIDs)

neo/tests/threaded/__init__.py

@@ -1027,8 +1027,12 @@ class NEOCluster(object):
             if not cell.isReadable()]
 
     def getZODBStorage(self, **kw):
-        kw['_app'] = kw.pop('client', self.client)
-        return Storage.Storage(None, self.name, **kw)
+        try:
+            app = kw.pop('client')
+            assert not kw, kw
+        except KeyError:
+            app = self._newClient(**kw) if kw else self.client
+        return Storage.Storage(None, self.name, _app=app)
 
     def importZODB(self, dummy_zodb=None, random=random):
         if dummy_zodb is None:

neo/tests/threaded/test.py

@@ -121,28 +121,6 @@ class Test(NEOThreadedTest):
                 self.assertFalse(cluster.storage.sqlCount('bigdata'))
             self.assertFalse(cluster.storage.sqlCount('data'))
 
-    @with_cluster()
-    def testDeleteObject(self, cluster):
-        if 1:
-            storage = cluster.getZODBStorage()
-            for clear_cache in 0, 1:
-                for tst in 'a.', 'bcd.':
-                    oid = storage.new_oid()
-                    serial = None
-                    for data in tst:
-                        txn = transaction.Transaction()
-                        storage.tpc_begin(txn)
-                        if data == '.':
-                            storage.deleteObject(oid, serial, txn)
-                        else:
-                            storage.store(oid, serial, data, '', txn)
-                        storage.tpc_vote(txn)
-                        serial = storage.tpc_finish(txn)
-                        if clear_cache:
-                            storage._cache.clear()
-                    self.assertRaises(POSException.POSKeyError,
-                        storage.load, oid, '')
-
     @with_cluster(storage_count=3, replicas=1, partitions=5)
     def testIterOIDs(self, cluster):
         storage = cluster.getZODBStorage()

neo/tests/threaded/testReplication.py

@@ -17,13 +17,13 @@
 
 import random, sys, threading, time
 import transaction
-from ZODB.POSException import ReadOnlyError, POSKeyError
+from ZODB.POSException import (
+    POSKeyError, ReadOnlyError, StorageTransactionError)
 import unittest
 from collections import defaultdict
 from functools import wraps
 from itertools import product
 from neo.lib import logging
-from neo.client.exception import NEOStorageError
 from neo.master.handlers.backup import BackupHandler
 from neo.storage.checker import CHECK_COUNT
 from neo.storage.database.manager import DatabaseManager
@@ -220,7 +220,7 @@ class ReplicationTests(NEOThreadedTest):
             counts[0] += 1
             if counts[0] > 1:
                 node_list = orig.im_self.nm.getClientList(only_identified=True)
-                node_list.remove(txn.getNode())
+                node_list.remove(txn.node)
                 node_list[0].getConnection().close()
             return orig(txn)
         with NEOCluster(partitions=np, replicas=0, storage_count=1) as upstream:
@@ -266,7 +266,7 @@ class ReplicationTests(NEOThreadedTest):
     def testBackupUpstreamStorageDead(self, backup):
         upstream = backup.upstream
         with ConnectionFilter() as f:
-            f.delayInvalidateObjects()
+            f.delayInvalidatePartitions()
             upstream.importZODB()(1)
         count = [0]
         def _connect(orig, conn):
@@ -1112,7 +1112,9 @@ class ReplicationTests(NEOThreadedTest):
         B = backup
         U = B.upstream
         Z = U.getZODBStorage()
-        #Zb = B.getZODBStorage()    # XXX see below about invalidations
+        with B.newClient() as client, self.assertRaises(ReadOnlyError):
+            client.last_tid
+        #Zb = B.getZODBStorage(read_only=True) # XXX see below about invalidations
 
         oid_list = []
         tid_list = []
@@ -1157,7 +1159,7 @@ class ReplicationTests(NEOThreadedTest):
                 # read data from B and verify it is what it should be
                 # XXX we open new ZODB storage every time because invalidations
                 # are not yet implemented in read-only mode.
-                Zb = B.getZODBStorage()
+                Zb = B.getZODBStorage(read_only=True)
                 for j, oid in enumerate(oid_list):
                     if cutoff <= i < recover and j >= cutoff:
                         self.assertRaises(POSKeyError, Zb.load, oid, '')
@@ -1170,7 +1172,6 @@ class ReplicationTests(NEOThreadedTest):
                 # not-yet-fully fetched backup state (transactions committed at
                 # [cutoff, recover) should not be there; otherwise transactions
                 # should be fully there)
-                Zb = B.getZODBStorage()
                 Btxn_list = list(Zb.iterator())
                 self.assertEqual(len(Btxn_list), cutoff if cutoff <= i < recover
                                                  else i+1)
@@ -1185,15 +1186,12 @@ class ReplicationTests(NEOThreadedTest):
 
                 # try to commit something to backup storage and make sure it is
                 # really read-only
-                Zb._cache.max_size = 0     # make store() do work in sync way
                 txn = transaction.Transaction()
                 self.assertRaises(ReadOnlyError, Zb.tpc_begin, txn)
                 self.assertRaises(ReadOnlyError, Zb.new_oid)
-                self.assertRaises(ReadOnlyError, Zb.store, oid_list[-1],
-                                            tid_list[-1], 'somedata', '', txn)
-                # tpc_vote first checks whether there were store replies -
+                # tpc_vote first checks whether the transaction has begun -
                 # thus not ReadOnlyError
-                self.assertRaises(NEOStorageError, Zb.tpc_vote, txn)
+                self.assertRaises(StorageTransactionError, Zb.tpc_vote, txn)
 
                 if i == loop // 2:
                     # Check that we survive a disconnection from upstream
@@ -1203,8 +1201,6 @@ class ReplicationTests(NEOThreadedTest):
                     conn.close()
                     self.tic()
 
-                # close storage because client app is otherwise shared in
-                # threaded tests and we need to refresh last_tid on next run
                 # (XXX see above about invalidations not working)
                 Zb.close()
 
@@ -1231,5 +1227,29 @@ class ReplicationTests(NEOThreadedTest):
         self.assertEqual(1, self.checkBackup(backup))
 
 
+    @backup_test(3)
+    def testDeleteObject(self, backup):
+        upstream = backup.upstream
+        storage = upstream.getZODBStorage()
+        for clear_cache in 0, 1:
+            for tst in 'a.', 'bcd.':
+                oid = storage.new_oid()
+                serial = None
+                for data in tst:
+                    txn = transaction.Transaction()
+                    storage.tpc_begin(txn)
+                    if data == '.':
+                        storage.deleteObject(oid, serial, txn)
+                    else:
+                        storage.store(oid, serial, data, '', txn)
+                    storage.tpc_vote(txn)
+                    serial = storage.tpc_finish(txn)
+                    self.tic()
+                    self.assertEqual(3, self.checkBackup(backup))
+                    if clear_cache:
+                        storage._cache.clear()
+                self.assertRaises(POSKeyError, storage.load, oid, '')
+
+
 if __name__ == "__main__":
     unittest.main()