nexedi / neoppod

When ConnectionPool._initNodeConnection fails a first time with:

  StorageError: protocol error: already connected

the following assertion failure happens when trying to reconnect before the
previous connection is actually closed (currently, only the node sending an
error message closes the connection, as commented in EventHandler):

  Traceback (most recent call last):
    File "neo/client/Storage.py", line 82, in load
      return self.app.load(oid)[:2]
    File "neo/client/app.py", line 367, in load
      data, tid, next_tid, _ = self._loadFromStorage(oid, tid, before_tid)
    File "neo/client/app.py", line 399, in _loadFromStorage
      askStorage)
    File "neo/client/app.py", line 293, in _askStorageForRead
      conn = cp.getConnForNode(node)
    File "neo/client/pool.py", line 98, in getConnForNode
      conn = self._initNodeConnection(node)
    File "neo/client/pool.py", line 48, in _initNodeConnection
      dispatcher=app.dispatcher)
    File "neo/lib/connection.py", line 704, in __init__
      super(MTClientConnection, self).__init__(*args, **kwargs)
    File "neo/lib/connection.py", line 602, in __init__
      node.setConnection(self)
    File "neo/lib/node.py", line 122, in setConnection
      attributeTracker.whoSet(self, '_connection'))
  AssertionError

This fixes:

  Traceback (most recent call last):
    File "neo/client/Storage.py", line 108, in tpc_vote
      return self.app.tpc_vote(transaction)
    File "neo/client/app.py", line 546, in tpc_vote
      self.waitStoreResponses(txn_context)
    File "neo/client/app.py", line 539, in waitStoreResponses
      _waitAnyTransactionMessage(txn_context)
    File "neo/client/app.py", line 160, in _waitAnyTransactionMessage
      self._handleConflicts(txn_context)
    File "neo/client/app.py", line 514, in _handleConflicts
      self._store(txn_context, oid, serial, data)
    File "neo/client/app.py", line 452, in _store
      self._waitAnyTransactionMessage(txn_context, False)
    File "neo/client/app.py", line 155, in _waitAnyTransactionMessage
      self._waitAnyMessage(queue, block=block)
    File "neo/client/app.py", line 142, in _waitAnyMessage
      _handlePacket(conn, packet, kw)
    File "neo/lib/threaded_app.py", line 133, in _handlePacket
      handler.dispatch(conn, packet, kw)
    File "neo/lib/handler.py", line 72, in dispatch
      method(conn, *args, **kw)
    File "neo/client/handlers/storage.py", line 143, in answerRebaseObject
      assert cached == data
  AssertionError

During rebase, writes could stay lockless although the partition was
replicated. Another transaction could then take locks prematurely, leading to
the following crash:

  Traceback (most recent call last):
    File "neo/lib/handler.py", line 72, in dispatch
      method(conn, *args, **kw)
    File "neo/storage/handlers/master.py", line 36, in notifyUnlockInformation
      self.app.tm.unlock(ttid)
    File "neo/storage/transactions.py", line 329, in unlock
      self.abort(ttid, even_if_locked=True)
    File "neo/storage/transactions.py", line 573, in abort
      not self._replicated.get(self.getPartition(oid))), x
  AssertionError: ('\x00\x00\x00\x00\x00\x03\x03v', '\x03\xca\xb44J\x13\x99\x88', '\x03\xca\xb44J\xe0\xdcU', {}, set(['\x00\x00\x00\x00\x00\x03\x03v']))

Flushing logs will help fixing NEO bugs (e.g. failed assertions).

This fixes:

  Traceback (innermost last):
    ...
    Module transaction._transaction, line 393, in _commitResources
      rm.tpc_vote(self)
    Module ZODB.Connection, line 797, in tpc_vote
      s = vote(transaction)
    Module neo.client.Storage, line 95, in tpc_vote
      return self.app.tpc_vote(transaction)
    Module neo.client.app, line 546, in tpc_vote
      self.waitStoreResponses(txn_context)
    Module neo.client.app, line 539, in waitStoreResponses
      _waitAnyTransactionMessage(txn_context)
    Module neo.client.app, line 160, in _waitAnyTransactionMessage
      self._handleConflicts(txn_context)
    Module neo.client.app, line 471, in _handleConflicts
      assert oid is None, (oid, serial)
  AssertionError: ('\x00\x00\x00\x00\x00\x02\n\xe3', '\x03\xca\xad\xcb!\x92\xb6\x9c')

