fix(pegboard): restore hibernating request ids

Author: NathanFlurry

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

@@ -358,6 +358,17 @@ pub async fn send_outbound(ctx: &ActivityCtx, input: &SendOutboundInput) -> Resu
 				.await?;
 		}
 		Allocation::Serverful { envoy_key } => {
+			let hibernating_requests = ctx
+				.op(crate::ops::actor::hibernating_request::list::Input {
+					actor_id: state.actor_id,
+				})
+				.await?
+				.into_iter()
+				.map(|request| protocol::HibernatingRequest {
+					gateway_id: request.gateway_id,
+					request_id: request.request_id,
+				})
+				.collect();
 			let command = protocol::Command::CommandStartActor(protocol::CommandStartActor {
 				config: protocol::ActorConfig {
 					name: state.name.clone(),
@@ -368,9 +379,7 @@ pub async fn send_outbound(ctx: &ActivityCtx, input: &SendOutboundInput) -> Resu
 						.as_ref()
 						.and_then(|x| BASE64_STANDARD.decode(x).ok()),
 				},
-				// Empty because request ids are ephemeral. This is intercepted by guard and
-				// populated before it reaches the runner
-				hibernating_requests: Vec::new(),
+				hibernating_requests,
 				preloaded_kv: None,
 			});
 

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

@@ -1314,6 +1314,10 @@ impl ActorContext {
 		self.0.sleep_requested.load(Ordering::SeqCst)
 	}
 
+	pub(crate) fn clear_sleep_requested(&self) {
+		self.0.sleep_requested.store(false, Ordering::SeqCst);
+	}
+
 	fn keep_awake_guard(&self) -> KeepAwakeGuard {
 		let region = self
 			.keep_awake_region()

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

@@ -2141,6 +2141,9 @@ impl ActorTask {
 			"actor lifecycle transition"
 		);
 		self.lifecycle = lifecycle;
+		if matches!(lifecycle, LifecycleState::Started) {
+			self.ctx.clear_sleep_requested();
+		}
 		self.ctx
 			.set_started(matches!(lifecycle, LifecycleState::Started));
 	}

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

@@ -1207,6 +1207,76 @@ mod moved_tests {
 		assert_eq!(remaining_conns[0].id(), hibernating_conn.id());
 	}
 
+	#[tokio::test]
+	async fn wake_start_clears_previous_sleep_request() {
+		let kv = new_in_memory();
+		let ctx = new_with_kv("actor-wake", "task-wake", Vec::new(), "local", kv.clone());
+		let (_lifecycle_tx, lifecycle_rx) = mpsc::channel(4);
+		let (_dispatch_tx, dispatch_rx) = mpsc::channel(4);
+		let (events_tx, events_rx) = mpsc::channel(4);
+		ctx.configure_lifecycle_events(Some(events_tx));
+
+		let factory = Arc::new(ActorFactory::new(
+			ActorConfig {
+				sleep_grace_period: Duration::from_millis(200),
+				sleep_grace_period_overridden: true,
+				..ActorConfig::default()
+			},
+			|start| {
+				Box::pin(async move {
+					let mut events = start.events;
+					while let Some(event) = events.recv().await {
+						match event {
+							ActorEvent::SerializeState { reply, .. } => {
+								reply.send(Ok(Vec::new()));
+							}
+							ActorEvent::RunGracefulCleanup { reply, .. } => {
+								reply.send(Ok(()));
+							}
+							_ => {}
+						}
+					}
+					Ok(())
+				})
+			},
+		));
+
+		let mut task = ActorTask::new(
+			"actor-wake".into(),
+			0,
+			lifecycle_rx,
+			dispatch_rx,
+			events_rx,
+			factory,
+			ctx.clone(),
+			None,
+			None,
+		);
+		let (start_tx, start_rx) = oneshot::channel();
+		task.handle_lifecycle(LifecycleCommand::Start { reply: start_tx })
+			.await;
+		start_rx
+			.await
+			.expect("start reply should send")
+			.expect("start should succeed");
+
+		ctx.sleep().expect("sleep request should succeed");
+		assert!(ctx.sleep_requested());
+
+		task.handle_stop(ShutdownKind::Sleep)
+			.await
+			.expect("sleep stop should succeed");
+		let (wake_tx, wake_rx) = oneshot::channel();
+		task.handle_lifecycle(LifecycleCommand::Start { reply: wake_tx })
+			.await;
+		wake_rx
+			.await
+			.expect("wake reply should send")
+			.expect("wake should succeed");
+
+		assert!(!ctx.sleep_requested());
+	}
+
 	#[tokio::test]
 	async fn sleep_shutdown_waits_for_on_state_change_before_final_save() {
 		let kv = new_in_memory();