chore(engine): add support for epoxy debugging

Author: NathanFlurry

Cargo.lock

@@ -9,26 +9,25 @@ license.workspace = true
 anyhow.workspace = true
 axum.workspace = true
 base64.workspace = true
-gas.workspace = true
 epoxy.workspace = true
-futures-util.workspace = true
+epoxy-protocol.workspace = true
+gas.workspace = true
+indexmap.workspace = true
+namespace.workspace = true
+pegboard.workspace = true
+pegboard-actor-kv.workspace = true
 rivet-api-builder.workspace = true
 rivet-api-types.workspace = true
 rivet-api-util.workspace = true
 rivet-config.workspace = true
 rivet-error.workspace = true
 rivet-pools.workspace = true
-rivet-util.workspace = true
 rivet-types.workspace = true
+rivet-util.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-indexmap.workspace = true
-
-namespace.workspace = true
-pegboard.workspace = true
-pegboard-actor-kv.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 universalpubsub.workspace = true
-uuid.workspace = true
 utoipa.workspace = true
+uuid.workspace = true

engine/packages/api-peer/Cargo.toml

@@ -1,5 +1,7 @@
 use anyhow::*;
+use epoxy_protocol::protocol::{ReplicaId, SlotId};
 use gas::prelude::*;
+use indexmap::IndexMap;
 use rivet_api_builder::ApiCtx;
 use serde::{Deserialize, Serialize};
 use universalpubsub::PublishOpts;
@@ -167,3 +169,312 @@ pub async fn set_epoxy_state(
 
 	Ok(SetEpoxyStateResponse {})
 }
