fix(rivetkit-core): error on sleep/destroy before startup or already-requested

Author: NathanFlurry

rivetkit-rust/packages/rivetkit-core/src/actor/context.rs

@@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering}
 use std::sync::{Arc, OnceLock};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
-use anyhow::Result;
+use anyhow::{Context as AnyhowContext, Result};
 use futures::future::BoxFuture;
 use parking_lot::{Mutex, RwLock};
 use rivet_envoy_client::handle::EnvoyHandle;
@@ -36,7 +36,7 @@ use crate::actor::task::{
 };
 use crate::actor::task_types::UserTaskKind;
 use crate::actor::work_registry::RegionGuard;
-use crate::error::ActorRuntime;
+use crate::error::{ActorLifecycle as ActorLifecycleError, ActorRuntime};
 use crate::inspector::{Inspector, InspectorSnapshot};
 use crate::kv::Kv;
 use crate::sqlite::SqliteDb;
@@ -384,9 +384,16 @@ impl ActorContext {
 		self
 	}
 
-	pub fn sleep(&self) {
+	pub fn sleep(&self) -> Result<()> {
+		if !self.0.sleep.started.load(Ordering::SeqCst) {
+			return Err(ActorLifecycleError::Starting.build())
+				.context("cannot request sleep before actor startup completes");
+		}
+		if self.0.sleep_requested.swap(true, Ordering::SeqCst) {
+			return Err(ActorLifecycleError::Stopping.build())
+				.context("sleep already requested for this generation");
+		}
 		self.cancel_sleep_timer();
-		self.0.sleep_requested.store(true, Ordering::SeqCst);
 		if Handle::try_current().is_ok() {
 			let ctx = self.clone();
 			let tracked = self.track_shutdown_task(async move {
@@ -396,15 +403,27 @@ impl ActorContext {
 				ctx.record_user_task_finished(UserTaskKind::SleepFinalize, started_at.elapsed());
 			});
 			if tracked {
-				return;
+				return Ok(());
 			}
 		}
 
 		self.request_sleep_from_envoy();
+		Ok(())
 	}
 
-	pub fn destroy(&self) {
-		self.mark_destroy_requested();
+	pub fn destroy(&self) -> Result<()> {
+		if !self.0.sleep.started.load(Ordering::SeqCst) {
+			return Err(ActorLifecycleError::Starting.build())
+				.context("cannot request destroy before actor startup completes");
+		}
+		if self.0.destroy_requested.swap(true, Ordering::SeqCst) {
+			return Err(ActorLifecycleError::Stopping.build())
+				.context("destroy already requested for this generation");
+		}
+		self.cancel_sleep_timer();
+		self.flush_on_shutdown();
+		self.0.destroy_completed.store(false, Ordering::SeqCst);
+		self.0.abort_signal.cancel();
 
 		let ctx = self.clone();
 		if Handle::try_current().is_ok() {
@@ -415,11 +434,12 @@ impl ActorContext {
 				ctx.record_user_task_finished(UserTaskKind::DestroyRequest, started_at.elapsed());
 			});
 			if tracked {
-				return;
+				return Ok(());
 			}
 		}
 
 		self.request_destroy_from_envoy();
+		Ok(())
 	}
 
 	pub fn mark_destroy_requested(&self) {

rivetkit-rust/packages/rivetkit-core/src/actor/sleep.rs

@@ -312,7 +312,13 @@ impl ActorContext {
 					sleep_timeout_ms = timeout.as_millis() as u64,
 					"sleep idle timer elapsed"
 				);
-				ctx.sleep();
+				if let Err(err) = ctx.sleep() {
+					tracing::debug!(
+						actor_id = %ctx.actor_id(),
+						?err,
+						"sleep idle timer request suppressed"
+					);
+				}
 			} else {
 				tracing::warn!(
 					actor_id = %ctx.actor_id(),
@@ -713,9 +719,11 @@ mod tests {
 	#[tokio::test(start_paused = true)]
 	async fn sleep_then_destroy_signal_tasks_do_not_leak_after_teardown() {
 		let ctx = ActorContext::new_for_sleep_tests("actor-sleep-destroy");
+		ctx.set_sleep_started(true);
 
-		ctx.sleep();
-		ctx.destroy();
+		ctx.sleep().expect("sleep should be accepted after startup");
+		ctx.destroy()
+			.expect("destroy should be accepted after startup");
 
 		assert_eq!(
 			ctx.shutdown_task_count(),

rivetkit-rust/packages/rivetkit-core/src/actor/task.rs

@@ -1952,7 +1952,13 @@ impl ActorTask {
 				sleep_timeout_ms = self.factory.config().sleep_timeout.as_millis() as u64,
 				"sleep idle deadline elapsed"
 			);
-			self.ctx.sleep();
+			if let Err(err) = self.ctx.sleep() {
+				tracing::debug!(
+					actor_id = %self.ctx.actor_id(),
+					?err,
+					"sleep idle deadline request suppressed"
+				);
+			}
 		} else {
 			tracing::warn!(
 				actor_id = %self.ctx.actor_id(),

rivetkit-rust/packages/rivetkit-core/tests/modules/context.rs

@@ -814,9 +814,11 @@ mod moved_tests {
 			crate::kv::Kv::new_in_memory(),
 		);
 
+		ctx.set_sleep_started(true);
+
 		assert_eq!(ctx.sleep_request_count(), 0);
 
-		ctx.sleep();
+		ctx.sleep().expect("sleep should be accepted after startup");
 		tokio::task::yield_now().await;
 
 		assert_eq!(ctx.sleep_request_count(), 1);

rivetkit-rust/packages/rivetkit-core/tests/modules/task.rs

@@ -2986,7 +2986,7 @@ mod moved_tests {
 		}));
 		let factory = Arc::new(ActorFactory::new(ActorConfig::default(), move |start| {
 			Box::pin(async move {
-				start.ctx.sleep();
+				start.ctx.sleep()?;
 				Ok(())
 			})
 		}));
@@ -3053,7 +3053,7 @@ mod moved_tests {
 		}));
 		let factory = Arc::new(ActorFactory::new(ActorConfig::default(), move |start| {
 			Box::pin(async move {
-				start.ctx.destroy();
+				start.ctx.destroy()?;
 				Ok(())
 			})
 		}));

rivetkit-rust/packages/rivetkit/src/context.rs

@@ -87,12 +87,12 @@ impl<A: Actor> Ctx<A> {
 		self.inner.save_state(deltas).await
 	}
 
-	pub fn sleep(&self) {
-		self.inner.sleep();
+	pub fn sleep(&self) -> Result<()> {
+		self.inner.sleep()
 	}
 
-	pub fn destroy(&self) {
-		self.inner.destroy();
+	pub fn destroy(&self) -> Result<()> {
+		self.inner.destroy()
 	}
 
 	#[deprecated(note = "no-op: use `keep_awake` or `wait_until` instead")]

rivetkit-typescript/packages/rivetkit-napi/src/actor_context.rs

@@ -431,13 +431,13 @@ impl ActorContext {
 	}
 
 	#[napi]
-	pub fn sleep(&self) {
-		self.inner.sleep();
+	pub fn sleep(&self) -> napi::Result<()> {
+		self.inner.sleep().map_err(napi_anyhow_error)
 	}
 
 	#[napi]
-	pub fn destroy(&self) {
-		self.inner.destroy();
+	pub fn destroy(&self) -> napi::Result<()> {
+		self.inner.destroy().map_err(napi_anyhow_error)
 	}
 
 	#[napi]