update comment

.golangci.yml

@@ -1,9 +1,18 @@
 version: "2"
 linters:
+  enable:
+    - errorlint
   settings:
     errcheck:
       exclude-functions:
         - (net/http.ResponseWriter).Write
+    errorlint:
+      errorf: true
+      # Do not enable `comparison` and `asserts`: they produce false positives,
+      # since many call sites intentionally compare sentinel errors directly (e.g. err == io.EOF)
+      # when the producer is documented to return them unwrapped. See https://github.com/VictoriaMetrics/VictoriaLogs/pull/1490
+      comparison: false
+      asserts: false
   exclusions:
     generated: lax
     presets:

app/vmagent/remotewrite/client.go

@@ -204,13 +204,9 @@ 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 + inmemoryWorkers)
+		return float64(concurrency)
 	})
-	for range inmemoryWorkers {
-		c.wg.Go(c.runWorkerForInmemoryQueue)
-	}
 	for range concurrency {
 		c.wg.Go(c.runWorker)
 	}
@@ -352,52 +348,6 @@ 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,9 +66,6 @@ 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 "+
@@ -926,7 +923,7 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, sanitizedURL string)
 	if maxInmemoryBlocks < 2 {
 		maxInmemoryBlocks = 2
 	}
-	fq := persistentqueue.MustOpenFastQueue(queuePath, sanitizedURL, maxInmemoryBlocks, maxPendingBytes, isPQDisabled, inmemoryQueueWorkers.GetOptionalArg(argIdx) > 0)
+	fq := persistentqueue.MustOpenFastQueue(queuePath, sanitizedURL, maxInmemoryBlocks, maxPendingBytes, isPQDisabled)
 	_ = 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, false)
+		fq := persistentqueue.MustOpenFastQueue(path, "test", 100, 0, false)
 		defer fs.MustRemoveDir(path)
 		defer fq.MustClose()
 

app/vmctl/utils.go

