lib/streamaggr: deduplicate data on total*, increase*, rate* outputs

docs/victoriametrics/changelog/CHANGELOG.md

@@ -26,6 +26,8 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 
 ## tip
 
+* BUGFIX: [stream aggregation](https://docs.victoriametrics.com/victoriametrics/stream-aggregation/): deduplicate data on `total`, `total_prometheus`, `increase`, `increase_prometheus`, `rate_sum` and `rate_avg` outputs.
+
 ## [v1.140.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.140.0)
 
 Released at 2026-04-10

lib/streamaggr/rate.go

@@ -103,7 +103,7 @@ func (av *rateAggrValue) pushSample(_ aggrConfig, sample *pushSample, key string
 		}
 		if sample.value >= sv.value {
 			state.increase += sample.value - sv.value
-		} else {
+		} else if sample.timestamp > state.timestamp {
 			// counter reset
 			state.increase += sample.value
 		}

lib/streamaggr/streamaggr_synctest_test.go

@@ -1,5 +1,3 @@
-//go:build synctest
-
 package streamaggr
 
 import (
@@ -347,9 +345,9 @@ foo:1m_total_prometheus 0
 	// total output for repeated series
 	f([]string{`
 foo 123
+bar{baz="qwe"} 2
 bar{baz="qwe"} 1.31
 bar{baz="qwe"} 4.34 1
-bar{baz="qwe"} 2
 foo{baz="qwe"} -5
 bar{baz="qwer"} 343
 bar{baz="qwer"} 344
@@ -366,14 +364,14 @@ foo:1m_total{baz="qwe"} 15
 	// total_prometheus output for repeated series
 	f([]string{`
 foo 123
-bar{baz="qwe"} 1.32
-bar{baz="qwe"} 4.34
+bar{baz="qwe"} 1
+bar{baz="qwe"} 4
 bar{baz="qwe"} 2
 foo{baz="qwe"} -5
 bar{baz="qwer"} 343
 bar{baz="qwer"} 344
 foo{baz="qwe"} 10
-`}, time.Minute, `bar:1m_total_prometheus{baz="qwe"} 5.02
+`}, time.Minute, `bar:1m_total_prometheus{baz="qwe"} 3
 bar:1m_total_prometheus{baz="qwer"} 1
 foo:1m_total_prometheus 0
 foo:1m_total_prometheus{baz="qwe"} 15
@@ -392,7 +390,7 @@ foo{baz="qwe"} -5
 bar{baz="qwer"} 343
 bar{baz="qwer"} 344
 foo{baz="qwe"} 10
-`}, time.Minute, `bar:1m_total 6.02
+`}, time.Minute, `bar:1m_total 4.02
 foo:1m_total 15
 `, `
 - interval: 1m
@@ -410,7 +408,7 @@ foo{baz="qwe"} -5
 bar{baz="qwer"} 343
 bar{baz="qwer"} 344
 foo{baz="qwe"} 10
-`}, time.Minute, `bar:1m_total_prometheus 6.02
+`}, time.Minute, `bar:1m_total_prometheus 4.02
 foo:1m_total_prometheus 15
 `, `
 - interval: 1m
@@ -443,14 +441,14 @@ foo:1m_increase_prometheus 0
 	// increase output for repeated series
 	f([]string{`
 foo 123
-bar{baz="qwe"} 1.32
-bar{baz="qwe"} 4.34
+bar{baz="qwe"} 1
+bar{baz="qwe"} 4
 bar{baz="qwe"} 2
 foo{baz="qwe"} -5
 bar{baz="qwer"} 343
 bar{baz="qwer"} 344
 foo{baz="qwe"} 10
-`}, time.Minute, `bar:1m_increase{baz="qwe"} 5.02
+`}, time.Minute, `bar:1m_increase{baz="qwe"} 3
 bar:1m_increase{baz="qwer"} 1
 foo:1m_increase 0
 foo:1m_increase{baz="qwe"} 15
@@ -462,14 +460,14 @@ foo:1m_increase{baz="qwe"} 15
 	// increase_prometheus output for repeated series
 	f([]string{`
 foo 123
-bar{baz="qwe"} 1.32
-bar{baz="qwe"} 4.34
+bar{baz="qwe"} 1
+bar{baz="qwe"} 4
 bar{baz="qwe"} 2
 foo{baz="qwe"} -5
 bar{baz="qwer"} 343
 bar{baz="qwer"} 344
 foo{baz="qwe"} 10
-`}, time.Minute, `bar:1m_increase_prometheus{baz="qwe"} 5.02
+`}, time.Minute, `bar:1m_increase_prometheus{baz="qwe"} 3
 bar:1m_increase_prometheus{baz="qwer"} 1
 foo:1m_increase_prometheus 0
 foo:1m_increase_prometheus{baz="qwe"} 15

lib/streamaggr/total.go

@@ -35,7 +35,7 @@ func (av *totalAggrValue) pushSample(c aggrConfig, sample *pushSample, key strin
 		}
 		if sample.value >= lv.value {
 			av.total += sample.value - lv.value
-		} else {
+		} else if sample.timestamp > lv.timestamp {
 			// counter reset
 			av.total += sample.value
 		}