fix(connection): will retry connection for each message

The channel will try to connect to the node for each message if a
connection has not yet been established or if the node has disconnected.

Author: aleksander-vedvik

channel.go

@@ -2,6 +2,7 @@ package gorums
 
 import (
 	"context"
+	"fmt"
 	"math"
 	"math/rand"
 	"sync"
@@ -35,7 +36,7 @@ type responseRouter struct {
 
 type channel struct {
 	sendQ           chan request
-	nodeID          uint32
+	node            *RawNode
 	mu              sync.Mutex
 	lastError       error
 	latency         time.Duration
@@ -50,29 +51,39 @@ type channel struct {
 	cancelStream    context.CancelFunc
 	responseRouters map[uint64]responseRouter
 	responseMut     sync.Mutex
+	connEstablished bool
 }
 
 func newChannel(n *RawNode) *channel {
 	return &channel{
 		sendQ:           make(chan request, n.mgr.opts.sendBuffer),
 		backoffCfg:      n.mgr.opts.backoff,
-		nodeID:          n.ID(),
+		node:            n,
 		latency:         -1 * time.Second,
 		rand:            rand.New(rand.NewSource(time.Now().UnixNano())),
 		responseRouters: make(map[uint64]responseRouter),
+		connEstablished: false,
 	}
 }
 
 func (c *channel) connect(ctx context.Context, conn *grpc.ClientConn) error {
-	var err error
 	c.parentCtx = ctx
+	go c.sendMsgs()
+	if conn == nil {
+		return fmt.Errorf("connection is nil")
+	}
+	return c.tryConnect(conn)
+}
+
+func (c *channel) tryConnect(conn *grpc.ClientConn) error {
+	var err error
 	c.streamCtx, c.cancelStream = context.WithCancel(c.parentCtx)
 	c.gorumsClient = ordering.NewGorumsClient(conn)
 	c.gorumsStream, err = c.gorumsClient.NodeStream(c.streamCtx)
 	if err != nil {
 		return err
 	}
-	go c.sendMsgs()
+	c.connEstablished = true
 	go c.recvMsgs()
 	return nil
 }
@@ -160,17 +171,23 @@ func (c *channel) sendMsgs() {
 			return
 		case req = <-c.sendQ:
 		}
+		// try to connect to the node if previous attempts
+		// have failed or if the node has disconnected
+		if !c.isConnected() {
+			// streamBroken will be set if the connection fails
+			c.tryReconnect()
+		}
 		// return error if stream is broken
 		if c.streamBroken.get() {
 			err := status.Errorf(codes.Unavailable, "stream is down")
-			c.routeResponse(req.msg.Metadata.MessageID, response{nid: c.nodeID, msg: nil, err: err})
+			c.routeResponse(req.msg.Metadata.MessageID, response{nid: c.node.ID(), msg: nil, err: err})
 			continue
 		}
 		// else try to send message
 		err := c.sendMsg(req)
 		if err != nil {
 			// return the error
-			c.routeResponse(req.msg.Metadata.MessageID, response{nid: c.nodeID, msg: nil, err: err})
+			c.routeResponse(req.msg.Metadata.MessageID, response{nid: c.node.ID(), msg: nil, err: err})
 		}
 	}
 }
@@ -189,7 +206,7 @@ func (c *channel) recvMsgs() {
 		} else {
 			c.streamMut.RUnlock()
 			err := status.FromProto(resp.Metadata.GetStatus()).Err()
-			c.routeResponse(resp.Metadata.MessageID, response{nid: c.nodeID, msg: resp.Message, err: err})
+			c.routeResponse(resp.Metadata.MessageID, response{nid: c.node.ID(), msg: resp.Message, err: err})
 		}
 
 		select {
@@ -200,11 +217,37 @@ func (c *channel) recvMsgs() {
 	}
 }
 
-func (c *channel) reconnect() {
+func (c *channel) tryReconnect() {
+	// a connection has never been established
+	if !c.connEstablished {
+		err := c.node.dial()
+		if err != nil {
+			c.streamBroken.set()
+			return
+		}
+		err = c.tryConnect(c.node.conn)
+		if err != nil {
+			c.streamBroken.set()
+			return
+		}
+	}
+	// the node has previously been connected
+	// but is now disconnected
+	if c.streamBroken.get() {
+		// try to reconnect only once
+		c.reconnect(1)
+	}
+}
+
+func (c *channel) reconnect(maxRetries ...int) {
 	c.streamMut.Lock()
 	defer c.streamMut.Unlock()
 	backoffCfg := c.backoffCfg
 
+	var maxretries float64 = -1
+	if len(maxRetries) > 0 {
+		maxretries = float64(maxRetries[0])
+	}
 	var retries float64
 	for {
 		var err error
@@ -217,6 +260,10 @@ func (c *channel) reconnect() {
 		}
 		c.cancelStream()
 		c.setLastErr(err)
+		if retries >= maxretries && maxretries > 0 {
+			c.streamBroken.set()
+			return
+		}
 		delay := float64(backoffCfg.BaseDelay)
 		max := float64(backoffCfg.MaxDelay)
 		for r := retries; delay < max && r > 0; r-- {
@@ -257,6 +304,10 @@ type atomicFlag struct {
 	flag int32
 }
 
+func (c *channel) isConnected() bool {
+	return c.connEstablished && !c.streamBroken.get()
+}
+
 func (f *atomicFlag) set()      { atomic.StoreInt32(&f.flag, 1) }
 func (f *atomicFlag) get() bool { return atomic.LoadInt32(&f.flag) == 1 }
 func (f *atomicFlag) clear()    { atomic.StoreInt32(&f.flag, 0) }

gorums.pb.go

@@ -119,7 +119,9 @@ func (m *RawManager) AddNode(node *RawNode) error {
 		m.logger.Printf("connecting to %s with id %d\n", node, node.id)
 	}
 	if err := node.connect(m); err != nil {
-		return fmt.Errorf("connection failed for %s: %w", node, err)
+		if m.logger != nil {
+			m.logger.Println(fmt.Errorf("connection failed for %s: %w. will retry later.", node, err))
+		}
 	}
 
 	m.mu.Lock()

mgr.go

@@ -61,28 +61,48 @@ func NewRawNodeWithID(addr string, id uint32) (*RawNode, error) {
 // connect to this node and associate it with the manager.
 func (n *RawNode) connect(mgr *RawManager) error {
 	n.mgr = mgr
+	n.channel = newChannel(n)
 	if n.mgr.opts.noConnect {
 		return nil
 	}
-	n.channel = newChannel(n)
-	var err error
-	ctx, cancel := context.WithTimeout(context.Background(), n.mgr.opts.nodeDialTimeout)
-	defer cancel()
-	n.conn, err = grpc.DialContext(ctx, n.addr, n.mgr.opts.grpcDialOpts...)
-	if err != nil {
-		return fmt.Errorf("dialing node failed: %w", err)
+	// ignoring the error because it will try to reconnect
+	// at a later time.
+	_ = n.dial()
+	ctx := n.ctxSetup()
+	if err := n.channel.connect(ctx, n.conn); err != nil {
+		return fmt.Errorf("starting stream failed: %w", err)
+	}
+	return nil
+}
+
+// dials the node if it has not been done previously
+func (n *RawNode) dial() error {
+	if n.conn == nil {
+		var err error
+		ctx, cancel := context.WithTimeout(context.Background(), n.mgr.opts.nodeDialTimeout)
+		defer cancel()
+		// error is ignored because we will retry the dial at a later time
+		n.conn, err = grpc.DialContext(ctx, n.addr, n.mgr.opts.grpcDialOpts...)
+		return err
 	}
+	return nil
+}
+
+// creates a context that governs the channel. It is
+// used to stop all channel goroutines and the NodeStream.
+//
+// this method should be run for each connection to ensure
+// fresh contexts. Reusing contexts could result in reusing
+// a cancelled context.
+func (n *RawNode) ctxSetup() context.Context {
 	md := n.mgr.opts.metadata.Copy()
 	if n.mgr.opts.perNodeMD != nil {
 		md = metadata.Join(md, n.mgr.opts.perNodeMD(n.id))
 	}
-	// a context for all of the streams
+	var ctx context.Context
 	ctx, n.cancel = context.WithCancel(context.Background())
 	ctx = metadata.NewOutgoingContext(ctx, md)
-	if err = n.channel.connect(ctx, n.conn); err != nil {
-		return fmt.Errorf("starting stream failed: %w", err)
-	}
-	return nil
+	return ctx
 }
 
 // close this node.

node.go

@@ -22,8 +22,9 @@ type managerOptions struct {
 
 func newManagerOptions() managerOptions {
 	return managerOptions{
-		backoff:    backoff.DefaultConfig,
-		sendBuffer: 0,
+		backoff:         backoff.DefaultConfig,
+		sendBuffer:      0,
+		nodeDialTimeout: 50 * time.Millisecond,
 	}
 }