raft

type AppEnts struct {
	// 'term' is provided by BaseMsg.Term
	// 'leaderId' is provied by BaseMsg.From
	BaseMsg

	// PrevLogIndex and PrevLogTerm will be used as consistency check.
	// Follower will only accept the AppEnts request if an entry with term
	// 'PrevLogTerm' and index 'PrevLogIndex' exists in its log.
	PrevLogIndex uint64
	PrevLogTerm  uint64

	// The entries that need to be appended to follower's log.
	// It can be nil if it's just a pure heartbeat message or a probing message.
	Entries []Entry

	// The index that can be safely committed by follower. Leader is the only one
	// who decides which index of entry can be committed, and leader will use this
	// field to piggyback the commit index to followers so followers can apply
	// committed commands to their state machines.
	LeaderCommit uint64
}

type AppEntsResp struct {
	BaseMsg
	// Whether its corresponding AppEnts request succeeded or not.
	Success bool

	// If its corresponding AppEnts succeeded, 'Index' will be the index of the last
	// matched entry in follower's log identified by that request, so leader can
	// update 'matchIndex' of follower.
	//
	// If its corresponding AppEnts failed, 'Index' will be the index used for
	// consisntecy check(AppEnts.PrevLogIndex) so leader can update follower's
	// 'nextIdx' for next probing.
	//
	// 'Index' field is not specified in paper because the paper phrases the
	// protocol in terms of RPC, response is paired with its request it can always
	// derive this 'Index' field from its corresponding request.
	// However, we decoupled our transport layer from Raft implementation so we
	// need to keep track of this ourselves.
	Index uint64

	// Hint from follower. If 'Hint' is not 0 leader should decrement the
	// 'nextIndex' to 'Hint'. We use 'Hint' to bypass all log entries that
	// a follower is missing, instead of sending one probing message per
	// log entry.
	// Stale response with a stale 'Hint' is fine. In that case a leader might
	// just probe from a stale position. We implement the follower and leader
	// in a way that processing a stale message is still correct.
	Hint uint64
}

type BaseMsg struct {
	Term     uint64 // The current term number of sender.
	To       string // The ID of receiver.
	From     string // The ID of sender.
	ToGUID   uint64 // The GUID of receiver.
	FromGUID uint64 // The GUID of sender.
	Epoch    uint64 // The epoch ID.
}

