fix(tests): added tests for Freeze/Reset

mutext_test.go

@@ -10,41 +10,41 @@ import (
 	"github.com/yaitoo/sqle"
 )
 
-func createCluster(ctx context.Context, num int) ([]string, []*Node, []func(), error) {
+func createCluster(num int) ([]string, []*Node, func(), error) {
 	var peers []string
 	var nodes []*Node
 	var clean []func()
 
+	release := func() {
+		for _, c := range clean {
+			c()
+		}
+	}
+
 	for i := 0; i < num; i++ {
 		db, fn, err := createSqlite3()
 		if err != nil {
-			return nil, nil, clean, err
+			return nil, nil, release, err
 		}
 		clean = append(clean, fn)
 		n := NewNode(getFreeAddr(), sqle.Open(db))
-		err = n.Start(ctx)
+		err = n.Start()
 		if err != nil {
-			return nil, nil, clean, err
+			return nil, nil, release, err
 		}
 		peers = append(peers, n.addr)
 		nodes = append(nodes, n)
+
+		clean = append(clean, n.Stop)
 	}
 
-	return peers, nodes, clean, nil
+	return peers, nodes, release, nil
 }
 
 func TestLock(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.TODO())
-	defer cancel()
-	peers, nodes, clean, err := createCluster(ctx, 5)
-
+	peers, nodes, clean, err := createCluster(5)
 	require.NoError(t, err)
-
-	defer func() {
-		for _, c := range clean {
-			c()
-		}
-	}()
+	defer clean()
 
 	tests := []struct {
 		name string
@@ -130,7 +130,9 @@ func TestLock(t *testing.T) {
 				m := New("lock_should_work", "wallet", "minority_nodes_are_down", WithPeers(peers...), WithTTL(10*time.Second))
 
 				nodes[0].Stop()
+				defer nodes[0].Start() // nolint: errcheck
 				nodes[1].Stop()
+				defer nodes[1].Start() // nolint: errcheck
 
 				err := m.Lock(context.TODO())
 
@@ -151,8 +153,11 @@ func TestLock(t *testing.T) {
 				m := New("lock_should_work", "wallet", "majority_nodes_are_down", WithPeers(peers...), WithTTL(10*time.Second))
 
 				nodes[0].Stop()
+				defer nodes[0].Start() // nolint: errcheck
 				nodes[1].Stop()
+				defer nodes[1].Start() // nolint: errcheck
 				nodes[2].Stop()
+				defer nodes[2].Start() // nolint: errcheck
 
 				err = m.Lock(context.TODO())
 				r.Error(err, async.ErrTooLessDone)
@@ -168,17 +173,9 @@ func TestLock(t *testing.T) {
 }
 
 func TestRenew(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.TODO())
-	defer cancel()
-	peers, nodes, clean, err := createCluster(ctx, 5)
-
+	peers, nodes, clean, err := createCluster(5)
 	require.NoError(t, err)
-
-	defer func() {
-		for _, c := range clean {
-			c()
-		}
-	}()
+	defer clean()
 
 	tests := []struct {
 		name string
@@ -290,7 +287,9 @@ func TestRenew(t *testing.T) {
 				r.Equal("renew_minority", m.lease.Key)
 
 				nodes[0].Stop()
+				defer nodes[0].Start() // nolint: errcheck
 				nodes[1].Stop()
+				defer nodes[1].Start() // nolint: errcheck
 				err = m.Renew(context.TODO())
 				r.NoError(err)
 
@@ -309,8 +308,12 @@ func TestRenew(t *testing.T) {
 				r.Equal("renew_majority", m.lease.Key)
 
 				nodes[0].Stop()
+				defer nodes[0].Start() // nolint: errcheck
 				nodes[1].Stop()
+				defer nodes[1].Start() // nolint: errcheck
 				nodes[2].Stop()
+				defer nodes[2].Start() // nolint: errcheck
+
 				err = m.Renew(context.TODO())
 				r.ErrorIs(err, async.ErrTooLessDone)
 
@@ -326,17 +329,9 @@ func TestRenew(t *testing.T) {
 }
 
 func TestUnlock(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.TODO())
-	defer cancel()
-	peers, nodes, clean, err := createCluster(ctx, 5)
-
+	peers, nodes, clean, err := createCluster(5)
 	require.NoError(t, err)
-
-	defer func() {
-		for _, c := range clean {
-			c()
-		}
-	}()
+	defer clean()
 
 	tests := []struct {
 		name string
@@ -407,7 +402,11 @@ func TestUnlock(t *testing.T) {
 				r.Equal("unlock_minority", m.lease.Key)
 
 				nodes[0].Stop()
+				defer nodes[0].Start() // nolint: errcheck
+
 				nodes[1].Stop()
+				defer nodes[1].Start() // nolint: errcheck
+
 				err = m.Unlock(context.TODO())
 				r.NoError(err)
 
@@ -428,8 +427,11 @@ func TestUnlock(t *testing.T) {
 				r.Equal("unlock_majority", m.lease.Key)
 
 				nodes[0].Stop()
+				defer nodes[0].Start() // nolint: errcheck
 				nodes[1].Stop()
+				defer nodes[1].Start() // nolint: errcheck
 				nodes[2].Stop()
+				defer nodes[2].Start() // nolint: errcheck
 
 				err = m.Unlock(context.TODO())
 				r.ErrorIs(err, async.ErrTooLessDone)
@@ -446,17 +448,9 @@ func TestUnlock(t *testing.T) {
 }
 
 func TestTopic(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.TODO())
-	defer cancel()
-	peers, nodes, clean, err := createCluster(ctx, 5)
-
+	peers, nodes, clean, err := createCluster(5)
 	require.NoError(t, err)
-
-	defer func() {
-		for _, c := range clean {
-			c()
-		}
-	}()
+	defer clean()
 
 	tests := []struct {
 		name string
@@ -516,7 +510,9 @@ func TestTopic(t *testing.T) {
 				r.Equal("freeze", m.lease.Key)
 
 				nodes[0].Stop()
+				defer nodes[0].Start() // nolint: errcheck
 				nodes[1].Stop()
+				defer nodes[1].Start() // nolint: errcheck
 
 				err = m.Freeze(context.Background(), "freeze")
 				r.NoError(err)
@@ -560,8 +556,11 @@ func TestTopic(t *testing.T) {
 				r.Equal("freeze", m.lease.Key)
 
 				nodes[0].Stop()
+				defer nodes[0].Start() // nolint: errcheck
 				nodes[1].Stop()
+				defer nodes[1].Start() // nolint: errcheck
 				nodes[2].Stop()
+				defer nodes[2].Start() // nolint: errcheck
 
 				err = m.Freeze(context.Background(), "freeze")
 				r.ErrorIs(err, async.ErrTooLessDone)

node.go

@@ -2,6 +2,7 @@ package dlm
 
 import (
 	"log/slog"
+	"net"
 	"net/rpc"
 	"sync"
 
@@ -14,7 +15,6 @@ func NewNode(addr string, db *sqle.DB, options ...NodeOption) *Node {
 		db:     db,
 		frozen: make(map[string]struct{}),
 		logger: slog.Default(),
-		close:  make(chan struct{}),
 	}
 
 	for _, o := range options {
@@ -25,12 +25,14 @@ func NewNode(addr string, db *sqle.DB, options ...NodeOption) *Node {
 }
 
 type Node struct {
-	mu     sync.Mutex
+	mu     sync.RWMutex
 	db     *sqle.DB
 	logger *slog.Logger
 	frozen map[string]struct{}
 
-	addr   string
-	server *rpc.Server
-	close  chan struct{}
+	stopped bool
+
+	addr     string
+	listener net.Listener
+	server   *rpc.Server
 }

node_svc.go

@@ -9,59 +9,68 @@ import (
 )
 
 // Start start the node and its RPC service
-func (n *Node) Start(ctx context.Context) error {
+func (n *Node) Start() error {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
 	l, err := net.Listen("tcp", n.addr)
 	if err != nil {
 		return err
 	}
 
-	_, err = n.db.ExecContext(ctx, CreateTableLease)
+	_, err = n.db.ExecContext(context.Background(), CreateTableLease)
 	if err != nil {
 		return err
 	}
 
-	_, err = n.db.ExecContext(ctx, CreateTableTopic)
+	_, err = n.db.ExecContext(context.Background(), CreateTableTopic)
 	if err != nil {
 		return err
 	}
 
 	n.server = rpc.NewServer()
-	go n.waitClose(ctx, l)
-	go n.waitRequest(l)
-
+	n.listener = l
+	go n.waitRequest()
+	n.stopped = false
+	n.logger.Info("dlm: node is running")
 	return n.server.RegisterName("dlm", n)
 }
 
 // Stop stop the node and its RPC service
 func (n *Node) Stop() {
-	go func() {
-		n.close <- struct{}{}
-	}()
-	n.logger.Info("dlm: node stopped")
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	n.listener.Close()
+	n.stopped = true
+	n.logger.Info("dlm: node is stopped")
+}
+
+func (n *Node) isStopped() bool {
+	n.mu.RLock()
+	defer n.mu.RUnlock()
+
+	return n.stopped
 }
 
-func (n *Node) waitRequest(l net.Listener) {
+func (n *Node) waitRequest() {
+
 	for {
-		conn, err := l.Accept()
+		conn, err := n.listener.Accept()
+
 		if err == nil {
+
+			if n.isStopped() {
+				return
+			}
+
 			go n.server.ServeConn(conn)
+
 		} else {
 			if errors.Is(err, net.ErrClosed) {
 				return
 			}
 
 			n.logger.Warn("dlm: wait request", slog.String("err", err.Error()), slog.String("addr", n.addr))
 		}
-
 	}
 }
-
-func (n *Node) waitClose(ctx context.Context, l net.Listener) {
-
-	select {
-	case <-n.close:
-	case <-ctx.Done():
-	}
-
-	l.Close()
-}

node_test.go

@@ -1,7 +1,6 @@
 package dlm
 
 import (
-	"context"
 	"database/sql"
 	"net"
 	"os"
@@ -48,16 +47,13 @@ func createSqlite3() (*sql.DB, func(), error) {
 }
 
 func TestLease(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
 	db, clean, err := createSqlite3()
 	require.NoError(t, err)
 	defer clean()
-
 	n := NewNode(getFreeAddr(), sqle.Open(db))
-	err = n.Start(ctx)
+	err = n.Start()
 	require.NoError(t, err)
+	defer n.Stop()
 
 	walletTerms := sqle.Duration(5 * time.Second)
 	userTerms := sqle.Duration(3 * time.Second)
@@ -233,16 +229,14 @@ func TestLease(t *testing.T) {
 }
 
 func TestNodeTopic(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
 	db, clean, err := createSqlite3()
 	require.NoError(t, err)
 	defer clean()
 
 	n := NewNode(getFreeAddr(), sqle.Open(db))
-	err = n.Start(ctx)
+	err = n.Start()
 	require.NoError(t, err)
+	defer n.Stop()
 
 	terms := sqle.Duration(5 * time.Second)