Update docs/victoriametrics/changelog/CHANGELOG.md

Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com>
Signed-off-by: Nikolay <nik@victoriametrics.com>

docs/victoriametrics/changelog/CHANGELOG.md

@@ -36,6 +36,7 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): properly apply limit to metrics metadata response. See [#11139](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11139).
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): fix a possible data race when processing OpenTelemetry metadata. See [#11238](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11238). Thanks to @nevgeny for contribution.
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): flush pending persistent queue data to chunk file before updating the metadata. This prevents the metadata writer offset from getting ahead of the chunk file size and avoids losing the persistent queue after an unclean shutdown. See [#11192](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11192).
+* BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): atomically write persistent queue metainfo to prevent possible file corruption on ungraceful shutdown. See [#11192](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11192).
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): fix increased CPU and memory usage when `-remoteWrite.urlRelabelConfig` or `-remoteWrite.streamAggr.config` flags are used. The bug was introduced in [#10854](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/10854) and existed since [v1.147.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.147.0). See [#11250](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11250).
 * BUGFIX: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): preserve newline formatting in alert and rule annotations on the Alerting page. See [#11171](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11171).
 * BUGFIX: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): hide `Total metric names` stats on [Cardinality Explorer](https://docs.victoriametrics.com/victoriametrics/#cardinality-explorer) page when user selects a specific metric or label to focus. See [#11154](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11154) for details. Thanks to @lghuy05 for the contribution.

lib/persistentqueue/persistentqueue.go

@@ -648,7 +648,7 @@ func (mi *metainfo) WriteToFile(path string) error {
 	if err != nil {
 		return fmt.Errorf("cannot marshal persistent queue metainfo %#v: %w", mi, err)
 	}
-	fs.MustWriteSync(path, data)
+	fs.MustWriteAtomic(path, data, true)
 	return nil
 }
 

lib/storage/partition.go

@@ -17,7 +17,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/syncwg"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
 )
 
@@ -131,9 +130,6 @@ type partition struct {
 	// wg.Add() must be called under partsLock after checking whether stopCh isn't closed.
 	// This should prevent from calling wg.Add() after stopCh is closed and wg.Wait() is called.
 	wg sync.WaitGroup
-
-	// Use syncwg instead of sync, since Add/Wait may be called from concurrent goroutines.
-	flushPendingItemsWG syncwg.WaitGroup
 }
 
 // partWrapper is a wrapper for the part.