With previous commit, there's no point anymore to distinguish storage nodes
for which we only check serials.

Not doing so was an incorrect optimization. Checking serials does take
write-locks and they must not be released when a client-storage connection
breaks between vote and lock, otherwise a concurrent transaction modifying such
serials may finish before.

When a client-storage connection breaks, the storage node discards data of all
ongoing transactions by the client. Therefore, a reconnection within the
context of the transaction is wrong, as it could lead to partially-written
transactions.

This fixes cases where such reconnection happened. The biggest issue was that
the mechanism to dispatch disconnection events only works when waiting for an
answer.

The client can still reconnect for other purposes but the new connection won't
be reused by transactions that already involved the storage node.

Without this new mechanism to detect oids that aren't write-locked,
a transaction could be committed successfully without detecting conflicts.
In the added test, the resulting value was 2, whereas it should be 5 if there
was no node failure.

This fixes stuck replication when a client loses connection to the master
during a commit.

The consequence was that the client never reconnected to that storage node.
On commits, writes to that node always failed, causing the master to
disconnect it.

Nothing wrong actually happens.

Traceback (most recent call last):
  File "neo/scripts/neostorage.py", line 32, in main
    app.run()
  File "neo/storage/app.py", line 194, in run
    self._run()
  File "neo/storage/app.py", line 225, in _run
    self.doOperation()
  File "neo/storage/app.py", line 310, in doOperation
    poll()
  File "neo/storage/app.py", line 134, in _poll
    self.em.poll(1)
  File "neo/lib/event.py", line 168, in poll
    self._poll(0)
  File "neo/lib/event.py", line 220, in _poll
    if conn.readable():
  File "neo/lib/connection.py", line 483, in readable
    self._closure()
  File "neo/lib/connection.py", line 541, in _closure
    self.close()
  File "neo/lib/connection.py", line 533, in close
    handler.connectionClosed(self)
  File "neo/storage/handlers/client.py", line 46, in connectionClosed
    app.tm.abortFor(conn.getUUID())
  File "neo/storage/transactions.py", line 594, in abortFor
    self.abort(ttid)
  File "neo/storage/transactions.py", line 570, in abort
    self._replicated.get(self.getPartition(oid))), x
AssertionError: ('\x00\x00\x00\x00\x00\x01a\xe5', '\x03\xcaZ\x04\x14o\x8e\xbb', '\x03\xcaZ\x04\x0eX{\xbb', {1: None, 21: '\x03\xcaZ\x04\x11\xc6\x94\xf6'}, set([]))

In case of storage disconnection, one packet (VoteTransaction) was not handled
the same way as other writes, and the failure was not reported to the master
so that it arbitrates the vote. The transaction was therefore partially
committed.

Here is what happened after simulating a network failure between a client and
a storage:

C8

