Merge pull request 'Garage v0.9' (#473) from next into main

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/473

1 files changed, 80 insertions, 9 deletions

diff --git a/src/model/garage.rs b/src/model/garage.rs
index a432aa7a..8c9a3af3 100644
--- a/src/model/garage.rs
+++ b/src/model/garage.rs
@@ -7,6 +7,7 @@ use garage_db as db;
 use garage_util::background::*;
 use garage_util::config::*;
 use garage_util::error::*;
+use garage_util::persister::PersisterShared;
 
 use garage_rpc::replication_mode::ReplicationMode;
 use garage_rpc::system::System;
@@ -17,6 +18,8 @@ use garage_table::replication::TableShardedReplication;
 use garage_table::*;
 
 use crate::s3::block_ref_table::*;
+use crate::s3::lifecycle_worker;
+use crate::s3::mpu_table::*;
 use crate::s3::object_table::*;
 use crate::s3::version_table::*;
 
@@ -57,11 +60,18 @@ pub struct Garage {
 	pub object_table: Arc<Table<ObjectTable, TableShardedReplication>>,
 	/// Counting table containing object counters
 	pub object_counter_table: Arc<IndexCounter<Object>>,
+	/// Table containing S3 multipart uploads
+	pub mpu_table: Arc<Table<MultipartUploadTable, TableShardedReplication>>,
+	/// Counting table containing multipart object counters
+	pub mpu_counter_table: Arc<IndexCounter<MultipartUpload>>,
 	/// Table containing S3 object versions
 	pub version_table: Arc<Table<VersionTable, TableShardedReplication>>,
 	/// Table containing S3 block references (not blocks themselves)
 	pub block_ref_table: Arc<Table<BlockRefTable, TableShardedReplication>>,
 
+	/// Persister for lifecycle worker info
+	pub lifecycle_persister: PersisterShared<lifecycle_worker::LifecycleWorkerPersisted>,
+
 	#[cfg(feature = "k2v")]
 	pub k2v: GarageK2V,
 }
@@ -82,8 +92,22 @@ impl Garage {
 		// Create meta dir and data dir if they don't exist already
 		std::fs::create_dir_all(&config.metadata_dir)
 			.ok_or_message("Unable to create Garage metadata directory")?;
-		std::fs::create_dir_all(&config.data_dir)
-			.ok_or_message("Unable to create Garage data directory")?;
+		match &config.data_dir {
+			DataDirEnum::Single(data_dir) => {
+				std::fs::create_dir_all(data_dir).ok_or_message(format!(
+					"Unable to create Garage data directory: {}",
+					data_dir.to_string_lossy()
+				))?;
+			}
+			DataDirEnum::Multiple(data_dirs) => {
+				for dir in data_dirs {
+					std::fs::create_dir_all(&dir.path).ok_or_message(format!(
+						"Unable to create Garage data directory: {}",
+						dir.path.to_string_lossy()
+					))?;
+				}
+			}
+		}
 
 		info!("Opening database...");
 		let mut db_path = config.metadata_dir.clone();
@@ -91,6 +115,11 @@ impl Garage {
 			// ---- Sled DB ----
 			#[cfg(feature = "sled")]
 			"sled" => {
+				if config.metadata_fsync {
+					return Err(Error::Message(format!(
+						"`metadata_fsync = true` is not supported with the Sled database engine"
+					)));
+				}
 				db_path.push("db");
 				info!("Opening Sled database at: {}", db_path.display());
 				let db = db::sled_adapter::sled::Config::default()
@@ -109,6 +138,15 @@ impl Garage {
 				db_path.push("db.sqlite");
 				info!("Opening Sqlite database at: {}", db_path.display());
 				let db = db::sqlite_adapter::rusqlite::Connection::open(db_path)
+					.and_then(|db| {
+						db.pragma_update(None, "journal_mode", &"WAL")?;
+						if config.metadata_fsync {
+							db.pragma_update(None, "synchronous", &"NORMAL")?;
+						} else {
+							db.pragma_update(None, "synchronous", &"OFF")?;
+						}
+						Ok(db)
+					})
 					.ok_or_message("Unable to open sqlite DB")?;
 				db::sqlite_adapter::SqliteDb::init(db)
 			}
@@ -136,8 +174,10 @@ impl Garage {
 				env_builder.max_readers(500);
 				env_builder.map_size(map_size);
 				unsafe {
-					env_builder.flag(heed::flags::Flags::MdbNoSync);
 					env_builder.flag(heed::flags::Flags::MdbNoMetaSync);
+					if !config.metadata_fsync {
+						env_builder.flag(heed::flags::Flags::MdbNoSync);
+					}
 				}
 				let db = match env_builder.open(&db_path) {
 					Err(heed::Error::Io(e)) if e.kind() == std::io::ErrorKind::OutOfMemory => {
@@ -182,6 +222,9 @@ impl Garage {
 		let replication_mode = ReplicationMode::parse(&config.replication_mode)
 			.ok_or_message("Invalid replication_mode in config file.")?;
 
+		info!("Initialize background variable system...");
+		let mut bg_vars = vars::BgVars::new();
+
 		info!("Initialize membership management system...");
 		let system = System::new(network_key, replication_mode, &config)?;
 
@@ -208,10 +251,12 @@ impl Garage {
 		let block_manager = BlockManager::new(
 			&db,
 			config.data_dir.clone(),
+			config.data_fsync,
 			config.compression_level,
 			data_rep_param,
 			system.clone(),
-		);
+		)?;
+		block_manager.register_bg_vars(&mut bg_vars);
 
 		// ---- admin tables ----
 		info!("Initialize bucket_table...");
@@ -248,6 +293,20 @@ impl Garage {
 			&db,
 		);
 
+		info!("Initialize multipart upload counter table...");
+		let mpu_counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), &db);
+
+		info!("Initialize multipart upload table...");
+		let mpu_table = Table::new(
+			MultipartUploadTable {
+				version_table: version_table.clone(),
+				mpu_counter_table: mpu_counter_table.clone(),
+			},
+			meta_rep_param.clone(),
+			system.clone(),
+			&db,
+		);
+
 		info!("Initialize object counter table...");
 		let object_counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), &db);
 
@@ -256,6 +315,7 @@ impl Garage {
 		let object_table = Table::new(
 			ObjectTable {
 				version_table: version_table.clone(),
+				mpu_table: mpu_table.clone(),
 				object_counter_table: object_counter_table.clone(),
 			},
 			meta_rep_param.clone(),
@@ -263,14 +323,15 @@ impl Garage {
 			&db,
 		);
 
+		info!("Load lifecycle worker state...");
+		let lifecycle_persister =
+			PersisterShared::new(&system.metadata_dir, "lifecycle_worker_state");
+		lifecycle_worker::register_bg_vars(&lifecycle_persister, &mut bg_vars);
+
 		// ---- K2V ----
 		#[cfg(feature = "k2v")]
 		let k2v = GarageK2V::new(system.clone(), &db, meta_rep_param);
 
-		// Initialize bg vars
-		let mut bg_vars = vars::BgVars::new();
-		block_manager.register_bg_vars(&mut bg_vars);
-
 		// -- done --
 		Ok(Arc::new(Self {
 			config,
@@ -284,14 +345,17 @@ impl Garage {
 			key_table,
 			object_table,
 			object_counter_table,
+			mpu_table,
+			mpu_counter_table,
 			version_table,
 			block_ref_table,
+			lifecycle_persister,
 			#[cfg(feature = "k2v")]
 			k2v,
 		}))
 	}
 
-	pub fn spawn_workers(&self, bg: &BackgroundRunner) {
+	pub fn spawn_workers(self: &Arc<Self>, bg: &BackgroundRunner) {
 		self.block_manager.spawn_workers(bg);
 
 		self.bucket_table.spawn_workers(bg);
@@ -300,9 +364,16 @@ impl Garage {
 
 		self.object_table.spawn_workers(bg);
 		self.object_counter_table.spawn_workers(bg);
+		self.mpu_table.spawn_workers(bg);
+		self.mpu_counter_table.spawn_workers(bg);
 		self.version_table.spawn_workers(bg);
 		self.block_ref_table.spawn_workers(bg);
 
+		bg.spawn_worker(lifecycle_worker::LifecycleWorker::new(
+			self.clone(),
+			self.lifecycle_persister.clone(),
+		));
+
 		#[cfg(feature = "k2v")]
 		self.k2v.spawn_workers(bg);
 	}