refactor: refactor sharded wal to handle coordinator

packages/utils/src/lib/create-runner-files.ts

@@ -1,8 +1,8 @@
 import { writeFile } from 'node:fs/promises';
 import path from 'node:path';
-import { threadId } from 'node:worker_threads';
 import type { RunnerFilesPaths } from '@code-pushup/models';
 import { ensureDirectoryExists, pluginWorkDir } from './file-system.js';
+import { getUniqueProcessThreadId } from './process-id.js';
 
 /**
  * Function to create timestamp nested plugin runner files for config and output.
@@ -14,9 +14,7 @@ export async function createRunnerFiles(
   pluginSlug: string,
   configJSON: string,
 ): Promise<RunnerFilesPaths> {
-  // Use timestamp + process ID + threadId
-  // This prevents race conditions when running the same plugin for multiple projects in parallel
-  const uniqueId = `${(performance.timeOrigin + performance.now()) * 10}-${process.pid}-${threadId}`;
+  const uniqueId = getUniqueProcessThreadId();
   const runnerWorkDir = path.join(pluginWorkDir(pluginSlug), uniqueId);
   const runnerConfigPath = path.join(runnerWorkDir, 'plugin-config.json');
   const runnerOutputPath = path.join(runnerWorkDir, 'runner-output.json');

packages/utils/src/lib/errors.ts

@@ -30,3 +30,21 @@ export function stringifyError(
   }
   return JSON.stringify(error);
 }
+
+/**
+ * Extends an error with a new message and keeps the original as the cause.
+ * This helps to keep the stacktrace intact and enables better debugging.
+ * @param error - The error to extend
+ * @param message - The new message to add to the error
+ * @returns A new error with the extended message and the original as cause
+ */
+export function extendError(
+  error: unknown,
+  message: string,
+  { appendMessage = false } = {},
+) {
+  const errorMessage = appendMessage
+    ? `${message}\n${stringifyError(error)}`
+    : message;
+  return new Error(errorMessage, { cause: error });
+}

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

@@ -31,3 +31,9 @@ export const SHARDED_WAL_COORDINATOR_ID_ENV_VAR =
  * Used as the base name for sharded WAL files (e.g., "trace" in "trace.json").
  */
 export const PROFILER_PERSIST_BASENAME = 'trace';
+
+/**
+ * Name for current measure.
+ * Used as the name for the sharded folder.
+ */
+export const PROFILER_MEASURE_NAME_ENV_VAR = 'CP_PROFILER_MEASURE_NAME';

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

@@ -16,10 +16,11 @@ import type {
   ActionTrackEntryPayload,
   MarkerPayload,
 } from '../user-timing-extensibility-api.type.js';
