Raft batching

server/filestore.go

@@ -8143,6 +8143,10 @@ func fileStoreMsgSizeEstimate(slen, maxPayload int) uint64 {
 	return uint64(emptyRecordLen + slen + 4 + maxPayload)
 }
 
+func (fs *fileStore) MsgSize(msg []byte) uint64 {
+	return fileStoreMsgSizeRaw(0, 0, len(msg))
+}
+
 // ResetState resets any state that's temporary. For example when changing leaders.
 func (fs *fileStore) ResetState() {
 	fs.mu.Lock()

server/jetstream_cluster_1_test.go

@@ -6973,7 +6973,7 @@ func TestJetStreamClusterCatchupMustStallWhenBehindOnApplies(t *testing.T) {
 	n := mset.node.(*raft)
 	n.Lock()
 	ae := n.buildAppendEntry(nil)
-	err = n.storeToWAL(ae)
+	_, _, err = n.storeToWAL(ae)
 	n.Unlock()
 	index, commit, applied := n.Progress()
 	require_NoError(t, err)

server/jetstream_cluster_4_test.go

@@ -4269,7 +4269,7 @@ func TestJetStreamClusterConsumerDontSendSnapshotOnLeaderChange(t *testing.T) {
 	rn.Lock()
 	entries := []*Entry{{EntryNormal, updateDeliveredBuffer()}, {EntryNormal, updateAcksBuffer()}}
 	ae := encode(t, rn.buildAppendEntry(entries))
-	err = rn.storeToWAL(ae)
+	_, _, err = rn.storeToWAL(ae)
 	minPindex := rn.pindex
 	rn.Unlock()
 	require_NoError(t, err)

server/memstore.go

@@ -2033,6 +2033,10 @@ func memStoreMsgSize(subj string, hdr, msg []byte) uint64 {
 	return memStoreMsgSizeRaw(len(subj), len(hdr), len(msg))
 }
 
+func (ms *memStore) MsgSize(msg []byte) uint64 {
+	return memStoreMsgSizeRaw(0, 0, len(msg))
+}
+
 // ResetState resets any state that's temporary. For example when changing leaders.
 func (ms *memStore) ResetState() {
 	ms.mu.Lock()