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 keys;
 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,
@@ -227,7 +217,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,
@@ -311,8 +305,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 {
@@ -333,8 +329,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

@@ -1,5 +1,5 @@
-use base64::Engine;
 use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
 use futures_util::StreamExt;
 use futures_util::TryStreamExt;
 use gas::prelude::*;
@@ -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::{destroy, Allocate, Destroy, Input, PendingAllocation, State};
 
 #[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(),
 		}
 	}
@@ -619,7 +616,10 @@ pub async fn reschedule_actor(
 	tracing::debug!(actor_id=?input.actor_id, "rescheduling actor");
 
 	// Determine next backoff sleep duration
-	let mut backoff = reschedule_backoff(state.reschedule_state.retry_count);
+	let mut backoff = reschedule_backoff(
+		state.reschedule_state.retry_count,
+		ctx.config().pegboard().base_retry_timeout(),
+	);
 
 	let (now, reset) = ctx
 		.v(2)
@@ -670,7 +670,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)
@@ -725,13 +726,17 @@ async fn compare_retry(ctx: &ActivityCtx, input: &CompareRetryInput) -> Result<(
 	let mut state = ctx.state::<State>()?;
 
 	let now = util::timestamp::now();
+
 	// If the last retry ts is more than RETRY_RESET_DURATION_MS ago, reset retry count
-	let reset = input.last_retry_ts < now - RETRY_RESET_DURATION_MS;
+	let reset = input.last_retry_ts < now - ctx.config().pegboard().retry_reset_duration();
 
 	if reset {
 		state.reschedule_ts = None;
 	} else {
-		let backoff = reschedule_backoff(input.retry_count);
+		let backoff = reschedule_backoff(
+			input.retry_count,
+			ctx.config().pegboard().base_retry_timeout(),
+		);
 		state.reschedule_ts = Some(now + i64::try_from(backoff.current_duration())?);
 	}
 
@@ -810,6 +815,6 @@ pub async fn set_complete(ctx: &ActivityCtx, input: &SetCompleteInput) -> Result
 	Ok(())
 }
 
-fn reschedule_backoff(retry_count: usize) -> util::backoff::Backoff {
-	util::backoff::Backoff::new_at(8, None, BASE_RETRY_TIMEOUT_MS, 500, retry_count)
+fn reschedule_backoff(retry_count: usize, base_retry_timeout: usize) -> util::backoff::Backoff {
+	util::backoff::Backoff::new_at(8, None, base_retry_timeout, 500, retry_count)
 }