DEBUG   recv failed for <SSLSocketConnectorIPv6 at 0x7f8198027f90 fileno 17 ('xxxx:xxxx:120:cd8::90a1', 53970), opened to ('xxxx:xxxx:60:4c2c::25c3', 39085)>: ECONNRESET (Connection reset by peer)
DEBUG   connection closed for <MTClientConnection(uuid=S2, address=[xxxx:xxxx:60:4c2c::25c3]:39085, handler=StorageEventHandler, closed, client) at 7f81939a0950>
DEBUG   connection started for <MTClientConnection(uuid=S2, address=[xxxx:xxxx:60:4c2c::25c3]:39085, handler=StorageEventHandler, fd=17, on_close=onConnectionClosed, connecting, client) at 7f8192eb17d0>
PACKET  #0x0000 RequestIdentification          > S2 ([xxxx:xxxx:60:4c2c::25c3]:39085)        | (<EnumItem CLIENT (2)>, -536870904, None, '...', [], 1535555463.455761)
DEBUG   SSL handshake done for <SSLSocketConnectorIPv6 at 0x7f8192eb1850 fileno 17 ('xxxx:xxxx:120:cd8::90a1', 54014), opened to ('xxxx:xxxx:60:4c2c::25c3', 39085)>: ECDHE-RSA-AES256-GCM-SHA384 256
DEBUG   connection completed for <MTClientConnection(uuid=S2, address=[xxxx:xxxx:60:4c2c::25c3]:39085, handler=StorageEventHandler, fd=17, on_close=onConnectionClosed, client) at 7f8192eb17d0> (from xxxx:xxxx:120:cd8::90a1:54014)
DEBUG   <SSLSocketConnectorIPv6 at 0x7f8192eb1850 fileno 17 ('xxxx:xxxx:120:cd8::90a1', 54014), opened to ('xxxx:xxxx:60:4c2c::25c3', 39085)> closed in recv
DEBUG   connection closed for <MTClientConnection(uuid=S2, address=[xxxx:xxxx:60:4c2c::25c3]:39085, handler=StorageEventHandler, closed, client) at 7f8192eb17d0>
ERROR   Connection to <StorageNode(uuid=S2, address=[xxxx:xxxx:60:4c2c::25c3]:39085, state=RUNNING, connection=None, not identified) at 7f81a8874690> failed

S2

DEBUG   accepted a connection from xxxx:xxxx:120:cd8::90a1:54014
DEBUG   SSL handshake done for <SSLSocketConnectorIPv6 at 0x7f657144a910 fileno 22 ('xxxx:xxxx:60:4c2c::25c3', 39085), opened from ('xxxx:xxxx:120:cd8::90a1', 54014)>: ECDHE-RSA-AES256-GCM-SHA384 256
DEBUG   connection completed for <ServerConnection(uuid=None, address=[xxxx:xxxx:120:cd8::90a1]:54014, handler=IdentificationHandler, fd=22, server) at 7f657144a090> (from xxxx:xxxx:60:4c2c::25c3:39085)
PACKET  #0x0000 RequestIdentification          < None ([xxxx:xxxx:120:cd8::90a1]:54014)         | (<EnumItem CLIENT (2)>, -536870904, None, '...', [], 1535555463.455761)
DEBUG   connection closed for <ServerConnection(uuid=None, address=[xxxx:xxxx:120:cd8::90a1]:54014, handler=IdentificationHandler, closed, server) at 7f657144a090>
WARNING A connection was lost during identification
ERROR   Pre-mortem data:
ERROR   Traceback (most recent call last):
ERROR     File "neo/storage/app.py", line 194, in run
ERROR       self._run()
ERROR     File "neo/storage/app.py", line 225, in _run
ERROR       self.doOperation()
ERROR     File "neo/storage/app.py", line 310, in doOperation
ERROR       poll()
ERROR     File "neo/storage/app.py", line 134, in _poll
ERROR       self.em.poll(1)
ERROR     File "neo/lib/event.py", line 160, in poll
ERROR       to_process.process()
ERROR     File "neo/lib/connection.py", line 499, in process
ERROR       self._handlers.handle(self, self._queue.pop(0))
ERROR     File "neo/lib/connection.py", line 85, in handle
ERROR       self._handle(connection, packet)
ERROR     File "neo/lib/connection.py", line 100, in _handle
ERROR       pending[0][1].packetReceived(connection, packet)
ERROR     File "neo/lib/handler.py", line 123, in packetReceived
ERROR       self.dispatch(*args)
ERROR     File "neo/lib/handler.py", line 72, in dispatch
ERROR       method(conn, *args, **kw)
ERROR     File "neo/storage/handlers/identification.py", line 56, in requestIdentification
ERROR       assert not node.isConnected(), node
ERROR   AssertionError: <ClientNode(uuid=C8, state=RUNNING, connection=<ServerConnection(uuid=C8, address=[xxxx:xxxx:120:cd8::90a1]:53970, handler=ClientOperationHandler, fd=18, on_close=onConnectionClosed, server) at 7f657147d7d0>) at 7f65714d6cd0>

