refactor: wip

Author: BioPhoton

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

@@ -8,8 +8,9 @@ import {
 import { objectToEntries } from '../transform.js';
 import {
   asOptions,
+  errorToMarkerPayload,
   markerPayload,
- errorToMarkerPayload } from '../user-timing-extensibility-api-utils.js';
+} from '../user-timing-extensibility-api-utils.js';
 import type {
   ActionTrackEntryPayload,
   MarkerPayload,
@@ -109,27 +110,44 @@ export class NodejsProfiler<
    */
 
   constructor(options: NodejsProfilerOptions<DomainEvents, Tracks>) {
+// Pick ProfilerBufferOptions
     const {
       captureBufferedEntries,
       flushThreshold,
       maxQueueSize,
+      ...allButBufferOptions
+    } = options;
+    // Pick ProfilerPersistOptions
+    const {
       format: profilerFormat,
       measureName,
       outDir = PROFILER_PERSIST_OUT_DIR,
       enabled,
       debug,
       ...profilerOptions
-    } = options;
+    } = allButBufferOptions;
 
     super({ ...profilerOptions, enabled, debug });
 
-    this.#initializeStorage(profilerFormat, {
+    const { encodePerfEntry, ...format } = profilerFormat;
+
+    this.#sharder = new ShardedWal<DomainEvents>({
+      debug,
+      dir: process.env[PROFILER_OUT_DIR_ENV_VAR] ?? outDir,
+      format: parseWalFormat<DomainEvents>(format),
+      coordinatorIdEnvVar: PROFILER_SHARDER_ID_ENV_VAR,
+      measureNameEnvVar: PROFILER_MEASURE_NAME_ENV_VAR,
+      groupId: measureName,
+    });
+
+    this.#shard = this.#sharder.shard();
+    this.#performanceObserverSink = new PerformanceObserverSink({
+      sink: this.#shard,
+      encodePerfEntry,
       captureBufferedEntries,
       flushThreshold,
       maxQueueSize,
-      measureName,
-      outDir,
-      debug,
+      debug: this.isDebugMode(),
     });
 
     this.#unsubscribeExitHandlers = subscribeProcessExit({
@@ -151,40 +169,6 @@ export class NodejsProfiler<
     }
   }
 
-  #initializeStorage(
-    profilerFormat: ProfilerFormat<DomainEvents>,
-    options: {
-      captureBufferedEntries?: boolean;
-      flushThreshold?: number;
-      maxQueueSize?: number;
-      measureName?: string;
-      outDir: string;
-      debug?: boolean;
-    },
-  ) {
-    const { encodePerfEntry, ...format } = profilerFormat;
-    const { captureBufferedEntries, flushThreshold, maxQueueSize, measureName, outDir, debug } = options;
-
-    this.#sharder = new ShardedWal<DomainEvents>({
-      debug,
-      dir: process.env[PROFILER_OUT_DIR_ENV_VAR] ?? outDir,
-      format: parseWalFormat<DomainEvents>(format),
-      coordinatorIdEnvVar: PROFILER_SHARDER_ID_ENV_VAR,
-      measureNameEnvVar: PROFILER_MEASURE_NAME_ENV_VAR,
-      groupId: measureName,
-    });
-
-    this.#shard = this.#sharder.shard();
-    this.#performanceObserverSink = new PerformanceObserverSink({
-      sink: this.#shard,
-      encodePerfEntry,
-      captureBufferedEntries,
-      flushThreshold,
-      maxQueueSize,
-      debug: this.isDebugMode(),
-    });
-  }
-
   /**
    * Creates a performance marker for a profiler state transition.
    */
@@ -255,22 +239,22 @@ export class NodejsProfiler<
     switch (transition) {
       case 'idle->running':
         super.setEnabled(true);
-        this.#shard.open();
-        this.#performanceObserverSink.subscribe();
+        this.#shard?.open();
+        this.#performanceObserverSink?.subscribe();
         break;
 
       case 'running->idle':
         super.setEnabled(false);
-        this.#performanceObserverSink.unsubscribe();
-        this.#shard.close();
+        this.#performanceObserverSink?.unsubscribe();
+        this.#shard?.close();
         break;
 
       case 'running->closed':
       case 'idle->closed':
         super.setEnabled(false);
-        this.#performanceObserverSink.unsubscribe();
-        this.#shard.close();
-        this.#sharder.finalizeIfCoordinator();
+        this.#performanceObserverSink?.unsubscribe();
+        this.#shard?.close();
+        this.#sharder?.finalizeIfCoordinator();
         this.#unsubscribeExitHandlers?.();
         break;
 
@@ -317,17 +301,17 @@ export class NodejsProfiler<
       state: sharderState,
       isCoordinator,
       ...sharderStats
-    } = this.#sharder.stats;
+    } = this.#sharder?.stats ?? {};
 
     return {
       profilerState: this.#state,
       debug: this.isDebugMode(),
       sharderState,
       ...sharderStats,
       isCoordinator,
-      shardOpen: !this.#shard.isClosed(),
-      shardPath: this.#shard.getPath(),
-      ...this.#performanceObserverSink.getStats(),
+      shardOpen: this.#shard?.isClosed(),
+      shardPath: this.#shard?.getPath(),
+      ...this.#performanceObserverSink?.getStats(),
     };
   }
 