@@ -885,9 +881,6 @@ func (pt *partition) MustClose() {
 func (pt *partition) DebugFlush() {
 	pt.idb.tb.DebugFlush()
 	pt.flushPendingRows(true)
-
-	// Wait for background flushers to finish.
-	pt.flushPendingItemsWG.Wait()
 }
 
 func (pt *partition) startInmemoryPartsMergers() {
@@ -1018,9 +1011,7 @@ func (pt *partition) pendingRowsFlusher() {
 }
 
 func (pt *partition) flushPendingRows(isFinal bool) {
-	pt.flushPendingItemsWG.Add(1)
 	pt.rawRows.flush(pt.flushRowssToInmemoryParts, isFinal)
-	pt.flushPendingItemsWG.Done()
 }
 
 func (pt *partition) flushInmemoryRowsToFiles() {

lib/storage/search_test.go

@@ -8,7 +8,6 @@ import (
 	"reflect"
 	"regexp"
 	"sort"
-	"sync"
 	"testing"
 	"testing/quick"
 	"time"
@@ -192,67 +191,6 @@ func TestSearch_VariousTimeRanges(t *testing.T) {
 	testStorageOpOnVariousTimeRanges(t, f)
 }
 
-// TestStorageAddFlushSearchDataConcurrently verifies that concurrent goroutines
-// can read their own writes.
-//
-// This test focuses on reading the data. For reading the index see
-// TestStorageAddFlushSearchMetricNamesConcurrently in storage_test.go.
-func TestStorageAddFlushSearchDataConcurrently(t *testing.T) {
-	defer testRemoveAll(t)
-
-	s := MustOpenStorage(t.Name(), OpenOptions{})
-	defer s.MustClose()
-
-	const numMetrics = 100
-	f := func(workerID int, tr TimeRange) error {
-		mrs := make([]MetricRow, numMetrics)
-		step := (tr.MaxTimestamp - tr.MinTimestamp) / int64(numMetrics)
-		for i := range numMetrics {
-			name := fmt.Sprintf("metric_%04d_%04d", workerID, i)
-			mn := MetricName{MetricGroup: []byte(name)}
-			mrs[i].MetricNameRaw = mn.marshalRaw(nil)
-			mrs[i].Timestamp = tr.MinTimestamp + int64(i)*step
-			mrs[i].Value = float64(i)
-		}
-
-		s.AddRows(mrs, defaultPrecisionBits)
-		s.DebugFlush()
-
-		tfs := NewTagFilters()
-		re := fmt.Sprintf(`metric_%04d.*`, workerID)
-		if err := tfs.Add(nil, []byte(re), false, true); err != nil {
-			return fmt.Errorf("tfs.Add(%q) failed unexpectedly: %w", re, err)
-		}
-		return testAssertSearchResult(s, tr, tfs, mrs)
-	}
-
-	const concurrency = 20
-	var wg sync.WaitGroup
-	errs := make([]error, concurrency)
-	for workerID := range concurrency {
-		wg.Go(func() {
-			for m := time.Month(1); m <= 12; m++ {
-				tr := TimeRange{
-					MinTimestamp: time.Date(2025, m, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
-					MaxTimestamp: time.Date(2025, m+1, 0, 0, 0, 0, 0, time.UTC).UnixMilli() - 1,
-				}
-				err := f(workerID, tr)
-				if err != nil {
-					errs[workerID] = fmt.Errorf("worker %d failed on tr=%v: %w", workerID, &tr, err)
-					break
-				}
-			}
-		})
-	}
-	wg.Wait()
-
-	for _, err := range errs {
-		if err != nil {
-			t.Errorf("%s", err)
-		}
-	}
-}
-
 func testSearchInternal(s *Storage, tr TimeRange, mrs []MetricRow) error {
 	for i := range 10 {
 		// Prepare TagFilters for search.

lib/storage/storage_test.go

@@ -4324,82 +4324,3 @@ func TestStorage_futureTimestamps(t *testing.T) {
 		})
 	})
 }
-
-// TestStorageAddFlushSearchMetricNamesConcurrently verifies that concurrent
-// goroutines can read their own writes.
-//
-// This test focuses on reading the index. For reading the data see
-// TestStorageAddFlushSearchDataConcurrently in search_test.go.
-func TestStorageAddFlushSearchMetricNamesConcurrently(t *testing.T) {
-	defer testRemoveAll(t)
-
-	s := MustOpenStorage(t.Name(), OpenOptions{})
-	defer s.MustClose()
-
-	const numMetrics = 100
-	f := func(workerID int, tr TimeRange) error {
-		mrs := make([]MetricRow, numMetrics)
-		want := make([]string, numMetrics)
-		step := (tr.MaxTimestamp - tr.MinTimestamp) / int64(numMetrics)
-		for i := range numMetrics {
-			timestamp := tr.MinTimestamp + int64(i)*step
-			name := fmt.Sprintf("metric_%04d_%d", workerID, timestamp)
-			want[i] = name
-			mn := MetricName{MetricGroup: []byte(name)}
-			mrs[i].MetricNameRaw = mn.marshalRaw(nil)
-			mrs[i].Timestamp = timestamp
-		}
-
-		s.AddRows(mrs, defaultPrecisionBits)
-		s.DebugFlush()
-
-		tfs := NewTagFilters()
-		re := fmt.Sprintf(`metric_%04d.*`, workerID)
-		if err := tfs.Add(nil, []byte(re), false, true); err != nil {
-			return fmt.Errorf("tfs.Add(%q) failed unexpectedly: %w", re, err)
-		}
-		got, err := s.SearchMetricNames(nil, []*TagFilters{tfs}, tr, 1e9, noDeadline)
-		if err != nil {
-			return fmt.Errorf("SearchMetricNames(%v) failed unexpectedly: %w", tfs, err)
-		}
-		for i, name := range got {
-			var mn MetricName
-			if err := mn.UnmarshalString(name); err != nil {
-				return fmt.Errorf("Could not unmarshal metric name %q: %w", name, err)
-			}
-			got[i] = string(mn.MetricGroup)
-		}
-		slices.Sort(got)
-
-		if diff := cmp.Diff(want, got); diff != "" {
-			return fmt.Errorf("unexpected metric names (-want, +got):\n%s", diff)
-		}
-		return nil
-	}
-
-	const concurrency = 20
-	var wg sync.WaitGroup
-	errs := make([]error, concurrency)
-	for workerID := range concurrency {
-		wg.Go(func() {
-			for m := time.Month(1); m <= 12; m++ {
-				tr := TimeRange{
-					MinTimestamp: time.Date(2025, m, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
-					MaxTimestamp: time.Date(2025, m+1, 0, 0, 0, 0, 0, time.UTC).UnixMilli() - 1,
-				}
-				err := f(workerID, tr)
-				if err != nil {
-					errs[workerID] = fmt.Errorf("worker %d failed on tr=%v: %w", workerID, &tr, err)
-					break
-				}
-			}
-		})
-	}
-	wg.Wait()
-
-	for _, err := range errs {
-		if err != nil {
-			t.Errorf("%s", err)
-		}
-	}
-}