fix: dont purge cache for actor keys (#3204)

* fix(api-public): do not pass user headers to forward to dc

* chore(guard): log runner routing

* fix: dont purge cache for actor keys

Author: NathanFlurry

packages/common/api-util/src/lib.rs

@@ -2,7 +2,7 @@ use anyhow::{Context, Result};
 use axum::{body::Body, response::Response};
 use futures_util::StreamExt;
 use rivet_api_builder::{ApiCtx, ErrorResponse, RawErrorResponse};
-use serde::{de::DeserializeOwned, Serialize};
+use serde::{Serialize, de::DeserializeOwned};
 use std::future::Future;
 
 mod errors;

packages/common/runtime/src/traces.rs

@@ -32,10 +32,7 @@ pub fn init_tracing_subscriber(otel_providers: &Option<OtelProviderGuard>) {
 			let otel_metric_layer = MetricsLayer::new(providers.meter_provider.clone())
 				.with_filter(build_filter_from_env_var("RUST_TRACE"));
 
-			(
-				Some(otel_trace_layer),
-				Some(otel_metric_layer),
-			)
+			(Some(otel_trace_layer), Some(otel_metric_layer))
 		}
 		None => (None, None),
 	};

packages/core/api-public/src/health.rs

@@ -1,5 +1,5 @@
 use anyhow::Result;
-use axum::{extract::Extension, response::IntoResponse, Json};
+use axum::{Json, extract::Extension, response::IntoResponse};
 use futures_util::StreamExt;
 use rivet_api_builder::ApiError;
 use serde::{Deserialize, Serialize};

packages/core/api-public/src/router.rs

@@ -3,12 +3,14 @@ use axum::{
 	middleware::{self, Next},
 	response::{IntoResponse, Redirect, Response},
 };
-use reqwest::header::{HeaderMap, AUTHORIZATION};
+use reqwest::header::{AUTHORIZATION, HeaderMap};
 use rivet_api_builder::{create_router, extract::FailedExtraction};
 use tower_http::cors::CorsLayer;
 use utoipa::OpenApi;
 
-use crate::{actors, ctx, datacenters, health, metadata, namespaces, runner_configs, runners, ui};
+use crate::{
+	actors, ctx, datacenters, health, metadata, namespaces, runner_configs, runners, ui,
+};
 
 #[derive(OpenApi)]
 #[openapi(

packages/services/epoxy/src/ops/explicit_prepare.rs

@@ -79,9 +79,15 @@ pub async fn epoxy_explicit_prepare(
 	let result = match analyze_prepare_responses(&highest_ballot_responses, instance) {
 		PrepareDecision::Commit(payload) => {
 			// EPaxos Step 29: Run Commit phase
-			let result =
-				crate::ops::propose::commit(ctx, &config, replica_id, &quorum_members, payload)
-					.await?;
+			let result = crate::ops::propose::commit(
+				ctx,
+				&config,
+				replica_id,
+				&quorum_members,
+				payload,
+				false,
+			)
+			.await?;
 			convert_proposal_result(result)
 		}
 		PrepareDecision::Accept(payload) => {
@@ -92,6 +98,7 @@ pub async fn epoxy_explicit_prepare(
 				replica_id,
 				&quorum_members,
 				payload,
+				false,
 			)
 			.await?;
 			convert_proposal_result(result)
@@ -324,7 +331,12 @@ async fn restart_phase1(
 	);
 
 	// Call the propose operation to restart consensus from Phase 1
-	let result = ctx.op(crate::ops::propose::Input { proposal }).await?;
+	let result = ctx
+		.op(crate::ops::propose::Input {
+			proposal,
+			purge_cache: false,
+		})
+		.await?;
 
 	// Convert ProposalResult to ExplicitPrepareResult
 	match result {

packages/services/epoxy/src/ops/propose.rs

@@ -24,6 +24,8 @@ pub enum CommandError {
 #[derive(Debug)]
 pub struct Input {
 	pub proposal: protocol::Proposal,
+	/// Only works in non-workflow contexts.
+	pub purge_cache: bool,
 }
 
 #[operation]
@@ -67,10 +69,26 @@ pub async fn epoxy_propose(ctx: &OperationCtx, input: &Input) -> Result<Proposal
 
 	match path {
 		Path::PathFast(protocol::PathFast { payload }) => {
-			commit(ctx, &config, replica_id, &quorum_members, payload).await
+			commit(
+				ctx,
+				&config,
+				replica_id,
+				&quorum_members,
+				payload,
+				input.purge_cache,
+			)
+			.await
 		}
 		Path::PathSlow(protocol::PathSlow { payload }) => {
-			run_paxos_accept(ctx, &config, replica_id, &quorum_members, payload).await
+			run_paxos_accept(
+				ctx,
+				&config,
+				replica_id,
+				&quorum_members,
+				payload,
+				input.purge_cache,
+			)
+			.await
 		}
 	}
 }
@@ -82,6 +100,7 @@ pub async fn run_paxos_accept(
 	replica_id: ReplicaId,
 	quorum_members: &[ReplicaId],
 	payload: Payload,
+	purge_cache: bool,
 ) -> Result<ProposalResult> {
 	// Clone payload for use after the closure
 	let payload_for_accepts = payload.clone();
@@ -113,6 +132,7 @@ pub async fn run_paxos_accept(
 			replica_id,
 			&quorum_members,
 			payload_for_accepts,
+			purge_cache,
 		)
 		.await
 	} else {
@@ -127,6 +147,7 @@ pub async fn commit(
 	replica_id: ReplicaId,
 	quorum_members: &[ReplicaId],
 	payload: Payload,
+	purge_cache: bool,
 ) -> Result<ProposalResult> {
 	// Commit locally
 	//

packages/services/epoxy/tests/common/utils.rs

@@ -14,6 +14,7 @@ pub async fn execute_command(
 			proposal: protocol::Proposal {
 				commands: vec![protocol::Command { kind: command }],
 			},
+			purge_cache: false,
 		})
 		.await?;
 

packages/services/epoxy/tests/proposal.rs

@@ -28,6 +28,7 @@ async fn proposal() {
 					}),
 				}],
 			},
+			purge_cache: false,
 		})
 		.await
 		.unwrap();

packages/services/pegboard/src/workflows/actor/actor_keys.rs

@@ -203,6 +203,7 @@ pub async fn propose(ctx: &ActivityCtx, input: &ProposeInput) -> Result<Proposal
 					}),
 				}],
 			},
+			purge_cache: false,
 		})
 		.await?;