raft: implement Raft consensus protocol

.github/CODEOWNERS

@@ -1,7 +1,12 @@
 # global code owners
-*   @TimSatke @tomarrell
+*                                       @TimSatke @tomarrell
 
+internal/database/storage/btree         @tomarrell
+internal/parser                         @TimSatke @SUMUKHA-PK
+internal/raft                           @SUMUKHA-PK
+internal/raft/cluster                   @TimSatke
+internal/raft/message                   @TimSatke
 internal/compiler @TimSatke
 internal/database/storage/btree @tomarrell
 internal/network @TimSatke
-internal/parser @TimSatke @SUMUKHA-PK
+internal/parser @TimSatke @SUMUKHA-PK

cmd/lbadd/main.go

@@ -160,7 +160,7 @@ func startNode(cmd *cobra.Command, args []string) {
 	if err := node.ListenAndServe(cmd.Context(), addr); err != nil {
 		log.Error().
 			Err(err).
-			Msg("listen and serve")
+			Msg("open")
 		os.Exit(ExitAbnormal)
 	}
 }

go.mod

@@ -3,18 +3,19 @@ module github.com/tomarrell/lbadd
 go 1.13
 
 require (
+	github.com/golang/protobuf v1.4.2
 	github.com/google/go-cmp v0.5.1
 	github.com/kr/text v0.2.0 // indirect
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
 	github.com/oklog/ulid v1.3.1
 	github.com/rs/zerolog v1.19.0
 	github.com/spf13/afero v1.3.2
 	github.com/spf13/cobra v1.0.0
-	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/stretchr/testify v1.6.1
-	golang.org/x/net v0.0.0-20200505041828-1ed23360d12c
-	golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a
+	golang.org/x/net v0.0.0-20200707034311-ab3426394381
+	golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208
 	golang.org/x/text v0.3.3
-	golang.org/x/tools v0.0.0-20200521211927-2b542361a4fc
-	gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f // indirect
+	golang.org/x/tools v0.0.0-20200713011307-fd294ab11aed
+	google.golang.org/protobuf v1.25.0
+	gotest.tools v2.2.0+incompatible
 )

internal/compiler/command/command.go

@@ -18,9 +18,9 @@ var _ Command = (*DropIndex)(nil)
 var _ Command = (*DropTrigger)(nil)
 var _ Command = (*DropView)(nil)
 var _ Command = (*Update)(nil)
-var _ Command = (*Insert)(nil)
 var _ Command = (*Join)(nil)
 var _ Command = (*Limit)(nil)
+var _ Command = (*Insert)(nil)
 
 // Command describes a structure that can be executed by the database executor.
 // Instead of using bytecode, we use a hierarchical structure for the executor.

internal/network/example_test.go

@@ -6,6 +6,7 @@ import (
 	"log"
 
 	"github.com/rs/zerolog"
+	"github.com/tomarrell/lbadd/internal/id"
 	"github.com/tomarrell/lbadd/internal/network"
 )
 
@@ -26,7 +27,8 @@ func ExampleServer() {
 
 	<-srv.Listening() // wait for the server to come up
 
-	client, _ := network.DialTCP(ctx, ":59513")
+	clientID := id.Create()
+	client, _ := network.DialTCP(ctx, clientID, ":59513")
 	defer func() {
 		_ = client.Close()
 	}()

internal/network/mocks/doc.go

@@ -0,0 +1,2 @@
+// Package mocks provides generated mock implementations for easy testing.
+package mocks

internal/network/server.go

@@ -2,9 +2,10 @@ package network
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"net"
+
+	"github.com/tomarrell/lbadd/internal/id"
 )
 
 // ConnHandler is a handler function for handling new connections. It will be
@@ -29,11 +30,16 @@ type Server interface {
 	// Addr returns the address that this server is listening to.
 	Addr() net.Addr
 
+	// OwnID returns the ID of this server. The remote ID of any connection is
+	// the own ID of another server.
+	OwnID() id.ID
 	// OnConnect sets a callback that will be executed whenever a new connection
 	// connects to this server.
 	OnConnect(ConnHandler)
 }
 
+//go:generate mockery -case=snake -name=Conn
+
 // Conn describes a network connection. One can send a message with Conn.Send,
 // and receive one with Conn.Receive. Unlike an io.Writer, the data that is
 // passed into Send is guaranteed to be returned in a single Receive call on the
