NOTES

Wire proto
----------

# all in network byte-order

PktHeader
	.id	u32	// autoincrement on ask (?)
	.code	u16	// defined by seqno if register(...) call in Packets
	.length	u32	// whole pkt length

# response_code = 0x8000 | code


List
	.len	u32
	[len]items

Dict
	.len	u32
	[len] key, value

Enum
	.value	s32	// None -> -1

String
	.len	u32
	[len] strdata

Address
	.host	String
	.port	u16	// present only if .host != ''

Bool
	.value	u8

Number
	.value	u32

Index
	.value	u64

PTID
	.value	u64	// 0 <-> None

Protocol
	Number		// encode <- version, decode -> check version

Checksum
	.checksum [20]u8

UUID
	.uuid	s32	// 0 <-> None

TID
	.tid	[8]byte	// None <-> \xff*8

POID = PTID

Float
	.value	float64	// '\xff' * 8 if None


~~~~~~~~

NodeList []
	type		NodeType
	address		Address
	uuid		UUID
	state		NodeState
	id_timestamp	Float

CellList []
	uuid	UUID
	state	CellState

RowList  []
	offset	Number	// u32
	cell_list CellList

HistoryList []
	serial	TID
	size	Number	// u32

UUIDList []
	uuid	UUID

TidList []
	tid	TID

OidList	[]
	oid	OID

~~~~~~~~

Notify
	.message	String

Error
	.code		Number
	.message	String

Ping
	ø
	_answer = PFEmpty

CloseClient
	ø

RequestIdentification
	.protocol_version	PProtocol
	.node_type		NodeType
	.uuid			UUID
	.address		Address
	.name			String
	.id_timestamp		Float

TODO



Storage related messages
------------------------

NotifyReady		S -> M

Recovery		PM -> S		S -> PM
LastIDs			PM -> S		S -> PM

PartitionTable		PM -> S		S -> PM
NotifyPartitionTable	PM -> S, C
PartitionChanges	PM -> S, C		// subset of NotifyPartitionTable (?)

StartOperation		PM -> S
StopOperation		PM -> S

UnfinishedTransactions	S -> PM		PM -> S
LockedTransactions	PM -> S		S -> PM

FinalTID		* -> S , C -> PM (?)
ValidateTransaction	PM -> S

NotifyTransactionFinished M -> S

LockInformation		PM -> S		S -> PM
UnlockInformation	PM -> S

StoreObject  C -> S
	.oid		OID
	.serial		TID	// original serial
	.data		String
	.data_serial	TID
	.tid		tid	// current txn id
	.unlock		bool

AnswerStoreObject  S -> C
	.conflicting	bool
	.oid		OID
	.serial		TID

GetObject  C -> S
	.oid
	.serial
	.tid

AnswerGetObject  S -> C
	.oid		OID
	.serial_start	TID
	.serial_end	TID
	.data		String
	.data_serial	TID


AbortTransaction	C -> S, PM
	.tid

StoreTransaction	C -> S		S -> C
	.tid			TID
	user, description, ext	String
	oidv			[]OID


VoteTransaction		C -> S		S -> C
	.tid


TIDList  C -> S
	first, last	Index	// u64 [first, last)
	partition	Number	// u32

AnswerTIDList  S -> C
	tidv	[]TID

TIDListFrom  C -> S
	min_tid, max_tid  TID
	length		  Number
	partition	  Number

AnswerTIDListFrom  S -> C
	tidv	[]TID

TransactionInformation  * -> S
	tid	TID

AnswerTransactionInformation  S -> *
	tid	TID
	user, description, extension	String
	packed	bool
	oidv	[]OID

ObjectHistory  C -> S
	oid	OID
	first, last	Index	// [first, last]

AnswerObjectHistory
	oid	  OID
	historyv  []strict{serial TID; size Number}

ObjectUndoSerial  C -> S
	tid		TID
	ltid		TID	// ?
	undone_tid	TID	// ?
	oidv		[]OID