@@ -74,9 +74,9 @@ func wrapErr(vmErr *vm.ImportError, verbose bool) error {
 		verboseMsg = "(enable `--verbose` output to get more details)"
 	}
 	if vmErr.Err == nil {
-		return fmt.Errorf("%s\n\tLatest delivered batch for timestamps range %d - %d %s\n%s",
+		return fmt.Errorf("%w\n\tLatest delivered batch for timestamps range %d - %d %s\n%s",
 			vmErr.Err, minTS, maxTS, verboseMsg, errTS)
 	}
-	return fmt.Errorf("%s\n\tImporting batch failed for timestamps range %d - %d %s\n%s",
+	return fmt.Errorf("%w\n\tImporting batch failed for timestamps range %d - %d %s\n%s",
 		vmErr.Err, minTS, maxTS, verboseMsg, errTS)
 }

apptest/client.go

@@ -156,14 +156,14 @@ func readAllAndClose(t *testing.T, responseBody io.ReadCloser) string {
 //
 // This type is expected to be embedded by the apps that serve metrics.
 type metricsClient struct {
-	cli *Client
-	url string
+	metricsCli *Client
+	url        string
 }
 
 func newMetricsClient(cli *Client, addr string) *metricsClient {
 	return &metricsClient{
-		cli: cli,
-		url: fmt.Sprintf("http://%s/metrics", addr),
+		metricsCli: cli,
+		url:        fmt.Sprintf("http://%s/metrics", addr),
 	}
 }
 
@@ -179,7 +179,7 @@ func (c *metricsClient) GetIntMetric(t *testing.T, metricName string) int {
 func (c *metricsClient) GetMetric(t *testing.T, metricName string) float64 {
 	t.Helper()
 
-	metrics, statusCode := c.cli.Get(t, c.url, nil)
+	metrics, statusCode := c.metricsCli.Get(t, c.url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -205,7 +205,7 @@ func (c *metricsClient) GetMetricsByPrefix(t *testing.T, prefix string) []float6
 
 	values := []float64{}
 
-	metrics, statusCode := c.cli.Get(t, c.url, nil)
+	metrics, statusCode := c.metricsCli.Get(t, c.url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -234,7 +234,7 @@ func (c *metricsClient) GetMetricsByRegexp(t *testing.T, re *regexp.Regexp) []fl
 
 	values := []float64{}
 
-	metrics, statusCode := c.cli.Get(t, c.url, nil)
+	metrics, statusCode := c.metricsCli.Get(t, c.url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -270,7 +270,7 @@ func (c *metricsClient) rpcRowsSentTotal(t *testing.T) int {
 }
 
 type vmselectClient struct {
-	cli                      *Client
+	vmselectCli              *Client
 	url                      func(op, path string, opts QueryOpts) string
 	metricNamesStatsResetURL string
 	tenantsURL               string
@@ -287,7 +287,7 @@ func (c *vmselectClient) PrometheusAPIV1Export(t *testing.T, query string, opts
 	values := opts.asURLValues()
 	values.Add("match[]", query)
 	values.Add("format", "promapi")
-	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1QueryResponse(t, res)
 }
 
@@ -302,7 +302,7 @@ func (c *vmselectClient) PrometheusAPIV1ExportNative(t *testing.T, query string,
 	values := opts.asURLValues()
 	values.Add("match[]", query)
 	values.Add("format", "promapi")
-	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	return []byte(res)
 }
 
@@ -315,7 +315,7 @@ func (c *vmselectClient) PrometheusAPIV1Query(t *testing.T, query string, opts Q
 	url := c.url("select", "prometheus/api/v1/query", opts)
 	values := opts.asURLValues()
 	values.Add("query", query)
-	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1QueryResponse(t, res)
 }
 
@@ -329,7 +329,7 @@ func (c *vmselectClient) PrometheusAPIV1QueryRange(t *testing.T, query string, o
 	url := c.url("select", "prometheus/api/v1/query_range", opts)
 	values := opts.asURLValues()
 	values.Add("query", query)
-	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1QueryResponse(t, res)
 }
 
@@ -342,7 +342,7 @@ func (c *vmselectClient) PrometheusAPIV1Series(t *testing.T, matchQuery string,
 	url := c.url("select", "prometheus/api/v1/series", opts)
 	values := opts.asURLValues()
 	values.Add("match[]", matchQuery)
-	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1SeriesResponse(t, res)
 }
 
@@ -354,7 +354,7 @@ func (c *vmselectClient) PrometheusAPIV1SeriesCount(t *testing.T, opts QueryOpts
 	t.Helper()
 	url := c.url("select", "prometheus/api/v1/series/count", opts)
 	values := opts.asURLValues()
-	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1SeriesCountResponse(t, res)
 }
 
@@ -367,7 +367,7 @@ func (c *vmselectClient) PrometheusAPIV1Labels(t *testing.T, matchQuery string,
 	url := c.url("select", "prometheus/api/v1/labels", opts)
 	values := opts.asURLValues()
 	values.Add("match[]", matchQuery)
-	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1LabelsResponse(t, res)
 }
 
@@ -382,7 +382,7 @@ func (c *vmselectClient) PrometheusAPIV1LabelValues(t *testing.T, labelName, mat
 	url := c.url("select", path, opts)
 	values := opts.asURLValues()
 	values.Add("match[]", matchQuery)
-	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1LabelValuesResponse(t, res)
 }
 
@@ -394,7 +394,7 @@ func (c *vmselectClient) PrometheusAPIV1Metadata(t *testing.T, metric string, li
 	values := opts.asURLValues()
 	values.Add("metric", metric)
 	values.Add("limit", strconv.Itoa(limit))
-	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1Metadata(t, res)
 }
 
@@ -408,7 +408,7 @@ func (c *vmselectClient) PrometheusAPIV1AdminTSDBDeleteSeries(t *testing.T, matc
 	url := c.url("delete", "prometheus/api/v1/admin/tsdb/delete_series", opts)
 	values := opts.asURLValues()
 	values.Add("match[]", matchQuery)
-	res, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
+	res, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusNoContent {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusNoContent, res)
 	}
@@ -426,7 +426,7 @@ func (c *vmselectClient) PrometheusAPIV1StatusMetricNamesStats(t *testing.T, lim
 	values.Add("limit", limit)
 	values.Add("le", le)
 	values.Add("match_pattern", matchPattern)
-	res, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
+	res, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
 	}
@@ -455,7 +455,7 @@ func (c *vmselectClient) PrometheusAPIV1StatusTSDB(t *testing.T, matchQuery stri
 	addNonEmpty("match[]", matchQuery)
 	addNonEmpty("topN", topN)
 	addNonEmpty("date", date)
-	res, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
+	res, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
 	}
@@ -476,7 +476,7 @@ func (c *vmselectClient) GraphiteMetricsIndex(t *testing.T, opts QueryOpts) Grap
 	t.Helper()
 
 	url := c.url("select", "graphite/metrics/index.json", opts)
-	res, statusCode := c.cli.Get(t, url, opts.Headers)
+	res, statusCode := c.vmselectCli.Get(t, url, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
 	}
@@ -499,7 +499,7 @@ func (c *vmselectClient) GraphiteMetricsFind(t *testing.T, query string, opts Qu
 	url := c.url("select", "graphite/metrics/find", opts)
 	values := opts.asURLValues()
 	values.Add("query", query)
-	resText, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
+	resText, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, resText)
 	}
@@ -522,7 +522,7 @@ func (c *vmselectClient) GraphiteMetricsExpand(t *testing.T, query string, opts
 	url := c.url("select", "graphite/metrics/expand", opts)
 	values := opts.asURLValues()
 	values.Add("query", query)
-	resText, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
+	resText, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, resText)
 	}
@@ -546,7 +546,7 @@ func (c *vmselectClient) GraphiteRender(t *testing.T, target string, opts QueryO
 	values := opts.asURLValues()
 	values.Add("format", "json")
 	values.Add("target", target)
-	resText, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
+	resText, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, resText)
 	}
@@ -567,7 +567,7 @@ func (c *vmselectClient) GraphiteTagsTagSeries(t *testing.T, record string, opts
 	url := c.url("select", "graphite/tags/tagSeries", opts)
 	values := opts.asURLValues()
 	values.Add("path", record)
-	_, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
+	_, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	if got, want := statusCode, http.StatusNotImplemented; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d", got, want)
 	}
@@ -584,7 +584,7 @@ func (c *vmselectClient) GraphiteTagsTagMultiSeries(t *testing.T, records []stri
 	for _, rec := range records {
 		values.Add("path", rec)
 	}
-	_, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
+	_, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
 	if got, want := statusCode, http.StatusNotImplemented; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d", got, want)
 	}
@@ -598,7 +598,7 @@ func (c *vmselectClient) GraphiteTagsTagMultiSeries(t *testing.T, records []stri
 func (c *vmselectClient) PrometheusAPIV1AdminStatusMetricNamesStatsReset(t *testing.T, opts QueryOpts) {
 	t.Helper()
 	values := opts.asURLValues()
-	res, statusCode := c.cli.PostForm(t, c.metricNamesStatsResetURL, values, opts.Headers)
+	res, statusCode := c.vmselectCli.PostForm(t, c.metricNamesStatsResetURL, values, opts.Headers)
 	if statusCode != http.StatusNoContent {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusNoContent, res)
 	}
@@ -608,7 +608,7 @@ func (c *vmselectClient) PrometheusAPIV1AdminStatusMetricNamesStatsReset(t *test
 // /admin/tenants endpoint.
 func (c *vmselectClient) APIV1AdminTenants(t *testing.T, opts QueryOpts) *AdminTenantsResponse {
 	t.Helper()
-	res, statusCode := c.cli.Get(t, c.tenantsURL, opts.Headers)
+	res, statusCode := c.vmselectCli.Get(t, c.tenantsURL, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
 	}
@@ -622,7 +622,7 @@ func (c *vmselectClient) APIV1AdminTenants(t *testing.T, opts QueryOpts) *AdminT
 }
 
 type vminsertClient struct {
-	cli                *Client
+	vminsertCli        *Client
 	url                func(op, path string, opts QueryOpts) string
 	openTSDBURL        func(op, path string, opts QueryOpts) string
 	graphiteListenAddr string
@@ -647,7 +647,7 @@ func (c *vminsertClient) PrometheusAPIV1ImportCSV(t *testing.T, records []string
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "text/plain")
 	c.sendBlocking(t, len(records), func() {
-		_, statusCode := c.cli.Post(t, url, data, headers)
+		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -671,7 +671,7 @@ func (c *vminsertClient) PrometheusAPIV1ImportNative(t *testing.T, data []byte,
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "text/plain")
 	c.sendBlocking(t, 1, func() {
-		_, statusCode := c.cli.Post(t, url, data, headers)
+		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -693,7 +693,7 @@ func (c *vminsertClient) PrometheusAPIV1Write(t *testing.T, wr prompb.WriteReque
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "application/x-protobuf")
 	c.sendBlocking(t, recordsCount, func() {
-		_, statusCode := c.cli.Post(t, url, data, headers)
+		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -745,7 +745,7 @@ func (c *vminsertClient) PrometheusAPIV1ImportPrometheus(t *testing.T, records [
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "text/plain")
 	c.sendBlocking(t, recordsCount, func() {
-		_, statusCode := c.cli.Post(t, url, data, headers)
+		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -771,7 +771,7 @@ func (c *vminsertClient) InfluxWrite(t *testing.T, records []string, opts QueryO
 	headers.Set("Content-Type", "text/plain")
 	c.sendBlocking(t, len(records), func() {
 		t.Helper()
-		_, statusCode := c.cli.Post(t, url, data, headers)
+		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -805,7 +805,7 @@ func (c *vminsertClient) OpentelemetryV1Metrics(t *testing.T, md otlppb.MetricsD
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "application/x-protobuf")
 	c.sendBlocking(t, recordsCount, func() {
-		_, statusCode := c.cli.Post(t, url, data, headers)
+		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
 		if statusCode != http.StatusOK {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 		}
@@ -830,7 +830,7 @@ func (c *vminsertClient) OpenTSDBAPIPut(t *testing.T, records []string, opts Que
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "application/json")
 	c.sendBlocking(t, len(records), func() {
-		_, statusCode := c.cli.Post(t, url, data, headers)
+		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -853,7 +853,7 @@ func (c *vminsertClient) ZabbixConnectorHistory(t *testing.T, records []string,
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "application/json")
 	c.sendBlocking(t, len(records), func() {
-		_, statusCode := c.cli.Post(t, url, data, headers)
+		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
 		if statusCode != http.StatusOK {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 		}
@@ -867,11 +867,11 @@ func (c *vminsertClient) ZabbixConnectorHistory(t *testing.T, records []string,
 // See https://docs.victoriametrics.com/victoriametrics/integrations/graphite/#ingesting
 func (c *vminsertClient) GraphiteWrite(t *testing.T, records []string, _ QueryOpts) {
 	t.Helper()
-	c.cli.Write(t, c.graphiteListenAddr, records)
+	c.vminsertCli.Write(t, c.graphiteListenAddr, records)
 }
 
 type vmstorageClient struct {
-	cli            *Client
+	vmstorageCli   *Client
 	httpListenAddr string
 }
 
@@ -881,7 +881,7 @@ func (c *vmstorageClient) ForceFlush(t *testing.T) {
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/internal/force_flush", c.httpListenAddr)
-	_, statusCode := c.cli.Get(t, url, nil)
+	_, statusCode := c.vmstorageCli.Get(t, url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -892,7 +892,7 @@ func (c *vmstorageClient) ForceMerge(t *testing.T) {
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/internal/force_merge", c.httpListenAddr)
-	_, statusCode := c.cli.Get(t, url, nil)
+	_, statusCode := c.vmstorageCli.Get(t, url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -905,7 +905,7 @@ func (c *vmstorageClient) ForceMerge(t *testing.T) {
 func (c *vmstorageClient) SnapshotCreate(t *testing.T) *SnapshotCreateResponse {
 	t.Helper()
 
-	data, statusCode := c.cli.Post(t, c.SnapshotCreateURL(), nil, nil)
+	data, statusCode := c.vmstorageCli.Post(t, c.SnapshotCreateURL(), nil, nil)
 	if got, want := statusCode, http.StatusOK; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
 	}
@@ -931,7 +931,7 @@ func (c *vmstorageClient) APIV1AdminTSDBSnapshot(t *testing.T) *APIV1AdminTSDBSn
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/api/v1/admin/tsdb/snapshot", c.httpListenAddr)
-	data, statusCode := c.cli.Post(t, url, nil, nil)
+	data, statusCode := c.vmstorageCli.Post(t, url, nil, nil)
 	if got, want := statusCode, http.StatusOK; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
 	}
@@ -952,7 +952,7 @@ func (c *vmstorageClient) SnapshotList(t *testing.T) *SnapshotListResponse {
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/snapshot/list", c.httpListenAddr)
-	data, statusCode := c.cli.Get(t, url, nil)
+	data, statusCode := c.vmstorageCli.Get(t, url, nil)
 	if got, want := statusCode, http.StatusOK; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
 	}
@@ -973,7 +973,7 @@ func (c *vmstorageClient) SnapshotDelete(t *testing.T, snapshotName string) *Sna
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/snapshot/delete?snapshot=%s", c.httpListenAddr, snapshotName)
-	data, statusCode := c.cli.Delete(t, url)
+	data, statusCode := c.vmstorageCli.Delete(t, url)
 	wantStatusCodes := map[int]bool{
 		http.StatusOK:                  true,
 		http.StatusInternalServerError: true,
@@ -998,7 +998,7 @@ func (c *vmstorageClient) SnapshotDeleteAll(t *testing.T) *SnapshotDeleteAllResp
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/snapshot/delete_all", c.httpListenAddr)
-	data, statusCode := c.cli.Post(t, url, nil, nil)
+	data, statusCode := c.vmstorageCli.Post(t, url, nil, nil)
 	if got, want := statusCode, http.StatusOK; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
 	}

apptest/tests/vmagent_remotewrite_test.go

@@ -332,11 +332,13 @@ 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 {
@@ -639,102 +641,3 @@ 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
-		},
-	)
-}

apptest/vminsert.go

@@ -77,7 +77,7 @@ type vminsertRuntimeValues struct {
 func newVminsert(app *app, cli *Client, rt vminsertRuntimeValues) *Vminsert {
 	metricsClient := newMetricsClient(cli, rt.httpListenAddr)
 	vminsertClient := &vminsertClient{
-		cli: cli,
+		vminsertCli: cli,
 		url: func(op, path string, opts QueryOpts) string {
 			return getClusterPath(rt.httpListenAddr, op, path, opts)
 		},

apptest/vmselect.go

@@ -48,7 +48,7 @@ func newVmselect(app *app, cli *Client, rt vmselectRuntimeValues) *Vmselect {
 		app:           app,
 		metricsClient: newMetricsClient(cli, rt.httpListenAddr),
 		vmselectClient: &vmselectClient{
-			cli: cli,
+			vmselectCli: cli,
 			url: func(op, path string, opts QueryOpts) string {
 				return getClusterPath(rt.httpListenAddr, op, path, opts)
 			},

apptest/vmsingle.go

@@ -58,11 +58,11 @@ func newVmsingle(app *app, cli *Client, rt vmsingleRuntimeValues) *Vmsingle {
 		app:           app,
 		metricsClient: newMetricsClient(cli, rt.httpListenAddr),
 		vmstorageClient: &vmstorageClient{
-			cli:            cli,
+			vmstorageCli:   cli,
 			httpListenAddr: rt.httpListenAddr,
 		},
 		vmselectClient: &vmselectClient{
-			cli: cli,
+			vmselectCli: cli,
 			url: func(op, path string, opts QueryOpts) string {
 				return fmt.Sprintf("http://%s/%s", rt.httpListenAddr, path)
 			},
@@ -70,7 +70,7 @@ func newVmsingle(app *app, cli *Client, rt vmsingleRuntimeValues) *Vmsingle {
 			tenantsURL:               "vmsingle-does-not-serve-tenants",
 		},
 		vminsertClient: &vminsertClient{
-			cli: cli,
+			vminsertCli: cli,
 			url: func(_, path string, _ QueryOpts) string {
 				return fmt.Sprintf("http://%s/%s", rt.httpListenAddr, path)
 			},

apptest/vmstorage.go

@@ -63,7 +63,7 @@ func newVmstorage(app *app, cli *Client, rt vmstorageRuntimeValues) *Vmstorage {
 		app:           app,
 		metricsClient: newMetricsClient(cli, rt.httpListenAddr),
 		vmstorageClient: &vmstorageClient{
-			cli:            cli,
+			vmstorageCli:   cli,
 			httpListenAddr: rt.httpListenAddr,
 		},
 		storageDataPath: rt.storageDataPath,

docs/victoriametrics/changelog/CHANGELOG.md

@@ -26,8 +26,6 @@ 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,37 +928,6 @@ 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/mergeset/table_search_timing_test.go

@@ -94,7 +94,7 @@ func benchmarkTableSearchKeysExt(b *testing.B, tb *Table, keys [][]byte, stripSu
 				}
 				ts.Seek(searchKey)
 				if !ts.NextItem() {
-					panic(fmt.Errorf("BUG: NextItem must return true for searchKeys[%d]=%q; err=%v", i, searchKey, ts.Error()))
+					panic(fmt.Errorf("BUG: NextItem must return true for searchKeys[%d]=%q; err=%w", i, searchKey, ts.Error()))
 				}
 				if !bytes.HasPrefix(ts.Item, searchKey) {
 					panic(fmt.Errorf("BUG: unexpected item found for searchKey[%d]=%q; got %q; want %q", i, searchKey, ts.Item, key))

lib/persistentqueue/fastqueue.go

@@ -26,8 +26,6 @@ type FastQueue struct {
 	// isPQDisabled is set to true when pq is disabled.
 	isPQDisabled bool
 
-	prioritizeInMemoryData bool
-
 	// pq is file-based queue
 	pq *queue
 
@@ -50,13 +48,12 @@ 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, prioritizeInMemoryData bool) *FastQueue {
+func MustOpenFastQueue(path, name string, maxInmemoryBlocks int, maxPendingBytes int64, isPQDisabled bool) *FastQueue {
 	pq := mustOpen(path, name, maxPendingBytes)
 	fq := &FastQueue{
-		pq:                     pq,
-		isPQDisabled:           isPQDisabled,
-		prioritizeInMemoryData: prioritizeInMemoryData,
-		ch:                     make(chan *bytesutil.ByteBuffer, maxInmemoryBlocks),
+		pq:           pq,
+		isPQDisabled: isPQDisabled,
+		ch:           make(chan *bytesutil.ByteBuffer, maxInmemoryBlocks),
 	}
 	fq.cond.L = &fq.mu
 	fq.lastInmemoryBlockReadTime = fasttime.UnixTimestamp()
@@ -100,7 +97,7 @@ func (fq *FastQueue) IsWriteBlocked() bool {
 	}
 	fq.mu.Lock()
 	defer fq.mu.Unlock()
-	return len(fq.ch) == cap(fq.ch) || (fq.pq.GetPendingBytes() > 0 && !fq.prioritizeInMemoryData)
+	return len(fq.ch) == cap(fq.ch) || fq.pq.GetPendingBytes() > 0
 }
 
 // UnblockAllReaders unblocks all the readers.
@@ -197,25 +194,18 @@ func (fq *FastQueue) tryWriteBlock(block []byte, ignoreDisabledPQ bool) bool {
 
 	isPQWriteAllowed := !fq.isPQDisabled || ignoreDisabledPQ
 
-	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
+	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 {
+			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.
@@ -226,7 +216,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
@@ -249,15 +239,16 @@ 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
-		}
-		if len(fq.ch) > 0 {
-			return fq.mustReadInMemoryBlockLocked(dst), true
+			continue
 		}
 		if fq.stopDeadline > 0 {
 			return dst, false
@@ -268,27 +259,6 @@ 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.
@@ -307,6 +277,9 @@ 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, false)
+		fq := MustOpenFastQueue(path, "foobar", 100, 0, 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, false)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, 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, false)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, 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, false)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, 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, false)
+		fq = MustOpenFastQueue(path, "foobar", capacity, 0, 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, false)
+		fq = MustOpenFastQueue(path, "foobar", capacity, 0, 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, false)
+	fq := MustOpenFastQueue(path, "foorbar", 123, 0, 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, false)
+	fq := MustOpenFastQueue(path, "foobar", 13, 0, 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, false)
+	fq := MustOpenFastQueue(path, "foobar", 5, 0, 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, false)
+	fq = MustOpenFastQueue(path, "foobar", 5, 0, 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, false)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, true)
 	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, false)
+	fq = MustOpenFastQueue(path, "foobar", capacity, 0, true)
 	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, false)
+	fq := MustOpenFastQueue(path, "foobar", capacity, 0, true)
 	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, false)
+	fq = MustOpenFastQueue(path, "foobar", capacity, 0, true)
 	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, false)
+			fq := MustOpenFastQueue(path, "foobar", iterationsCount*2, 0, 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, false)
+			fq := MustOpenFastQueue(path, "foobar", iterationsCount*cgroup.AvailableCPUs()*2, 0, false)
 			defer func() {
 				fq.MustClose()
 				fs.MustRemoveDir(path)