chore: do not merge: disable reschedule backoff

Author: MasterPtato

.gitattributes

@@ -12,6 +12,8 @@
 # Prevent from counting in the language statistics
 engine/artifacts/** linguist-generated=true
 engine/sdks/** linguist-generated=true
+engine/sdks/typescript/runner/** linguist-generated=false
+engine/sdks/typescript/test-runner/** linguist-generated=false
 engine/sdks/schema/** linguist-generated=false
 
 website/public/llms.txt linguist-generated=true

Cargo.lock

@@ -1,8 +1,5 @@
 use anyhow::Result;
-use axum::{
-	http::HeaderMap,
-	response::{IntoResponse, Response},
-};
+use axum::response::{IntoResponse, Response};
 use rivet_api_builder::{
 	ApiError,
 	extract::{Extension, Json, Query},
@@ -77,11 +74,10 @@ pub struct GetOrCreateResponse {
 )]
 pub async fn get_or_create(
 	Extension(ctx): Extension<ApiCtx>,
-	headers: HeaderMap,
 	Query(query): Query<GetOrCreateQuery>,
 	Json(body): Json<GetOrCreateRequest>,
 ) -> Response {
-	match get_or_create_inner(ctx, headers, query, body).await {
+	match get_or_create_inner(ctx, query, body).await {
 		Ok(response) => Json(response).into_response(),
 		Err(err) => ApiError::from(err).into_response(),
 	}
@@ -90,7 +86,6 @@ pub async fn get_or_create(
 #[tracing::instrument(skip_all)]
 async fn get_or_create_inner(
 	ctx: ApiCtx,
-	headers: HeaderMap,
 	query: GetOrCreateQuery,
 	body: GetOrCreateRequest,
 ) -> Result<GetOrCreateResponse> {

engine/packages/api-public/src/actors/get_or_create.rs

@@ -87,7 +87,7 @@ async fn fanout_inner(ctx: ApiCtx) -> Result<FanoutResponse> {
 				}
 			} else {
 				// Remote datacenter - HTTP request
-				match send_health_checks(&ctx, &dc).await {
+				match send_health_checks(&dc).await {
 					Ok(response) => DatacenterHealth {
 						datacenter_label: dc.datacenter_label,
 						datacenter_name: dc.name.clone(),
@@ -129,7 +129,6 @@ async fn fanout_inner(ctx: ApiCtx) -> Result<FanoutResponse> {
 
 #[tracing::instrument(skip_all)]
 async fn send_health_checks(
-	ctx: &ApiCtx,
 	dc: &rivet_config::config::topology::Datacenter,
 ) -> Result<HealthResponse> {
 	let client = rivet_pools::reqwest::client().await?;

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

@@ -114,7 +114,7 @@ pub async fn fetch_serverless_runner_metadata(
 		});
 	}
 
-	let payload = serde_json::from_str::<ServerlessMetadataPayload>(&body_raw).map_err(|err| {
+	let payload = serde_json::from_str::<ServerlessMetadataPayload>(&body_raw).map_err(|_| {
 		ServerlessMetadataError::InvalidResponseJson {
 			body: body_for_user,
 		}

engine/packages/api-public/src/runner_configs/utils.rs

@@ -79,15 +79,8 @@ 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,
-				false,
-			)
-			.await?;
+			let result =
+				crate::ops::propose::commit(ctx, &config, replica_id, payload, false).await?;
 			convert_proposal_result(result)
 		}
 		PrepareDecision::Accept(payload) => {

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

@@ -1,2 +1,3 @@
 pub mod get_local;
 pub mod get_optimistic;
+pub mod purge_local;

engine/packages/epoxy/src/ops/kv/mod.rs

@@ -0,0 +1,28 @@
+use anyhow::*;
+use epoxy_protocol::protocol::ReplicaId;
+use gas::prelude::*;
+
+use crate::keys;
+
+#[derive(Debug)]
+pub struct Input {
+	pub replica_id: ReplicaId,
+	pub keys: Vec<Vec<u8>>,
+}
+
+#[operation]
+pub async fn epoxy_kv_purge_local(ctx: &OperationCtx, input: &Input) -> Result<()> {
+	ctx.udb()?
+		.run(|tx| async move {
+			let tx = tx.with_subspace(keys::subspace(input.replica_id));
+
+			for key in &input.keys {
+				tx.delete(&keys::keys::KvOptimisticCacheKey::new(key.clone()));
+			}
+
+			Ok(())
+		})
+		.await?;
+
+	Ok(())
+}

engine/packages/epoxy/src/ops/kv/purge_local.rs

@@ -1,4 +1,6 @@
 use anyhow::*;
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
 use epoxy_protocol::protocol::{self, Path, Payload, ReplicaId};
 use gas::prelude::*;
 use rivet_api_builder::prelude::*;
@@ -68,15 +70,7 @@ 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,
-				input.purge_cache,
-			)
-			.await
+			commit(ctx, &config, replica_id, payload, input.purge_cache).await
 		}
 		Path::PathSlow(protocol::PathSlow { payload }) => {
 			run_paxos_accept(
@@ -125,15 +119,7 @@ pub async fn run_paxos_accept(
 
 	// EPaxos Step 20
 	if quorum >= utils::calculate_quorum(quorum_members.len(), utils::QuorumType::Slow) {
-		commit(
-			ctx,
-			&config,
-			replica_id,
-			&quorum_members,
-			payload_for_accepts,
-			purge_cache,
-		)
-		.await
+		commit(ctx, &config, replica_id, payload_for_accepts, purge_cache).await
 	} else {
 		Ok(ProposalResult::ConsensusFailed)
 	}
@@ -144,7 +130,6 @@ pub async fn commit(
 	ctx: &OperationCtx,
 	config: &protocol::ClusterConfig,
 	replica_id: ReplicaId,
-	quorum_members: &[ReplicaId],
 	payload: Payload,
 	purge_cache: bool,
 ) -> Result<ProposalResult> {
@@ -183,6 +168,27 @@ pub async fn commit(
 		}
 	});
 
+	if purge_cache {
+		let keys = payload
+			.proposal
+			.commands
+			.iter()
+			.map(replica::utils::extract_key_from_command)
+			.flatten()
+			.map(|key| BASE64.encode(key))
+			.collect::<Vec<_>>();
+
+		// Purge optimistic cache for all dcs
+		if !keys.is_empty() {
+			let ctx = ctx.clone();
+			tokio::spawn(async move {
+				if let Err(err) = purge_optimistic_cache(ctx, keys).await {
+					tracing::error!(?err, "failed purging optimistic cache");
+				}
+			});
+		}
+	}
+
 	if let Some(cmd_err) = cmd_err {
 		Ok(ProposalResult::CommandError(cmd_err))
 	} else {
@@ -325,3 +331,22 @@ async fn send_commits(
 
 	Ok(())
 }
+
+async fn purge_optimistic_cache(ctx: OperationCtx, keys: Vec<String>) -> Result<()> {
+	for dc in &ctx.config().topology().datacenters {
+		let workflow_id = ctx
+			.workflow(crate::workflows::purger::Input {
+				replica_id: dc.datacenter_label as u64,
+			})
+			.tag("replica_id", dc.datacenter_label as u64)
+			.unique()
+			.dispatch()
+			.await?;
+		ctx.signal(crate::workflows::purger::Purge { keys: keys.clone() })
+			.to_workflow_id(workflow_id)
+			.send()
+			.await?;
+	}
+
+	Ok(())
+}

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

@@ -150,6 +150,16 @@ pub async fn message_request(
 				value: result.value,
 			})
 		}
+		protocol::RequestKind::KvPurgeRequest(req) => {
+			// Handle KV purge request
+			ctx.op(ops::kv::purge_local::Input {
+				replica_id: current_replica_id,
+				keys: req.keys.clone(),
+			})
+			.await?;
+
+			protocol::ResponseKind::KvPurgeResponse
+		}
 	};
 
 	Ok(protocol::Response { kind })