fix(pb): rewrite runner wf to handle batch signals (#3480)

Author: MasterPtato

engine/packages/pegboard-runner/src/conn.rs

@@ -154,7 +154,7 @@ pub async fn init_conn(
 			};
 
 		// Forward to runner wf
-		ctx.signal(pegboard::workflows::runner::Forward { inner: packet })
+		ctx.signal(pegboard::workflows::runner2::Forward { inner: packet })
 			.to_workflow_id(workflow_id)
 			.send()
 			.await

engine/packages/pegboard-runner/src/ping_task.rs

@@ -55,7 +55,7 @@ async fn update_runner_ping(ctx: &StandaloneCtx, conn: &Conn) -> Result<()> {
 		if let RunnerEligibility::ReEligible = notif.eligibility {
 			tracing::debug!(runner_id=?notif.runner_id, "runner has become eligible again");
 
-			ctx.signal(pegboard::workflows::runner::CheckQueue {})
+			ctx.signal(pegboard::workflows::runner2::CheckQueue {})
 				.to_workflow_id(notif.workflow_id)
 				.send()
 				.await?;

engine/packages/pegboard-runner/src/ws_to_tunnel_task.rs

@@ -358,7 +358,7 @@ async fn handle_message(
 		| protocol::ToServer::ToServerEvents(_)
 		| protocol::ToServer::ToServerAckCommands(_)
 		| protocol::ToServer::ToServerStopping => {
-			ctx.signal(pegboard::workflows::runner::Forward {
+			ctx.signal(pegboard::workflows::runner2::Forward {
 				inner: protocol::ToServer::try_from(msg)
 					.context("failed to convert message for workflow forwarding")?,
 			})

engine/packages/pegboard/src/workflows/actor/mod.rs

@@ -320,7 +320,7 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 											.await?;
 
 											// Send signal to stop actor now that we know it will be sleeping
-											ctx.signal(crate::workflows::runner::Command {
+											ctx.signal(crate::workflows::runner2::Command {
 												inner: protocol::Command::CommandStopActor(
 													protocol::CommandStopActor {
 														actor_id: input.actor_id.to_string(),
@@ -349,7 +349,7 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 											})
 											.await?;
 
-											ctx.signal(crate::workflows::runner::Command {
+											ctx.signal(crate::workflows::runner2::Command {
 												inner: protocol::Command::CommandStopActor(
 													protocol::CommandStopActor {
 														actor_id: input.actor_id.to_string(),
@@ -492,7 +492,7 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 								})
 								.await?;
 
-								ctx.signal(crate::workflows::runner::Command {
+								ctx.signal(crate::workflows::runner2::Command {
 									inner: protocol::Command::CommandStopActor(protocol::CommandStopActor {
 										actor_id: input.actor_id.to_string(),
 										generation: state.generation,
@@ -506,7 +506,7 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 						Main::Destroy(_) => {
 							// If allocated, send stop actor command
 							if let Some(runner_workflow_id) = state.runner_workflow_id {
-								ctx.signal(crate::workflows::runner::Command {
+								ctx.signal(crate::workflows::runner2::Command {
 									inner: protocol::Command::CommandStopActor(protocol::CommandStopActor {
 										actor_id: input.actor_id.to_string(),
 										generation: state.generation,
@@ -623,7 +623,7 @@ async fn handle_stopped(
 	if let (StoppedVariant::Lost { .. }, Some(old_runner_workflow_id)) =
 		(&variant, old_runner_workflow_id)
 	{
-		ctx.signal(crate::workflows::runner::Command {
+		ctx.signal(crate::workflows::runner2::Command {
 			inner: protocol::Command::CommandStopActor(protocol::CommandStopActor {
 				actor_id: input.actor_id.to_string(),
 				generation: state.generation,

engine/packages/pegboard/src/workflows/actor/runtime.rs

@@ -517,7 +517,7 @@ pub async fn spawn_actor(
 				.send()
 				.await?;
 
-			ctx.signal(crate::workflows::runner::Command {
+			ctx.signal(crate::workflows::runner2::Command {
 				inner: protocol::Command::CommandStartActor(protocol::CommandStartActor {
 					actor_id: input.actor_id.to_string(),
 					generation,
@@ -576,7 +576,7 @@ pub async fn spawn_actor(
 					})
 					.await?;
 
-					ctx.signal(crate::workflows::runner::Command {
+					ctx.signal(crate::workflows::runner2::Command {
 						inner: protocol::Command::CommandStartActor(protocol::CommandStartActor {
 							actor_id: input.actor_id.to_string(),
 							generation,

engine/packages/pegboard/src/workflows/mod.rs

@@ -1,2 +1,3 @@
 pub mod actor;
 pub mod runner;
+pub mod runner2;