fix(sdk-metrics): merge uncollected delta accumulations

CHANGELOG.md

@@ -26,6 +26,7 @@ For experimental package changes, see the [experimental CHANGELOG](experimental/
 
 * fix(core): added falsy check to make otel core work with browser where webpack config had process as false or null [#3613](https://github.com/open-telemetry/opentelemetry-js/issues/3613) @ravindra-dyte
 * fix(instrumentation-http): include query params in http.target [#3646](https://github.com/open-telemetry/opentelemetry-js/pull/3646) @kobi-co
+* fix(sdk-metrics): merge uncollected delta accumulations [#3667](https://github.com/open-telemetry/opentelemetry-js/pull/3667) @legendecas
 
 ### :books: (Refine Doc)
 

packages/sdk-metrics/src/state/DeltaMetricProcessor.ts

@@ -57,6 +57,7 @@ export class DeltaMetricProcessor<T extends Maybe<Accumulation>> {
           this._aggregator.createAccumulation(collectionTime);
         accumulation?.record(value);
         let delta = accumulation;
+        // Diff with recorded cumulative memo.
         if (this._cumulativeMemoStorage.has(attributes, hashCode)) {
           // has() returned true, previous is present.
           // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
@@ -66,6 +67,16 @@ export class DeltaMetricProcessor<T extends Maybe<Accumulation>> {
           )!;
           delta = this._aggregator.diff(previous, accumulation);
         }
+        // Merge with uncollected active delta.
+        if (this._activeCollectionStorage.has(attributes, hashCode)) {
+          // has() returned true, previous is present.
+          // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+          const active = this._activeCollectionStorage.get(
+            attributes,
+            hashCode
+          )!;
+          delta = this._aggregator.merge(active, delta);
+        }
 
         // Save the current record and the delta record.
         this._cumulativeMemoStorage.set(attributes, accumulation, hashCode);

packages/sdk-metrics/test/state/DeltaMetricProcessor.test.ts

@@ -100,6 +100,26 @@ describe('DeltaMetricProcessor', () => {
         assert.strictEqual(accumulation?.toPointValue(), 11);
       }
     });
+
+    it('should merge with active delta of accumulations', () => {
+      const metricProcessor = new DeltaMetricProcessor(new SumAggregator(true));
+
+      {
+        const measurements = new AttributeHashMap<number>();
+        measurements.set({}, 10);
+        metricProcessor.batchCumulate(measurements, [0, 0]);
+      }
+
+      {
+        const measurements = new AttributeHashMap<number>();
+        measurements.set({}, 20);
+        metricProcessor.batchCumulate(measurements, [1, 1]);
+      }
+
+      const accumulations = metricProcessor.collect();
+      const accumulation = accumulations.get({});
+      assert.strictEqual(accumulation?.toPointValue(), 20);
+    });
   });
 
   describe('collect', () => {