type Config struct {
	// ID of the node. Users must assign a unique ID to each Raft node in a group.
	ID string

	// The ID of the cluster. For now we only use it as a metric label to make our
	// monitoring system easier to aggregate and analyze metrics from raft nodes
	// of a same cluster.
	// TODO: use it to guard against misconfiguration of a cluster?
	ClusterID string

	// CandidateTimeout is used with 'RandomElectionRange' by a candidate to start
	// an election of next term, in units of 'tick'. If a candidate can't be elected
	// during time "CandidateTimeout + [0, RandomElectionRange]" it will start a new
	// election for next term.
	CandidateTimeout uint32

	// FollowerTimeout is used with 'RandomElectionRange' by a follower to start an
	// election of next term, in units of 'tick'. If a follower can't hear anything
	// from its leader during time "FollowerTimeout + [0, RandomElectionRange]" it
	// will start a new election of next term.
	FollowerTimeout uint32

	// LeaderStepdownTimeout is used to control how long the leader can maintain
	// its leadership without being able to contact a quorum of nodes, in units of
	// 'tick'. Once a leader loses the contact from a quorum after this amount
	// of time it will step down as leader. 'LeaderStepdownTimeout' must be larger
	// than 'HeartbeatTimeout' if it's not 0, otherwise the leader might falsely
	// think it loses its leadership and step down, which will make the whole system
	// unavailable until a new leader is elected. This is a voluntary decision and
	// do not depend on this to guarantee there's only one leader exists in cluster.
	// This is only used to step down a leader node who might already lose its
	// leadership at some point, so there might be multiple leaders(though only
	// one is effective) exist at the same time.
	//
	// If 'LeaderStepdownTimeout' is 0 then leader will keep its leadership until
	// a higher term is detected.
	//
	// The motivation of using 'LeaderStepdownTimeout' is when a network partiton
	// happens a stale leader may stay in leader state indefinitely even there's
	// a newer leader gets elected on the other side. Without realizing it might
	// already lost its leadership the stale leader will keep returning timeout
	// error to clients who are connecting to it. But if it steps down from leader
	// state at some point then clients can get more useful feedbacks(errors about
	// not being leader anymore) and fail over to the real leader node sooner.
	LeaderStepdownTimeout uint32

	// RandomElectionRange is a range of the randomness we introduced to leader
	// election to avoid the problem of split votes. Before electing itself as
	// leader, nodes need to wait for an extra time in range [0, RandomElectionRange]
	// to avoid split votes.
	RandomElectionRange uint32

	// HeartbeatTimeout is the interval of AppEnts requests, in units of 'tick'.
	// Leader must send at least one AppEnts request every this interval to maintain
	// its leadership. HeartbeatTimeout must be smaller than FollowerTimeout,
	// otherwise an elected leader can not maintain its leadership for a long time.
	HeartbeatTimeout uint32

	// SnapshotTimeout specifies how long it's expected to transfer snapshot AND
	// recover state from the snapshot. Leader will try to send messages to
	// followers at some intervals to maintain its leadership or retransmit the
	// message that might be lost. Because snapshot messages might take a long
	// time to transfer and apply so we use a different timeout to avoid
	// retransmitting unnecessary snapshot messages. The timeout should be set to
	// be large enough to cover the time spent in transferring and recovery.
	//
	// TODO: It might be hard to guess a good value beforehand, we could adjust the
	// timeout value dynamically using:
	//  - based on size of snapshot
	//  - use exponential back-off
	// Some relevant discussions in ZooKeeper group(https://issues.apache.org/jira/browse/ZOOKEEPER-1977)
	// (I don't think this has been addressed by them)
	SnapshotTimeout uint32

	// Duration per tick. We separated the logic time("tick") and real time(duration
	// per tick) to make our implementation more testable.
	DurationPerTick time.Duration

	// GenSeed is the function that takes a Raft ID as parameter and generates a
	// seed number for random number generator of that node. If users don't
	// specify one, a default seed generator will be used.
	GenSeed func(string) int64

	// MaxNumEntsPerAppEnts specifies the maximum number of entries per AppEnts
	// message. We don't want to saturate disk/network IO when synchronizing a
	// straggler.
	MaxNumEntsPerAppEnts uint32

	// LogEntriesAfterSnapshot specifies the number of entries to leave in the
	// log after a snapshot is taken. Once a snapshot is taken we are OK to
	// trim all log entries which are included in snapshot, but doing that might
	// force a leader to ship its entire snapshot to a follower when a follower
	// is missing anything in the snapshot. By keeping some entries in the log
	// after a snapshot, we can send those entries to followers to synchronize
	// them, instead of having to send an entire snapshot.
	// LogEntriesAfterSnapshot is the maximum number of entries we can trim(included
	// in snapshot) but keep them untrimmed in the log.
	LogEntriesAfterSnapshot uint64

	// SnapshotThreshold specifies how many applied commands there must be since last
	// snapshhot before we perform a new snapshot. If it's 0 then no snapshot will
	// be performed. For example, if you set "SnapshotThreshold" to 1000, then after
	// every 1000 commands applied to state machine a new snapshot will be taken.
	// There's a trade-off on this number -- the larger it is, the less frequently
	// we'll take snapshot so that it would cost less IO, but it might take longer
	// time in in recovery to apply all log entries that are not included in snapshot.
	SnapshotThreshold uint64

	// MaximumPendingProposals specifies the maximum number of pending proposals
	// allowed. Further requests will be blocked once the size of pending proposals
	// reaches this limit. If it is set to 0, no limit will be enforced.
	MaximumPendingProposals uint32

	// MaximumProposalBatch is the maximum number of proposals leader will try to
	// process them at once from proposal channel. The larger it is, the larger
	// throughput you might get, but also larger latency per request you might
	// experience.
	MaximumProposalBatch uint32
}