AnswerObjectUndoSerial  S -> C
	{} oid ->
		.current_serial	TID
		.undo_serial	TID
		.is_current	bool

HasLock  C -> S
	tid	TID
	oid	OID

AnswerHasLock  S -> C
	oid		OID
	lock_state	LockState	// not_locket, granted, granted_to_other:w

CheckCurrentSerial  C -> S	AnswerCheckCurrentSerial  S -> C
	tid	TID			conflicting	bool
	serial	TID			oid		OID
	oid	OID			serial		TID

Pack	C -> M -> S
	tid	TID

CheckPartition	M -> S
	partition		Number
	upstream_name		String
	address			Address
	min_tid, max_tid	TID

CheckTIDRange	S -> S				AnswerCheckTIDRange	S -> S
	partition		Number			count		Number
	length			Number			checksum	Checksum
	min_tid, max_tid	TID			max_tid		TID

CheckSerialRange  S -> S

ParitionCorrupted  S -> M
	partition	Number
	uuidv		[]UUID



~~~~~~~~
(?)
NotifyNodeIntormation
NodeInformation

TODO


Storage communications
----------------------

* bootstrap (connect to primary master):

>M	RequestIdentification
<M	AcceptIdentification	(AnswerRequestIdentification)

* init (under master command go through recovery / verification)

(BaseMasterHandler)
<M	AskRecovery
>M	AnswerRecovery		(ptid, backup_tid, truncate_tid)

<M	AskPartitionTable
>M	AnswerPartitionTable	(ptid, []PtRow)

# neoctl start
<M	NotifyNodeInformation	(S1.state=RUNNING)
# S: "I was told I'm RUNNING"

<M	NotifyClusterInformation	(state=VERIFYING)
<M	NotifyNodeInformation		(S1.state=RUNNING)	# again, why?
# S: "I was told I'm RUNNING"

<M	NotifyPartitionTable	(ptid=1, `node 0: S1, R`)
# S saves PT info locally

# M asks about unfinished transactions
<M	AskLockedTransactions
>M	AnswerLockedTransactions	{} ttid -> tid	# in example we have empty

<M	AskLastIDs
>M	AnswerLastIDs			(last_oid, last_tid)

<M	NotifyClusterInformation	(state=RUNNING)

<M	StartOperation
>M	NotifyReady

--- also in base/init handlers:
<M	StopOperation
<M	AskFinalTID
# >M	answerUnfinishedTransactions (only upon S request)
# <M	ReelectPrimary			?
<M	Truncate
<M	Validatetransactions


* operation (serve clients):


Tables
------

- config
	.name	str
	.value	str
	(name, nid, partitions, ptid, replicas, version, zodb=pickle...)

- pt
	.rid	int	// = row id = part of oid space
	.nid	int
	.state	tinyint	// = cell state

	pkey (rid, nid)

# committed txns
- trans
	.partition	smallint
	.tid		bigint
	.packed		bool
	.oids		mediumblob	// []oid
	.user		blob
	.description	blob
	.ext		blob
	.ttid		bigint		// XXX ?

	pkey (partition, tid)

# committed object metadata
- obj
	.partition	smallint
	.oid		bigint
	.tid		bigint
	.data_id	bigint | NULL	// -> data.id
	.value_tid	bigint | NULL	// XXX ?	(у нас NULL)

	pkey (partition, tid, oid)
	key (partition, oid, tid)
	key data_id

# object data
- data
	.id		bigint
	.hash		sha1		// UNIQUE (hash, compression)
	.compression	tinyint
	.value		mediumblob

	key id
	key (hash, compression)	// <- from UNIQUE ^^^

- (bigdata)

# uncommitted transactions
# (= trans)
- ttrans
	.partition	smallint
	.tid		bigint
	.packed		bool
	.oids		mediumblob	// []oid
	.user		blob
	.description	blob
	.ext		blob
	ttid		bigint		// XXX ?

# uncommitted object metadata
# (= obj)
- tobj


Partition Table (general & current-py)
--------------------------------------

	#Np (how-many partitions)    #R (replication factor)