+import { getShardedPath } from '../wal-sharded.js';
 import {
   type AppendableSink,
+  type WalRecord,
   WriteAheadLogFile,
-  getShardedPath,
 } from '../wal.js';
 import {
   PROFILER_DEBUG_ENV_VAR,
@@ -73,7 +74,7 @@ export type NodejsProfilerOptions<
  * @template Tracks - Record type defining available track names and their configurations
  */
 export class NodejsProfiler<
-  DomainEvents extends string | object,
+  DomainEvents extends WalRecord,
   Tracks extends Record<string, ActionTrackEntryPayload> = Record<
     string,
     ActionTrackEntryPayload
@@ -118,6 +119,7 @@ export class NodejsProfiler<
         filename ??
         path.join(
           process.cwd(),
+          // @TODO remove in PR https://github.com/code-pushup/cli/pull/1231 in favour of class method getShardedFileName
           getShardedPath({
             dir: 'tmp/profiles',
             groupId: getUniqueTimeId(),

packages/utils/src/lib/wal-sharded.int.test.ts

@@ -0,0 +1,258 @@
+import fs from 'node:fs';
+import path from 'node:path';
+import { SHARDED_WAL_COORDINATOR_ID_ENV_VAR } from './profiler/constants.js';
+import { ShardedWal } from './wal-sharded.js';
+import { type WalFormat, type WalRecord, stringCodec } from './wal.js';
+
+describe('ShardedWal Integration', () => {
+  const testDir = path.join(
+    process.cwd(),
+    'tmp',
+    'int',
+    'utils',
+    'wal-sharded',
+  );
+  const makeMockFormat = <T extends WalRecord>(
+    overrides: Partial<WalFormat<T>>,
+  ): WalFormat<T> => {
+    const {
+      baseName = 'wal',
+      walExtension = '.log',
+      finalExtension = '.json',
+      codec = stringCodec<T>(),
+      finalizer = records => `${JSON.stringify(records)}\n`,
+    } = overrides;
+
+    return {
+      baseName,
+      walExtension,
+      finalExtension,
+      codec,
+      finalizer,
+    };
+  };
+  let shardedWal: ShardedWal;
+
+  beforeEach(() => {
+    if (fs.existsSync(testDir)) {
+      fs.rmSync(testDir, { recursive: true, force: true });
+    }
+    fs.mkdirSync(testDir, { recursive: true });
+  });
+
+  afterEach(() => {
+    if (shardedWal) {
+      shardedWal.cleanupIfCoordinator();
+    }
+    if (fs.existsSync(testDir)) {
+      fs.rmSync(testDir, { recursive: true, force: true });
+    }
+  });
+
+  it('should create and finalize shards correctly', () => {
+    shardedWal = new ShardedWal({
+      debug: false,
+      dir: testDir,
+      format: makeMockFormat({
+        baseName: 'trace',
+      }),
+      coordinatorIdEnvVar: SHARDED_WAL_COORDINATOR_ID_ENV_VAR,
+      groupId: 'create-finalize',
+    });
+
+    const shard1 = shardedWal.shard();
+    shard1.open();
+    shard1.append('record1');
+    shard1.append('record2');
+    shard1.close();
+
+    const shard2 = shardedWal.shard();
+    shard2.open();
+    shard2.append('record3');
+    shard2.close();
+
+    shardedWal.finalize();
+
+    const finalFile = path.join(
+      testDir,
+      shardedWal.groupId,
+      `trace.create-finalize.json`,
+    );
+    expect(fs.existsSync(finalFile)).toBeTrue();
+
+    const content = fs.readFileSync(finalFile, 'utf8');
+    const records = JSON.parse(content.trim());
+    expect(records).toEqual(['record1', 'record2', 'record3']);
+  });
+
+  it('should merge multiple shards correctly', () => {
+    shardedWal = new ShardedWal({
+      debug: false,
+      dir: testDir,
+      format: makeMockFormat({
+        baseName: 'merged',
+      }),
+      coordinatorIdEnvVar: SHARDED_WAL_COORDINATOR_ID_ENV_VAR,
+      groupId: 'merge-shards',
+    });
+
+    // eslint-disable-next-line functional/no-loop-statements
+    for (let i = 1; i <= 5; i++) {
+      const shard = shardedWal.shard();
+      shard.open();
+      shard.append(`record-from-shard-${i}`);
+      shard.close();
+    }
+
+    shardedWal.finalize();
+
+    const finalFile = path.join(
+      testDir,
+      shardedWal.groupId,
+      `merged.merge-shards.json`,
+    );
+    const content = fs.readFileSync(finalFile, 'utf8');
+    const records = JSON.parse(content.trim());
+    expect(records).toHaveLength(5);
+    expect(records[0]).toBe('record-from-shard-1');
+    expect(records[4]).toBe('record-from-shard-5');
+  });
+
+  it('should expose recovery details in stats when debug is true', () => {
+    shardedWal = new ShardedWal({
+      debug: true,
+      dir: testDir,
+      format: makeMockFormat({
+        baseName: 'test',
+      }),
+      coordinatorIdEnvVar: SHARDED_WAL_COORDINATOR_ID_ENV_VAR,
+      groupId: 'invalid-entries',
+    });
+
+    const shard = shardedWal.shard();
+    shard.open();
+    shard.append('valid1');
+    shard.append('invalid');
+    shard.append('valid2');
+    shard.close();
+
+    shardedWal.finalize();
+    // When debug is true, lastRecovery should contain recovery results
+    expect(shardedWal.stats.lastRecovery).toHaveLength(1);
+    expect(shardedWal.stats.lastRecovery[0]).toMatchObject({
+      file: expect.stringContaining('test.'),
+      result: expect.objectContaining({
+        records: expect.arrayContaining(['valid1', 'invalid', 'valid2']),
+        errors: [],
+        partialTail: null,
+      }),
+    });
+
+    const finalFile = path.join(
+      testDir,
+      shardedWal.groupId,
+      `test.invalid-entries.json`,
+    );
+    const content = fs.readFileSync(finalFile, 'utf8');
+    const records = JSON.parse(content.trim());
+    expect(records).toEqual(['valid1', 'invalid', 'valid2']);
+  });
+
+  it('should cleanup shard files after finalization', () => {
+    shardedWal = new ShardedWal({
+      debug: false,
+      dir: testDir,
+      format: makeMockFormat({
+        baseName: 'cleanup-test',
+      }),
+      coordinatorIdEnvVar: SHARDED_WAL_COORDINATOR_ID_ENV_VAR,
+      groupId: 'cleanup-test',
+    });
+
+    const shard1 = shardedWal.shard();
+    shard1.open();
+    shard1.append('record1');
+    shard1.close();
+
+    const shard2 = shardedWal.shard();
+    shard2.open();
+    shard2.append('record2');
+    shard2.close();
+
+    shardedWal.finalize();
+
+    const finalFile = path.join(
+      testDir,
+      shardedWal.groupId,
+      `cleanup-test.cleanup-test.json`,
+    );
+    expect(fs.existsSync(finalFile)).toBeTrue();
+
+    shardedWal.cleanupIfCoordinator();
+
+    const groupDir = path.join(testDir, shardedWal.groupId);
+    const files = fs.readdirSync(groupDir);
+    expect(files).not.toContain(expect.stringMatching(/cleanup-test.*\.log$/));
+    expect(files).toContain(`cleanup-test.cleanup-test.json`);
+  });
+
+  it('should use custom options in finalizer', () => {
+    shardedWal = new ShardedWal({
+      debug: false,
+      dir: testDir,
+      format: makeMockFormat({
+        baseName: 'custom',
+        finalizer: (records, opt) =>
+          `${JSON.stringify({ records, metadata: opt })}\n`,
+      }),
+      coordinatorIdEnvVar: SHARDED_WAL_COORDINATOR_ID_ENV_VAR,
+      groupId: 'custom-finalizer',
+    });
+
+    const shard = shardedWal.shard();
+    shard.open();
+    shard.append('record1');
+    shard.close();
+
+    shardedWal.finalize({ version: '2.0', timestamp: Date.now() });
+
+    const finalFile = path.join(
+      testDir,
+      shardedWal.groupId,
+      `custom.custom-finalizer.json`,
+    );
+    const content = fs.readFileSync(finalFile, 'utf8');
+    const result = JSON.parse(content.trim());
+    expect(result.records).toEqual(['record1']);
+    expect(result.metadata).toEqual({
+      version: '2.0',
+      timestamp: expect.any(Number),
+    });
+  });
+
+  it('should handle empty shards correctly', () => {
+    shardedWal = new ShardedWal({
+      debug: false,
+      dir: testDir,
+      format: makeMockFormat({
+        baseName: 'empty',
+      }),
+      coordinatorIdEnvVar: SHARDED_WAL_COORDINATOR_ID_ENV_VAR,
+      groupId: 'empty-shards',
+    });
+
+    const groupDir = path.join(testDir, shardedWal.groupId);
+    fs.mkdirSync(groupDir, { recursive: true });
+
+    shardedWal.finalize();
+
+    const finalFile = path.join(
+      testDir,
+      shardedWal.groupId,
+      `empty.${shardedWal.groupId}.json`,
+    );
+    expect(fs.existsSync(finalFile)).toBeTrue();
+    const content = fs.readFileSync(finalFile, 'utf8');
+    expect(content.trim()).toBe('[]');
+  });
+});