refactor: add better flush handling

Author: John Doe

packages/utils/src/lib/performance-observer.int.test.ts

@@ -179,4 +179,49 @@ describe('PerformanceObserverSink', () => {
 
     expect(sink.getWrittenItems()).toHaveLength(2);
   });
+
+  it('cursor logic prevents duplicate processing of performance entries', () => {
+    const observer = new PerformanceObserverSink(options);
+    observer.subscribe();
+
+    performance.mark('first-mark');
+    performance.mark('second-mark');
+    expect(encode).not.toHaveBeenCalled();
+    observer.flush();
+    expect(sink.getWrittenItems()).toStrictEqual([
+      'first-mark:mark',
+      'second-mark:mark',
+    ]);
+
+    expect(encode).toHaveBeenCalledTimes(2);
+    expect(encode).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({ name: 'first-mark' }),
+    );
+    expect(encode).toHaveBeenNthCalledWith(
+      2,
+      expect.objectContaining({ name: 'second-mark' }),
+    );
+
+    performance.mark('third-mark');
+    performance.measure('first-measure');
+
+    observer.flush();
+    expect(sink.getWrittenItems()).toStrictEqual([
+      'first-mark:mark',
+      'second-mark:mark',
+      'third-mark:mark',
+      'first-measure:measure',
+    ]);
+
+    expect(encode).toHaveBeenCalledTimes(4);
+    expect(encode).toHaveBeenNthCalledWith(
+      3,
+      expect.objectContaining({ name: 'third-mark' }),
+    );
+    expect(encode).toHaveBeenNthCalledWith(
+      4,
+      expect.objectContaining({ name: 'first-measure' }),
+    );
+  });
 });

packages/utils/src/lib/performance-observer.ts

@@ -1,12 +1,13 @@
 import {
-  type EntryType,
   type PerformanceEntry,
   PerformanceObserver,
   type PerformanceObserverEntryList,
   performance,
 } from 'node:perf_hooks';
 import type { Buffered, Encoder, Observer, Sink } from './sink-source.types.js';
 
+const OBSERVED_TYPES = ['mark', 'measure'] as const;
+type ObservedEntryType = 'mark' | 'measure';
 export const DEFAULT_FLUSH_THRESHOLD = 20;
 
 export type PerformanceObserverOptions<T> = {
@@ -24,16 +25,21 @@ export class PerformanceObserverSink<T>
   #flushThreshold: number;
   #sink: Sink<T, unknown>;
   #observer: PerformanceObserver | undefined;
-  #observedTypes: EntryType[] = ['mark', 'measure'];
-  #getEntries = (list: PerformanceObserverEntryList) =>
-    this.#observedTypes.flatMap(t => list.getEntriesByType(t));
-  #observedCount: number = 0;
+
+  #pendingCount = 0;
+
+  // "cursor" per type: how many we already wrote from the global buffer
+  #written: Map<ObservedEntryType, number>;
 
   constructor(options: PerformanceObserverOptions<T>) {
-    this.#encode = options.encode;
-    this.#sink = options.sink;
-    this.#buffered = options.buffered ?? false;
-    this.#flushThreshold = options.flushThreshold ?? DEFAULT_FLUSH_THRESHOLD;
+    const { encode, sink, buffered, flushThreshold } = options;
+    this.#encode = encode;
+    this.#written = new Map<ObservedEntryType, number>(
+      OBSERVED_TYPES.map(t => [t, 0]),
+    );
+    this.#sink = sink;
+    this.#buffered = buffered ?? false;
+    this.#flushThreshold = flushThreshold ?? DEFAULT_FLUSH_THRESHOLD;
   }
 
   encode(entry: PerformanceEntry): T[] {
@@ -45,37 +51,51 @@ export class PerformanceObserverSink<T>
       return;
     }
 
-    this.#observer = new PerformanceObserver(list => {
-      const entries = this.#getEntries(list);
-      this.#observedCount += entries.length;
-      if (this.#observedCount >= this.#flushThreshold) {
-        this.flush(entries);
-      }
-    });
+    // The only used to trigger the flush it is not processing the entries just counting them
+    this.#observer = new PerformanceObserver(
+      (list: PerformanceObserverEntryList) => {
+        const batchCount = OBSERVED_TYPES.reduce(
+          (n, t) => n + list.getEntriesByType(t).length,
+          0,
+        );
+
+        this.#pendingCount += batchCount;
+        if (this.#pendingCount >= this.#flushThreshold) {
+          this.flush();
+        }
+      },
+    );
 
     this.#observer.observe({
