feat: make actors reschedule indefinitely with backoff

packages/common/chirp-workflow/core/src/utils.rs

@@ -50,12 +50,11 @@ pub mod time {
 	impl TsToMillis for Instant {
 		fn to_millis(self) -> GlobalResult<i64> {
 			let now_instant = Instant::now();
-			let now_system_time = SystemTime::now();
 
 			let system_time = if self >= now_instant {
-				now_system_time.checked_add(self.duration_since(now_instant))
+				SystemTime::now().checked_add(self.duration_since(now_instant))
 			} else {
-				now_system_time.checked_sub(now_instant.duration_since(self))
+				SystemTime::now().checked_sub(now_instant.duration_since(self))
 			};
 
 			let ms = unwrap!(system_time, "invalid timestamp")
@@ -67,6 +66,12 @@ pub mod time {
 		}
 	}
 
+	impl TsToMillis for tokio::time::Instant {
+		fn to_millis(self) -> GlobalResult<i64> {
+			self.into_std().to_millis()
+		}
+	}
+
 	impl TsToMillis for SystemTime {
 		fn to_millis(self) -> GlobalResult<i64> {
 			let ms = self

packages/services/ds/db/servers/migrations/20241115184305_add_durable.up.sql

@@ -1,3 +1,5 @@
 ALTER TABLE servers
 	RENAME COLUMN kill_timeout_ms TO lifecycle_kill_timeout_ms,
-	ADD COLUMN lifecycle_durable BOOLEAN DEFAULT false;
+	ADD COLUMN lifecycle_durable BOOLEAN NOT NULL DEFAULT false,
+	ADD COLUMN reschedule_retry_count INT NOT NULL DEFAULT 0,
+	ADD COLUMN last_reschedule_retry_ts INT;

packages/services/ds/src/workflows/server/mod.rs

@@ -542,7 +542,7 @@ struct GetServerMetaInput {
 	image_id: Uuid,
 }
 
-#[derive(Debug, Serialize, Deserialize, Hash)]
+#[derive(Clone, Debug, Serialize, Deserialize, Hash)]
 struct GetServerMetaOutput {
 	project_id: Uuid,
 	project_slug: String,
@@ -582,7 +582,8 @@ async fn get_server_meta(
 	let project_id = unwrap!(env.game_id).as_uuid();
 	let projects_res = op!([ctx] game_get {
 		game_ids: vec![project_id.into()],
-	}).await?;
+	})
+	.await?;
 	let project = unwrap!(projects_res.games.first());
 
 	Ok(GetServerMetaOutput {
@@ -646,6 +647,37 @@ async fn update_image(ctx: &ActivityCtx, input: &UpdateImageInput) -> GlobalResu
 	Ok(())
 }
 
+#[derive(Debug, Serialize, Deserialize, Hash)]
+struct UpdateRescheduleRetryInput {
+	server_id: Uuid,
+	reset: bool,
+}
+
+#[activity(UpdateRescheduleRetry)]
+async fn update_reschedule_retry(
+	ctx: &ActivityCtx,
+	input: &UpdateRescheduleRetryInput,
+) -> GlobalResult<i64> {
+	let (retry_count,) = sql_fetch_one!(
+		[ctx, (i64,)]
+		"
+		UPDATE db_ds.servers
+		SET
+			reschedule_retry_count = COALESCE($2, reschedule_retry_count + 1),
+			last_reschedule_retry_ts = COALESCE($3, last_reschedule_retry_ts)
+		WHERE server_id = $1
+		-- Return value before update
+		RETURNING reschedule_retry_count - 1
+		",
+		input.server_id,
+		input.reset.then_some(0),
+		(!input.reset).then(util::timestamp::now),
+	)
+	.await?;
+
+	Ok(retry_count)
+}
+
 #[message("ds_server_create_complete")]
 pub struct CreateComplete {}