feat(pegboard): move pegboard serverless to wf (#3416)

Fixes RVT-5269

Author: abcxff

Cargo.lock

@@ -1,7 +1,7 @@
 
 [workspace]
 resolver = "2"
-members = ["engine/packages/actor-kv","engine/packages/api-builder","engine/packages/api-peer","engine/packages/api-public","engine/packages/api-types","engine/packages/api-util","engine/packages/bootstrap","engine/packages/cache","engine/packages/cache-purge","engine/packages/cache-result","engine/packages/clickhouse-inserter","engine/packages/clickhouse-user-query","engine/packages/config","engine/packages/dump-openapi","engine/packages/engine","engine/packages/env","engine/packages/epoxy","engine/packages/error","engine/packages/error-macros","engine/packages/gasoline","engine/packages/gasoline-macros","engine/packages/guard","engine/packages/guard-core","engine/packages/internal","engine/packages/logs","engine/packages/metrics","engine/packages/namespace","engine/packages/pegboard","engine/packages/pegboard-gateway","engine/packages/pegboard-runner","engine/packages/pegboard-serverless","engine/packages/pools","engine/packages/runtime","engine/packages/service-manager","engine/packages/telemetry","engine/packages/test-deps","engine/packages/test-deps-docker","engine/packages/tracing-reconfigure","engine/packages/types","engine/packages/universaldb","engine/packages/universalpubsub","engine/packages/util","engine/packages/util-id","engine/packages/workflow-worker","engine/sdks/rust/api-full","engine/sdks/rust/data","engine/sdks/rust/epoxy-protocol","engine/sdks/rust/runner-protocol","engine/sdks/rust/ups-protocol"]
+members = ["engine/packages/actor-kv","engine/packages/api-builder","engine/packages/api-peer","engine/packages/api-public","engine/packages/api-types","engine/packages/api-util","engine/packages/bootstrap","engine/packages/cache","engine/packages/cache-purge","engine/packages/cache-result","engine/packages/clickhouse-inserter","engine/packages/clickhouse-user-query","engine/packages/config","engine/packages/dump-openapi","engine/packages/engine","engine/packages/env","engine/packages/epoxy","engine/packages/error","engine/packages/error-macros","engine/packages/gasoline","engine/packages/gasoline-macros","engine/packages/guard","engine/packages/guard-core","engine/packages/logs","engine/packages/metrics","engine/packages/namespace","engine/packages/pegboard","engine/packages/pegboard-gateway","engine/packages/pegboard-runner","engine/packages/pools","engine/packages/runtime","engine/packages/service-manager","engine/packages/telemetry","engine/packages/test-deps","engine/packages/test-deps-docker","engine/packages/tracing-reconfigure","engine/packages/types","engine/packages/universaldb","engine/packages/universalpubsub","engine/packages/util","engine/packages/util-id","engine/packages/workflow-worker","engine/sdks/rust/api-full","engine/sdks/rust/data","engine/sdks/rust/epoxy-protocol","engine/sdks/rust/runner-protocol","engine/sdks/rust/ups-protocol"]
 
 [workspace.package]
 version = "2.0.25"
@@ -333,9 +333,6 @@ path = "engine/packages/guard"
 [workspace.dependencies.rivet-guard-core]
 path = "engine/packages/guard-core"
 
-[workspace.dependencies.internal]
-path = "engine/packages/internal"
-
 [workspace.dependencies.rivet-logs]
 path = "engine/packages/logs"
 
@@ -354,9 +351,6 @@ path = "engine/packages/pegboard-gateway"
 [workspace.dependencies.pegboard-runner]
 path = "engine/packages/pegboard-runner"
 
-[workspace.dependencies.pegboard-serverless]
-path = "engine/packages/pegboard-serverless"
-
 [workspace.dependencies.rivet-pools]
 path = "engine/packages/pools"
 

Cargo.toml

@@ -24,11 +24,11 @@ serde.workspace = true
 serde_json.workspace = true
 indexmap.workspace = true
 
-tokio.workspace = true
-tracing.workspace = true
 namespace.workspace = true
 pegboard.workspace = true
 pegboard-actor-kv.workspace = true
+tokio.workspace = true
+tracing.workspace = true
 universalpubsub.workspace = true
 uuid.workspace = true
 utoipa.workspace = true

engine/artifacts/errors/serverless_runner_pool.not_found.json

@@ -29,23 +29,6 @@ pub async fn cache_purge(
 	Ok(CachePurgeResponse {})
 }
 
-#[derive(Serialize)]
-#[serde(deny_unknown_fields)]
-pub struct BumpServerlessAutoscalerResponse {}
-
-pub async fn bump_serverless_autoscaler(
-	ctx: ApiCtx,
-	_path: (),
-	_query: (),
-	_body: (),
-) -> Result<BumpServerlessAutoscalerResponse> {
-	ctx.msg(rivet_types::msgs::pegboard::BumpServerlessAutoscaler {})
-		.send()
-		.await?;
-
-	Ok(BumpServerlessAutoscalerResponse {})
-}
-
 #[derive(Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct SetTracingConfigRequest {

engine/artifacts/openapi.json

@@ -35,10 +35,6 @@ pub async fn router(
 			.route("/runners/names", get(runners::list_names))
 			// MARK: Internal
 			.route("/cache/purge", post(internal::cache_purge))
-			.route(
-				"/bump-serverless-autoscaler",
-				post(internal::bump_serverless_autoscaler),
-			)
 			.route(
 				"/epoxy/coordinator/replica-reconfigure",
 				post(internal::epoxy_replica_reconfigure),

engine/packages/api-peer/Cargo.toml

@@ -26,7 +26,7 @@ pub async fn list(ctx: ApiCtx, _path: ListPath, query: ListQuery) -> Result<List
 
 	if !runner_names.is_empty() {
 		let runner_configs = ctx
-			.op(namespace::ops::runner_config::get::Input {
+			.op(pegboard::ops::runner_config::get::Input {
 				runners: runner_names
 					.into_iter()
 					.map(|name| (namespace.namespace_id, name))
@@ -63,7 +63,7 @@ pub async fn list(ctx: ApiCtx, _path: ListPath, query: ListQuery) -> Result<List
 		};
 
 		let runner_configs = ctx
-			.op(namespace::ops::runner_config::list::Input {
+			.op(pegboard::ops::runner_config::list::Input {
 				namespace_id: namespace.namespace_id,
 				variant,
 				after_name,
@@ -121,7 +121,7 @@ pub async fn upsert(
 		.ok_or_else(|| namespace::errors::Namespace::NotFound.build())?;
 
 	let endpoint_config_changed = ctx
-		.op(namespace::ops::runner_config::upsert::Input {
+		.op(pegboard::ops::runner_config::upsert::Input {
 			namespace_id: namespace.namespace_id,
 			name: path.runner_name,
 			config: body.0.into(),
@@ -159,7 +159,7 @@ pub async fn delete(ctx: ApiCtx, path: DeletePath, query: DeleteQuery) -> Result
 		.await?
 		.ok_or_else(|| namespace::errors::Namespace::NotFound.build())?;
 
-	ctx.op(namespace::ops::runner_config::delete::Input {
+	ctx.op(pegboard::ops::runner_config::delete::Input {
 		namespace_id: namespace.namespace_id,
 		name: path.runner_name,
 	})

engine/packages/api-peer/src/internal.rs

@@ -88,7 +88,7 @@ async fn refresh_metadata_inner(
 		.collect();
 
 	let runner_configs = ctx
-		.op(namespace::ops::runner_config::get::Input {
+		.op(pegboard::ops::runner_config::get::Input {
 			runners,
 			bypass_cache: true,
 		})

engine/packages/api-peer/src/router.rs

@@ -56,6 +56,25 @@ pub struct Pegboard {
 	///
 	/// **Experimental**
 	pub hibernating_request_eligible_threshold: Option<i64>,
+	/// Time to delay a serverless runner from attempting a new outbound connection after a connection failure.
+	///
+	/// Unit is in milliseconds.
+	///
+	/// **Experimental**
+	pub serverless_base_retry_timeout: Option<usize>,
+	/// How long a serverless runner goes without connection failures before it's retry count is reset to 0,
+	/// effectively resetting its backoff to 0.
+	///
+	/// Unit is in milliseconds.
+	///
+	/// **Experimental**
+	pub serverless_retry_reset_duration: Option<i64>,
+	/// Maximum exponent for the serverless backoff calculation.
+	///
+	/// This controls the maximum backoff duration when serverlessly connecting to runners.
+	///
+	/// **Experimental**
+	pub serverless_backoff_max_exponent: Option<usize>,
 }
 
 impl Pegboard {
@@ -91,4 +110,17 @@ impl Pegboard {
 		self.hibernating_request_eligible_threshold
 			.unwrap_or(90_000)
 	}
+
+	pub fn serverless_base_retry_timeout(&self) -> usize {
+		self.serverless_base_retry_timeout.unwrap_or(2000)
+	}
+
+	pub fn serverless_retry_reset_duration(&self) -> i64 {
+		self.serverless_retry_reset_duration
+			.unwrap_or(10 * 60 * 1000)
+	}
+
+	pub fn serverless_backoff_max_exponent(&self) -> usize {
+		self.serverless_backoff_max_exponent.unwrap_or(8)
+	}
 }