feat(utils): add performance observer (#1206)

Author: BioPhoton

packages/utils/mocks/sink.mock.ts

@@ -0,0 +1,30 @@
+import type { Sink } from '../src/lib/sink-source.types';
+
+export class MockSink implements Sink<string, string> {
+  private writtenItems: string[] = [];
+  private closed = false;
+
+  open(): void {
+    this.closed = false;
+  }
+
+  write(input: string): void {
+    this.writtenItems.push(input);
+  }
+
+  close(): void {
+    this.closed = true;
+  }
+
+  isClosed(): boolean {
+    return this.closed;
+  }
+
+  encode(input: string): string {
+    return `${input}-${this.constructor.name}-encoded`;
+  }
+
+  getWrittenItems(): string[] {
+    return [...this.writtenItems];
+  }
+}

packages/utils/src/lib/clock-epoch.int.test.ts

@@ -20,16 +20,6 @@ describe('epochClock', () => {
     expect(c.fromDateNowMs).toBeFunction();
   });
 
-  it('should support performance clock by default for epochNowUs', () => {
-    const c = epochClock();
-    expect(c.timeOriginMs).toBe(performance.timeOrigin);
-    const nowUs = c.epochNowUs();
-    expect(nowUs).toBe(Math.round(nowUs));
-    const expectedUs = Date.now() * 1000;
-
-    expect(nowUs).toBeWithin(expectedUs - 2000, expectedUs + 1000);
-  });
-
   it('should convert epoch milliseconds to microseconds correctly', () => {
     const c = epochClock();
     const epochMs = Date.now();

packages/utils/src/lib/clock-epoch.unit.test.ts

@@ -33,7 +33,7 @@ describe('epochClock', () => {
   it('should support performance clock by default for epochNowUs', () => {
     const c = epochClock();
     expect(c.timeOriginMs).toBe(500_000);
-    expect(c.epochNowUs()).toBe(1_000_000_000); // timeOrigin + (Date.now() - timeOrigin) = Date.now()
+    expect(c.epochNowUs()).toBe(500_000_000); // timeOrigin + performance.now() = timeOrigin + 0
   });
 
   it.each([

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

@@ -0,0 +1,182 @@
+import { type PerformanceEntry, performance } from 'node:perf_hooks';
+import {
+  type MockedFunction,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  vi,
+} from 'vitest';
+import { MockSink } from '../../mocks/sink.mock';
+import {
+  type PerformanceObserverOptions,
+  PerformanceObserverSink,
+} from './performance-observer.js';
+
+describe('PerformanceObserverSink', () => {
+  let encode: MockedFunction<(entry: PerformanceEntry) => string[]>;
+  let sink: MockSink;
+  let options: PerformanceObserverOptions<string>;
+
+  const awaitObserverCallback = () =>
+    new Promise(resolve => setTimeout(resolve, 10));
+
+  beforeEach(() => {
+    sink = new MockSink();
+    encode = vi.fn((entry: PerformanceEntry) => [
+      `${entry.name}:${entry.entryType}`,
+    ]);
+
+    options = {
+      sink,
+      encode,
+    };
+
+    performance.clearMarks();
+    performance.clearMeasures();
+  });
+
+  it('creates instance with required options', () => {
+    expect(() => new PerformanceObserverSink(options)).not.toThrow();
+  });
+
+  it('internal PerformanceObserver should process observed entries', () => {
+    const observer = new PerformanceObserverSink(options);
+    observer.subscribe();
+
+    performance.mark('test-mark');
+    performance.measure('test-measure');
+    observer.flush();
+    expect(encode).toHaveBeenCalledTimes(2);
+    expect(encode).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({
+        name: 'test-mark',
+        entryType: 'mark',
+      }),
+    );
+    expect(encode).toHaveBeenNthCalledWith(
+      2,
+      expect.objectContaining({
+        name: 'test-measure',
+        entryType: 'measure',
+      }),
+    );
+  });
+
+  it('internal PerformanceObserver calls flush if flushThreshold exceeded', async () => {
+    const observer = new PerformanceObserverSink({
+      ...options,
+      flushThreshold: 3,
+    });
+    observer.subscribe();
+
+    performance.mark('test-mark1');
+    performance.mark('test-mark2');
+    performance.mark('test-mark3');
+
+    await awaitObserverCallback();
+
+    expect(encode).toHaveBeenCalledTimes(3);
+  });
+
+  it('flush flushes observed entries when subscribed', () => {
+    const observer = new PerformanceObserverSink(options);
+    observer.subscribe();
+
+    performance.mark('test-mark1');
+    performance.mark('test-mark2');
+    expect(sink.getWrittenItems()).toStrictEqual([]);
+
+    observer.flush();
+    expect(sink.getWrittenItems()).toStrictEqual([
+      'test-mark1:mark',
+      'test-mark2:mark',
+    ]);
+  });
+
+  it('flush calls encode for each entry', () => {
+    const observer = new PerformanceObserverSink(options);
+    observer.subscribe();
+
+    performance.mark('test-mark1');
+    performance.mark('test-mark2');
+
+    observer.flush();
+
+    expect(encode).toHaveBeenCalledWith(
+      expect.objectContaining({
+        name: 'test-mark1',
+        entryType: 'mark',
+      }),
+    );
+    expect(encode).toHaveBeenCalledWith(
+      expect.objectContaining({
+        name: 'test-mark2',
+        entryType: 'mark',
+      }),
+    );
+  });
+
+  it('unsubscribe stops observing performance entries', async () => {
+    const observer = new PerformanceObserverSink({
+      ...options,
+      flushThreshold: 1,
+    });
+
+    observer.subscribe();
+    performance.mark('subscribed-mark1');
+    performance.mark('subscribed-mark2');
+    await awaitObserverCallback();
+    expect(encode).toHaveBeenCalledTimes(2);
+
+    observer.unsubscribe();
+    performance.mark('unsubscribed-mark1');
+    performance.mark('unsubscribed-mark2');
+    await awaitObserverCallback();
+    expect(encode).toHaveBeenCalledTimes(2);
+  });
+
+  it('should observe performance entries and write them to the sink on flush', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    observer.subscribe();
+    performance.mark('test-mark');
+    observer.flush();
+    expect(sink.getWrittenItems()).toHaveLength(1);
+  });
+
+  it('should observe buffered performance entries when buffered is enabled', async () => {
+    const observer = new PerformanceObserverSink({
+      ...options,
+      buffered: true,
+    });
+
+    performance.mark('test-mark-1');
+    performance.mark('test-mark-2');
+    await new Promise(resolve => setTimeout(resolve, 10));
+    observer.subscribe();
+    await new Promise(resolve => setTimeout(resolve, 10));
+    expect(performance.getEntries()).toHaveLength(2);
+    observer.flush();
+    expect(sink.getWrittenItems()).toHaveLength(2);
+  });
+
+  it('handles multiple encoded items per performance entry', () => {
+    const multiEncodeFn = vi.fn(e => [
+      `${e.entryType}-item1`,
+      `${e.entryType}item2`,
+    ]);
+    const observer = new PerformanceObserverSink({
+      ...options,
+      encode: multiEncodeFn,
+    });
+
+    observer.subscribe();
+
+    performance.mark('test-mark');
+    observer.flush();
+
+    expect(sink.getWrittenItems()).toHaveLength(2);
+  });
+});

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

@@ -0,0 +1,92 @@
+import {
+  type EntryType,
+  type PerformanceEntry,
+  PerformanceObserver,
+  type PerformanceObserverEntryList,
+  performance,
+} from 'node:perf_hooks';
+import type { Buffered, Encoder, Observer, Sink } from './sink-source.types.js';
+
+export const DEFAULT_FLUSH_THRESHOLD = 20;
+
+export type PerformanceObserverOptions<T> = {
+  sink: Sink<T, unknown>;
+  encode: (entry: PerformanceEntry) => T[];
+  buffered?: boolean;
+  flushThreshold?: number;
+};
+
+export class PerformanceObserverSink<T>
+  implements Observer, Buffered, Encoder<PerformanceEntry, T[]>
+{
+  #encode: (entry: PerformanceEntry) => T[];
+  #buffered: boolean;
+  #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;
+
+  constructor(options: PerformanceObserverOptions<T>) {
+    this.#encode = options.encode;
+    this.#sink = options.sink;
+    this.#buffered = options.buffered ?? false;
+    this.#flushThreshold = options.flushThreshold ?? DEFAULT_FLUSH_THRESHOLD;
+  }
+
+  encode(entry: PerformanceEntry): T[] {
+    return this.#encode(entry);
+  }
+
+  subscribe(): void {
+    if (this.#observer) {
+      return;
+    }
+
+    this.#observer = new PerformanceObserver(list => {
+      const entries = this.#getEntries(list);
+      this.#observedCount += entries.length;
+      if (this.#observedCount >= this.#flushThreshold) {
+        this.flush(entries);
+      }
+    });
+
+    this.#observer.observe({
+      entryTypes: this.#observedTypes,
+      buffered: this.#buffered,
+    });
+  }
+
+  flush(entriesToProcess?: PerformanceEntry[]): 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);
+      });
+    });
+
+    // In real PerformanceObserver, entries remain in the global buffer
+    // They are only cleared when explicitly requested via performance.clearMarks/clearMeasures
+
+    this.#observedCount = 0;
+  }
+
+  unsubscribe(): void {
+    if (!this.#observer) {
+      return;
+    }
+    this.#observer?.disconnect();
+    this.#observer = undefined;
+  }
+
+  isSubscribed(): boolean {
+    return this.#observer !== undefined;
+  }
+}