feat: basic envoy tunnel impl

Author: MasterPtato

Cargo.lock

@@ -34,6 +34,8 @@ pub async fn delete(ctx: ApiCtx, path: DeletePath, query: DeleteQuery) -> Result
 		}),
 	)?;
 
+	let namespace = namespace_res.ok_or_else(|| namespace::errors::Namespace::NotFound.build())?;
+
 	let actor = actors_res
 		.actors
 		.into_iter()
@@ -45,13 +47,12 @@ pub async fn delete(ctx: ApiCtx, path: DeletePath, query: DeleteQuery) -> Result
 		return Err(pegboard::errors::Actor::NotFound.build());
 	}
 
-	let namespace = namespace_res.ok_or_else(|| namespace::errors::Namespace::NotFound.build())?;
-
 	// Verify the actor belongs to the specified namespace
 	if actor.namespace_id != namespace.namespace_id {
 		return Err(pegboard::errors::Actor::NotFound.build());
 	}
 
+	// TODO: Actor v2
 	let res = ctx
 		.signal(pegboard::workflows::actor::Destroy {})
 		.to_workflow::<pegboard::workflows::actor::Workflow>()

engine/artifacts/config-schema.json

@@ -125,7 +125,7 @@ pub struct Pegboard {
 	/// Max time since last seen before actor is considered stale, in milliseconds.
 	pub envoy_event_demuxer_max_last_seen_threshold: Option<u64>,
 	/// Max response payload size in bytes from actors.
-	pub envoy_max_response_payload_body_size: Option<usize>,
+	pub envoy_max_response_payload_size: Option<usize>,
 	/// Ping interval for envoy updates in milliseconds.
 	pub envoy_update_ping_interval: Option<u64>,
 	/// How long after last ping before considering a envoy ineligible for allocation.
@@ -282,8 +282,8 @@ impl Pegboard {
 			.unwrap_or(30_000)
 	}
 
-	pub fn envoy_max_response_payload_body_size(&self) -> usize {
-		self.envoy_max_response_payload_body_size
+	pub fn envoy_max_response_payload_size(&self) -> usize {
+		self.envoy_max_response_payload_size
 			.unwrap_or(20 * 1024 * 1024) // 20 MiB
 	}
 

engine/packages/api-peer/src/actors/delete.rs

@@ -148,6 +148,8 @@ where
 			} else {
 				tracing::debug!(?tags, "unique workflow already exists");
 			}
+		} else {
+			tracing::debug!(?actual_workflow_id, "dispatched workflow");
 		}
 
 		if workflow_id == actual_workflow_id {

engine/packages/config/src/config/pegboard.rs

@@ -17,7 +17,11 @@ impl SharedState {
 	}
 
 	pub async fn start(&self) -> Result<()> {
-		self.pegboard_gateway.start().await?;
+		tokio::try_join!(
+			self.pegboard_gateway.start(),
+			self.pegboard_gateway2.start(),
+		)?;
+
 		Ok(())
 	}
 }

engine/packages/gasoline/src/builder/common/workflow.rs

@@ -334,6 +334,7 @@ pub async fn handle_init(
 				serverless_drain_grace_period: conn
 					.is_serverless
 					.then(|| pb.serverless_drain_grace_period() as i64),
+				max_response_payload_size: pb.envoy_max_response_payload_size() as u64,
 			},
 		}));
 	let init_msg_serialized = init_msg.serialize(conn.protocol_version)?;

engine/packages/guard/src/shared_state.rs

@@ -454,12 +454,7 @@ async fn handle_tunnel_message(
 	let inner_data_len = tunnel_message_inner_data_len(&msg.message_kind);
 
 	// Enforce incoming payload size
-	if inner_data_len
-		> ctx
-			.config()
-			.pegboard()
-			.envoy_max_response_payload_body_size()
-	{
+	if inner_data_len > ctx.config().pegboard().envoy_max_response_payload_size() {
 		return Err(errors::WsError::InvalidPacket("payload too large".to_string()).build());
 	}
 

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

@@ -308,7 +308,7 @@ impl SharedState {
 		while let Ok(NextOutput::Message(msg)) = sub.next().await {
 			tracing::trace!(
 				payload_len = msg.payload.len(),
-				"received message from pubsub"
+				"received message from envoy"
 			);
 
 			match versioned::ToGateway::deserialize_with_embedded_version(&msg.payload) {

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

@@ -358,8 +358,17 @@ struct CheckEnvoyLivenessInput {
 	envoy_key: String,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+struct CheckEnvoyLivenessOutput {
+	expired: bool,
+	now: i64,
+}
+
 #[activity(CheckEnvoyLiveness)]
-async fn check_envoy_liveness(ctx: &ActivityCtx, input: &CheckEnvoyLivenessInput) -> Result<bool> {
+async fn check_envoy_liveness(
+	ctx: &ActivityCtx,
+	input: &CheckEnvoyLivenessInput,
+) -> Result<CheckEnvoyLivenessOutput> {
 	let state = ctx.state::<State>()?;
 	let envoy_lost_threshold = ctx.config().pegboard().envoy_lost_threshold();
 
@@ -378,7 +387,7 @@ async fn check_envoy_liveness(ctx: &ActivityCtx, input: &CheckEnvoyLivenessInput
 			let now = util::timestamp::now();
 			let expired = last_ping_ts < now - envoy_lost_threshold;
 
-			Ok(expired)
+			Ok(CheckEnvoyLivenessOutput { expired, now })
 		})
 		.custom_instrument(tracing::info_span!("actor_check_envoy_liveness_tx"))
 		.await
@@ -392,7 +401,7 @@ async fn listen_for_signals(
 	metrics_workflow_id: Id,
 ) -> Result<Vec<Main>> {
 	// Listen for signals based on transition
-	let signals = match &state.transition {
+	let signals = match &mut state.transition {
 		Transition::Allocating {
 			lost_timeout_ts, ..
 		}
@@ -432,20 +441,22 @@ async fn listen_for_signals(
 			// Listen for signals with periodic liveness check timeout
 			let signals = ctx
 				.listen_n_until::<Main>(
-					last_liveness_check_ts + ctx.config().pegboard().envoy_lost_threshold(),
+					*last_liveness_check_ts + ctx.config().pegboard().envoy_lost_threshold(),
 					256,
 				)
 				.await?;
 
 			// Perform liveness check
 			if signals.is_empty() {
-				let expired = ctx
+				let res = ctx
 					.activity(CheckEnvoyLivenessInput {
 						envoy_key: envoy.envoy_key.clone(),
 					})
 					.await?;
 
-				if expired {
+				*last_liveness_check_ts = res.now;
+
+				if res.expired {
 					vec![Main::Lost(Lost {
 						generation: state.generation,
 						reason: LostReason::EnvoyConnectionLost,