feat(pb): add reschedule_ts to actors

engine/packages/pegboard/src/ops/actor/get.rs

@@ -87,10 +87,11 @@ pub async fn pegboard_actor_get(ctx: &OperationCtx, input: &Input) -> Result<Out
 			crash_policy: actor_state.crash_policy,
 
 			create_ts: actor_state.create_ts,
-			pending_allocation_ts: actor_state.pending_allocation_ts,
 			start_ts: actor_state.start_ts,
-			sleep_ts: actor_state.sleep_ts,
+			pending_allocation_ts: actor_state.pending_allocation_ts,
 			connectable_ts: actor_state.connectable_ts,
+			sleep_ts: actor_state.sleep_ts,
+			reschedule_ts: actor_state.reschedule_ts,
 			destroy_ts: actor_state.destroy_ts,
 		});
 	}

engine/packages/pegboard/src/ops/actor/list_for_ns.rs

@@ -198,6 +198,7 @@ pub async fn pegboard_actor_list_for_ns(ctx: &OperationCtx, input: &Input) -> Re
 			start_ts: actor_state.start_ts,
 			sleep_ts: actor_state.sleep_ts,
 			connectable_ts: actor_state.connectable_ts,
+			reschedule_ts: actor_state.reschedule_ts,
 			destroy_ts: actor_state.destroy_ts,
 		});
 	}

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

@@ -58,6 +58,8 @@ pub struct State {
 	pub complete_ts: Option<i64>,
 	pub connectable_ts: Option<i64>,
 	pub pending_allocation_ts: Option<i64>,
+	#[serde(default)]
+	pub reschedule_ts: Option<i64>,
 	pub destroy_ts: Option<i64>,
 
 	// Null if not allocated
@@ -93,6 +95,7 @@ impl State {
 			sleep_ts: None,
 			connectable_ts: None,
 			complete_ts: None,
+			reschedule_ts: None,
 			destroy_ts: None,
 
 			runner_id: None,

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

@@ -619,17 +619,12 @@ pub async fn reschedule_actor(
 	tracing::debug!(actor_id=?input.actor_id, "rescheduling actor");
 
 	// Determine next backoff sleep duration
-	let mut backoff = util::backoff::Backoff::new_at(
-		8,
-		None,
-		BASE_RETRY_TIMEOUT_MS,
-		500,
-		state.reschedule_state.retry_count,
-	);
+	let mut backoff = reschedule_backoff(state.reschedule_state.retry_count);
 
 	let (now, reset) = ctx
 		.v(2)
 		.activity(CompareRetryInput {
+			retry_count: state.reschedule_state.retry_count,
 			last_retry_ts: state.reschedule_state.last_retry_ts,
 		})
 		.await?;
@@ -720,15 +715,27 @@ pub async fn clear_pending_allocation(
 
 #[derive(Debug, Serialize, Deserialize, Hash)]
 struct CompareRetryInput {
+	#[serde(default)]
+	retry_count: usize,
 	last_retry_ts: i64,
 }
 
 #[activity(CompareRetry)]
 async fn compare_retry(ctx: &ActivityCtx, input: &CompareRetryInput) -> Result<(i64, bool)> {
-	let now = util::timestamp::now();
+	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
-	Ok((now, input.last_retry_ts < now - RETRY_RESET_DURATION_MS))
+	let reset = input.last_retry_ts < now - RETRY_RESET_DURATION_MS;
+
+	if reset {
+		state.reschedule_ts = None;
+	} else {
+		let backoff = reschedule_backoff(input.retry_count);
+		state.reschedule_ts = Some(now + i64::try_from(backoff.current_duration())?);
+	}
+
+	Ok((now, reset))
 }
 
 #[derive(Debug, Serialize, Deserialize, Hash)]
@@ -740,7 +747,9 @@ pub struct SetStartedInput {
 pub async fn set_started(ctx: &ActivityCtx, input: &SetStartedInput) -> Result<()> {
 	let mut state = ctx.state::<State>()?;
 
-	state.start_ts = Some(util::timestamp::now());
+	if state.start_ts.is_none() {
+		state.start_ts = Some(util::timestamp::now());
+	}
 	state.connectable_ts = Some(util::timestamp::now());
 
 	ctx.udb()?
@@ -800,3 +809,7 @@ 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)
+}

engine/packages/types/src/actors.rs

@@ -15,11 +15,20 @@ pub struct Actor {
 	pub runner_name_selector: String,
 	pub crash_policy: CrashPolicy,
 
+	/// Denotes when the actor was first created.
 	pub create_ts: i64,
+	/// Denotes when the actor was first made connectable. Null if never.
 	pub start_ts: Option<i64>,
+	/// Denotes when the actor started waiting for an allocation.
 	pub pending_allocation_ts: Option<i64>,
+	/// Denotes when the actor was last connectable. Null if actor is not running.
 	pub connectable_ts: Option<i64>,
+	/// Denotes when the actor entered a sleeping state.
 	pub sleep_ts: Option<i64>,
+	/// Denotes when the actor will try to allocate again. If this is set, the actor will not attempt to
+	/// allocate until the given timestamp.
+	pub reschedule_ts: Option<i64>,
+	/// Denotes when the actor was destroyed.
 	pub destroy_ts: Option<i64>,
 }
 

engine/packages/util/src/backoff.rs

@@ -85,8 +85,7 @@ impl Backoff {
 			return None;
 		}
 
-		let next_wait = self.wait * 2usize.pow(self.i.min(self.max_exponent) as u32)
-			+ rand::thread_rng().gen_range(0..self.randomness);
+		let next_wait = self.current_duration() + rand::thread_rng().gen_range(0..self.randomness);
 		self.sleep_until += Duration::from_millis(next_wait as u64);
 
 		self.i += 1;