Milestones:

k: The target replication count and MUST be odd.

Failover chain: The ordered collection of physical service instances
for the same logical service.

Quorum: The set of (k+1)/2 physical service instances for the same
logical service with an agreement on the lastCommitTime.

Configuration:

   - port on which to talk to the peers.

   - protocol version?

   - ???

BufferDescriptor:

   - int:bufferId
 
   - long:chksum

   - ByteBuffer (will be direct; may be a slice).

1. Write replication (write(oldadd,newaddr,chksum,bd), extend(length),
   prepare(commitCounter), commit(rootBlock), abort(commitCounter)).

   - The integration point for write() is the WriteCacheService's newWriteCache()
     method.  This returns a WriteCache whose writeOnChannel() method must be
     overridden to write on the next node in the failover chain which is also
     a member of the quorum.  The pipeline write can be issued first, obtaining
     a Future.  Then issue the write on the local disk, which will block on the
     IO.  Then wait on the Future.  This will minimize the latency.

   - The integration point for extend() is the RWStrategy or WORMStrategy.
   
   - The integration point for commitNow() / abort() is the AbstractJournal. 
     These translate down into other messages at the HA glue layer, include
     prepare2Phase(commitCounter,bd), commit2Phase(commitCounter), and
     abort2Phase(commitCount).  The master is the only one who would validate
     a high level transaction.  By the time the master calls commitNow(long)
     it has validated the transaction and has a specific commitTime and can
     form the exact root block image.
     
        - prepare2Phase will read the new root block from the master, log the
          old root block, and optionally flush to the disk. 
        
     	- commit2Phase will write the new root block and flush to the disk.
     	
     	- abort2Phase will reload from the current root block.

   - Pattern is send a FutureTask and wait on the Future.

   - Write pipeline with nio using direct buffers.

   - 2/3-phase commit with quorum voting "yes".  otherwise you are
     under quorum on this commit and we need to send an abort message
     to the members of the failover chain for that commit (this
     suggests that we need to send the commitCounter and/or commit
     time with the prepare and the abort; the commit message will have
     the root block).
     
   - The write pipeline should be reconfigured each time a new quorum is
     established.  The configuration should include specific end points
     (addr,port) for each node.  Each node makes a request from its 
     predecessor in the pipeline.  Data streams through from node to node
     as it arrives, sending deltas down the socket in efficient chunk sizes
     for the network protocol but not blocking to wait for the entire record
     to arrive at a node before streaming to the downstream requester.
     
     The ResourceService bears some similarity to the write pipeline, but is
     focused on replication of entire files.  Also, while the write pipeline
     always flows from the master along the quorum order, it is expected that
     index builds will occur on the node with affinity for the shard and then
     replicated to the failover nodes.  Index segment files are significantly
     larger than raw records, so the overhead to establish a write order for
     each file shipped is acceptable in this case.  However, this is not an
     acceptable overhead for the write pipeline.
     
     We need to explore resynchronization in more depth.  This differs from
     the write pipeline because any node in the quorum can be read on.  In
     this way it is more like failover reads. 

2. Handle bad reads (read(addr,bd))

    - Read on a peer in the quorum and alerting interface since node
     may be flakey.

3. Quorum membership (A quorum protects against choosing a new master
   with stale state following a federation restart after failure of
   one or more nodes).

   - Need a Quorum object. Who is in the quorum and who is the master.
     The failover chain is every node which is live according to zk.
     The quorum is only the nodes that are online for client requests
     and is formed from an agreement among (k+1)/2 nodes over state.

   - The write pipeline must skip over nodes which are not in the
     quorum.

   - Once a quorum membership is established, if the master is not
     part of the quorum, then it closes and then reopens its zookeeper
     connection in order to relinquish its role as the master.

   - Once a quorum membership is established, anyone who is not in the
     quorum must resynchronize before they can join the quorum.
     
   - Identification of a quorum will be coordinated using zookeeper.  Any
     node which is not a member of a quorum will create an ephemeral znode
     under the following zpath:
     
     	.../logicalService/quorums/[commitTime]/[epheremalChild]
     	
     where "commitTime" is the commitTime of its current root block.  A quorum
     exists (has "met") when there are (k+1)/2 children for a specific commitTime.
     Any node which is not part of a "met" quorum must undergo resynchronization
     to bring it into the quorum.  This can be either leading resynchronization
     (discarding committed state) or trailing resynchronization (acquiring state
     committed in the quorum).  
     
     The failover chain is simply the ordered set of (k+1)/2 ephemeral znodes
     under a given commitTime znode, e.g., the current met quorum.
     
     The quorum is only renegotiated if the #of members in the quorum falls 
     below (k+1)/2.  Until then, the ephemeral znodes remain associated with
     the commitTime under which the quorum was formed even though subsequent
     commits may have flowed through the system.  Once a quorum is met, other
     nodes must synchronize with the quorums current state and then join the
     quorum under its commitTime.
     
     Note that the commitTime is simply a token.  Its semantics are purely those
     of the commitTime for which the quorum was established.  Once the quorum is
     active, new commits will flow through the quorum.  However the commitTime
     token for the quorum DOES NOT update even though new commits have been
     processed.  Only if the quorum breaks (falls below (k+1)/2 members) do
     nodes publish themselves under their then current commitTime, thus starting
     the search for a new quorum.
     
     In order to manage quorums, each node must watch the znode hierarchy under
     .../logicalService/quorums.  There is existing code for this sort of thing.
     
     In addition to the above zpath, each logical service also maintains a list
     of all physical instances.  The ServicesManagerService (SMS) is responsible
     for creating and destroying service instances as necessary to maintain a
     given failover capacity (k).  It uses persistent child znodes to represent
     the known physical service instances for a given logical service.  It
     currently uses ephemeral znodes to manage a master election, but I think
     that this needs to be changed to be quorum based as described above.
     
   - ? Master specific caches must be discarded if the quorum breaks or if the
     master fails over to another node.  This is because the failover nodes
     are not always maintaining these caches.
     
