itest: add itest covering the leader healthcheck

itest/list_on_test.go

@@ -233,6 +233,10 @@ var allTestCases = []*lntest.TestCase{
 		Name:     "etcd failover",
 		TestFunc: testEtcdFailover,
 	},
+	{
+		Name:     "leader health check",
+		TestFunc: testLeaderHealthCheck,
+	},
 	{
 		Name:     "hold invoice force close",
 		TestFunc: testHoldInvoiceForceClose,

itest/lnd_etcd_failover_test.go

@@ -4,6 +4,11 @@
 package itest
 
 import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"sync"
 	"testing"
 	"time"
 
@@ -137,3 +142,167 @@ func testEtcdFailoverCase(ht *lntest.HarnessTest, kill bool) {
 	// process.
 	ht.Shutdown(carol2)
 }
+
+// Proxy is a simple TCP proxy that forwards all traffic between a local and a
+// remote address. We use it to simulate a network partition in the leader
+// health check test.
+type Proxy struct {
+	listenAddr string
+	targetAddr string
+	ctx        context.Context
+	cancel     context.CancelFunc
+	wg         sync.WaitGroup
+	stopped    chan struct{}
+}
+
+// NewProxy creates a new Proxy instance with a provided context.
+func NewProxy(ctx context.Context, listenAddr, targetAddr string) *Proxy {
+	ctx, cancel := context.WithCancel(ctx)
+	return &Proxy{
+		listenAddr: listenAddr,
+		targetAddr: targetAddr,
+		ctx:        ctx,
+		cancel:     cancel,
+		stopped:    make(chan struct{}),
+	}
+}
+
+// Start starts the proxy. It listens on the listen address and forwards all
+// traffic to the target address.
+func (p *Proxy) Start(t *testing.T) {
+	listener, err := net.Listen("tcp", p.listenAddr)
+	require.NoError(t, err, "Failed to listen on %s", p.listenAddr)
+	t.Logf("Proxy is listening on %s", p.listenAddr)
+
+	p.wg.Add(1)
+	go func() {
+		defer func() {
+			close(p.stopped)
+			p.wg.Done()
+		}()
+
+		for {
+			select {
+			case <-p.ctx.Done():
+				listener.Close()
+				return
+			default:
+			}
+
+			conn, err := listener.Accept()
+			if err != nil {
+				if p.ctx.Err() != nil {
+					// Context is done, exit the loop
+					return
+				}
+				t.Logf("Proxy failed to accept connection: %v", err)
+				continue
+			}
+
+			p.wg.Add(1)
+			go p.handleConnection(t, conn)
+		}
+	}()
+}
+
+// handleConnection handles an accepted connection and forwards all traffic
+// between the listener and target.
+func (p *Proxy) handleConnection(t *testing.T, conn net.Conn) {
+	targetConn, err := net.Dial("tcp", p.targetAddr)
+	require.NoError(t, err, "Failed to connect to target %s", p.targetAddr)
+
+	defer func() {
+		conn.Close()
+		targetConn.Close()
+		p.wg.Done()
+	}()
+
+	done := make(chan struct{})
+
+	p.wg.Add(2)
+	go func() {
+		defer p.wg.Done()
+		io.Copy(targetConn, conn)
+	}()
+
+	go func() {
+		defer p.wg.Done()
+		io.Copy(conn, targetConn)
+		close(done)
+	}()
+
+	select {
+	case <-p.ctx.Done():
+	case <-done:
+	}
+}
+
+// Stop stops the proxy and waits for all connections to be closed and all
+// goroutines to be stopped.
+func (p *Proxy) Stop(t *testing.T) {
+	t.Log("Stopping proxy", time.Now())
+	p.cancel()
+	p.wg.Wait()
+	<-p.stopped
+	t.Log("Proxy stopped", time.Now())
+}
+
+// testLeaderHealthCheck tests that a node is properly shut down when the leader
+// health check fails.
+func testLeaderHealthCheck(ht *lntest.HarnessTest) {
+	clientPort := port.NextAvailablePort()
+
+	// Let's start a test etcd instance that we'll redirect through a proxy.
+	etcdCfg, cleanup, err := kvdb.StartEtcdTestBackend(
+		ht.T.TempDir(), uint16(clientPort),
+		uint16(port.NextAvailablePort()), "",
+	)
+	require.NoError(ht, err, "Failed to start etcd instance")
+
+	// Start a proxy that will forward all traffic to the etcd instance.
+	clientAddr := fmt.Sprintf("localhost:%d", clientPort)
+	proxyAddr := fmt.Sprintf("localhost:%d", port.NextAvailablePort())
+
+	ctx, cancel := context.WithCancel(ht.Context())
+	defer cancel()
+
+	proxy := NewProxy(ctx, proxyAddr, clientAddr)
+	proxy.Start(ht.T)
+
+	// With the proxy in place, we can now configure the etcd client to
+	// connect to the proxy instead of the etcd instance.
+	etcdCfg.Host = "http://" + proxyAddr
+
+	defer cleanup()
+
+	// Make leader election session TTL 5 sec to make the test run fast.
+	const leaderSessionTTL = 5
+
+	observer, err := cluster.MakeLeaderElector(
+		ht.Context(), cluster.EtcdLeaderElector, "observer",
+		lncfg.DefaultEtcdElectionPrefix, leaderSessionTTL, etcdCfg,
+	)
+	require.NoError(ht, err, "Cannot start election observer")
+
+	// Start Carol with cluster support.
+	password := []byte("the quick brown fox jumps the lazy dog")
+	stateless := false
+	cluster := true
+
+	carol, _, _ := ht.NewNodeWithSeedEtcd(
+		"Carol", etcdCfg, password, stateless, cluster,
+		leaderSessionTTL,
+	)
+
+	// Make sure Carol is indeed the leader.
+	assertLeader(ht, observer, "Carol")
+
+	// Stop the proxy so that we simulate a network partition which
+	// consequently will make the leader health check fail and force Carol
+	// to shut down.
+	proxy.Stop(ht.T)
+
+	// Wait for Carol to stop. If the health check wouldn't properly work
+	// this call would timeout and trigger a test failure.
+	carol.WaitForProcessExit()
+}

itest/lnd_no_etcd_dummy_failover_test.go

@@ -8,3 +8,7 @@ import "github.com/lightningnetwork/lnd/lntest"
 // testEtcdFailover is an empty itest when LND is not compiled with etcd
 // support.
 func testEtcdFailover(ht *lntest.HarnessTest) {}
+
+// testLeaderHealthCheck is an empty itest when LND is not compiled with etcd
+// support.
+func testLeaderHealthCheck(ht *lntest.HarnessTest) {}