Update lib/persistentqueue/persistentqueue.go

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

.github/workflows/build.yml

@@ -65,7 +65,7 @@ jobs:
             arch: amd64
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           persist-credentials: false
 

.github/workflows/changelog-linter.yml

@@ -13,7 +13,7 @@ jobs:
       contents: read
     runs-on: 'ubuntu-latest'
     steps:
-      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           # needed for proper diff
           fetch-depth: 0

.github/workflows/check-commit-signed.yml

@@ -12,7 +12,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           fetch-depth: 0  # we need full history for commit verification
           persist-credentials: false

.github/workflows/check-licenses.yml

@@ -17,7 +17,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           persist-credentials: false
 

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

@@ -31,7 +31,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           persist-credentials: false
 
@@ -54,14 +54,14 @@ jobs:
           restore-keys: go-artifacts-${{ runner.os }}-codeql-analyze-${{ steps.go.outputs.go-version }}-
 
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@e46ed2cbd01164d986452f91f178727624ae40d7  # v4.35.3
+        uses: github/codeql-action/init@8aad20d150bbac5944a9f9d289da16a4b0d87c1e  # v4.36.2
         with:
           languages: go
 
       - name: Autobuild
-        uses: github/codeql-action/autobuild@e46ed2cbd01164d986452f91f178727624ae40d7  # v4.35.3
+        uses: github/codeql-action/autobuild@8aad20d150bbac5944a9f9d289da16a4b0d87c1e  # v4.36.2
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@e46ed2cbd01164d986452f91f178727624ae40d7  # v4.35.3
+        uses: github/codeql-action/analyze@8aad20d150bbac5944a9f9d289da16a4b0d87c1e  # v4.36.2
         with:
           category: 'language:go'

.github/workflows/docs.yaml

@@ -18,13 +18,13 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           path: __vm
           persist-credentials: false
 
       - name: Checkout private code
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           repository: VictoriaMetrics/vmdocs
           token: ${{ secrets.VM_BOT_GH_TOKEN }}

.github/workflows/test.yml

@@ -34,7 +34,7 @@ jobs:
     runs-on: 'vm-runner'
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           persist-credentials: false
 
@@ -79,7 +79,7 @@ jobs:
 
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           persist-credentials: false
 
@@ -106,7 +106,7 @@ jobs:
 
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           persist-credentials: false
 

.github/workflows/vmui.yml

@@ -32,7 +32,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:
           persist-credentials: false
 

app/vmagent/remotewrite/remotewrite.go

