fix(config): allow configuring topo dcs via map, fix pg ssl mode config

Author: MasterPtato

Cargo.lock

@@ -10,18 +10,17 @@ export function generateDatacenterRivetEngine(
 ) {
 	const clickhouseHost =
 		context.config.networkMode === "host" ? "127.0.0.1" : "clickhouse";
-	const datacenters = [];
+	const datacenters: Record<string, any> = {};
 
 	for (const dc of context.config.datacenters) {
 		const serviceHost = context.getServiceHost("rivet-engine", dc.name, 0);
-		datacenters.push({
-			name: dc.name,
+		datacenters[dc.name] = {
 			datacenter_label: dc.id,
 			is_leader: dc.id === 1,
 			peer_url: `http://${serviceHost}:${API_PEER_PORT}`,
 			public_url: `http://${serviceHost}:${GUARD_PORT}`,
 			valid_hosts: [`${serviceHost}`, `127.0.0.1`, `localhost`],
-		});
+		};
 	}
 
 	// Generate a separate config file for each engine node

engine/artifacts/config-schema.json

@@ -59,66 +59,73 @@ async fn fanout_inner(ctx: ApiCtx) -> Result<FanoutResponse> {
 	// Require datacenter read permissions to access health status
 	ctx.auth().await?;
 
-	let dcs = &ctx.config().topology().datacenters;
+	let dcs = ctx
+		.config()
+		.topology()
+		.datacenters
+		.iter()
+		.cloned()
+		.collect::<Vec<_>>();
 
 	tracing::debug!(datacenters = dcs.len(), "starting health fanout");
 
-	let results = futures_util::stream::iter(dcs.clone().into_iter().map(|dc| {
-		let ctx = ctx.clone();
-
-		async move {
-			let start = Instant::now();
-
-			if dc.datacenter_label == ctx.config().dc_label() {
-				// Local datacenter - check directly
-				let response = HealthResponse {
-					runtime: "engine".to_string(),
-					status: "ok".to_string(),
-					version: env!("CARGO_PKG_VERSION").to_string(),
-				};
-
-				DatacenterHealth {
-					datacenter_label: dc.datacenter_label,
-					datacenter_name: dc.name.clone(),
-					status: HealthStatus::Ok,
-					rtt_ms: Some(start.elapsed().as_secs_f64() * 1000.0),
-					response: Some(response),
-					error: None,
-				}
-			} else {
-				// Remote datacenter - HTTP request
-				match send_health_checks(&dc).await {
-					Ok(response) => DatacenterHealth {
+	let results = futures_util::stream::iter(dcs)
+		.map(|dc| {
+			let ctx = ctx.clone();
+
+			async move {
+				let start = Instant::now();
+
+				if dc.datacenter_label == ctx.config().dc_label() {
+					// Local datacenter - check directly
+					let response = HealthResponse {
+						runtime: "engine".to_string(),
+						status: "ok".to_string(),
+						version: env!("CARGO_PKG_VERSION").to_string(),
+					};
+
+					DatacenterHealth {
 						datacenter_label: dc.datacenter_label,
 						datacenter_name: dc.name.clone(),
 						status: HealthStatus::Ok,
 						rtt_ms: Some(start.elapsed().as_secs_f64() * 1000.0),
 						response: Some(response),
 						error: None,
-					},
-					Err(err) => {
-						tracing::warn!(
-							?dc.datacenter_label,
-							?err,
-							"health check failed for datacenter"
-						);
-
-						DatacenterHealth {
+					}
+				} else {
+					// Remote datacenter - HTTP request
+					match send_health_checks(&dc).await {
+						Ok(response) => DatacenterHealth {
 							datacenter_label: dc.datacenter_label,
 							datacenter_name: dc.name.clone(),
-							status: HealthStatus::Error,
+							status: HealthStatus::Ok,
 							rtt_ms: Some(start.elapsed().as_secs_f64() * 1000.0),
-							response: None,
-							error: Some(err.to_string()),
+							response: Some(response),
+							error: None,
+						},
+						Err(err) => {
+							tracing::warn!(
+								?dc.datacenter_label,
+								?err,
+								"health check failed for datacenter"
+							);
+
+							DatacenterHealth {
+								datacenter_label: dc.datacenter_label,
+								datacenter_name: dc.name.clone(),
+								status: HealthStatus::Error,
+								rtt_ms: Some(start.elapsed().as_secs_f64() * 1000.0),
+								response: None,
+								error: Some(err.to_string()),
+							}
 						}
 					}
 				}
 			}
-		}
-	}))
-	.buffer_unordered(16)
-	.collect::<Vec<_>>()
-	.await;
+		})
+		.buffer_unordered(16)
+		.collect::<Vec<_>>()
+		.await;
 
 	tracing::debug!(results = results.len(), "health fanout completed");
 

engine/docker/template/src/services/edge/rivet-engine.ts

@@ -39,7 +39,13 @@ pub async fn delete(
 async fn delete_inner(ctx: ApiCtx, path: DeletePath, query: DeleteQuery) -> Result<DeleteResponse> {
 	ctx.auth().await?;
 
-	let dcs = ctx.config().topology().datacenters.clone();
+	let dcs = ctx
+		.config()
+		.topology()
+		.datacenters
+		.iter()
+		.cloned()
+		.collect::<Vec<_>>();
 	futures_util::stream::iter(dcs)
 		.map(|dc| {
 			let ctx = ctx.clone();

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

@@ -112,7 +112,13 @@ where
 	A: Fn(u16, I, &mut R),
 	R: Default + Send + 'static,
 {
-	let dcs = ctx.config().topology().datacenters.clone();
+	let dcs = ctx
+		.config()
+		.topology()
+		.datacenters
+		.iter()
+		.cloned()
+		.collect::<Vec<_>>();
 
 	let results = futures_util::stream::iter(dcs)
 		.map(|dc| {

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

@@ -203,6 +203,31 @@ impl Root {
 			}
 		}
 
+		// Set name property for map topology
+		if let Some(topology) = &mut self.topology {
+			match &mut topology.datacenters {
+				DatacentersRepr::Map(m) => {
+					for (name, dc) in m {
+						if !dc.name.is_empty() {
+							bail!(
+								"datacenter '{}' cannot have the `name` property set because it is automatically derived from key",
+								dc.name
+							);
+						}
+
+						dc.name = name.clone();
+					}
+				}
+				DatacentersRepr::List(l) => {
+					for (i, dc) in l.iter().enumerate() {
+						if dc.name.is_empty() {
+							bail!("datacenter at index {} must have a name", i);
+						}
+					}
+				}
+			}
+		}
+
 		// Validate force_shutdown_duration is greater than worker and guard shutdown durations
 		let worker = self.runtime.worker_shutdown_duration();
 		let guard = self.runtime.guard_shutdown_duration();

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

@@ -1,15 +1,16 @@
 use anyhow::{Context, Result};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 use url::Url;
 
 #[derive(Debug, Serialize, Deserialize, Clone, JsonSchema)]
 #[serde(deny_unknown_fields)]
 pub struct Topology {
 	/// Must be included in `datacenters`
 	pub datacenter_label: u16,
-	/// List of all datacenters, including this datacenter.
-	pub datacenters: Vec<Datacenter>,
+	/// Map of all datacenters, including this datacenter.
+	pub datacenters: DatacentersRepr,
 }
 
 impl Topology {
@@ -47,30 +48,88 @@ impl Default for Topology {
 	fn default() -> Self {
 		Topology {
 			datacenter_label: 1,
-			datacenters: vec![Datacenter {
-				name: "default".into(),
-				datacenter_label: 1,
-				is_leader: true,
-				public_url: Url::parse(&format!(
-					"http://127.0.0.1:{}",
-					crate::defaults::ports::GUARD
-				))
-				.unwrap(),
-				peer_url: Url::parse(&format!(
-					"http://127.0.0.1:{}",
-					crate::defaults::ports::API_PEER
-				))
-				.unwrap(),
-				proxy_url: None,
-				valid_hosts: None,
-			}],
+			datacenters: DatacentersRepr::Map(
+				[(
+					"default".to_string(),
+					Datacenter {
+						name: "default".into(),
+						datacenter_label: 1,
+						is_leader: true,
+						public_url: Url::parse(&format!(
+							"http://127.0.0.1:{}",
+							crate::defaults::ports::GUARD
+						))
+						.unwrap(),
+						peer_url: Url::parse(&format!(
+							"http://127.0.0.1:{}",
+							crate::defaults::ports::API_PEER
+						))
+						.unwrap(),
+						proxy_url: None,
+						valid_hosts: None,
+					},
+				)]
+				.into(),
+			),
+		}
+	}
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
+#[serde(untagged)]
+pub enum DatacentersRepr {
+	Map(HashMap<String, Datacenter>),
+	/// Deprecated.
+	List(Vec<Datacenter>),
+}
+
+impl DatacentersRepr {
+	pub fn iter(&self) -> DatacentersIter<'_> {
+		self.into_iter()
+	}
+
+	pub fn len(&self) -> usize {
+		match self {
+			DatacentersRepr::Map(m) => m.len(),
+			DatacentersRepr::List(l) => l.len(),
+		}
+	}
+}
+
+pub enum DatacentersIter<'a> {
+	Map(std::collections::hash_map::Values<'a, String, Datacenter>),
+	List(std::slice::Iter<'a, Datacenter>),
+}
+
+impl<'a> Iterator for DatacentersIter<'a> {
+	type Item = &'a Datacenter;
+
+	fn next(&mut self) -> Option<Self::Item> {
+		match self {
+			DatacentersIter::Map(iter) => iter.next(),
+			DatacentersIter::List(iter) => iter.next(),
+		}
+	}
+}
+
+impl<'a> IntoIterator for &'a DatacentersRepr {
+	type Item = &'a Datacenter;
+	type IntoIter = DatacentersIter<'a>;
+
+	fn into_iter(self) -> Self::IntoIter {
+		match self {
+			DatacentersRepr::Map(map) => DatacentersIter::Map(map.values()),
+			DatacentersRepr::List(vec) => DatacentersIter::List(vec.iter()),
 		}
 	}
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone, JsonSchema)]
 #[serde(deny_unknown_fields)]
 pub struct Datacenter {
+	/// When configuring `datacenters` via a hashmap this is automatically derived from the key. Required
+	/// when configuring via list (which is deprecated)
+	#[serde(default = "String::new")]
 	pub name: String,
 	pub datacenter_label: u16,
 	pub is_leader: bool,