fix(tsm1): limit concurrent WAL encodings to reduce memory pressure under heavy write load (#20814)

Co-authored-by: zhaoyun.248 <zhaoyun.248@bytedance.com>

pkg/pool/bytes.go

@@ -45,7 +45,7 @@ func (p *Bytes) Put(c []byte) {
 // LimitedBytes is a pool of byte slices that can be re-used.  Slices in
 // this pool will not be garbage collected when not in use.  The pool will
 // hold onto a fixed number of byte slices of a maximum size.  If the pool
-// is empty and max pool size has not been allocated yet, it will return a
+// is empty or the required size is larger than max size, it will return a
 // new byte slice.  Byte slices added to the pool that are over the max size
 // are dropped.
 type LimitedBytes struct {

tsdb/engine/tsm1/wal.go

@@ -112,7 +112,9 @@ type WAL struct {
 	SegmentSize int
 
 	// statistics for the WAL
-	stats   *WALStatistics
+	stats *WALStatistics
+
+	// limiter limits the max concurrency of waiting WAL writes.
 	limiter limiter.Fixed
 }
 
@@ -409,6 +411,9 @@ func (l *WAL) writeToLog(entry WALEntry) (int, error) {
 	// limit how many concurrent encodings can be in flight.  Since we can only
 	// write one at a time to disk, a slow disk can cause the allocations below
 	// to increase quickly.  If we're backed up, wait until others have completed.
+	l.limiter.Take()
+	defer l.limiter.Release()
+
 	bytes := bytesPool.Get(entry.MarshalSize())
 
 	b, err := entry.Encode(bytes)

tsdb/series_partition.go

@@ -88,7 +88,7 @@ func (p *SeriesPartition) Open() error {
 		p.index = NewSeriesIndex(p.IndexPath())
 		if err := p.index.Open(); err != nil {
 			return err
-		} else if p.index.Recover(p.segments); err != nil {
+		} else if err := p.index.Recover(p.segments); err != nil {
 			return err
 		}