raft: lift max size from raftLog

There is no more built-in flow control in the raftLog type. This commit
moves the final piece of it out. There is only one place where this
limit is still used: hasUnappliedConfChanges check scans the log when
campaigning.

Epic: none
Release note: none

Author: pav-kv

pkg/raft/log.go

@@ -83,36 +83,20 @@ type raftLog struct {
 	applied uint64
 
 	logger raftlogger.Logger
-
-	// maxApplyingEntsSize limits the outstanding byte size of the messages
-	// returned from calls to nextCommittedEnts that have not been acknowledged
-	// by a call to appliedTo.
-	maxApplyingEntsSize entryEncodingSize
 }
 
-// newLog returns log using the given storage and default options. It
-// recovers the log to the state that it just commits and applies the
-// latest snapshot.
+// newLog returns a raft log initialized to the state in the given storage.
 func newLog(storage Storage, logger raftlogger.Logger) *raftLog {
-	return newLogWithSize(storage, logger, noLimit)
-}
-
-// newLogWithSize returns a log using the given storage and max
-// message size.
-func newLogWithSize(
-	storage Storage, logger raftlogger.Logger, maxApplyingEntsSize entryEncodingSize,
-) *raftLog {
 	compacted, lastIndex := storage.Compacted(), storage.LastIndex()
 	lastTerm, err := storage.Term(lastIndex)
 	if err != nil {
 		panic(err) // TODO(pav-kv): the storage should always cache the last term.
 	}
 	last := entryID{term: lastTerm, index: lastIndex}
 	return &raftLog{
-		storage:             storage,
-		unstable:            newUnstable(last, logger),
-		termCache:           newTermCache(termCacheSize, last),
-		maxApplyingEntsSize: maxApplyingEntsSize,
+		storage:   storage,
+		unstable:  newUnstable(last, logger),
+		termCache: newTermCache(termCacheSize, last),
 
 		// Initialize our committed and applied pointers to the time of the last
 		// compaction.

pkg/raft/log_test.go

@@ -373,7 +373,6 @@ func TestNextCommittedEnts(t *testing.T) {
 }
 
 func TestAcceptApplying(t *testing.T) {
-	maxSize := entryEncodingSize(100)
 	snap := pb.Snapshot{
 		Metadata: pb.SnapshotMetadata{Term: 1, Index: 3},
 	}
@@ -398,7 +397,7 @@ func TestAcceptApplying(t *testing.T) {
 		t.Run("", func(t *testing.T) {
 			storage := NewMemoryStorage()
 			require.NoError(t, storage.ApplySnapshot(snap))
-			raftLog := newLogWithSize(storage, raftlogger.DiscardLogger, maxSize)
+			raftLog := newLog(storage, raftlogger.DiscardLogger)
 			require.True(t, raftLog.append(init))
 			require.NoError(t, storage.Append(init.sub(3, 4)))
 			raftLog.checkInvariants(t)

pkg/raft/node_test.go

@@ -645,11 +645,8 @@ func TestCommitPagination(t *testing.T) {
 
 func TestCommitPaginationWithAsyncStorageWrites(t *testing.T) {
 	s := newTestMemoryStorage(withPeers(1))
-	cfg := newTestConfig(1, 10, 1, s)
-	cfg.MaxCommittedSizePerReady = 2048
+	rn := newTestRawNode(1, 10, 1, s)
 
-	rn, err := NewRawNode(cfg)
-	require.NoError(t, err)
 	require.NoError(t, rn.Campaign())
 
 	// Persist vote.

pkg/raft/raft.go

@@ -323,6 +323,10 @@ type raft struct {
 
 	maxMsgSize         entryEncodingSize
 	maxUncommittedSize entryPayloadSize
+	// maxCommittedPageSize limits the size of committed entries that can be
+	// loaded into memory in hasUnappliedConfChanges.
+	// TODO(#131559): avoid this loading in the first place, and remove this.
+	maxCommittedPageSize entryEncodingSize
 
 	config               quorum.Config
 	trk                  tracker.ProgressTracker
@@ -445,7 +449,7 @@ func newRaft(c *Config) *raft {
 	if err := c.validate(); err != nil {
 		panic(err.Error())
 	}
-	raftlog := newLogWithSize(c.Storage, c.Logger, entryEncodingSize(c.MaxCommittedSizePerReady))
+	raftlog := newLog(c.Storage, c.Logger)
 	hs, cs, err := c.Storage.InitialState()
 	if err != nil {
 		panic(err) // TODO(bdarnell)
@@ -457,6 +461,7 @@ func newRaft(c *Config) *raft {
 		raftLog:                     raftlog,
 		maxMsgSize:                  entryEncodingSize(c.MaxSizePerMsg),
 		maxUncommittedSize:          entryPayloadSize(c.MaxUncommittedEntriesSize),
+		maxCommittedPageSize:        entryEncodingSize(c.MaxCommittedSizePerReady),
 		lazyReplication:             c.LazyReplication,
 		electionTimeout:             c.ElectionTick,
 		electionTimeoutJitter:       c.ElectionJitterTick,
@@ -1474,13 +1479,7 @@ func (r *raft) hasUnappliedConfChanges() bool {
 	// Scan all unapplied committed entries to find a config change. Paginate the
 	// scan, to avoid a potentially unlimited memory spike.
 	lo, hi := r.raftLog.applied, r.raftLog.committed
-	// Reuse the maxApplyingEntsSize limit because it is used for similar purposes
-	// (limiting the read of unapplied committed entries) when raft sends entries
-	// via the Ready struct for application.
-	// TODO(pavelkalinnikov): find a way to budget memory/bandwidth for this scan
-	// outside the raft package.
-	pageSize := r.raftLog.maxApplyingEntsSize
-	if err := r.raftLog.scan(lo, hi, pageSize, func(ents []pb.Entry) error {
+	if err := r.raftLog.scan(lo, hi, r.maxCommittedPageSize, func(ents []pb.Entry) error {
 		for i := range ents {
 			if ents[i].Type == pb.EntryConfChange || ents[i].Type == pb.EntryConfChangeV2 {
 				found = true