test that demonstrates liveness failure on leader disk stall

If the leader's disk stalls, we can't replicate logs to followers. But the
heartbeat RPCs will still be sent concurrently, and this resets the heartbeat
timeout on the followers. This is intentional in order to avoid leadership
flapping in the face of temporary disk stalls, but has the side effect of
preventing recovery from a more persistent disk stall.

Author: tgross

raft_test.go

@@ -1018,10 +1018,10 @@ func TestRaft_RestoreSnapshotOnStartup_Monotonic(t *testing.T) {
 	conf := inmemConfig(t)
 	conf.TrailingLogs = 10
 	opts := &MakeClusterOpts{
-		Peers:         1,
-		Bootstrap:     true,
-		Conf:          conf,
-		MonotonicLogs: true,
+		Peers:               1,
+		Bootstrap:           true,
+		Conf:                conf,
+		LogstoreWrapperFunc: NewMockMonotonicLogStore,
 	}
 	c := MakeClusterCustom(t, opts)
 	defer c.Close()
@@ -1446,10 +1446,10 @@ func snapshotAndRestore(t *testing.T, offset uint64, monotonicLogStore bool, res
 	var c *cluster
 	numPeers := 3
 	optsMonotonic := &MakeClusterOpts{
-		Peers:         numPeers,
-		Bootstrap:     true,
-		Conf:          conf,
-		MonotonicLogs: true,
+		Peers:               numPeers,
+		Bootstrap:           true,
+		Conf:                conf,
+		LogstoreWrapperFunc: NewMockMonotonicLogStore,
 	}
 	if monotonicLogStore {
 		c = MakeClusterCustom(t, optsMonotonic)
@@ -2730,7 +2730,7 @@ func TestRaft_LogStoreIsMonotonic(t *testing.T) {
 
 	// Now create a new MockMonotonicLogStore using the leader logs and expect
 	// it to work.
-	log = &MockMonotonicLogStore{s: leader.logs}
+	log = NewMockMonotonicLogStore(leader.logs)
 	mcast, ok = log.(MonotonicLogStore)
 	require.True(t, ok)
 	assert.True(t, mcast.IsMonotonic())

replication_test.go

@@ -0,0 +1,147 @@
+// Copyright IBM Corp. 2013, 2026
+// SPDX-License-Identifier: MPL-2.0
+
+package raft
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLeaderHeartbeatsWithStalledDisk(t *testing.T) {
+	c := MakeClusterCustom(t, &MakeClusterOpts{
+		Peers:               3,
+		Bootstrap:           true,
+		LogstoreWrapperFunc: newBlockingLogStore,
+	})
+	t.Cleanup(c.Close)
+
+	go func() {
+		i := 0
+		ticker := time.NewTicker(500 * time.Millisecond)
+		for {
+			select {
+			case <-t.Context().Done():
+				return
+			case <-ticker.C:
+				i++
+				leader := c.Leader()
+				fut := leader.Apply(fmt.Appendf([]byte{}, "test%d", i), 0)
+				if err := fut.Error(); err != nil {
+					t.Logf("got error trying to write: %v", err)
+				} else {
+					t.Logf("write %d ok", i)
+				}
+			}
+		}
+	}()
+
+	t.Log("waiting for 5 seconds before partitioning leader")
+	time.Sleep(time.Second * 5)
+
+	oldLeader := c.Leader()
+	oldLeaderID := oldLeader.leaderID
+	oldLeaderTerm := oldLeader.getCurrentTerm()
+	c.Partition([]ServerAddress{c.Leader().localAddr})
+
+	var newLeaderTerm uint64
+	ctx, cancel := context.WithTimeout(t.Context(), 10*c.propagateTimeout)
+	t.Cleanup(cancel)
+DONE:
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatal("election didn't happen!")
+		default:
+			newLeader := c.Leader()
+			if newLeader.leaderID != oldLeaderID {
+				t.Log("leader has stepped down!")
+				newLeaderTerm = newLeader.getCurrentTerm()
+				require.NotEqual(t, newLeaderTerm, oldLeaderTerm)
+				cancel()
+				break DONE
+			}
+		}
+	}
+
+	require.Len(t, c.WaitForFollowers(1), 1)
+
+	t.Log("leader was elected. healing parition")
+
+	// reconnect the partitioned node
+	c.FullyConnect()
+	time.Sleep(3 * c.propagateTimeout)
+
+	leaderTerm := c.Leader().getCurrentTerm()
+	require.Equal(t, newLeaderTerm, leaderTerm)
+
+	t.Log("blocking disk on leader")
+
+	leader := c.Leader()
+	leaderID := leader.leaderID
+	leaderStore := leader.logs.(*blockingLogStore)
+	leaderStore.block()
+	t.Cleanup(leaderStore.unblock)
+
+	ctx, cancel = context.WithTimeout(t.Context(), 5*time.Second)
+	t.Cleanup(cancel)
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatal("leader did not step down!")
+		default:
+			if c.Leader().leaderID != leaderID {
+				t.Log("leader has stepped down!")
+				return
+			}
+		}
+	}
+}
+
+// blockingLogStore wraps a LogStore and blocks GetLog calls on demand,
+// simulating disk IO stalls.
+type blockingLogStore struct {
+	LogStore
+	mu       sync.Mutex
+	blocked  atomic.Bool
+	unblockC chan struct{}
+}
+
+func newBlockingLogStore(inner LogStore) LogStore {
+	return &blockingLogStore{
+		LogStore: inner,
+		unblockC: make(chan struct{}),
+	}
+}
+
+func (b *blockingLogStore) block() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.blocked.Store(true)
+	b.unblockC = make(chan struct{})
+}
+
+func (b *blockingLogStore) unblock() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.blocked.Load() {
+		b.blocked.Store(false)
+		close(b.unblockC)
+	}
+}
+
+func (b *blockingLogStore) GetLog(index uint64, log *Log) error {
+	if b.blocked.Load() {
+		b.mu.Lock()
+		ch := b.unblockC
+		b.mu.Unlock()
+		<-ch
+	}
+	return b.LogStore.GetLog(index, log)
+}

