Tuck maybe_bootstrap_checkpoint() back into replication service. (#366)

Use strong Url type.

Signed-off-by: Jason Volk <jason@zemos.net>

Author: jevolk

src/api/client/replication.rs

@@ -30,6 +30,7 @@ use serde_json::json;
 use tokio::time::sleep;
 use tuwunel_core::{Err, Result, err, utils::time::now};
 use tuwunel_database::{WalFrame, is_wal_gap_error};
+use url::Url;
 
 /// Query parameters for `GET /_tuwunel/replication/wal`.
 #[derive(Debug, Deserialize)]
@@ -57,7 +58,7 @@ pub(crate) async fn replication_status(
 		.server
 		.config
 		.rocksdb_primary_url
-		.as_deref()
+		.as_ref()
 		.filter(|_| !services.replication.is_promoted())
 		.and(Some("secondary"))
 		.unwrap_or("primary");
@@ -272,7 +273,7 @@ pub(crate) async fn replication_promote(
 #[derive(Debug, Deserialize, Serialize)]
 pub(crate) struct DemoteBody {
 	/// URL of the new primary to replicate from (e.g. `http://host:8008`).
-	pub primary_url: String,
+	pub primary_url: Url,
 }
 
 /// `POST /_tuwunel/replication/demote`
@@ -295,7 +296,7 @@ pub(crate) async fn replication_demote(
 	State(services): State<crate::State>,
 	Json(body): Json<DemoteBody>,
 ) -> impl IntoResponse {
-	if body.primary_url.is_empty() {
+	if body.primary_url.as_str().is_empty() {
 		return Err!(HttpJson(BAD_REQUEST, {"error": "primary_url is required"}));
 	}
 

src/core/config/mod.rs

@@ -1350,19 +1350,16 @@ pub struct Config {
 	/// Path for the secondary instance's own RocksDB log files. Required when
 	/// `rocksdb_secondary` is true and the primary DB is not on a shared
 	/// filesystem. Must be a writable directory local to this host.
-	#[serde(default)]
 	pub rocksdb_secondary_path: Option<PathBuf>,
 
 	/// URL of the primary instance for WAL-streaming replication.
 	/// Example: `https://primary.example.com`
 	/// Required on secondary instances that use WAL streaming.
-	#[serde(default)]
-	pub rocksdb_primary_url: Option<String>,
+	pub rocksdb_primary_url: Option<Url>,
 
 	/// Shared secret token for replication endpoint authentication.
 	/// Both primary and secondary must have the same value.
 	/// Leave unset to disable the replication HTTP endpoints entirely.
-	#[serde(default)]
 	pub rocksdb_replication_token: Option<String>,
 
 	/// How long (in seconds) the primary retains WAL segments beyond what

src/router/run.rs

@@ -7,7 +7,7 @@ use futures::{FutureExt, future::join, pin_mut};
 use tuwunel_core::{
 	Error, Result, Server, debug, debug_error, debug_info, error, info, utils::BoolExt,
 };
-use tuwunel_service::Services;
+use tuwunel_service::{Services, replication::maybe_bootstrap_checkpoint};
 
 use crate::{handle::ServerHandle, serve};
 
@@ -79,6 +79,7 @@ pub(crate) async fn run(services: Arc<Services>) -> Result {
 pub(crate) async fn start(server: Arc<Server>) -> Result<Arc<Services>> {
 	debug!("Starting...");
 
+	maybe_bootstrap_checkpoint(&server).await?;
 	let services = Services::build(server).await?.start().await?;
 
 	#[cfg(all(feature = "systemd", target_os = "linux"))]

src/service/replication/bootstrap.rs

@@ -0,0 +1,117 @@
+use std::{fs, io::Cursor, sync::Arc, time::Duration};
+
+use bytes::Bytes;
+use reqwest::Client;
+use tar::Archive;
+use tuwunel_core::{Err, Result, Server, err, info};
+use url::Url;
+
+pub async fn maybe_bootstrap_checkpoint(server: &Arc<Server>) -> Result {
+	let Some(primary_url) = server.config.rocksdb_primary_url.as_ref() else {
+		return Ok(());
+	};
+
+	bootstrap_checkpoint(server, primary_url).await
+}
+
+#[tracing::instrument(
+	name = "bootstrap",
+	level = "info",
+	ret,
+	skip_all,
+	fields(primary_url)
+)]
+async fn bootstrap_checkpoint(server: &Arc<Server>, primary_url: &Url) -> Result {
+	let db_path = server.config.database_path.clone();
+	let sidecar = db_path
+		.parent()
+		.unwrap_or(&db_path)
+		.join("_replication_needs_bootstrap");
+
+	let current_file = db_path.join("CURRENT");
+	let needs_bootstrap = sidecar.exists()
+		|| !current_file.exists()
+		|| fs::metadata(&current_file)
+			.as_ref()
+			.map(fs::Metadata::len)
+			.unwrap_or(0)
+			.eq(&0);
+
+	if !needs_bootstrap {
+		return Ok(());
+	}
+
+	let token = server
+		.config
+		.rocksdb_replication_token
+		.as_deref()
+		.unwrap_or_default();
+
+	info!("Pre-open bootstrap: downloading checkpoint from {primary_url}");
+
+	let client = Client::builder()
+		.connect_timeout(Duration::from_secs(10))
+		.build()
+		.map_err(|e| err!(Database("Failed to build HTTP client: {e}")))?;
+
+	let resp = client
+		.get(format!("{primary_url}/_tuwunel/replication/checkpoint"))
+		.header("x-tuwunel-replication-token", token)
+		.send()
+		.await
+		.map_err(|e| err!(Database("Checkpoint request failed: {e}")))?;
+
+	if !resp.status().is_success() {
+		return Err!(Database("Primary returned {} for checkpoint", resp.status()));
+	}
+
+	let seq: u64 = resp
+		.headers()
+		.get("x-tuwunel-checkpoint-sequence")
+		.and_then(|v| v.to_str().ok())
+		.and_then(|s| s.parse().ok())
+		.unwrap_or(0);
+
+	let tar_bytes: Bytes = resp
+		.bytes()
+		.await
+		.map_err(|e| err!(Database("Reading checkpoint body: {e}")))?;
+
+	let parent = db_path.parent().unwrap_or(&db_path);
+	let staging = parent.join("_replication_staging");
+	let backup = parent.join("_replication_backup");
+
+	if staging.exists() {
+		fs::remove_dir_all(&staging).map_err(|e| err!(Database("Removing staging dir: {e}")))?;
+	}
+
+	fs::create_dir_all(&staging).map_err(|e| err!(Database("Creating staging dir: {e}")))?;
+
+	let cursor = Cursor::new(&*tar_bytes);
+	let mut archive = Archive::new(cursor);
+	archive
+		.unpack(&staging)
+		.map_err(|e| err!(Database("Unpacking checkpoint: {e}")))?;
+
+	let checkpoint_src = staging.join("checkpoint");
+
+	if backup.exists() {
+		fs::remove_dir_all(&backup).map_err(|e| err!(Database("Removing backup: {e}")))?;
+	}
+
+	if db_path.exists() {
+		fs::rename(&db_path, &backup).map_err(|e| err!(Database("Moving db to backup: {e}")))?;
+	}
+
+	fs::rename(&checkpoint_src, &db_path)
+		.map_err(|e| err!(Database("Moving checkpoint to db_path: {e}")))?;
+
+	fs::remove_dir_all(&staging).ok();
+
+	fs::write(&sidecar, seq.to_string())
+		.map_err(|e| err!(Database("Writing bootstrap sidecar: {e}")))?;
+
+	info!("Pre-open bootstrap complete; resume_seq = {seq}. RocksDB will open clean checkpoint.");
+
+	Ok(())
+}

src/service/replication/mod.rs

@@ -22,6 +22,8 @@
 //!   -> on demote(url): exit standby, bootstrap from new primary, resume stream
 //! ```
 
+mod bootstrap;
+
 use std::{
 	fs,
 	sync::{
@@ -39,7 +41,9 @@ use tokio::{
 };
 use tuwunel_core::{Err, Result, Server, err, error, expected, implement, info, warn};
 use tuwunel_database::{Database, WalFrame, is_wal_gap_error};
+use url::Url;
 
+pub use self::bootstrap::maybe_bootstrap_checkpoint;
 use crate::{
 	service::{Args, make_name},
 	services::OnceServices,
@@ -62,7 +66,7 @@ pub struct Service {
 
 	/// Runtime-overridden primary URL set by `demote()`. Takes precedence over
 	/// `config.rocksdb_primary_url` when set.
-	dynamic_primary_url: RwLock<Option<String>>,
+	dynamic_primary_url: RwLock<Option<Url>>,
 
 	/// Wakes the standby loop immediately when `demote()` is called.
 	demote_notify: Notify,
@@ -282,7 +286,7 @@ pub fn is_promoted(&self) -> bool { self.promoted.load(Ordering::Acquire) }
 ///
 /// Returns `Err` if the instance is not currently promoted.
 #[implement(Service)]
-pub async fn demote(&self, new_primary_url: String) -> Result {
+pub async fn demote(&self, new_primary_url: Url) -> Result {
 	if !self.promoted.load(Ordering::Acquire) {
 		return Err!(Database("This instance is not currently promoted; cannot demote."));
 	}
@@ -304,7 +308,7 @@ pub async fn demote(&self, new_primary_url: String) -> Result {
 /// Stream WAL frames from the primary until disconnect, promotion, or
 /// error.
 #[implement(Service)]
-async fn run_stream(&self, primary_url: &str) -> Result {
+async fn run_stream(&self, primary_url: &Url) -> Result {
 	let resume_seq = self.get_replication_resume_seq()?;
 	let url = format!("{primary_url}/_tuwunel/replication/wal?since={resume_seq}");
 	let resp = self

src/service/services.rs

@@ -73,7 +73,6 @@ pub struct Services {
 
 #[implement(Services)]
 pub async fn build(server: Arc<Server>) -> Result<Arc<Self>> {
-	Self::maybe_bootstrap_checkpoint(&server).await?;
 	let db = Database::open(&server).await?;
 	let services = Arc::new(OnceServices::default());
 	let args = Args {
@@ -275,106 +274,3 @@ pub async fn memory_usage(&self) -> Result<String> {
 		})
 		.await
 }
-
-#[implement(Services)]
-async fn maybe_bootstrap_checkpoint(server: &Arc<Server>) -> Result {
-	use bytes::Bytes;
-
-	let primary_url = match server.config.rocksdb_primary_url.as_ref() {
-		| Some(url) => url.clone(),
-		| None => return Ok(()),
-	};
-
-	let db_path = server.config.database_path.clone();
-	let sidecar = db_path
-		.parent()
-		.unwrap_or(&db_path)
-		.join("_replication_needs_bootstrap");
-
-	let current_file = db_path.join("CURRENT");
-	let needs_bootstrap = sidecar.exists() || !current_file.exists() || {
-		std::fs::metadata(&current_file)
-			.map(|m| m.len() == 0)
-			.unwrap_or(true)
-	};
-
-	if !needs_bootstrap {
-		return Ok(());
-	}
-
-	let token = server
-		.config
-		.rocksdb_replication_token
-		.as_deref()
-		.unwrap_or("");
-
-	info!("Pre-open bootstrap: downloading checkpoint from {primary_url}");
-
-	let client = reqwest::Client::builder()
-		.connect_timeout(std::time::Duration::from_secs(10))
-		.build()
-		.map_err(|e| tuwunel_core::err!(Database("Failed to build HTTP client: {e}")))?;
-
-	let resp = client
-		.get(format!("{primary_url}/_tuwunel/replication/checkpoint"))
-		.header("x-tuwunel-replication-token", token)
-		.send()
-		.await
-		.map_err(|e| tuwunel_core::err!(Database("Checkpoint request failed: {e}")))?;
-
-	if !resp.status().is_success() {
-		return Err(tuwunel_core::err!(Database(
-			"Primary returned {} for checkpoint",
-			resp.status()
-		)));
-	}
-
-	let seq: u64 = resp
-		.headers()
-		.get("x-tuwunel-checkpoint-sequence")
-		.and_then(|v| v.to_str().ok())
-		.and_then(|s| s.parse().ok())
-		.unwrap_or(0);
-
-	let tar_bytes: Bytes = resp
-		.bytes()
-		.await
-		.map_err(|e| tuwunel_core::err!(Database("Reading checkpoint body: {e}")))?;
-
-	let parent = db_path.parent().unwrap_or(&db_path);
-	let staging = parent.join("_replication_staging");
-	let backup = parent.join("_replication_backup");
-
-	if staging.exists() {
-		std::fs::remove_dir_all(&staging)
-			.map_err(|e| tuwunel_core::err!(Database("Removing staging dir: {e}")))?;
-	}
-	std::fs::create_dir_all(&staging)
-		.map_err(|e| tuwunel_core::err!(Database("Creating staging dir: {e}")))?;
-
-	let cursor = std::io::Cursor::new(&*tar_bytes);
-	let mut archive = tar::Archive::new(cursor);
-	archive
-		.unpack(&staging)
-		.map_err(|e| tuwunel_core::err!(Database("Unpacking checkpoint: {e}")))?;
-
-	let checkpoint_src = staging.join("checkpoint");
-
-	if backup.exists() {
-		std::fs::remove_dir_all(&backup)
-			.map_err(|e| tuwunel_core::err!(Database("Removing backup: {e}")))?;
-	}
-	if db_path.exists() {
-		std::fs::rename(&db_path, &backup)
-			.map_err(|e| tuwunel_core::err!(Database("Moving db to backup: {e}")))?;
-	}
-	std::fs::rename(&checkpoint_src, &db_path)
-		.map_err(|e| tuwunel_core::err!(Database("Moving checkpoint to db_path: {e}")))?;
-
-	let _: std::io::Result<()> = std::fs::remove_dir_all(&staging);
-	std::fs::write(&sidecar, seq.to_string())
-		.map_err(|e| tuwunel_core::err!(Database("Writing bootstrap sidecar: {e}")))?;
-
-	info!("Pre-open bootstrap complete; resume_seq = {seq}. RocksDB will open clean checkpoint.");
-	Ok(())
-}

tuwunel-example.toml

@@ -1130,19 +1130,19 @@
 # `rocksdb_secondary` is true and the primary DB is not on a shared
 # filesystem. Must be a writable directory local to this host.
 #
-#rocksdb_secondary_path = false
+#rocksdb_secondary_path =
 
 # URL of the primary instance for WAL-streaming replication.
 # Example: `https://primary.example.com`
 # Required on secondary instances that use WAL streaming.
 #
-#rocksdb_primary_url = false
+#rocksdb_primary_url =
 
 # Shared secret token for replication endpoint authentication.
 # Both primary and secondary must have the same value.
 # Leave unset to disable the replication HTTP endpoints entirely.
 #
-#rocksdb_replication_token = false
+#rocksdb_replication_token =
 
 # How long (in seconds) the primary retains WAL segments beyond what
 # local recovery requires. Gives the secondary a window to reconnect