Class ShardedWal

packages/utils/src/lib/wal-sharded.ts:115–453

 */

export class ShardedWal<T extends WalRecord = WalRecord> {
  static instanceCount = 0;

  readonly #id: string = getUniqueInstanceId({
    next() {
      return ++ShardedWal.instanceCount;
    },
  });
  readonly groupId: string;
  readonly #debug: boolean = false;
  readonly #format: WalFormat<T>;
  readonly #dir: string = process.cwd();
  readonly #coordinatorIdEnvVar: string;
  #state: 'active' | 'finalized' | 'cleaned' = 'active';
  #lastRecovery: {
    file: string;
    result: RecoverResult<T | InvalidEntry<string>>;
  }[] = [];
  #createdShardFiles: string[] = [];

  /**
   * Initialize the given environment variable if not already set.
   * This must be done as early as possible before any user code runs.
   * Sets envVarName to the current instance ID if not already defined.
   *
   * @param envVarName - Environment variable name for storing coordinator ID
   * @param instanceID - The instance ID to set as coordinator
   */
  static setCoordinatorProcess(envVarName: string, instanceID: string): void {
    if (!process.env[envVarName]) {
      process.env[envVarName] = instanceID;
    }
  }

  /**
   * Determines if this process is the leader WAL process.
   *
   * The leader is the process that first enabled profiling over the given env var.
   * All descendant processes inherit the environment.
   *
   * @param envVarName - Environment variable name for storing coordinator ID
   * @param instanceID - The instance ID to check
   * @returns true if this is the leader WAL process, false otherwise
   */
  static isCoordinatorProcess(envVarName: string, instanceID: string): boolean {
    return process.env[envVarName] === instanceID;
  }

  /**
   * Create a sharded WAL manager.
   *
   * @param opt.dir - Base directory to store shard files (defaults to process.cwd())
   * @param opt.format - WAL format configuration
   * @param opt.groupId - Group ID for sharding (defaults to generated group ID)
   * @param opt.coordinatorIdEnvVar - Environment variable name for storing coordinator ID (defaults to CP_SHARDED_WAL_COORDINATOR_ID)
   * @param opt.autoCoordinator - Whether to auto-set the coordinator ID on construction (defaults to true)
   */
  constructor(opt: {