[Server] add fixes to cluster on closing a connection

Author: unit-adm

README.md

@@ -36,7 +36,7 @@ Unitdb supports Get, Put, Delete operations. It also supports encryption, batch
 Samples are available in the examples directory for reference.
 
 ## Clustering
-To bring up the Unitdb cluster start 2 or more nodes, 3 nodes or more nodes are recommended for fault tolerance.
+To bring up the Unitdb cluster start 2 or more nodes. For fault tolerance 3 nodes or more are recommended.
 
 ```
 > ./bin/unitdb -listen=:6060 -grpc_listen=:6080 -cluster_self=one -db_path=/tmp/unitdb/node1

server/.gitignore

@@ -83,6 +83,8 @@ type ClusterSess struct {
 	// IP address of the client. For long polling this is the IP of the last poll
 	RemoteAddr string
 
+	// protocol - NONE (unset), RPC, GRPC, GRPC_WEB, WEBSOCK
+	Proto lp.Proto
 	// Connection ID
 	ConnID uid.LID
 
@@ -305,11 +307,13 @@ func (c *Cluster) Master(msg *ClusterReq, rejected *bool) error {
 			}
 
 			log.Info("cluster.Master", "new connection request"+string(msg.Conn.ConnID))
-			conn = Globals.Service.newRpcConn(node, msg.Conn.ConnID, msg.Conn.ClientID)
+			conn = Globals.Service.newRpcConn(node, msg.Conn.Proto, msg.Conn.ConnID, msg.Conn.ClientID)
 			go conn.rpcWriteLoop()
 		}
 		// Update session params which may have changed since the last call.
+		conn.proto = msg.Conn.Proto
 		conn.connid = msg.Conn.ConnID
+		conn.clientid = msg.Conn.ClientID
 
 		switch msg.Type {
 		case message.SUBSCRIBE:
@@ -410,6 +414,7 @@ func (c *Cluster) routeToContract(msg lp.Packet, topic *security.Topic, msgType
 			Message:   m,
 			Conn: &ClusterSess{
 				//RemoteAddr: conn.(),
+				Proto:    conn.proto,
 				ConnID:   conn.connid,
 				ClientID: conn.clientid}})
 }
@@ -525,7 +530,10 @@ func (c *_Conn) rpcWriteLoop() {
 				// channel closed
 				return
 			}
-			m, err := lp.Encode(c.proto, msg)
+			if c.adp == nil {
+				return
+			}
+			m, err := lp.Encode(c.adp, msg)
 			if err != nil {
 				log.Error("conn.writeRpc", err.Error())
 				return

server/internal/cluster.go

@@ -39,8 +39,9 @@ import (
 type _Conn struct {
 	sync.Mutex
 	tracked uint32 // Whether the connection was already tracked or not.
-	// protocol - NONE (unset), RPC, GRPC, WEBSOCK, CLUSTER
-	proto  lp.ProtoAdapter
+	// protocol - NONE (unset), RPC, GRPC, GRPC_WEB, WEBSOCK, CLUSTER
+	proto  lp.Proto
+	adp    lp.ProtoAdapter
 	socket net.Conn
 	// send     chan []byte
 	send               chan lp.Packet
@@ -65,16 +66,16 @@ type _Conn struct {
 }
 
 func (s *_Service) newConn(t net.Conn, proto lp.Proto) *_Conn {
-	var lineProto lp.ProtoAdapter
+	var adp lp.ProtoAdapter
 	switch proto {
 	case lp.GRPC:
-		lineProto = &grpc.LineProto{}
+		adp = &grpc.LineProto{}
 	case lp.GRPC_WEB:
-		lineProto = &grpc.LineProto{}
+		adp = &grpc.LineProto{}
 	}
 
 	c := &_Conn{
-		proto:      lineProto,
+		adp:        adp,
 		socket:     t,
 		MessageIds: message.NewMessageIds(),
 		send:       make(chan lp.Packet, 1), // buffered
@@ -96,8 +97,16 @@ func (s *_Service) newConn(t net.Conn, proto lp.Proto) *_Conn {
 }
 
 // newRpcConn a new connection in cluster
-func (s *_Service) newRpcConn(conn interface{}, connid uid.LID, clientid uid.ID) *_Conn {
+func (s *_Service) newRpcConn(conn interface{}, proto lp.Proto, connid uid.LID, clientid uid.ID) *_Conn {
+	var adp lp.ProtoAdapter
+	switch proto {
+	case lp.GRPC:
+		adp = &grpc.LineProto{}
+	case lp.GRPC_WEB:
+		adp = &grpc.LineProto{}
+	}
 	c := &_Conn{
+		adp:        adp,
 		connid:     connid,
 		clientid:   clientid,
 		MessageIds: message.NewMessageIds(),
@@ -307,7 +316,7 @@ func (c *_Conn) storeInbound(m lp.Packet) {
 		blockId := uint64(c.clientid.Contract())
 		k := uint64(c.inboundID(m.Info().MessageID))<<32 + blockId
 		fmt.Println("inbound: type, key, qos", m.Type(), k, m.Info().Qos)
-		store.Log.PersistInbound(c.proto, k, m)
+		store.Log.PersistInbound(c.adp, k, m)
 	}
 }
 
@@ -316,7 +325,7 @@ func (c *_Conn) storeOutbound(m lp.Packet) {
 		blockId := uint64(c.clientid.Contract())
 		k := uint64(c.inboundID(m.Info().MessageID))<<32 + blockId
 		fmt.Println("inbound: type, key, qos", m.Type(), k, m.Info().Qos)
-		store.Log.PersistOutbound(c.proto, k, m)
+		store.Log.PersistOutbound(c.adp, k, m)
 	}
 }
 

server/internal/conn.go

@@ -52,7 +52,7 @@ func (c *_Conn) readLoop() error {
 		c.socket.SetDeadline(time.Now().Add(time.Second * 120))
 
 		// Decode an incoming packet
-		pkt, err := lp.ReadPacket(c.proto, reader)
+		pkt, err := lp.ReadPacket(c.adp, reader)
 		if err != nil {
 			fmt.Println("readPacket: err", err)
 			return err
@@ -205,7 +205,7 @@ func (c *_Conn) writeLoop(ctx context.Context) {
 				// Channel closed.
 				return
 			}
-			m, err := lp.Encode(c.proto, msg)
+			m, err := lp.Encode(c.adp, msg)
 			if err != nil {
 				log.Error("conn.writeLoop", err.Error())
 				return
@@ -216,7 +216,7 @@ func (c *_Conn) writeLoop(ctx context.Context) {
 				// Channel closed.
 				return
 			}
-			m, err := lp.Encode(c.proto, msg)
+			m, err := lp.Encode(c.adp, msg)
 			if err != nil {
 				log.Error("conn.writeLoop", err.Error())
 				return
@@ -396,7 +396,7 @@ func (c *_Conn) resume() {
 	// contract is used as blockId and key prefix
 	keys := store.Log.Keys()
 	for _, k := range keys {
-		msg := store.Log.Get(c.proto, k)
+		msg := store.Log.Get(c.adp, k)
 		if msg == nil {
 			continue
 		}

server/internal/hdl_conn.go

@@ -42,7 +42,7 @@ func main() {
 	var listenOn = flag.String("listen", "", "Override address and port to listen on for HTTP(S) clients.")
 	var listenGrpcOn = flag.String("grpc_listen", "", "Override address and port to listen on for GRPC clients.")
 	var clusterSelf = flag.String("cluster_self", "", "Override the name of the current cluster node")
-	var dbPath = flag.String("db_path", "tmp/unitdb", "Override the db path.")
+	var dbPath = flag.String("db_path", "/tmp/unitdb", "Override the db path.")
 	var varzPath = flag.String("varz", "/varz", "Expose runtime stats at the given endpoint, e.g. /varz. Disabled if not set")
 	flag.Parse()
 
@@ -70,7 +70,7 @@ func main() {
 	}
 
 	// Set up gRPC server, if one is configured
-	if *listenGrpcOn == "" {
+	if *listenGrpcOn != "" {
 		cfg.GrpcListen = *listenGrpcOn
 	}
 

server/main.go

@@ -41,7 +41,7 @@
 	"cluster_config": {
 		// Name of this node. Can be assigned from the command line.
 		// Empty string disables clustering.
-		"self": "",
+		"self": "one",
 
 		// List of available nodes.
 		"nodes": [