Traceback (most recent call last):
  File "neo/tests/functional/testClient.py", line 241, in testExport
    self.assertEqual(dump, self.__dump(neo_db.storage, list))
  File "neo/tests/functional/testClient.py", line 210, in __dump
    for t in storage.iterator()}
  File "neo/tests/functional/testClient.py", line 207, in <dictcomp>
    return {u64(t.tid): sorted((u64(o.oid), o.data_txn and u64(o.data_txn),
  File "neo/client/iterator.py", line 69, in iterator
    max_tid, chunk = app.transactionLog(start, stop, CHUNK_LENGTH)
  File "neo/client/app.py", line 841, in transactionLog
    Packets.AskTIDsFrom(start, stop, limit, offset))
  File "neo/client/app.py", line 296, in _askStorageForRead
    return askStorage(conn, packet)
  File "neo/client/app.py", line 164, in _askStorage
    return self._ask(conn, packet, handler=self.storage_handler, **kw)
  File "neo/lib/threaded_app.py", line 144, in _ask
    _handlePacket(qconn, qpacket, kw, handler)
  File "neo/lib/threaded_app.py", line 133, in _handlePacket
    handler.dispatch(conn, packet, kw)
  File "neo/lib/handler.py", line 72, in dispatch
    method(conn, *args, **kw)
  File "neo/lib/handler.py", line 208, in error
    getattr(self, Errors[code])(conn, message)
  File "neo/lib/handler.py", line 227, in backendNotImplemented
    raise NotImplementedError(message)
NotImplementedError: neo.storage.database.importer.ImporterDatabaseManager does not implement getReplicationTIDList

Resulting trace files can be reused to simulate a workload
and test new cache algorithms.

Besides the use of another module for option parsing, the main change is that
there's no more Config class that mixes configuration for different components.
Application classes now takes a simple 'dict' with parsed values.

The changes in 'neoctl' are somewhat ugly, because command-line options are not
defined on the command-line class, but this component is likely to disappear
in the future.

It remains possible to pass options via a configuration file. The code is a bit
complex but isolated in neo.lib.config

For SSL, the code may be simpler if we change for a single --ssl option that
takes 3 paths. Not done to not break compatibility. Hence, the hack with
an extra OptionList class in neo.lib.app

A new functional test tests the 'neomigrate' script, instead of just the
internal API to migrate data.

This mainly concerns the storage node, and depending on how its polling loop is
changed, the following crash could happen again during replication:

  File "neo/scripts/neostorage.py", line 66, in main
    app.run()
  File "neo/storage/app.py", line 147, in run
    self._run()
  File "neo/storage/app.py", line 178, in _run
    self.doOperation()
  File "neo/storage/app.py", line 258, in doOperation
    _poll(0)
  File "neo/lib/event.py", line 231, in _poll
    conn.writable()
  File "neo/lib/connection.py", line 418, in writable
    if self.connector.send():
  File "neo/lib/connector.py", line 179, in send
    n = self.socket.send(msg)
  File "ssl.py", line 719, in send
    v = self._sslobj.write(data)
OverflowError: string longer than 2147483647 byte

This commit adds a contraint when tweaking the partition table with replicas,
so that cells of each partition are assigned as far as possible from each
other, e.g. not on the same machine even if each one has several disks, and
in any case not on the same storage device.

Currently, the topology path of each node is automatically calculated by the
storage backend. Both MySQL and SQLite return a 2-tuple (host, st_dev).
To be improved:
- Add a storage option to override the path: the 'tweak' algorithm can already
  handle topology paths of any length, so something like (room, machine, disk)
  could be done easily.
- Write OS-specific code to determine the real hardware behind st_dev
  (e.g. 2 different 'st_dev' values may actually refer to the same disk,
   because of layers like partitioning, device-mapper, loop, btrfs subvolumes,
   and so on).
- Make 'neoctl' report in some way if the PT is optimal. Meanwhile,
  if it isn't, the master only logs a WARNING during tweak.

This is a follow-up of commit b3dd69730cf0e4273e1be33ee3a5ee382836b3b3
("Optimize resumption of replication by starting from a greater TID").
I missed the case where a storage node is restarted while it is replicating:
it lost the TID where it was interrupted.

Although we commit after each replicated chunk, to avoid transferring again
all the data from the beginning, it could still waste time to check that
the data are already replicated.

Otherwise, they were either lost or flushed to a file of a next test.

The bug is likely to be in the test rather than in NEO.