From cf9690af0e560471e6af943b8ce0be8fe39d2d6c Mon Sep 17 00:00:00 2001 From: Aleksandar Petrov <8142643+aleks-p@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:55:00 -0300 Subject: [PATCH 1/2] fix(metastore): disable raft pipeline replication On raft v1.7.3 a single failed AppendEntries can deadlock the leader's pipeline replication to one follower, permanently. The transport sizes both netPipeline channels at MaxRPCsInFlight-2, so the default of 2 leaves them unbuffered: when a follower answers Success=false, pipelineDecode returns and closes finishCh, leaving doneCh with no reader; the SEND loop then races between the closed finishCh and a ready triggerCh, and if it picks triggerCh it issues one more send that parks on inprogressCh while decodeResponses parks on doneCh. pipeline.Close is deferred inside pipelineReplicate, which is stuck in that select, so shutdownCh never closes. replicate() never returns either, so its deferred close(stopHeartbeat) never runs: heartbeats keep flowing, the follower stays a quiet Follower that never forces an election, and it is simply never replicated to again. It fails readiness on ErrLagBehind until the leader is restarted -- restarting the follower does not help, because the wedge is on the leader. The follower's own timeoutLogStore is one way to produce the triggering Success=false, since raft's appendEntries handler returns without setting Success when StoreLogs fails. Pinning MaxRPCsInFlight below raft's minInFlightForPipelining makes AppendEntriesPipeline return ErrPipelineReplicationNotSupported, which replicate() handles by staying in synchronous RPC mode. Raising it instead would only widen the window, since both channels stay bounded. --- pkg/metastore/raftnode/node.go | 35 +- pkg/metastore/raftnode/transport_test.go | 31 ++ .../test/metastore_pipeline_deadlock_test.go | 336 ++++++++++++++++++ 3 files changed, 397 insertions(+), 5 deletions(-) create mode 100644 pkg/metastore/raftnode/transport_test.go create mode 100644 pkg/metastore/test/metastore_pipeline_deadlock_test.go diff --git a/pkg/metastore/raftnode/node.go b/pkg/metastore/raftnode/node.go index 5e30164271..8ed8f81e38 100644 --- a/pkg/metastore/raftnode/node.go +++ b/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 diff --git a/pkg/metastore/raftnode/transport_test.go b/pkg/metastore/raftnode/transport_test.go new file mode 100644 index 0000000000..35315102b7 --- /dev/null +++ b/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) +} diff --git a/pkg/metastore/test/metastore_pipeline_deadlock_test.go b/pkg/metastore/test/metastore_pipeline_deadlock_test.go new file mode 100644 index 0000000000..03b389df6c --- /dev/null +++ b/pkg/metastore/test/metastore_pipeline_deadlock_test.go @@ -0,0 +1,336 @@ +package test + +import ( + "context" + "crypto/rand" + "errors" + "os" + "path/filepath" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/grafana/dskit/flagext" + "github.com/hashicorp/raft" + "github.com/oklog/ulid/v2" + "github.com/stretchr/testify/require" + + metastorev1 "github.com/grafana/pyroscope/api/gen/proto/go/metastore/v1" + "github.com/grafana/pyroscope/v2/pkg/metastore" + "github.com/grafana/pyroscope/v2/pkg/metastore/raftnode/raftnodepb" + "github.com/grafana/pyroscope/v2/pkg/objstore/providers/memory" +) + +// TestRaftPipelineDeadlock covers the workaround pinned by +// raftMaxRPCsInFlight: on raft v1.7.3, a brief log store stall on a follower +// can deadlock the leader's pipeline replication to that follower forever. +// +// In raft v1.7.3 the network transport sizes both netPipeline channels at +// MaxRPCsInFlight-2, so the default of 2 leaves them unbuffered. Given a +// follower that fails an AppendEntries, five things happen in order: +// +// 1. decodeResponses takes future #1 off inprogressCh and blocks decoding. +// 2. pipelineSend issues #2, which blocks handing off to inprogressCh. +// 3. #1's response comes back with Success=false, so Raft.pipelineDecode +// returns and closes finishCh. doneCh now has no reader. +// 4. decodeResponses loops, takes #2 (unblocking the sender), and blocks +// decoding it. +// 5. The SEND loop selects between the closed finishCh and a ready +// triggerCh. If it picks triggerCh it issues #3, which parks on +// inprogressCh; #2's response then parks decodeResponses on doneCh. +// +// pipeline.Close is deferred inside pipelineReplicate, which is stuck in step +// 5, so shutdownCh never closes and neither goroutine escapes. replicate() +// never returns either, so its deferred close(stopHeartbeat) never runs: +// heartbeats to the victim keep flowing, it stays a quiet Follower that never +// forces an election, and it is simply never replicated to again. The stalled +// replica then fails readiness on ErrLagBehind indefinitely. Restarting it +// does not help, because the wedge is on the leader. +// +// Step 5 is a select race, so the fault is applied over several rounds. +// Measured at roughly nine times in ten per round; each round needs the peer +// healthy first, since raft only re-enters pipeline mode after a successful +// replicateTo sets allowPipeline. Hence a fault that rejects writes rather +// than delaying them — see failInjector. +// +// The assertion is that the follower catches up again once its log store is +// healthy, which is what readiness depends on. The goroutine stacks are only +// consulted to explain a failure. +func TestRaftPipelineDeadlock(t *testing.T) { + const ( + clusterSize = 3 + writers = 8 + rounds = 4 + failFor = time.Second + recoverIn = 15 * time.Second + // Mirrors a raft.log-store-timeout well under raft.transport-timeout, + // which is the ordering that produces this failure in production. + failAfter = 200 * time.Millisecond + ) + + faults := make([]*failInjector, clusterSize) + for i := range faults { + faults[i] = &failInjector{delay: failAfter} + } + + cfg := new(metastore.Config) + flagext.DefaultValues(cfg) + + var mu sync.Mutex + nodeIdx := 0 + cfg.Raft.LogStoreMiddleware = func(store raft.LogStore) raft.LogStore { + mu.Lock() + i := nodeIdx + nodeIdx++ + mu.Unlock() + faults[i].store = store + return faults[i] + } + + ms := NewMetastoreSet(t, cfg, clusterSize, memory.NewInMemBucket()) + defer func() { + for _, f := range faults { + f.clear() + } + // Shutting down a raft node whose replication goroutine is wedged + // blocks forever, which would turn a reproduced failure into an + // unhelpful test binary timeout. + done := make(chan struct{}) + go func() { + defer close(done) + ms.Close() + }() + select { + case <-done: + case <-time.After(20 * time.Second): + t.Log("teardown timed out; raft shutdown is blocked on the wedged node") + } + }() + + leaderIdx := findLeader(t, ms) + followerIdx := (leaderIdx + 1) % clusterSize + t.Logf("leader=node-%d victim follower=node-%d", leaderIdx, followerIdx) + + writer := newBlockWriter(ms, leaderIdx, writers) + defer writer.stop() + + // Sustained writes push every follower through a successful replicateTo, + // which sets allowPipeline and moves replication into pipeline mode. + // Without this the leader is still in RPC mode when we stall, and the + // code under test never runs. + writer.start() + require.Eventually(t, func() bool { + return followerCaughtUp(t, ms, leaderIdx, followerIdx) + }, 20*time.Second, 250*time.Millisecond, "follower did not catch up with the leader") + t.Log("cluster healthy; replication would be pipelining if it were enabled") + + for round := 1; round <= rounds; round++ { + faults[followerIdx].fail() + time.Sleep(failFor) + faults[followerIdx].clear() + + // The follower's log store is healthy again and writes keep flowing, + // so a follower that is merely behind catches up in well under a + // second. One the leader has stopped replicating to never will. + if !awaitCatchUp(t, ms, leaderIdx, followerIdx, recoverIn) { + if deadlocked, dump := pipelineDeadlockStacks(); deadlocked { + t.Logf("round %d: raft pipeline replication is deadlocked\n%s", round, dump) + } else { + t.Logf("round %d: follower is not catching up, but the raft pipeline "+ + "goroutines are not in the expected interlock", round) + } + t.Logf("full goroutine dump written to %s", writeGoroutineDump(t)) + logLag(t, ms, leaderIdx, followerIdx, "victim") + t.Fatal("the leader has permanently stopped replicating to the follower") + } + } +} + +// awaitCatchUp waits for the follower's applied index to reach the leader's +// commit index, which is the same condition Metastore.CheckReady enforces and +// therefore what decides pod readiness in a real cell. +func awaitCatchUp(t *testing.T, ms MetastoreSet, leaderIdx, followerIdx int, within time.Duration) bool { + t.Helper() + deadline := time.Now().Add(within) + for time.Now().Before(deadline) { + if followerCaughtUp(t, ms, leaderIdx, followerIdx) { + return true + } + time.Sleep(100 * time.Millisecond) + } + return false +} + +func followerCaughtUp(t *testing.T, ms MetastoreSet, leaderIdx, followerIdx int) bool { + t.Helper() + leader := pipelineNodeInfo(t, ms, leaderIdx) + follower := pipelineNodeInfo(t, ms, followerIdx) + if leader == nil || follower == nil { + return false + } + return follower.GetAppliedIndex() >= leader.GetCommitIndex() +} + +func logLag(t *testing.T, ms MetastoreSet, leaderIdx, followerIdx int, what string) { + t.Helper() + leader := pipelineNodeInfo(t, ms, leaderIdx) + follower := pipelineNodeInfo(t, ms, followerIdx) + t.Logf("%s: leader commit=%d, follower applied=%d, lag=%d entries", + what, leader.GetCommitIndex(), follower.GetAppliedIndex(), + leader.GetCommitIndex()-follower.GetAppliedIndex()) +} + +func pipelineNodeInfo(t *testing.T, ms MetastoreSet, idx int) *raftnodepb.NodeInfo { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + resp, err := ms.Instances[idx].NodeInfo(ctx, &raftnodepb.NodeInfoRequest{}) + if err != nil { + return nil + } + return resp.GetNode() +} + +// pipelineDeadlockStacks looks for the interlock described above: +// netPipeline.AppendEntries parked on the inprogressCh handoff while +// netPipeline.decodeResponses is parked on the doneCh handoff to a +// pipelineDecode that has already returned. Requiring both halves keeps this +// from firing on a merely slow pipeline. +// +// Both sites are channel sends inside a select, which the runtime reports as +// "[select]" rather than "[chan send]", so the frames are what identify them. +// A healthy decodeResponses is distinguishable because it sits inside +// decodeResponse reading from the connection, not in its own select. +func pipelineDeadlockStacks() (bool, string) { + buf := make([]byte, 1<<24) + buf = buf[:runtime.Stack(buf, true)] + + var sendBlocked, decodeBlocked []string + for g := range strings.SplitSeq(string(buf), "\n\n") { + if !strings.Contains(g, "[select]") { + continue + } + switch { + case strings.Contains(g, "netPipeline).AppendEntries"): + sendBlocked = append(sendBlocked, g) + case strings.Contains(g, "netPipeline).decodeResponses") && + !strings.Contains(g, "raft.decodeResponse("): + decodeBlocked = append(decodeBlocked, g) + } + } + if len(sendBlocked) == 0 || len(decodeBlocked) == 0 { + return false, "" + } + return true, strings.Join(append(sendBlocked, decodeBlocked...), "\n\n") +} + +func writeGoroutineDump(t *testing.T) string { + t.Helper() + buf := make([]byte, 1<<24) + buf = buf[:runtime.Stack(buf, true)] + path := filepath.Join(t.TempDir(), "goroutines.txt") + if err := os.WriteFile(path, buf, 0o644); err != nil { + t.Logf("could not write goroutine dump: %v", err) + return "" + } + return path +} + +// blockWriter keeps a steady stream of concurrent AddBlock calls going +// through the leader. +type blockWriter struct { + ms MetastoreSet + leaderIdx int + writers int + seq atomic.Uint64 + cancel context.CancelFunc + wg sync.WaitGroup +} + +func newBlockWriter(ms MetastoreSet, leaderIdx, writers int) *blockWriter { + return &blockWriter{ms: ms, leaderIdx: leaderIdx, writers: writers} +} + +func (w *blockWriter) start() { + ctx, cancel := context.WithCancel(context.Background()) + w.cancel = cancel + for i := 0; i < w.writers; i++ { + w.wg.Go(func() { + for ctx.Err() == nil { + callCtx, callCancel := context.WithTimeout(ctx, 5*time.Second) + _, _ = w.ms.Instances[w.leaderIdx].AddBlock(callCtx, &metastorev1.AddBlockRequest{ + Block: &metastorev1.BlockMeta{ + Id: ulid.MustNew(w.seq.Add(1), rand.Reader).String(), + }, + }) + callCancel() + } + }) + } +} + +func (w *blockWriter) stop() { + if w.cancel == nil { + return + } + w.cancel() + w.wg.Wait() + w.cancel = nil +} + +// failInjector wraps a raft.LogStore so writes can be made slow and then +// rejected, without ever touching the underlying store. +// +// Both halves matter, and for different reasons. +// +// Rejecting rather than eventually writing keeps the follower's log +// consistent, so it catches up as soon as the fault clears and each round +// starts from a healthy peer. A fault that merely delays the write would land +// it out of band and livelock the follower on "non-monotonic log entries" — +// a different bug (#5307) that would both mask this one and make every round +// after the first meaningless. +// +// The delay is what actually creates the deadlock window. It has to be long +// enough that decodeResponses is still busy with one response when the next +// send arrives, so that send parks on inprogressCh; failing instantly instead +// lets every handoff complete and the interlock never forms. In production +// that delay is supplied by timeoutLogStore, which fails a stalled write after +// raft.log-store-timeout. It must stay well below the transport timeout: if +// the leader's read deadline fired first, decodeResponse would call +// conn.Release() and the next send would fail fast on a closed connection, +// letting the leader unwind cleanly. +type failInjector struct { + store raft.LogStore + failing atomic.Bool + delay time.Duration +} + +func (f *failInjector) fail() { f.failing.Store(true) } +func (f *failInjector) clear() { f.failing.Store(false) } + +func (f *failInjector) FirstIndex() (uint64, error) { return f.store.FirstIndex() } +func (f *failInjector) LastIndex() (uint64, error) { return f.store.LastIndex() } +func (f *failInjector) GetLog(idx uint64, log *raft.Log) error { return f.store.GetLog(idx, log) } +func (f *failInjector) DeleteRange(min, max uint64) error { return f.store.DeleteRange(min, max) } + +func (f *failInjector) IsMonotonic() bool { + if m, ok := f.store.(raft.MonotonicLogStore); ok { + return m.IsMonotonic() + } + return false +} + +func (f *failInjector) StoreLog(log *raft.Log) error { + return f.StoreLogs([]*raft.Log{log}) +} + +func (f *failInjector) StoreLogs(logs []*raft.Log) error { + if f.failing.Load() { + time.Sleep(f.delay) + return errors.New("simulated slow log store failure") + } + return f.store.StoreLogs(logs) +} From 8a3ada442518fae89f1dccb26c8c97f9bc744043 Mon Sep 17 00:00:00 2001 From: Aleksandar Petrov <8142643+aleks-p@users.noreply.github.com> Date: Fri, 14 Aug 2026 15:54:21 -0300 Subject: [PATCH 2/2] test(metastore): allocate free ports in the metastore test harness NewMetastoreSet hardcoded raft ports 10500/10502/10504. It is used from both pkg/metastore/test and pkg/segmentwriter, and go test runs packages in parallel, so the two could bind the same ports and fail with "address already in use". That was rare while every caller was short-lived. TestRaftPipelineDeadlock holds the ports long enough to make it reliable, so it surfaced as a consistent CI failure in pkg/segmentwriter rather than an occasional flake. --- pkg/metastore/test/create.go | 33 ++++++++++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/pkg/metastore/test/create.go b/pkg/metastore/test/create.go index 9dd662bd21..f170dbd80f 100644 --- a/pkg/metastore/test/create.go +++ b/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) {