@@ -1064,7 +1064,7 @@ func (rwctx *remoteWriteCtx) TryPushTimeSeries(tss []prompb.TimeSeries, forceDro
 
 	copyTimeSeriesIfNeeded := func() {
 		if v == nil {
-			v := tssPool.Get().(*[]prompb.TimeSeries)
+			v = tssPool.Get().(*[]prompb.TimeSeries)
 			tss = append(*v, tss...)
 		}
 	}

app/vmui/packages/vmui/src/components/ExploreAlerts/BaseAlert/index.tsx

@@ -98,7 +98,7 @@ const BaseAlert = ({ item, group }: BaseAlertProps) => {
       {!!Object.keys(item.annotations || {}).length && (
         <>
           <span className="vm-alerts-title">Annotations</span>
-          <table>
+          <table className="vm-annotations-table">
             <colgroup>
               <col className="vm-col-md"/>
               <col/>

app/vmui/packages/vmui/src/components/ExploreAlerts/BaseRule/index.tsx

@@ -121,7 +121,7 @@ const BaseRule = ({ item, group }: BaseRuleProps) => {
       {!!Object.keys(item?.annotations || {}).length && (
         <>
           <span className="vm-alerts-title">Annotations</span>
-          <table>
+          <table className="vm-annotations-table">
             <colgroup>
               <col className="vm-col-md"/>
               <col/>

app/vmui/packages/vmui/src/pages/ExploreAlerts/style.scss

@@ -62,6 +62,13 @@
     padding-right: 40px;
   }
 
+  .vm-annotations-table {
+    tbody > tr > td {
+      vertical-align: top;
+      white-space: pre-line;
+    }
+  }
+
   display: flex;
   flex-direction: column;
   align-items: stretch;

docs/victoriametrics/Articles.md

@@ -114,6 +114,8 @@ See also [case studies](https://docs.victoriametrics.com/victoriametrics/casestu
 * [QCon London 2026: Wrangling Telemetry at Scale, a Guide to Self-Hosted Observability](https://www.infoq.com/news/2026/03/self-hosted-observability/)
 * [How We Made Telemetry Queries 10x Faster: Chunk-Split Caching for Metrics, Logs, and Traces](https://mirastacklabs.ai/blog/chunk-split-caching/)
 * [Claude Code: creating Kubernetes debugging AI Agent for VictoriaMetrics](https://rtfm.co.ua/en/claude-code-creating-kubernetes-debugging-ai-agent-for-victoriametrics/)
+* [OpenTelemetry: OTel Collectors in Kubernetes and VictoriaMetrics Stack integration](https://itnext.io/opentelemetry-otel-collectors-in-kubernetes-and-victoriametrics-stack-integration-d907ed0a15a0)
+* [VictoriaMetrics vs Prometheus: my default, and when I still pick Prometheus](https://jorijn.com/en/blog/victoriametrics-vs-prometheus/)
 
 ## Third-party articles and slides about VictoriaLogs
 

docs/victoriametrics/changelog/CHANGELOG.md

@@ -28,6 +28,7 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 
 * FEATURE: [MetricsQL](https://docs.victoriametrics.com/victoriametrics/metricsql/): support `fill` modifiers to allow missing series on either side of a binary operation to be filled with a provided default value. See [#10598](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10598).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): Improve background discovery performance for [http_sd](https://docs.victoriametrics.com/victoriametrics/sd_configs/#http_sd_configs) discovery. See [#8838](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8838).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): restore broken persistent queue chunk file from the last valid written block in case of ungraceful vmagent shutdown. See [#11192](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11192).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): allow overriding `max_scrape_size` on a per-target basis via the `__max_scrape_size__` label during target relabeling. See [#11188](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11188).
 * FEATURE: [vmstorage](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): add `-maxBackfillAge` command-line flag for limiting ingestion of samples with historical timestamps, for example, when older data has been moved between storage tiers (nvme/hdd, hot/cold). See [#11199](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11199). Thanks to @AshwinRamaniPsg for contribution.
 
@@ -37,6 +38,8 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * BUGFIX: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): keep only one header navigation dropdown (`Explore`, `Tools`) open at a time. Previously, hovering across two dropdowns could briefly leave both open due to the close delay. See [#11224](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11224). Thanks to @antedotee for contribution.
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): fix a possible data race when processing OpenTelemetry metadata. See [#11238](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11238). Thanks to @nevgeny for contribution.
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): flush pending persistent queue data to chunk file before updating the metadata. This prevents the metadata writer offset from getting ahead of the chunk file size and avoids losing the persistent queue after an unclean shutdown. See [#11192](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11192).
+* BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): fix increased CPU and memory usage when `-remoteWrite.urlRelabelConfig` or `-remoteWrite.streamAggr.config` flags are used. The bug was introduced in [#10854](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/10854) and existed since [v1.147.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.147.0). See [#11250](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11250).
+* BUGFIX: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): preserve newline formatting in alert and rule annotations on the Alerting page. See [#11171](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11171).
 
 ## [v1.147.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.147.0)
 
@@ -58,6 +61,7 @@ Released at 2026-07-06
 * FEATURE: [MetricsQL](https://docs.victoriametrics.com/victoriametrics/metricsql/): improve the selection algorithm of [buckets_limit](https://docs.victoriametrics.com/victoriametrics/metricsql/#buckets_limit) to remove consecutive empty buckets at the beginning and end to obtain more accurate min and max values. See [#10417](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10417).
 * FEATURE: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): expose `vmalert_group_rule_results_limit` metric to indicate the number of alerts or recording results that a single rule within the group can produce. See [#11179](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11179). Thanks to @vinyas-bharadwaj for the contribution.
 * FEATURE: [alerts](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules): add `AlertingRuleResultsApproachingLimit` and `RecordingRuleResultsApproachingLimit` alerting rules to [vmalert alerts](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules/alerts-vmalert.yml). These alerts notify when a rule's last evaluation samples exceed 90% of the configured group results limit. See [#11179](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11179). Thanks to @vinyas-bharadwaj for the contribution.
+* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): add support for [Monitoring Data eXchange (MDX)](https://docs.victoriametrics.com/victoriametrics/vmagent/#monitoring-data-exchange): the ability to route only metrics from VictoriaMetrics services to a specific `-remoteWrite.url`. MDX is useful for building monitoring-of-monitoring where one remote storage should receive the full metric stream and another should receive only VictoriaMetrics metrics. Enable per destination with `-remoteWrite.mdx.enable=true`. See [#10600](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10600).
 
 * BUGFIX: all VictoriaMetrics components: cancel in-flight HTTP requests shortly before `-http.maxGracefulShutdownDuration` elapses during graceful shutdown, so they can drain and the shutdown completes cleanly within that window instead of timing out and exiting via `logger.Fatalf` -> `os.Exit`. This prevents skipping the storage flush and losing in-memory data when long-lived requests are in flight (such as VictoriaLogs live tailing). See [#1502](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1502).
 * BUGFIX: `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): properly check values range for the limits configured with flags `-maxLabelsPerTimeseries`, `-maxLabelNameLen` and `-maxLabelValueLen`. It must be in range `1..65535`. See [#11128](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11128).
@@ -80,7 +84,6 @@ Released at 2026-06-22
 * FEATURE: [vmctl](https://docs.victoriametrics.com/victoriametrics/vmctl/): add `-vm-headers` and `-vm-bearer-token` flags for authenticating requests to the VictoriaMetrics import destination. The flags are available in `opentsdb`, `influx`, `remote-read`, `prometheus`, `mimir`, and `thanos` vmctl sub-commands. See [#8897](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8897).
 * FEATURE: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): log calls to [/api/v1/admin/tsdb/delete_series](https://docs.victoriametrics.com/victoriametrics/url-examples/#apiv1admintsdbdelete_series) API handler. This should help to identify events of metrics deletion from the database. See [#11104](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11104).
 * FEATURE: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): add the `last` value to graph legend statistics. See [#10759](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/10759).
-* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): add support for [Monitoring Data eXchange (MDX)](https://docs.victoriametrics.com/victoriametrics/vmagent/#monitoring-data-exchange): the ability to route only metrics from VictoriaMetrics services to a specific `-remoteWrite.url`. MDX is useful for building monitoring-of-monitoring where one remote storage should receive the full metric stream and another should receive only VictoriaMetrics metrics. Enable per destination with `-remoteWrite.mdx.enable=true`. See [#10600](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10600).
 
 * BUGFIX: [enterprise](https://docs.victoriametrics.com/enterprise/) [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmstorage` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): properly expose metric `vm_retention_filters_partitions_scheduled_rows`. See [#11138](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11138)
 * BUGFIX: [stream aggregation](https://docs.victoriametrics.com/victoriametrics/stream-aggregation/): fix issue with producing aggregated samples with identical timestamps between flushes. See [#10808](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/10808).

docs/victoriametrics/stream-aggregation/README.md

@@ -645,9 +645,9 @@ See also [why you shouldn't put an aggregator behind a load balancer](https://do
 
 # Troubleshooting
 
-- [Unexpected spikes for `total` or `increase` outputs](#data-delay-and-staleness).
+- [Unexpected spikes for `total` or `increase` outputs](#staleness).
 - [Excessively large values for `total*`, `increase*`, and `rate*` outputs](#counter-resets).
-- [Lower than expected values for `total_prometheus` and `increase_prometheus` outputs](#data-delay-and-staleness).
+- [Lower than expected values for `total_prometheus` and `increase_prometheus` outputs](#staleness).
 - [High memory usage and CPU usage](#high-resource-usage).
 - [Unexpected results in vmagent cluster mode](#cluster-mode).
 - [Inaccurate aggregation results for histograms](#aggregation-windows)

lib/persistentqueue/persistentqueue.go

@@ -25,6 +25,8 @@ const MaxBlockSize = 32 * 1024 * 1024
 // DefaultChunkFileSize represents default chunk file size
 const DefaultChunkFileSize = (MaxBlockSize + 8) * 16
 
+const blockHeaderSize = 8
+
 var chunkFileNameRegex = regexp.MustCompile("^[0-9A-F]{16}$")
 
 // queue represents persistent queue.
@@ -129,7 +131,7 @@ func mustOpen(path, name string, maxPendingBytes int64) *queue {
 }
 
 func mustOpenInternal(path, name string, chunkFileSize, maxBlockSize, maxPendingBytes uint64) *queue {
-	if chunkFileSize < 8 || chunkFileSize-8 < maxBlockSize {
+	if chunkFileSize < blockHeaderSize || chunkFileSize-blockHeaderSize < maxBlockSize {
 		logger.Panicf("BUG: too small chunkFileSize=%d for maxBlockSize=%d; chunkFileSize must fit at least one block", chunkFileSize, maxBlockSize)
 	}
 	if maxBlockSize <= 0 {
@@ -280,14 +282,30 @@ func tryOpeningQueue(path, name string, chunkFileSize, maxBlockSize, maxPendingB
 			q.writerFlushedOffset = mi.WriterOffset
 			if fileSize := fs.MustFileSize(q.writerPath); fileSize != q.writerLocalOffset {
 				if fileSize < q.writerLocalOffset {
-					logger.Errorf("%q size (%d bytes) is smaller than the writer offset (%d bytes); removing the file",
-						q.writerPath, fileSize, q.writerLocalOffset)
-					fs.MustRemovePath(q.writerPath)
-					continue
+					validChunkFileSize := mustGetChunkValidDataSize(q.writerPath, q.maxBlockSize)
+					logger.Warnf("%q size (%d bytes) is smaller than the writer offset (%d bytes); "+
+						"this may be the case on unclean shutdown (OOM, `kill -9`, hardware reset); resetting writer to fileSize: %d",
+						q.writerPath, fileSize, q.writerLocalOffset, validChunkFileSize)
+					mi.WriterOffset = offset + validChunkFileSize
+					q.writerOffset = mi.WriterOffset
+					q.writerLocalOffset = mi.WriterOffset % q.chunkFileSize
+					q.writerFlushedOffset = mi.WriterOffset
+					// The recovered writer offset may end up smaller than mi.ReaderOffset.
+					// Do not clamp the reader offset in this case: it means the reader has already
+					// consumed data that is now lost, so the remaining queue state cannot be trusted.
+					// The readerOffset > writerOffset check below will fail queue opening,
+					// and the caller (mustOpen) will drop the queue and recreate it from scratch.
+				} else {
+					logger.Warnf("%q size (%d bytes) is bigger than writer offset (%d bytes); "+
+						"this may be the case on unclean shutdown (OOM, `kill -9`, hardware reset); trying to fix it by adjusting fileSize to %d",
+						q.writerPath, fileSize, q.writerLocalOffset, q.writerLocalOffset)
+				}
+				if err := os.Truncate(q.writerPath, int64(q.writerLocalOffset)); err != nil {
+					logger.Panicf("FATAL: cannot truncate chunk: %q to size: %d: %s", q.writerPath, q.writerLocalOffset, err)
+				}
+				if err := mi.WriteToFile(metainfoPath); err != nil {
+					logger.Panicf("FATAL: cannot update metainfo file: %q: %s", metainfoPath, err)
 				}
-				logger.Warnf("%q size (%d bytes) is bigger than writer offset (%d bytes); "+
-					"this may be the case on unclean shutdown (OOM, `kill -9`, hardware reset); trying to fix it by adjusting fileSize to %d",
-					q.writerPath, fileSize, q.writerLocalOffset, q.writerLocalOffset)
 			}
 			w, err := filestream.OpenWriterAt(q.writerPath, int64(q.writerLocalOffset), false)
 			if err != nil {
@@ -357,7 +375,7 @@ func (q *queue) MustWriteBlock(block []byte) {
 	}
 	if q.maxPendingBytes > 0 {
 		// Drain the oldest blocks until the number of pending bytes becomes enough for the block.
-		blockSize := uint64(len(block) + 8)
+		blockSize := uint64(len(block) + blockHeaderSize)
 		maxPendingBytes := q.maxPendingBytes
 		if blockSize < maxPendingBytes {
 			maxPendingBytes -= blockSize
@@ -395,7 +413,7 @@ func (q *queue) writeBlock(block []byte) error {
 	defer func() {
 		writeDurationSeconds.Add(time.Since(startTime).Seconds())
 	}()
-	if q.writerLocalOffset+q.maxBlockSize+8 > q.chunkFileSize {
+	if q.writerLocalOffset+q.maxBlockSize+blockHeaderSize > q.chunkFileSize {
 		if err := q.nextChunkFileForWrite(); err != nil {
 			return fmt.Errorf("cannot create next chunk file: %w", err)
 		}
@@ -408,7 +426,7 @@ func (q *queue) writeBlock(block []byte) error {
 	err := q.write(header.B)
 	headerBufPool.Put(header)
 	if err != nil {
-		return fmt.Errorf("cannot write header with size 8 bytes to %q: %w", q.writerPath, err)
+		return fmt.Errorf("cannot write header with size %d bytes to %q: %w", blockHeaderSize, q.writerPath, err)
 	}
 
 	// Write block contents.
@@ -469,7 +487,7 @@ func (q *queue) readBlock(dst []byte) ([]byte, error) {
 	defer func() {
 		readDurationSeconds.Add(time.Since(startTime).Seconds())
 	}()
-	if q.readerLocalOffset+q.maxBlockSize+8 > q.chunkFileSize {
+	if q.readerLocalOffset+q.maxBlockSize+blockHeaderSize > q.chunkFileSize {
 		if err := q.nextChunkFileForRead(); err != nil {
 			return dst, fmt.Errorf("cannot open next chunk file: %w", err)
 		}
@@ -478,12 +496,12 @@ func (q *queue) readBlock(dst []byte) ([]byte, error) {
 again:
 	// Read block len.
 	header := headerBufPool.Get()
-	header.B = bytesutil.ResizeNoCopyMayOverallocate(header.B, 8)
+	header.B = bytesutil.ResizeNoCopyMayOverallocate(header.B, blockHeaderSize)
 	err := q.readFull(header.B)
 	blockLen := encoding.UnmarshalUint64(header.B)
 	headerBufPool.Put(header)
 	if err != nil {
-		logger.Errorf("skipping corrupted %q, since header with size 8 bytes cannot be read from it: %s", q.readerPath, err)
+		logger.Errorf("skipping corrupted %q, since header with size %d bytes cannot be read from it: %s", q.readerPath, blockHeaderSize, err)
 		if err := q.skipBrokenChunkFile(); err != nil {
 			return dst, err
 		}
@@ -669,3 +687,39 @@ func (mi *metainfo) ReadFromFile(path string) error {
 	}
 	return nil
 }
+
+func mustGetChunkValidDataSize(filePath string, maxBlockSize uint64) uint64 {
+	f, err := os.Open(filePath)
+	if err != nil {
+		logger.Panicf("FATAL: cannot open file: %s", err)
+	}
+	defer fs.MustClose(f)
+
+	fi, err := f.Stat()
+	if err != nil {
+		logger.Panicf("FATAL: cannot read file stat: %s", err)
+	}
+	fileSize := uint64(fi.Size())
+
+	var offset uint64
+	var header [blockHeaderSize]byte
+	for {
+		if offset+blockHeaderSize > fileSize {
+			return offset
+		}
+		if _, err := io.ReadFull(f, header[:]); err != nil {
+			return offset
+		}
+		blockLen := encoding.UnmarshalUint64(header[:])
+		if blockLen == 0 || blockLen > maxBlockSize {
+			return offset
+		}
+		if offset+blockHeaderSize+blockLen > fileSize {
+			return offset
+		}
+		offset += blockHeaderSize + blockLen
+		if _, err := f.Seek(int64(offset), io.SeekStart); err != nil {
+			return offset
+		}
+	}
+}

lib/persistentqueue/persistentqueue_test.go

@@ -137,6 +137,243 @@ func TestQueueOpen(t *testing.T) {
 		q.MustClose()
 		fs.MustRemoveDir(path)
 	})
+	t.Run("damaged-writer-file-tail", func(t *testing.T) {
+		path := "damaged-writer-file-tail"
+		fs.MustRemoveDir(path)
+
+		q := mustOpen(path, "foobar", 0)
+		block1 := []byte("valid block 1")
+		block2 := []byte("valid block 2")
+		q.MustWriteBlock(block1)
+		q.MustWriteBlock(block2)
+		q.MustClose()
+
+		chunkPath := filepath.Join(path, fmt.Sprintf("%016X", 0))
+		chunkFileSize := fs.MustFileSize(chunkPath)
+
+		if err := os.Truncate(chunkPath, int64(chunkFileSize)-1); err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		q = mustOpen(path, "foobar", 0)
+
+		var buf []byte
+		var ok bool
+		buf, ok = q.MustReadBlockNonblocking(buf[:0])
+		if !ok {
+			t.Fatalf("expected block to be readable")
+		}
+		if string(buf) != string(block1) {
+			t.Fatalf("unexpected block got: %q, want: %q", buf, block1)
+		}
+
+		_, ok = q.MustReadBlockNonblocking(buf[:0])
+		if ok {
+			t.Fatalf("expected second block to be dropped")
+		}
+		q.MustClose()
+
+		expectedChunkFileSize := uint64(blockHeaderSize + len(block1))
+		chunkFileSize = fs.MustFileSize(chunkPath)
+		if chunkFileSize != expectedChunkFileSize {
+			t.Fatalf("unexpected chunk file size: got %d; want %d", chunkFileSize, expectedChunkFileSize)
+		}
+
+		fs.MustRemoveDir(path)
+	})
+	t.Run("damaged-writer-file-extra-tail", func(t *testing.T) {
+		path := "damaged-writer-file-extra-tail"
+		fs.MustRemoveDir(path)
+
+		q := mustOpen(path, "foobar", 0)
+		blocks := [][]byte{
+			[]byte("valid block 1"),
+			[]byte("valid block 2"),
+		}
+		for _, block := range blocks {
+			q.MustWriteBlock(block)
+		}
+		q.MustClose()
+
+		chunkPath := filepath.Join(path, fmt.Sprintf("%016X", 0))
+		originChunkFileSize := fs.MustFileSize(chunkPath)
+
+		f, err := os.OpenFile(chunkPath, os.O_WRONLY|os.O_APPEND, 0)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if _, err := f.Write([]byte("corrupted block")); err != nil {
+			t.Fatalf("cannot append corrupted tail: %s", err)
+		}
+		if err := f.Close(); err != nil {
+			t.Fatalf("cannot close chunk file: %s", err)
+		}
+
+		q = mustOpen(path, "foobar", 0)
+
+		var buf []byte
+		var ok bool
+		for _, block := range blocks {
+			buf, ok = q.MustReadBlockNonblocking(buf[:0])
+			if !ok {
+				t.Fatalf("expected block to be readable")
+			}
+			if string(buf) != string(block) {
+				t.Fatalf("unexpected block got: %q, want: %q", buf, block)
+			}
+		}
+		q.MustClose()
+		chunkFileSize := fs.MustFileSize(chunkPath)
+		if chunkFileSize != originChunkFileSize {
+			t.Fatalf("unexpected chunk file size: got %d; want %d", chunkFileSize, originChunkFileSize)
+		}
+		fs.MustRemoveDir(path)
+	})
+	t.Run("damaged-writer-file-partial-header", func(t *testing.T) {
+		path := "damaged-writer-file-partial-header"
+		fs.MustRemoveDir(path)
+
+		q := mustOpen(path, "foobar", 0)
+		block1 := []byte("valid block 1")
+		block2 := []byte("valid block 2")
+		q.MustWriteBlock(block1)
+		q.MustWriteBlock(block2)
+		q.MustClose()
+
+		chunkPath := filepath.Join(path, fmt.Sprintf("%016X", 0))
+		// Keep block1 intact plus only 3 bytes of the second block's header.
+		newSize := int64(blockHeaderSize + len(block1) + 3)
+		if err := os.Truncate(chunkPath, newSize); err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		q = mustOpen(path, "foobar", 0)
+		buf, ok := q.MustReadBlockNonblocking(nil)
+		if !ok {
+			t.Fatalf("expected block to be readable")
+		}
+		if string(buf) != string(block1) {
+			t.Fatalf("unexpected block got: %q, want: %q", buf, block1)
+		}
+		if _, ok := q.MustReadBlockNonblocking(buf[:0]); ok {
+			t.Fatalf("expected second block to be dropped")
+		}
+		q.MustClose()
+
+		expectedChunkFileSize := uint64(blockHeaderSize + len(block1))
+		if chunkFileSize := fs.MustFileSize(chunkPath); chunkFileSize != expectedChunkFileSize {
+			t.Fatalf("unexpected chunk file size: got %d; want %d", chunkFileSize, expectedChunkFileSize)
+		}
+		fs.MustRemoveDir(path)
+	})
+	t.Run("damaged-writer-file-corrupted-header", func(t *testing.T) {
+		path := "damaged-writer-file-corrupted-header"
+		fs.MustRemoveDir(path)
+
+		q := mustOpen(path, "foobar", 0)
+		block1 := []byte("valid block 1")
+		block2 := []byte("valid block 2")
+		block3 := []byte("valid block 3")
+		q.MustWriteBlock(block1)
+		q.MustWriteBlock(block2)
+		q.MustWriteBlock(block3)
+		q.MustClose()
+
+		chunkPath := filepath.Join(path, fmt.Sprintf("%016X", 0))
+		chunkFileSize := fs.MustFileSize(chunkPath)
+
+		// Overwrite the second block's header with an impossible blockLen
+		// and drop the last byte so the recovery scan is triggered.
+		f, err := os.OpenFile(chunkPath, os.O_WRONLY, 0)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		garbage := [blockHeaderSize]byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF}
+		if _, err := f.WriteAt(garbage[:], int64(blockHeaderSize+len(block1))); err != nil {
+			t.Fatalf("cannot corrupt block header: %s", err)
+		}
+		if err := f.Close(); err != nil {
+			t.Fatalf("cannot close chunk file: %s", err)
+		}
+		if err := os.Truncate(chunkPath, int64(chunkFileSize)-1); err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		q = mustOpen(path, "foobar", 0)
+		buf, ok := q.MustReadBlockNonblocking(nil)
+		if !ok {
+			t.Fatalf("expected block to be readable")
+		}
+		if string(buf) != string(block1) {
+			t.Fatalf("unexpected block got: %q, want: %q", buf, block1)
+		}
+		if _, ok := q.MustReadBlockNonblocking(buf[:0]); ok {
+			t.Fatalf("expected remaining blocks to be dropped")
+		}
+		q.MustClose()
+
+		expectedChunkFileSize := uint64(blockHeaderSize + len(block1))
+		if chunkFileSize := fs.MustFileSize(chunkPath); chunkFileSize != expectedChunkFileSize {
+			t.Fatalf("unexpected chunk file size: got %d; want %d", chunkFileSize, expectedChunkFileSize)
+		}
+		fs.MustRemoveDir(path)
+	})
+	t.Run("damaged-writer-file-second-chunk", func(t *testing.T) {
+		path := "damaged-writer-file-second-chunk"
+		fs.MustRemoveDir(path)
+
+		const maxBlockSize = 64
+		const chunkFileSize = (maxBlockSize + blockHeaderSize) * 2
+
+		newBlock := func(fill byte) []byte {
+			b := make([]byte, 32)
+			for i := range b {
+				b[i] = fill
+			}
+			return b
+		}
+		block1 := newBlock('1')
+		block2 := newBlock('2')
+		block3 := newBlock('3')
+		block4 := newBlock('4')
+
+		q := mustOpenInternal(path, "foobar", chunkFileSize, maxBlockSize, 0)
+		q.MustWriteBlock(block1)
+		q.MustWriteBlock(block2)
+		// blocks 3 and 4 land in the second chunk file
+		q.MustWriteBlock(block3)
+		q.MustWriteBlock(block4)
+		q.MustClose()
+
+		secondChunkPath := filepath.Join(path, fmt.Sprintf("%016X", chunkFileSize))
+		secondChunkSize := fs.MustFileSize(secondChunkPath)
+		if err := os.Truncate(secondChunkPath, int64(secondChunkSize)-1); err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		q = mustOpenInternal(path, "foobar", chunkFileSize, maxBlockSize, 0)
+		var buf []byte
+		var ok bool
+		for _, block := range [][]byte{block1, block2, block3} {
+			buf, ok = q.MustReadBlockNonblocking(buf[:0])
+			if !ok {
+				t.Fatalf("expected block to be readable")
+			}
+			if string(buf) != string(block) {
+				t.Fatalf("unexpected block got: %q, want: %q", buf, block)
+			}
+		}
+		if _, ok := q.MustReadBlockNonblocking(buf[:0]); ok {
+			t.Fatalf("expected last block to be dropped")
+		}
+		q.MustClose()
+
+		expectedSize := uint64(blockHeaderSize + len(block3))
+		if gotSize := fs.MustFileSize(secondChunkPath); gotSize != expectedSize {
+			t.Fatalf("unexpected second chunk file size: got %d; want %d", gotSize, expectedSize)
+		}
+		fs.MustRemoveDir(path)
+	})
 }
 
 func TestQueueResetIfEmpty(t *testing.T) {

lib/storage/storage_synctest_test.go

@@ -551,7 +551,7 @@ func TestStorageAddRows_nextDayIndexPrefill(t *testing.T) {
 		nextDaySlowInserts := m.SlowPerDayIndexInserts
 		slowInserts := nextDaySlowInserts - currDaySlowInserts
 		if slowInserts >= numSeries {
-			t.Errorf("unexpected amount of slow inserts: got %d, want < %d", slowInserts, numSeries)
+			t.Fatalf("unexpected amount of slow inserts: got %d, want < %d", slowInserts, numSeries)
 		}
 
 	})
@@ -842,7 +842,7 @@ func TestStorageLastPartitionMetrics(t *testing.T) {
 		s.AddRows(mrs, defaultPrecisionBits)
 		s.DebugFlush()
 		if got := s.newTimeseriesCreated.Load(); got != want {
-			t.Errorf("unexpected number of new timeseries: got %d, want %d", got, want)
+			t.Fatalf("unexpected number of new timeseries: got %d, want %d", got, want)
 		}
 		// wait for merged parts to be attached to the table
 		time.Sleep(time.Minute)
@@ -1218,7 +1218,7 @@ func TestStorage_denyQueriesOutsideRetention(t *testing.T) {
 		slices.Sort(gotData)
 
 		if diff := cmp.Diff(wantData, gotData); diff != "" {
-			t.Errorf("unexpected tag value suffixes (-want, +got):\n%s", diff)
+			t.Fatalf("unexpected tag value suffixes (-want, +got):\n%s", diff)
 		}
 	}
 
@@ -1487,3 +1487,100 @@ func TestStorageAddRows_MaxBackfillAge(t *testing.T) {
 		s.MustClose()
 	})
 }
+
+func TestStorageAddRows_currHourMetricIDs(t *testing.T) {
+	defer testRemoveAll(t)
+
+	f := func(t *testing.T, disablePerDayIndex bool) {
+		synctest.Test(t, func(t *testing.T) {
+			s := MustOpenStorage(t.Name(), OpenOptions{
+				DisablePerDayIndex: disablePerDayIndex,
+			})
+			defer s.MustClose()
+
+			now := time.Now().UTC()
+			currHourTR := TimeRange{
+				MinTimestamp: time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.UTC).UnixMilli(),
+				MaxTimestamp: time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 59, 59, 999_999_999, time.UTC).UnixMilli(),
+			}
+			currHour := uint64(currHourTR.MinTimestamp / 1000 / 3600)
+			prevHourTR := TimeRange{
+				MinTimestamp: currHourTR.MinTimestamp - msecPerHour,
+				MaxTimestamp: currHourTR.MaxTimestamp - msecPerHour,
+			}
+			rng := rand.New(rand.NewSource(1))
+
+			// Test current hour metricIDs population when data ingestion takes the
+			// slow path. The database is empty, therefore the index and the
+			// tsidCache contain no metricIDs, therefore the data ingestion will
+			// take slow path.
+
+			mrs := testGenerateMetricRowsWithPrefix(rng, 1000, "slow_path", currHourTR)
+			s.AddRows(mrs, defaultPrecisionBits)
+			s.DebugFlush()
+			s.updateCurrHourMetricIDs(currHour)
+			if got, want := s.currHourMetricIDs.Load().m.Len(), 1000; got != want {
+				t.Fatalf("[slow path] unexpected current hour metric ID count: got %d, want %d", got, want)
+			}
+
+			// Test current hour metricIDs population when data ingestion takes the
+			// fast path (when the metricIDs are found in the tsidCache)
+
+			// First insert samples to populate the tsidCache. The samples belong to
+			// the previous hour, therefore the metricIDs won't be added to
+			// currHourMetricIDs.
+			mrs = testGenerateMetricRowsWithPrefix(rng, 1000, "fast_path", prevHourTR)
+			s.AddRows(mrs, defaultPrecisionBits)
+			s.DebugFlush()
+			s.updateCurrHourMetricIDs(currHour)
+			if got, want := s.currHourMetricIDs.Load().m.Len(), 1000; got != want {
+				t.Fatalf("[fast path] unexpected current hour metric ID count after ingesting samples for previous hour: got %d, want %d", got, want)
+			}
+
+			// Now ingest the same metrics. This time the metricIDs will be found in
+			// tsidCache so the ingestion will take the fast path.
+			mrs = testGenerateMetricRowsWithPrefix(rng, 1000, "fast_path", currHourTR)
+			s.AddRows(mrs, defaultPrecisionBits)
+			s.DebugFlush()
+			s.updateCurrHourMetricIDs(currHour)
+			if got, want := s.currHourMetricIDs.Load().m.Len(), 2000; got != want {
+				t.Fatalf("[fast path] unexpected current hour metric ID count: got %d, want %d", got, want)
+			}
+
+			// Test current hour metricIDs population when data ingestion takes the
+			// slower path (when the metricIDs are not found in the tsidCache but
+			// found in the index)
+
+			// First insert samples to populate the index. The samples belong to
+			// the previous hour, therefore the metricIDs won't be added to
+			// currHourMetricIDs.
+			mrs = testGenerateMetricRowsWithPrefix(rng, 1000, "slower_path", prevHourTR)
+			s.AddRows(mrs, defaultPrecisionBits)
+			s.DebugFlush()
+			s.updateCurrHourMetricIDs(currHour)
+			if got, want := s.currHourMetricIDs.Load().m.Len(), 2000; got != want {
+				t.Fatalf("[slower path] unexpected current hour metric ID count after ingesting samples for previous hour: got %d, want %d", got, want)
+			}
+			// Inserted samples were also added to the tsidCache. Drop it to
+			// enforce the fallback to index search.
+			s.resetAndSaveTSIDCache()
+
+			// Now ingest the same metrics. This time the metricIDs will be searched
+			// and found in index so the ingestion will take the slower path.
+			mrs = testGenerateMetricRowsWithPrefix(rng, 1000, "slower_path", currHourTR)
+			s.AddRows(mrs, defaultPrecisionBits)
+			s.DebugFlush()
+			s.updateCurrHourMetricIDs(currHour)
+			if got, want := s.currHourMetricIDs.Load().m.Len(), 3000; got != want {
+				t.Fatalf("[slower path] unexpected current hour metric ID count: got %d, want %d", got, want)
+			}
+		})
+	}
+
+	t.Run("disablePerDayIndex=false", func(t *testing.T) {
+		f(t, false)
+	})
+	t.Run("disablePerDayIndex=true", func(t *testing.T) {
+		f(t, true)
+	})
+}

lib/storage/storage_test.go

@@ -3477,103 +3477,6 @@ func TestStorageAddRows_SamplesWithZeroDate(t *testing.T) {
 	})
 }
 
-func TestStorageAddRows_currHourMetricIDs(t *testing.T) {
-	defer testRemoveAll(t)
-
-	f := func(t *testing.T, disablePerDayIndex bool) {
-		t.Helper()
-
-		s := MustOpenStorage(t.Name(), OpenOptions{
-			DisablePerDayIndex: disablePerDayIndex,
-		})
-		defer s.MustClose()
-
-		now := time.Now().UTC()
-		currHourTR := TimeRange{
-			MinTimestamp: time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.UTC).UnixMilli(),
-			MaxTimestamp: time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 59, 59, 999_999_999, time.UTC).UnixMilli(),
-		}
-		currHour := uint64(currHourTR.MinTimestamp / 1000 / 3600)
-		prevHourTR := TimeRange{
-			MinTimestamp: currHourTR.MinTimestamp - 3600*1000,
-			MaxTimestamp: currHourTR.MaxTimestamp - 3600*1000,
-		}
-		rng := rand.New(rand.NewSource(1))
-
-		// Test current hour metricIDs population when data ingestion takes the
-		// slow path. The database is empty, therefore the index and the
-		// tsidCache contain no metricIDs, therefore the data ingestion will
-		// take slow path.
-
-		mrs := testGenerateMetricRowsWithPrefix(rng, 1000, "slow_path", currHourTR)
-		s.AddRows(mrs, defaultPrecisionBits)
-		s.DebugFlush()
-		s.updateCurrHourMetricIDs(currHour)
-		if got, want := s.currHourMetricIDs.Load().m.Len(), 1000; got != want {
-			t.Errorf("[slow path] unexpected current hour metric ID count: got %d, want %d", got, want)
-		}
-
-		// Test current hour metricIDs population when data ingestion takes the
-		// fast path (when the metricIDs are found in the tsidCache)
-
-		// First insert samples to populate the tsidCache. The samples belong to
-		// the previous hour, therefore the metricIDs won't be added to
-		// currHourMetricIDs.
-		mrs = testGenerateMetricRowsWithPrefix(rng, 1000, "fast_path", prevHourTR)
-		s.AddRows(mrs, defaultPrecisionBits)
-		s.DebugFlush()
-		s.updateCurrHourMetricIDs(currHour)
-		if got, want := s.currHourMetricIDs.Load().m.Len(), 1000; got != want {
-			t.Errorf("[fast path] unexpected current hour metric ID count after ingesting samples for previous hour: got %d, want %d", got, want)
-		}
-
-		// Now ingest the same metrics. This time the metricIDs will be found in
-		// tsidCache so the ingestion will take the fast path.
-		mrs = testGenerateMetricRowsWithPrefix(rng, 1000, "fast_path", currHourTR)
-		s.AddRows(mrs, defaultPrecisionBits)
-		s.DebugFlush()
-		s.updateCurrHourMetricIDs(currHour)
-		if got, want := s.currHourMetricIDs.Load().m.Len(), 2000; got != want {
-			t.Errorf("[fast path] unexpected current hour metric ID count: got %d, want %d", got, want)
-		}
-
-		// Test current hour metricIDs population when data ingestion takes the
-		// slower path (when the metricIDs are not found in the tsidCache but
-		// found in the index)
-
-		// First insert samples to populate the index. The samples belong to
-		// the previous hour, therefore the metricIDs won't be added to
-		// currHourMetricIDs.
-		mrs = testGenerateMetricRowsWithPrefix(rng, 1000, "slower_path", prevHourTR)
-		s.AddRows(mrs, defaultPrecisionBits)
-		s.DebugFlush()
-		s.updateCurrHourMetricIDs(currHour)
-		if got, want := s.currHourMetricIDs.Load().m.Len(), 2000; got != want {
-			t.Errorf("[slower path] unexpected current hour metric ID count after ingesting samples for previous hour: got %d, want %d", got, want)
-		}
-		// Inserted samples were also added to the tsidCache. Drop it to
-		// enforce the fallback to index search.
-		s.resetAndSaveTSIDCache()
-
-		// Now ingest the same metrics. This time the metricIDs will be searched
-		// and found in index so the ingestion will take the slower path.
-		mrs = testGenerateMetricRowsWithPrefix(rng, 1000, "slower_path", currHourTR)
-		s.AddRows(mrs, defaultPrecisionBits)
-		s.DebugFlush()
-		s.updateCurrHourMetricIDs(currHour)
-		if got, want := s.currHourMetricIDs.Load().m.Len(), 3000; got != want {
-			t.Errorf("[slower path] unexpected current hour metric ID count: got %d, want %d", got, want)
-		}
-	}
-
-	t.Run("disablePerDayIndex=false", func(t *testing.T) {
-		f(t, false)
-	})
-	t.Run("disablePerDayIndex=true", func(t *testing.T) {
-		f(t, true)
-	})
-}
-
 // testSearchMetricIDs returns metricIDs for the given tfss and tr.
 //
 // The returned metricIDs are sorted. The function panics in in case of error.