fix(metastore): disable raft pipeline replication

pkg/metastore/raftnode/node.go

@@ -73,6 +73,23 @@ const (
 	defaultTransportConnPoolSize = 10
 	defaultTransportTimeout      = 10 * time.Second
 	defaultLogStoreTimeout       = 10 * time.Second
+
+	// Anything below raft's minInFlightForPipelining (2) makes
+	// AppendEntriesPipeline return ErrPipelineReplicationNotSupported, which
+	// replicate() handles by staying in synchronous RPC mode.
+	//
+	// Pipelining is unusable on raft v1.7.3: it sizes both netPipeline
+	// channels at MaxRPCsInFlight-2, so the default of 2 leaves them
+	// unbuffered, and a follower that fails a single AppendEntries can
+	// deadlock replication to itself permanently. See
+	// TestRaftPipelineDeadlock for the full interlock. Raising this instead
+	// of lowering it would only widen the window, since both channels stay
+	// bounded. Revert once the upstream bug is fixed.
+	//
+	// The cost is small: replication still ships up to MaxAppendEntries
+	// entries per round trip, and a single entry commits in one round trip
+	// either way.
+	raftMaxRPCsInFlight = 1
 )
 
 func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
@@ -160,11 +177,7 @@ func NewNode(
 	if err != nil {
 		return nil, err
 	}
-	n.transport, err = raft.NewTCPTransport(
-		config.BindAddress, addr,
-		int(config.TransportConnPoolSize),
-		config.TransportTimeout,
-		os.Stderr)
+	n.transport, err = newTransport(config, addr)
 	if err != nil {
 		return nil, err
 	}
@@ -176,6 +189,18 @@ func NewNode(
 	return &n, nil
 }
 
+// newTransport creates the raft TCP transport. A nil Logger makes raft build
+// the same "raft-net" hclog on stderr that the plain NewTCPTransport
+// constructor would.
+func newTransport(config Config, advertise net.Addr) (*raft.NetworkTransport, error) {
+	return raft.NewTCPTransportWithConfig(config.BindAddress, advertise,
+		&raft.NetworkTransportConfig{
+			MaxPool:         int(config.TransportConnPoolSize),
+			Timeout:         config.TransportTimeout,
+			MaxRPCsInFlight: raftMaxRPCsInFlight,
+		})
+}
+
 func (n *Node) Init() (err error) {
 	raftConfig := raft.DefaultConfig()
 	// TODO: Wrap gokit

pkg/metastore/raftnode/transport_test.go

@@ -0,0 +1,31 @@
+package raftnode
+
+import (
+	"net"
+	"testing"
+
+	"github.com/hashicorp/raft"
+	"github.com/stretchr/testify/require"
+)
+
+// TestTransport_PipeliningDisabled pins the workaround in newTransport.
+// Raft's pipeline replication deadlocks on v1.7.3 (see the
+// raftMaxRPCsInFlight comment, and TestRaftPipelineDeadlock in
+// pkg/metastore/test for the end-to-end reproduction), so the transport must
+// refuse to pipeline and leave replicate() in synchronous RPC mode.
+func TestTransport_PipeliningDisabled(t *testing.T) {
+	advertise, err := net.ResolveTCPAddr("tcp", "localhost:0")
+	require.NoError(t, err)
+
+	transport, err := newTransport(Config{
+		BindAddress:           "localhost:0",
+		TransportConnPoolSize: defaultTransportConnPoolSize,
+		TransportTimeout:      defaultTransportTimeout,
+	}, advertise)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, transport.Close()) })
+
+	// Raft checks maxInFlight before dialing, so no peer has to exist.
+	_, err = transport.AppendEntriesPipeline("peer", transport.LocalAddr())
+	require.ErrorIs(t, err, raft.ErrPipelineReplicationNotSupported)
+}

pkg/metastore/test/create.go

@@ -3,6 +3,7 @@ package test
 import (
 	"context"
 	"fmt"
+	"net"
 	"testing"
 	"time"
 
@@ -35,9 +36,12 @@ func NewMetastoreSet(t *testing.T, cfg *metastore.Config, n int, bucket objstore
 	raftAddresses := make([]string, n)
 	raftIds := make([]string, n)
 	bootstrapPeers := make([]string, n)
+	raftPorts := freeLocalPorts(t, n)
 	for i := 0; i < n; i++ {
+		// gRPC runs over in-memory listeners keyed by this string, so it is
+		// only a label and never bound.
 		grpcAddresses[i] = fmt.Sprintf("localhost:%d", 10500+i)
-		raftAddresses[i] = fmt.Sprintf("localhost:%d", 10500+2*i)
+		raftAddresses[i] = fmt.Sprintf("localhost:%d", raftPorts[i])
 		raftIds[i] = fmt.Sprintf("node-%d", i)
 		bootstrapPeers[i] = fmt.Sprintf("%s/%s", raftAddresses[i], raftIds[i])
 	}
@@ -141,6 +145,33 @@ func NewMetastoreSet(t *testing.T, cfg *metastore.Config, n int, bucket objstore
 	return res
 }
 
+// freeLocalPorts picks n ports that are free right now. Raft needs real TCP
+// listeners and every peer address has to be known before any node starts, so
+// they cannot simply be bound as :0. Fixed ports are not an option either:
+// this helper is used from more than one package, and go test runs packages
+// in parallel, so they would intermittently collide with "address already in
+// use".
+func freeLocalPorts(t *testing.T, n int) []int {
+	t.Helper()
+	ports := make([]int, n)
+	listeners := make([]*net.TCPListener, n)
+	for i := range ports {
+		addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
+		require.NoError(t, err)
+		l, err := net.ListenTCP("tcp", addr)
+		require.NoError(t, err)
+		listeners[i] = l
+		ports[i] = l.Addr().(*net.TCPAddr).Port
+	}
+	// Every listener stays open until all ports have been picked, so the OS
+	// cannot hand out the same one twice; they are only released once the
+	// full set is known.
+	for _, l := range listeners {
+		require.NoError(t, l.Close())
+	}
+	return ports
+}
+
 func MockStaticDiscovery(t *testing.T, servers []discovery.Server) *mockdiscovery.MockDiscovery {
 	d := mockdiscovery.NewMockDiscovery(t)
 	d.On("Subscribe", mock.Anything).Run(func(args mock.Arguments) {