-      entryTypes: this.#observedTypes,
+      entryTypes: OBSERVED_TYPES,
       buffered: this.#buffered,
     });
   }
 
-  flush(entriesToProcess?: PerformanceEntry[]): void {
+  flush(): void {
     if (!this.#observer) {
       return;
     }
 
-    const entries = entriesToProcess || this.#getEntries(performance);
-    entries.forEach(entry => {
-      const encoded = this.encode(entry);
-      encoded.forEach(item => {
-        this.#sink.write(item);
-      });
-    });
+    OBSERVED_TYPES.forEach(t => {
+      const written = this.#written.get(t) ?? 0;
+      const fresh = performance.getEntriesByType(t).slice(written);
+
+      try {
+        fresh
+          .flatMap(entry => this.encode(entry))
+          .forEach(item => this.#sink.write(item));
 
-    // In real PerformanceObserver, entries remain in the global buffer
-    // They are only cleared when explicitly requested via performance.clearMarks/clearMeasures
+        this.#written.set(t, written + fresh.length);
+      } catch (error) {
+        throw new Error(
+          'PerformanceObserverSink failed to write items to sink.',
+          { cause: error },
+        );
+      }
+    });
 
-    this.#observedCount = 0;
+    this.#pendingCount = 0;
   }
 
   unsubscribe(): void {

packages/utils/src/lib/performance-observer.unit.test.ts

@@ -41,6 +41,40 @@ describe('PerformanceObserverSink', () => {
     expect(MockPerformanceObserver.instances).toHaveLength(0);
   });
 
+  it('creates instance with default flushThreshold when not provided', () => {
+    expect(
+      () =>
+        new PerformanceObserverSink({
+          sink,
+          encode,
+        }),
+    ).not.toThrow();
+    expect(MockPerformanceObserver.instances).toHaveLength(0);
+    // Instance creation covers the default flushThreshold assignment
+  });
+
+  it('automatically flushes when pendingCount reaches flushThreshold', () => {
+    const observer = new PerformanceObserverSink({
+      sink,
+      encode,
+      flushThreshold: 2, // Set threshold to 2
+    });
+    observer.subscribe();
+
+    const mockObserver = MockPerformanceObserver.lastInstance();
+
+    // Emit 1 entry - should not trigger flush yet (pendingCount = 1 < 2)
+    mockObserver?.emitMark('first-mark');
+    expect(sink.getWrittenItems()).toStrictEqual([]);
+
+    // Emit 1 more entry - should trigger flush (pendingCount = 2 >= 2)
+    mockObserver?.emitMark('second-mark');
+    expect(sink.getWrittenItems()).toStrictEqual([
+      'first-mark:mark',
+      'second-mark:mark',
+    ]);
+  });
+
   it('creates instance with all options without starting to observe', () => {
     expect(
       () =>
@@ -219,4 +253,56 @@ describe('PerformanceObserverSink', () => {
     expect(perfObserver?.disconnect).toHaveBeenCalledTimes(1);
     expect(MockPerformanceObserver.instances).toHaveLength(0);
   });
+
+  it('flush wraps sink write errors with descriptive error message', () => {
+    const failingSink = {
+      write: vi.fn(() => {
+        throw new Error('Sink write failed');
+      }),
+    };
+
+    const observer = new PerformanceObserverSink({
+      sink: failingSink as any,
+      encode,
+      flushThreshold: 1,
+    });
+
+    observer.subscribe();
+
+    performance.mark('test-mark');
+
+    expect(() => observer.flush()).toThrow(
+      expect.objectContaining({
+        message: 'PerformanceObserverSink failed to write items to sink.',
+        cause: expect.objectContaining({
+          message: 'Sink write failed',
+        }),
+      }),
+    );
+  });
+
+  it('flush wraps encode errors with descriptive error message', () => {
+    const failingEncode = vi.fn(() => {
+      throw new Error('Encode failed');
+    });
+
+    const observer = new PerformanceObserverSink({
+      sink,
+      encode: failingEncode,
+      flushThreshold: 1,
+    });
+
+    observer.subscribe();
+
+    performance.mark('test-mark');
+
+    expect(() => observer.flush()).toThrow(
+      expect.objectContaining({
+        message: 'PerformanceObserverSink failed to write items to sink.',
+        cause: expect.objectContaining({
+          message: 'Encode failed',
+        }),
+      }),
+    );
+  });
 });