type Entry struct {
	Term  uint64
	Index uint64
	Cmd   []byte
	Type  uint8
}

type FSM interface {
	// Apply will be called once a command is committed in a Raft group. This callback
	// is called from a single goroutine to ensure that the state updates are applied
	// in the same order as they arrived. Application must apply the command in the
	// callback and can optionally return the result to it so that the one who proposed
	// the command can get the result from Pending.Res.
	//
	// Note that the behavior of the command should be deterministic, otherwise the
	// applications might end up in an inconsistent state.
	Apply(Entry) (result interface{})

	// OnLeadershipChange will be called to notify applications the change of
	// leader status of Raft. 'isLeader' tells application whether it is in
	// leader state or in non-leader state. 'term' tells application the
	// current term of the Raft node. 'leader' tells application the Raft
	// address of the leader node, it will be an empty string if the leader
	// is unknown yet, but once the the leader is known Raft will call this
	// callback with the address of the leader.
	//
	// The term number can be used in an application specific way. For example,
	// it's possible that multiple Raft leaders exist at the same time, but only
	// one leader can be effective, it is because internally Raft uses term
	// number to detect stale leaders, applications who also want one elected
	// primary to intereact with outsiders can also attach term number in their
	// messages so outsiders can use it to detect stale primaries. Another use
	// case is if applications want to implement the logic like read and then
	// modify based on the state which were read, they can also attach the term
	// number in modify command so when modify command is applied applications
	// can compare the term number attached in modify command to the term number
	// of the command itself so if they don't match there might be a potential
	// conflict modification from another leader. By guaranteeing that the term
	// when applications read the state and the term of the command which will
	// modify state machine are the same, there can't be a conflict change from
	// another leader in between.
	OnLeadershipChange(isLeader bool, term uint64, leader string)

	// OnMembershipChange will be called to notify applications of membership
	// changes over time.
	OnMembershipChange(membership Membership)

	// Snapshot is called once a snapshot is needed to be taken to compact log.
	// Applications should return an implementation of "Snapshoter" whose method
	// "Save" will be called to dump the actual state in a different goroutine.
	// The state that will be dumped in "Snapshoter.Save" MUST be the state at
	// the point the callback "Snapshot" is called.
	//
	// Once the "Snapshot" is returned further mutations might be applied in Apply
	// callback. However, these mutations should not affect the state that is
	// being dumped in "Snapshoter.Save".
	//
	// If any erorr is returned no snapshot will be taken.
	Snapshot() (Snapshoter, error)

	// SnapshotRestore is called once applications need to restore its state from
	// snapshot. Once it's called applications need to discard its current state
	// and restore its state from the given reader. Also the index and term number
	// of last applied command to the snapshot is passed to the applications.
	SnapshotRestore(reader io.Reader, lastIndex, lastTerm uint64)
}

type InstallSnapshot struct {
	// 'term' is provided by BaseMsg.Term
	// 'leaderId' is provied by BaseMsg.From
	BaseMsg

	// The index of last applied command included in this snapshot.
	LastIndex uint64
	// The term of last applied command included in this snapshot.
	LastTerm uint64

	// The latest applied membership change in snapshot.
	Membership Membership

	// The payload of a snapshot.
	//
	// This message must be created with Body pointing to actual payload.
	// Transport is responsible for sending it out on sender side and pointing
	// it to actual payload on receiver side.
	//
	// Body MUST be closed after the messge is sent out on sender side or
	// processed on receiver side.
	Body io.ReadCloser
}

