execStreamingInContainer function

Future<void> execStreamingInContainer(String name, List<String> command, String buildJobId, String runId, String token, { required Future<bool> isCancelled(), Duration timeout = maxJobTimeout, })

Implementation

Future<void> execStreamingInContainer(
  String name,
  List<String> command,
  String buildJobId,
  String runId,
  String token, {
  required Future<bool> Function() isCancelled,
  Duration timeout = maxJobTimeout,
}) async {
  final process = await Process.start('docker', ['exec', name, ...command]);

  await process.stdin.close();

  final stdoutCompleter = Completer<void>();
  final stderrCompleter = Completer<void>();
  final outputErrors = <String>[];
  var hasSuccessfulStep = false;

  void processLine(String line) {
    final trimmed = line.trim();
    if (trimmed.isEmpty || _isNoisyLine(trimmed)) return;

    if (trimmed.contains('✅') || trimmed.contains('Job succeeded')) {
      hasSuccessfulStep = true;
    }

    if (_isActError(trimmed)) {
      outputErrors.add(trimmed);
    }

    final cleanLine = stripActPrefix(trimmed);
    logInfo(buildJobId, runId, cleanLine);
  }

  process.stdout.transform(utf8.decoder).listen((data) {
    final masked = data.replaceAll(token, '***').trim();
    if (masked.isNotEmpty) {
      for (final line in LineSplitter.split(masked)) {
        processLine(line);
      }
    }
  }, onDone: () => stdoutCompleter.complete());

  process.stderr.transform(utf8.decoder).listen((data) {
    final masked = data.replaceAll(token, '***').trim();
    if (masked.isNotEmpty) {
      for (final line in LineSplitter.split(masked)) {
        processLine(line);
      }
    }
  }, onDone: () => stderrCompleter.complete());

  final startTime = DateTime.now();
  var isTimedOut = false;

  final cancelTimer = Timer.periodic(const Duration(seconds: 5), (_) async {
    if (DateTime.now().difference(startTime) > timeout) {
      isTimedOut = true;
      process.kill(ProcessSignal.sigterm);
      return;
    }
    if (await isCancelled()) {
      process.kill(ProcessSignal.sigterm);
    }
  });

  final exitCode = await process.exitCode;
  cancelTimer.cancel();
  await stdoutCompleter.future;
  await stderrCompleter.future;

  if (isTimedOut) {
    throw TimeoutException(
      'Job execution exceeded timeout of ${timeout.inMinutes} minutes.',
    );
  }

  if (exitCode != 0) {
    throw Exception('act exited with code $exitCode');
  }

  if (outputErrors.isNotEmpty) {
    throw Exception('act reported errors:\n${outputErrors.join('\n')}');
  }

  if (!hasSuccessfulStep) {
    throw Exception(
      'act exited with code 0 but no steps were executed. '
      'Check that your workflow has a valid runs-on key.',
    );
  }
}