refactor: use sharded WAL

Author: BioPhoton

packages/utils/docs/profiler.md

@@ -259,7 +259,7 @@ const saved = profiler.measure('save-user', () => saveToDb(user), {
 
 This profiler extends all options and API from Profiler with automatic process exit handling for buffered performance data.
 
-The NodeJSProfiler automatically subscribes to performance observation and installs exit handlers that flush buffered data on process termination (signals, fatal errors, or normal exit).
+The NodeJSProfiler automatically subscribes to performance observation and installs exit handlers that flush buffered data on process termination (signals, fatal errors, or normal exit). It uses a `ShardedWal` internally to coordinate multiple WAL shards across processes/files.
 
 ## Configuration
 
@@ -273,12 +273,16 @@ new NodejsProfiler<DomainEvents, Tracks>(options: NodejsProfilerOptions<DomainEv
 
 **Options:**
 
-| Property                 | Type                                    | Default    | Description                                                                     |
-| ------------------------ | --------------------------------------- | ---------- | ------------------------------------------------------------------------------- |
-| `encodePerfEntry`        | `PerformanceEntryEncoder<DomainEvents>` | _required_ | Function that encodes raw PerformanceEntry objects into domain-specific types   |
-| `captureBufferedEntries` | `boolean`                               | `true`     | Whether to capture performance entries that occurred before observation started |
-| `flushThreshold`         | `number`                                | `20`       | Threshold for triggering queue flushes based on queue length                    |
-| `maxQueueSize`           | `number`                                | `10_000`   | Maximum number of items allowed in the queue before new entries are dropped     |
+| Property                 | Type                                    | Default          | Description                                                                          |
+| ------------------------ | --------------------------------------- | ---------------- | ------------------------------------------------------------------------------------ |
+| `format`                 | `Partial<WalFormat<DomainEvents>>`      | _required_       | WAL format configuration for sharded write-ahead logging                             |
+| `measureName`            | `string`                                | _auto-generated_ | Optional folder name for sharding. If not provided, a new group ID will be generated |
+| `outDir`                 | `string`                                | `'tmp/profiles'` | Output directory for WAL shards and final files                                      |
+| `outBaseName`            | `string`                                | _optional_       | Override the base name for WAL files (overrides format.baseName)                     |
+| `encodePerfEntry`        | `PerformanceEntryEncoder<DomainEvents>` | _required_       | Function that encodes raw PerformanceEntry objects into domain-specific types        |
+| `captureBufferedEntries` | `boolean`                               | `true`           | Whether to capture performance entries that occurred before observation started      |
+| `flushThreshold`         | `number`                                | `20`             | Threshold for triggering queue flushes based on queue length                         |
+| `maxQueueSize`           | `number`                                | `10_000`         | Maximum number of items allowed in the queue before new entries are dropped          |
 
 ## API Methods
 

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

@@ -19,3 +19,21 @@ export const PROFILER_ENABLED_ENV_VAR = 'CP_PROFILING';
  * ```
  */
 export const PROFILER_DEBUG_ENV_VAR = 'CP_PROFILER_DEBUG';
+
+/**
+ * Default output directory for persisted profiler data.
+ * Matches the default persist output directory from models.
+ */
+export const PERSIST_OUT_DIR = '.code-pushup';
+
+/**
+ * Default filename (without extension) for persisted profiler data.
+ * Matches the default persist filename from models.
+ */
+export const PERSIST_OUT_FILENAME = 'report';
+
+/**
+ * Default base name for WAL files.
+ * Used as the base name for sharded WAL files (e.g., "trace").
+ */
+export const PERSIST_OUT_BASENAME = 'trace';

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

@@ -1,7 +1,6 @@
-import { MockTraceEventFileSink } from '../../../mocks/sink.mock.js';
 import type { PerformanceEntryEncoder } from '../performance-observer.js';
 import type { ActionTrackEntryPayload } from '../user-timing-extensibility-api.type.js';
-import { NodejsProfiler, Profiler } from './profiler.js';
+import { NodeJsProfiler, Profiler } from './profiler.js';
 
 describe('Profiler Integration', () => {
   let profiler: Profiler<Record<string, ActionTrackEntryPayload>>;
@@ -306,25 +305,31 @@ describe('NodeJS Profiler Integration', () => {
     return [];
   };
 
-  let mockSink: MockTraceEventFileSink;
-  let nodejsProfiler: NodejsProfiler<string>;
+  let nodejsProfiler: NodeJsProfiler<string>;
 
   beforeEach(() => {
-    mockSink = new MockTraceEventFileSink();
-
-    nodejsProfiler = new NodejsProfiler({
+    nodejsProfiler = new NodeJsProfiler({
       prefix: 'test',
       track: 'test-track',
-      sink: mockSink,
+      format: {
+        baseName: 'test',
+        walExtension: '.log',
+        finalExtension: '.log',
+        codec: {
+          encode: (v: string) => v,
+          decode: (v: string) => v,
+        },
+        finalizer: (records: (string | { __invalid: true; raw: string })[]) =>
+          records.filter((r): r is string => typeof r === 'string').join('\n'),
+      },
       encodePerfEntry: simpleEncoder,
       enabled: true,
     });
   });
 
   it('should initialize with sink opened when enabled', () => {
-    expect(mockSink.isClosed()).toBeFalse();
     expect(nodejsProfiler.isEnabled()).toBeTrue();
-    expect(mockSink.open).toHaveBeenCalledOnce();
+    expect(nodejsProfiler.stats.walOpen).toBeTrue();
   });
 
   it('should create performance entries and write to sink', () => {
@@ -340,41 +345,52 @@ describe('NodeJS Profiler Integration', () => {
         return 'async-result';
       }),
     ).resolves.toBe('async-result');
+
+    const stats = nodejsProfiler.stats;
+    await expect(JSON.stringify(stats, null, 2)).toMatchFileSnapshot(
+      '__snapshots__/profiler.int.test.async-operations.json',
+    );
   });
 
   it('should disable profiling and close sink', () => {
     nodejsProfiler.setEnabled(false);
     expect(nodejsProfiler.isEnabled()).toBeFalse();
-    expect(mockSink.isClosed()).toBeTrue();
-    expect(mockSink.close).toHaveBeenCalledOnce();
+    expect(nodejsProfiler.stats.walOpen).toBeFalse();
 
     expect(nodejsProfiler.measure('disabled-test', () => 'success')).toBe(
       'success',
     );
-
-    expect(mockSink.getWrittenItems()).toHaveLength(0);
   });
 
   it('should re-enable profiling correctly', () => {
     nodejsProfiler.setEnabled(false);
     nodejsProfiler.setEnabled(true);
 
     expect(nodejsProfiler.isEnabled()).toBeTrue();
-    expect(mockSink.isClosed()).toBeFalse();
-    expect(mockSink.open).toHaveBeenCalledTimes(2);
+    expect(nodejsProfiler.stats.walOpen).toBeTrue();
 
     expect(nodejsProfiler.measure('re-enabled-test', () => 42)).toBe(42);
   });
 
   it('should support custom tracks', () => {
-    const profilerWithTracks = new NodejsProfiler({
+    const profilerWithTracks = new NodeJsProfiler({
       prefix: 'api-server',
       track: 'HTTP',
       tracks: {
         db: { track: 'Database', color: 'secondary' },
         cache: { track: 'Cache', color: 'primary' },
       },
-      sink: mockSink,
+      format: {
+        baseName: 'test',
+        walExtension: '.log',
+        finalExtension: '.log',
+        codec: {
+          encode: (v: string) => v,
+          decode: (v: string) => v,
+        },
+        finalizer: (records: (string | { __invalid: true; raw: string })[]) =>
+          records.filter((r): r is string => typeof r === 'string').join('\n'),
+      },
       encodePerfEntry: simpleEncoder,
     });
 
@@ -386,10 +402,20 @@ describe('NodeJS Profiler Integration', () => {
   });
 
   it('should capture buffered entries when buffered option is enabled', () => {
-    const bufferedProfiler = new NodejsProfiler({
+    const bufferedProfiler = new NodeJsProfiler({
       prefix: 'buffered-test',
       track: 'Test',
-      sink: mockSink,
+      format: {
+        baseName: 'test',
+        walExtension: '.log',
+        finalExtension: '.log',
+        codec: {
+          encode: (v: string) => v,
+          decode: (v: string) => v,
+        },
+        finalizer: (records: (string | { __invalid: true; raw: string })[]) =>
+          records.filter((r): r is string => typeof r === 'string').join('\n'),
+      },
       encodePerfEntry: simpleEncoder,
       captureBufferedEntries: true,
       enabled: true,
@@ -407,10 +433,20 @@ describe('NodeJS Profiler Integration', () => {
   });
 
   it('should return correct getStats with dropped and written counts', () => {
-    const statsProfiler = new NodejsProfiler({
+    const statsProfiler = new NodeJsProfiler({
       prefix: 'stats-test',
       track: 'Stats',
-      sink: mockSink,
+      format: {
+        baseName: 'test',
+        walExtension: '.log',
+        finalExtension: '.log',
+        codec: {
+          encode: (v: string) => v,
+          decode: (v: string) => v,
+        },
+        finalizer: (records: (string | { __invalid: true; raw: string })[]) =>
+          records.filter((r): r is string => typeof r === 'string').join('\n'),
+      },
       encodePerfEntry: simpleEncoder,
       maxQueueSize: 2,
       flushThreshold: 2,
@@ -431,10 +467,20 @@ describe('NodeJS Profiler Integration', () => {
   });
 
   it('should provide comprehensive queue statistics via getStats', () => {
-    const profiler = new NodejsProfiler({
+    const profiler = new NodeJsProfiler({
       prefix: 'stats-profiler',
       track: 'Stats',
-      sink: mockSink,
+      format: {
+        baseName: 'test',
+        walExtension: '.log',
+        finalExtension: '.log',
+        codec: {
+          encode: (v: string) => v,
+          decode: (v: string) => v,
+        },
+        finalizer: (records: (string | { __invalid: true; raw: string })[]) =>
+          records.filter((r): r is string => typeof r === 'string').join('\n'),
+      },
       encodePerfEntry: simpleEncoder,
       maxQueueSize: 3,
       flushThreshold: 2,
@@ -462,4 +508,13 @@ describe('NodeJS Profiler Integration', () => {
     expect(finalStats.isSubscribed).toBeFalse();
     expect(finalStats.queued).toBe(0);
   });
+
+  it('should handle async operations', async () => {
+    await expect(
+      nodejsProfiler.measureAsync('async-test', async () => {
+        await new Promise(resolve => setTimeout(resolve, 1));
+        return 'async-result';
+      }),
+    ).resolves.toBe('async-result');
+  });
 });