Assessor_2.0/server/import-worker-client.ts

import { Worker } from 'node:worker_threads';
import type { ImportedDataset } from '../src/domain/dataset';
import { ImportError } from '../src/shared/errors';
import type { ImportExecutionRequest, ImportExecutor } from './import-executor';

interface WorkerResult {
  ok: boolean;
  dataset?: ImportedDataset;
  error?: { message: string; diagnostics: ConstructorParameters<typeof ImportError>[1] };
}

interface QueuedJob {
  request: ImportExecutionRequest;
  resolve(value: ImportedDataset): void;
  reject(reason: unknown): void;
}

export interface NodeImportWorkerExecutorOptions {
  workerUrl: URL;
  timeoutMs?: number;
  maximumConcurrent?: number;
  maximumQueued?: number;
  maximumQueuedBytes?: number;
  maximumOldGenerationMb?: number;
  maximumYoungGenerationMb?: number;
}

export class NodeImportWorkerExecutor implements ImportExecutor {
  private readonly timeoutMs: number;
  private readonly maximumConcurrent: number;
  private readonly maximumQueued: number;
  private readonly maximumQueuedBytes: number;
  private readonly maximumOldGenerationMb: number;
  private readonly maximumYoungGenerationMb: number;
  private readonly queue: QueuedJob[] = [];
  private readonly workers = new Set<Worker>();
  private active = 0;
  private queuedBytes = 0;
  private closed = false;

  constructor(private readonly options: NodeImportWorkerExecutorOptions) {
    this.timeoutMs = options.timeoutMs ?? 30_000;
    this.maximumConcurrent = options.maximumConcurrent ?? 2;
    this.maximumQueued = options.maximumQueued ?? 8;
    this.maximumQueuedBytes = options.maximumQueuedBytes ?? 256 * 1024 * 1024;
    this.maximumOldGenerationMb = options.maximumOldGenerationMb ?? 512;
    this.maximumYoungGenerationMb = options.maximumYoungGenerationMb ?? 64;
  }

  execute(request: ImportExecutionRequest): Promise<ImportedDataset> {
    if (this.closed) return Promise.reject(this.executionError('parser-executor-closed', 'Der Parserdienst wird beendet.'));
    const requestBytes = request.source.bytes.byteLength;
    if (this.active >= this.maximumConcurrent
      && (this.queue.length >= this.maximumQueued || this.queuedBytes + requestBytes > this.maximumQueuedBytes)) {
      return Promise.reject(this.executionError('parser-capacity-limit', 'Die lokale Parserwarteschlange ist ausgelastet. Bitte später erneut versuchen.'));
    }
    return new Promise<ImportedDataset>((resolve, reject) => {
      this.queue.push({ request, resolve, reject });
      this.queuedBytes += requestBytes;
      this.pump();
    });
  }

  async close(): Promise<void> {
    this.closed = true;
    const error = this.executionError('parser-executor-closed', 'Der Parserdienst wurde beendet.');
    this.queue.splice(0).forEach(job => job.reject(error));
    this.queuedBytes = 0;
    await Promise.all([...this.workers].map(worker => worker.terminate().catch(() => undefined)));
  }

  private executionError(code: string, message: string): ImportError {
    return new ImportError(message, [{ severity: 'error', code, message }]);
  }

  private pump(): void {
    while (!this.closed && this.active < this.maximumConcurrent && this.queue.length) {
      const job = this.queue.shift()!;
      this.queuedBytes -= job.request.source.bytes.byteLength;
      this.active += 1;
      void this.run(job).finally(() => {
        this.active -= 1;
        this.pump();
      });
    }
  }

  private run(job: QueuedJob): Promise<void> {
    return new Promise<void>(resolveJob => {
      const worker = new Worker(this.options.workerUrl, {
        workerData: job.request,
        resourceLimits: {
          maxOldGenerationSizeMb: this.maximumOldGenerationMb,
          maxYoungGenerationSizeMb: this.maximumYoungGenerationMb,
          stackSizeMb: 8
        }
      });
      this.workers.add(worker);
      let settled = false;
      const settle = (action: () => void) => {
        if (settled) return;
        settled = true;
        clearTimeout(timer);
        this.workers.delete(worker);
        action();
        void worker.terminate().finally(resolveJob);
      };
      const timer = setTimeout(() => settle(() => job.reject(this.executionError(
        'parser-time-limit', `Der Parser wurde nach ${this.timeoutMs} Millisekunden beendet.`
      ))), this.timeoutMs);
      worker.once('message', (result: WorkerResult) => settle(() => {
        if (result.ok && result.dataset) job.resolve(result.dataset);
        else job.reject(new ImportError(result.error?.message ?? 'Parser-Worker fehlgeschlagen.', result.error?.diagnostics ?? []));
      }));
      worker.once('error', error => settle(() => {
        const outOfMemory = 'code' in error && error.code === 'ERR_WORKER_OUT_OF_MEMORY';
        job.reject(this.executionError(
          outOfMemory ? 'parser-memory-limit' : 'parser-worker-error',
          outOfMemory ? 'Der Parser hat sein Speicherlimit überschritten.' : 'Der isolierte Parserprozess ist fehlgeschlagen.'
        ));
      }));
      worker.once('exit', code => {
        if (!settled) settle(() => job.reject(this.executionError('parser-worker-exit', `Der Parser-Worker wurde mit Status ${code} beendet.`)));
      });
    });
  }
}