testing.go

@@ -136,6 +136,10 @@ type MockMonotonicLogStore struct {
 	s LogStore
 }
 
+func NewMockMonotonicLogStore(logs LogStore) LogStore {
+	return &MockMonotonicLogStore{s: logs}
+}
+
 // IsMonotonic implements the MonotonicLogStore interface.
 func (m *MockMonotonicLogStore) IsMonotonic() bool {
 	return true
@@ -717,13 +721,13 @@ WAIT:
 
 // NOTE: This is exposed for middleware testing purposes and is not a stable API
 type MakeClusterOpts struct {
-	Peers           int
-	Bootstrap       bool
-	Conf            *Config
-	ConfigStoreFSM  bool
-	MakeFSMFunc     func() FSM
-	LongstopTimeout time.Duration
-	MonotonicLogs   bool
+	Peers               int
+	Bootstrap           bool
+	Conf                *Config
+	ConfigStoreFSM      bool
+	MakeFSMFunc         func() FSM
+	LongstopTimeout     time.Duration
+	LogstoreWrapperFunc func(LogStore) LogStore
 }
 
 // makeCluster will return a cluster with the given config and number of peers.
@@ -805,8 +809,8 @@ func makeCluster(t *testing.T, opts *MakeClusterOpts) *cluster {
 		snap := c.snaps[i]
 		trans := c.trans[i]
 
-		if opts.MonotonicLogs {
-			logs = &MockMonotonicLogStore{s: logs}
+		if opts.LogstoreWrapperFunc != nil {
+			logs = opts.LogstoreWrapperFunc(logs)
 		}
 
 		peerConf := opts.Conf