chore(pegboard): allow configuring base_retry_timeout, actor_start_threshold, actor_stop_threshold, and retry_reset_duration

Author: NathanFlurry

engine/packages/config/src/config/mod.rs

@@ -11,6 +11,7 @@ pub mod clickhouse;
 pub mod db;
 pub mod guard;
 pub mod logs;
+pub mod pegboard;
 pub mod pubsub;
 pub mod telemetry;
 pub mod topology;
@@ -24,6 +25,7 @@ pub use clickhouse::*;
 pub use db::Database;
 pub use guard::*;
 pub use logs::*;
+pub use pegboard::*;
 pub use pubsub::PubSub;
 pub use telemetry::*;
 pub use topology::*;
@@ -72,6 +74,9 @@ pub struct Root {
 	#[serde(default)]
 	pub api_peer: Option<ApiPeer>,
 
+	#[serde(default)]
+	pub pegboard: Option<Pegboard>,
+
 	#[serde(default)]
 	pub logs: Option<Logs>,
 
@@ -107,6 +112,7 @@ impl Default for Root {
 			guard: None,
 			api_public: None,
 			api_peer: None,
+			pegboard: None,
 			logs: None,
 			topology: None,
 			database: None,
@@ -136,6 +142,11 @@ impl Root {
 		self.api_peer.as_ref().unwrap_or(&DEFAULT)
 	}
 
+	pub fn pegboard(&self) -> &Pegboard {
+		static DEFAULT: LazyLock<Pegboard> = LazyLock::new(Pegboard::default);
+		self.pegboard.as_ref().unwrap_or(&DEFAULT)
+	}
+
 	pub fn logs(&self) -> &Logs {
 		static DEFAULT: LazyLock<Logs> = LazyLock::new(Logs::default);
 		self.logs.as_ref().unwrap_or(&DEFAULT)

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

@@ -0,0 +1,50 @@
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Serialize, Deserialize, Clone, Default, JsonSchema)]
+#[serde(deny_unknown_fields)]
+pub struct Pegboard {
+	/// Time to delay an actor from rescheduling after a rescheduling failure.
+	///
+	/// Unit is in milliseconds.
+	///
+	/// **Experimental**
+	pub base_retry_timeout: Option<usize>,
+	/// How long to wait after creating and not receiving a starting state before setting actor as lost.
+	///
+	/// Unit is in milliseconds.
+	///
+	/// **Experimental**
+	pub actor_start_threshold: Option<i64>,
+	/// How long to wait after stopping and not receiving a stop state before setting actor as lost.
+	///
+	/// Unit is in milliseconds.
+	///
+	/// **Experimental**
+	pub actor_stop_threshold: Option<i64>,
+	/// How long an actor goes without retries before it's retry count is reset to 0, effectively resetting its
+	/// backoff to 0.
+	///
+	/// Unit is in milliseconds.
+	///
+	/// **Experimental**
+	pub retry_reset_duration: Option<i64>,
+}
+
+impl Pegboard {
+	pub fn base_retry_timeout(&self) -> usize {
+		self.base_retry_timeout.unwrap_or(2000)
+	}
+
+	pub fn actor_start_threshold(&self) -> i64 {
+		self.actor_start_threshold.unwrap_or(30_000)
+	}
+
+	pub fn actor_stop_threshold(&self) -> i64 {
+		self.actor_stop_threshold.unwrap_or(30_000)
+	}
+
+	pub fn retry_reset_duration(&self) -> i64 {
+		self.retry_reset_duration.unwrap_or(10 * 60 * 1000)
+	}
+}

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

@@ -10,16 +10,6 @@ mod destroy;
 mod runtime;
 mod setup;
 
-/// Time to delay an actor from rescheduling after a rescheduling failure.
-const BASE_RETRY_TIMEOUT_MS: usize = 2000;
-/// How long to wait after creating and not receiving a starting state before setting actor as lost.
-const ACTOR_START_THRESHOLD_MS: i64 = util::duration::seconds(30);
-/// How long to wait after stopping and not receiving a stop state before setting actor as lost.
-const ACTOR_STOP_THRESHOLD_MS: i64 = util::duration::seconds(30);
-/// How long an actor goes without retries before it's retry count is reset to 0, effectively resetting its
-/// backoff to 0.
-const RETRY_RESET_DURATION_MS: i64 = util::duration::minutes(10);
-
 #[derive(Clone, Debug, Serialize, Deserialize, Hash)]
 pub struct Input {
 	pub actor_id: Id,
@@ -224,7 +214,11 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 		runtime::SpawnActorOutput::Allocated {
 			runner_id,
 			runner_workflow_id,
-		} => runtime::LifecycleState::new(runner_id, runner_workflow_id),
+		} => runtime::LifecycleState::new(
+			runner_id,
+			runner_workflow_id,
+			ctx.config().pegboard().actor_start_threshold(),
+		),
 		runtime::SpawnActorOutput::Sleep => {
 			ctx.activity(runtime::SetSleepingInput {
 				actor_id: input.actor_id,
@@ -308,8 +302,10 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 								}) => match intent {
 									protocol::ActorIntent::ActorIntentSleep => {
 										if !state.sleeping {
-											state.gc_timeout_ts =
-												Some(util::timestamp::now() + ACTOR_STOP_THRESHOLD_MS);
+											state.gc_timeout_ts = Some(
+												util::timestamp::now()
+													+ ctx.config().pegboard().actor_stop_threshold(),
+											);
 											state.sleeping = true;
 
 											ctx.activity(runtime::SetSleepingInput {
@@ -330,8 +326,10 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 										}
 									}
 									protocol::ActorIntent::ActorIntentStop => {
-										state.gc_timeout_ts =
-											Some(util::timestamp::now() + ACTOR_STOP_THRESHOLD_MS);
+										state.gc_timeout_ts = Some(
+											util::timestamp::now()
+												+ ctx.config().pegboard().actor_stop_threshold(),
+										);
 
 										ctx.activity(runtime::SetNotConnectableInput {
 											actor_id: input.actor_id,

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

@@ -15,10 +15,7 @@ use universaldb::utils::{FormalKey, IsolationLevel::*};
 
 use crate::{keys, metrics, workflows::runner::RUNNER_ELIGIBLE_THRESHOLD_MS};
 
-use super::{
-	ACTOR_START_THRESHOLD_MS, Allocate, BASE_RETRY_TIMEOUT_MS, Destroy, Input, PendingAllocation,
-	RETRY_RESET_DURATION_MS, State, destroy,
-};
+use super::{Allocate, Destroy, Input, PendingAllocation, State, destroy};
 
 #[derive(Deserialize, Serialize)]
 pub struct LifecycleState {
@@ -42,7 +39,7 @@ pub struct LifecycleState {
 }
 
 impl LifecycleState {
-	pub fn new(runner_id: Id, runner_workflow_id: Id) -> Self {
+	pub fn new(runner_id: Id, runner_workflow_id: Id, actor_start_threshold: i64) -> Self {
 		LifecycleState {
 			generation: 0,
 			runner_id: Some(runner_id),
@@ -51,7 +48,7 @@ impl LifecycleState {
 			will_wake: false,
 			wake_for_alarm: false,
 			alarm_ts: None,
-			gc_timeout_ts: Some(util::timestamp::now() + ACTOR_START_THRESHOLD_MS),
+			gc_timeout_ts: Some(util::timestamp::now() + actor_start_threshold),
 			reschedule_state: RescheduleState::default(),
 		}
 	}
@@ -622,7 +619,7 @@ pub async fn reschedule_actor(
 	let mut backoff = util::backoff::Backoff::new_at(
 		8,
 		None,
-		BASE_RETRY_TIMEOUT_MS,
+		ctx.config().pegboard().base_retry_timeout(),
 		500,
 		state.reschedule_state.retry_count,
 	);
@@ -675,7 +672,8 @@ pub async fn reschedule_actor(
 		state.runner_workflow_id = Some(*runner_workflow_id);
 
 		// Reset gc timeout once allocated
-		state.gc_timeout_ts = Some(util::timestamp::now() + ACTOR_START_THRESHOLD_MS);
+		state.gc_timeout_ts =
+			Some(util::timestamp::now() + ctx.config().pegboard().actor_start_threshold());
 	}
 
 	Ok(spawn_res)
@@ -727,8 +725,11 @@ struct CompareRetryInput {
 async fn compare_retry(ctx: &ActivityCtx, input: &CompareRetryInput) -> Result<(i64, bool)> {
 	let now = util::timestamp::now();
 
-	// If the last retry ts is more than RETRY_RESET_DURATION_MS ago, reset retry count
-	Ok((now, input.last_retry_ts < now - RETRY_RESET_DURATION_MS))
+	// If the last retry ts is more than retry_reset_duration ago, reset retry count
+	Ok((
+		now,
+		input.last_retry_ts < now - ctx.config().pegboard().retry_reset_duration(),
+	))
 }
 
 #[derive(Debug, Serialize, Deserialize, Hash)]