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

@@ -1,4 +1,4 @@
-// Copyright IBM Corp. 2013, 2025
+// Copyright IBM Corp. 2013, 2026
 // SPDX-License-Identifier: MPL-2.0
 
 package raft
@@ -1021,10 +1021,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()
@@ -1642,10 +1642,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)
@@ -2926,7 +2926,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

@@ -1,4 +1,4 @@
-// Copyright IBM Corp. 2013, 2025
+// Copyright IBM Corp. 2013, 2026
 // SPDX-License-Identifier: MPL-2.0
 
 package raft
@@ -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,15 +721,15 @@ 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
-	CommitTrackingLogs bool
-	PropagateError     bool // If true, return errors instead of calling t.Fatal
+	Peers               int
+	Bootstrap           bool
+	Conf                *Config
+	ConfigStoreFSM      bool
+	MakeFSMFunc         func() FSM
+	LongstopTimeout     time.Duration
+	LogstoreWrapperFunc func(LogStore) LogStore
+	CommitTrackingLogs  bool
+	PropagateError      bool // If true, return errors instead of calling t.Fatal
 }
 
 // makeCluster will return a cluster with the given config and number of peers.
@@ -810,8 +814,8 @@ func makeCluster(t *testing.T, opts *MakeClusterOpts) (*cluster, error) {
 		snap := c.snaps[i]
 		trans := c.trans[i]
 
-		if opts.MonotonicLogs {
-			logs = &MockMonotonicLogStore{s: logs}
+		if opts.LogstoreWrapperFunc != nil {
+			logs = opts.LogstoreWrapperFunc(logs)
 		} else if opts.CommitTrackingLogs {
 			logs = NewInmemCommitTrackingStore()
 		}