From 50ecc9030d93a79546d72541f3137702af665023 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Bia=C5=82y?= Date: Fri, 28 Aug 2026 15:10:28 +0200 Subject: [PATCH 1/3] feat(auto): live engine events, forward-compatible decoding, process control Deliver Pulumi engine events as they happen instead of only after the CLI exits, and stop a successful deployment from being reported as a failure. - OnEvent / EventLog / OnProcessStart options on preview/up/refresh/destroy - new EventLogs: tails the --event-log file on a daemon thread; parse errors are collected into `parseErrors` instead of failing the operation - Other(String) on DiffKind/ProgressType/OpType for unknown wire values; fix OpTypeFormat.write to emit JsString so it survives as a Map key - OperationFailedError keeps diagnostics/failures on the failure path, where the log was previously never parsed - ChildProcess exposes the pulumi process; interrupt() sends SIGINT, the only signal pulumi treats as a graceful cancel - shell.apply reimplemented as spawn + collect + join (from os-lib's call) - results gained resourcePreEvents/resourceOperations/failures/parseErrors tailf 0.1.0 -> 0.2.0, fixes a lot of tailf issues --- auto/project.scala | 5 +- .../scala/besom/auto/internal/AutoError.scala | 40 ++ .../besom/auto/internal/ChildProcess.scala | 40 ++ .../scala/besom/auto/internal/EventLogs.scala | 146 ++++++ .../scala/besom/auto/internal/Events.scala | 16 +- .../scala/besom/auto/internal/Stack.scala | 443 +++++++++++++----- .../scala/besom/auto/internal/shell.scala | 93 ++-- .../besom/auto/internal/EventLogsTest.scala | 161 +++++++ .../auto/internal/LocalWorkspaceTest.scala | 181 ++++++- .../besom/auto/internal/OptionsTest.scala | 84 ++++ .../scala/besom/auto/internal/ShellTest.scala | 78 +++ .../scala/besom/auto/internal/StackTest.scala | 45 ++ 12 files changed, 1169 insertions(+), 163 deletions(-) create mode 100644 auto/src/main/scala/besom/auto/internal/ChildProcess.scala create mode 100644 auto/src/main/scala/besom/auto/internal/EventLogs.scala create mode 100644 auto/src/test/scala/besom/auto/internal/EventLogsTest.scala create mode 100644 auto/src/test/scala/besom/auto/internal/OptionsTest.scala diff --git a/auto/project.scala b/auto/project.scala index 027d3ab3..7c6d2cc0 100644 --- a/auto/project.scala +++ b/auto/project.scala @@ -7,14 +7,15 @@ //> using dep org.virtuslab::besom-core:0.5.2-SNAPSHOT //> using dep org.virtuslab::besom-model:0.5.2-SNAPSHOT //> using dep org.virtuslab::scala-yaml:0.3.1 +//> using dep com.lihaoyi::geny:1.1.1 //> using dep com.lihaoyi::os-lib:0.11.8 //> using dep com.lihaoyi::os-lib-watch:0.11.8 //> using dep org.eclipse.jgit:org.eclipse.jgit:6.8.0.202311291450-r //> using dep org.eclipse.jgit:org.eclipse.jgit.ssh.jsch:6.8.0.202311291450-r -//> using dep org.slf4j:slf4j-nop:2.0.17 // TODO library should not have bindings for slf4j -//> using dep ma.chinespirit::tailf:0.1.0 +//> using dep ma.chinespirit::tailf:0.2.0 //> using test.dep org.scalameta::munit:1.2.4 +//> using test.dep org.slf4j:slf4j-nop:2.0.17 //> using publish.name "besom-auto" //> using publish.organization "org.virtuslab" diff --git a/auto/src/main/scala/besom/auto/internal/AutoError.scala b/auto/src/main/scala/besom/auto/internal/AutoError.scala index 44b19c35..18d6fa44 100644 --- a/auto/src/main/scala/besom/auto/internal/AutoError.scala +++ b/auto/src/main/scala/besom/auto/internal/AutoError.scala @@ -15,6 +15,46 @@ object AutoError: def apply(message: String, cause: Throwable) = new AutoError(Some(message), Some(cause)) def apply(cause: Throwable) = new AutoError(None, Some(cause)) +/** Raised when a `pulumi` lifecycle operation (up/preview/refresh/destroy) fails. + * + * The engine event log is parsed best-effort even on the failure path, so [[diagnostics]] and [[failures]] - the data that explains *why* + * the operation failed - stay available. The bulky stdout/stderr dump lives on the [[ShellAutoError]] in [[cause]]; the copies here are + * for programmatic access. + * + * @param operation + * the lifecycle operation that failed, one of `preview`, `up`, `refresh`, `destroy` + * @param exitCode + * the exit code of the `pulumi` process + * @param stdout + * the standard output of the `pulumi` process + * @param stderr + * the standard error of the `pulumi` process + * @param resourcePreEvents + * the resource operations the engine started before it gave up + * @param resourceOperations + * the resource operations that completed before the engine gave up + * @param failures + * the resource operations that failed + * @param diagnostics + * the diagnostic messages emitted by the engine and the providers + * @param parseErrors + * the event log lines that could not be decoded + */ +@SerialVersionUID(1L) +case class OperationFailedError( + message: Option[String], + cause: Option[Throwable], + operation: String, + exitCode: Int, + stdout: String, + stderr: String, + resourcePreEvents: List[ResourcePreEvent], + resourceOperations: List[ResOutputsEvent], + failures: List[ResOpFailedEvent], + diagnostics: List[DiagnosticEvent], + parseErrors: List[EventLogParseError] +) extends BaseAutoError(message, cause) + @SerialVersionUID(1L) case class ShellAutoError( message: Option[String], diff --git a/auto/src/main/scala/besom/auto/internal/ChildProcess.scala b/auto/src/main/scala/besom/auto/internal/ChildProcess.scala new file mode 100644 index 00000000..010cf9fa --- /dev/null +++ b/auto/src/main/scala/besom/auto/internal/ChildProcess.scala @@ -0,0 +1,40 @@ +package besom.auto.internal + +/** A handle to a `pulumi` process started by besom-auto, handed to callers that opted in with `ShellOption.OnStart` (or the per-operation + * `OnProcessStart` options) so that they can implement cancellation. + * + * The handle is live only for as long as the operation that produced it runs. + */ +final class ChildProcess private[auto] (val underlying: os.SubProcess): + + /** The operating system process id. */ + def pid: Long = underlying.wrapped.pid() + + /** Whether the process is still running. */ + def isAlive: Boolean = underlying.isAlive() + + /** Sends `SIGINT`, which is the only signal `pulumi` treats as "cancel gracefully" - it logs `^C received; cancelling` and unwinds the + * current step rather than dying mid-operation. Sending it a second time is what `pulumi` itself escalates to immediate termination, so + * forwarding a terminal's Ctrl-C straight through gives the usual two stage behaviour. + * + * The JDK exposes no signal API, so on Unix this shells out to `kill -INT`; on Windows, where there is no equivalent, it falls back to + * [[terminate]]. + */ + def interrupt(): Unit = + if isWindows then terminate() + else + // best effort - if the process is already gone kill exits non-zero and there is nothing to cancel + os.proc("kill", "-INT", pid.toString).call(check = false) + () + + /** Sends `SIGTERM` and, if the process is still alive after the grace period, `SIGKILL`. Note that `pulumi` does *not* treat `SIGTERM` as + * a cancellation - use [[interrupt]] for that. + */ + def terminate(): Unit = underlying.destroy() + + /** Sends `SIGKILL` immediately. The stack is very likely to be left with a pending operation. */ + def kill(): Unit = underlying.destroy(shutdownGracePeriod = 0) + + private def isWindows: Boolean = System.getProperty("os.name", "").toLowerCase.startsWith("windows") + +end ChildProcess diff --git a/auto/src/main/scala/besom/auto/internal/EventLogs.scala b/auto/src/main/scala/besom/auto/internal/EventLogs.scala new file mode 100644 index 00000000..52ad20b0 --- /dev/null +++ b/auto/src/main/scala/besom/auto/internal/EventLogs.scala @@ -0,0 +1,146 @@ +package besom.auto.internal + +import scala.util.Try +import scala.util.control.NonFatal + +/** A line of the engine event log that could not be decoded. + * + * Parse errors are collected instead of failing the operation they belong to - Pulumi is free to add new event types and new values of + * existing enums at any time and a successful deployment must not be reported as a failure because of it. + * + * @param lineNumber + * the 1-based number of the offending line in the event log + * @param line + * the raw content of the offending line + * @param error + * the decoding error + */ +case class EventLogParseError(lineNumber: Int, line: String, error: Exception) + +/** The result of decoding an engine event log. + * + * @param events + * all events that could be decoded, in the order the engine emitted them + * @param parseErrors + * all lines that could not be decoded + */ +case class ParsedEventLog(events: List[EngineEvent], parseErrors: List[EventLogParseError]) +object ParsedEventLog: + val empty: ParsedEventLog = ParsedEventLog(Nil, Nil) + +/** Everything related to reading Pulumi's `--event-log` file, both post-hoc and live. */ +private[auto] object EventLogs: + + /** Reads and decodes an engine event log. + * + * Undecodable lines are collected in [[ParsedEventLog.parseErrors]] and never fail the read; only a failure to read the file itself + * produces a `Left`. + * + * @param path + * the path of the event log + * @return + * the decoded log or an error if the file could not be read + */ + def parse(path: os.Path): Either[Exception, ParsedEventLog] = + Try(os.read.lines(path)).toEither.left + .map(e => AutoError(s"Failed to read event log: $path", e)) + .map { lines => + val (events, errors) = lines.iterator.zipWithIndex + .filter { case (line, _) => line.nonEmpty } + .foldLeft((List.empty[EngineEvent], List.empty[EventLogParseError])) { case ((events, errors), (line, idx)) => + EngineEvent.fromJson(line) match + case Right(event) => (event :: events, errors) + case Left(error) => (events, EventLogParseError(idx + 1, line, error) :: errors) + } + ParsedEventLog(events.reverse, errors.reverse) + } + end parse + + /** Finds the [[SummaryEvent]] emitted at the end of an operation. + * + * @param events + * the decoded events + * @param parseErrors + * the parse errors from the same log, used to explain a missing summary caused by a wire format drift + * @return + * the summary event or an error if there was none + */ + def summary(events: List[EngineEvent], parseErrors: List[EventLogParseError] = Nil): Either[Exception, SummaryEvent] = + events + .collectFirst { case e if e.summaryEvent.isDefined => e.summaryEvent.get } + .toRight { + val suffix = + if parseErrors.isEmpty then "" + else s" (${parseErrors.size} event log line(s) failed to parse, the summary event may be among them)" + AutoError(s"No summary event found in event log$suffix") + } + end summary + + def resourcePreEvents(events: List[EngineEvent]): List[ResourcePreEvent] = events.flatMap(_.resourcePreEvent) + def resourceOutputs(events: List[EngineEvent]): List[ResOutputsEvent] = events.flatMap(_.resOutputsEvent) + def failures(events: List[EngineEvent]): List[ResOpFailedEvent] = events.flatMap(_.resOpFailedEvent) + def diagnostics(events: List[EngineEvent]): List[DiagnosticEvent] = events.flatMap(_.diagnosticEvent) + + /** Runs `body` while tailing `path`, handing every decoded engine event to `handler` as it is written. + * + * The follower is opened before `body` runs, so no event written by the subprocess can be missed. When `body` returns the follower is + * stopped, which drains whatever is already on disk before signalling EOF, so events written just before the process exited are still + * delivered. + * + * Undecodable lines are dropped here - [[parse]] accounts for them post-hoc in `parseErrors`. Exceptions thrown by `handler` are + * swallowed so that a single misbehaving consumer cannot end the stream. + * + * Caveat: `tailf`'s follower treats a file that got shorter than the current read position as rotated and restarts from offset 0. + * Nothing truncates a Pulumi event log mid-run, but it is the only path that could replay events, so consumers should stay idempotent + * per URN. + * + * @param path + * the event log to tail, which must already exist + * @param handler + * the consumer of the events, or `None` to run `body` without tailing at all + * @param body + * the operation producing the events + * @return + * the result of `body`, or a `Left` if the follower could not be opened + */ + def around[A](path: os.Path, handler: Option[EngineEvent => Unit])(body: => Either[Exception, A]): Either[Exception, A] = + handler match + case None => body + case Some(_) if !os.exists(path) => + // tailf would happily wait for the file to appear; for us its absence is a setup error, not something to wait out + Left(AutoError(s"Cannot stream engine events, event log does not exist: $path")) + case Some(onEvent) => + shell.tail(path).flatMap { follower => + val reader = new Thread( + new Runnable: + def run(): Unit = + try + val lines = scala.io.Source.fromInputStream(follower)(using scala.io.Codec.UTF8).getLines() + while lines.hasNext do + val line = lines.next() + if line.nonEmpty then + EngineEvent.fromJson(line) match + case Right(event) => + try onEvent(event) + catch case NonFatal(_) => () // a broken consumer must not end the stream + case Left(_) => () // accounted for post-hoc by parse + catch case NonFatal(_) => () // the follower was closed from under us, nothing left to read + , + s"besom-auto-event-tail-${path.last}" + ) + reader.setDaemon(true) + reader.start() + + try body + finally + try + follower.stop() // drain what is already written, then EOF + reader.join(5000) // a reader parked in waitNewInput wakes within one rereadSleep + catch case NonFatal(_) => () + finally + try follower.close() + catch case NonFatal(_) => () + } + end around + +end EventLogs diff --git a/auto/src/main/scala/besom/auto/internal/Events.scala b/auto/src/main/scala/besom/auto/internal/Events.scala index 2c116469..fb145262 100644 --- a/auto/src/main/scala/besom/auto/internal/Events.scala +++ b/auto/src/main/scala/besom/auto/internal/Events.scala @@ -13,6 +13,12 @@ enum DiffKind(val value: String): case Update extends DiffKind("update") case UpdateReplace extends DiffKind("update-replace") + /** A diff kind emitted by the engine that this version of besom-auto does not know about. + * + * Kept so that a newer Pulumi CLI can not fail decoding of an otherwise valid event log. + */ + case Other(unknownValue: String) extends DiffKind(unknownValue) + def forcesReplacement: Boolean = value.endsWith("-replace") end DiffKind object DiffKind: @@ -23,7 +29,7 @@ object DiffKind: case "delete-replace" => DiffKind.DeleteReplace case "update" => DiffKind.Update case "update-replace" => DiffKind.UpdateReplace - case other => throw DeserializationException(s"Unknown DiffKind: $other") + case other => DiffKind.Other(other) given RootJsonFormat[DiffKind] with def write(obj: DiffKind): JsValue = JsString(obj.value) @@ -36,12 +42,18 @@ end DiffKind enum ProgressType(val value: String): case PluginDownload extends ProgressType("plugin-download") case PluginInstall extends ProgressType("plugin-install") + + /** A progress type emitted by the engine that this version of besom-auto does not know about. + * + * Kept so that a newer Pulumi CLI can not fail decoding of an otherwise valid event log. + */ + case Other(unknownValue: String) extends ProgressType(unknownValue) end ProgressType object ProgressType: def from(value: String): ProgressType = value match case "plugin-download" => ProgressType.PluginDownload case "plugin-install" => ProgressType.PluginInstall - case other => throw DeserializationException(s"Unknown ProgressType: $other") + case other => ProgressType.Other(other) given RootJsonFormat[ProgressType] with def write(obj: ProgressType): JsValue = JsString(obj.value) diff --git a/auto/src/main/scala/besom/auto/internal/Stack.scala b/auto/src/main/scala/besom/auto/internal/Stack.scala index 6f3513da..10be74db 100644 --- a/auto/src/main/scala/besom/auto/internal/Stack.scala +++ b/auto/src/main/scala/besom/auto/internal/Stack.scala @@ -87,52 +87,52 @@ case class Stack(name: String, workspace: Workspace): Seq(s"--exec-kind=${ExecKind.AutoInline}", s"--client=$address") else Seq(s"--exec-kind=${ExecKind.AutoLocal}") + val shellOpts: Seq[shell.ShellOption] = opts.onProcessStart.map(shell.ShellOption.OnStart(_)).toSeq + for - eventsPath <- eventLogsPath("preview") - r <- { + eventsPath <- eventLogsPath("preview", opts.eventLog) + r <- EventLogs.around(eventsPath, opts.onEvent) { val watchArgs: Seq[String] = Seq("--event-log=" + eventsPath) val args: Seq[String] = Seq("preview") ++ sharedArgs ++ kindArgs ++ watchArgs - pulumi(args)().left.map(AutoError("Preview failed", _)) // after this line pulumi cli is no longer running + pulumi(args)(shellOpts*).left.map(operationFailed("preview", eventsPath)) // after this line pulumi cli is no longer running } - events <- parseEventLog(eventsPath) - summary <- extractSummary(events) + parsed <- EventLogs.parse(eventsPath) + summary <- EventLogs.summary(parsed.events, parsed.parseErrors) yield PreviewResult( stdout = r.out, stderr = r.err, summary = summary.resourceChanges, - resourceChanges = extractResourcePreEvents(events), - diagnostics = extractDiagnostics(events) + resourceChanges = EventLogs.resourcePreEvents(parsed.events), + diagnostics = EventLogs.diagnostics(parsed.events), + parseErrors = parsed.parseErrors ) end preview - private def parseEventLog(path: os.Path): Either[Exception, List[EngineEvent]] = - val lines = os.read.lines(path).filter(_.nonEmpty) - val parsed = lines.map(EngineEvent.fromJson(_)) - val errors = parsed.collect { case Left(e) => e } - if errors.nonEmpty then - val err = errors.foldLeft(AutoError(s"Failed to parse ${errors.size} engine events")) { case (acc, e) => - acc.addSuppressed(e) - acc - } - Left(err) - else Right(parsed.collect { case Right(e) => e }.toList) - - private def extractSummary(events: List[EngineEvent]): Either[Exception, SummaryEvent] = - events - .collectFirst { case e if e.summaryEvent.isDefined => e.summaryEvent.get } - .toRight(AutoError("No summary event found in event log")) - - private def extractResourcePreEvents(events: List[EngineEvent]): List[ResourcePreEvent] = - events.flatMap(_.resourcePreEvent) - - private def extractDiagnostics(events: List[EngineEvent]): List[DiagnosticEvent] = - events.flatMap(_.diagnosticEvent) - - private def extractFailures(events: List[EngineEvent]): List[ResOpFailedEvent] = - events.flatMap(_.resOpFailedEvent) - - private def extractResourceOutputs(events: List[EngineEvent]): List[ResOutputsEvent] = - events.flatMap(_.resOutputsEvent) + /** Turns a failed `pulumi` invocation into an error that still carries whatever the engine managed to report before giving up. + * + * The event log is parsed best-effort here because the operation short-circuits on a non-zero exit code, which is exactly when the + * diagnostics and resource failures matter most. + */ + private def operationFailed(operation: String, eventsPath: os.Path)(error: ShellAutoError | AutoError): Exception = + error match + case err: ShellAutoError => + val parsed = EventLogs.parse(eventsPath).getOrElse(ParsedEventLog.empty) + OperationFailedError( + message = Some(s"${operation.capitalize} failed"), + cause = Some(err), + operation = operation, + exitCode = err.exitCode, + stdout = err.stdout, + stderr = err.stderr, + resourcePreEvents = EventLogs.resourcePreEvents(parsed.events), + resourceOperations = EventLogs.resourceOutputs(parsed.events), + failures = EventLogs.failures(parsed.events), + diagnostics = EventLogs.diagnostics(parsed.events), + parseErrors = parsed.parseErrors + ) + // the remote workspace pre/post command callbacks - no pulumi process, no event log + case err: AutoError => AutoError(s"${operation.capitalize} failed", err) + end operationFailed /** Create or update the resources in a stack by executing the program in the Workspace. Update updates the resources in a stack by * executing the program in the Workspace associated with this stack, if one is provided. @@ -167,16 +167,15 @@ case class Stack(name: String, workspace: Workspace): Seq(s"--exec-kind=${ExecKind.AutoInline}", s"--client=$address") else Seq(s"--exec-kind=${ExecKind.AutoLocal}") + val shellOpts: Seq[shell.ShellOption] = opts.onProcessStart.map(shell.ShellOption.OnStart(_)).toSeq + for - eventsPath <- eventLogsPath("up") - r <- { + eventsPath <- eventLogsPath("up", opts.eventLog) + r <- EventLogs.around(eventsPath, opts.onEvent) { val watchArgs: Seq[String] = Seq("--event-log=" + eventsPath) val args: Seq[String] = Seq("up", "--yes", "--skip-preview") ++ sharedArgs ++ kindArgs ++ watchArgs - pulumi(args)( -// FIXME: missing streams, implement progressStreams and errorProgressStreams -// shell.Option.Stdout(opts.progressStreams), -// shell.Option.Stderr(opts.errorProgressStreams) - ).left.map(AutoError("Up failed", _)) + // FIXME: missing streams, implement progressStreams and errorProgressStreams + pulumi(args)(shellOpts*).left.map(operationFailed("up", eventsPath)) } outputs <- outputs history <- history( @@ -185,15 +184,17 @@ case class Stack(name: String, workspace: Workspace): /* If it's a remote workspace, don't set ShowSecrets to prevent attempting to load the project file. */ Option.when(opts.showSecrets && !isRemote)(HistoryOption.ShowSecrets).toSeq* ).flatMap(_.headOption.toRight(AutoError("Failed to get history, result was empty"))) - events <- parseEventLog(eventsPath) + parsed <- EventLogs.parse(eventsPath) yield UpResult( stdout = r.out, stderr = r.err, outputs = outputs, summary = history, - resourceOperations = extractResourceOutputs(events), - failures = extractFailures(events), - diagnostics = extractDiagnostics(events) + resourceOperations = EventLogs.resourceOutputs(parsed.events), + failures = EventLogs.failures(parsed.events), + diagnostics = EventLogs.diagnostics(parsed.events), + resourcePreEvents = EventLogs.resourcePreEvents(parsed.events), + parseErrors = parsed.parseErrors ) end up @@ -220,12 +221,14 @@ case class Stack(name: String, workspace: Workspace): ++ (if workspace.program.isDefined then Seq(s"--exec-kind=${ExecKind.AutoInline}") else Seq(s"--exec-kind=${ExecKind.AutoLocal}")) ++ remoteArgs // Apply the remote args, if needed + val shellOpts: Seq[shell.ShellOption] = opts.onProcessStart.map(shell.ShellOption.OnStart(_)).toSeq + for - eventsPath <- eventLogsPath("refresh") - r <- { + eventsPath <- eventLogsPath("refresh", opts.eventLog) + r <- EventLogs.around(eventsPath, opts.onEvent) { val watchArgs: Seq[String] = Seq("--event-log=" + eventsPath) val args: Seq[String] = Seq("refresh", "--yes", "--skip-preview") ++ sharedArgs ++ watchArgs - pulumi(args)().left.map(AutoError("Refresh failed", _)) + pulumi(args)(shellOpts*).left.map(operationFailed("refresh", eventsPath)) } history <- history( pageSize = 1, @@ -233,12 +236,16 @@ case class Stack(name: String, workspace: Workspace): /* If it's a remote workspace, don't set ShowSecrets to prevent attempting to load the project file. */ Option.when(opts.showSecrets && !isRemote)(HistoryOption.ShowSecrets).toSeq* ).flatMap(_.headOption.toRight(AutoError("Failed to get history, result was empty"))) - events <- parseEventLog(eventsPath) + parsed <- EventLogs.parse(eventsPath) yield RefreshResult( stdout = r.out, stderr = r.err, summary = history, - diagnostics = extractDiagnostics(events) + diagnostics = EventLogs.diagnostics(parsed.events), + resourcePreEvents = EventLogs.resourcePreEvents(parsed.events), + resourceOperations = EventLogs.resourceOutputs(parsed.events), + failures = EventLogs.failures(parsed.events), + parseErrors = parsed.parseErrors ) end refresh @@ -264,12 +271,14 @@ case class Stack(name: String, workspace: Workspace): ++ (if workspace.program.isDefined then Seq(s"--exec-kind=${ExecKind.AutoInline}") else Seq(s"--exec-kind=${ExecKind.AutoLocal}")) ++ remoteArgs // Apply the remote args, if needed + val shellOpts: Seq[shell.ShellOption] = opts.onProcessStart.map(shell.ShellOption.OnStart(_)).toSeq + for - eventsPath <- eventLogsPath("destroy") - r <- { + eventsPath <- eventLogsPath("destroy", opts.eventLog) + r <- EventLogs.around(eventsPath, opts.onEvent) { val watchArgs: Seq[String] = Seq("--event-log=" + eventsPath) val args: Seq[String] = Seq("destroy", "--yes", "--skip-preview") ++ sharedArgs ++ watchArgs - pulumi(args)().left.map(AutoError("Destroy failed", _)) + pulumi(args)(shellOpts*).left.map(operationFailed("destroy", eventsPath)) } history <- history( pageSize = 1, @@ -277,13 +286,16 @@ case class Stack(name: String, workspace: Workspace): /* If it's a remote workspace, don't set ShowSecrets to prevent attempting to load the project file. */ Option.when(opts.showSecrets && !isRemote)(HistoryOption.ShowSecrets).toSeq* ).flatMap(_.headOption.toRight(AutoError("Failed to get history, result was empty"))) - events <- parseEventLog(eventsPath) + parsed <- EventLogs.parse(eventsPath) yield DestroyResult( stdout = r.out, stderr = r.err, summary = history, - failures = extractFailures(events), - diagnostics = extractDiagnostics(events) + failures = EventLogs.failures(parsed.events), + diagnostics = EventLogs.diagnostics(parsed.events), + resourcePreEvents = EventLogs.resourcePreEvents(parsed.events), + resourceOperations = EventLogs.resourceOutputs(parsed.events), + parseErrors = parsed.parseErrors ) end destroy @@ -542,14 +554,30 @@ case class Stack(name: String, workspace: Workspace): case _ => throw AutoError(s"Unknown workspace type: ${workspace.getClass.getTypeName}") end remoteArgs - private def eventLogsPath(command: String): Either[Exception, os.Path] = + /** Creates the file `pulumi --event-log` will write to. + * + * The file has to exist before `pulumi` starts so that a follower can be attached to it without racing the engine. + * + * @param command + * the lifecycle operation the log belongs to, used to name the temporary directory + * @param requested + * a caller supplied path, which is truncated if it already exists; when not provided a fresh temporary directory is used + */ + private def eventLogsPath(command: String, requested: NotProvidedOr[os.Path]): Either[Exception, os.Path] = Try { - val logsDir = os.temp.dir(prefix = s"automation-logs-$command-") - val path = logsDir / "eventlog.txt" - os.write(path, "") + val path = requested.asOption match + case Some(p) => + os.makeDir.all(p / os.up) + os.write.over(p, "") // os.write throws if the file already exists + p + case None => + val logsDir = os.temp.dir(prefix = s"automation-logs-$command-") + val p = logsDir / "eventlog.txt" + os.write(p, "") + p if !os.exists(path) then throw AutoError(s"Failed to create event log file: $path") path - }.toEither.left.map(e => AutoError("Failed to create temporary directory for event logs", e)) + }.toEither.left.map(e => AutoError(s"Failed to create event log file for $command", e)) end eventLogsPath private def startLanguageRuntimeServer(): String = @@ -673,9 +701,28 @@ object PreviewOption: */ // case class ErrorProgressStreams(writers: os.ProcessOutput*) extends PreviewOption - /** Allows specifying one or more channels to receive the Pulumi event stream + /** Delivers every engine event to `handler` as the operation produces it, rather than only once it has finished. + * + * The handler is invoked on a besom-auto owned daemon thread that tails the engine event log, so it must be thread safe; exceptions it + * throws are swallowed so that a broken consumer cannot end the stream. Consumers should also stay idempotent per URN - the only way an + * event can be seen twice is a truncated log, which nothing does mid-run, but the tailing reader would restart from the beginning if it + * happened. */ -// case class EventStreams(channels: EngineEvent*) extends PreviewOption + case class OnEvent(handler: EngineEvent => Unit) extends PreviewOption + + /** Writes the engine event log to the given path instead of a temporary directory, so that it survives the operation. + * + * The file is created, or truncated if it already exists, before `pulumi` starts. + */ + case class EventLog(path: os.Path) extends PreviewOption + + /** Hands `handler` a [[ChildProcess]] handle to the `pulumi` process so that the operation can be cancelled - see + * [[ChildProcess.interrupt]] for the graceful, Ctrl-C equivalent. + * + * The handler runs on the calling thread right after the process is spawned and before it is waited on, so it must not block - stash the + * handle and return. + */ + case class OnProcessStart(handler: ChildProcess => Unit) extends PreviewOption /** Specifies the agent responsible for the update, stored in backends as "environment.exec.agent" */ @@ -724,8 +771,12 @@ end PreviewOption * allows specifying one or more io.Writers to redirect incremental preview stdout * @param errorProgressStreams * allows specifying one or more io.Writers to redirect incremental preview stderr - * @param eventStreams - * allows specifying one or more channels to receive the Pulumi event stream + * @param onEvent + * receives every engine event as the operation produces it + * @param eventLog + * writes the engine event log to the given path instead of a temporary directory + * @param onProcessStart + * receives a handle to the `pulumi` process, so that the operation can be cancelled * @param userAgent * the agent responsible for the update, stored in backends as "environment.exec.agent" * @param color @@ -748,7 +799,9 @@ private[auto] case class PreviewOptions( debugLogOpts: LoggingOptions = LoggingOptions(), // progressStreams: List[os.ProcessOutput] = List.empty, // TODO: implement multiple writers // errorProgressStreams: List[os.ProcessOutput] = List.empty, // TODO: implement multiple writers -// eventStreams: List[EngineEvent] = List.empty, // TODO: implement EngineEvent + onEvent: Option[EngineEvent => Unit] = None, + eventLog: NotProvidedOr[os.Path] = NotProvided, + onProcessStart: Option[ChildProcess => Unit] = None, userAgent: NotProvidedOr[String] = NotProvided, color: NotProvidedOr[Color] = NotProvided, plan: NotProvidedOr[os.Path] = NotProvided, @@ -775,7 +828,9 @@ object PreviewOptions: // TODO: missing streams // case PreviewOption.ProgressStreams(writers) :: tail => from(tail*).copy(progressStreams = writers) // case PreviewOption.ErrorProgressStreams(writers) :: tail => from(tail*).copy(errorProgressStreams = writers) -// case PreviewOption.EventStreams(channels) :: tail => from(tail*).copy(eventStreams = channels) + case PreviewOption.OnEvent(handler) :: tail => from(tail*).copy(onEvent = Some(handler)) + case PreviewOption.EventLog(path) :: tail => from(tail*).copy(eventLog = path) + case PreviewOption.OnProcessStart(handler) :: tail => from(tail*).copy(onProcessStart = Some(handler)) case PreviewOption.UserAgent(agent) :: tail => from(tail*).copy(userAgent = agent) case PreviewOption.Color(color) :: tail => from(tail*).copy(color = color) case PreviewOption.Plan(path) :: tail => from(tail*).copy(plan = path) @@ -834,9 +889,29 @@ object UpOption: /** Allows specifying one or more io.Writers to redirect incremental update stderr */ // case class ErrorProgressStreams(writers: os.ProcessOutput*) extends UpOption - /** Allows specifying one or more channels to receive the Pulumi event stream + + /** Delivers every engine event to `handler` as the operation produces it, rather than only once it has finished. + * + * The handler is invoked on a besom-auto owned daemon thread that tails the engine event log, so it must be thread safe; exceptions it + * throws are swallowed so that a broken consumer cannot end the stream. Consumers should also stay idempotent per URN - the only way an + * event can be seen twice is a truncated log, which nothing does mid-run, but the tailing reader would restart from the beginning if it + * happened. + */ + case class OnEvent(handler: EngineEvent => Unit) extends UpOption + + /** Writes the engine event log to the given path instead of a temporary directory, so that it survives the operation. + * + * The file is created, or truncated if it already exists, before `pulumi` starts. */ -// case class EventStreams(channels: EngineEvent*) extends UpOption + case class EventLog(path: os.Path) extends UpOption + + /** Hands `handler` a [[ChildProcess]] handle to the `pulumi` process so that the operation can be cancelled - see + * [[ChildProcess.interrupt]] for the graceful, Ctrl-C equivalent. + * + * The handler runs on the calling thread right after the process is spawned and before it is waited on, so it must not block - stash the + * handle and return. + */ + case class OnProcessStart(handler: ChildProcess => Unit) extends UpOption /** Specifies the agent responsible for the update, stored in backends as "environment.exec.agent" */ @@ -889,8 +964,12 @@ end UpOption * allows specifying one or more io.Writers to redirect incremental preview stdout * @param errorProgressStreams * allows specifying one or more io.Writers to redirect incremental preview stderr - * @param eventStreams - * allows specifying one or more channels to receive the Pulumi event stream + * @param onEvent + * receives every engine event as the operation produces it + * @param eventLog + * writes the engine event log to the given path instead of a temporary directory + * @param onProcessStart + * receives a handle to the `pulumi` process, so that the operation can be cancelled * @param userAgent * specifies the agent responsible for the update, stored in backends as "environment.exec.agent" * @param color @@ -915,7 +994,9 @@ case class UpOptions( debugLogOpts: LoggingOptions = LoggingOptions(), // progressStreams: List[os.ProcessOutput] = List.empty, // TODO: implement multiple writers // errorProgressStreams: List[os.ProcessOutput] = List.empty, // TODO: implement multiple writers -// eventStreams: List[EngineEvent] = List.empty, // TODO: implement EngineEvent + onEvent: Option[EngineEvent => Unit] = None, + eventLog: NotProvidedOr[os.Path] = NotProvided, + onProcessStart: Option[ChildProcess => Unit] = None, userAgent: NotProvidedOr[String] = NotProvided, color: NotProvidedOr[Color] = NotProvided, plan: NotProvidedOr[os.Path] = NotProvided, @@ -938,7 +1019,9 @@ object UpOptions: // TODO: missing streams // case UpOption.ProgressStreams(writers) :: tail => from(tail*).copy(progressStreams = writers) // case UpOption.ErrorProgressStreams(writers) :: tail => from(tail*).copy(errorProgressStreams = writers) -// case UpOption.EventStreams(channels) :: tail => from(tail*).copy(eventStreams = channels) + case UpOption.OnEvent(handler) :: tail => from(tail*).copy(onEvent = Some(handler)) + case UpOption.EventLog(path) :: tail => from(tail*).copy(eventLog = path) + case UpOption.OnProcessStart(handler) :: tail => from(tail*).copy(onProcessStart = Some(handler)) case UpOption.UserAgent(agent) :: tail => from(tail*).copy(userAgent = agent) case UpOption.Color(color) :: tail => from(tail*).copy(color = color) case UpOption.Plan(path) :: tail => from(tail*).copy(plan = path) @@ -982,9 +1065,29 @@ object RefreshOption: /** Allows specifying one or more io.Writers to redirect incremental refresh stderr */ // case class ErrorProgressStreams(writers: os.ProcessOutput*) extends RefreshOption - /** Allows specifying one or more channels to receive the Pulumi event stream + + /** Delivers every engine event to `handler` as the operation produces it, rather than only once it has finished. + * + * The handler is invoked on a besom-auto owned daemon thread that tails the engine event log, so it must be thread safe; exceptions it + * throws are swallowed so that a broken consumer cannot end the stream. Consumers should also stay idempotent per URN - the only way an + * event can be seen twice is a truncated log, which nothing does mid-run, but the tailing reader would restart from the beginning if it + * happened. + */ + case class OnEvent(handler: EngineEvent => Unit) extends RefreshOption + + /** Writes the engine event log to the given path instead of a temporary directory, so that it survives the operation. + * + * The file is created, or truncated if it already exists, before `pulumi` starts. + */ + case class EventLog(path: os.Path) extends RefreshOption + + /** Hands `handler` a [[ChildProcess]] handle to the `pulumi` process so that the operation can be cancelled - see + * [[ChildProcess.interrupt]] for the graceful, Ctrl-C equivalent. + * + * The handler runs on the calling thread right after the process is spawned and before it is waited on, so it must not block - stash the + * handle and return. */ -// case class EventStreams(channels: EngineEvent*) extends RefreshOption + case class OnProcessStart(handler: ChildProcess => Unit) extends RefreshOption /** Specifies additional settings for debug logging */ @@ -1024,8 +1127,12 @@ end RefreshOption * allows specifying one or more io.Writers to redirect incremental refresh stdout * @param errorProgressStreams * allows specifying one or more io.Writers to redirect incremental refresh stderr - * @param eventStreams - * allows specifying one or more channels to receive the Pulumi event stream + * @param onEvent + * receives every engine event as the operation produces it + * @param eventLog + * writes the engine event log to the given path instead of a temporary directory + * @param onProcessStart + * receives a handle to the `pulumi` process, so that the operation can be cancelled * @param userAgent * specifies the agent responsible for the refresh, stored in backends as "environment.exec.agent" * @param color @@ -1041,7 +1148,9 @@ case class RefreshOptions( debugLogOpts: LoggingOptions = LoggingOptions(), // progressStreams: List[os.ProcessOutput] = List.empty, // TODO: implement multiple writers // errorProgressStreams: List[os.ProcessOutput] = List.empty, // TODO: implement multiple writers -// eventStreams: List[EngineEvent] = List.empty, // TODO: implement EngineEvent + onEvent: Option[EngineEvent => Unit] = None, + eventLog: NotProvidedOr[os.Path] = NotProvided, + onProcessStart: Option[ChildProcess => Unit] = None, userAgent: NotProvidedOr[String] = NotProvided, color: NotProvidedOr[Color] = NotProvided, showSecrets: Boolean = false @@ -1058,12 +1167,14 @@ object RefreshOptions: // TODO: missing streams // case RefreshOption.ProgressStreams(writers) :: tail => from(tail*).copy(progressStreams = writers) // case RefreshOption.ErrorProgressStreams(writers) :: tail => from(tail*).copy(errorProgressStreams = writers) -// case RefreshOption.EventStreams(channels) :: tail => from(tail*).copy(eventStreams = channels) - case RefreshOption.UserAgent(agent) :: tail => from(tail*).copy(userAgent = agent) - case RefreshOption.Color(color) :: tail => from(tail*).copy(color = color) - case RefreshOption.ShowSecrets :: tail => from(tail*).copy(showSecrets = true) - case Nil => RefreshOptions() - case o => throw AutoError(s"Unknown refresh option: $o") + case RefreshOption.OnEvent(handler) :: tail => from(tail*).copy(onEvent = Some(handler)) + case RefreshOption.EventLog(path) :: tail => from(tail*).copy(eventLog = path) + case RefreshOption.OnProcessStart(handler) :: tail => from(tail*).copy(onProcessStart = Some(handler)) + case RefreshOption.UserAgent(agent) :: tail => from(tail*).copy(userAgent = agent) + case RefreshOption.Color(color) :: tail => from(tail*).copy(color = color) + case RefreshOption.ShowSecrets :: tail => from(tail*).copy(showSecrets = true) + case Nil => RefreshOptions() + case o => throw AutoError(s"Unknown refresh option: $o") end RefreshOptions @@ -1097,9 +1208,29 @@ object DestroyOption: /** Allows specifying one or more io.Writers to redirect incremental destroy stderr */ // case class ErrorProgressStreams(writers: os.ProcessOutput*) extends DestroyOption - /** Allows specifying one or more channels to receive the Pulumi event stream + + /** Delivers every engine event to `handler` as the operation produces it, rather than only once it has finished. + * + * The handler is invoked on a besom-auto owned daemon thread that tails the engine event log, so it must be thread safe; exceptions it + * throws are swallowed so that a broken consumer cannot end the stream. Consumers should also stay idempotent per URN - the only way an + * event can be seen twice is a truncated log, which nothing does mid-run, but the tailing reader would restart from the beginning if it + * happened. */ -// case class EventStreams(channels: EngineEvent*) extends DestroyOption + case class OnEvent(handler: EngineEvent => Unit) extends DestroyOption + + /** Writes the engine event log to the given path instead of a temporary directory, so that it survives the operation. + * + * The file is created, or truncated if it already exists, before `pulumi` starts. + */ + case class EventLog(path: os.Path) extends DestroyOption + + /** Hands `handler` a [[ChildProcess]] handle to the `pulumi` process so that the operation can be cancelled - see + * [[ChildProcess.interrupt]] for the graceful, Ctrl-C equivalent. + * + * The handler runs on the calling thread right after the process is spawned and before it is waited on, so it must not block - stash the + * handle and return. + */ + case class OnProcessStart(handler: ChildProcess => Unit) extends DestroyOption /** Specifies additional settings for debug logging */ @@ -1139,8 +1270,12 @@ end DestroyOption * allows specifying one or more io.Writers to redirect incremental destroy stdout * @param errorProgressStreams * allows specifying one or more io.Writers to redirect incremental destroy stderr - * @param eventStreams - * allows specifying one or more channels to receive the Pulumi event stream + * @param onEvent + * receives every engine event as the operation produces it + * @param eventLog + * writes the engine event log to the given path instead of a temporary directory + * @param onProcessStart + * receives a handle to the `pulumi` process, so that the operation can be cancelled * @param userAgent * specifies the agent responsible for the destroy, stored in backends as "environment.exec.agent" * @param color @@ -1156,7 +1291,9 @@ case class DestroyOptions( debugLogOpts: LoggingOptions = LoggingOptions(), // progressStreams: List[os.ProcessOutput] = List.empty, // TODO: implement multiple writers // errorProgressStreams: List[os.ProcessOutput] = List.empty, // TODO: implement multiple writers -// eventStreams: List[EngineEvent] = List.empty, // TODO: implement EngineEvent + onEvent: Option[EngineEvent => Unit] = None, + eventLog: NotProvidedOr[os.Path] = NotProvided, + onProcessStart: Option[ChildProcess => Unit] = None, userAgent: NotProvidedOr[String] = NotProvided, color: NotProvidedOr[Color] = NotProvided, showSecrets: Boolean = false @@ -1172,12 +1309,14 @@ object DestroyOptions: case DestroyOption.DebugLogging(debugOpts) :: tail => from(tail*).copy(debugLogOpts = debugOpts) // case DestroyOption.ProgressStreams(writers) :: tail => from(tail*).copy(progressStreams = writers) // case DestroyOption.ErrorProgressStreams(writers) :: tail => from(tail*).copy(errorProgressStreams = writers) -// case DestroyOption.EventStreams(channels) :: tail => from(tail*).copy(eventStreams = channels) - case DestroyOption.UserAgent(agent) :: tail => from(tail*).copy(userAgent = agent) - case DestroyOption.Color(color) :: tail => from(tail*).copy(color = color) - case DestroyOption.ShowSecrets :: tail => from(tail*).copy(showSecrets = true) - case Nil => DestroyOptions() - case o => throw AutoError(s"Unknown destroy option: $o") + case DestroyOption.OnEvent(handler) :: tail => from(tail*).copy(onEvent = Some(handler)) + case DestroyOption.EventLog(path) :: tail => from(tail*).copy(eventLog = path) + case DestroyOption.OnProcessStart(handler) :: tail => from(tail*).copy(onProcessStart = Some(handler)) + case DestroyOption.UserAgent(agent) :: tail => from(tail*).copy(userAgent = agent) + case DestroyOption.Color(color) :: tail => from(tail*).copy(color = color) + case DestroyOption.ShowSecrets :: tail => from(tail*).copy(showSecrets = true) + case Nil => DestroyOptions() + case o => throw AutoError(s"Unknown destroy option: $o") end DestroyOptions @@ -1277,13 +1416,21 @@ end Color * standard error * @param summary * the expected changes + * @param resourceChanges + * the resource operations the engine would perform + * @param diagnostics + * the diagnostic messages emitted by the engine and the providers + * @param parseErrors + * the engine event log lines that could not be decoded - non-empty means the operation still succeeded but some of the data above may be + * incomplete */ case class PreviewResult( stdout: String, stderr: String, summary: Map[OpType, Int], resourceChanges: List[ResourcePreEvent] = Nil, - diagnostics: List[DiagnosticEvent] = Nil + diagnostics: List[DiagnosticEvent] = Nil, + parseErrors: List[EventLogParseError] = Nil ): def permalink: Either[Exception, String] = ??? // TODO: implement GetPermalink end PreviewResult @@ -1298,6 +1445,17 @@ end PreviewResult * the stack outputs * @param summary * the deployed changes + * @param resourceOperations + * the resource operations that completed, with their outputs + * @param failures + * the resource operations that failed + * @param diagnostics + * the diagnostic messages emitted by the engine and the providers + * @param resourcePreEvents + * the start of each resource operation, emitted before the operation was performed + * @param parseErrors + * the engine event log lines that could not be decoded - non-empty means the operation still succeeded but some of the data above may be + * incomplete */ case class UpResult( stdout: String, @@ -1306,7 +1464,9 @@ case class UpResult( summary: UpdateSummary, resourceOperations: List[ResOutputsEvent] = Nil, failures: List[ResOpFailedEvent] = Nil, - diagnostics: List[DiagnosticEvent] = Nil + diagnostics: List[DiagnosticEvent] = Nil, + resourcePreEvents: List[ResourcePreEvent] = Nil, + parseErrors: List[EventLogParseError] = Nil ): def permalink: Either[Exception, String] = ??? // TODO: implement GetPermalink end UpResult @@ -1319,12 +1479,27 @@ end UpResult * standard error * @param summary * the deployed changes + * @param diagnostics + * the diagnostic messages emitted by the engine and the providers + * @param resourcePreEvents + * the start of each resource operation, emitted before the operation was performed + * @param resourceOperations + * the resource operations that completed, with their outputs + * @param failures + * the resource operations that failed + * @param parseErrors + * the engine event log lines that could not be decoded - non-empty means the operation still succeeded but some of the data above may be + * incomplete */ case class RefreshResult( stdout: String, stderr: String, summary: UpdateSummary, - diagnostics: List[DiagnosticEvent] = Nil + diagnostics: List[DiagnosticEvent] = Nil, + resourcePreEvents: List[ResourcePreEvent] = Nil, + resourceOperations: List[ResOutputsEvent] = Nil, + failures: List[ResOpFailedEvent] = Nil, + parseErrors: List[EventLogParseError] = Nil ): def permalink: Either[Exception, String] = ??? // TODO: implement GetPermalink end RefreshResult @@ -1337,13 +1512,27 @@ end RefreshResult * standard error * @param summary * the deployed changes + * @param failures + * the resource operations that failed + * @param diagnostics + * the diagnostic messages emitted by the engine and the providers + * @param resourcePreEvents + * the start of each resource operation, emitted before the operation was performed + * @param resourceOperations + * the resource operations that completed, with their outputs + * @param parseErrors + * the engine event log lines that could not be decoded - non-empty means the operation still succeeded but some of the data above may be + * incomplete */ case class DestroyResult( stdout: String, stderr: String, summary: UpdateSummary, failures: List[ResOpFailedEvent] = Nil, - diagnostics: List[DiagnosticEvent] = Nil + diagnostics: List[DiagnosticEvent] = Nil, + resourcePreEvents: List[ResourcePreEvent] = Nil, + resourceOperations: List[ResOutputsEvent] = Nil, + parseErrors: List[EventLogParseError] = Nil ): def permalink: Either[Exception, String] = ??? // TODO: implement GetPermalink end DestroyResult @@ -1502,7 +1691,7 @@ end SummaryEvent /** OpType describes the type of operation performed to a resource managed by Pulumi. Should generally mirror `deploy.StepOp` in the engine. */ -enum OpType(value: String): +enum OpType(val value: String): override def toString: String = value /** Indicates no change was made. */ @@ -1549,39 +1738,45 @@ enum OpType(value: String): /** Indicates replacement of an existing resource with an imported resource. */ case ImportReplacement extends OpType("import-replacement") + + /** An operation type emitted by the engine that this version of besom-auto does not know about. + * + * Kept so that a newer Pulumi CLI can not fail decoding of an otherwise valid event log - notably a single unknown key in + * [[SummaryEvent.resourceChanges]] would otherwise sink the whole summary event. + */ + case Other(unknownValue: String) extends OpType(unknownValue) end OpType object OpType: implicit object OpTypeFormat extends RootJsonFormat[OpType] { - def write(e: OpType): JsObject = JsObject( - "value" -> JsString(e.toString) - ) + // has to be a JsString - OpType is used as a Map key in SummaryEvent.resourceChanges and besom-json's + // mapFormat throws a SerializationException unless the key format writes one + def write(e: OpType): JsValue = JsString(e.value) def read(value: JsValue): OpType = { value match { - case JsString(value) => - OpType.from(value).left.map(DeserializationException("OpType expected", _)).fold(throw _, identity) - case n => throw DeserializationException(s"OpType expected, got: $n") + case JsString(value) => OpType.from(value) + case n => throw DeserializationException(s"OpType expected, got: $n") } } } - def from(value: String): Either[Exception, OpType] = + def from(value: String): OpType = value match - case "same" => Right(Same) - case "create" => Right(Create) - case "update" => Right(Update) - case "delete" => Right(Delete) - case "replace" => Right(Replace) - case "create-replacement" => Right(CreateReplacement) - case "delete-replaced" => Right(DeleteReplaced) - case "read" => Right(Read) - case "read-replacement" => Right(ReadReplacement) - case "refresh" => Right(Refresh) - case "discard" => Right(ReadDiscard) - case "discard-replaced" => Right(DiscardReplaced) - case "remove-pending-replace" => Right(RemovePendingReplace) - case "import" => Right(Import) - case "import-replacement" => Right(ImportReplacement) - case _ => Left(Exception(s"Unknown OpType: $value")) + case "same" => Same + case "create" => Create + case "update" => Update + case "delete" => Delete + case "replace" => Replace + case "create-replacement" => CreateReplacement + case "delete-replaced" => DeleteReplaced + case "read" => Read + case "read-replacement" => ReadReplacement + case "refresh" => Refresh + case "discard" => ReadDiscard + case "discard-replaced" => DiscardReplaced + case "remove-pending-replace" => RemovePendingReplace + case "import" => Import + case "import-replacement" => ImportReplacement + case other => Other(other) end from end OpType diff --git a/auto/src/main/scala/besom/auto/internal/shell.scala b/auto/src/main/scala/besom/auto/internal/shell.scala index cd908ada..af033b51 100644 --- a/auto/src/main/scala/besom/auto/internal/shell.scala +++ b/auto/src/main/scala/besom/auto/internal/shell.scala @@ -24,21 +24,42 @@ object shell: if res.exitCode == 0 then Right(res) else Left(res.asError) end Result + /** Runs a command to completion. + * + * This is a faithful transcription of os-lib's `os.proc(...).call(...)` - which is itself `spawn` + collect + `join` - with one + * addition: [[ShellOptions.onStart]] is handed a [[ChildProcess]] right after the subprocess is spawned, which is what makes + * cancellation possible. The handler runs on the calling thread before the process is joined, so it must not block - stash the handle + * and return. + */ def apply(command: os.Shellable*)(opts: ShellOption*): Either[ShellAutoError, Result] = val options = ShellOptions.from(opts*) - val result = os - .proc(command*) - .call( - cwd = options.cwd.asOption.orNull, - env = options.env, - stdin = options.stdin, - stdout = options.stdout, - stderr = options.stderr, - mergeErrIntoOut = options.mergeErrIntoOut, - timeout = options.timeout, - check = options.check, - propagateEnv = options.propagateEnv - ) + + val chunks = new java.util.concurrent.ConcurrentLinkedQueue[Either[geny.Bytes, geny.Bytes]] + + val p = os.proc(command*) + val sub = p.spawn( + cwd = options.cwd.asOption.orNull, + env = options.env, + stdin = options.stdin, + // a caller supplied ProcessOutput wins, exactly as in os-lib - we only collect when the stream is left at os.Pipe + stdout = + if options.stdout ne os.Pipe then options.stdout + else os.ProcessOutput.ReadBytes((buf, n) => chunks.add(Left(new geny.Bytes(java.util.Arrays.copyOf(buf, n))))), + stderr = + if options.stderr ne os.Pipe then options.stderr + else os.ProcessOutput.ReadBytes((buf, n) => chunks.add(Right(new geny.Bytes(java.util.Arrays.copyOf(buf, n))))), + mergeErrIntoOut = options.mergeErrIntoOut, + propagateEnv = options.propagateEnv + ) + + options.onStart.foreach(_(ChildProcess(sub))) + + sub.join(timeout = options.timeout, timeoutGracePeriod = 100) + + import scala.jdk.CollectionConverters.* + val result = os.CommandResult(p.commandChunks, sub.exitCode(), chunks.iterator.asScala.toIndexedSeq) + if result.exitCode != 0 && options.check then throw os.SubprocessException(result) + Result.from(result, options.env) end apply @@ -78,6 +99,14 @@ object shell: */ case object DontPropagateEnv extends ShellOption + /** A handler receiving a [[ChildProcess]] handle to the spawned subprocess, for callers that want to be able to cancel it. + * + * The handler runs on the calling thread right after the subprocess is spawned and before it is joined, so it must not block - stash + * the handle and return. + */ + case class OnStart(handler: ChildProcess => Unit) extends ShellOption + end ShellOption + /** Options for the subprocess execution. * @param cwd * the working directory of the subprocess @@ -97,6 +126,8 @@ object shell: * whether to check the subprocess exit code and throw an exception if it is non-zero * @param propagateEnv * whether to propagate the current environment variables to the subprocess + * @param onStart + * an optional handler receiving a handle to the spawned subprocess, invoked on the calling thread before the process is joined */ case class ShellOptions( cwd: NotProvidedOr[os.Path] = NotProvided, @@ -107,7 +138,8 @@ object shell: mergeErrIntoOut: Boolean = false, timeout: Long = -1, check: Boolean = false, // in contrast to os lib we default to false, because we use our own error handling - propagateEnv: Boolean = true + propagateEnv: Boolean = true, + onStart: Option[ChildProcess => Unit] = None ) object ShellOptions: @@ -122,6 +154,7 @@ object shell: case ShellOption.Timeout(timeout) :: tail => from(tail).copy(timeout = timeout) case ShellOption.Check :: tail => from(tail).copy(check = true) case ShellOption.DontPropagateEnv :: tail => from(tail).copy(propagateEnv = false) + case ShellOption.OnStart(handler) :: tail => from(tail).copy(onStart = Some(handler)) case ShellOption.Env(env) :: tail => { val old = from(tail*) old.copy(env = old.env ++ env) @@ -170,22 +203,20 @@ object shell: end pulumi - import ma.chinespirit.tailf.Tail - - // FIXME probably requires a redesign? - def tail(path: os.Path): Either[Exception, Iterator[String] & AutoCloseable] = - Tail - .follow(path.toIO) - .left - .map(e => Exception(s"Failed to open $path for tailing", e)) - .map { is => - new Iterator[String] with AutoCloseable: - val src = scala.io.Source.fromInputStream(is) - val it = src.getLines() - - def hasNext: Boolean = it.hasNext - def next(): String = it.next() - def close(): Unit = src.close() - } + import ma.chinespirit.tailf.{Follower, Tail} + + /** Opens a file for tailing, in the `tail -f` sense - the returned [[Follower]] is an `InputStream` that blocks at the end of the file + * instead of signalling EOF. + * + * The [[Follower]] itself is handed back rather than an iterator so that callers get `stop()`, which drains what is already written and + * only then signals EOF, in addition to the abrupt `close()`. + * + * @param path + * the file to tail, which must already exist + * @return + * the follower or an error if the file could not be opened + */ + def tail(path: os.Path): Either[Exception, Follower] = + Tail.follow(path.toIO).left.map(e => Exception(s"Failed to open $path for tailing", e)) end shell diff --git a/auto/src/test/scala/besom/auto/internal/EventLogsTest.scala b/auto/src/test/scala/besom/auto/internal/EventLogsTest.scala new file mode 100644 index 00000000..1f26cb48 --- /dev/null +++ b/auto/src/test/scala/besom/auto/internal/EventLogsTest.scala @@ -0,0 +1,161 @@ +package besom.auto.internal + +import java.util.concurrent.{CountDownLatch, TimeUnit} +import scala.collection.mutable + +class EventLogsTest extends munit.FunSuite: + + private def event(sequence: Int, urn: String = "urn:pulumi:dev::proj::pkg:index:Res::res"): String = + s"""{"sequence":$sequence,"timestamp":1700000000,"resourcePreEvent":{"metadata":{"op":"create","urn":"$urn","type":"pkg:index:Res","provider":"","old":null,"new":null,"detailedDiff":null}}}""" + + private val summaryLine = + """{"sequence":99,"timestamp":1700000000,"summaryEvent":{"maybeCorrupt":false,"durationSeconds":3,"resourceChanges":{"create":1},"PolicyPacks":{}}}""" + + // ── parse ───────────────────────────────────────────────────────────── + + test("parse collects undecodable lines instead of failing the whole log") { + val path = os.temp.dir() / "eventlog.txt" + os.write( + path, + Seq( + event(1), // 1: fine + "not json at all", // 2: garbage + event(2), // 3: fine + "", // 4: blank, skipped entirely + """{"sequence":3,"timestamp":""", // 5: truncated, as a killed process would leave it + summaryLine // 6: fine + ).mkString("\n") + ) + + val parsed = EventLogs.parse(path).getOrElse(fail("parse should not fail on undecodable lines")) + + assertEquals(parsed.events.map(_.sequence), List(1, 2, 99)) + assertEquals(parsed.parseErrors.map(_.lineNumber), List(2, 5)) + assertEquals(parsed.parseErrors.map(_.line), List("not json at all", """{"sequence":3,"timestamp":""")) + } + + test("parse fails only when the file cannot be read") { + val missing = os.temp.dir() / "nope.txt" + assert(EventLogs.parse(missing).isLeft) + } + + test("summary mentions the parse error count when the summary line went missing") { + val path = os.temp.dir() / "eventlog.txt" + os.write(path, Seq(event(1), "garbage").mkString("\n")) + + val parsed = EventLogs.parse(path).getOrElse(fail("parse failed")) + val err = EventLogs.summary(parsed.events, parsed.parseErrors).left.getOrElse(fail("expected a missing summary")) + + assert(err.getMessage.contains("1 event log line(s) failed to parse"), s"got: ${err.getMessage}") + // and stays quiet when there is nothing to blame + assert(!EventLogs.summary(parsed.events).left.getOrElse(fail("expected a missing summary")).getMessage.contains("failed to parse")) + } + + test("summary finds the summary event") { + val path = os.temp.dir() / "eventlog.txt" + os.write(path, Seq(event(1), summaryLine).mkString("\n")) + + val parsed = EventLogs.parse(path).getOrElse(fail("parse failed")) + val summary = EventLogs.summary(parsed.events, parsed.parseErrors).getOrElse(fail("expected a summary")) + assertEquals(summary.resourceChanges, Map[OpType, Int](OpType.Create -> 1)) + assertEquals(EventLogs.resourcePreEvents(parsed.events).size, 1) + } + + // ── around ──────────────────────────────────────────────────────────── + + test("around runs the body unchanged when there is no handler") { + val path = os.temp.dir() / "eventlog.txt" + os.write(path, "") + assertEquals(EventLogs.around(path, None)(Right("done")), Right("done")) + } + + test("around fails without running the body when the log cannot be opened") { + var ran = false + val res = EventLogs.around(os.temp.dir() / "missing.txt", Some(_ => ())) { + ran = true + Right(()) + } + assert(res.isLeft, s"expected a Left, got: $res") + assert(!ran, "the body must not run when the follower could not be opened") + } + + test("around delivers events live and drains what was written just before the body returned") { + val path = os.temp.dir() / "eventlog.txt" + os.write(path, "") + + val received = mutable.ListBuffer.empty[Int] + val sawFirst = new CountDownLatch(1) + val sawSecond = new CountDownLatch(1) + + val result = EventLogs.around( + path, + Some { e => + received.synchronized(received += e.sequence) + sawFirst.countDown() + if received.synchronized(received.size) >= 2 then sawSecond.countDown() + } + ) { + os.write.append(path, event(1) + "\n") + // liveness: this must arrive while the body is still running + assert(sawFirst.await(10, TimeUnit.SECONDS), "the first event was not delivered before the body returned") + + os.write.append(path, event(2) + "\n") + assert(sawSecond.await(10, TimeUnit.SECONDS), "the second event was not delivered before the body returned") + + // written last thing before returning - stop() must drain it rather than lose it + os.write.append(path, summaryLine + "\n") + Right("done") + } + + assertEquals(result, Right("done")) + assertEquals(received.synchronized(received.toList), List(1, 2, 99)) + } + + test("around keeps delivering after a handler throws") { + val path = os.temp.dir() / "eventlog.txt" + os.write(path, "") + + val received = mutable.ListBuffer.empty[Int] + val done = new CountDownLatch(1) + + val result = EventLogs.around( + path, + Some { e => + if e.sequence == 1 then throw new RuntimeException("boom") + received.synchronized(received += e.sequence) + if e.sequence == 99 then done.countDown() + } + ) { + os.write.append(path, Seq(event(1), event(2), summaryLine).mkString("", "\n", "\n")) + assert(done.await(10, TimeUnit.SECONDS), "delivery stopped after the handler threw") + Right(()) + } + + assertEquals(result, Right(())) + assertEquals(received.synchronized(received.toList), List(2, 99)) + } + + test("around drops undecodable lines rather than failing the operation") { + val path = os.temp.dir() / "eventlog.txt" + os.write(path, "") + + val received = mutable.ListBuffer.empty[Int] + val done = new CountDownLatch(1) + + val result = EventLogs.around( + path, + Some { e => + received.synchronized(received += e.sequence) + if e.sequence == 99 then done.countDown() + } + ) { + os.write.append(path, Seq(event(1), "garbage", summaryLine).mkString("", "\n", "\n")) + assert(done.await(10, TimeUnit.SECONDS), "delivery stopped at the undecodable line") + Right(()) + } + + assertEquals(result, Right(())) + assertEquals(received.synchronized(received.toList), List(1, 99)) + } + +end EventLogsTest diff --git a/auto/src/test/scala/besom/auto/internal/LocalWorkspaceTest.scala b/auto/src/test/scala/besom/auto/internal/LocalWorkspaceTest.scala index 49c82b81..a7bdf85f 100644 --- a/auto/src/test/scala/besom/auto/internal/LocalWorkspaceTest.scala +++ b/auto/src/test/scala/besom/auto/internal/LocalWorkspaceTest.scala @@ -4,8 +4,34 @@ import besom.model.FullyQualifiedStackName import besom.test.* import besom.util.eitherOps +import scala.concurrent.duration.* + class LocalWorkspaceTest extends munit.FunSuite: + // these shell out to git, go and pulumi and download plugins, which does not fit in munit's 30s default + override def munitTimeout: Duration = 5.minutes + + /** Collects the engine events delivered by `OnEvent`, keeping enough timing to tell live delivery from a post-hoc dump. */ + private class EventRecorder: + private val events = scala.collection.mutable.ListBuffer.empty[EngineEvent] + @volatile private var firstEventAt: Long = -1L + @volatile private var callReturnedAt: Long = -1L + + val record: EngineEvent => Unit = e => + events.synchronized { + if firstEventAt < 0 then firstEventAt = System.nanoTime() + events += e + } + + def markCallReturned(): Unit = callReturnedAt = System.nanoTime() + def all: List[EngineEvent] = events.synchronized(events.toList) + def preEvents: List[ResourcePreEvent] = all.flatMap(_.resourcePreEvent) + def firstArrivedBeforeReturn: Boolean = firstEventAt > 0 && callReturnedAt > 0 && firstEventAt < callReturnedAt + end EventRecorder + + /** The explicit parent URN of a step, empty for the root stack resource. */ + private def parentOf(meta: StepEventMetadata): Option[String] = meta.`new`.orElse(meta.old).map(_.parent) + FunFixture[FullyQualifiedStackName]( setup = t => fqsn(this.getClass, t), teardown = _ => () @@ -23,6 +49,9 @@ class LocalWorkspaceTest extends munit.FunSuite: ) } + val upEvents = EventRecorder() + val upAgainEvents = EventRecorder() + val res = for stack <- createStackRemoteSource( stackName, @@ -43,13 +72,17 @@ class LocalWorkspaceTest extends munit.FunSuite: LocalWorkspaceOption.PulumiHome(pulumiHomeDir), LocalWorkspaceOption.EnvVars(shell.pulumi.env.PulumiConfigPassphraseEnv -> "test") ) - prevRes <- stack.preview() - upRes <- stack.up() + prevRes <- stack.preview() + upRes <- stack.up(UpOption.OnEvent(upEvents.record)) + _ = upEvents.markCallReturned() + // a second, no-op up is the only way to observe `same` steps, which the "full tree" premise depends on + upAgainRes <- stack.up(UpOption.OnEvent(upAgainEvents.record)) + _ = upAgainEvents.markCallReturned() destroyRes <- stack.destroy() - yield (prevRes, upRes, destroyRes) + yield (prevRes, upRes, upAgainRes, destroyRes) res.fold( e => fail(e.getMessage, e), - (prevRes, upRes, destroyRes) => { + (prevRes, upRes, upAgainRes, destroyRes) => { // Preview: summary (backward compat) assertEquals(prevRes.summary, Map(OpType.Create -> 1)) @@ -87,13 +120,153 @@ class LocalWorkspaceTest extends munit.FunSuite: assert(upRes.resourceOperations.nonEmpty, s"Expected non-empty resourceOperations, got: ${upRes.resourceOperations}") assert(upRes.failures.isEmpty, s"Expected no failures, got: ${upRes.failures}") + // Up: resourcePreEvents restores the *start* of each operation + assert(upRes.resourcePreEvents.nonEmpty, s"Expected non-empty resourcePreEvents, got: ${upRes.resourcePreEvents}") + assertEquals(upRes.parseErrors, Nil) + // Destroy: should succeed without failures assertEquals(destroyRes.summary.result, Some("succeeded")) assert(destroyRes.failures.isEmpty, s"Expected no failures on destroy, got: ${destroyRes.failures}") assert(destroyRes.diagnostics != null) + assert(destroyRes.resourcePreEvents.nonEmpty, s"Expected non-empty resourcePreEvents on destroy") + assertEquals(destroyRes.parseErrors, Nil) + + // ── live event delivery ───────────────────────────────────────── + val live = upEvents.all + assert(live.nonEmpty, "no engine events were delivered live") + + // (a) liveness - the first event has to arrive while up is still running + assert(upEvents.firstArrivedBeforeReturn, "the first engine event did not arrive before up returned") + + // the live stream must agree with the post-hoc parse, that is what makes it a drop-in replacement + assertEquals(live.flatMap(_.resOutputsEvent).map(_.metadata.urn), upRes.resourceOperations.map(_.metadata.urn)) + assertEquals(upEvents.preEvents.map(_.metadata.urn), upRes.resourcePreEvents.map(_.metadata.urn)) + + // (b) `same` steps are present in the log - it precedes Pulumi's display filter + val sameSteps = upAgainEvents.preEvents.filter(_.metadata.op == OpType.Same) + assertEquals(upAgainRes.summary.result, Some("succeeded")) + assert( + sameSteps.nonEmpty, + s"Expected `same` steps in a no-op up, got ops: ${upAgainEvents.preEvents.map(_.metadata.op)}" + ) + + // (c) the root stack resource carries an explicit, empty parent + // (this program has no other resources - the nested case is covered by the resource tree test below) + val roots = upEvents.preEvents.filter(_.metadata.`type` == "pulumi:pulumi:Stack") + assert(roots.nonEmpty, "no pulumi:pulumi:Stack ResourcePreEvent") + assertEquals(roots.flatMap(e => parentOf(e.metadata)).distinct, List("")) + } + ) + } + FunFixture[FullyQualifiedStackName]( + setup = t => fqsn(this.getClass, t), + teardown = _ => () + ).test("engine events describe the full resource tree") { generatedStackName => + val stackName = FullyQualifiedStackName("goproj", generatedStackName.stack) + val pulumiHomeDir = os.temp.dir() / ".pulumi" + loginLocal(pulumiHomeDir) + + // component resources need no provider plugin, so a three level tree costs nothing beyond the SDK the fixture already vendors + val program = + """|package main + | + |import "github.com/pulumi/pulumi/sdk/v3/go/pulumi" + | + |type Group struct{ pulumi.ResourceState } + | + |func main() { + | pulumi.Run(func(ctx *pulumi.Context) error { + | parent := &Group{} + | if err := ctx.RegisterComponentResource("besom:test:Group", "parent", parent); err != nil { + | return err + | } + | child := &Group{} + | if err := ctx.RegisterComponentResource("besom:test:Group", "child", child, pulumi.Parent(parent)); err != nil { + | return err + | } + | grandchild := &Group{} + | if err := ctx.RegisterComponentResource("besom:test:Group", "grandchild", grandchild, pulumi.Parent(child)); err != nil { + | return err + | } + | return nil + | }) + |} + |""".stripMargin + + val binName = "treeBinary" + val bin = if System.getProperty("os.name").startsWith("Windows") then binName + ".exe" else binName + val binaryBuilder = (ws: Workspace) => { + os.write.over(ws.workDir / "main.go", program) + shell("go", "build", "-o", bin, "main.go")(shell.ShellOption.Cwd(ws.workDir)).bimap( + e => e.withMessage("go build failed"), + _ => () + ) + } + + val upEvents = EventRecorder() + + val res = for + stack <- createStackRemoteSource( + stackName, + GitRepo( + url = "https://github.com/pulumi/test-repo.git", + projectPath = "goproj", + setup = binaryBuilder + ), + LocalWorkspaceOption.Project( + Project( + name = "goproj", + runtime = ProjectRuntimeInfo(name = "go", options = Map("binary" -> binName)) + ) + ), + LocalWorkspaceOption.PulumiHome(pulumiHomeDir), + LocalWorkspaceOption.EnvVars(shell.pulumi.env.PulumiConfigPassphraseEnv -> "test") + ) + upRes <- stack.up(UpOption.OnEvent(upEvents.record)) + _ = upEvents.markCallReturned() + _ <- stack.destroy() + yield upRes + + res.fold( + e => fail(e.getMessage, e), + upRes => { + assertEquals(upRes.summary.result, Some("succeeded")) + assertEquals(upRes.parseErrors, Nil) + assert(upEvents.firstArrivedBeforeReturn, "the first engine event did not arrive before up returned") + + val pre = upEvents.preEvents + assertEquals(pre.map(_.metadata.urn), upRes.resourcePreEvents.map(_.metadata.urn)) + + def urnOf(name: String) = pre.map(_.metadata.urn).find(_.endsWith(s"::$name")).getOrElse(fail(s"no pre event for $name")) + val (parentUrn, childUrn, grandchildUrn) = (urnOf("parent"), urnOf("child"), urnOf("grandchild")) + val byUrn = pre.map(e => e.metadata.urn -> e.metadata).toMap + + // (c) the explicit parent URN is populated - empty for the root stack, set for everything nested under it + val (roots, nested) = pre.partition(_.metadata.`type` == "pulumi:pulumi:Stack") + assert(roots.nonEmpty, "no pulumi:pulumi:Stack ResourcePreEvent") + assertEquals(roots.flatMap(e => parentOf(e.metadata)).distinct, List("")) + val rootUrn = roots.head.metadata.urn + + assertEquals(nested.size, 3, s"expected the three components, got: ${nested.map(_.metadata.urn)}") + assert( + nested.forall(e => parentOf(e.metadata).exists(_.nonEmpty)), + s"Expected every nested resource to carry a parent, got: ${nested.map(e => e.metadata.urn -> parentOf(e.metadata))}" + ) + // and the parent URNs reconstruct exactly the tree the program declared + assertEquals(parentOf(byUrn(parentUrn)), Some(rootUrn)) + assertEquals(parentOf(byUrn(childUrn)), Some(parentUrn)) + assertEquals(parentOf(byUrn(grandchildUrn)), Some(childUrn)) + + // (d) ordering - a parent's ResourcePreEvent always precedes its children's, so a renderer needs no orphan buffering + val order = pre.map(_.metadata.urn).zipWithIndex.toMap + val orphans = pre.filter { e => + parentOf(e.metadata).filter(_.nonEmpty).exists(parent => !order.get(parent).exists(_ < order(e.metadata.urn))) + } + assertEquals(orphans.map(_.metadata.urn), Nil, "a child ResourcePreEvent preceded its parent's") } ) } + FunFixture[FullyQualifiedStackName]( setup = t => fqsn(this.getClass, t), teardown = _ => () diff --git a/auto/src/test/scala/besom/auto/internal/OptionsTest.scala b/auto/src/test/scala/besom/auto/internal/OptionsTest.scala new file mode 100644 index 00000000..4870a07c --- /dev/null +++ b/auto/src/test/scala/besom/auto/internal/OptionsTest.scala @@ -0,0 +1,84 @@ +package besom.auto.internal + +class OptionsTest extends munit.FunSuite: + + private val onEvent: EngineEvent => Unit = _ => () + private val onStart: ChildProcess => Unit = _ => () + private val logPath = os.pwd / "eventlog.txt" + + test("PreviewOptions.from event and process options") { + val opts = PreviewOptions.from( + PreviewOption.OnEvent(onEvent), + PreviewOption.EventLog(logPath), + PreviewOption.OnProcessStart(onStart) + ) + assertEquals(opts.onEvent, Some(onEvent)) + assertEquals(opts.eventLog, logPath) + assertEquals(opts.onProcessStart, Some(onStart)) + + val defaults = PreviewOptions.from() + assertEquals(defaults.onEvent, None) + assertEquals(defaults.eventLog.asOption, None) + assertEquals(defaults.onProcessStart, None) + } + + test("UpOptions.from event and process options") { + val opts = UpOptions.from( + UpOption.OnEvent(onEvent), + UpOption.EventLog(logPath), + UpOption.OnProcessStart(onStart) + ) + assertEquals(opts.onEvent, Some(onEvent)) + assertEquals(opts.eventLog, logPath) + assertEquals(opts.onProcessStart, Some(onStart)) + + val defaults = UpOptions.from() + assertEquals(defaults.onEvent, None) + assertEquals(defaults.eventLog.asOption, None) + assertEquals(defaults.onProcessStart, None) + } + + test("RefreshOptions.from event and process options") { + val opts = RefreshOptions.from( + RefreshOption.OnEvent(onEvent), + RefreshOption.EventLog(logPath), + RefreshOption.OnProcessStart(onStart) + ) + assertEquals(opts.onEvent, Some(onEvent)) + assertEquals(opts.eventLog, logPath) + assertEquals(opts.onProcessStart, Some(onStart)) + + val defaults = RefreshOptions.from() + assertEquals(defaults.onEvent, None) + assertEquals(defaults.eventLog.asOption, None) + assertEquals(defaults.onProcessStart, None) + } + + test("DestroyOptions.from event and process options") { + val opts = DestroyOptions.from( + DestroyOption.OnEvent(onEvent), + DestroyOption.EventLog(logPath), + DestroyOption.OnProcessStart(onStart) + ) + assertEquals(opts.onEvent, Some(onEvent)) + assertEquals(opts.eventLog, logPath) + assertEquals(opts.onProcessStart, Some(onStart)) + + val defaults = DestroyOptions.from() + assertEquals(defaults.onEvent, None) + assertEquals(defaults.eventLog.asOption, None) + assertEquals(defaults.onProcessStart, None) + } + + test("repeated options resolve the same way as the pre-existing ones") { + val other: EngineEvent => Unit = _ => () + val opts = UpOptions.from( + UpOption.OnEvent(onEvent), + UpOption.OnEvent(other), + UpOption.Message("first"), + UpOption.Message("second") + ) + assertEquals(opts.onEvent, if opts.message == "first" then Some(onEvent) else Some(other)) + } + +end OptionsTest diff --git a/auto/src/test/scala/besom/auto/internal/ShellTest.scala b/auto/src/test/scala/besom/auto/internal/ShellTest.scala index 24a0dc59..a0fbf023 100644 --- a/auto/src/test/scala/besom/auto/internal/ShellTest.scala +++ b/auto/src/test/scala/besom/auto/internal/ShellTest.scala @@ -26,4 +26,82 @@ class ShellTest extends munit.FunSuite: assertEquals(options.mergeErrIntoOut, true) assertEquals(options.check, true) assertEquals(options.propagateEnv, false) + assertEquals(options.onStart, None) } + + test("ShellOptions.from OnStart") { + val handler: ChildProcess => Unit = _ => () + assertEquals(ShellOptions.from(ShellOption.OnStart(handler)).onStart, Some(handler)) + } + + test("out/err are collected separately") { + val res = shell("sh", "-c", "echo out; echo err 1>&2")().getOrElse(fail("command failed")) + assertEquals(res.exitCode, 0) + assertEquals(res.out.trim, "out") + assertEquals(res.err.trim, "err") + } + + test("mergeErrIntoOut folds stderr into stdout") { + val res = shell("sh", "-c", "echo err 1>&2")(ShellOption.MergeErrIntoOut).getOrElse(fail("command failed")) + assertEquals(res.out.trim, "err") + assertEquals(res.err.trim, "") + } + + test("a caller supplied stdout wins over collection") { + val lines = scala.collection.mutable.ListBuffer.empty[String] + val res = shell("sh", "-c", "echo out")( + ShellOption.Stdout(os.ProcessOutput.Readlines(l => lines.synchronized(lines += l))) + ).getOrElse(fail("command failed")) + + assertEquals(lines.synchronized(lines.toList), List("out")) + assertEquals(res.out, "") // nothing was collected, the caller took it + } + + test("a non-zero exit code is reported as a ShellAutoError") { + val err = shell("sh", "-c", "echo boom 1>&2; exit 3")().left.getOrElse(fail("expected a failure")) + assertEquals(err.exitCode, 3) + assertEquals(err.stderr.trim, "boom") + } + + test("Check throws instead of returning a Left") { + intercept[os.SubprocessException] { + shell("sh", "-c", "exit 3")(ShellOption.Check) + } + } + + test("OnStart hands over a live process handle") { + @volatile var handle: Option[ChildProcess] = None + + val res = shell("sh", "-c", "echo hello")(ShellOption.OnStart(p => handle = Some(p))) + .getOrElse(fail("command failed")) + + assertEquals(res.out.trim, "hello") + val p = handle.getOrElse(fail("OnStart was never invoked")) + assert(p.pid > 0, s"expected a real pid, got ${p.pid}") + assert(!p.isAlive, "the process should be done by the time the call returned") + } + + test("OnStart handle can interrupt a running process") { + @volatile var handle: Option[ChildProcess] = None + val started = new java.util.concurrent.CountDownLatch(1) + + val runner = new Thread(() => + shell("sh", "-c", "sleep 30")(ShellOption.OnStart { p => + handle = Some(p) + started.countDown() + }) + () + ) + runner.setDaemon(true) + runner.start() + + assert(started.await(10, java.util.concurrent.TimeUnit.SECONDS), "the process never started") + val p = handle.getOrElse(fail("OnStart was never invoked")) + assert(p.isAlive, "the process should still be running") + + p.interrupt() + runner.join(10000) + assert(!runner.isAlive, "the call did not return after the process was interrupted") + assert(!p.isAlive, "the process should be gone") + } +end ShellTest diff --git a/auto/src/test/scala/besom/auto/internal/StackTest.scala b/auto/src/test/scala/besom/auto/internal/StackTest.scala index ce85e5c9..d156b595 100644 --- a/auto/src/test/scala/besom/auto/internal/StackTest.scala +++ b/auto/src/test/scala/besom/auto/internal/StackTest.scala @@ -381,4 +381,49 @@ class EngineEventJSONTest extends munit.FunSuite { assertEquals(ProgressType.from("plugin-download"), ProgressType.PluginDownload) assertEquals(ProgressType.from("plugin-install"), ProgressType.PluginInstall) } + + // ── Forward compatibility ─────────────────────────────────────────── + + test("unknown DiffKind decodes to Other instead of failing") { + assertEquals(DiffKind.from("teleport"), DiffKind.Other("teleport")) + assertEquals(JsString("teleport").convertTo[DiffKind], DiffKind.Other("teleport")) + // the value is preserved, so the -replace convention keeps working for kinds we have never seen + assert(DiffKind.Other("teleport-replace").forcesReplacement) + assert(!DiffKind.Other("teleport").forcesReplacement) + + val json = + """{"op":"create","urn":"urn:pulumi:dev::proj::pulumi:pulumi:Stack::proj-dev","type":"pulumi:pulumi:Stack","provider":"","old":null,"new":null,"detailedDiff":{"foo":{"diffKind":"teleport","inputDiff":true}}}""" + val meta = json.parseJson[StepEventMetadata].getOrElse(fail("Failed to parse")) + assertEquals(meta.detailedDiff.flatMap(_.get("foo")).map(_.diffKind), Some(DiffKind.Other("teleport"))) + } + + test("unknown ProgressType decodes to Other instead of failing") { + assertEquals(ProgressType.from("plugin-teleport"), ProgressType.Other("plugin-teleport")) + + val json = + """{"sequence":7,"timestamp":1700000000,"progressEvent":{"type":"plugin-teleport","id":"aws","message":"beaming","received":1,"total":2,"done":false}}""" + val event = EngineEvent.fromJson(json).getOrElse(fail("Failed to parse")) + assertEquals(event.progressEvent.map(_.`type`), Some(ProgressType.Other("plugin-teleport"))) + } + + test("unknown OpType decodes to Other instead of failing") { + assertEquals(OpType.from("teleport"), OpType.Other("teleport")) + assertEquals(JsString("teleport").convertTo[OpType], OpType.Other("teleport")) + assertEquals(OpType.Other("teleport").value, "teleport") + } + + test("OpType round-trips as a Map key") { + // besom-json's mapFormat requires the key format to write a JsString, otherwise SummaryEvent serialization blows up + val changes: Map[OpType, Int] = Map(OpType.Create -> 1, OpType.Same -> 2, OpType.Other("teleport") -> 3) + assertEquals(changes.toJson, JsObject("create" -> JsNumber(1), "same" -> JsNumber(2), "teleport" -> JsNumber(3))) + assertEquals(changes.toJson.convertTo[Map[OpType, Int]], changes) + } + + test("SummaryEvent with an unknown op in resourceChanges still parses") { + val json = + """{"sequence":42,"timestamp":1700000000,"summaryEvent":{"maybeCorrupt":false,"durationSeconds":3,"resourceChanges":{"create":1,"teleport":2},"PolicyPacks":{}}}""" + val event = EngineEvent.fromJson(json).getOrElse(fail("Failed to parse")) + val summary = event.summaryEvent.getOrElse(fail("Expected a summary event")) + assertEquals(summary.resourceChanges, Map[OpType, Int](OpType.Create -> 1, OpType.Other("teleport") -> 2)) + } } From 72ac3e4eb9e998e8f6bb71d0a6f4e07ffd103e78 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Bia=C5=82y?= Date: Fri, 28 Aug 2026 16:12:28 +0200 Subject: [PATCH 2/3] post-review fixes --- .../scala/besom/auto/internal/EventLogs.scala | 12 +- .../scala/besom/auto/internal/Stack.scala | 12 +- .../scala/besom/auto/internal/shell.scala | 13 ++- .../besom/auto/internal/EventLogsTest.scala | 24 ++++ .../auto/internal/LocalWorkspaceTest.scala | 104 ++++++++++++++++-- .../scala/besom/auto/internal/ShellTest.scala | 14 +++ 6 files changed, 159 insertions(+), 20 deletions(-) diff --git a/auto/src/main/scala/besom/auto/internal/EventLogs.scala b/auto/src/main/scala/besom/auto/internal/EventLogs.scala index 52ad20b0..f46ab3af 100644 --- a/auto/src/main/scala/besom/auto/internal/EventLogs.scala +++ b/auto/src/main/scala/besom/auto/internal/EventLogs.scala @@ -90,6 +90,10 @@ private[auto] object EventLogs: * Undecodable lines are dropped here - [[parse]] accounts for them post-hoc in `parseErrors`. Exceptions thrown by `handler` are * swallowed so that a single misbehaving consumer cannot end the stream. * + * Every event is delivered before this returns. `stop()` takes effect at EOF and the reader re-checks it after each `rereadSleep` + * (100ms), so the wait is short - but it is unbounded, which means a `handler` that never returns blocks the operation from returning. + * That is the reason handlers must not block. + * * Caveat: `tailf`'s follower treats a file that got shorter than the current read position as rotated and restarts from offset 0. * Nothing truncates a Pulumi event log mid-run, but it is the only path that could replay events, so consumers should stay idempotent * per URN. @@ -133,13 +137,11 @@ private[auto] object EventLogs: try body finally - try - follower.stop() // drain what is already written, then EOF - reader.join(5000) // a reader parked in waitNewInput wakes within one rereadSleep - catch case NonFatal(_) => () + follower.stop() // takes effect at EOF, so the reader drains what is already written and then ends + try reader.join() // no deadline: cutting the reader short here would silently drop events it still holds finally try follower.close() - catch case NonFatal(_) => () + catch case NonFatal(_) => () // releasing the fd must not mask the operation's own outcome } end around diff --git a/auto/src/main/scala/besom/auto/internal/Stack.scala b/auto/src/main/scala/besom/auto/internal/Stack.scala index 10be74db..b166c0d7 100644 --- a/auto/src/main/scala/besom/auto/internal/Stack.scala +++ b/auto/src/main/scala/besom/auto/internal/Stack.scala @@ -720,7 +720,8 @@ object PreviewOption: * [[ChildProcess.interrupt]] for the graceful, Ctrl-C equivalent. * * The handler runs on the calling thread right after the process is spawned and before it is waited on, so it must not block - stash the - * handle and return. + * handle and return. Exceptions it throws are swallowed: the operation still runs to completion, the caller simply ends up without a + * handle. */ case class OnProcessStart(handler: ChildProcess => Unit) extends PreviewOption @@ -909,7 +910,8 @@ object UpOption: * [[ChildProcess.interrupt]] for the graceful, Ctrl-C equivalent. * * The handler runs on the calling thread right after the process is spawned and before it is waited on, so it must not block - stash the - * handle and return. + * handle and return. Exceptions it throws are swallowed: the operation still runs to completion, the caller simply ends up without a + * handle. */ case class OnProcessStart(handler: ChildProcess => Unit) extends UpOption @@ -1085,7 +1087,8 @@ object RefreshOption: * [[ChildProcess.interrupt]] for the graceful, Ctrl-C equivalent. * * The handler runs on the calling thread right after the process is spawned and before it is waited on, so it must not block - stash the - * handle and return. + * handle and return. Exceptions it throws are swallowed: the operation still runs to completion, the caller simply ends up without a + * handle. */ case class OnProcessStart(handler: ChildProcess => Unit) extends RefreshOption @@ -1228,7 +1231,8 @@ object DestroyOption: * [[ChildProcess.interrupt]] for the graceful, Ctrl-C equivalent. * * The handler runs on the calling thread right after the process is spawned and before it is waited on, so it must not block - stash the - * handle and return. + * handle and return. Exceptions it throws are swallowed: the operation still runs to completion, the caller simply ends up without a + * handle. */ case class OnProcessStart(handler: ChildProcess => Unit) extends DestroyOption diff --git a/auto/src/main/scala/besom/auto/internal/shell.scala b/auto/src/main/scala/besom/auto/internal/shell.scala index af033b51..5d5268fe 100644 --- a/auto/src/main/scala/besom/auto/internal/shell.scala +++ b/auto/src/main/scala/besom/auto/internal/shell.scala @@ -2,6 +2,8 @@ package besom.auto.internal import besom.util.* +import scala.util.control.NonFatal + object shell: case class Result private ( command: Seq[String], @@ -29,7 +31,7 @@ object shell: * This is a faithful transcription of os-lib's `os.proc(...).call(...)` - which is itself `spawn` + collect + `join` - with one * addition: [[ShellOptions.onStart]] is handed a [[ChildProcess]] right after the subprocess is spawned, which is what makes * cancellation possible. The handler runs on the calling thread before the process is joined, so it must not block - stash the handle - * and return. + * and return. Exceptions it throws are swallowed; the command still runs to completion. */ def apply(command: os.Shellable*)(opts: ShellOption*): Either[ShellAutoError, Result] = val options = ShellOptions.from(opts*) @@ -52,7 +54,11 @@ object shell: propagateEnv = options.propagateEnv ) - options.onStart.foreach(_(ChildProcess(sub))) + // a broken consumer must neither leak the process it was handed nor kill it - skipping the join below would leave a live + // `pulumi up` running until JVM exit, and destroying it would leave the stack locked with a pending operation. Same contract + // as the OnEvent handlers: a consumer's exception cannot break the command it is observing. + try options.onStart.foreach(_(ChildProcess(sub))) + catch case NonFatal(_) => () sub.join(timeout = options.timeout, timeoutGracePeriod = 100) @@ -102,7 +108,8 @@ object shell: /** A handler receiving a [[ChildProcess]] handle to the spawned subprocess, for callers that want to be able to cancel it. * * The handler runs on the calling thread right after the subprocess is spawned and before it is joined, so it must not block - stash - * the handle and return. + * the handle and return. Exceptions it throws are swallowed: the command it was handed still runs to completion, the caller simply + * ends up without a handle. */ case class OnStart(handler: ChildProcess => Unit) extends ShellOption end ShellOption diff --git a/auto/src/test/scala/besom/auto/internal/EventLogsTest.scala b/auto/src/test/scala/besom/auto/internal/EventLogsTest.scala index 1f26cb48..1cadb914 100644 --- a/auto/src/test/scala/besom/auto/internal/EventLogsTest.scala +++ b/auto/src/test/scala/besom/auto/internal/EventLogsTest.scala @@ -135,6 +135,30 @@ class EventLogsTest extends munit.FunSuite: assertEquals(received.synchronized(received.toList), List(2, 99)) } + test("around drains a slow handler completely before returning") { + val path = os.temp.dir() / "eventlog.txt" + os.write(path, "") + + val received = mutable.ListBuffer.empty[Int] + val total = 12 + + // the handler is slow enough that it is certainly still draining when the body returns and stop() is called, + // so returning early would lose events - the wait after stop() is deliberately unbounded + val result = EventLogs.around( + path, + Some { e => + Thread.sleep(150) + received.synchronized(received += e.sequence) + } + ) { + os.write.append(path, (1 to total).map(event(_)).mkString("", "\n", "\n")) + Right(()) + } + + assertEquals(result, Right(())) + assertEquals(received.synchronized(received.toList), (1 to total).toList) + } + test("around drops undecodable lines rather than failing the operation") { val path = os.temp.dir() / "eventlog.txt" os.write(path, "") diff --git a/auto/src/test/scala/besom/auto/internal/LocalWorkspaceTest.scala b/auto/src/test/scala/besom/auto/internal/LocalWorkspaceTest.scala index a7bdf85f..c6c0f726 100644 --- a/auto/src/test/scala/besom/auto/internal/LocalWorkspaceTest.scala +++ b/auto/src/test/scala/besom/auto/internal/LocalWorkspaceTest.scala @@ -23,7 +23,11 @@ class LocalWorkspaceTest extends munit.FunSuite: events += e } - def markCallReturned(): Unit = callReturnedAt = System.nanoTime() + /** Runs `call`, stamping the moment it returned - on the failure path too, so liveness stays checkable there. */ + def timing[A](call: => A): A = + try call + finally callReturnedAt = System.nanoTime() + def all: List[EngineEvent] = events.synchronized(events.toList) def preEvents: List[ResourcePreEvent] = all.flatMap(_.resourcePreEvent) def firstArrivedBeforeReturn: Boolean = firstEventAt > 0 && callReturnedAt > 0 && firstEventAt < callReturnedAt @@ -73,11 +77,9 @@ class LocalWorkspaceTest extends munit.FunSuite: LocalWorkspaceOption.EnvVars(shell.pulumi.env.PulumiConfigPassphraseEnv -> "test") ) prevRes <- stack.preview() - upRes <- stack.up(UpOption.OnEvent(upEvents.record)) - _ = upEvents.markCallReturned() + upRes <- upEvents.timing(stack.up(UpOption.OnEvent(upEvents.record))) // a second, no-op up is the only way to observe `same` steps, which the "full tree" premise depends on - upAgainRes <- stack.up(UpOption.OnEvent(upAgainEvents.record)) - _ = upAgainEvents.markCallReturned() + upAgainRes <- upAgainEvents.timing(stack.up(UpOption.OnEvent(upAgainEvents.record))) destroyRes <- stack.destroy() yield (prevRes, upRes, upAgainRes, destroyRes) res.fold( @@ -222,9 +224,8 @@ class LocalWorkspaceTest extends munit.FunSuite: LocalWorkspaceOption.PulumiHome(pulumiHomeDir), LocalWorkspaceOption.EnvVars(shell.pulumi.env.PulumiConfigPassphraseEnv -> "test") ) - upRes <- stack.up(UpOption.OnEvent(upEvents.record)) - _ = upEvents.markCallReturned() - _ <- stack.destroy() + upRes <- upEvents.timing(stack.up(UpOption.OnEvent(upEvents.record))) + _ <- stack.destroy() yield upRes res.fold( @@ -267,6 +268,93 @@ class LocalWorkspaceTest extends munit.FunSuite: ) } + FunFixture[FullyQualifiedStackName]( + setup = t => fqsn(this.getClass, t), + teardown = _ => () + ).test("a failed up reports OperationFailedError carrying the engine events") { generatedStackName => + val stackName = FullyQualifiedStackName("goproj", generatedStackName.stack) + val pulumiHomeDir = os.temp.dir() / ".pulumi" + loginLocal(pulumiHomeDir) + + // registers one component, then fails the program - enough to produce pre-events and a diagnostic before the non-zero exit + val program = + """|package main + | + |import ( + | "fmt" + | + | "github.com/pulumi/pulumi/sdk/v3/go/pulumi" + |) + | + |type Group struct{ pulumi.ResourceState } + | + |func main() { + | pulumi.Run(func(ctx *pulumi.Context) error { + | group := &Group{} + | if err := ctx.RegisterComponentResource("besom:test:Group", "doomed", group); err != nil { + | return err + | } + | return fmt.Errorf("deliberate failure from the test program") + | }) + |} + |""".stripMargin + + val binName = "failingBinary" + val binaryBuilder = (ws: Workspace) => { + os.write.over(ws.workDir / "main.go", program) + shell("go", "build", "-o", binName, "main.go")(shell.ShellOption.Cwd(ws.workDir)).bimap( + e => e.withMessage("go build failed"), + _ => () + ) + } + + val upEvents = EventRecorder() + + val res = for + stack <- createStackRemoteSource( + stackName, + GitRepo(url = "https://github.com/pulumi/test-repo.git", projectPath = "goproj", setup = binaryBuilder), + LocalWorkspaceOption.Project( + Project(name = "goproj", runtime = ProjectRuntimeInfo(name = "go", options = Map("binary" -> binName))) + ), + LocalWorkspaceOption.PulumiHome(pulumiHomeDir), + LocalWorkspaceOption.EnvVars(shell.pulumi.env.PulumiConfigPassphraseEnv -> "test") + ) + upRes <- upEvents.timing(stack.up(UpOption.OnEvent(upEvents.record), UpOption.Color(Color.Never))) + yield upRes + + res match + case Right(up) => fail(s"expected the up to fail, got: ${up.summary.result}") + case Left(e: OperationFailedError) => + assertEquals(e.operation, "up") + assertNotEquals(e.exitCode, 0) + assert(e.getMessage.startsWith("Up failed"), s"unexpected message: ${e.getMessage}") + + // the bulky dump stays on the ShellAutoError cause + assert(e.cause.exists(_.isInstanceOf[ShellAutoError]), s"expected a ShellAutoError cause, got: ${e.cause}") + + // ── the point of the type: the log is parsed on the failure path, so this data is not lost ── + assert(e.resourcePreEvents.nonEmpty, "expected resourcePreEvents to survive the failure") + assert( + e.resourcePreEvents.exists(_.metadata.urn.endsWith("::doomed")), + s"expected the component's pre-event, got: ${e.resourcePreEvents.map(_.metadata.urn)}" + ) + assert(e.diagnostics.nonEmpty, "expected diagnostics to survive the failure") + assert( + e.diagnostics.exists(_.message.contains("deliberate failure from the test program")), + s"expected the program's error among the diagnostics, got: ${e.diagnostics.map(_.message)}" + ) + assertEquals(e.parseErrors, Nil) + assert(e.stdout.nonEmpty || e.stderr.nonEmpty, "expected the process output to be carried") + + // live delivery keeps working on the failure path too + assert(upEvents.preEvents.nonEmpty, "no engine events were delivered live before the failure") + assert(upEvents.firstArrivedBeforeReturn, "the first engine event did not arrive before up returned") + assertEquals(upEvents.preEvents.map(_.metadata.urn), e.resourcePreEvents.map(_.metadata.urn)) + + case Left(other) => fail(s"expected an OperationFailedError, got ${other.getClass.getName}: ${other.getMessage.take(300)}") + } + FunFixture[FullyQualifiedStackName]( setup = t => fqsn(this.getClass, t), teardown = _ => () diff --git a/auto/src/test/scala/besom/auto/internal/ShellTest.scala b/auto/src/test/scala/besom/auto/internal/ShellTest.scala index a0fbf023..5143d465 100644 --- a/auto/src/test/scala/besom/auto/internal/ShellTest.scala +++ b/auto/src/test/scala/besom/auto/internal/ShellTest.scala @@ -81,6 +81,20 @@ class ShellTest extends munit.FunSuite: assert(!p.isAlive, "the process should be done by the time the call returned") } + test("a throwing OnStart handler neither leaks nor kills the process") { + @volatile var handle: Option[ChildProcess] = None + + // skipping the join would leave the child running; destroying it would lose the output + val res = shell("sh", "-c", "echo hello")(ShellOption.OnStart { p => + handle = Some(p) + throw new RuntimeException("boom") + }).getOrElse(fail("a broken OnStart handler must not fail the command")) + + assertEquals(res.exitCode, 0) + assertEquals(res.out.trim, "hello") + assert(!handle.getOrElse(fail("OnStart was never invoked")).isAlive, "the process was left running") + } + test("OnStart handle can interrupt a running process") { @volatile var handle: Option[ChildProcess] = None val started = new java.util.concurrent.CountDownLatch(1) From b4d9f2dce07ec04a9579c1bfb12ff976eca6ed91 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Bia=C5=82y?= Date: Fri, 28 Aug 2026 21:05:34 +0200 Subject: [PATCH 3/3] fix linux signalling bug --- auto/src/main/scala/besom/auto/internal/ChildProcess.scala | 4 ++++ auto/src/test/scala/besom/auto/internal/ShellTest.scala | 4 +++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/auto/src/main/scala/besom/auto/internal/ChildProcess.scala b/auto/src/main/scala/besom/auto/internal/ChildProcess.scala index 010cf9fa..68852cd9 100644 --- a/auto/src/main/scala/besom/auto/internal/ChildProcess.scala +++ b/auto/src/main/scala/besom/auto/internal/ChildProcess.scala @@ -19,6 +19,10 @@ final class ChildProcess private[auto] (val underlying: os.SubProcess): * * The JDK exposes no signal API, so on Unix this shells out to `kill -INT`; on Windows, where there is no equivalent, it falls back to * [[terminate]]. + * + * The signal goes to the process besom-auto spawned, not to its descendants. That is exactly right for `pulumi`, which is always spawned + * directly and unwinds its own children - but it does mean a caller that interposes a shell (`sh -c "pulumi ..."`) would break + * cancellation, since the shell would receive the signal, ignore it while waiting on its foreground child, and never pass it on. */ def interrupt(): Unit = if isWindows then terminate() diff --git a/auto/src/test/scala/besom/auto/internal/ShellTest.scala b/auto/src/test/scala/besom/auto/internal/ShellTest.scala index 5143d465..82b7a210 100644 --- a/auto/src/test/scala/besom/auto/internal/ShellTest.scala +++ b/auto/src/test/scala/besom/auto/internal/ShellTest.scala @@ -99,8 +99,10 @@ class ShellTest extends munit.FunSuite: @volatile var handle: Option[ChildProcess] = None val started = new java.util.concurrent.CountDownLatch(1) + // `sleep` is spawned directly, not via `sh -c`: a shell is a separate process that ignores SIGINT while it waits on a + // foreground child, and on Linux `/bin/sh` (dash) does not exec, so the signal would reach the shell and never the sleep val runner = new Thread(() => - shell("sh", "-c", "sleep 30")(ShellOption.OnStart { p => + shell("sleep", "30")(ShellOption.OnStart { p => handle = Some(p) started.countDown() })