@@ -42,9 +48,8 @@ type Server interface {
 type Conn interface {
 	io.Closer
 
-	// ID returns the ID of this connection. It can be used to uniquely identify
-	// this connection globally.
-	ID() ID
+	// RemoteID returns the own ID of the server that this connection points to.
+	RemoteID() id.ID
 	// Send sends the given payload to the remote part of this connection. The
 	// message will not be chunked, and can be read with a single call to
 	// Conn.Receive.
@@ -53,10 +58,3 @@ type Conn interface {
 	// is a byte slice that was sent with a single call to Conn.Send.
 	Receive(context.Context) ([]byte, error)
 }
-
-// ID describes an identifier that is used for connections. An ID has to be
-// unique application-wide. IDs must not be re-used.
-type ID interface {
-	fmt.Stringer
-	Bytes() []byte
-}

internal/network/tcp_conn.go

@@ -5,8 +5,10 @@ import (
 	"fmt"
 	"net"
 	"sync"
+	"sync/atomic"
 	"time"
 
+	"github.com/tomarrell/lbadd/internal/id"
 	"golang.org/x/net/context"
 	"golang.org/x/sync/errgroup"
 )
@@ -22,8 +24,8 @@ var (
 var _ Conn = (*tcpConn)(nil)
 
 type tcpConn struct {
-	id     ID
-	closed bool
+	remoteID id.ID
+	closed   int32
 
 	readLock   sync.Mutex
 	writeLock  sync.Mutex
@@ -32,7 +34,7 @@ type tcpConn struct {
 
 // DialTCP dials to the given address, assuming a TCP network. The returned Conn
 // is ready to use.
-func DialTCP(ctx context.Context, addr string) (Conn, error) {
+func DialTCP(ctx context.Context, ownID id.ID, addr string) (Conn, error) {
 	// dial the remote endpoint
 	var d net.Dialer
 	conn, err := d.DialContext(ctx, "tcp", addr)
@@ -43,38 +45,50 @@ func DialTCP(ctx context.Context, addr string) (Conn, error) {
 	// create a new connection object
 	tcpConn := newTCPConn(conn)
 
-	// receive the connection ID from the remote endpoint and apply it
-	myID, err := tcpConn.Receive(ctx)
+	// receive the remote ID from the remote endpoint and apply it
+	remoteID, err := tcpConn.Receive(ctx)
 	if err != nil {
 		_ = tcpConn.Close()
-		return nil, fmt.Errorf("receive ID: %w", err)
+		return nil, fmt.Errorf("receive remote ID: %w", err)
 	}
-	parsedID, err := parseID(myID)
+	parsedID, err := id.Parse(remoteID)
 	if err != nil {
 		_ = tcpConn.Close()
-		return nil, fmt.Errorf("parse ID: %w", err)
+		return nil, fmt.Errorf("parse remote ID: %w", err)
+	}
+	tcpConn.remoteID = parsedID
+
+	// send own ID to remote endpoint
+	err = tcpConn.Send(ctx, ownID.Bytes())
+	if err != nil {
+		_ = tcpConn.Close()
+		return nil, fmt.Errorf("send own ID: %w", err)
 	}
-	tcpConn.id = parsedID
 
 	// return the connection object
 	return tcpConn, nil
 }
 
+// NewTCPConn wraps the underlying connection into a tcpConn.
+func NewTCPConn(underlying net.Conn) Conn {
+	return newTCPConn(underlying)
+}
+
 func newTCPConn(underlying net.Conn) *tcpConn {
-	id := createID()
+	id := id.Create()
 	conn := &tcpConn{
-		id:         id,
+		remoteID:   id,
 		underlying: underlying,
 	}
 	return conn
 }
 
-func (c *tcpConn) ID() ID {
-	return c.id
+func (c *tcpConn) RemoteID() id.ID {
+	return c.remoteID
 }
 
 func (c *tcpConn) Send(ctx context.Context, payload []byte) error {
-	if c.closed {
+	if atomic.LoadInt32(&c.closed) == 1 {
 		return ErrClosed
 	}
 
@@ -137,7 +151,7 @@ func (c *tcpConn) sendAsync(payload []byte) chan error {
 }
 
 func (c *tcpConn) Receive(ctx context.Context) ([]byte, error) {
-	if c.closed {
+	if atomic.LoadInt32(&c.closed) == 1 {
 		return nil, ErrClosed
 	}
 
@@ -202,7 +216,7 @@ func (c *tcpConn) receiveAsync() chan interface{} {
 }
 
 func (c *tcpConn) Close() error {
-	c.closed = true
+	atomic.StoreInt32(&c.closed, 1)
 
 	// release all resources
 	ctx := context.Background()