type Log interface {
	// The index of first entry in log. If there's no entry in log,
	// empty will be returned as true.
	FirstIndex() (index uint64, empty bool)

	// The index of last entry in log. If there's no entry in log,
	// empty will be returned as true.
	LastIndex() (index uint64, empty bool)

	// GetBound returns the first index and last index of a log if it's not empty,
	// otherwise the third return value "empty" will be returned as true.
	GetBound() (firstIndex, lastIndex uint64, empty bool)

	// Append appends multiple entries to log. It's user's responsibility to
	// guarantee the appended entries have contiguous indices with existing
	// entries in log. If any errors occur, process will log an error and die.
	Append(...Entry)

	// Truncate truncates all log entries after(not include) the entry with the
	// given index. If any errors occur, process will log an error and die.
	Truncate(index uint64)

	// Trim deletes all entries up to(include) the given index.
	// This is used for log compaction. If any errors occur, process will log an
	// error and die.
	Trim(index uint64)

	// Term returns the term number of an entry at the given index. It's caller's
	// responsibility to guarantee an entry with the given index is in the log,
	// otherwise we'll log an error message and die.
	Term(index uint64) uint64

	// Entries returns a slice of entries in range [beg, end). If 'end' exceeds
	// the index of last entry in log then the log should return as many entries
	// as it can. But users need to guarantee the index 'beg' exists in log,
	// otherwise we will panic.
	Entries(beg, end uint64) []Entry

	// GetIterator returns an iterator that can be used to iterate the log,
	// starting at the entry with ID 'first' or the first entry after where
	// it would be if there is no entry with that ID.
	GetIterator(first uint64) LogIterator

	// Close releases any resources used by the log. The log must not
	// be used after calling Close.
	Close()
}

type LogIterator interface {
	// Next advances the iterator. It returns true if it was able to
	// advance to the next entry or false if there are no more entries
	// or an error occurred. Use Err() to tell the difference.
	Next() bool

	// Entry returns the current entry. Next must be called before every
	// call to Entry.
	Entry() Entry

	// Err returns any error that occurred during iteration or nil if no
	// error occurred.
	Err() error

	// Close releases any resources associated with the iterator and
	// returns an error, if any occurred during iteration. The iterator
	// must not be used after Close is called.
	Close() error
}

type Membership struct {
	// Members is the cluster members.
	Members []string

	// Epoch is an id for this instance of raft. It will never change after the
	// initial configuration. It is used to detect misconfigurations and prevent
	// corruption.
	Epoch uint64

	// Index is the index of the change in log.
	// (When serialized in logs/snapshots, Index is unused.)
	Index uint64

	// Term is the term of the change in log.
	// (When serialized in logs/snapshots, Term is unused.)
	Term uint64
}

type Msg interface {
	GetTerm() uint64
	SetTerm(uint64)
	GetTo() string
	SetTo(string)
	GetFrom() string
	SetFrom(string)
	GetToGUID() uint64
	SetToGUID(uint64)
	GetFromGUID() uint64
	SetFromGUID(uint64)
	GetEpoch() uint64
	SetEpoch(uint64)
}

type Pending struct {
	// Res is the result of executing a command. When a command is applied to
	// the state machine, FSM.Apply is called. The return value of that function
	// is stored here.
	Res interface{}

	//  Err will be set if any error occured.
	Err error

	// Done is the channel that will be signaled when the command concludes.
	Done chan struct{}
	// contains filtered or unexported fields
}

type Raft struct {
	// contains filtered or unexported fields
}

type SnapshotFileReader interface {
	io.ReadCloser

	// GetMetadata returns the meatadata of this snapshot file.
	GetMetadata() SnapshotMetadata
}

type SnapshotFileWriter interface {
	io.Writer

	// Commit should be called once we have written all state to the file.
	// If everything goes smoothly and "Commit" succeeds, the snapshot file will
	// become "effective"(durable and visible).
	Commit() error

	// Abort aborts a pending snapshot file.
	Abort() error

	// GetMetadata returns the metadata of the snapshot file.
	GetMetadata() SnapshotMetadata
}

