Merge branch 'main' into feat/utils/profiler-perf-buffer

# Conflicts:
#	packages/utils/mocks/sink.mock.ts
#	packages/utils/src/lib/performance-observer.int.test.ts
#	packages/utils/src/lib/performance-observer.ts
#	packages/utils/src/lib/performance-observer.unit.test.ts
#	packages/utils/src/lib/profiler/constants.ts
#	packages/utils/src/lib/profiler/profiler.ts
#	packages/utils/src/lib/profiler/profiler.unit.test.ts
#	packages/utils/src/lib/sink-source.type.ts

Author: BioPhoton

packages/utils/eslint.config.js

@@ -12,6 +12,12 @@ export default tseslint.config(
       },
     },
   },
+  {
+    files: ['packages/utils/src/lib/**/wal*.ts'],
+    rules: {
+      'n/no-sync': 'off',
+    },
+  },
   {
     files: ['**/*.json'],
     rules: {

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

@@ -3,7 +3,7 @@ import { defaultClock, epochClock } from './clock-epoch.js';
 describe('epochClock', () => {
   it('should create epoch clock with defaults', () => {
     const c = epochClock();
-    expect(c.timeOriginMs).toBe(500_000);
+    expect(c.timeOriginMs).toBe(1_700_000_000_000);
     expect(c.tid).toBe(2);
     expect(c.pid).toBe(10_001);
     expect(c.fromEpochMs).toBeFunction();
@@ -32,8 +32,8 @@ describe('epochClock', () => {
 
   it('should support performance clock by default for epochNowUs', () => {
     const c = epochClock();
-    expect(c.timeOriginMs).toBe(500_000);
-    expect(c.epochNowUs()).toBe(500_000_000);
+    expect(c.timeOriginMs).toBe(1_700_000_000_000);
+    expect(c.epochNowUs()).toBe(1_700_000_000_000_000);
   });
 
   it.each([
@@ -55,8 +55,8 @@ describe('epochClock', () => {
   });
 
   it.each([
-    [0, 500_000_000],
-    [1000, 501_000_000],
+    [0, 1_700_000_000_000_000],
+    [1000, 1_700_000_001_000_000],
   ])(
     'should convert performance milliseconds to microseconds',
     (perfMs, expected) => {

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

@@ -1,21 +1,21 @@
 import { type PerformanceEntry, performance } from 'node:perf_hooks';
 import type { MockedFunction } from 'vitest';
-import { MockSink } from '../../mocks/sink.mock';
+import { MockFileSink } from '../../mocks/sink.mock';
 import {
   type PerformanceObserverOptions,
   PerformanceObserverSink,
 } from './performance-observer.js';
 
 describe('PerformanceObserverSink', () => {
   let encode: MockedFunction<(entry: PerformanceEntry) => string[]>;
-  let sink: MockSink;
+  let sink: MockFileSink;
   let options: PerformanceObserverOptions<string>;
 
   const awaitObserverCallback = () =>
     new Promise(resolve => setTimeout(resolve, 10));
 
   beforeEach(() => {
-    sink = new MockSink();
+    sink = new MockFileSink();
     sink.open();
     encode = vi.fn((entry: PerformanceEntry) => [
       `${entry.name}:${entry.entryType}`,

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

@@ -6,6 +6,7 @@ import {
 import { isEnvVarEnabled } from './env.js';
 import { PROFILER_DEBUG_ENV_VAR } from './profiler/constants.js';
 import type { Buffered, Observer, Sink } from './sink-source.type';
+import type { AppendableSink } from './wal.js';
 
 /**
  * Encoder that converts PerformanceEntry to domain events.
@@ -163,7 +164,7 @@ export type PerformanceObserverOptions<T> = {
  * @implements {Observer} - Lifecycle management interface
  * @implements {Buffered} - Queue statistics interface
  */
-export class PerformanceObserverSink<T> implements Observer, Buffered {
+export class PerformanceObserverSink<T> {
   /** Encoder function for transforming PerformanceEntry objects into domain types */
   #encodePerfEntry: PerformanceEntryEncoder<T>;
 
@@ -352,7 +353,7 @@ export class PerformanceObserverSink<T> implements Observer, Buffered {
 
     this.#queue.forEach(item => {
       try {
-        this.#sink.write(item);
+        this.#sink.append(item);
         this.#written++;
       } catch {
         failedItems.push(item);

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

@@ -1,14 +1,15 @@
 import { type PerformanceEntry, performance } from 'node:perf_hooks';
 import type { MockedFunction } from 'vitest';
 import { MockPerformanceObserver } from '@code-pushup/test-utils';
-import { MockSink } from '../../mocks/sink.mock';
+import { MockFileSink } from '../../mocks/sink.mock';
 import {
   DEFAULT_FLUSH_THRESHOLD,
   DEFAULT_MAX_QUEUE_SIZE,
   type PerformanceObserverOptions,
   PerformanceObserverSink,
   validateFlushThreshold,
 } from './performance-observer.js';
+import type { Codec } from './wal.js';
 
 describe('validateFlushThreshold', () => {
   it.each([
@@ -55,12 +56,12 @@ describe('validateFlushThreshold', () => {
 
 describe('PerformanceObserverSink', () => {
   let encodePerfEntry: MockedFunction<(entry: PerformanceEntry) => string[]>;
-  let sink: MockSink;
+  let sink: MockFileSink;
   let options: PerformanceObserverOptions<string>;
 
   beforeEach(() => {
     vi.clearAllMocks();
-    sink = new MockSink();
+    sink = new MockFileSink();
     sink.open();
     encodePerfEntry = vi.fn((entry: PerformanceEntry) => [
       `${entry.name}:${entry.entryType}`,
@@ -663,6 +664,28 @@ describe('PerformanceObserverSink', () => {
     );
   });
 
+  it('accepts custom sinks with append method', () => {
+    const collectedItems: string[] = [];
+    const customSink = {
+      // eslint-disable-next-line functional/immutable-data
+      append: (item: string) => collectedItems.push(item),
+    };
+
+    const observer = new PerformanceObserverSink({
+      sink: customSink,
+      encode: (entry: PerformanceEntry) => [`${entry.name}:${entry.duration}`],
+    });
+
+    observer.subscribe();
+
+    const mockObserver = MockPerformanceObserver.lastInstance();
+    mockObserver?.emitMark('test-mark');
+
+    observer.flush();
+
+    expect(collectedItems).toContain('test-mark:0');
+  });
+
   it('tracks addedSinceLastFlush counter correctly', () => {
     const observer = new PerformanceObserverSink({
       sink,

packages/utils/src/lib/profiler/constants.ts

@@ -1,2 +1,7 @@
 export const PROFILER_ENABLED_ENV_VAR = 'CP_PROFILING';
 export const PROFILER_DEBUG_ENV_VAR = 'CP_PROFILER_DEBUG';
+export const PROFILER_COORDINATOR_FLAG_ENV_VAR = 'CP_PROFILER_COORDINATOR';
+export const PROFILER_ORIGIN_PID_ENV_VAR = 'CP_PROFILER_ORIGIN_PID';
+export const PROFILER_DIRECTORY_ENV_VAR = 'CP_PROFILER_DIR';
+export const PROFILER_BASE_NAME = 'trace';
+export const PROFILER_DIRECTORY = './tmp/profiles';

packages/utils/src/lib/profiler/profiler.ts

@@ -1,4 +1,6 @@
 import { performance } from 'node:perf_hooks';
+import process from 'node:process';
+import { threadId } from 'node:worker_threads';
 import { isEnvVarEnabled } from '../env.js';
 import {
   type PerformanceObserverOptions,
@@ -26,6 +28,14 @@ import {
   PROFILER_ENABLED_ENV_VAR,
 } from './constants.js';
 
+/**
+ * Generates a unique profiler ID based on performance time origin, process ID, thread ID, and instance count.
+ */
+export function getProfilerId() {
+  // eslint-disable-next-line functional/immutable-data
+  return `${Math.round(performance.timeOrigin)}.${process.pid}.${threadId}.${++Profiler.instanceCount}`;
+}
+
 /**
  * Configuration options for creating a Profiler instance.
  *
@@ -69,6 +79,8 @@ export type ProfilerOptions<T extends ActionTrackConfigs = ActionTrackConfigs> =
  *
  */
 export class Profiler<T extends ActionTrackConfigs> {
+  static instanceCount = 0;
+  readonly id = getProfilerId();
   #enabled: boolean = false;
   readonly #defaults: ActionTrackEntryPayload;
   readonly tracks: Record<keyof T, ActionTrackEntryPayload> | undefined;

packages/utils/src/lib/profiler/profiler.unit.test.ts

@@ -1,4 +1,5 @@
 import { performance } from 'node:perf_hooks';
+import { threadId } from 'node:worker_threads';
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { MockTraceEventFileSink } from '../../../mocks/sink.mock.js';
 import type { PerformanceEntryEncoder } from '../performance-observer.js';
@@ -9,8 +10,19 @@ import {
   type NodejsProfilerOptions,
   Profiler,
   type ProfilerOptions,
+  getProfilerId,
 } from './profiler.js';
 
+describe('getProfilerId', () => {
+  it('should generate a unique id per process', () => {
+    expect(getProfilerId()).toBe(
+      `${Math.round(performance.timeOrigin)}.${process.pid}.${threadId}.1`,
+    );
+    expect(getProfilerId()).toBe(
+      `${Math.round(performance.timeOrigin)}.${process.pid}.${threadId}.2`,
+    );
+  });
+});
 describe('Profiler', () => {
   const getProfiler = (overrides?: Partial<ProfilerOptions>) =>
     new Profiler({