4. Resynchronization.  There are two cases: leading and trailing.  We
   can resynchronize from anyone in the quorum (and could do scattered
   reads on the quorum).  

   (*) The RW and WORM each require some atomic decision making about
   what delta is required and each must not update their root blocks
   or invalidate existing state until they are current with the quorum
   to avoid creating a database state which is incoherent (one that
   can not be reopened from its root blocks).

   The node will begin to start accepting writes from the pipeline as
   soon as it has computed its deltas, but will not join the quorum
   until it is current.  "Current" means all deltas are cleared (this
   is defined for the WORM as a file byte delta and for the RW as an
   allocation bit map delta). We need to do something interesting,
   such as continuous approximation, since the delta computation is
   not atomic across the failover set.
   
   ? When do we update the root blocks during the resynchronization protocol?
     By the time the node is synchronized, both of its root blocks should be
     current.  However, we do not want to write both root blocks at once. Each
     root block should involve its own commit point so we always have one good
     root block, even in the event of a failure during the root block update.

   - WORM
     
     Leading.  Just get the root blocks and the file length. If the
     file is too long, then truncate it.  Update the root blocks and
     you are done.  The file extent comes with each write cache block,
     so we can even get this from the write replication protocol.
     Provide a failsafe check against arbitrary truncation of the
     journal.

     Trailing.  Send the file delta.  This should be ultra direct.
     Just transmit the delta in the file from the channel to the
     socket.  Update the root blocks once the file delta is up to
     date.

   - RW.  The RW store requires allocation block checkpoints (aka
     "session").  For example, maintaining 24 hours of session.

     Wrong session.  If you are past the session checkpoint then you
     can not do incremental synchronization and you must request the
     entire store.  This is true whether you are leading or
     trailing. (We could write out session checkpoints onto the store
     itself to allow deltas which cross a restart or other session
     boundary and define an index over the session boundaries for fast
     access to those records.)

     Otherwise, first obtain a lock which prevents the master from
     expiring the session during resynchronization.

     For the RW store the one non-atomic change is in saving the
     allocation blocks, especially if we have resized the file.  From
     transaction to transaction this is not a problem since the
     committed data is never re-used in the next transaction.  The
     final update should always be to the rootblock, but at the point
     the allocation blocks are written this could be out of sync.  The
     solution is to write the blocks in a two-phase update.  First,
     temporarily extend the file to store the new allocation blocks,
     then write the root block to reference these.  This secures the
     data, now we can write the allocation blocks to the correct area,
     update the rootblock and lastly set the correct file extent.

          [See below, but also note that I would like to avoid
	   updating the root blocks until the quorum join commit so as
	   to not change the lastCommitTime or otherwise spoil the
	   state of the store.  Martyn seems to have mapped out a
	   solution below.  Another way to approach this might be to
	   "patch" the root blocks with the correct offset and meta
	   bits addr.  Another is to explicitly stage the RW store
	   through each extension using a set of extension records
	   written onto the store.]

     trailing. With a current session, we know that no current data
     has been overwritten.
     
       1) If the file needs extention then we do that first, and
          update the rootblock.  This creates the space for the
          updates.

       2) Compute delta using current in-memory allocation blocks and
          new in-memory allocation blocks.

       3) Request delta state and add writes.

       4) Temporarily extend file to write new allocation blocks and
          update rootblock. (committing new state).

       5) Overwrite 'old' allocation blocks with new blocks and update
          rootblock.

       6) Set correct file extent.

     leading.  In the same session we will not have overwritten data,
     so we need to address any file extention and synchronize the
     allocation blocks and metaallocation bits.
     
     Since the metallocation bits are referenced from the main heap,
     updating the rootblock is all that is necessary to reference the
     correct state.
     
     We will use the two-phase update to sync the allocation blocks
     
     1) extend the file and write the updated blocks to the extended
        file

     2) update the root block to point to extended bits (and flush)

     3) copy allocation blocks to correct area

     4) final update of rootblock

     5) set correct file extent

5. Robust messaging and liveness.

