BREAK THIS UP: WIP patches

Author: NathanFlurry

Cargo.toml

@@ -57,7 +57,8 @@ members = [
   "engine/sdks/rust/epoxy-protocol",
   "engine/sdks/rust/test-envoy",
   "engine/sdks/rust/ups-protocol",
-  "rivetkit-typescript/packages/rivetkit-native"
+  "rivetkit-typescript/packages/rivetkit-native",
+  "rivetkit-typescript/packages/sqlite-native"
 ]
 
   [workspace.package]

engine/packages/api-peer/src/namespaces.rs

@@ -97,15 +97,6 @@ pub async fn create(
 ) -> Result<CreateResponse> {
 	let namespace_id = Id::new_v1(ctx.config().dc_label());
 
-	ctx.workflow(namespace::workflows::namespace::Input {
-		namespace_id,
-		name: body.name.clone(),
-		display_name: body.display_name.clone(),
-	})
-	.tag("namespace_id", namespace_id)
-	.dispatch()
-	.await?;
-
 	let mut create_sub = ctx
 		.subscribe::<namespace::workflows::namespace::CreateComplete>((
 			"namespace_id",
@@ -116,6 +107,15 @@ pub async fn create(
 		.subscribe::<namespace::workflows::namespace::Failed>(("namespace_id", namespace_id))
 		.await?;
 
+	ctx.workflow(namespace::workflows::namespace::Input {
+		namespace_id,
+		name: body.name.clone(),
+		display_name: body.display_name.clone(),
+	})
+	.tag("namespace_id", namespace_id)
+	.dispatch()
+	.await?;
+
 	tokio::select! {
 		res = create_sub.next() => { res?; },
 		res = fail_sub.next() => {

engine/packages/guard/src/routing/pegboard_gateway/resolve_actor_query.rs

@@ -208,7 +208,8 @@ async fn resolve_query_target_dc_label(
 
 fn serialize_actor_key(key: &[String]) -> Result<String> {
 	const EMPTY_KEY: &str = "/";
-	const KEY_SEPARATOR: char = '/';
+	const KEY_SEPARATOR: &str = "/";
+	const KEY_SEPARATOR_CHAR: char = '/';
 
 	if key.is_empty() {
 		return Ok(EMPTY_KEY.to_string());
@@ -221,11 +222,13 @@ fn serialize_actor_key(key: &[String]) -> Result<String> {
 			continue;
 		}
 
-		let escaped = part.replace('\\', "\\\\").replace(KEY_SEPARATOR, "\\/");
+		let escaped = part
+			.replace('\\', "\\\\")
+			.replace(KEY_SEPARATOR_CHAR, "\\/");
 		escaped_parts.push(escaped);
 	}
 
-	Ok(escaped_parts.join(EMPTY_KEY))
+	Ok(escaped_parts.join(KEY_SEPARATOR))
 }
 
 fn is_duplicate_key_error(err: &anyhow::Error) -> bool {

engine/packages/pegboard-gateway/src/keepalive_task.rs

@@ -20,13 +20,22 @@ pub async fn task(
 	request_id: protocol::RequestId,
 	mut keepalive_abort_rx: watch::Receiver<()>,
 ) -> Result<LifecycleResult> {
-	let mut ping_interval = tokio::time::interval(Duration::from_millis(
-		(ctx.config()
-			.pegboard()
-			.hibernating_request_eligible_threshold()
-			/ 2)
-		.try_into()?,
-	));
+	ctx.op(pegboard::ops::actor::hibernating_request::upsert::Input {
+		actor_id,
+		gateway_id,
+		request_id,
+	})
+	.await?;
+	shared_state.keepalive_hws(request_id).await?;
+
+	let ping_interval_ms = (ctx
+		.config()
+		.pegboard()
+		.hibernating_request_eligible_threshold()
+		/ 2)
+	.max(1);
+	let mut ping_interval =
+		tokio::time::interval(Duration::from_millis(ping_interval_ms.try_into()?));
 	ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
 	loop {

engine/packages/pegboard-gateway/src/lib.rs

@@ -826,7 +826,7 @@ impl PegboardGateway {
 			})
 			.await?
 		{
-			if actor.runner_id.is_some() {
+			if !actor.sleeping && actor.runner_id.is_some() {
 				tracing::debug!("actor became ready during hibernation");
 
 				return Ok(HibernationResult::Continue);

engine/packages/pegboard-gateway/src/shared_state.rs

@@ -151,7 +151,7 @@ impl SharedState {
 			gateway_id,
 			receiver_subject,
 			in_flight_requests: HashMap::new(),
-			hibernation_timeout: pegboard_config.hibernating_request_eligible_threshold(),
+			hibernation_timeout: pegboard_config.hibernating_request_eligible_threshold().max(1),
 			gc_interval: Duration::from_millis(pegboard_config.gateway_gc_interval_ms()),
 			tunnel_ping_timeout: pegboard_config.gateway_tunnel_ping_timeout_ms(),
 			hws_message_ack_timeout: Duration::from_millis(

engine/packages/pegboard-gateway2/src/keepalive_task.rs

@@ -20,13 +20,22 @@ pub async fn task(
 	request_id: protocol::RequestId,
 	mut keepalive_abort_rx: watch::Receiver<()>,
 ) -> Result<LifecycleResult> {
-	let mut ping_interval = tokio::time::interval(Duration::from_millis(
-		(ctx.config()
-			.pegboard()
-			.hibernating_request_eligible_threshold()
-			/ 2)
-		.try_into()?,
-	));
+	ctx.op(pegboard::ops::actor::hibernating_request::upsert::Input {
+		actor_id,
+		gateway_id,
+		request_id,
+	})
+	.await?;
+	shared_state.keepalive_hws(request_id).await?;
+
+	let ping_interval_ms = (ctx
+		.config()
+		.pegboard()
+		.hibernating_request_eligible_threshold()
+		/ 2)
+	.max(1);
+	let mut ping_interval =
+		tokio::time::interval(Duration::from_millis(ping_interval_ms.try_into()?));
 	ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
 	loop {

engine/packages/pegboard-gateway2/src/lib.rs

@@ -832,7 +832,7 @@ impl PegboardGateway2 {
 			})
 			.await?
 		{
-			if actor.envoy_key.is_some() {
+			if !actor.sleeping && actor.envoy_key.is_some() {
 				tracing::debug!("actor became ready during hibernation");
 
 				return Ok(HibernationResult::Continue);

engine/packages/pegboard-gateway2/src/shared_state.rs

@@ -98,7 +98,7 @@ impl SharedState {
 			gateway_id,
 			receiver_subject,
 			in_flight_requests: HashMap::new(),
-			hibernation_timeout: pegboard_config.hibernating_request_eligible_threshold(),
+			hibernation_timeout: pegboard_config.hibernating_request_eligible_threshold().max(1),
 			gc_interval: Duration::from_millis(pegboard_config.gateway_gc_interval_ms()),
 			tunnel_ping_timeout: pegboard_config.gateway_tunnel_ping_timeout_ms(),
 			hws_message_ack_timeout: Duration::from_millis(

engine/packages/pegboard/src/ops/actor/hibernating_request/delete.rs

@@ -16,6 +16,13 @@ pub async fn pegboard_actor_hibernating_request_delete(
 	ctx: &OperationCtx,
 	input: &Input,
 ) -> Result<()> {
+	tracing::info!(
+		actor_id=%input.actor_id,
+		gateway_id=%protocol::util::id_to_string(&input.gateway_id),
+		request_id=%protocol::util::id_to_string(&input.request_id),
+		"deleting hibernating request"
+	);
+
 	ctx.udb()?
 		.run(|tx| async move {
 			let tx = tx.with_subspace(keys::subspace());