Cell
	.node   (-> .uuid, .addr)
	.state

	.backup_tid         # last tid this cell has all data for
	.replicating        # currently replicating up to this (.replicating) tid

PT
	.id↑
	.partition_list [#Np] of []Cell
	.count_dict     {} node -> #node_used_in_pt


	 Pt
	+-+
	| |
	+-+  +----------+ +------------+ +-----+
	| |  |node,state| |node2,state2| |cell3| ...
	+-+  +----------+ +------------+ +-----+
     Np	| |
	+-+
	| |
	+-+     oid -> PTentry (as PT[oid % Np]
	| |     tid
	+-+


----------------------------------------

BaseApplication
    .em     EventManager
    .nm     NodeManager

(master) Application < BaseApplication
    .tm     (master) TransactionManager
    .pt     (master) PartitionTable

    .listening_conn         ListeningConnection
    .primary                bool    # "am I primary"
    .primary_master_node    ?
    .cluster_state

    .current_manager    None | RecoveryManager | VerificationManager
    .backup_app         BackupApplication

    .backup_tid
    .truncate_tid

    .admin_handler              AdministrationHandler
    .secondary_master_handler   SecondaryMasterHandler
    .client_service_handler     ClientServiceHandler
    .storage_service_handler    StorageServiceHandler

(storage) Application < BaseApplication
    .tm     (storage) TransactionManager
    .dm     DatabaseManager     (Importer | MySQL | SQLite)

    .pt     ? PartitionTable

    .checker        Checker         # checks sha1([]tid  -- XXX only tid list, not other metadata & data)
    .replicator     Replicator

    .listening_conn ListeningConnection
    .master_conn
    .master_node

    # (queued events are executed at some txn boundaries & locks)
    .event_queue        deque   (key, callable, msg_id, conn, args)
    .event_queue_dict   {}      key -> count(.event_queue, key)

    .task_queue         deque

    .operational        bool

    .ready              bool    # when .operational=T and "got all informations"


Handlers
--------

AdministrationHandler < MasterHandler
    .connectionLost(...)
    .setClusterSate(...)
    .setNodeState(...)
    .addPendingNodes(...)
    .tweakPartitionTable(...)
    .truncate(conn, tid)
    .checkReplicas(...)

SecondaryMasterHandler < MasterHandler
    .connectionLost(...)
    .announcePrimary(...)
    .reelectPrimary(...)

ClientServiceHandler < MasterHandler
    .connectionLost(...)
    .askNodeInformation(conn)           # send informations about master and storages only
    .askBeginTransaction(conn, tid)
    .askNewOIDs(conn, num_oids)
    .askFinalTID(conn, ttid)
    .askPack(conn, tid)
    .abortTransaction(conn, tid)

StorageServiceHandler < BaseServiceHandler
    .connectionCompleted(conn, new)
    .connectionLost(conn, new_state)
    .askUnfinishedTransactions(conn)
    .answerInformationLocked(conn, ttid)
    .notifyPartitionCorrupted(conn, partition, cell_list)
    .notifyReplicationDone(conn, offset, tid)
    .answerPack(conn, status)


MasterHandler < EventHandler
    .connectionCompleted(...)
    .requestIdentification(...)
    .askClusterState(...)
    .askRecovery(...)
    .askLastIDs(...)
    .askLastTransaction(...)
    .askNodeInformation(...)
    .askPartitionTable(...)

BaseServiceHandler < MasterHandler
    .connectionCompleted(...)
    .connectionLost(...)
    .notifyReady(conn)

EventHandler
    .packetReceived(conn, pkt, **kw)
        -> .dispatch(conn, pkt, **kw)
            -> conn.setPeerId(pkt.getId())
            -> getattr(self, pkt.handler_method_name) (conn, *pkt.decode(), **kw)

           except ...:
               reply error



IdentificationHandler < MasterHandler

Connections
-----------

ListeningConnection < BaseConnection

    .__init__(self, app, handler, addr):
        connector = self.ConnectorClass(addr)
        BaseConnection.__init__(self, app, connector, addr)
        connector.makeListeningConnection()                     # ~= self.socket.bind(addr) + listen()
        self.em.register(self)

    .readable(self):
        connector, addr = self.connector.accept()
        conn = ServerConnection(self.em, self._handlers.getHandler(), connector, addr)
        conn._connected()


BaseConnection
    .ConnectorClass = SocketConnector   # by default

    .em             EventManager
    .connector      self.ConnectorClass()   # by default ?
    .addr
    ._handlers      HandlerSwitcher


HandlerSwitcher
    # pending handlers and related requests
    ._pending       [] of [{}(msg_id -> answer_class, timeout, on_timeout, kw), handler]

    .emit(request, timeout, on_timeout, kw)
    .handle(connection, pkt)
        -> ._handle(connection, pkt)

    ._handle(connection, pkt):
        msg_id = pkt.getId()
        request_dict, handler = ._pending[0]
        if !pkt.isResponse():
            handler.packetReceived(connection, pkt)

        klass, _, _, kw = request_pop(msg_id)
        if klass && isinstance(pkt, klass) || pkt.isError():
            handler.packetReceived(connection, packet, kw)
        else:
            # unexpected answer

        # skip all intermediate handlers and apply to "latest" (fff6cf45)
        while len(._pending) > 1 && !._pending[0].handler:
            del ._pending[0]


ServerConnection <Connection
    KEEP_ALICE = Connection.KEEP_ALIVE + 5
    server = True

# main worker class for handling data exchanging
Connection <BaseConnection
    .read_buf       ReadBuffer  ([][]byte  = iovec)
    ._queue         []
    ._parser_state  ParserState

    .writable()
        -> .connector.send()

    .readable()
        -> .connector.receive(.read_buf)
           while .read_buf:
               pkt = Packets.parse(.read_buf, ._parser_state)
               ._queue.append(pkt)
           return bool(._queue)

    # called from under EventManager.poll()     # incoming messages
    .process()
        -> ._handlers.handle(self, ._queue.pop(0))

    ._addPacket(pkt):
        .connector.queue(pkt.encode())
        .em.addWriter(self)

    .notify(pkt):
        pkt.setId(next-msg-id)
        ._addPacket(pkt)

    .ask(pkt, timeout=30, on_timeout=None, **kw):
        .notify(pkt)
        ._handlers.emit(pkt, timeout, on_timeout, kw)

    .answer(pkt, msg_id=None):
        pkt.setId(msgid ?: self.getPeerId())
        ._addPacket(pkt)


------------

EventManager
    .connection_dict        {}  fd -> Connection
    .reader_set             set
    .writer_set             set
    .epoll
    ._pending_processing    []Connection    # fetched from .epoll

    .register(conn)
    .unregister(conn, close=False)

    .poll(blocking=1):
        ._poll(blocking)
        to_process = ._pending_processing.pop(0)
        to_process.process()

    ._poll(blocking):
        event_list = .epoll.poll(blocking)
        for fd, event in event_list:
            conn = .connection_dict[fd]
            if event & EPOLLIN:
                if conn.readable():
                    ._addPendingConnection(conn)    # queues to ._pending_processing
            # EPOLLOUT -> wlist -> conn.writable()
            # EPOLLERR -> elist -> conn.readable() -> ._addPendingConnection(conn)

    .addReader(conn)
    .removeReader(conn)
    .addWriter(conn)
    .removeWriter(conn)


--------

# M waits for S* to connect so PT is filled
RecoveryManager < MasterHandler
    ...

# Finish all transactions for which we know that tpc_finish was called but not fully processed
#
# ( abort(?) / commit half-way finished transactions  (commit if we know tpc_finish was started)
#   XXX no actual abort happens since .verifyData():
#     Transaction not locked. No need to tell nodes to delete it,
#     since they drop any unfinished data just before being
#     operational. )
VerificationManager < BaseServiceHandler
    ...