- If you are not part of the quorum, then what do you with a read
  request?  With a write request?  How does the client encapsulate
  handling redirects?  We could send back the quorum metadata, which
  says which nodes are in the quorum, whether or not a quorum exists,
  and the master is only the master if the quorum exists.

  ? How do we tell the client when the quorum changes such that it
    could have use another node for its read request?

  ? Maybe we should forward read requests to a peer in the quorum
    since all the client is aware of is which nodes are in the
    failover chain?

    Write requests to a node which is in the master position but is
    not in the quorum should block until a quorum is established,
    which may require the node to yield the master role. If the node
    yields master role to another node, then it will forward the
    request to the new master once that node comes online as the
    master.

    Requests can still be dropped if a node has the request and it
    dies.  How does the client ever notice this?  By a timeout on the
    FutureTask?  How does the client deal with retries for reads and
    writes?  It can cancel the FutureTask and submit a new one to
    another node.  The buffer descriptor must be associated with a
    message identifier to prevent read/write on a buffer whose
    contents have been changed.

  This is related to the concept of shard affinity, which is also not
  directly visible to clients.  There is also the question of how
  clients are able to make any requests at all... e.g.,
  sail#runQuery(...) or whatever.  Load balancing of requests for a
  journal needs to be handled at the application level.

- Do we send messages to everyone in the failover chain and only have those
  who are in the quorum respond or do we only send messages to everyone in
  the quorum?  On the one hand a quorum state transition could cause a message
  to be missed.  On the other hand, sending messages to a node which is dead
  to RMI but still zookeeper could cause significant latency while the RMI
  message times out.

- There is a broad question about "liveness" and heart beats which is related
  to these issues.

Open questions:

- How to make the resynchronization process one which guarantees eventual
  consistency without allowing any intermediate bad states?   This is partly
  a question of distributed atomic decision making and partly a question for
  the RW store, which has more complex state transitions than the WORM.

- Can we do 2 phase commits for a quorum w/ the existing AbstractTask
  and WriteExecutorService lock (this might only show up as a problem
  for the DS since we don't tend to use this for the journal).

- Under quorum recovery: 

   - Quorum could be defined by the minimum value of the
     lastCommitTime across (k+1)/2 peers?  This is an interesting
     notion which is made possible by the temporal database structure.
     This could get around the 2-phase commit requirement.  We still
     want the (k+1)/2 bit because that protects us against choosing a
     quorum with older data when we could have waited and gotten a
     quorum with newer data.  E.g., treat the "minimum" as a fallback
     posture if we can not achieve a quorum otherwise.  This could
     happen if a 2-phase commit fails on the "commit" message.  Note
     that a rollback at this level should probably be applied to all
     indices having data on the shards which have gone under the
     quorum.

   - Another option is to choose the most recent lastCommitTime and
     replicate the data to another node. As long as the root blocks
     are good and all data can be read this should be a safe
     under-quorum recovery option.

   - Should we allow the system to operate under quorum?  You can read
     from historical data up to the time when you lost the quorum.
     However, if you write on a logical node which is under quorum
     then your writes are at risk if your node goes down and an older
     node comes up.

- Can we kill a client's zk connection if it is not reachable or are
  we going to wait until it times out?

============================================================

Quorum next steps:

   I am closing out the zookeeper quorum integration.  It appears to
   work fine and in exact the same manner for the mock quorum fixture
   and the zookeeper integration with the quorum logic.

   I want to revisit some aspects of the quorum's internal behavior in
   the context of addressing the resynchronization protocol for the
   services in the quorum.  The basic issue is that there is some
   concurrent indeterminism in the ordering of events as observed by
   the local quorum objects for each service.  I think that the right
   way to handle this may be to split the quorum into a model class
   with atomic set/get methods for its state, whose implementations
   could be trivially unit tested, and a behavior class which is
   responsible for bringing the quorum state for a service to the
   target "run state" for that service.

   I am thinking of target run states as {MEMBER, SYNCHRONIZE, JOIN}.
   MEMBER would correspond to a service which was online but not
   synchronized.  SYNCHRONIZE would correspond to a service which was
   attempting to join a quorum and was making up for the delta in its
   persistent state so it could join.  JOIN would be a service which
   was synchronized.

   Since the details of any such run levels really interact with the
   resynchronization protocol and other issues such as hot spare
   recruitment, I'd like to put them off until we take up the
   resynchronization logic.


TODO:

     - RW is always using 6 buffers.  This must be a configuration
       option so we can stress test the WriteCacheService under heavy
       write loads and mixed write/read loads with lots of concurrency
       and only a few buffers.  We need to do this to look for
       deadlocks.

     - Handle read errors by reading on a peer.  Note that some file
       systems will retry MANY times, which could hang the caller
       (timed reads?  how?).  Consider doing replacement writes over
       the bad region (again, we have a problem with timeouts.  I am
       not even sure if Java IO is really interruptable once it gets
       down to the OS layer).  This might be a problem for Windows and
       HA.

     - Report read errors in support of decision making about failure.
