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

Author: MasterPtato

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

@@ -166,7 +166,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

@@ -50,7 +50,7 @@ pub async fn task(
 			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

@@ -342,7 +342,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

@@ -319,7 +319,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(),
@@ -348,7 +348,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(),
@@ -495,7 +495,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,
@@ -509,7 +509,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,
@@ -628,7 +628,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

@@ -514,7 +514,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,
@@ -563,7 +563,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;