app/vmagent: add new flag to prioritize recently ingested data

This commit adds a new flag remoteWrite.inmemoryQueueWorkers. Which
 dedicates set of workers for processing only
in-memory part of vmagent persistentqueue. It should help to mitigate
an issue when stale data at file-based queue prevents from ingestion
recent data.

 There is a downside in case of remote storage is not reachable,
it's possible to get only a part of data ingested and other part queued
into file-based queue. But it should be acceptable, because there is no
strong guarantees for the data ingestion order.

Fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8833

.github/workflows/codeql-analysis-go.yml

@@ -52,14 +52,14 @@ jobs:
           restore-keys: go-artifacts-${{ runner.os }}-codeql-analyze-${{ steps.go.outputs.go-version }}-
 
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@7211b7c8077ea37d8641b6271f6a365a22a5fbfa  # v4.36.0
+        uses: github/codeql-action/init@e46ed2cbd01164d986452f91f178727624ae40d7  # v4.35.3
         with:
           languages: go
 
       - name: Autobuild
-        uses: github/codeql-action/autobuild@7211b7c8077ea37d8641b6271f6a365a22a5fbfa  # v4.36.0
+        uses: github/codeql-action/autobuild@e46ed2cbd01164d986452f91f178727624ae40d7  # v4.35.3
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@7211b7c8077ea37d8641b6271f6a365a22a5fbfa  # v4.36.0
+        uses: github/codeql-action/analyze@e46ed2cbd01164d986452f91f178727624ae40d7  # v4.35.3
         with:
           category: 'language:go'

app/vmagent/remotewrite/client.go

@@ -204,9 +204,13 @@ func (c *client) init(argIdx, concurrency int, sanitizedURL string) {
 	c.packetsDropped = metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_packets_dropped_total{url=%q}`, c.sanitizedURL))
 	c.retriesCount = metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_retries_count_total{url=%q}`, c.sanitizedURL))
 	c.sendDuration = metrics.GetOrCreateFloatCounter(fmt.Sprintf(`vmagent_remotewrite_send_duration_seconds_total{url=%q}`, c.sanitizedURL))
+	inmemoryWorkers := inmemoryQueueWorkers.GetOptionalArg(argIdx)
 	metrics.GetOrCreateGauge(fmt.Sprintf(`vmagent_remotewrite_queues{url=%q}`, c.sanitizedURL), func() float64 {
-		return float64(concurrency)
+		return float64(concurrency + inmemoryWorkers)
 	})
+	for range inmemoryWorkers {
+		c.wg.Go(c.runWorkerForInmemoryQueue)
+	}
 	for range concurrency {
 		c.wg.Go(c.runWorker)
 	}
@@ -348,6 +352,52 @@ func (c *client) runWorker() {
 	}
 }
 