type SnapshotManager interface {
	// BeginSnapshot returns a "SnapshotFileWriter" to which applications can dump
	// their state.
	BeginSnapshot(SnapshotMetadata) (SnapshotFileWriter, error)

	// GetSnapshot returns a "SnapshotFileReader" of the latest snapshot file from
	// which applications can restore their state. If reader is returned as nil it
	// means there's no snapshot file found.
	GetSnapshot() SnapshotFileReader

	// GetSnapshotMetadata returns the metadata of the latest file, and
	// (optionally) a path to a file on disk where the snapshot is stored. If
	// there's no snapshot file found 'NilSnapshotMetadata', "" will be
	// returned.
	// The file path is only indended for backups. Other packages shouldn't
	// assume anything about the snapshot file format, and not all storage
	// implementations may provide it.
	GetSnapshotMetadata() (SnapshotMetadata, string)
}

type SnapshotMetadata struct {
	// The last index of applied command to the snapshot.
	LastIndex uint64
	// The term of last index of applied command to the snapshot.
	LastTerm uint64
	// Membership information in snapshot.
	Membership *Membership
}

type Snapshoter interface {
	// Save is the callback that will be called by Raft. When this callback is
	// called applications should write the serialized state to the given
	// "writer". This callback will be called concurrently with "FSM.Apply" so
	// it should be implemented in a way that concurrent mutations made in
	// "FSM.Apply" will have no effects to the state that is being dumped.
	//
	// If any error is returned the snapshot file will be discarded.
	Save(writer io.Writer) error

	// Release will be called by Raft once Raft is done(either succeed or fail)
	// with the snapshot. Applications should release any resources that are
	// assoicated the snapshot when this callback is called.
	Release()
}

type State interface {
	// SaveState persists states 'voteFor' and 'currentTerm' to disk atomically.
	SaveState(voteFor string, term uint64)

	// GetState returns the persisted states "voteFor" and "currentTerm"
	GetState() (voteFor string, term uint64)

	// GetCurrentTerm returns state "currentTerm".
	GetCurrentTerm() uint64

	// SetCurrentTerm modifies state "currentTerm", state "voteFor" will not be
	// changed.
	SetCurrentTerm(uint64)

	// GetVoteFor returns state "voteFor".
	GetVoteFor() string

	// SetVoteFor modifies state "voteFor", state "currentTerm" will not be changed.
	SetVoteFor(string)

	// GetMyGUID gets this node's GUID.
	GetMyGUID() uint64

	// GetGUIDFor gets the known GUID for the given node, or an empty string if
	// no messages from that node have been received.
	GetGUIDFor(id string) uint64

	// SetGUIDFor sets the known GUID for the given node.
	SetGUIDFor(id string, guid uint64)

	// FilterGUIDs forgets about GUIDs of nodes that are not in the given set of ids.
	FilterGUIDs(ids []string)
}

type Storage struct {
	State
	SnapshotManager
	// contains filtered or unexported fields
}

type Transport interface {
	// Addr returns the local address of the transport.
	Addr() string

	// Receive returns a channel for receiving incoming messages
	// from the network.
	Receive() <-chan Msg

	// Send asynchronously sends message 'm' to the node referred
	// by 'm.GetTo()' through the transport.
	Send(m Msg)

	// Close closes all connections.
	Close() error
}

type TransportConfig struct {
	// Transport address that the raft instance listens on.
	Addr string

	// Capacity of the incoming message channel that is returned in
	// 'Transport.Receive()'.
	MsgChanCap int
}

type VoteReq struct {
	// 'term' is provided by BaseMsg.Term
	// 'candidateId' is provied by BaseMsg.From
	BaseMsg

	// The index and term number of last entry in candidate's log. Candidate can
	// be elected as leader only if it has "sufficient" up-to-dated log history.
	LastLogIndex uint64
	LastLogTerm  uint64
}

type VoteResp struct {
	BaseMsg
	// Whether its corresponding VoteReq is granted or not.
	Granted bool
}