+
+#[derive(Serialize, Deserialize)]
+pub struct GetEpoxyReplicaDebugResponse {
+	pub config: epoxy::types::ClusterConfig,
+	pub computed: EpoxyReplicaDebugComputed,
+	pub state: EpoxyReplicaDebugState,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct EpoxyReplicaDebugComputed {
+	pub this_replica_id: ReplicaId,
+	pub this_replica_status: Option<epoxy::types::ReplicaStatus>,
+	pub is_coordinator: bool,
+	pub quorum_members: Vec<ReplicaId>,
+	pub quorum_sizes: EpoxyQuorumSizes,
+	pub replica_counts: EpoxyReplicaCounts,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct EpoxyQuorumSizes {
+	pub fast: usize,
+	pub slow: usize,
+	pub all: usize,
+	pub any: usize,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct EpoxyReplicaCounts {
+	pub active: usize,
+	pub learning: usize,
+	pub joining: usize,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct EpoxyReplicaDebugState {
+	pub ballot: EpoxyBallot,
+	pub instance_number: u64,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct EpoxyBallot {
+	pub epoch: u64,
+	pub ballot: u64,
+	pub replica_id: ReplicaId,
+}
+
+/// Returns debug information for this epoxy replica including cluster config, computed values, and state.
+///
+/// This reads from the replica's local UDB state, which may differ from the coordinator's
+/// view if the replica hasn't been reconfigured yet.
+pub async fn get_epoxy_replica_debug(
+	ctx: ApiCtx,
+	_path: (),
+	_query: (),
+) -> Result<GetEpoxyReplicaDebugResponse> {
+	let replica_id = ctx.config().epoxy_replica_id();
+
+	let (config, ballot, instance_number) = ctx
+		.udb()?
+		.run(|tx| async move {
+			let config = epoxy::utils::read_config(&tx, replica_id).await?;
+			let ballot = epoxy::replica::ballot::get_ballot(&tx, replica_id).await?;
+			let instance_number = epoxy::utils::read_instance_number(&tx, replica_id).await?;
+			Result::Ok((config, ballot, instance_number))
+		})
+		.await?;
+
+	// Compute derived values from config
+	let this_replica_status = config
+		.replicas
+		.iter()
+		.find(|r| r.replica_id == replica_id)
+		.map(|r| epoxy::types::ReplicaStatus::from(r.status.clone()));
+	let is_coordinator = config.coordinator_replica_id == replica_id;
+
+	let quorum_members = epoxy::utils::get_quorum_members(&config);
+	let quorum_member_count = quorum_members.len();
+
+	let quorum_sizes = EpoxyQuorumSizes {
+		fast: epoxy::utils::calculate_quorum(quorum_member_count, epoxy::utils::QuorumType::Fast),
+		slow: epoxy::utils::calculate_quorum(quorum_member_count, epoxy::utils::QuorumType::Slow),
+		all: epoxy::utils::calculate_quorum(quorum_member_count, epoxy::utils::QuorumType::All),
+		any: epoxy::utils::calculate_quorum(quorum_member_count, epoxy::utils::QuorumType::Any),
+	};
+
+	let replica_counts = EpoxyReplicaCounts {
+		active: config
+			.replicas
+			.iter()
+			.filter(|r| matches!(r.status, epoxy_protocol::protocol::ReplicaStatus::Active))
+			.count(),
+		learning: config
+			.replicas
+			.iter()
+			.filter(|r| matches!(r.status, epoxy_protocol::protocol::ReplicaStatus::Learning))
+			.count(),
+		joining: config
+			.replicas
+			.iter()
+			.filter(|r| matches!(r.status, epoxy_protocol::protocol::ReplicaStatus::Joining))
+			.count(),
+	};
+
+	Ok(GetEpoxyReplicaDebugResponse {
+		config: config.into(),
+		computed: EpoxyReplicaDebugComputed {
+			this_replica_id: replica_id,
+			this_replica_status,
+			is_coordinator,
+			quorum_members,
+			quorum_sizes,
+			replica_counts,
+		},
+		state: EpoxyReplicaDebugState {
+			ballot: EpoxyBallot {
+				epoch: ballot.epoch,
+				ballot: ballot.ballot,
+				replica_id: ballot.replica_id,
+			},
+			instance_number,
+		},
+	})
+}
+
+// MARK: Key debug
+#[derive(Serialize, Deserialize, Clone)]
+pub struct EpoxyKeyDebugPath {
+	pub key: String,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct GetEpoxyKeyDebugResponse {
+	pub replica_id: ReplicaId,
+	pub key: String,
+	pub instances: Vec<EpoxyKeyInstance>,
+	pub instances_by_status: IndexMap<String, usize>,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct EpoxyKeyInstance {
+	pub replica_id: ReplicaId,
+	pub slot_id: SlotId,
+	pub log_entry: Option<EpoxyKeyLogEntry>,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct EpoxyKeyLogEntry {
+	pub status: String,
+	pub ballot: EpoxyBallot,
+	pub seq: u64,
+	pub deps: Vec<EpoxyInstance>,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct EpoxyInstance {
+	pub replica_id: ReplicaId,
+	pub slot_id: SlotId,
+}
+
+/// Returns debug information for a specific key on this replica.
+///
+/// Shows all instances that have touched this key and their log entry states.
+pub async fn get_epoxy_key_debug(
+	ctx: ApiCtx,
+	path: EpoxyKeyDebugPath,
+	_query: (),
+) -> Result<GetEpoxyKeyDebugResponse> {
+	let replica_id = ctx.config().epoxy_replica_id();
+
+	// Decode key from base64
+	let key_bytes = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, &path.key)
+		.context("invalid base64 key")?;
+
+	let instances = ctx
+		.udb()?
+		.run(|tx| {
+			let key_bytes = key_bytes.clone();
+			async move {
+				let protocol_instances =
+					epoxy::utils::read_key_instances(&tx, replica_id, key_bytes).await?;
+
+				let mut instances = Vec::new();
+				for instance in protocol_instances {
+					let log_entry =
+						epoxy::utils::read_log_entry(&tx, replica_id, &instance).await?;
+					instances.push(EpoxyKeyInstance {
+						replica_id: instance.replica_id,
+						slot_id: instance.slot_id,
+						log_entry: log_entry.map(|entry| EpoxyKeyLogEntry {
+							status: format!("{:?}", entry.state),
+							ballot: EpoxyBallot {
+								epoch: entry.ballot.epoch,
+								ballot: entry.ballot.ballot,
+								replica_id: entry.ballot.replica_id,
+							},
+							seq: entry.seq,
+							deps: entry
+								.deps
+								.into_iter()
+								.map(|d| EpoxyInstance {
+									replica_id: d.replica_id,
+									slot_id: d.slot_id,
+								})
+								.collect(),
+						}),
+					});
+				}
+
+				Result::Ok(instances)
+			}
+		})
+		.await?;
+
+	// Compute instances by status
+	let mut instances_by_status = IndexMap::new();
+	for instance in &instances {
+		let status = instance
+			.log_entry
+			.as_ref()
+			.map(|e| e.status.clone())
+			.unwrap_or_else(|| "Unknown".to_string());
+		*instances_by_status.entry(status).or_insert(0) += 1;
+	}
+
+	Ok(GetEpoxyKeyDebugResponse {
+		replica_id,
+		key: path.key,
+		instances,
+		instances_by_status,
+	})
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct GetEpoxyKeyDebugFanoutResponse {
+	pub replicas: IndexMap<ReplicaId, GetEpoxyKeyDebugResponse>,
+	pub errors: IndexMap<ReplicaId, String>,
+}
+
+/// Returns debug information for a specific key across all replicas in the cluster.
+///
+/// Fans out to all replicas and aggregates their responses.
+pub async fn get_epoxy_key_debug_fanout(
+	ctx: ApiCtx,
+	path: EpoxyKeyDebugPath,
+	_query: (),
+) -> Result<GetEpoxyKeyDebugFanoutResponse> {
+	let replica_id = ctx.config().epoxy_replica_id();
+
+	// Get cluster config to find all replicas
+	let config: epoxy::types::ClusterConfig = ctx
+		.udb()?
+		.run(|tx| async move {
+			let config = epoxy::utils::read_config(&tx, replica_id).await?;
+			Result::Ok(config.into())
+		})
+		.await?;
+
+	// Get local response first
+	let local_response = get_epoxy_key_debug(ctx.clone(), path.clone(), ()).await?;
+
+	let mut replicas = IndexMap::new();
+	let mut errors = IndexMap::new();
+
+	// Add local response
+	replicas.insert(replica_id, local_response);
+
+	// Fan out to other replicas
+	let client = rivet_pools::reqwest::client().await?;
+
+	for replica_config in &config.replicas {
+		if replica_config.replica_id == replica_id {
+			continue;
+		}
+
+		let url = format!(
+			"{}/epoxy/replica/key/{}",
+			replica_config.api_peer_url, path.key
+		);
+
+		let response_result = client.get(&url).send().await;
+		match response_result {
+			std::result::Result::Ok(response) => {
+				if response.status().is_success() {
+					match response.json::<GetEpoxyKeyDebugResponse>().await {
+						std::result::Result::Ok(resp) => {
+							replicas.insert(replica_config.replica_id, resp);
+						}
+						std::result::Result::Err(e) => {
+							errors.insert(
+								replica_config.replica_id,
+								format!("failed to parse response: {}", e),
+							);
+						}
+					}
+				} else {
+					errors.insert(
+						replica_config.replica_id,
+						format!("request failed with status: {}", response.status()),
+					);
+				}
+			}
+			std::result::Result::Err(e) => {
+				errors.insert(replica_config.replica_id, format!("request failed: {}", e));
+			}
+		}
+	}
+
+	Ok(GetEpoxyKeyDebugFanoutResponse { replicas, errors })
+}

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

@@ -42,6 +42,18 @@ pub async fn router(
 			)
 			.route("/epoxy/coordinator/state", get(internal::get_epoxy_state))
 			.route("/epoxy/coordinator/state", post(internal::set_epoxy_state))
+			.route(
+				"/epoxy/replica/debug",
+				get(internal::get_epoxy_replica_debug),
+			)
+			.route(
+				"/epoxy/replica/key/{key}",
+				get(internal::get_epoxy_key_debug),
+			)
+			.route(
+				"/epoxy/replica/key/{key}/fanout",
+				get(internal::get_epoxy_key_debug_fanout),
+			)
 			.route("/debug/tracing/config", put(internal::set_tracing_config))
 	})
 	.await

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

@@ -11,6 +11,7 @@ path = "src/main.rs"
 
 [dependencies]
 anyhow.workspace = true
+base64.workspace = true
 chrono.workspace = true
 clap.workspace = true
 colored_json.workspace = true