From f9a2a74c090a16ba755d52664f8b65e920fc5199 Mon Sep 17 00:00:00 2001 From: Maksym Tsymbarov Date: Fri, 26 Jun 2026 14:08:03 +0200 Subject: [PATCH] Fixed real-time aggregation window end to track wall-clock time --- ui-ngx/src/app/core/api/data-aggregator.ts | 27 ++++++++++++---------- 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/ui-ngx/src/app/core/api/data-aggregator.ts b/ui-ngx/src/app/core/api/data-aggregator.ts index 9a648bbcaf..8cd761f3dc 100644 --- a/ui-ngx/src/app/core/api/data-aggregator.ts +++ b/ui-ngx/src/app/core/api/data-aggregator.ts @@ -310,22 +310,25 @@ export class DataAggregator { const delta = Math.floor(this.elapsed / this.aggregationTimeout); if (delta || !this.data || dataChanged) { const tickTs = delta * this.aggregationTimeout; + let currentEndTs: number; if (this.subsTw.quickInterval) { const startEndTime = calculateIntervalStartEndTime(this.subsTw.quickInterval, this.subsTw.timezone); this.startTs = startEndTime[0] + this.subsTw.tsOffset; this.endTs = startEndTime[1] + this.subsTw.tsOffset; + currentEndTs = this.endTs; } else { this.startTs += tickTs; this.endTs += tickTs; + currentEndTs = this.endTs + (this.elapsed - tickTs); } if (this.subsTw.aggregation.type !== AggregationType.NONE) { - this.updateLastInterval(); + this.updateLastInterval(currentEndTs); } - this.data = this.updateData(); + this.data = this.updateData(currentEndTs); this.elapsed = this.elapsed - delta * this.aggregationTimeout; } } else { - this.data = this.updateData(); + this.data = this.updateData(this.endTs); } if (this.onDataCb && (!this.ignoreDataUpdateOnIntervalTick || this.updatedData)) { this.onDataCb(this.data, detectChanges); @@ -336,7 +339,7 @@ export class DataAggregator { } } - private updateData(): IndexedData { + private updateData(currentEndTs: number): IndexedData { this.dataBuffer = []; this.tsKeys.forEach((key) => { if (!this.dataBuffer[key.id]) { @@ -358,7 +361,7 @@ export class DataAggregator { } deletedKeys.push(aggStartTs); this.updatedData = true; - } else if (aggData.ts < this.endTs || noAggregation) { + } else if (aggData.ts < currentEndTs || noAggregation) { const kvPair: DataEntry = [aggData.ts, aggData.aggValue, aggData.interval]; keyData.push(kvPair); } @@ -366,7 +369,7 @@ export class DataAggregator { deletedKeys.forEach(ts => aggKeyData.delete(ts)); keyData.sort((set1, set2) => set1[0] - set2[0]); if (this.subsTw.aggregation.stateData) { - this.updateStateBounds(keyData, deepClone(this.lastPrevKvPairData[id])); + this.updateStateBounds(keyData, currentEndTs, deepClone(this.lastPrevKvPairData[id])); } if (keyData.length > this.subsTw.aggregation.limit) { keyData = keyData.slice(keyData.length - this.subsTw.aggregation.limit); @@ -376,7 +379,7 @@ export class DataAggregator { return this.dataBuffer; } - private updateStateBounds(keyData: DataSet, lastPrevKvPair: DataEntry) { + private updateStateBounds(keyData: DataSet, currentEndTs: number, lastPrevKvPair: DataEntry) { if (lastPrevKvPair) { lastPrevKvPair[0] = this.startTs; lastPrevKvPair[2] = [this.startTs, this.startTs]; @@ -397,10 +400,10 @@ export class DataAggregator { } if (keyData.length) { let lastKvPair = keyData[keyData.length - 1]; - if (lastKvPair[0] < this.endTs) { + if (lastKvPair[0] < currentEndTs) { lastKvPair = deepClone(lastKvPair); - lastKvPair[0] = this.endTs; - lastKvPair[2] = [this.endTs, this.endTs]; + lastKvPair[0] = currentEndTs; + lastKvPair[2] = [currentEndTs, currentEndTs]; keyData.push(lastKvPair); } } @@ -479,11 +482,11 @@ export class DataAggregator { } } - private updateLastInterval() { + private updateLastInterval(currentEndTs: number) { for (const idStr of Object.keys(this.aggregationMap.aggMap)) { const id = Number(idStr); const aggKeyData = this.aggregationMap.aggMap[id]; - aggKeyData.updateLastInterval(this.endTs); + aggKeyData.updateLastInterval(currentEndTs); } }