<commit_before>package graft

import (
	"fmt"
	"math/rand"
	"net"
	"sync"
	"time"

	"golang.org/x/net/context"

	"github.com/lostz/graft/protocol"
	"github.com/uber-go/zap"

	"google.golang.org/grpc"
)

//Node ...
type Node struct {
	electTimer   *time.Timer
	id           string
	ip           string
	port         int
	leader       string
	logPath      string
	mu           sync.Mutex
	peers        []string
	quit         chan chan struct{}
	server       *grpc.Server
	state        State
	StateChg     chan StateChange
	term         uint64
	vote         string
	VoteRequests chan *protocol.VoteRequest
	VoteResponse chan *protocol.VoteResponse
	HeartBeats   chan *protocol.HeartbeatRequest
}

func (n *Node) broadcastVote() {
	for _, peer := range n.peers {
		n.votePeer(peer)
	}
}

func (n *Node) addPeer(peer string) {
	n.mu.Lock()
	defer n.mu.Unlock()
	if peer == n.id {
		return
	}
	for _, p := range n.peers {
		if p == peer {
			return
		}
	}
	n.peers = append(n.peers, peer)
}

func (n *Node) votePeer(peer string) {
	conn, err := grpc.Dial(peer, []grpc.DialOption{grpc.WithTimeout(300 * time.Millisecond), grpc.WithInsecure()}...)
	if err != nil {
		logger.Error(
			"grpc dial",
			zap.String("err", err.Error()),
		)
		return
	}
	defer conn.Close()
	c := protocol.NewRaftClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond)
	defer cancel()
	c.SendVoteRequest(ctx, &protocol.VoteRequest{Term: n.term, Candidate: n.id})
	return
}

func (n *Node) broadcastHearbeat() {

	for _, peer := range n.peers {
		go n.heartbeatPeer(peer)

	}
}

func (n *Node) heartbeatPeer(peer string) {
	conn, err := grpc.Dial(peer, []grpc.DialOption{grpc.WithTimeout(1000 * time.Millisecond), grpc.WithInsecure()}...)
	if err != nil {
		logger.Error(
			"grpc dial",
			zap.String("err", err.Error()),
		)
		//peer offline
		return
	}
	defer conn.Close()
	c := protocol.NewRaftClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond)
	defer cancel()
	c.Heartbeat(ctx, &protocol.HeartbeatRequest{Term: n.term, Leader: n.id})

}

func (n *Node) clearTimers() {
	if n.electTimer != nil {
		n.electTimer.Stop()
		n.electTimer = nil
	}
}

//Close wait until the state is processed
func (n *Node) Close() {
	if n.State() == CLOSED {
		return
	}
	n.server.GracefulStop()
	n.waitOnLoopFinish()
	n.clearTimers()
	n.closeLog()

}

//Heartbeat from leader
func (n *Node) Heartbeat(context context.Context, requset *protocol.HeartbeatRequest) (*protocol.Response, error) {
	if requset.Term < n.term {
		return &protocol.Response{}, nil
	}
	saveState := false
	if n.State() == LEADER {
		n.switchToFollower("")
	}
	n.mu.Lock()
	n.term = requset.Term
	n.vote = ""
	n.mu.Unlock()
	saveState = true
	n.resetElectionTimeout()
	if saveState {
		if err := n.writeState(); err != nil {
			logger.Error(
				"write state",
				zap.Uint64("term", n.term),
				zap.String("error", err.Error()),
			)
		}
	}
	n.switchToFollower(requset.Leader)
	return &protocol.Response{}, nil
}

func (n *Node) handleHeartBeat(hb *protocol.HeartbeatRequest) bool {
	if hb.Term < n.term {
		return false
	}
	saveState := false
	stepDown := false
	if hb.Term > n.term {
		n.term = hb.Term
		n.vote = ""
		stepDown = true
		saveState = true
	}
	if n.State() == CANDIDATE && hb.Term >= n.term {
		n.term = hb.Term
		n.vote = ""
		stepDown = true
		saveState = true
	}
	n.resetElectionTimeout()
	if saveState {
		if err := n.writeState(); err != nil {
			logger.Error(
				"write state",
				zap.Uint64("term", n.term),
				zap.String("error", err.Error()),
			)
			stepDown = true
		}
	}
	return stepDown

}