@@ -336,6 +320,6 @@ export class NodejsProfiler<
     if (this.#state === 'closed') {
       return; // No-op if closed
     }
-    this.#performanceObserverSink.flush();
+    this.#performanceObserverSink?.flush();
   }
 }

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

@@ -67,7 +67,7 @@ export class ShardedWal<T extends WalRecord = WalRecord> {
     },
   });
   readonly groupId = getUniqueTimeId();
-  readonly #debug = false;
+  readonly #debug: boolean = false;
   readonly #format: WalFormat<T>;
   readonly #dir: string = process.cwd();
   readonly #coordinatorIdEnvVar: string;
@@ -117,7 +117,7 @@ export class ShardedWal<T extends WalRecord = WalRecord> {
    * @param opt.measureNameEnvVar - Environment variable name for coordinating groupId across processes (optional)
    */
   constructor(opt: {
-    debug: boolean;
+    debug?: boolean;
     dir?: string;
     format: WalFormat<T>;
     groupId?: string;
@@ -128,12 +128,17 @@ export class ShardedWal<T extends WalRecord = WalRecord> {
     const {
       dir,
       format,
+      debug,
       groupId,
       coordinatorIdEnvVar,
       autoCoordinator = true,
       measureNameEnvVar,
     } = opt;
 
+    if (debug != null) {
+      this.#debug = debug;
+    }
+
     // Determine groupId: use provided, then env var, or generate
     // eslint-disable-next-line functional/no-let
     let resolvedGroupId: string;
@@ -295,6 +300,11 @@ export class ShardedWal<T extends WalRecord = WalRecord> {
       .map(entry => path.join(groupDir, entry));
   }
 
+  /** Get shard file paths created by this instance */
+  private getCreatedShardFiles() {
+    return this.#createdShardFiles.filter(f => fs.existsSync(f));
+  }
+
   /**
    * Finalize all shards by merging them into a single output file.
    * Recovers all records from all shards, validates no errors, and writes merged result.
@@ -355,9 +365,11 @@ export class ShardedWal<T extends WalRecord = WalRecord> {
       return;
     }
 
-    this.shardFiles().forEach(f => {
-      fs.unlinkSync(f);
-    });
+    this.getCreatedShardFiles()
+      .filter(f => fs.existsSync(f))
+      .forEach(f => {
+        fs.unlinkSync(f);
+      });
 
     this.#state = 'cleaned';
   }
@@ -367,13 +379,13 @@ export class ShardedWal<T extends WalRecord = WalRecord> {
       lastRecover: this.#lastRecovery,
       state: this.#state,
       groupId: this.groupId,
-      shardCount: this.shardFiles().length,
+      shardCount: this.getCreatedShardFiles().length,
       isCoordinator: this.isCoordinator(),
       isFinalized: this.isFinalized(),
       isCleaned: this.isCleaned(),
       finalFilePath: this.getFinalFilePath(),
-      shardFileCount: this.shardFiles().length,
-      shardFiles: this.shardFiles(),
+      shardFileCount: this.getCreatedShardFiles().length,
+      shardFiles: this.getCreatedShardFiles(),
     };
   }
 

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

@@ -502,27 +502,7 @@ describe('ShardedWal', () => {
 
       expect(sw.stats.shardFiles).toHaveLength(0);
       sw.shard();
-      expect(sw.stats.shardFiles).toHaveLength(1);
-
-      sw.cleanupIfCoordinator();
-      expect(sw.getState()).toBe('cleaned');
-      expect(sw.stats.shardFiles).toHaveLength(1);
-    });
-
-    it('should support cleanupIfCoordinator method', () => {
-      vol.fromJSON({
-        '/shards/20231114-221320-000/test.20231114-221320-000.10001.2.1.log':
-          'content1',
-      });
-
-      const sw = getShardedWal({
-        dir: '/shards',
-        format: { baseName: 'test', walExtension: '.log' },
-      });
-
       expect(sw.stats.shardFiles).toHaveLength(0);
-      sw.shard();
-      expect(sw.stats.shardFiles).toHaveLength(1);
 
       sw.cleanupIfCoordinator();
       expect(sw.getState()).toBe('cleaned');

packages/utils/src/lib/wal.ts

@@ -14,12 +14,12 @@ export type Codec<I, O = string> = {
 
 export type InvalidEntry<O = string> = { __invalid: true; raw: O };
 
-type CodecInput<C extends Codec<any, any>> =
-  C extends Codec<infer I, any> ? I : never;
-type CodecOutput<C extends Codec<any, any>> =
-  C extends Codec<any, infer O> ? O : never;
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
+type CodecInput<C> = C extends Codec<infer I, infer O> ? I : never;
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
+type CodecOutput<C> = C extends Codec<infer I, infer O> ? O : never;
 
-export type TolerantCodec<C extends Codec<any, any>> = Codec<
+export type TolerantCodec<C> = Codec<
   CodecInput<C> | InvalidEntry<CodecOutput<C>>,
   CodecOutput<C>
 >;