feat: Use OptimisticTransactionDB for the database engine

Fixes #1131

Co-authored-by: Jade Ellis <jade@ellis.link>

.forgejo/workflows/build-debian.yml

@@ -35,6 +35,7 @@ jobs:
         uses: actions/checkout@v6
         with:
           fetch-depth: 0
+          ref: ${{ github.ref_name }}
 
       - name: Cache Cargo registry
         uses: actions/cache@v4

.forgejo/workflows/build-fedora.yml

@@ -33,6 +33,7 @@ jobs:
         uses: actions/checkout@v6
         with:
           fetch-depth: 0
+          ref: ${{ github.ref_name }}
 
 
       - name: Cache DNF packages

docs/deploying/generic.mdx

@@ -134,7 +134,7 @@ ### Example systemd Unit File
 ## Creating the Continuwuity configuration file
 
 Now you need to create the Continuwuity configuration file in
-`/etc/continuwuity/continuwuity.toml`. You can find an example configuration at
+`/etc/conduwuit/conduwuit.toml`. You can find an example configuration at
 [conduwuit-example.toml](../reference/config.mdx).
 
 **Please take a moment to read the config. You need to change at least the

src/core/config/mod.rs

@@ -1224,12 +1224,6 @@ pub struct Config {
 	#[serde(default)]
 	pub rocksdb_repair: bool,
 
-	#[serde(default)]
-	pub rocksdb_read_only: bool,
-
-	#[serde(default)]
-	pub rocksdb_secondary: bool,
-
 	/// Enables idle CPU priority for compaction thread. This is not enabled by
 	/// default to prevent compaction from falling too far behind on busy
 	/// systems.

src/database/engine.rs

@@ -19,7 +19,7 @@
 
 use conduwuit::{Err, Result, debug, info, warn};
 use rocksdb::{
-	AsColumnFamilyRef, BoundColumnFamily, DBCommon, DBWithThreadMode, MultiThreaded,
+	AsColumnFamilyRef, BoundColumnFamily, DBCommon, MultiThreaded, OptimisticTransactionDB,
 	WaitForCompactOptions,
 };
 
@@ -33,13 +33,11 @@ pub struct Engine {
 	pub(crate) db: Db,
 	pub(crate) pool: Arc<Pool>,
 	pub(crate) ctx: Arc<Context>,
-	pub(super) read_only: bool,
-	pub(super) secondary: bool,
 	pub(crate) checksums: bool,
 	corks: AtomicU32,
 }
 
-pub(crate) type Db = DBWithThreadMode<MultiThreaded>;
+pub(crate) type Db = OptimisticTransactionDB<MultiThreaded>;
 
 impl Engine {
 	#[tracing::instrument(
@@ -129,14 +127,6 @@ pub fn current_sequence(&self) -> u64 {
 
 		sequence
 	}
-
-	#[inline]
-	#[must_use]
-	pub fn is_read_only(&self) -> bool { self.secondary || self.read_only }
-
-	#[inline]
-	#[must_use]
-	pub fn is_secondary(&self) -> bool { self.secondary }
 }
 
 impl Drop for Engine {

src/database/engine/backup.rs

@@ -12,9 +12,8 @@ pub fn backup(&self) -> Result {
 	let mut engine = self.backup_engine()?;
 	let config = &self.ctx.server.config;
 	if config.database_backups_to_keep > 0 {
-		let flush = !self.is_read_only();
 		engine
-			.create_new_backup_flush(&self.db, flush)
+			.create_new_backup_flush(&self.db, true)
 			.map_err(map_err)?;
 
 		let engine_info = engine.get_backup_info();

src/database/engine/memory_usage.rs

@@ -1,7 +1,7 @@
 use std::fmt::Write;
 
 use conduwuit::{Result, implement};
-use rocksdb::perf::get_memory_usage_stats;
+use rocksdb::perf::MemoryUsageBuilder;
 
 use super::Engine;
 use crate::or_else;
@@ -9,16 +9,21 @@
 #[implement(Engine)]
 pub fn memory_usage(&self) -> Result<String> {
 	let mut res = String::new();
-	let stats = get_memory_usage_stats(Some(&[&self.db]), Some(&[&*self.ctx.row_cache.lock()]))
-		.or_else(or_else)?;
+
+	let mut builder = MemoryUsageBuilder::new().or_else(or_else)?;
+	builder.add_db(&self.db);
+	builder.add_cache(&self.ctx.row_cache.lock());
+
+	let usage = builder.build().or_else(or_else)?;
+
 	let mibs = |input| f64::from(u32::try_from(input / 1024).unwrap_or(0)) / 1024.0;
 	writeln!(
 		res,
 		"Memory buffers: {:.2} MiB\nPending write: {:.2} MiB\nTable readers: {:.2} MiB\nRow \
 		 cache: {:.2} MiB",
-		mibs(stats.mem_table_total),
-		mibs(stats.mem_table_unflushed),
-		mibs(stats.mem_table_readers_total),
+		mibs(usage.approximate_mem_table_total()),
+		mibs(usage.approximate_mem_table_unflushed()),
+		mibs(usage.approximate_mem_table_readers_total()),
 		mibs(u64::try_from(self.ctx.row_cache.lock().get_usage())?),
 	)?;
 

src/database/engine/open.rs

@@ -35,14 +35,7 @@ pub(crate) async fn open(ctx: Arc<Context>, desc: &[Descriptor]) -> Result<Arc<S
 	}
 
 	debug!("Opening database...");
-	let db = if config.rocksdb_read_only {
-		Db::open_cf_descriptors_read_only(&db_opts, path, cfds, false)
-	} else if config.rocksdb_secondary {
-		Db::open_cf_descriptors_as_secondary(&db_opts, path, path, cfds)
-	} else {
-		Db::open_cf_descriptors(&db_opts, path, cfds)
-	}
-	.or_else(or_else)?;
+	let db = Db::open_cf_descriptors(&db_opts, path, cfds).or_else(or_else)?;
 
 	info!(
 		columns = num_cfds,
@@ -55,8 +48,6 @@ pub(crate) async fn open(ctx: Arc<Context>, desc: &[Descriptor]) -> Result<Arc<S
 		db,
 		pool: ctx.pool.clone(),
 		ctx: ctx.clone(),
-		read_only: config.rocksdb_read_only,
-		secondary: config.rocksdb_secondary,
 		checksums: config.rocksdb_checksums,
 		corks: AtomicU32::new(0),
 	}))

src/database/map/insert.rs

@@ -219,7 +219,7 @@ pub fn insert_batch<'a, I, K, V>(&'a self, iter: I)
 	K: AsRef<[u8]> + Sized + Debug + 'a,
 	V: AsRef<[u8]> + Sized + 'a,
 {
-	let mut batch = WriteBatchWithTransaction::<false>::default();
+	let mut batch = WriteBatchWithTransaction::<true>::default();
 	for (key, val) in iter {
 		batch.put_cf(&self.cf(), key.as_ref(), val.as_ref());
 	}

src/database/mod.rs

@@ -77,14 +77,6 @@ pub fn iter(&self) -> impl Iterator<Item = (&MapsKey, &MapsVal)> + Send + '_ {
 
 	#[inline]
 	pub fn keys(&self) -> impl Iterator<Item = &MapsKey> + Send + '_ { self.maps.keys() }
-
-	#[inline]
-	#[must_use]
-	pub fn is_read_only(&self) -> bool { self.db.is_read_only() }
-
-	#[inline]
-	#[must_use]
-	pub fn is_secondary(&self) -> bool { self.db.is_secondary() }
 }
 
 impl Index<&str> for Database {

src/service/emergency/mod.rs

@@ -37,10 +37,6 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 	}
 
 	async fn worker(self: Arc<Self>) -> Result {
-		if self.services.globals.is_read_only() {
-			return Ok(());
-		}
-
 		if self.services.config.ldap.enable {
 			warn!("emergency password feature not available with LDAP enabled.");
 			return Ok(());

src/service/globals/mod.rs

@@ -171,7 +171,4 @@ pub fn user_is_local(&self, user_id: &UserId) -> bool {
 	pub fn server_is_ours(&self, server_name: &ServerName) -> bool {
 		server_name == self.server_name()
 	}
-
-	#[inline]
-	pub fn is_read_only(&self) -> bool { self.db.db.is_read_only() }
 }

src/service/services.rs

@@ -130,7 +130,7 @@ pub async fn start(self: &Arc<Self>) -> Result<Arc<Self>> {
 
 		// reset dormant online/away statuses to offline, and set the server user as
 		// online
-		if self.server.config.allow_local_presence && !self.db.is_read_only() {
+		if self.server.config.allow_local_presence {
 			self.presence.unset_all_presence().await;
 			_ = self
 				.presence
@@ -146,7 +146,7 @@ pub async fn stop(&self) {
 		info!("Shutting down services...");
 
 		// set the server user as offline
-		if self.server.config.allow_local_presence && !self.db.is_read_only() {
+		if self.server.config.allow_local_presence {
 			_ = self
 				.presence
 				.ping_presence(&self.globals.server_user, &ruma::presence::PresenceState::Offline)

src/service/uiaa/mod.rs

@@ -11,7 +11,7 @@
 use ruma::{
 	CanonicalJsonValue, DeviceId, OwnedDeviceId, OwnedUserId, UserId,
 	api::client::{
-		error::ErrorKind,
+		error::{ErrorKind, StandardErrorBody},
 		uiaa::{AuthData, AuthType, Password, UiaaInfo, UserIdentifier},
 	},
 };
@@ -104,6 +104,7 @@ pub fn create(
 }
 
 #[implement(Service)]
+#[allow(clippy::useless_let_if_seq)]
 pub async fn try_auth(
 	&self,
 	user_id: &UserId,
@@ -163,17 +164,39 @@ pub async fn try_auth(
 			let user_id = user_id_from_username;
 
 			// Check if password is correct
+			let mut password_verified = false;
+
+			// First try local password hash verification
 			if let Ok(hash) = self.services.users.password_hash(&user_id).await {
-				let hash_matches = hash::verify_password(password, &hash).is_ok();
-				if !hash_matches {
-					uiaainfo.auth_error = Some(ruma::api::client::error::StandardErrorBody {
-						kind: ErrorKind::forbidden(),
-						message: "Invalid username or password.".to_owned(),
-					});
-					return Ok((false, uiaainfo));
+				password_verified = hash::verify_password(password, &hash).is_ok();
+			}
+
+			// If local password verification failed, try LDAP authentication
+			#[cfg(feature = "ldap")]
+			if !password_verified && self.services.config.ldap.enable {
+				// Search for user in LDAP to get their DN
+				if let Ok(dns) = self.services.users.search_ldap(&user_id).await {
+					if let Some((user_dn, _is_admin)) = dns.first() {
+						// Try to authenticate with LDAP
+						password_verified = self
+							.services
+							.users
+							.auth_ldap(user_dn, password)
+							.await
+							.is_ok();
+					}
 				}
 			}
 
+			if !password_verified {
+				uiaainfo.auth_error = Some(StandardErrorBody {
+					kind: ErrorKind::forbidden(),
+					message: "Invalid username or password.".to_owned(),
+				});
+
+				return Ok((false, uiaainfo));
+			}
+
 			// Password was correct! Let's add it to `completed`
 			uiaainfo.completed.push(AuthType::Password);
 		},
@@ -197,7 +220,7 @@ pub async fn try_auth(
 				},
 				| Err(e) => {
 					error!("ReCaptcha verification failed: {e:?}");
-					uiaainfo.auth_error = Some(ruma::api::client::error::StandardErrorBody {
+					uiaainfo.auth_error = Some(StandardErrorBody {
 						kind: ErrorKind::forbidden(),
 						message: "ReCaptcha verification failed.".to_owned(),
 					});
@@ -210,7 +233,7 @@ pub async fn try_auth(
 			if tokens.contains(t.token.trim()) {
 				uiaainfo.completed.push(AuthType::RegistrationToken);
 			} else {
-				uiaainfo.auth_error = Some(ruma::api::client::error::StandardErrorBody {
+				uiaainfo.auth_error = Some(StandardErrorBody {
 					kind: ErrorKind::forbidden(),
 					message: "Invalid registration token.".to_owned(),
 				});