func (n *Node) handleVoteRequest(vreq *protocol.VoteRequest) bool {
	deny := &protocol.VoteResponse{Term: n.term, Granted: false}
	n.addPeer(vreq.Candidate)
	if vreq.Term < n.term {
		n.sendVoteResponse(vreq.Candidate, deny)
		return false
	}
	saveState := false
	stepDown := false<commit_msg>fix heartbeat bug<commit_after>package graft

import (
	"fmt"
	"math/rand"
	"net"
	"sync"
	"time"

	"golang.org/x/net/context"

	"github.com/lostz/graft/protocol"
	"github.com/uber-go/zap"

	"google.golang.org/grpc"
)

//Node ...
type Node struct {
	electTimer   *time.Timer
	id           string
	ip           string
	port         int
	leader       string
	logPath      string
	mu           sync.Mutex
	peers        []string
	quit         chan chan struct{}
	server       *grpc.Server
	state        State
	StateChg     chan StateChange
	term         uint64
	vote         string
	VoteRequests chan *protocol.VoteRequest
	VoteResponse chan *protocol.VoteResponse
	HeartBeats   chan *protocol.HeartbeatRequest
}

func (n *Node) broadcastVote() {
	for _, peer := range n.peers {
		n.votePeer(peer)
	}
}

func (n *Node) votePeer(peer string) {
	conn, err := grpc.Dial(peer, []grpc.DialOption{grpc.WithTimeout(300 * time.Millisecond), grpc.WithInsecure()}...)
	if err != nil {
		logger.Error(
			"grpc dial",
			zap.String("err", err.Error()),
		)
		return
	}
	defer conn.Close()
	c := protocol.NewRaftClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond)
	defer cancel()
	c.SendVoteRequest(ctx, &protocol.VoteRequest{Term: n.term, Candidate: n.id})
	return
}

func (n *Node) broadcastHearbeat() {

	for _, peer := range n.peers {
		go n.heartbeatPeer(peer)

	}
}

func (n *Node) heartbeatPeer(peer string) {
	conn, err := grpc.Dial(peer, []grpc.DialOption{grpc.WithTimeout(1000 * time.Millisecond), grpc.WithInsecure()}...)
	if err != nil {
		logger.Error(
			"grpc dial",
			zap.String("err", err.Error()),
		)
		//peer offline
		return
	}
	defer conn.Close()
	c := protocol.NewRaftClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond)
	defer cancel()
	c.Heartbeat(ctx, &protocol.HeartbeatRequest{Term: n.term, Leader: n.id})

}

func (n *Node) clearTimers() {
	if n.electTimer != nil {
		n.electTimer.Stop()
		n.electTimer = nil
	}
}

//Close wait until the state is processed
func (n *Node) Close() {
	if n.State() == CLOSED {
		return
	}
	n.server.GracefulStop()
	n.waitOnLoopFinish()
	n.clearTimers()
	n.closeLog()

}

//Heartbeat from leader
func (n *Node) Heartbeat(context context.Context, requset *protocol.HeartbeatRequest) (*protocol.Response, error) {
	n.HeartBeats <- requset
	return &protocol.Response{}, nil
}

func (n *Node) handleHeartBeat(hb *protocol.HeartbeatRequest) bool {
	if hb.Term < n.term {
		return false
	}
	saveState := false
	stepDown := false
	if hb.Term > n.term {
		n.term = hb.Term
		n.vote = ""
		stepDown = true
		saveState = true
	}
	if n.State() == CANDIDATE && hb.Term >= n.term {
		n.term = hb.Term
		n.vote = ""
		stepDown = true
		saveState = true
	}
	n.resetElectionTimeout()
	if saveState {
		if err := n.writeState(); err != nil {
			logger.Error(
				"write state",
				zap.Uint64("term", n.term),
				zap.String("error", err.Error()),
			)
			stepDown = true
		}
	}
	return stepDown

}

func (n *Node) handleVoteRequest(vreq *protocol.VoteRequest) bool {
	deny := &protocol.VoteResponse{Term: n.term, Granted: false}
	if vreq.Term < n.term {
		n.sendVoteResponse(vreq.Candidate, deny)
		return false
	}
	saveState := false
	stepDown := false
{'fraction_non_alphanumeric': 0.14653851511352556, 'fraction_numerical': 0.004457445326647165, 'mean_word_length': 3.233490566037736, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7398622', 'n_tokens_mistral': 2880, 'n_tokens_neox': 2702, 'n_words': 602}