raft: remove AsyncStorageWrites flag

Epic: none
Release note: none

Author: pav-kv

pkg/kv/kvserver/replica_raft.go

@@ -1079,9 +1079,10 @@ func (r *Replica) handleRaftReadyRaftMuLocked(
 	//
 	// Note that the Entries slice in the MsgStorageApply cannot refer to entries
 	// that are also in the Entries slice in the MsgStorageAppend. Raft will not
-	// allow unstable entries to be applied when AsyncStorageWrites is enabled.
+	// allow unstable entries to be applied.
+	// TODO(pav-kv): Reconsider if this can be relaxed.
 	//
-	// If we disable AsyncStorageWrites in the future, this property will no
+	// If we disable async storage writes in the future, this property will no
 	// longer be true, and the two slices could overlap. For example, this can
 	// happen when a follower is being caught up on committed commands. We could
 	// acknowledge these commands early even though they aren't durably in the

pkg/kv/kvserver/store.go

@@ -405,7 +405,6 @@ func newRaftConfig(
 	return &raft.Config{
 		ID:                          id,
 		Applied:                     uint64(appliedIndex),
-		AsyncStorageWrites:          true,
 		ElectionTick:                storeCfg.RaftElectionTimeoutTicks,
 		ElectionJitterTick:          storeCfg.RaftElectionTimeoutJitterTicks,
 		HeartbeatTick:               storeCfg.RaftHeartbeatIntervalTicks,

pkg/raft/node.go

@@ -97,6 +97,9 @@ type Ready struct {
 
 	// MustSync indicates whether the HardState and Entries must be durably
 	// written to disk or if a non-durable write is permissible.
+	//
+	// TODO(pav-kv): This flag isn't used at the moment, and the user code
+	// determines MustSync from the content of Messages. Make the API explicit.
 	MustSync bool
 }
 

pkg/raft/node_test.go

@@ -297,7 +297,6 @@ func TestNodeStart(t *testing.T) {
 		ElectionJitterTick: 10,
 		HeartbeatTick:      1,
 		Storage:            storage,
-		AsyncStorageWrites: true,
 		MaxSizePerMsg:      noLimit,
 		MaxInflightMsgs:    256,
 		StoreLiveness:      raftstoreliveness.AlwaysLive{},
@@ -369,7 +368,6 @@ func TestNodeRestart(t *testing.T) {
 		ElectionJitterTick: 10,
 		HeartbeatTick:      1,
 		Storage:            storage,
-		AsyncStorageWrites: true,
 		MaxSizePerMsg:      noLimit,
 		MaxInflightMsgs:    256,
 		StoreLiveness:      raftstoreliveness.AlwaysLive{},
@@ -422,7 +420,6 @@ func TestNodeRestartFromSnapshot(t *testing.T) {
 		ElectionJitterTick: 10,
 		HeartbeatTick:      1,
 		Storage:            s,
-		AsyncStorageWrites: true,
 		MaxSizePerMsg:      noLimit,
 		MaxInflightMsgs:    256,
 		StoreLiveness:      raftstoreliveness.AlwaysLive{},
@@ -449,7 +446,6 @@ func TestNodeAdvance(t *testing.T) {
 		ElectionJitterTick: 10,
 		HeartbeatTick:      1,
 		Storage:            storage,
-		AsyncStorageWrites: true,
 		MaxSizePerMsg:      noLimit,
 		MaxInflightMsgs:    256,
 		StoreLiveness:      raftstoreliveness.AlwaysLive{},
@@ -554,7 +550,6 @@ func TestNodeProposeAddLearnerNode(t *testing.T) {
 func TestAppendPagination(t *testing.T) {
 	const maxSizePerMsg = 2048
 	n := newNetworkWithConfig(func(c *Config) {
-		c.AsyncStorageWrites = true
 		c.MaxSizePerMsg = maxSizePerMsg
 	}, nil, nil, nil)
 

pkg/raft/raft.go

@@ -46,11 +46,11 @@ const (
 	None pb.PeerID = 0
 	// LocalAppendThread is a reference to a local thread that saves unstable
 	// log entries and snapshots to stable storage. The identifier is used as a
-	// target for MsgStorageAppend messages when AsyncStorageWrites is enabled.
+	// target for MsgStorageAppend messages.
 	LocalAppendThread pb.PeerID = math.MaxUint64
 	// LocalApplyThread is a reference to a local thread that applies committed
 	// log entries to the local state machine. The identifier is used as a
-	// target for MsgStorageApply messages when AsyncStorageWrites is enabled.
+	// target for MsgStorageApply messages.
 	LocalApplyThread pb.PeerID = math.MaxUint64 - 1
 )
 
@@ -159,9 +159,9 @@ type Config struct {
 	// threads are not responsible for understanding the response messages, only
 	// for delivering them to the correct target after performing the storage
 	// write.
-	// TODO(#129411): deprecate !AsyncStorageWrites mode as it's not used in
-	// CRDB.
-	AsyncStorageWrites bool
+	// TODO(pav-kv): this comment is a remnant of the AsyncStorageWrites option,
+	// which is now implicitly always true. Move the comment to a better place.
+
 	// LazyReplication instructs raft to hold off constructing MsgApp messages
 	// eagerly in reaction to Step() calls.
 	//
@@ -185,7 +185,8 @@ type Config struct {
 	//
 	// Despite its name (preserved for compatibility), this quota applies across
 	// Ready structs to encompass all outstanding entries in unacknowledged
-	// MsgStorageApply messages when AsyncStorageWrites is enabled.
+	// MsgStorageApply messages.
+	// TODO(pav-kv): make the name better.
 	MaxCommittedSizePerReady uint64
 	// MaxUncommittedEntriesSize limits the aggregate byte size of the
 	// uncommitted entries that may be appended to a leader's log. Once this
@@ -558,13 +559,11 @@ func (r *raft) hardState() pb.HardState {
 // next Ready handling cycle, except in one condition below.
 //
 // Certain message types are scheduled for being sent *after* the unstable state
-// is durably persisted in storage. If AsyncStorageWrites config flag is true,
-// the responsibility of upholding this condition is on the application, so the
-// message will be handed over via the next Ready as usually; if false, the
-// message will skip one Ready handling cycle, and will be sent after the
-// application has persisted the state.
-//
-// TODO(pav-kv): remove this special case after !AsyncStorageWrites is removed.
+// is durably persisted in storage. These messages are nevertheless included in
+// Ready.Messages, and the responsibility of upholding this condition is on the
+// application.
+// TODO(pav-kv): make this requirement explicit in the API, instead of mixing
+// the two kinds of messages together.
 func (r *raft) send(m pb.Message) {
 	if m.From == None {
 		m.From = r.id

pkg/raft/raft_test.go

@@ -5455,7 +5455,6 @@ func newTestConfig(
 		ElectionJitterTick: election,
 		HeartbeatTick:      heartbeat,
 		Storage:            storage,
-		AsyncStorageWrites: true,
 		MaxSizePerMsg:      noLimit,
 		MaxInflightMsgs:    256,
 		StoreLiveness:      storeLiveness,

pkg/raft/rafttest/interaction_env_handler_add_nodes.go

@@ -35,6 +35,10 @@ func (env *InteractionEnv) handleAddNodes(t *testing.T, d datadriven.TestData) e
 	n := firstAsInt(t, d)
 	var snap pb.Snapshot
 	cfg := raftConfigStub()
+	// NB: the datadriven tests use the async storage API, but have an option to
+	// sync writes immediately in imitation of the previous synchronous API.
+	var asyncWrites bool
+
 	for _, arg := range d.CmdArgs[1:] {
 		for i := range arg.Vals {
 			switch arg.Key {
@@ -56,7 +60,7 @@ func (env *InteractionEnv) handleAddNodes(t *testing.T, d datadriven.TestData) e
 			case "content":
 				arg.Scan(t, i, &snap.Data)
 			case "async-storage-writes":
-				arg.Scan(t, i, &cfg.AsyncStorageWrites)
+				arg.Scan(t, i, &asyncWrites)
 			case "lazy-replication":
 				arg.Scan(t, i, &cfg.LazyReplication)
 			case "prevote":
@@ -81,7 +85,7 @@ func (env *InteractionEnv) handleAddNodes(t *testing.T, d datadriven.TestData) e
 			}
 		}
 	}
-	return env.AddNodes(n, cfg, snap)
+	return env.AddNodes(n, cfg, snap, asyncWrites)
 }
 
 type snapOverrideStorage struct {
@@ -100,7 +104,9 @@ var _ raft.Storage = snapOverrideStorage{}
 
 // AddNodes adds n new nodes initialized from the given snapshot (which may be
 // empty), and using the cfg as template. They will be assigned consecutive IDs.
-func (env *InteractionEnv) AddNodes(n int, cfg raft.Config, snap pb.Snapshot) error {
+func (env *InteractionEnv) AddNodes(
+	n int, cfg raft.Config, snap pb.Snapshot, asyncWrites bool,
+) error {
 	bootstrap := !reflect.DeepEqual(snap, pb.Snapshot{})
 	for i := 0; i < n; i++ {
 		id := pb.PeerID(1 + len(env.Nodes))
@@ -144,11 +150,6 @@ func (env *InteractionEnv) AddNodes(n int, cfg raft.Config, snap pb.Snapshot) er
 			cfg.CRDBVersion = cluster.MakeTestingClusterSettings().Version
 		}
 
-		// NB: the datadriven tests use the async storage API, but have an option to
-		// sync writes immediately in imitation of the previous synchronous API.
-		asyncWrites := cfg.AsyncStorageWrites
-		cfg.AsyncStorageWrites = true
-
 		cfg.StoreLiveness = newStoreLiveness(env.Fabric, id)
 
 		cfg.Metrics = raft.NewMetrics()

pkg/raft/rafttest/interaction_env_handler_process_ready.go

@@ -71,9 +71,6 @@ func (env *InteractionEnv) ProcessReady(idx int) error {
 		return nil
 	}
 
-	if !n.Config.AsyncStorageWrites {
-		panic("expected AsyncStorageWrites")
-	}
 	env.Output.WriteString(raft.DescribeReady(rd, defaultEntryFormatter))
 
 	for _, m := range rd.Messages {

pkg/raft/rafttest/node.go

@@ -53,7 +53,6 @@ func startNode(id raftpb.PeerID, peers []raft.Peer, iface iface) *node {
 		ElectionJitterTick:        50,
 		HeartbeatTick:             1,
 		Storage:                   st,
-		AsyncStorageWrites:        true,
 		MaxSizePerMsg:             1024 * 1024,
 		MaxInflightMsgs:           256,
 		MaxUncommittedEntriesSize: 1 << 30,
@@ -216,7 +215,6 @@ func (n *node) restart() {
 		ElectionJitterTick:        10,
 		HeartbeatTick:             1,
 		Storage:                   n.storage,
-		AsyncStorageWrites:        true,
 		MaxSizePerMsg:             1024 * 1024,
 		MaxInflightMsgs:           256,
 		MaxUncommittedEntriesSize: 1 << 30,

pkg/raft/rawnode.go

@@ -52,9 +52,6 @@ func NewRawNode(config *Config) (*RawNode, error) {
 	rn := &RawNode{
 		raft: r,
 	}
-	if !config.AsyncStorageWrites {
-		panic("synchronous storage writes are no longer supported")
-	}
 	ss := r.softState()
 	rn.prevSoftSt = &ss
 	rn.prevHardSt = r.hardState()
@@ -267,7 +264,9 @@ func (rn *RawNode) Ready() Ready {
 
 // MustSync returns true if the hard state and count of Raft entries indicate
 // that a synchronous write to persistent storage is required.
-// NOTE: MustSync isn't used under AsyncStorageWrites mode.
+//
+// TODO(pav-kv): MustSync isn't used, because all writes are asynchronous.
+// Remove this, or repurpose it to fit the asynchronous writes API.
 func MustSync(st, prevst pb.HardState, entsnum int) bool {
 	// Persistent state on all servers:
 	// (Updated on stable storage before responding to RPCs)
@@ -299,8 +298,7 @@ func needStorageAppendRespMsg(rd Ready) bool {
 // newStorageAppendMsg creates the message that should be sent to the local
 // append thread to instruct it to append log entries, write an updated hard
 // state, and apply a snapshot. The message also carries a set of responses
-// that should be delivered after the rest of the message is processed. Used
-// with AsyncStorageWrites.
+// that should be delivered after the rest of the message is processed.
 func newStorageAppendMsg(r *raft, rd Ready) pb.Message {
 	m := pb.Message{
 		Type:    pb.MsgStorageAppend,
@@ -429,7 +427,7 @@ func needStorageApplyMsg(rd Ready) bool { return len(rd.CommittedEntries) > 0 }
 // newStorageApplyMsg creates the message that should be sent to the local
 // apply thread to instruct it to apply committed log entries. The message
 // also carries a response that should be delivered after the rest of the
-// message is processed. Used with AsyncStorageWrites.
+// message is processed.
 func newStorageApplyMsg(r *raft, rd Ready) pb.Message {
 	ents := rd.CommittedEntries
 	return pb.Message{