feat(toolchain): remote build logs

abcxff · abcxff · commit bcf92bd8f3a7 · 2025-06-11T23:16:51.000-04:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/cloud/packages/ci-manager/src/executors/docker.ts b/cloud/packages/ci-manager/src/executors/docker.ts
@@ -38,6 +38,11 @@ export async function runDockerBuild(
 		`Starting kaniko with args: docker ${kanikoArgs.join(" ")}`,
 	);
 
+	buildStore.updateStatus(buildId, {
+		type: "running",
+		data: { docker: {} }
+	});
+
 	return new Promise<void>((resolve, reject) => {
 		const dockerProcess = spawn("docker", kanikoArgs, {
 			stdio: ["pipe", "pipe", "pipe"]
diff --git a/cloud/packages/ci-manager/src/executors/rivet.ts b/cloud/packages/ci-manager/src/executors/rivet.ts
@@ -78,6 +78,13 @@ export async function runRivetBuild(
 		const actorId = createResponse.actor.id;
 		buildStore.addLog(buildId, `Created Rivet actor: ${actorId}`);
 
+		buildStore.updateStatus(buildId, {
+			type: "running",
+			data: {
+				rivet: { actorId }
+			}
+		});
+
 		await pollActorStatus(
 			buildStore,
 			client,
diff --git a/cloud/packages/ci-manager/src/kaniko-runner.ts b/cloud/packages/ci-manager/src/kaniko-runner.ts
@@ -17,7 +17,12 @@ export async function runKanikoBuild(
 
 	await mkdir(dirname(build.contextPath!), { recursive: true });
 
-	buildStore.updateStatus(buildId, { type: "running", data: {} });
+	buildStore.updateStatus(buildId, {
+		type: "running",
+		data: {
+			noRunner: {}
+		}
+	});
 
 	const executionMode = process.env.KANIKO_EXECUTION_MODE || "docker";
 	buildStore.addLog(buildId, `Using execution mode: ${executionMode}`);
diff --git a/cloud/packages/ci-manager/src/types.ts b/cloud/packages/ci-manager/src/types.ts
@@ -1,9 +1,25 @@
 import { z } from "zod";
 import { NO_SEP_CHAR_REGEX, UNIT_SEP_CHAR } from "./common";
 
+export const RunnerSchema = z.union([
+    z.object({
+        rivet: z.object({
+            actorId: z.string(),
+        })
+    }),
+    z.object({
+        docker: z.object({})
+    }),
+    z.object({
+        noRunner: z.object({})
+    })
+]);
+
+export type Runner = z.infer<typeof RunnerSchema>;
+
 export const StatusSchema = z.discriminatedUnion("type", [
     z.object({ type: z.literal("starting"), data: z.object({}) }),
-    z.object({ type: z.literal("running"), data: z.object({}) }),
+    z.object({ type: z.literal("running"), data: RunnerSchema }),
     z.object({ type: z.literal("finishing"), data: z.object({}) }),
     z.object({ type: z.literal("converting"), data: z.object({}) }),
     z.object({ type: z.literal("uploading"), data: z.object({}) }),
diff --git a/packages/toolchain/cli/Cargo.toml b/packages/toolchain/cli/Cargo.toml
@@ -16,7 +16,6 @@ path = "src/main.rs"
 [dependencies]
 anyhow = "1.0"
 async-posthog.workspace = true
-base64 = "0.22.1"
 clap = { version = "4.5.9", features = ["derive"] }
 ctrlc = "3.4.5"
 deno-embed.workspace = true
diff --git a/packages/toolchain/cli/src/commands/actor/create.rs b/packages/toolchain/cli/src/commands/actor/create.rs
@@ -103,7 +103,7 @@ pub struct Opts {
 
 	/// Specify which log stream to display
 	#[clap(long)]
-	log_stream: Option<crate::util::actor::logs::LogStream>,
+	log_stream: Option<toolchain::util::actor::logs::LogStream>,
 
 	/// Deploy the build before creating the actor
 	#[clap(long)]
@@ -299,17 +299,18 @@ impl Opts {
 
 		// Tail logs
 		if self.logs {
-			crate::util::actor::logs::tail(
+			toolchain::util::actor::logs::tail(
 				&ctx,
-				crate::util::actor::logs::TailOpts {
+				toolchain::util::actor::logs::TailOpts {
 					environment: &env,
 					actor_id: response.actor.id,
 					stream: self
 						.log_stream
 						.clone()
-						.unwrap_or(crate::util::actor::logs::LogStream::All),
+						.unwrap_or(toolchain::util::actor::logs::LogStream::All),
 					follow: true,
 					timestamps: true,
+					exit_on_ctrl_c: true,
 				},
 			)
 			.await?;
diff --git a/packages/toolchain/cli/src/commands/actor/logs.rs b/packages/toolchain/cli/src/commands/actor/logs.rs
@@ -16,7 +16,7 @@ pub struct Opts {
 
 	/// Specify which log stream to display (stdout, stderr, or all)
 	#[clap(long, short = 's')]
-	stream: Option<crate::util::actor::logs::LogStream>,
+	stream: Option<toolchain::util::actor::logs::LogStream>,
 
 	/// Disable timestamp display in logs
 	#[clap(long)]
@@ -36,17 +36,18 @@ impl Opts {
 		let actor_id =
 			Uuid::parse_str(&self.id).map_err(|_| errors::UserError::new("invalid id uuid"))?;
 
-		crate::util::actor::logs::tail(
+		toolchain::util::actor::logs::tail(
 			&ctx,
-			crate::util::actor::logs::TailOpts {
+			toolchain::util::actor::logs::TailOpts {
 				environment: &env,
 				actor_id,
 				stream: self
 					.stream
 					.clone()
-					.unwrap_or(crate::util::actor::logs::LogStream::All),
+					.unwrap_or(toolchain::util::actor::logs::LogStream::All),
 				follow: !self.no_follow,
 				timestamps: !self.no_timestamps,
+				exit_on_ctrl_c: true
 			},
 		)
 		.await?;
diff --git a/packages/toolchain/cli/src/util/mod.rs b/packages/toolchain/cli/src/util/mod.rs
@@ -1,4 +1,3 @@
-pub mod actor;
 pub mod deploy;
 pub mod env;
 pub mod login;
diff --git a/packages/toolchain/toolchain/Cargo.toml b/packages/toolchain/toolchain/Cargo.toml
@@ -9,6 +9,7 @@ edition.workspace = true
 [dependencies]
 anyhow = "1.0"
 async-stream = "0.3.3"
+base64 = "0.22.1"
 clap = { version = "4.5", features = ["derive"] }
 console = "0.15"
 const_format = "0.2.32"
diff --git a/packages/toolchain/toolchain/src/util/actor/logs.rs b/packages/toolchain/toolchain/src/util/actor/logs.rs
@@ -4,7 +4,10 @@ use clap::ValueEnum;
 use std::time::Duration;
 use tokio::signal;
 use tokio::sync::watch;
-use toolchain::rivet_api::{apis, models};
+use crate::{
+	rivet_api::{apis, models},
+	ToolchainCtx
+};
 use uuid::Uuid;
 
 #[derive(ValueEnum, Clone)]
@@ -23,25 +26,28 @@ pub struct TailOpts<'a> {
 	pub stream: LogStream,
 	pub follow: bool,
 	pub timestamps: bool,
+	pub exit_on_ctrl_c: bool,
 }
 
 /// Reads the logs of an actor.
-pub async fn tail(ctx: &toolchain::ToolchainCtx, opts: TailOpts<'_>) -> Result<()> {
+pub async fn tail(ctx: &ToolchainCtx, opts: TailOpts<'_>) -> Result<()> {
 	let (stdout_fetched_tx, stdout_fetched_rx) = watch::channel(false);
 	let (stderr_fetched_tx, stderr_fetched_rx) = watch::channel(false);
 
+	let exit_on_ctrl_c = opts.exit_on_ctrl_c;
+	
 	tokio::select! {
 		result = tail_streams(ctx, &opts, stdout_fetched_tx, stderr_fetched_tx) => result,
 		result = poll_actor_state(ctx, &opts, stdout_fetched_rx, stderr_fetched_rx) => result,
-		_ = signal::ctrl_c() => {
+		_ = signal::ctrl_c(), if exit_on_ctrl_c => {
 			Ok(())
 		}
 	}
 }
 
 /// Reads the streams of an actor's logs.
 async fn tail_streams(
-	ctx: &toolchain::ToolchainCtx,
+	ctx: &ToolchainCtx,
 	opts: &TailOpts<'_>,
 	stdout_fetched_tx: watch::Sender<bool>,
 	stderr_fetched_tx: watch::Sender<bool>,
@@ -66,7 +72,7 @@ async fn tail_streams(
 
 /// Reads a specific stream of an actor's log.
 async fn tail_stream(
-	ctx: &toolchain::ToolchainCtx,
+	ctx: &ToolchainCtx,
 	opts: &TailOpts<'_>,
 	stream: models::ActorsQueryLogStream,
 	log_fetched_tx: watch::Sender<bool>,
@@ -138,7 +144,7 @@ async fn tail_stream(
 ///
 /// Using this in a `tokio::select` will make all other tasks cancel when the actor finishes.
 async fn poll_actor_state(
-	ctx: &toolchain::ToolchainCtx,
+	ctx: &ToolchainCtx,
 	opts: &TailOpts<'_>,
 	mut stdout_fetched_rx: watch::Receiver<bool>,
 	mut stderr_fetched_rx: watch::Receiver<bool>,
diff --git a/packages/toolchain/toolchain/src/util/actor/mod.rs b/packages/toolchain/toolchain/src/util/actor/mod.rs
diff --git a/packages/toolchain/toolchain/src/util/docker/build_remote.rs b/packages/toolchain/toolchain/src/util/docker/build_remote.rs
@@ -505,7 +505,7 @@ async fn upload_build_context(
 }
 
 async fn poll_build_status(
-	_ctx: &ToolchainCtx,
+	ctx: &ToolchainCtx,
 	task: task::TaskCtx,
 	build_id: &str,
 	ci_manager_endpoint: &str,
@@ -514,11 +514,16 @@ async fn poll_build_status(
 
 	// Poll build status until completion
 	task.log("[Remote Build] Polling build status...");
-	let max_attempts = 900; // 30 minutes
-	let interval = Duration::from_secs(2);
+	let max_timeout = Duration::from_secs(30 * 60);
+	let poll_interval = Duration::from_secs(2);
+	let start_time = std::time::Instant::now();
+
+	loop {
+		if start_time.elapsed() >= max_timeout {
+			bail!("Build polling timeout after 30 minutes");
+		}
 
-	for attempt in 0..max_attempts {
-		tokio::time::sleep(interval).await;
+		tokio::time::sleep(poll_interval).await;
 
 		let response = reqwest::Client::new()
 			.get(&format!("{}/builds/{}", server_url, build_id))
@@ -539,8 +544,6 @@ async fn poll_build_status(
 						.and_then(|t| t.as_str())
 						.unwrap_or("unknown");
 
-					task.log(format!("[Remote Build] Build status: {}", status));
-
 					match status {
 						"success" => {
 							task.log("[Remote Build] Build completed successfully");
@@ -562,33 +565,71 @@ async fn poll_build_status(
 								.unwrap_or("Unknown error");
 							bail!("Remote build failed: {}", reason);
 						}
+						"running" => {
+							task.log(format!("[Remote Build] Build status: {}", status));
+
+							// If we know its running, and have its actor ID,
+							// we can stop polling and start streaming logs
+							// until the actor exists.
+							let runner = build_info
+								.get("status")
+								.and_then(|s| s.get("data"))
+								.and_then(|d| d.get("rivet"));
+							if let Some(runner) = runner {
+								let actor_id_raw = runner
+									.get("actorId")
+									.and_then(|a| a.as_str())
+									.context("Rivet runner missing actorId")?;
+
+								let actor_id = Uuid::parse_str(&actor_id_raw)
+									.context("Failed to parse actor ID from rivet runner")?;
+
+								task.log("[Remote Build] Streaming build logs from actor.");
+
+								match crate::util::actor::logs::tail(
+									&ctx,
+									crate::util::actor::logs::TailOpts {
+										environment: CI_ENVIRONMENT_ID,
+										actor_id,
+										stream: crate::util::actor::logs::LogStream::All,
+										follow: true,
+										timestamps: true,
+										exit_on_ctrl_c: false
+									},
+								)
+								.await {
+									Ok(_) => {
+										task.log("[Remote Build] Build logs streaming completed.");
+									}
+									Err(e) => {
+										task.log(format!(
+											"[Remote Build] Failed to stream build logs: {}",
+											e
+										));
+									}
+								}
+							}
+						}
 						_ => {
+							task.log(format!("[Remote Build] Build status: {}", status));
 							// Continue polling for other statuses (pending, running, etc.)
 						}
 					}
 				} else {
 					task.log(format!(
-						"[Remote Build] Poll attempt {} failed: HTTP {}",
-						attempt + 1,
+						"[Remote Build] Poll failed: HTTP {}",
 						res.status()
 					));
 				}
 			}
 			Err(e) => {
 				task.log(format!(
-					"[Remote Build] Poll attempt {} failed: {}",
-					attempt + 1,
+					"[Remote Build] Poll failed: {}",
 					e
 				));
 			}
 		}
-
-		if attempt == max_attempts - 1 {
-			bail!("Build polling timeout after {} attempts", max_attempts);
-		}
 	}
-
-	bail!("timed out polling status")
 }
 
 async fn _get_build_by_tags(
diff --git a/packages/toolchain/toolchain/src/util/mod.rs b/packages/toolchain/toolchain/src/util/mod.rs
@@ -1,3 +1,4 @@
+pub mod actor;
 pub mod api;
 pub mod build;
 pub mod cmd;

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,3 @@`
`1`		`-pub mod actor;`
`2`	`1`	`pub mod deploy;`
`3`	`2`	`pub mod env;`
`4`	`3`	`pub mod login;`