+func (c *client) runWorkerForInmemoryQueue() {
+	var ok bool
+	var block []byte
+	ch := make(chan bool, 1)
+	for {
+		block, ok = c.fq.MustReadInMemoryBlockBlocking(block[:0])
+		if !ok {
+			return
+		}
+		go func() {
+			startTime := time.Now()
+			ch <- c.sendBlock(block)
+			c.sendDuration.Add(time.Since(startTime).Seconds())
+		}()
+		select {
+		case ok := <-ch:
+			if ok {
+				// The block has been sent successfully
+				continue
+			}
+			// Return unsent block to the queue.
+			c.fq.MustWriteBlockIgnoreDisabledPQ(block)
+			return
+		case <-c.stopCh:
+			// c must be stopped. Wait up to 5 seconds for the in-flight request to complete.
+			// If it succeeds, drain the remaining in-memory queue before returning.
+			stopCtx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+			defer cancel()
+
+			select {
+			case ok := <-ch:
+				if !ok {
+					// Return unsent block to the queue.
+					c.fq.MustWriteBlockIgnoreDisabledPQ(block)
+				} else {
+					c.drainInMemoryQueue(stopCtx, block[:0])
+				}
+			case <-stopCtx.Done():
+				// Return unsent block to the queue.
+				c.fq.MustWriteBlockIgnoreDisabledPQ(block)
+			}
+			return
+		}
+	}
+}
+
 func (c *client) doRequest(url string, body []byte) (*http.Response, error) {
 	req, err := c.newRequest(url, body)
 	if err != nil {

app/vmagent/remotewrite/remotewrite.go

@@ -66,6 +66,9 @@ var (
 	queues = flagutil.NewArrayInt("remoteWrite.queues", cgroup.AvailableCPUs()*2, "The number of concurrent queues to each -remoteWrite.url. Set more queues if default number of queues "+
 		"isn't enough for sending high volume of collected data to remote storage. "+
 		"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage")
+	inmemoryQueueWorkers = flagutil.NewArrayInt("remoteWrite.inmemoryQueueWorkers", 0, "The number of additional workers per each -remoteWrite.url, which send only recently ingested data from the in-memory queue, "+
+		"while the file-based queue at -remoteWrite.tmpDataPath is drained by workers configured via -remoteWrite.queues. "+
+		"This reduces delivery lag for fresh samples when the file-based queue contains a backlog accumulated during remote storage outages.")
 	showRemoteWriteURL = flag.Bool("remoteWrite.showURL", false, "Whether to show -remoteWrite.url in the exported metrics. "+
 		"It is hidden by default, since it can contain sensitive info such as auth key")
 	maxPendingBytesPerURL = flagutil.NewArrayBytes("remoteWrite.maxDiskUsagePerURL", 0, "The maximum file-based buffer size in bytes at -remoteWrite.tmpDataPath "+
@@ -923,7 +926,7 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, sanitizedURL string)
 	if maxInmemoryBlocks < 2 {
 		maxInmemoryBlocks = 2
 	}
-	fq := persistentqueue.MustOpenFastQueue(queuePath, sanitizedURL, maxInmemoryBlocks, maxPendingBytes, isPQDisabled)
+	fq := persistentqueue.MustOpenFastQueue(queuePath, sanitizedURL, maxInmemoryBlocks, maxPendingBytes, isPQDisabled, inmemoryQueueWorkers.GetOptionalArg(argIdx) > 0)
 	_ = metrics.GetOrCreateGauge(fmt.Sprintf(`vmagent_remotewrite_pending_data_bytes{path=%q, url=%q}`, queuePath, sanitizedURL), func() float64 {
 		return float64(fq.GetPendingBytes())
 	})

app/vmagent/remotewrite/remotewrite_test.go

@@ -77,7 +77,7 @@ func TestRemoteWriteContext_TryPushTimeSeries(t *testing.T) {
 
 		path := "fast-queue-write-test"
 		fs.MustRemoveDir(path)
-		fq := persistentqueue.MustOpenFastQueue(path, "test", 100, 0, false)
+		fq := persistentqueue.MustOpenFastQueue(path, "test", 100, 0, false, false)
 		defer fs.MustRemoveDir(path)
 		defer fq.MustClose()
 

app/vmstorage/main.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
@@ -21,7 +23,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/stringsutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vminsertapi"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmselectapi"
-	"github.com/VictoriaMetrics/metrics"
 )
 
 var (
@@ -152,7 +153,7 @@ func Init(vmselectMaxConcurrentRequests int, resetCacheIfNeeded func(mrs []stora
 		LogNewSeries:                *logNewSeries,
 	}
 	strg := storage.MustOpenStorage(*storageDataPath, opts)
-	vmStorage = newVMStorage(strg, vmselectMaxConcurrentRequests, resetCacheIfNeeded)
+	vmStorage = newVMStorageSingleNode(strg, vmselectMaxConcurrentRequests, resetCacheIfNeeded)
 
 	var m storage.Metrics
 	strg.UpdateMetrics(&m)
@@ -174,15 +175,15 @@ func Init(vmselectMaxConcurrentRequests int, resetCacheIfNeeded func(mrs []stora
 	GetSearch = vmStorage.GetSearch
 	PutSearch = vmStorage.PutSearch
 	RequestHandler = vmStorage.requestHandler
-	DebugFlush = vmStorage.s.DebugFlush
+	DebugFlush = vmStorage.vms.s.DebugFlush
 }
 
 var storageMetrics *metrics.Set
 
 var (
-	// vmStorage is an instance of vmstorage used by vminsert and
+	// vmStorageSingleNode is an instance of vmstorage used by vminsert and
 	// vmselect for writing and reading data.
-	vmStorage      *VMStorage
+	vmStorage      *VMStorageSingleNode
 	VMInsertAPI    vminsertapi.API
 	VMSelectAPI    vmselectapi.API
 	GetSearch      func(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (*storage.Search, int, error)
@@ -208,12 +209,15 @@ func Stop() {
 	logger.Infof("the vmstorage has been stopped")
 }
 
+func (vmssn *VMStorageSingleNode) requestHandler(w http.ResponseWriter, r *http.Request) bool {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.requestHandler(w, r)
+}
+
 // requestHandler is a storage request handler.
 // TODO(@rtm0): Move to a separate file, request_handler.go
 func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) bool {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	path := r.URL.Path
 	if path == "/internal/force_merge" {
 		if !httpserver.CheckAuthFlag(w, r, forceMergeAuthKey) {
@@ -358,10 +362,14 @@ var (
 )
 
 // TODO(@rtm0): Move to metrics.go.
-func (vms *VMStorage) writeStorageMetrics(w io.Writer) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
+func (vmssn *VMStorageSingleNode) writeStorageMetrics(w io.Writer) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	vmssn.vms.writeStorageMetrics(w)
+}
 
+// TODO(@rtm0): Move to metrics.go.
+func (vms *VMStorage) writeStorageMetrics(w io.Writer) {
 	strg := vms.s
 	var m storage.Metrics
 	strg.UpdateMetrics(&m)

app/vmstorage/vmstorage.go

@@ -1,7 +1,6 @@
 package vmstorage
 
 import (
-	"errors"
 	"flag"
 	"fmt"
 	"sync"
@@ -15,7 +14,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/syncwg"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmselectapi"
 )
@@ -36,7 +34,7 @@ var (
 // newVMStorage creates a new instance of of VMStorage.
 //
 // The created VMStorage instance takes ownership of s.
-func newVMStorage(s *storage.Storage, vmselectMaxConcurrentRequests int, resetCacheIfNeeded func(mrs []storage.MetricRow)) *VMStorage {
+func newVMStorage(s *storage.Storage, vmselectMaxConcurrentRequests int) *VMStorage {
 	if err := encoding.CheckPrecisionBits(uint8(*precisionBits)); err != nil {
 		logger.Fatalf("invalid -precisionBits=%d: %s", *precisionBits, err)
 	}
@@ -51,8 +49,6 @@ func newVMStorage(s *storage.Storage, vmselectMaxConcurrentRequests int, resetCa
 		maxUniqueTimeseries:           *maxUniqueTimeseries,
 		maxUniqueTimeSeriesCalculated: maxUniqueTimeseriesCalculated,
 		staleSnapshotsRemoverCh:       make(chan struct{}),
-		wg:                            syncwg.WaitGroup{},
-		resetCacheIfNeeded:            resetCacheIfNeeded,
 	}
 	vms.initStaleSnapshotsRemover()
 	return vms
@@ -82,17 +78,6 @@ type VMStorage struct {
 	maxUniqueTimeSeriesCalculated int
 	staleSnapshotsRemoverCh       chan struct{}
 	staleSnapshotsRemoverWG       sync.WaitGroup
-
-	// wg is used to wrap every storage call into wg.Add(1) ... wg.Done()
-	// for proper graceful shutdown when Stop is called.
-	//
-	// Use syncwg instead of sync, since Add is called from concurrent
-	// goroutines.
-	wg syncwg.WaitGroup
-
-	// resetCacheIfNeeded is a callback for automatic resetting of response
-	// cache if needed.
-	resetCacheIfNeeded func(mrs []storage.MetricRow)
 }
 
 func (vms *VMStorage) initStaleSnapshotsRemover() {
@@ -118,7 +103,6 @@ func (vms *VMStorage) initStaleSnapshotsRemover() {
 func (vms *VMStorage) Stop() {
 	close(vms.staleSnapshotsRemoverCh)
 	vms.staleSnapshotsRemoverWG.Wait()
-	vms.wg.WaitAndBlock()
 	vms.s.MustClose()
 }
 
@@ -127,14 +111,6 @@ func (vms *VMStorage) Stop() {
 // The caller should limit the number of concurrent calls to WriteRows() in
 // order to limit memory usage.
 func (vms *VMStorage) WriteRows(rows []storage.MetricRow) error {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
-	if vms.s.IsReadOnly() {
-		return errReadOnly
-	}
-	vms.resetCacheIfNeeded(rows)
-
 	vms.s.AddRows(rows, uint8(*precisionBits))
 	return nil
 }
@@ -144,41 +120,26 @@ func (vms *VMStorage) WriteRows(rows []storage.MetricRow) error {
 // The caller should limit the number of concurrent calls to WriteMetadata() in
 // order to limit memory usage.
 func (vms *VMStorage) WriteMetadata(rows []metricsmetadata.Row) error {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
-	if vms.s.IsReadOnly() {
-		return errReadOnly
-	}
 	vms.s.AddMetadataRows(rows)
 	return nil
 }
 
-var errReadOnly = errors.New("the storage is in read-only mode; check -storage.minFreeDiskSpaceBytes command-line flag value")
-
 // IsReadOnly returns true is the storage is in read-only mode.
 func (vms *VMStorage) IsReadOnly() bool {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
 	return vms.s.IsReadOnly()
 }
 
 func (vms *VMStorage) InitSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (vmselectapi.BlockIterator, error) {
-	vms.wg.Add(1)
-
 	tr := sq.GetTimeRange()
 	maxMetrics := vms.getMaxMetrics(sq.MaxMetrics)
 	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
 	if err != nil {
-		vms.wg.Done()
 		return nil, err
 	}
 	if len(tfss) == 0 {
-		vms.wg.Done()
 		return nil, fmt.Errorf("missing tag filters")
 	}
 	bi := getBlockIterator()
-	bi.wgDone = vms.wg.Done
 	bi.sr.Init(qt, vms.s, tfss, tr, maxMetrics, deadline)
 	if err := bi.sr.Error(); err != nil {
 		bi.MustClose()
@@ -200,9 +161,8 @@ func (vms *VMStorage) getMaxMetrics(searchQueryLimit int) int {
 
 // blockIterator implements vmselectapi.BlockIterator
 type blockIterator struct {
-	sr     storage.Search
-	mb     storage.MetricBlock
-	wgDone func()
+	sr storage.Search
+	mb storage.MetricBlock
 }
 
 var blockIteratorsPool sync.Pool
@@ -211,8 +171,6 @@ func (bi *blockIterator) MustClose() {
 	bi.sr.MustClose()
 	bi.mb.MetricName = nil
 	bi.mb.Block.Reset()
-	bi.wgDone()
-	bi.wgDone = nil
 	blockIteratorsPool.Put(bi)
 }
 
@@ -239,63 +197,8 @@ func (bi *blockIterator) Error() error {
 	return bi.sr.Error()
 }
 
-// GetSearch sets up an instance of storage search and returns it to the caller
-// along with the max series count that the search can return.
-//
-// This method is not part of the vmselectapi.API and must only be used by
-// vmsingle HTTP handlers.
-//
-// Callers of this method must call PutSearch() once the search instance is not
-// needed anymore.
-func (vms *VMStorage) GetSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (*storage.Search, int, error) {
-	vms.wg.Add(1)
-
-	tr := sq.GetTimeRange()
-	maxMetrics := vms.getMaxMetrics(sq.MaxMetrics)
-	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
-	if err != nil {
-		vms.wg.Done()
-		return nil, 0, err
-	}
-
-	sr := getSearch()
-	maxSeriesCount := sr.Init(qt, vms.s, tfss, tr, sq.MaxMetrics, deadline)
-	return sr, maxSeriesCount, nil
-}
-
-// PutSearch resets the search once it is not needed anymore and puts it aside
-// for future reuse.
-//
-// This method is not part of the vmselectapi.API and must only be used by
-// vmsingle HTTP handlers.
-//
-// The method must only be used on search instances that have been created with
-// GetSearch().
-func (vms *VMStorage) PutSearch(sr *storage.Search) {
-	putSearch(sr)
-	vms.wg.Done()
-}
-
-func getSearch() *storage.Search {
-	v := ssPool.Get()
-	if v == nil {
-		return &storage.Search{}
-	}
-	return v.(*storage.Search)
-}
-
-func putSearch(sr *storage.Search) {
-	sr.MustClose()
-	ssPool.Put(sr)
-}
-
-var ssPool sync.Pool
-
 // SearchMetricNames returns metric names for the given tfss on the given tr.
 func (vms *VMStorage) SearchMetricNames(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) ([]string, error) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	tr := sq.GetTimeRange()
 	maxMetrics := sq.MaxMetrics
 	if maxMetrics <= 0 {
@@ -316,9 +219,6 @@ func (vms *VMStorage) SearchMetricNames(qt *querytracer.Tracer, sq *storage.Sear
 // SearchLabelValues searches for label values for the given labelName, tfss and
 // tr.
 func (vms *VMStorage) LabelValues(qt *querytracer.Tracer, sq *storage.SearchQuery, labelName string, maxLabelValues int, deadline uint64) ([]string, error) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	tr := sq.GetTimeRange()
 	if maxLabelValues <= 0 || maxLabelValues > *maxTagValues {
 		maxLabelValues = *maxTagValues
@@ -344,9 +244,6 @@ func (vms *VMStorage) LabelValues(qt *querytracer.Tracer, sq *storage.SearchQuer
 // similar APIs.
 func (vms *VMStorage) TagValueSuffixes(qt *querytracer.Tracer, _, _ uint32, tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte,
 	maxSuffixes int, deadline uint64) ([]string, error) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	if maxSuffixes <= 0 || maxSuffixes > *maxTagValueSuffixesPerSearch {
 		maxSuffixes = *maxTagValueSuffixesPerSearch
 	}
@@ -363,9 +260,6 @@ func (vms *VMStorage) TagValueSuffixes(qt *querytracer.Tracer, _, _ uint32, tr s
 
 // SearchLabelNames searches for tag keys matching the given tfss on tr.
 func (vms *VMStorage) LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLabelNames int, deadline uint64) ([]string, error) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	tr := sq.GetTimeRange()
 	if maxLabelNames <= 0 || maxLabelNames > *maxTagKeys {
 		maxLabelNames = *maxTagKeys
@@ -384,8 +278,6 @@ func (vms *VMStorage) LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery
 }
 
 func (vms *VMStorage) SeriesCount(_ *querytracer.Tracer, _, _ uint32, deadline uint64) (uint64, error) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
 	return vms.s.GetSeriesCount(deadline)
 }
 
@@ -395,9 +287,6 @@ func (vms *VMStorage) Tenants(_ *querytracer.Tracer, _ storage.TimeRange, _ uint
 
 // GetTSDBStatus returns TSDB status for given filters on the given date.
 func (vms *VMStorage) TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery, focusLabel string, topN int, deadline uint64) (*storage.TSDBStatus, error) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	tr := sq.GetTimeRange()
 	maxMetrics := sq.MaxMetrics
 	if maxMetrics <= 0 {
@@ -417,9 +306,6 @@ func (vms *VMStorage) TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery
 //
 // Returns the number of deleted series.
 func (vms *VMStorage) DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (int, error) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	tr := sq.GetTimeRange()
 	maxMetrics := sq.MaxMetrics
 	if maxMetrics <= 0 {
@@ -438,26 +324,17 @@ func (vms *VMStorage) DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQue
 }
 
 func (vms *VMStorage) RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow, _ uint64) error {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	vms.s.RegisterMetricNames(qt, mrs)
 	return nil
 }
 
 // GetMetricNamesUsageStats returns metric name usage stats.
 func (vms *VMStorage) GetMetricNamesUsageStats(qt *querytracer.Tracer, _ *storage.TenantToken, limit, le int, matchPattern string, _ uint64) (metricnamestats.StatsResult, error) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	return vms.s.GetMetricNamesStats(qt, limit, le, matchPattern), nil
 }
 
 // ResetMetricNamesStats resets state for metric names usage tracker
 func (vms *VMStorage) ResetMetricNamesUsageStats(qt *querytracer.Tracer, _ uint64) error {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
-
 	vms.s.ResetMetricNamesStats(qt)
 	return nil
 }
@@ -494,8 +371,6 @@ func (vms *VMStorage) setupTfss(qt *querytracer.Tracer, sq *storage.SearchQuery,
 }
 
 func (vms *VMStorage) GetMetadataRecords(qt *querytracer.Tracer, _ *storage.TenantToken, limit int, metricName string, _ uint64) ([]*metricsmetadata.Row, error) {
-	vms.wg.Add(1)
-	defer vms.wg.Done()
 	return vms.s.GetMetadataRows(qt, limit, metricName), nil
 }
 

app/vmstorage/vmstorage_single_node.go

@@ -0,0 +1,213 @@
+package vmstorage
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/syncwg"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmselectapi"
+)
+
+// newVMStorageSingleNode creates a new instance of of VMStorage for vmsingle.
+func newVMStorageSingleNode(s *storage.Storage, maxConcurrentRequests int, resetCacheIfNeeded func(mrs []storage.MetricRow)) *VMStorageSingleNode {
+	vms := newVMStorage(s, maxConcurrentRequests)
+	return &VMStorageSingleNode{
+		vms:                vms,
+		wg:                 syncwg.WaitGroup{},
+		resetCacheIfNeeded: resetCacheIfNeeded,
+	}
+}
+
+type VMStorageSingleNode struct {
+	vms *VMStorage
+
+	// wg is used to wrap every storage call into wg.Add(1) ... wg.Done()
+	// for proper graceful shutdown when Stop is called.
+	//
+	// Use syncwg instead of sync, since Add is called from concurrent
+	// goroutines.
+	wg syncwg.WaitGroup
+
+	// resetCacheIfNeeded is a callback for automatic resetting of response
+	// cache if needed.
+	resetCacheIfNeeded func(mrs []storage.MetricRow)
+}
+
+func (vmssn *VMStorageSingleNode) Stop() {
+	vmssn.wg.WaitAndBlock()
+	vmssn.vms.Stop()
+}
+
+// WriteRows writes metric rows to the storage.
+//
+// Returns an error if the storage is in read-only mode.
+//
+// The caller should limit the number of concurrent calls to WriteRows() in
+// order to limit memory usage.
+func (vmssn *VMStorageSingleNode) WriteRows(rows []storage.MetricRow) error {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+
+	if vmssn.vms.IsReadOnly() {
+		return errReadOnly
+	}
+	vmssn.resetCacheIfNeeded(rows)
+	return vmssn.vms.WriteRows(rows)
+}
+
+// WriteMetadata writes metrics metadata to storage.
+//
+// Returns an error if the storage is in read-only mode.
+//
+// The caller should limit the number of concurrent calls to WriteMetadata() in
+// order to limit memory usage.
+func (vmssn *VMStorageSingleNode) WriteMetadata(rows []metricsmetadata.Row) error {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+
+	if vmssn.vms.IsReadOnly() {
+		return errReadOnly
+	}
+	return vmssn.vms.WriteMetadata(rows)
+}
+
+var errReadOnly = errors.New("the storage is in read-only mode; check -storage.minFreeDiskSpaceBytes command-line flag value")
+
+func (vmssn *VMStorageSingleNode) IsReadOnly() bool {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.IsReadOnly()
+}
+
+func (vmssn *VMStorageSingleNode) InitSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (vmselectapi.BlockIterator, error) {
+	return nil, fmt.Errorf("not implemented in vmsingle")
+}
+
+// GetSearch sets up an instance of storage search and returns it to the caller
+// along with the max series count that the search can return.
+//
+// This method is not part of the vmselectapi.API and must only be used by
+// vmsingle HTTP handlers.
+//
+// Callers of this method must call PutSearch() once the search instance is not
+// needed anymore.
+func (vmssn *VMStorageSingleNode) GetSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (*storage.Search, int, error) {
+	vmssn.wg.Add(1)
+
+	tr := sq.GetTimeRange()
+	maxMetrics := vmssn.vms.getMaxMetrics(sq.MaxMetrics)
+	tfss, err := vmssn.vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
+	if err != nil {
+		vmssn.wg.Done()
+		return nil, 0, err
+	}
+
+	sr := getSearch()
+	maxSeriesCount := sr.Init(qt, vmssn.vms.s, tfss, tr, sq.MaxMetrics, deadline)
+	return sr, maxSeriesCount, nil
+}
+
+// PutSearch resets the search once it is not needed anymore and puts it aside
+// for future reuse.
+//
+// This method is not part of the vmselectapi.API and must only be used by
+// vmsingle HTTP handlers.
+//
+// The method must only be used on search instances that have been created with
+// GetSearch().
+func (vmssn *VMStorageSingleNode) PutSearch(sr *storage.Search) {
+	putSearch(sr)
+	vmssn.wg.Done()
+}
+
+func getSearch() *storage.Search {
+	v := ssPool.Get()
+	if v == nil {
+		return &storage.Search{}
+	}
+	return v.(*storage.Search)
+}
+
+func putSearch(sr *storage.Search) {
+	sr.MustClose()
+	ssPool.Put(sr)
+}
+
+var ssPool sync.Pool
+
+func (vmssn *VMStorageSingleNode) SearchMetricNames(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) ([]string, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.SearchMetricNames(qt, sq, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) LabelValues(qt *querytracer.Tracer, sq *storage.SearchQuery, labelName string, maxLabelValues int, deadline uint64) ([]string, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.LabelValues(qt, sq, labelName, maxLabelValues, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) TagValueSuffixes(qt *querytracer.Tracer, accountID, projectID uint32, tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte, maxSuffixes int, deadline uint64) ([]string, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.TagValueSuffixes(qt, accountID, projectID, tr, tagKey, tagValuePrefix, delimiter, maxSuffixes, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLabelNames int, deadline uint64) ([]string, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.LabelNames(qt, sq, maxLabelNames, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) SeriesCount(qt *querytracer.Tracer, accountID, projectID uint32, deadline uint64) (uint64, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.SeriesCount(qt, accountID, projectID, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) Tenants(qt *querytracer.Tracer, tr storage.TimeRange, deadline uint64) ([]string, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.Tenants(qt, tr, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery, focusLabel string, topN int, deadline uint64) (*storage.TSDBStatus, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.TSDBStatus(qt, sq, focusLabel, topN, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (int, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.DeleteSeries(qt, sq, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow, deadline uint64) error {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.RegisterMetricNames(qt, mrs, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) GetMetricNamesUsageStats(qt *querytracer.Tracer, tt *storage.TenantToken, limit, le int, matchPattern string, deadline uint64) (metricnamestats.StatsResult, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.GetMetricNamesUsageStats(qt, tt, limit, le, matchPattern, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) ResetMetricNamesUsageStats(qt *querytracer.Tracer, deadline uint64) error {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.ResetMetricNamesUsageStats(qt, deadline)
+}
+
+func (vmssn *VMStorageSingleNode) GetMetadataRecords(qt *querytracer.Tracer, tt *storage.TenantToken, limit int, metricName string, deadline uint64) ([]*metricsmetadata.Row, error) {
+	vmssn.wg.Add(1)
+	defer vmssn.wg.Done()
+	return vmssn.vms.GetMetadataRecords(qt, tt, limit, metricName, deadline)
+}

app/vmstorage/vmstorage_test.go

@@ -48,7 +48,7 @@ func TestGetMaxMetrics(t *testing.T) {
 		t.Helper()
 		*maxUniqueTimeseries = storageMaxUniqueTimeseries
 		s := storage.MustOpenStorage(t.Name(), storage.OpenOptions{})
-		vms := newVMStorage(s, maxConcurrentRequests, func(mrs []storage.MetricRow) {})
+		vms := newVMStorage(s, maxConcurrentRequests)
 		defer vms.Stop()
 		maxMetrics := vms.getMaxMetrics(searchQueryLimit)
 		if maxMetrics != expect {

apptest/tests/vmagent_remotewrite_test.go

@@ -332,13 +332,11 @@ func TestSingleVMAgentDropOnOverload(t *testing.T) {
 	vmagent.APIV1ImportPrometheusNoWaitFlush(t, []string{
 		"foo_bar 1 1652169600000", // 2022-05-10T08:00:00Z
 	}, apptest.QueryOpts{})
-
 	waitFor(
 		func() bool {
 			return vmagent.RemoteWriteRequests(t, url1) == 1 && vmagent.RemoteWriteRequests(t, url2) == 1
 		},
 	)
-
 	// Send 2 more requests, the first RW endpoint should receive everything, the second should add them to the queue
 	// since worker is busy with the first request.
 	for i := range 2 {
@@ -641,3 +639,102 @@ func TestSingleVMAgentMultitenancy(t *testing.T) {
 		t.Fatalf("expected vmagent_tenant_inserted_rows_total to have value 1 for accountID=5, projectID=0")
 	}
 }
+
+func TestSingleVMAgentPriorizeRecentData(t *testing.T) {
+	tc := apptest.NewTestCase(t)
+	defer tc.Stop()
+
+	remoteWriteSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer remoteWriteSrv.Close()
+
+	remoteWriteSrv2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer remoteWriteSrv2.Close()
+
+	vmagent := tc.MustStartDefaultRWVmagent("vmagent", []string{
+		fmt.Sprintf(`-remoteWrite.url=%s/api/v1/write`, remoteWriteSrv.URL),
+		fmt.Sprintf(`-remoteWrite.url=%s/api/v1/write`, remoteWriteSrv2.URL),
+		"-remoteWrite.disableOnDiskQueue=true",
+		// use only 1 worker to get a full queue faster
+		"-remoteWrite.queues=1",
+		"-remoteWrite.flushInterval=1ms",
+		"-remoteWrite.inmemoryQueueWorkers=1",
+		// fastqueue size is roughly memory.Allowed() / len(urls) / *maxRowsPerBlock / 100
+		// Use very large maxRowsPerBlock to get fastqueue of minimal length(2).
+		// See initRemoteWriteCtxs function in remotewrite.go for details.
+		"-remoteWrite.maxRowsPerBlock=1000000000",
+		"-remoteWrite.tmpDataPath=" + tc.Dir() + "/vmagent",
+
+		// Delay retry logic to avoid race conditions with waitFor assertions.
+		// It improves the test stability on resource-constrained runners.
+		// Should be bigger than retries * period
+		"-remoteWrite.retryMinInterval=3s",
+	})
+
+	const (
+		retries = 20
+		period  = 100 * time.Millisecond
+	)
+
+	waitFor := func(f func() bool) {
+		t.Helper()
+		for range retries {
+			if f() {
+				return
+			}
+			time.Sleep(period)
+		}
+		t.Fatalf("timed out waiting for retry #%d", retries)
+	}
+
+	// Real remote write URLs are hidden in metrics
+	url1 := "1:secret-url"
+	url2 := "2:secret-url"
+
+	// Wait until first request got flushed to remote write server
+	vmagent.APIV1ImportPrometheusNoWaitFlush(t, []string{
+		"foo_bar 1 1652169600000", // 2022-05-10T08:00:00Z
+	}, apptest.QueryOpts{})
+	waitFor(
+		func() bool {
+			return vmagent.RemoteWriteRequests(t, url1) == 1 && vmagent.RemoteWriteRequests(t, url2) == 1
+		},
+	)
+	// Wait until second request got flushed to remote write server
+	// since there are 2 independent queues (general and in-memory) with minimal capacity of 1
+	vmagent.APIV1ImportPrometheusNoWaitFlush(t, []string{
+		"foo_bar 1 1652169600000", // 2022-05-10T08:00:00Z
+	}, apptest.QueryOpts{})
+	waitFor(
+		func() bool {
+			return vmagent.RemoteWriteRequests(t, url1) == 2 && vmagent.RemoteWriteRequests(t, url2) == 2
+		},
+	)
+	// Send 2 more requests, the first RW endpoint should receive everything, the second should add them to the queue
+	// since worker is busy with the first request.
+	for i := range 2 {
+		vmagent.APIV1ImportPrometheusNoWaitFlush(t, []string{
+			"foo_bar 1 1652169600000", // 2022-05-10T08:00:00Z
+		}, apptest.QueryOpts{})
+
+		waitFor(
+			func() bool {
+				return vmagent.RemoteWriteRequests(t, url1) == 3+i && vmagent.RemoteWritePendingInmemoryBlocks(t, url2) == 1+i
+			},
+		)
+	}
+
+	// Send one more request.
+	vmagent.APIV1ImportPrometheusNoWaitFlush(t, []string{
+		"foo_bar 1 1652169600000", // 2022-05-10T08:00:00Z
+	}, apptest.QueryOpts{})
+
+	waitFor(
+		func() bool {
+			return vmagent.RemoteWriteRequests(t, url1) == 5 && vmagent.RemoteWriteSamplesDropped(t, url2) > 0
+		},
+	)
+}

deployment/docker/vmanomaly/vmanomaly-integration/compose.yml

@@ -59,7 +59,7 @@ services:
       - '--external.alert.source=explore?orgId=1&left=["now-1h","now","VictoriaMetrics",{"expr": },{"mode":"Metrics"},{"ui":[true,true,true,"none"]}]'
     restart: always
   vmanomaly:
-    image: victoriametrics/vmanomaly:v1.29.5
+    image: victoriametrics/vmanomaly:v1.29.4
     depends_on:
       - "victoriametrics"
     ports:

deployment/docker/vmanomaly/vmanomaly-integration/vmanomaly_guide_dashboard.json

@@ -43,7 +43,7 @@
         "content": "If you don't observe any data initially, please wait a few minutes for it to appear. \n\nUpon the first running the guide (if there is not enough node_exporter monitoring data collected in your system), you may notice a significant number of false positive anomalies found. The predictions will become more accurate with at least two weeks' (full `fit_window`) worth of data provided to vmanomaly.\n\nEach row displays information for a distinct mode. The query used for anomaly detection is `sum(rate(node_cpu_seconds_total[5m])) by (mode, instance, job)`.\n",
         "mode": "markdown"
       },
-      "pluginVersion": "12.2.0",
+      "pluginVersion": "10.2.1",
       "title": "Overview",
       "type": "text"
     },

docs/anomaly-detection/CHANGELOG.md

@@ -14,21 +14,6 @@ aliases:
 ---
 Please find the changelog for VictoriaMetrics Anomaly Detection below.
 
-## v1.29.5
-Released: 2026-06-11
-
-- UI: Updated [vmanomaly UI](https://docs.victoriametrics.com/anomaly-detection/ui/) from [v1.7.0](https://docs.victoriametrics.com/anomaly-detection/ui/#v170) to [v1.7.1](https://docs.victoriametrics.com/anomaly-detection/ui/#v171), see respective [release notes](https://docs.victoriametrics.com/anomaly-detection/ui/#v171) for details.
-
-- IMPROVEMENT: Redesigned [hot reload](https://docs.victoriametrics.com/anomaly-detection/components/#hot-reload) config change detection to content-based polling with configurable `-configCheckInterval`, improving reliability for Kubernetes ConfigMap symlink rotations and other filesystems where event delivery can be inconsistent.
-
-- IMPROVEMENT: Refined config validation errors for broken or invalid config sections, so startup and reload failures point to the affected section more clearly (e.g. YAML indentation typos).
-
-- IMPROVEMENT: Tightened config validation for [`PeriodicScheduler`](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/#periodic-scheduler) `infer_every` and [`IsolationForestModel`](https://docs.victoriametrics.com/anomaly-detection/components/models/#isolation-forest-multivariate) `contamination`, including clearer handling of missing scheduler intervals, numeric contamination strings, and invalid non-finite values.
-
-- BUGFIX: Fixed a multiprocessing startup issue with `settings.n_workers > 1` that could leave scheduled data fetch or successive inference jobs stuck and repeatedly skipped by internal scheduler.
-
-- BUGFIX: Bounded [`VmReader`](https://docs.victoriametrics.com/anomaly-detection/components/reader/#vm-reader) and [`VLogsReader`](https://docs.victoriametrics.com/anomaly-detection/components/reader/#victorialogs-reader) data fetch and post-fetch processing waits so stalled datasource reads or multiprocessing dataframe creation no longer keep [`PeriodicScheduler`](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/#periodic-scheduler) `data_fetch` jobs running indefinitely. Previously, such stuck jobs could keep internal scheduler's `max_instances=1` slot per (scheduler, query) pair occupied, causing future data fetch, fit, or infer runs to be skipped until vmanomaly was restarted. The config validator now also warns when the configured reader timeout budget can exceed the connected scheduler interval.
-
 ## v1.29.4
 Released: 2026-05-15
 

docs/anomaly-detection/FAQ.md

@@ -423,7 +423,7 @@ services:
   # ...
   vmanomaly:
     container_name: vmanomaly
-    image: victoriametrics/vmanomaly:v1.29.5
+    image: victoriametrics/vmanomaly:v1.29.4
     # ...
     restart: always
     volumes:
@@ -641,7 +641,7 @@ options:
 Here’s an example of using the config splitter to divide configurations based on the `extra_filters` argument from the reader section:
 
 ```sh
-docker pull victoriametrics/vmanomaly:v1.29.5 && docker image tag victoriametrics/vmanomaly:v1.29.5 vmanomaly
+docker pull victoriametrics/vmanomaly:v1.29.4 && docker image tag victoriametrics/vmanomaly:v1.29.4 vmanomaly
 ```
 
 ```sh

docs/anomaly-detection/Migration.md

@@ -45,7 +45,8 @@ There are 2 types of compatibility to consider when migrating in stateful mode:
 
 | Group start | Group end | Compatibility | Notes |
 |---------|--------- |------------|-------|
-| [v1.29.1](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1291) | [v1.29.5](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1295) | Fully Compatible | - |
+| [v1.29.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1293) | Latest* | Fully Compatible | Just a placeholder for new releases |
+| [v1.29.1](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1291) | [v1.29.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1293) | Fully Compatible | - |
 | [v1.28.7](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1287) | [v1.29.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1290) | Partially compatible* | Dumped models of class [prophet](https://docs.victoriametrics.com/anomaly-detection/components/models/#prophet) and [seasonal quantile](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-seasonal-quantile) have problems with loading to [v1.29.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1290) due to dropped `pytz` library. **Upgrading directly from v1.28.7 to [v1.29.1](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1291) with a fix is suggested** |
 | [v1.26.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1262) | [v1.28.7](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1287) | Fully Compatible | [v1.28.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1280) introduced [rolling](https://docs.victoriametrics.com/anomaly-detection/components/models/#rolling-models) model class drop in favor of [online](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-models) models (`rolling_quantile` and `std` models), however, it does not impact compatibility, as artifacts were not produced by default for rolling models. Also, offline `mad` and `zscore` models are redirecting to their respective online counterparts since [v1.28.4](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1284). |
 | [v1.25.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1253) | [v1.26.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1270) | Partially Compatible* | [v1.25.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1253) introduced `forecast_at` argument for base [univariate](https://docs.victoriametrics.com/anomaly-detection/components/models/#univariate-models) and `Prophet` [models](https://docs.victoriametrics.com/anomaly-detection/components/models/#prophet), however, itself remains backward-reversible from newer states like [v1.26.2](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1262), [v1.27.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1270). (All models except `isolation_forest_multivariate` class will be dropped) |

docs/anomaly-detection/QuickStart.md

@@ -37,39 +37,29 @@ The `vmanomaly` service supports a set of command-line arguments to configure it
 > Single-dashed command-line argument {{% available_from "v1.23.3" anomaly %}} format can be used, e.g. `-license.forceOffline` in addition to `--license.forceOffline`. This aligns better with other VictoriaMetrics ecosystem components. Mixing the two styles is also supported, e.g. `-license.forceOffline --loggerLevel INFO`.
 
 ```shellhelp
-usage: vmanomaly.py [-h] [--license STRING | --licenseFile PATH] [--license.forceOffline] [--loggerLevel {DEBUG,INFO,WARNING,ERROR,FATAL}] [--watch] [-configCheckInterval DURATION] [--dryRun] [--outputSpec PATH] config [config ...]
+usage: vmanomaly.py [-h] [--license STRING | --licenseFile PATH] [--license.forceOffline] [--loggerLevel {DEBUG,WARNING,FATAL,ERROR,INFO}] [--watch] [--dryRun] [--outputSpec PATH] config [config ...]
 
 VictoriaMetrics Anomaly Detection Service
 
 positional arguments:
-  config                YAML config file(s) or directories containing YAML files. Multiple files will recursively merge each other
-                        values so multiple configs can be combined. If a directory is provided, all `.yaml` files inside will be
-                        merged, without recursion. Default: vmanomaly.yaml is expected in the current directory.
+  config                YAML config file(s) or directories containing YAML files. Multiple files will recursively merge each other values so multiple configs can be combined. If a directory is provided,
+                        all `.yaml` files inside will be merged, without recursion. Default: vmanomaly.yaml is expected in the current directory.
 
 options:
   -h                    Show this help message and exit
-  --license STRING      License key for VictoriaMetrics Enterprise. See https://victoriametrics.com/products/enterprise/trial/ to
-                        obtain a trial license.
-  --licenseFile PATH    Path to file with license key for VictoriaMetrics Enterprise. See
-                        https://victoriametrics.com/products/enterprise/trial/ to obtain a trial license.
+  --license STRING      License key for VictoriaMetrics Enterprise. See https://victoriametrics.com/products/enterprise/trial/ to obtain a trial license.
+  --licenseFile PATH    Path to file with license key for VictoriaMetrics Enterprise. See https://victoriametrics.com/products/enterprise/trial/ to obtain a trial license.
   --license.forceOffline
-                        Whether to force offline verification for VictoriaMetrics Enterprise license key, which has been passed either
-                        via -license or via -licenseFile command-line flag. The issued license key must support offline verification
-                        feature. Contact info@victoriametrics.com if you need offline license verification.
-  --loggerLevel {DEBUG,INFO,WARNING,ERROR,FATAL}
-                        Minimum level to log. Possible values: {'DEBUG', 'INFO', 'WARNING', 'ERROR', 'FATAL'}.
-  --watch               Watch config files for changes and trigger hot reloads. Watches the specified config file or directory for
-                        modifications, deletions, or additions. Upon detecting changes, triggers config reload. If new config
-                        validation fails, continues with previous valid config and state.
-  -configCheckInterval DURATION
-                        Interval for checking watched config files for content changes. Default: 30s.
-  --dryRun              Validate only: parse + merge all YAML(s) and run schema checks, then exit. Does not require a license to run.
-                        Does not expose metrics, or launch vmanomaly service(s).
+                        Whether to force offline verification for VictoriaMetrics Enterprise license key, which has been passed either via -license or via -licenseFile command-line flag. The issued
+                        license key must support offline verification feature. Contact info@victoriametrics.com if you need offline license verification.
+  --loggerLevel {DEBUG,WARNING,FATAL,ERROR,INFO}
+                        Minimum level to log. Possible values: DEBUG, INFO, WARNING, ERROR, FATAL.
+  --watch               Watch config files for changes and trigger hot reloads. Watches the specified config file or directory for modifications, deletions, or additions. Upon detecting changes,
+                        triggers config reload. If new config validation fails, continues with previous valid config and state.
+  --dryRun              Validate only: parse + merge all YAML(s) and run schema checks, then exit. Does not require a license to run. Does not expose metrics, or launch vmanomaly service(s).
   --outputSpec PATH     Target location of .yaml output spec.
 ```
 
-{{% available_from "v1.29.5" anomaly %}} When `--watch` is enabled, config changes are detected by fixed-interval content polling instead of filesystem event delivery. The polling frequency is controlled by `-configCheckInterval` (default: `30s`). The same option can also be passed as `--configCheckInterval`, `--config.check.interval`, `--config-check-interval`, `--config_check_interval`, or in key-value form such as `configCheckInterval=30s`.
-
 You can specify these options when running `vmanomaly` to fine-tune logging levels or handle licensing configurations, as per your requirements.
 
 ### Licensing
@@ -132,7 +122,7 @@ Below are the steps to get `vmanomaly` up and running inside a Docker container:
 1. Pull Docker image:
 
 ```sh
-docker pull victoriametrics/vmanomaly:v1.29.5
+docker pull victoriametrics/vmanomaly:v1.29.4
 ```
 
 2. Create the license file with your license key.
@@ -152,7 +142,7 @@ docker run -it \
     -v ./license:/license \
     -v ./config.yaml:/config.yaml \
     -p 8490:8490 \
-    victoriametrics/vmanomaly:v1.29.5 \
+    victoriametrics/vmanomaly:v1.29.4 \
     /config.yaml \
     --licenseFile=/license \
     --loggerLevel=INFO \
@@ -169,7 +159,7 @@ docker run -it \
     -e VMANOMALY_DATA_DUMPS_DIR=/tmp/vmanomaly/data \
     -e VMANOMALY_MODEL_DUMPS_DIR=/tmp/vmanomaly/models \
     -p 8490:8490 \
-    victoriametrics/vmanomaly:v1.29.5 \
+    victoriametrics/vmanomaly:v1.29.4 \
     /config.yaml \
     --licenseFile=/license \
     --loggerLevel=INFO \
@@ -182,7 +172,7 @@ services:
   # ...
   vmanomaly:
     container_name: vmanomaly
-    image: victoriametrics/vmanomaly:v1.29.5
+    image: victoriametrics/vmanomaly:v1.29.4
     # ...
     restart: always
     volumes:

docs/anomaly-detection/UI.md

@@ -315,7 +315,7 @@ docker run -it --rm \
   -e VMANOMALY_MCP_SERVER_URL=http://mcp-vmanomaly:8081/mcp \
   -p 8080:8080 \
   -p 8490:8490 \
-  victoriametrics/vmanomaly:v1.29.5 \
+  victoriametrics/vmanomaly:v1.29.4 \
   vmanomaly_config.yaml
 ```
 
@@ -640,17 +640,6 @@ If the **results** look good and the **model configuration should be deployed in
 
 ## Changelog
 
-### v1.7.1
-Released: 2026-06-11
-
-vmanomaly version: [v1.29.5](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1295)
-
-- FEATURE: Added bulk Apply/Decline actions for [Copilot](#ai-assistance) chat suggestions.
-
-- BUGFIX: Fixed modal windows closing when the mouse is released outside the window during text selection.
-
-- BUGFIX: Fixed tooltip hover behavior so tooltips do not disappear while the cursor moves into the hover content.
-
 ### v1.7.0
 Released: 2026-05-15
 

docs/anomaly-detection/components/README.md

@@ -143,14 +143,11 @@ server:
 
 > This feature is better used in conjunction with [stateful service](https://docs.victoriametrics.com/anomaly-detection/components/settings/#state-restoration) to preserve the state of the models and schedulers between restarts and reuse what can be reused, thus avoiding unnecessary re-training of models, re-initialization of schedulers and re-reading of data.
 
-{{% available_from "v1.25.0" anomaly %}} Service supports hot reload of configuration files, which allows for automatic reloading of configurations on config files change without the need of explicit service restart. This can be enabled via the `--watch` [CLI argument](https://docs.victoriametrics.com/anomaly-detection/quickstart/#command-line-arguments). `vmanomaly_config_reload_enabled` flag in [self-monitoring metrics](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#startup-metrics) will be set to 1 (if enabled) or 0 (if disabled).
-
-> [!NOTE]
-> {{% deprecated_from "v1.29.5" anomaly %}} File system event-based hot reload has been deprecated in favor of content-based polling with configurable `-configCheckInterval` due to reliability issues with Kubernetes ConfigMap symlink rotations and other filesystems where event delivery can be inconsistent. If you were using file system event-based hot reload, please switch to content-based polling by enabling `--watch` flag and configuring `-configCheckInterval` as needed.
+{{% available_from "v1.25.0" anomaly %}} Service supports hot reload of configuration files, which allows for automatic reloading of configurations on config files change filesystem events without the need of explicit service restart. This can be enabled via the `--watch` [CLI argument](https://docs.victoriametrics.com/anomaly-detection/quickstart/#command-line-arguments). `vmanomaly_config_reload_enabled` flag in [self-monitoring metrics](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#startup-metrics) will be set to 1 (if enabled) or 0 (if disabled).
 
 ### How it works
 
-It works by checking watched `.yml|.yaml` file contents in the specified files or directories on the configured interval `-configCheckInterval` (default is `30s`) {{% available_from "v1.29.5" anomaly %}}. When a content change is detected, the service will attempt to reload the configuration files after the existing debounce window, rebuild the [global config](https://docs.victoriametrics.com/anomaly-detection/scaling-vmanomaly/#global-configuration) and reinitialize the components. If the reload is successful, the `vmanomaly_config_reloads_total` metric will be incremented for `status="success"` label, otherwise it will be incremented with `status="failure"` label and a respective error message on config validation failure(s) will be logged.
+It works by watching for file system events, such as modifications, creations, or deletions of `.yml|.yaml` files in the specified directories. When a change is detected, the service will attempt to reload the configuration files, rebuild the [global config](https://docs.victoriametrics.com/anomaly-detection/scaling-vmanomaly/#global-configuration) and reinitialize the components. If the reload is successful, the `vmanomaly_config_reloads_total` metric will be incremented for `status="success"` label, otherwise it will be incremented with `status="failure"` label and a respective error message on config validation failure(s) will be logged.
 
 > If the reload fails, the service will log an error message indicating the reason for the failure, and the **previous configuration will remain active until a successful reload occurs** to preserve the service's stability. This means that if there are errors in the new configuration, the service will continue to operate with the last valid configuration until the issues are resolved.
 

docs/anomaly-detection/components/models.md

@@ -449,9 +449,9 @@ models:
 
 > The `decay` argument works only in combination with [online models](#online-models) like [`ZScoreOnlineModel`](#online-z-score) or [`OnlineQuantileModel`](#online-seasonal-quantile).
 
-The `decay` {{% available_from "v1.23.0" anomaly %}} argument is used to control the (exponential) **decay factor** for online models, which determines how quickly the model adapts to new data. It is a positive float value from `(0.0, 1.0]` interval, where:
-- Value `1.0` means no decay (the model treats all data points equally, without giving more weight to recent ones). This is the default value for backward compatibility.
-- Values less than `1.0` mean that the model will give more weight to recent data, effectively "forgetting" older data over time.
+The `decay` {{% available_from "v1.23.0" anomaly %}} argument is used to control the (exponential) **decay factor** for online models, which determines how quickly the model adapts to new data. It is a float value between `0.0` and `1.0`, where:
+- `1.0` means no decay (the model treats all data equally, without giving more weight to recent data). This is the default value for backward compatibility.
+- Less than `1.0` means that the model will give more weight to recent data, effectively "forgetting" older data over time.
 
 Roughly speaking, for the recent N datapoints model processes `decay` = `d` means that these datapoints will contribute to the model as [1 - d^X] percent of total importance, for example decay of
 - `0.99` means that 100 recent datapoints will contribute as [1 - 0.99^100] = 63.23% of total importance
@@ -998,7 +998,7 @@ Here we use Isolation Forest implementation from `scikit-learn` [library](https:
 
 * `class` (string) - model class name `"model.isolation_forest.IsolationForestMultivariateModel"` (or `isolation_forest_multivariate` with class alias support {{% available_from "v1.13.0" anomaly %}})
 
-* `contamination` (float or string, optional) - The amount of contamination of the data set, i.e. the proportion of outliers in the data set. Used when fitting to define the threshold on the scores of the samples. Default value - "auto". Should be either `"auto"` or be in the range (0.0, 0.5]. {{% available_from "v1.29.5" anomaly %}} Numeric strings, such as `"0.01"`, are accepted, while invalid non-finite values, such as `nan`, `inf`, and `-inf`, are rejected during config validation.
+* `contamination` (float or string, optional) - The amount of contamination of the data set, i.e. the proportion of outliers in the data set. Used when fitting to define the threshold on the scores of the samples. Default value - "auto". Should be either `"auto"` or be in the range (0.0, 0.5].
 
 * `seasonal_features` (list of string) - List of seasonality to encode through [cyclical encoding](https://towardsdatascience.com/cyclical-features-encoding-its-about-time-ce23581845ca), i.e. `dow` (day of week). **Introduced in [1.12.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1120)**. 
   - Empty by default for backward compatibility.
@@ -1265,7 +1265,7 @@ monitoring:
 Let's pull the docker image for `vmanomaly`:
 
 ```sh
-docker pull victoriametrics/vmanomaly:v1.29.5
+docker pull victoriametrics/vmanomaly:v1.29.4
 ```
 
 Now we can run the docker container putting as volumes both config and model file:
@@ -1279,7 +1279,7 @@ docker run -it \
 -v $(PWD)/license:/license \
 -v $(PWD)/custom_model.py:/vmanomaly/model/custom.py \
 -v $(PWD)/custom.yaml:/config.yaml \
-victoriametrics/vmanomaly:v1.29.5 /config.yaml \
+victoriametrics/vmanomaly:v1.29.4 /config.yaml \
 --licenseFile=/license
 --watch
 ```

docs/anomaly-detection/guides/guide-vmanomaly-vmalert/README.md

@@ -10,12 +10,12 @@ sitemap:
 
 - To use *vmanomaly*, part of the enterprise package, a license key is required. Obtain your key [here](https://victoriametrics.com/products/enterprise/trial/) for this tutorial or for enterprise use.
 - In the tutorial, we'll be using the following VictoriaMetrics components:
-  -  [VictoriaMetrics Single-Node](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) (v1.145.0)
-  -  [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/) (v1.145.0)
-  -  [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) (v1.145.0)
-- [Grafana](https://grafana.com/) (v12.2.0)
+  -  [VictoriaMetrics Single-Node](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) (v1.137.0)
+  -  [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/) (v1.137.0)
+  -  [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) (v1.137.0)
+- [Grafana](https://grafana.com/) (v.10.2.1)
 - [Docker](https://docs.docker.com/get-docker/) and [Docker Compose](https://docs.docker.com/compose/)
-- [Node exporter](https://github.com/prometheus/node_exporter#node-exporter) (v1.9.1) and [Alertmanager](https://prometheus.io/docs/alerting/latest/alertmanager/) (v0.28.1)
+- [Node exporter](https://github.com/prometheus/node_exporter#node-exporter) (v1.7.0) and [Alertmanager](https://prometheus.io/docs/alerting/latest/alertmanager/) (v0.27.0)
 
 ![typical setup diagram](guide-vmanomaly-vmalert_overview.webp)
 
@@ -323,7 +323,7 @@ Let's wrap it all up together into the `docker-compose.yml` file.
 services:
   vmagent:
     container_name: vmagent
-    image: victoriametrics/vmagent:v1.145.0
+    image: victoriametrics/vmagent:v1.137.0
     depends_on:
       - "victoriametrics"
     ports:
@@ -340,7 +340,7 @@ services:
 
   victoriametrics:
     container_name: victoriametrics
-    image: victoriametrics/victoria-metrics:v1.145.0
+    image: victoriametrics/victoria-metrics:v1.137.0
     ports:
       - 8428:8428
     volumes:
@@ -356,7 +356,7 @@ services:
 
   grafana:
     container_name: grafana
-    image: grafana/grafana:12.2.0
+    image: grafana/grafana-oss:10.2.1
     depends_on:
       - "victoriametrics"
     ports:
@@ -373,7 +373,7 @@ services:
 
   vmalert:
     container_name: vmalert
-    image: victoriametrics/vmalert:v1.145.0
+    image: victoriametrics/vmalert:v1.137.0
     depends_on:
       - "victoriametrics"
     ports:
@@ -395,7 +395,7 @@ services:
     restart: always
   vmanomaly:
     container_name: vmanomaly
-    image: victoriametrics/vmanomaly:v1.29.5
+    image: victoriametrics/vmanomaly:v1.29.4
     depends_on:
       - "victoriametrics"
     ports:
@@ -412,7 +412,7 @@ services:
       - "--licenseFile=/license"
   alertmanager:
     container_name: alertmanager
-    image: prom/alertmanager:v0.28.1
+    image: prom/alertmanager:v0.27.0
     volumes:
       - ./alertmanager.yml:/config/alertmanager.yml
     command:
@@ -424,7 +424,7 @@ services:
     restart: always
 
   node-exporter:
-    image: quay.io/prometheus/node-exporter:v1.9.1
+    image: quay.io/prometheus/node-exporter:v1.7.0
     container_name: node-exporter
     ports:
       - 9100:9100

docs/victoriametrics/changelog/CHANGELOG.md

@@ -26,6 +26,8 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 
 ## tip
 
+* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): add a new flag `-remoteWrite.inmemoryQueueWorkers` to prioritize recently ingested data over historical data stored at file-based [persistent queue](https://docs.victoriametrics.com/victoriametrics/vmagent/#on-disk-persistence-and-data-processing-order). See [#8833](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8833)
+
 * BUGFIX: [stream aggregation](https://docs.victoriametrics.com/victoriametrics/stream-aggregation/): fix issue with producing aggregated samples with identical timestamps between flushes. See PR [#10808](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/10808) for details.
 
 ## [v1.145.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.145.0)

docs/victoriametrics/vmagent.md

@@ -928,6 +928,37 @@ vmagent will generate the following persistent queue folders:
 2_0AAFDF53E314A72A
 ```
 
+### On-disk persistence and data processing order
+
+By default, vmagent processes data in FIFO order. If data has been written
+to the on-disk queue, it must be flushed to the remote storage before newly
+ingested data can be forwarded there. During long outages, vmagent may
+accumulate large amounts of data in the file-based queue, which can introduce
+a significant lag between the moment data is collected by vmagent and the
+moment it becomes visible at the remote storage.
+
+This behavior can be changed with the `-remoteWrite.inmemoryQueueWorkers`
+command-line flag. When set to a non-zero value, vmagent starts the given
+number of additional workers, which send only recently ingested data from
+the in-memory queue, while the workers configured via `-remoteWrite.queues`
+drain the file-based backlog concurrently. This reduces the delivery lag for
+fresh samples after remote storage outages or slowdowns. The flag can be set
+individually per each `-remoteWrite.url`.
+
+Note that these workers are started in addition to the workers configured
+via `-remoteWrite.queues`, so the total number of concurrent connections to
+the remote storage becomes the sum of both flags. Take this into account if
+the remote storage limits the number of concurrent requests.
+
+This flag has the following possible limitations:
+
+* Samples may arrive at the remote storage out of order, since recent data
+  can be delivered before the older backlogged data. Do not use this option
+  if the remote storage doesn't accept out-of-order samples.
+* Recent data isn't guaranteed to take the fast path: if the in-memory queue
+  is full, newly ingested data is still written to the file-based queue and
+  is delivered in FIFO order by the generic workers.
+
 ### Disabling On-disk persistence
 
 There are cases when it is better to disable on-disk persistence for pending data on the `vmagent` side:

lib/persistentqueue/fastqueue.go

@@ -26,6 +26,8 @@ type FastQueue struct {
 	// isPQDisabled is set to true when pq is disabled.
 	isPQDisabled bool
 
+	prioritizeInMemoryData bool
+
 	// pq is file-based queue
 	pq *queue
 
@@ -48,12 +50,13 @@ type FastQueue struct {
 // reaches maxPendingSize.
 // if isPQDisabled is set to true, then write requests that exceed in-memory buffer capacity are rejected.
 // in-memory queue part can be stored on disk during graceful shutdown.
-func MustOpenFastQueue(path, name string, maxInmemoryBlocks int, maxPendingBytes int64, isPQDisabled bool) *FastQueue {
+func MustOpenFastQueue(path, name string, maxInmemoryBlocks int, maxPendingBytes int64, isPQDisabled bool, prioritizeInMemoryData bool) *FastQueue {
 	pq := mustOpen(path, name, maxPendingBytes)
 	fq := &FastQueue{
-		pq:           pq,
-		isPQDisabled: isPQDisabled,
-		ch:           make(chan *bytesutil.ByteBuffer, maxInmemoryBlocks),
+		pq:                     pq,
+		isPQDisabled:           isPQDisabled,
+		prioritizeInMemoryData: prioritizeInMemoryData,
+		ch:                     make(chan *bytesutil.ByteBuffer, maxInmemoryBlocks),
 	}
 	fq.cond.L = &fq.mu
 	fq.lastInmemoryBlockReadTime = fasttime.UnixTimestamp()
@@ -97,7 +100,7 @@ func (fq *FastQueue) IsWriteBlocked() bool {
 	}
 	fq.mu.Lock()
 	defer fq.mu.Unlock()
-	return len(fq.ch) == cap(fq.ch) || fq.pq.GetPendingBytes() > 0
+	return len(fq.ch) == cap(fq.ch) || (fq.pq.GetPendingBytes() > 0 && !fq.prioritizeInMemoryData)
 }
 
 // UnblockAllReaders unblocks all the readers.
@@ -194,18 +197,25 @@ func (fq *FastQueue) tryWriteBlock(block []byte, ignoreDisabledPQ bool) bool {
 
 	isPQWriteAllowed := !fq.isPQDisabled || ignoreDisabledPQ
 
-	fq.flushInmemoryBlocksToFileIfNeededLocked()
-	if n := fq.pq.GetPendingBytes(); n > 0 {
-		// The file-based queue isn't drained yet. This means that in-memory queue cannot be used yet.
-		// So put the block to file-based queue.
-		if len(fq.ch) > 0 {
-			logger.Panicf("BUG: the in-memory queue must be empty when the file-based queue is non-empty; it contains %d pending bytes", n)
+	if !isPQWriteAllowed && fq.pq.GetPendingBytes() > 0 {
+		// fast path: there is pending data at file-based queue,
+		// it must be drained before in-memory queue could be used.
+		// File-based queue could be non-empty after vmagent restart
+		// and vmagent couldn't flush in-memory queue during shutdown.
+		return false
+	}
+	if !fq.prioritizeInMemoryData {
+		fq.flushInmemoryBlocksToFileIfNeededLocked()
+		if fq.pq.GetPendingBytes() > 0 {
+			// The file-based queue isn't drained yet. This means that in-memory queue cannot be used yet.
+			// So put the block to file-based queue.
+
+			if !isPQWriteAllowed {
+				return false
+			}
+			fq.pq.MustWriteBlock(block)
+			return true
 		}
-		if !isPQWriteAllowed {
-			return false
-		}
-		fq.pq.MustWriteBlock(block)
-		return true
 	}
 	if len(fq.ch) == cap(fq.ch) {
 		// There is no space left in the in-memory queue. Put the data to file-based queue.
@@ -216,7 +226,7 @@ func (fq *FastQueue) tryWriteBlock(block []byte, ignoreDisabledPQ bool) bool {
 		fq.pq.MustWriteBlock(block)
 		return true
 	}
-	// Fast path - put the block to in-memory queue.
+
 	bb := blockBufPool.Get()
 	bb.B = append(bb.B[:0], block...)
 	fq.ch <- bb
@@ -239,16 +249,15 @@ func (fq *FastQueue) MustReadBlock(dst []byte) ([]byte, bool) {
 		if fq.stopDeadline > 0 && fasttime.UnixTimestamp() > fq.stopDeadline {
 			return dst, false
 		}
-		if len(fq.ch) > 0 {
-			return fq.mustReadInMemoryBlockLocked(dst), true
-		}
 		if n := fq.pq.GetPendingBytes(); n > 0 {
 			data, ok := fq.pq.MustReadBlockNonblocking(dst)
 			if ok {
 				return data, true
 			}
 			dst = data
-			continue
+		}
+		if len(fq.ch) > 0 {
+			return fq.mustReadInMemoryBlockLocked(dst), true
 		}
 		if fq.stopDeadline > 0 {
 			return dst, false
@@ -259,6 +268,27 @@ func (fq *FastQueue) MustReadBlock(dst []byte) ([]byte, bool) {
 	}
 }
 
+// MustReadInMemoryBlockBlocking reads the next block from the in-memory queue into dst and returns it.
+// It blocks until a block is available or the stop deadline is exceeded, in which case it returns (dst, false).
+func (fq *FastQueue) MustReadInMemoryBlockBlocking(dst []byte) ([]byte, bool) {
+	fq.mu.Lock()
+	defer fq.mu.Unlock()
+
+	for {
+		if fq.stopDeadline > 0 && fasttime.UnixTimestamp() > fq.stopDeadline {
+			return dst, false
+		}
+		if len(fq.ch) > 0 {
+			return fq.mustReadInMemoryBlockLocked(dst), true
+		}
+		if fq.stopDeadline > 0 {
+			return dst, false
+		}
+		// There are no blocks. Wait for new block.
+		fq.cond.Wait()
+	}
+}
+
 // MustReadInMemoryBlock reads the next block from the in-memory queue into dst and returns it.
 // It returns (dst, true) if a block was available, or (nil, false) if the in-memory queue is empty.
 // It does not block waiting for new blocks.
@@ -277,9 +307,6 @@ func (fq *FastQueue) mustReadInMemoryBlockLocked(dst []byte) []byte {
 	if len(fq.ch) == 0 {
 		logger.Panicf("BUG: the function must not be called when in-memory queue is empty. Caller should verify the queue len upfront")
 	}
-	if n := fq.pq.GetPendingBytes(); n > 0 {
-		logger.Panicf("BUG: the file-based queue must be empty when the in-memory queue is non-empty; it contains %d pending bytes", n)
-	}
 	bb := <-fq.ch
 	fq.pendingInmemoryBytes -= uint64(len(bb.B))
 	fq.lastInmemoryBlockReadTime = fasttime.UnixTimestamp()

lib/persistentqueue/fastqueue_test.go

@@ -13,7 +13,7 @@ func TestFastQueueOpenClose(_ *testing.T) {
 	path := "fast-queue-open-close"
 	fs.MustRemoveDir(path)
 	for range 10 {
-		fq := MustOpenFastQueue(path, "foobar", 100, 0, false)
+		fq := MustOpenFastQueue(path, "foobar", 100, 0, false, false)
 		fq.MustClose()
 	}
 	fs.MustRemoveDir(path)
@@ -24,7 +24,7 @@ func TestFastQueueWriteReadInmemory(t *testing.T) {
 	fs.MustRemoveDir(path)
 
 	capacity := 100
-	fq := MustOpenFastQueue(path, "foobar", capacity, 0, false)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, false, false)
 	if n := fq.GetInmemoryQueueLen(); n != 0 {
 		t.Fatalf("unexpected non-zero inmemory queue size:  %d", n)
 	}
@@ -57,7 +57,7 @@ func TestFastQueueWriteReadMixed(t *testing.T) {
 	fs.MustRemoveDir(path)
 
 	capacity := 100
-	fq := MustOpenFastQueue(path, "foobar", capacity, 0, false)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, false, false)
 	if n := fq.GetPendingBytes(); n != 0 {
 		t.Fatalf("the number of pending bytes must be 0; got %d", n)
 	}
@@ -93,7 +93,7 @@ func TestFastQueueWriteReadWithCloses(t *testing.T) {
 	fs.MustRemoveDir(path)
 
 	capacity := 100
-	fq := MustOpenFastQueue(path, "foobar", capacity, 0, false)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, false, false)
 	if n := fq.GetPendingBytes(); n != 0 {
 		t.Fatalf("the number of pending bytes must be 0; got %d", n)
 	}
@@ -106,7 +106,7 @@ func TestFastQueueWriteReadWithCloses(t *testing.T) {
 
 		blocks = append(blocks, block)
 		fq.MustClose()
-		fq = MustOpenFastQueue(path, "foobar", capacity, 0, false)
+		fq = MustOpenFastQueue(path, "foobar", capacity, 0, false, false)
 	}
 	if n := fq.GetPendingBytes(); n == 0 {
 		t.Fatalf("the number of pending bytes must be greater than 0")
@@ -120,7 +120,7 @@ func TestFastQueueWriteReadWithCloses(t *testing.T) {
 			t.Fatalf("unexpected block read; got %q; want %q", buf, block)
 		}
 		fq.MustClose()
-		fq = MustOpenFastQueue(path, "foobar", capacity, 0, false)
+		fq = MustOpenFastQueue(path, "foobar", capacity, 0, false, false)
 	}
 	if n := fq.GetPendingBytes(); n != 0 {
 		t.Fatalf("the number of pending bytes must be 0; got %d", n)
@@ -133,7 +133,7 @@ func TestFastQueueReadUnblockByClose(t *testing.T) {
 	path := "fast-queue-read-unblock-by-close"
 	fs.MustRemoveDir(path)
 
-	fq := MustOpenFastQueue(path, "foorbar", 123, 0, false)
+	fq := MustOpenFastQueue(path, "foorbar", 123, 0, false, false)
 	resultCh := make(chan error)
 	go func() {
 		data, ok := fq.MustReadBlock(nil)
@@ -163,7 +163,7 @@ func TestFastQueueReadUnblockByWrite(t *testing.T) {
 	path := "fast-queue-read-unblock-by-write"
 	fs.MustRemoveDir(path)
 
-	fq := MustOpenFastQueue(path, "foobar", 13, 0, false)
+	fq := MustOpenFastQueue(path, "foobar", 13, 0, false, false)
 	block := "foodsafdsaf sdf"
 	resultCh := make(chan error)
 	go func() {
@@ -197,7 +197,7 @@ func TestFastQueueReadWriteConcurrent(t *testing.T) {
 	path := "fast-queue-read-write-concurrent"
 	fs.MustRemoveDir(path)
 
-	fq := MustOpenFastQueue(path, "foobar", 5, 0, false)
+	fq := MustOpenFastQueue(path, "foobar", 5, 0, false, false)
 
 	var blocks []string
 	blocksMap := make(map[string]bool)
@@ -259,7 +259,7 @@ func TestFastQueueReadWriteConcurrent(t *testing.T) {
 	readersWG.Wait()
 
 	// Collect the remaining data
-	fq = MustOpenFastQueue(path, "foobar", 5, 0, false)
+	fq = MustOpenFastQueue(path, "foobar", 5, 0, false, false)
 	resultCh := make(chan error)
 	go func() {
 		for len(blocksMap) > 0 {
@@ -293,7 +293,7 @@ func TestFastQueueWriteReadWithDisabledPQ(t *testing.T) {
 	fs.MustRemoveDir(path)
 
 	capacity := 20
-	fq := MustOpenFastQueue(path, "foobar", capacity, 0, true)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, true, false)
 	if n := fq.GetInmemoryQueueLen(); n != 0 {
 		t.Fatalf("unexpected non-zero inmemory queue size:  %d", n)
 	}
@@ -310,7 +310,7 @@ func TestFastQueueWriteReadWithDisabledPQ(t *testing.T) {
 	}
 
 	fq.MustClose()
-	fq = MustOpenFastQueue(path, "foobar", capacity, 0, true)
+	fq = MustOpenFastQueue(path, "foobar", capacity, 0, true, false)
 	for _, block := range blocks {
 		buf, ok := fq.MustReadBlock(nil)
 		if !ok {
@@ -329,7 +329,7 @@ func TestFastQueueWriteReadWithIgnoreDisabledPQ(t *testing.T) {
 	fs.MustRemoveDir(path)
 
 	capacity := 20
-	fq := MustOpenFastQueue(path, "foobar", capacity, 0, true)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, true, false)
 	if n := fq.GetInmemoryQueueLen(); n != 0 {
 		t.Fatalf("unexpected non-zero inmemory queue size:  %d", n)
 	}
@@ -351,7 +351,7 @@ func TestFastQueueWriteReadWithIgnoreDisabledPQ(t *testing.T) {
 	}
 
 	fq.MustClose()
-	fq = MustOpenFastQueue(path, "foobar", capacity, 0, true)
+	fq = MustOpenFastQueue(path, "foobar", capacity, 0, true, false)
 	for _, block := range blocks {
 		buf, ok := fq.MustReadBlock(nil)
 		if !ok {

lib/persistentqueue/fastqueue_timing_test.go

@@ -17,7 +17,7 @@ func BenchmarkFastQueueThroughputSerial(b *testing.B) {
 			b.SetBytes(int64(blockSize) * iterationsCount)
 			path := fmt.Sprintf("bench-fast-queue-throughput-serial-%d", blockSize)
 			fs.MustRemoveDir(path)
-			fq := MustOpenFastQueue(path, "foobar", iterationsCount*2, 0, false)
+			fq := MustOpenFastQueue(path, "foobar", iterationsCount*2, 0, false, false)
 			defer func() {
 				fq.MustClose()
 				fs.MustRemoveDir(path)
@@ -38,7 +38,7 @@ func BenchmarkFastQueueThroughputConcurrent(b *testing.B) {
 			b.SetBytes(int64(blockSize) * iterationsCount)
 			path := fmt.Sprintf("bench-fast-queue-throughput-concurrent-%d", blockSize)
 			fs.MustRemoveDir(path)
-			fq := MustOpenFastQueue(path, "foobar", iterationsCount*cgroup.AvailableCPUs()*2, 0, false)
+			fq := MustOpenFastQueue(path, "foobar", iterationsCount*cgroup.AvailableCPUs()*2, 0, false, false)
 			defer func() {
 				fq.MustClose()
 				fs.MustRemoveDir(path)