diff --git a/src/cmd/src/error.rs b/src/cmd/src/error.rs index 82273c0b67d..7b0decf05ca 100644 --- a/src/cmd/src/error.rs +++ b/src/cmd/src/error.rs @@ -85,6 +85,13 @@ pub enum Error { source: datanode::error::Error, }, + #[snafu(display("Failed to close the metadata kv backend"))] + CloseMetadataKvBackend { + #[snafu(implicit)] + location: Location, + source: common_meta::error::Error, + }, + #[snafu(display("Failed to start flownode"))] StartFlownode { #[snafu(implicit)] @@ -333,6 +340,7 @@ impl ErrorExt for Error { Error::StartDatanode { source, .. } => source.status_code(), Error::StartFrontend { source, .. } => source.status_code(), Error::ShutdownDatanode { source, .. } => source.status_code(), + Error::CloseMetadataKvBackend { source, .. } => source.status_code(), Error::ShutdownFrontend { source, .. } => source.status_code(), Error::StartMetaServer { source, .. } => source.status_code(), Error::ShutdownMetaServer { source, .. } => source.status_code(), diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index 2b07db019ab..0521aa9404f 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -39,7 +39,7 @@ use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl}; use common_meta::ddl_manager::{DdlManager, DdlManagerConfiguratorRef, DdlManagerRef}; use common_meta::key::flow::FlowMetadataManager; use common_meta::key::{TableMetadataManager, TableMetadataManagerRef}; -use common_meta::kv_backend::KvBackendRef; +use common_meta::kv_backend::{ClosableKvBackendRef, KvBackendRef}; use common_meta::node_manager::{FlownodeRef, NodeManagerRef}; use common_meta::procedure_executor::{LocalProcedureExecutor, ProcedureExecutorRef}; use common_meta::region_keeper::MemoryRegionKeeper; @@ -225,6 +225,8 @@ pub struct Instance { procedure_manager: ProcedureManagerRef, leader_services_controller: Box, leader_services_context: LeaderServicesContext, + /// Closed last on shutdown, see [ClosableKvBackend](common_meta::kv_backend::ClosableKvBackend). + metadata_kv_backend: ClosableKvBackendRef, // Keep the logging guard to prevent the worker from being dropped. _guard: Vec, } @@ -279,7 +281,7 @@ impl App for Instance { /// ones; the first error is returned and the rest are logged. async fn stop(&mut self) -> Result<()> { let mut first_error = None; - let mut record = |result: Result<()>| { + let mut record_shutdown_error = |result: Result<()>| { if let Err(err) = result { if first_error.is_none() { first_error = Some(err); @@ -289,14 +291,14 @@ impl App for Instance { } }; - record( + record_shutdown_error( self.frontend .shutdown() .await .context(error::ShutdownFrontendSnafu), ); - record( + record_shutdown_error( self.leader_services_controller .stop( self.procedure_manager.clone(), @@ -305,20 +307,28 @@ impl App for Instance { .await, ); - record( + record_shutdown_error( self.datanode .shutdown() .await .context(error::ShutdownDatanodeSnafu), ); - record( + record_shutdown_error( self.flownode .shutdown() .await .context(error::ShutdownFlownodeSnafu), ); + // Close the metadata after attempting to stop all components. + record_shutdown_error( + self.metadata_kv_backend + .close() + .await + .context(error::CloseMetadataKvBackendSnafu), + ); + info!("Datanode instance stopped."); first_error.map_or(Ok(()), Err) @@ -543,10 +553,11 @@ impl StartCommand { .context(OtherSnafu)?; let metadata_dir = metadata_store_dir(data_home); - let kv_backend = creator + let metadata_kv_backend = creator .metadata_kv_backend_creator .create(metadata_dir, &opts) .await?; + let kv_backend: KvBackendRef = metadata_kv_backend.clone(); let (procedure_manager, event_recorder_handle) = standalone::build_procedure_manager(kv_backend.clone(), opts.procedure); @@ -790,6 +801,7 @@ impl StartCommand { procedure_manager, leader_services_controller: creator.leader_services_controller, leader_services_context, + metadata_kv_backend, _guard: vec![], }; let result = InstanceCreatorResult { @@ -848,15 +860,25 @@ impl NodeManagerCreator for DefaultNodeManagerCreator { /// configuration types. #[async_trait] pub trait MetadataKvBackendCreator: Send + Sync { - async fn create(&self, metadata_dir: String, opts: &StandaloneOptions) -> Result; + /// Standalone closes the backend on shutdown, so wrappers must forward + /// [close](common_meta::kv_backend::ClosableKvBackend::close) to the backend they wrap. + async fn create( + &self, + metadata_dir: String, + opts: &StandaloneOptions, + ) -> Result; } pub struct DefaultMetadataKvBackendCreator; #[async_trait] impl MetadataKvBackendCreator for DefaultMetadataKvBackendCreator { - async fn create(&self, metadata_dir: String, opts: &StandaloneOptions) -> Result { - standalone::build_metadata_kvbackend(metadata_dir, opts.metadata_store) + async fn create( + &self, + metadata_dir: String, + opts: &StandaloneOptions, + ) -> Result { + standalone::open_metadata_kvbackend(metadata_dir, opts.metadata_store) .context(error::BuildMetadataKvbackendSnafu) } } @@ -1161,6 +1183,7 @@ mod tests { opts.storage.data_home = data_home.path().to_str().unwrap().to_string(); let creator = InstanceCreator::default() .with_leader_services_controller(Box::new(FailingLeaderServices)); + let metadata_dir = metadata_store_dir(&opts.storage.data_home); let (mut instance, _) = StartCommand::build_with(opts, vec![], creator) .await .unwrap(); @@ -1181,6 +1204,11 @@ mod tests { log_store.append_batch(vec![]).await, Err(log_store::error::Error::ObjectStoreWalStopped { .. }) )); + // The metadata backend was closed although earlier steps failed: raft-engine locks its + // directory while open, so reopening it only succeeds once it was released. + let metadata = + standalone::open_metadata_kvbackend(metadata_dir, Default::default()).unwrap(); + metadata.close().await.unwrap(); } #[tokio::test] diff --git a/src/common/meta/src/kv_backend.rs b/src/common/meta/src/kv_backend.rs index 10e51578505..1baa86f9ef1 100644 --- a/src/common/meta/src/kv_backend.rs +++ b/src/common/meta/src/kv_backend.rs @@ -41,6 +41,18 @@ pub mod txn; pub mod util; pub type KvBackendRef = Arc + Send + Sync>; +/// A [KvBackend] holding resources that must be released explicitly on shutdown, because +/// components sharing it may keep it alive until the process exits. +/// +/// Wrappers of a closable backend must forward [ClosableKvBackend::close] to it. +#[async_trait] +pub trait ClosableKvBackend: KvBackend { + /// Releases the backend's resources; any later use fails. + async fn close(&self) -> Result<(), Error>; +} + +pub type ClosableKvBackendRef = Arc; + #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))] // The default meta table name, default is "greptime_metakv". pub const DEFAULT_META_TABLE_NAME: &str = "greptime_metakv"; diff --git a/src/datanode/src/datanode.rs b/src/datanode/src/datanode.rs index f257f9adedb..f035ecb92ea 100644 --- a/src/datanode/src/datanode.rs +++ b/src/datanode/src/datanode.rs @@ -102,6 +102,8 @@ pub struct Datanode { leases_notifier: Option>, plugins: Plugins, object_store_log_store: Option>, + /// The raft-engine WAL, if any. Stopped after the region server, see [Datanode::shutdown]. + raft_engine_log_store: Option>, } impl Datanode { @@ -184,6 +186,13 @@ impl Datanode { ); } record_shutdown_error(&mut first_error, self.region_server.stop().await); + // The stopped engines no longer hold the raft-engine WAL, so it can be closed. + if let Some(log_store) = self.raft_engine_log_store.take() { + record_shutdown_error( + &mut first_error, + close_raft_engine_log_store(log_store).await, + ); + } first_error.map_or(Ok(()), Err) } @@ -213,10 +222,15 @@ pub struct DatanodeBuilder { topic_stats_reporter: Option>, open_regions_writable_override: Option, local_file_access: LocalFileAccess, + /// Set as soon as the raft-engine log store exists so that a failed build can close it. + raft_engine_log_store: Option>, #[cfg(feature = "enterprise")] extension_range_provider_factory: Option, /// Set as soon as the object store log store exists so that a failed build can stop it. object_store_log_store: Option>, + /// Set as soon as the region server exists so that a failed build can stop its engines, + /// which hold the log stores. + region_server: Option, } impl DatanodeBuilder { @@ -230,6 +244,8 @@ impl DatanodeBuilder { cache_registry: None, open_regions_writable_override: None, local_file_access: LocalFileAccess::Disabled, + raft_engine_log_store: None, + region_server: None, #[cfg(feature = "enterprise")] extension_range_provider_factory: None, topic_stats_reporter: None, @@ -304,14 +320,26 @@ impl DatanodeBuilder { self.try_build().await } - /// Builds the datanode and stops the object store log store if a later step fails. + /// Builds the datanode and, if a later step fails, stops the engines and the log stores it + /// opened. async fn try_build(&mut self) -> Result { let result = self.build_inner().await; - if result.is_err() - && let Some(log_store) = &self.object_store_log_store - && let Err(err) = log_store.stop().await - { - warn!(err; "Failed to stop the object store log store after a failed build"); + if result.is_err() { + if let Some(log_store) = &self.object_store_log_store + && let Err(err) = log_store.stop().await + { + warn!(err; "Failed to stop the object store log store after a failed build"); + } + if let Some(region_server) = self.region_server.take() + && let Err(err) = region_server.stop().await + { + warn!(err; "Failed to stop the region server after a failed build"); + } + if let Some(log_store) = self.raft_engine_log_store.take() + && let Err(err) = close_raft_engine_log_store(log_store).await + { + warn!(err; "Failed to close the raft-engine log store after a failed build"); + } } result } @@ -372,6 +400,7 @@ impl DatanodeBuilder { cache_registry.clone(), ) .await?; + self.region_server = Some(region_server.clone()); // A region acknowledges only writes whose WAL append completed within its lease from // the metasrv. if controlled_by_metasrv && let Some(mito) = region_server.mito_engine() { @@ -447,6 +476,7 @@ impl DatanodeBuilder { None }; + self.region_server = None; Ok(Datanode { services: ServerHandlers::default(), heartbeat_task, @@ -456,6 +486,7 @@ impl DatanodeBuilder { leases_notifier, plugins: self.plugins.clone(), object_store_log_store: self.object_store_log_store.take(), + raft_engine_log_store: self.raft_engine_log_store.take(), }) } @@ -759,6 +790,7 @@ impl DatanodeBuilder { let log_store = Self::build_raft_engine_log_store(&opts.storage.data_home, raft_engine_config) .await?; + self.raft_engine_log_store = Some(log_store.clone()); let builder = MitoEngineBuilder::new( &opts.storage.data_home, @@ -919,6 +951,25 @@ pub(crate) fn record_shutdown_error(first_error: &mut Option, resu } } +/// Closes the raft-engine WAL, which takes its last reference: raft-engine only closes the +/// active log file when its engine is dropped. If something else still holds the WAL, it is +/// stopped instead and the file stays open until that reference is dropped. +async fn close_raft_engine_log_store(log_store: Arc) -> Result<()> { + let result = match Arc::try_unwrap(log_store) { + Ok(log_store) => log_store.close().await, + Err(log_store) => { + warn!( + "The raft-engine WAL is still referenced elsewhere ({} references), stopping it without closing its active log file", + Arc::strong_count(&log_store) - 1 + ); + log_store.stop().await + } + }; + result + .map_err(BoxedError::new) + .context(ShutdownInstanceSnafu) +} + /// Returns the object store that holds the WAL objects: the default one for an /// empty `storage_provider`, otherwise the configured provider with that name. fn wal_object_store( @@ -1176,6 +1227,7 @@ mod tests { use common_test_util::temp_dir::{create_named_temp_file, create_temp_dir}; use common_wal::config::DatanodeWalConfig; use common_wal::config::object_store::{AckMode, ObjectStoreWalConfig, STANDALONE_GENERATION}; + use common_wal::config::raft_engine::RaftEngineConfig; use common_wal::options::{ObjectStoreWalOptions, WAL_OPTIONS_KEY, WalOptions}; use log_store::ObjectStoreLogStore; use log_store::error::Error as LogStoreError; @@ -1308,6 +1360,16 @@ mod tests { std::fs::read_dir(dir).unwrap().next().is_none() } + /// Opens the raft-engine WAL under `data_home` again, which only succeeds once the previous + /// engine was released (raft-engine holds a lock on its directory while open). + async fn assert_raft_engine_wal_reopens(data_home: &str) { + let reopened = + DatanodeBuilder::build_raft_engine_log_store(data_home, &RaftEngineConfig::default()) + .await + .unwrap(); + reopened.stop().await.unwrap(); + } + /// Returns true if the store rejects even an empty append, which it does only once stopped. async fn is_stopped(log_store: &ObjectStoreLogStore) -> bool { matches!( @@ -1672,6 +1734,7 @@ mod tests { assert!(datanode.object_store_log_store.is_none()); assert!(data_home.path().join("wal").exists()); datanode.shutdown().await.unwrap(); + assert_raft_engine_wal_reopens(data_home.path().to_str().unwrap()).await; } #[tokio::test] @@ -1743,6 +1806,60 @@ mod tests { assert!(is_stopped(&log_store).await); } + #[tokio::test] + async fn test_shutdown_closes_raft_engine_wal_of_written_region() { + common_telemetry::init_default_ut_logging(); + let data_home = create_temp_dir("raft-engine-wal-written-region"); + let mut datanode = datanode_builder( + data_home.path().to_str().unwrap(), + DatanodeWalConfig::RaftEngine(RaftEngineConfig::default()), + Arc::new(MemoryKvBackend::new()), + ) + .build() + .await + .unwrap(); + let region_server = datanode.region_server(); + let request = CreateRequestBuilder::new().build(); + let schema = rows_schema(&request); + region_server + .handle_request(REGION, RegionRequest::Create(request)) + .await + .unwrap(); + put_rows(®ion_server, &schema, 0..3, false).await; + + datanode.shutdown().await.unwrap(); + + // The open region and its writes must not keep the WAL referenced: if the datanode does + // not hold the last reference, it only stops the WAL and the reopen below fails. + assert_raft_engine_wal_reopens(data_home.path().to_str().unwrap()).await; + } + + #[tokio::test] + async fn test_shutdown_closes_raft_engine_wal_after_engine_failures() { + common_telemetry::init_default_ut_logging(); + let data_home = create_temp_dir("raft-engine-wal-shutdown-failure"); + let builder = datanode_builder( + data_home.path().to_str().unwrap(), + DatanodeWalConfig::RaftEngine(RaftEngineConfig::default()), + Arc::new(MemoryKvBackend::new()), + ); + let mut datanode = builder.build().await.unwrap(); + let (engine, _) = MockRegionEngine::with_custom_apply_fn("failing", |engine| { + engine.handle_stop_mock_fn = Some(Box::new(|| { + error::UnexpectedSnafu { + violated: "stop failed", + } + .fail() + })); + }); + datanode.region_server().register_engine(engine); + + let err = datanode.shutdown().await.unwrap_err(); + + assert_matches!(err, Error::StopRegionEngine { .. }); + assert_raft_engine_wal_reopens(data_home.path().to_str().unwrap()).await; + } + #[tokio::test] async fn test_failed_build_stops_object_store_log_store() { common_telemetry::init_default_ut_logging(); @@ -1763,6 +1880,27 @@ mod tests { assert!(is_stopped(log_store).await); } + #[tokio::test] + async fn test_failed_build_closes_raft_engine_log_store() { + common_telemetry::init_default_ut_logging(); + let data_home = create_temp_dir("raft-engine-wal-failed-build"); + let kv_backend = Arc::new(MemoryKvBackend::new()); + // Regions registered in the metadata but missing on disk fail the region open + // that runs after the log store and the engines are built. + setup_table_datanode(&(kv_backend.clone() as _)).await; + let mut builder = datanode_builder( + data_home.path().to_str().unwrap(), + DatanodeWalConfig::RaftEngine(RaftEngineConfig::default()), + kv_backend, + ); + + assert!(builder.try_build().await.is_err()); + + assert!(builder.region_server.is_none()); + assert!(builder.raft_engine_log_store.is_none()); + assert_raft_engine_wal_reopens(data_home.path().to_str().unwrap()).await; + } + const REGION: RegionId = RegionId::new(1024, 1); fn enqueued_wal_config(flush_interval: Duration) -> ObjectStoreWalConfig { diff --git a/src/log-store/src/error.rs b/src/log-store/src/error.rs index 2fac46efe56..3215db2ce21 100644 --- a/src/log-store/src/error.rs +++ b/src/log-store/src/error.rs @@ -85,6 +85,14 @@ pub enum Error { location: Location, }, + #[snafu(display("Failed to join a blocking task of the log store"))] + Join { + #[snafu(source)] + error: tokio::task::JoinError, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Log store not started yet"))] IllegalState { #[snafu(implicit)] @@ -529,6 +537,7 @@ impl ErrorExt for Error { | InvalidWalEntryRange { .. } => StatusCode::InvalidArguments, StartWalTask { .. } | StopWalTask { .. } + | Join { .. } | IllegalState { .. } | NoMaxValue { .. } | Cast { .. } diff --git a/src/log-store/src/raft_engine.rs b/src/log-store/src/raft_engine.rs index fb9445bfe9c..a1765ffd1c3 100644 --- a/src/log-store/src/raft_engine.rs +++ b/src/log-store/src/raft_engine.rs @@ -12,22 +12,38 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::raft_engine::protos::logstore::EntryImpl; - mod backend; pub mod log_store; +use std::sync::Arc; + pub use backend::RaftEngineBackend; pub use raft_engine::Config; +use raft_engine::Engine; +use snafu::ResultExt; use store_api::logstore::entry::{Entry, NaiveEntry}; use store_api::logstore::provider::Provider; use store_api::storage::RegionId; +use crate::error::{JoinSnafu, Result}; +use crate::raft_engine::protos::logstore::EntryImpl; + #[allow(renamed_and_removed_lints, mismatched_lifetime_syntaxes, unused_parens)] pub mod protos { include!(concat!(env!("OUT_DIR"), concat!("/", "protos/", "mod.rs"))); } +/// Drops `engine` on a blocking thread, as that blocks: dropping the last reference joins the +/// engine's threads and trims and syncs the active log file, which is what closes it. +/// +/// No `Engine::sync` before: on raft-engine 0.4.2 it writes an empty batch, which returns before +/// the fsync (see `durable_sync`). +pub(crate) async fn close_engine(engine: Arc) -> Result<()> { + common_runtime::spawn_blocking_global(move || drop(engine)) + .await + .context(JoinSnafu) +} + impl EntryImpl { pub fn create(id: u64, ns: u64, data: Vec) -> Self { Self { @@ -56,3 +72,23 @@ impl From for Entry { }) } } + +/// Total size of the raft-engine log files in `dir`. +/// +/// raft-engine preallocates the active log file ahead of the writes, where the platform and +/// file system support it, and only trims that tail when the engine is closed. +#[cfg(test)] +pub(crate) fn log_files_size(dir: &std::path::Path) -> u64 { + std::fs::read_dir(dir) + .unwrap() + .map(|entry| entry.unwrap().path()) + .filter(|path| path.extension().is_some_and(|ext| ext == "raftlog")) + .map(|path| std::fs::metadata(path).unwrap().len()) + .sum() +} + +/// Size above which the log files are taken to still hold the tail raft-engine preallocates +/// for the active file. This is a detection threshold, below the 2 MiB raft-engine actually +/// allocates, not that allocation size. +#[cfg(test)] +pub(crate) const PREALLOCATION_THRESHOLD: u64 = 1024 * 1024; diff --git a/src/log-store/src/raft_engine/backend.rs b/src/log-store/src/raft_engine/backend.rs index 461465cf18e..1ed3b4315f8 100644 --- a/src/log-store/src/raft_engine/backend.rs +++ b/src/log-store/src/raft_engine/backend.rs @@ -23,7 +23,7 @@ use common_config::KvBackendConfig; use common_error::ext::BoxedError; use common_meta::error as meta_error; use common_meta::kv_backend::txn::{Txn, TxnOp, TxnOpResponse, TxnRequest, TxnResponse}; -use common_meta::kv_backend::{KvBackend, TxnService}; +use common_meta::kv_backend::{ClosableKvBackend, KvBackend, TxnService}; use common_meta::rpc::KeyValue; use common_meta::rpc::store::{ BatchDeleteRequest, BatchDeleteResponse, BatchGetRequest, BatchGetResponse, BatchPutRequest, @@ -33,17 +33,19 @@ use common_meta::rpc::store::{ use common_meta::util::get_next_prefix_key; use common_runtime::RepeatedTask; use raft_engine::{Config, Engine, LogBatch, ReadableSize, RecoveryMode}; -use snafu::{IntoError, ResultExt}; +use snafu::{IntoError, OptionExt, ResultExt}; -use crate::error::{self, Error, IoSnafu, RaftEngineSnafu, StartWalTaskSnafu}; +use crate::error::{self, Error, IoSnafu, RaftEngineSnafu, StartWalTaskSnafu, StopWalTaskSnafu}; +use crate::raft_engine::close_engine; use crate::raft_engine::log_store::PurgeExpiredFilesFunction; pub(crate) const SYSTEM_NAMESPACE: u64 = 0; /// RaftEngine based [KvBackend] implementation. pub struct RaftEngineBackend { - engine: RwLock>, - _gc_task: RepeatedTask, + /// `None` once the backend is closed, see [ClosableKvBackend::close]. + engine: RwLock>>, + gc_task: RepeatedTask, } fn ensure_dir(dir: &str) -> error::Result<()> { @@ -88,7 +90,7 @@ impl RaftEngineBackend { let gc_task = RepeatedTask::new( config.purge_interval, Box::new(PurgeExpiredFilesFunction { - engine: engine.clone(), + engine: Arc::downgrade(&engine), }), ); gc_task @@ -96,12 +98,37 @@ impl RaftEngineBackend { .context(StartWalTaskSnafu { name: "gc_task" })?; Ok(Self { - engine: RwLock::new(engine), - _gc_task: gc_task, + engine: RwLock::new(Some(engine)), + gc_task, }) } } +/// raft-engine only trims the space it preallocated at the tail of the active log file when +/// the [Engine] is dropped. This backend is shared by many components, some of which live in +/// background tasks until the process exits, so without an explicit close the next start +/// finds that tail and truncates it as a broken batch. +#[async_trait::async_trait] +impl ClosableKvBackend for RaftEngineBackend { + /// Stops the purge task and closes the engine. + async fn close(&self) -> meta_error::Result<()> { + self.gc_task + .stop() + .await + .context(StopWalTaskSnafu { name: "gc_task" }) + .map_err(BoxedError::new) + .context(meta_error::ExternalSnafu)?; + let engine = self.engine.write().unwrap().take(); + match engine { + Some(engine) => close_engine(engine) + .await + .map_err(BoxedError::new) + .context(meta_error::ExternalSnafu), + None => Ok(()), + } + } +} + #[async_trait::async_trait] impl TxnService for RaftEngineBackend { type Error = meta_error::Error; @@ -119,10 +146,11 @@ impl TxnService for RaftEngineBackend { // outside "get" or "put" etc. It doesn't serve the purpose of mutating some Rust data, so // the variable is not "mut". Suppress the clippy warning because of this. #[allow(clippy::readonly_write_lock)] - let engine = self.engine.write().unwrap(); + let guard = self.engine.write().unwrap(); + let engine = get_open_engine(&guard)?; for cmp in compare { - let existing_value = engine_get(&engine, &cmp.key)?.map(|kv| kv.value); + let existing_value = engine_get(engine, &cmp.key)?.map(|kv| kv.value); if !cmp.compare_value(existing_value.as_ref()) { succeeded = false; break; @@ -141,7 +169,7 @@ impl TxnService for RaftEngineBackend { } TxnOp::Get(key) => { - let value = engine_get(&engine, &key)?.map(|kv| kv.value); + let value = engine_get(engine, &key)?.map(|kv| kv.value); let kvs = value .map(|value| KeyValue { key, value }) .into_iter() @@ -153,7 +181,7 @@ impl TxnService for RaftEngineBackend { } TxnOp::Delete(key) => { - let prev = engine_get(&engine, &key)?; + let prev = engine_get(engine, &key)?; batch.delete(SYSTEM_NAMESPACE, key); let deleted = if prev.is_some() { 1 } else { 0 }; Ok(TxnOpResponse::ResponseDelete(DeleteRangeResponse { @@ -212,9 +240,7 @@ impl KvBackend for RaftEngineBackend { let mut more = false; let mut iter = 0; - self.engine - .read() - .unwrap() + get_open_engine(&self.engine.read().unwrap())? .scan_raw_messages( SYSTEM_NAMESPACE, start_key.as_deref(), @@ -251,11 +277,12 @@ impl KvBackend for RaftEngineBackend { let mut prev = None; // Engine::write assures that one batch is written atomically. The read/write lock is // just to prevent race condition between put and txn. - let engine = self.engine.read().unwrap(); + let guard = self.engine.read().unwrap(); + let engine = get_open_engine(&guard)?; if prev_kv { - prev = engine_get(&engine, &key)?; + prev = engine_get(engine, &key)?; } - engine_put(&engine, key, value)?; + engine_put(engine, key, value)?; Ok(PutResponse { prev_kv: prev }) } @@ -269,9 +296,10 @@ impl KvBackend for RaftEngineBackend { vec![] }; - let engine = self.engine.read().unwrap(); + let guard = self.engine.read().unwrap(); + let engine = get_open_engine(&guard)?; for kv in kvs { - if prev_kv && let Some(kv) = engine_get(&engine, &kv.key)? { + if prev_kv && let Some(kv) = engine_get(engine, &kv.key)? { prev_kvs.push(kv); } batch @@ -295,7 +323,9 @@ impl KvBackend for RaftEngineBackend { kvs: Vec::with_capacity(req.keys.len()), }; for key in req.keys { - let Some(value) = self.engine.read().unwrap().get(SYSTEM_NAMESPACE, &key) else { + let Some(value) = + get_open_engine(&self.engine.read().unwrap())?.get(SYSTEM_NAMESPACE, &key) + else { continue; }; response.kvs.push(KeyValue { key, value }); @@ -324,9 +354,10 @@ impl KvBackend for RaftEngineBackend { let mut prev_kvs = vec![]; let mut deleted = 0; - let engine = self.engine.read().unwrap(); + let guard = self.engine.read().unwrap(); + let engine = get_open_engine(&guard)?; for kv in range_resp.kvs { - engine_delete(&engine, &kv.key)?; + engine_delete(engine, &kv.key)?; if prev_kv { prev_kvs.push(kv); } @@ -348,9 +379,10 @@ impl KvBackend for RaftEngineBackend { vec![] }; let mut batch = LogBatch::with_capacity(keys.len()); - let engine = self.engine.read().unwrap(); + let guard = self.engine.read().unwrap(); + let engine = get_open_engine(&guard)?; for key in keys { - if prev_kv && let Some(prev) = engine_get(&engine, &key)? { + if prev_kv && let Some(prev) = engine_get(engine, &key)? { prev_kvs.push(prev); } batch.delete(SYSTEM_NAMESPACE, key); @@ -364,25 +396,35 @@ impl KvBackend for RaftEngineBackend { } async fn get(&self, key: &[u8]) -> Result, Self::Error> { - engine_get(&self.engine.read().unwrap(), key) + engine_get(get_open_engine(&self.engine.read().unwrap())?, key) } async fn exists(&self, key: &[u8]) -> Result { - Ok(engine_get(&self.engine.read().unwrap(), key)?.is_some()) + Ok(engine_get(get_open_engine(&self.engine.read().unwrap())?, key)?.is_some()) } async fn delete(&self, key: &[u8], prev_kv: bool) -> Result, Self::Error> { - let engine = self.engine.read().unwrap(); + let guard = self.engine.read().unwrap(); + let engine = get_open_engine(&guard)?; let prev = if prev_kv { - engine_get(&engine, key)? + engine_get(engine, key)? } else { None }; - engine_delete(&engine, key)?; + engine_delete(engine, key)?; Ok(prev) } } +/// Returns the engine, or an error once the backend is closed. +fn get_open_engine(engine: &Option>) -> meta_error::Result<&Engine> { + engine + .as_deref() + .context(error::IllegalStateSnafu) + .map_err(BoxedError::new) + .context(meta_error::ExternalSnafu) +} + fn engine_get(engine: &Engine, key: &[u8]) -> meta_error::Result> { let res = engine.get(SYSTEM_NAMESPACE, key); Ok(res.map(|value| KeyValue { @@ -430,11 +472,69 @@ mod tests { use common_test_util::temp_dir::create_temp_dir; use super::*; + use crate::raft_engine::{PREALLOCATION_THRESHOLD, log_files_size}; fn build_kv_backend(dir: String) -> RaftEngineBackend { RaftEngineBackend::try_open_with_cfg(dir, &KvBackendConfig::default()).unwrap() } + #[tokio::test] + async fn test_close_closes_active_log_file() { + let dir = create_temp_dir("raft-engine-kv-close-test"); + let backend = build_kv_backend(dir.path().to_str().unwrap().to_string()); + backend + .put(PutRequest { + key: b"key".to_vec(), + value: b"value".to_vec(), + prev_kv: false, + }) + .await + .unwrap(); + let preallocated = log_files_size(dir.path()) >= PREALLOCATION_THRESHOLD; + + backend.close().await.unwrap(); + if preallocated { + assert!(log_files_size(dir.path()) < PREALLOCATION_THRESHOLD); + } + assert!(backend.get(b"key").await.is_err()); + + // Closing released the engine, so the directory can be reopened with the data intact. + let reopened = build_kv_backend(dir.path().to_str().unwrap().to_string()); + assert_eq!( + b"value".to_vec(), + reopened.get(b"key").await.unwrap().unwrap().value + ); + reopened.close().await.unwrap(); + } + + #[tokio::test] + async fn test_drop_closes_active_log_file() { + let dir = create_temp_dir("raft-engine-kv-drop-test"); + let backend = build_kv_backend(dir.path().to_str().unwrap().to_string()); + backend + .put(PutRequest { + key: b"key".to_vec(), + value: b"value".to_vec(), + prev_kv: false, + }) + .await + .unwrap(); + let preallocated = log_files_size(dir.path()) >= PREALLOCATION_THRESHOLD; + + // Callers that own the backend, like the CLI tools, close it by dropping it: the + // purge task on the global runtime must not keep the engine alive. + drop(backend); + if preallocated { + assert!(log_files_size(dir.path()) < PREALLOCATION_THRESHOLD); + } + let reopened = build_kv_backend(dir.path().to_str().unwrap().to_string()); + assert_eq!( + b"value".to_vec(), + reopened.get(b"key").await.unwrap().unwrap().value + ); + reopened.close().await.unwrap(); + } + #[tokio::test] async fn test_raft_engine_kv() { let dir = create_temp_dir("raft-engine-kv"); diff --git a/src/log-store/src/raft_engine/log_store.rs b/src/log-store/src/raft_engine/log_store.rs index dcc1109b569..f4e0d28cf9b 100644 --- a/src/log-store/src/raft_engine/log_store.rs +++ b/src/log-store/src/raft_engine/log_store.rs @@ -14,7 +14,8 @@ use std::collections::{HashMap, hash_map}; use std::fmt::{Debug, Formatter}; -use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Weak}; use std::time::Duration; use async_stream::stream; @@ -27,6 +28,8 @@ use store_api::logstore::entry::{Entry, Id as EntryId, NaiveEntry}; use store_api::logstore::provider::{Provider, RaftEngineProvider}; use store_api::logstore::{AppendBatchResponse, LogStore, SendableEntryStream}; use store_api::storage::RegionId; +use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; use crate::error::{ AddEntryLogBatchSnafu, DiscontinuousLogIndexSnafu, Error, FetchEntrySnafu, @@ -35,6 +38,7 @@ use crate::error::{ }; use crate::metrics; use crate::raft_engine::backend::SYSTEM_NAMESPACE; +use crate::raft_engine::close_engine; use crate::raft_engine::protos::logstore::{EntryImpl, NamespaceImpl}; const NAMESPACE_PREFIX: &str = "$sys/"; @@ -44,12 +48,21 @@ pub struct RaftEngineLogStore { sync_period: Option, read_batch_size: usize, engine: Arc, + /// Tasks feeding the streams returned by [LogStore::read]. Each holds the engine until its + /// stream is drained, so `stop` cancels them and waits for them before [Self::close] can + /// release it. + readers: TaskTracker, + readers_cancel: CancellationToken, gc_task: RepeatedTask, sync_task: RepeatedTask, } +/// Purges expired log files periodically. +/// +/// Holds the engine weakly: the task runs on the global runtime, which outlives its owner, and +/// a strong reference there would keep the engine (and its active log file) open until exit. pub struct PurgeExpiredFilesFunction { - pub engine: Arc, + pub engine: Weak, } #[async_trait::async_trait] @@ -59,7 +72,10 @@ impl TaskFunction for PurgeExpiredFilesFunction { } async fn call(&mut self) -> Result<()> { - match self.engine.purge_expired_files().context(RaftEngineSnafu) { + let Some(engine) = self.engine.upgrade() else { + return Ok(()); + }; + match engine.purge_expired_files().context(RaftEngineSnafu) { Ok(res) => { // TODO(hl): the retval of purge_expired_files indicates the namespaces need to be compact, // which is useful when monitoring regions failed to flush it's memtable to SSTs. @@ -106,8 +122,9 @@ fn durable_sync(engine: &Engine, batch: &mut LogBatch) -> raft_engine::Result, + engine: Weak, /// Batch reused across ticks to avoid allocating a new one each period. /// `None` after a failed sync; a fresh batch is created on the next tick. batch: Option, @@ -116,7 +133,9 @@ pub struct SyncWalTaskFunction { #[async_trait::async_trait] impl TaskFunction for SyncWalTaskFunction { async fn call(&mut self) -> std::result::Result<(), Error> { - let engine = self.engine.clone(); + let Some(engine) = self.engine.upgrade() else { + return Ok(()); + }; let mut batch = self .batch .take() @@ -145,9 +164,9 @@ impl TaskFunction for SyncWalTaskFunction { } impl SyncWalTaskFunction { - pub fn new(engine: Arc) -> Self { + pub fn new(engine: &Arc) -> Self { Self { - engine, + engine: Arc::downgrade(engine), batch: None, } } @@ -170,13 +189,13 @@ impl RaftEngineLogStore { let gc_task = RepeatedTask::new( config.purge_interval, Box::new(PurgeExpiredFilesFunction { - engine: engine.clone(), + engine: Arc::downgrade(&engine), }), ); let sync_task = RepeatedTask::new( config.sync_period.unwrap_or(Duration::from_secs(5)), - Box::new(SyncWalTaskFunction::new(engine.clone())), + Box::new(SyncWalTaskFunction::new(&engine)), ); let log_store = Self { @@ -184,6 +203,8 @@ impl RaftEngineLogStore { sync_period: config.sync_period, read_batch_size: config.read_batch_size, engine, + readers: TaskTracker::new(), + readers_cancel: CancellationToken::new(), gc_task, sync_task, }; @@ -195,6 +216,17 @@ impl RaftEngineLogStore { self.gc_task.started() } + /// Stops the log store and closes the engine. + /// + /// raft-engine only trims the space it preallocated at the tail of the active log file when + /// the [Engine] is dropped. Consuming the log store makes this the last reference, so the + /// file is closed here instead of whenever the engine happens to be dropped, which may not + /// happen before the process exits. + pub async fn close(self) -> Result<()> { + self.stop().await?; + close_engine(self.engine).await + } + fn start(&self) -> Result<()> { self.gc_task .start(common_runtime::global_runtime()) @@ -215,7 +247,7 @@ impl RaftEngineLogStore { /// Returns the `LogBatch` converted along with the last entry id /// to append in each namespace(region). fn entries_to_batch( - &self, + engine: &Engine, entries: Vec, ) -> Result<(LogBatch, HashMap)> { // Records the last entry id for each region's entries. @@ -240,7 +272,7 @@ impl RaftEngineLogStore { } hash_map::Entry::Vacant(v) => { // this entry is the first in batch of given region. - if let Some(first_index) = self.engine.first_index(region_id.as_u64()) { + if let Some(first_index) = engine.first_index(region_id.as_u64()) { // ensure the first in batch does not override compacted entry. ensure!( entry_id > first_index, @@ -252,7 +284,7 @@ impl RaftEngineLogStore { ); } // ensure the first in batch does not form a hole in raft-engine. - if let Some(last_index) = self.engine.last_index(region_id.as_u64()) { + if let Some(last_index) = engine.last_index(region_id.as_u64()) { ensure!( entry_id == last_index + 1, DiscontinuousLogIndexSnafu { @@ -305,7 +337,15 @@ impl LogStore for RaftEngineLogStore { self.sync_task .stop() .await - .context(StopWalTaskSnafu { name: "sync_task" }) + .context(StopWalTaskSnafu { name: "sync_task" })?; + + // Readers hold the engine until their streams are drained, which a caller that keeps + // a stream may never do: cancel them and wait for them to finish. + self.readers_cancel.cancel(); + self.readers.close(); + self.readers.wait().await; + + Ok(()) } /// Appends a batch of entries to logstore. `RaftEngineLogStore` assures the atomicity of @@ -324,7 +364,7 @@ impl LogStore for RaftEngineLogStore { return Ok(AppendBatchResponse::default()); } - let (mut batch, last_entry_ids) = self.entries_to_batch(entries)?; + let (mut batch, last_entry_ids) = Self::entries_to_batch(&self.engine, entries)?; let _ = self .engine .write(&mut batch, self.sync_write) @@ -364,7 +404,14 @@ impl LogStore for RaftEngineLogStore { ); let max_batch_size = self.read_batch_size; let (tx, mut rx) = tokio::sync::mpsc::channel(max_batch_size); - let _handle = common_runtime::spawn_global(async move { + let cancel = self.readers_cancel.clone(); + // Set when the read is cancelled before the last entry, so the stream ends with an error + // instead of an end of stream that would look like the whole WAL was read. A flag rather + // than an error sent on the channel: the reader is usually cancelled while blocked on a + // full channel, where there is no room for one. + let truncated = Arc::new(AtomicBool::new(false)); + let reader_truncated = truncated.clone(); + let _handle = common_runtime::spawn_global(self.readers.track_future(async move { while start_index <= last_index { let mut vec = Vec::with_capacity(max_batch_size); match engine @@ -385,18 +432,30 @@ impl LogStore for RaftEngineLogStore { if let Some(last_entry) = vec.last() { start_index = last_entry.id + 1; } - // reader side closed, cancel following reads - if tx.send(Ok(vec)).await.is_err() { + // Stop on shutdown, or when the reader side is closed. + let sent = tokio::select! { + biased; + _ = cancel.cancelled() => { + reader_truncated.store(true, Ordering::Release); + false + } + sent = tx.send(Ok(vec)) => sent.is_ok(), + }; + if !sent { break; } } Err(e) => { - let _ = tx.send(Err(e)).await; + tokio::select! { + biased; + _ = cancel.cancelled() => reader_truncated.store(true, Ordering::Release), + _ = tx.send(Err(e)) => {} + } break; } } } - }); + })); let s = stream!({ while let Some(res) = rx.recv().await { @@ -404,6 +463,10 @@ impl LogStore for RaftEngineLogStore { yield Ok(res.into_iter().map(Entry::from).collect::>()); } + // The channel closes once the reader task ends, after it set the flag. + if truncated.load(Ordering::Acquire) { + yield IllegalStateSnafu.fail(); + } }); Ok(Box::pin(s)) } @@ -587,6 +650,7 @@ mod tests { use crate::error::Error; use crate::raft_engine::log_store::RaftEngineLogStore; use crate::raft_engine::protos::logstore::EntryImpl; + use crate::raft_engine::{PREALLOCATION_THRESHOLD, log_files_size}; #[tokio::test] async fn test_open_logstore() { @@ -656,7 +720,7 @@ mod tests { .await .unwrap(); - let mut task = SyncWalTaskFunction::new(logstore.engine.clone()); + let mut task = SyncWalTaskFunction::new(&logstore.engine); for _ in 0..3 { task.call().await.unwrap(); let batch = task.batch.as_ref().expect("batch is kept after a sync"); @@ -768,6 +832,103 @@ mod tests { res } + #[tokio::test] + async fn test_close_with_retained_read_stream() { + let dir = create_temp_dir("raft-engine-logstore-retained-reader-test"); + // One entry per batch and a one-batch channel: the reader blocks on its second send. + let config = RaftEngineConfig { + read_batch_size: 1, + ..Default::default() + }; + let logstore = + RaftEngineLogStore::try_new(dir.path().to_str().unwrap().to_string(), &config) + .await + .unwrap(); + for id in 1..=4 { + logstore + .append(EntryImpl::create(id, 1, id.to_string().into_bytes()).into()) + .await + .unwrap(); + } + // Kept but never consumed, as a caller that holds a stream across shutdown would. + let stream = logstore + .read(&Provider::raft_engine_provider(1), 1) + .await + .unwrap(); + let preallocated = log_files_size(dir.path()) >= PREALLOCATION_THRESHOLD; + + tokio::time::timeout(Duration::from_secs(10), logstore.close()) + .await + .expect("close must not wait for the stream to be consumed") + .unwrap(); + if preallocated { + assert!(log_files_size(dir.path()) < PREALLOCATION_THRESHOLD); + } + + // The reader released the engine, so the directory can be reopened with all entries. + let reopened = + RaftEngineLogStore::try_new(dir.path().to_str().unwrap().to_string(), &config) + .await + .unwrap(); + let entries = collect_entries( + reopened + .read(&Provider::raft_engine_provider(1), 1) + .await + .unwrap(), + ) + .await; + assert_eq!(4, entries.len()); + reopened.close().await.unwrap(); + + // The cancelled read ends with an error, not with an end of stream that would pass for a + // complete read of the WAL. + let batches = stream.collect::>().await; + assert!(batches.len() < 4); + assert!(matches!( + batches.last(), + Some(Err(Error::IllegalState { .. })) + )); + } + + #[tokio::test] + async fn test_close_closes_active_log_file() { + let dir = create_temp_dir("raft-engine-logstore-close-test"); + let logstore = RaftEngineLogStore::try_new( + dir.path().to_str().unwrap().to_string(), + &RaftEngineConfig::default(), + ) + .await + .unwrap(); + logstore + .append(EntryImpl::create(1, 1, "1".as_bytes().to_vec()).into()) + .await + .unwrap(); + let preallocated = log_files_size(dir.path()) >= PREALLOCATION_THRESHOLD; + + logstore.close().await.unwrap(); + if preallocated { + assert!(log_files_size(dir.path()) < PREALLOCATION_THRESHOLD); + } + + // Closing released the engine, so the directory can be reopened with the entry intact. + let reopened = RaftEngineLogStore::try_new( + dir.path().to_str().unwrap().to_string(), + &RaftEngineConfig::default(), + ) + .await + .unwrap(); + let entries = collect_entries( + reopened + .read(&Provider::raft_engine_provider(1), 1) + .await + .unwrap(), + ) + .await; + assert_eq!(1, entries.len()); + assert_eq!(1, entries[0].entry_id()); + reopened.close().await.unwrap(); + } + #[tokio::test] async fn test_reopen() { let dir = create_temp_dir("raft-engine-logstore-reopen-test"); diff --git a/src/mito2/src/engine.rs b/src/mito2/src/engine.rs index d22cd3f9af2..7e215cf31a2 100644 --- a/src/mito2/src/engine.rs +++ b/src/mito2/src/engine.rs @@ -105,7 +105,7 @@ mod puffin_index; use std::any::Any; use std::collections::{HashMap, HashSet}; -use std::sync::Arc; +use std::sync::{Arc, RwLock}; use std::time::Instant; use api::region::RegionResponse; @@ -156,9 +156,9 @@ use crate::compaction::CompactionState; use crate::config::{MitoConfig, ScanMemoryMode}; use crate::engine::puffin_index::{IndexEntryContext, collect_index_entries_from_puffin}; use crate::error::{ - IncrementalQueryStaleSnafu, InvalidRequestSnafu, JoinSnafu, MitoManifestInfoSnafu, RecvSnafu, - RegionNotFoundSnafu, Result, SequenceRangeUnsupportedSnafu, SerdeJsonSnafu, - SerializeColumnMetadataSnafu, SnapshotFenceStaleSnafu, + EngineStoppedSnafu, IncrementalQueryStaleSnafu, InvalidRequestSnafu, JoinSnafu, + MitoManifestInfoSnafu, RecvSnafu, RegionNotFoundSnafu, Result, SequenceRangeUnsupportedSnafu, + SerdeJsonSnafu, SerializeColumnMetadataSnafu, SnapshotFenceStaleSnafu, }; #[cfg(feature = "enterprise")] use crate::extension::BoxedExtensionRangeProviderFactory; @@ -274,7 +274,7 @@ impl<'a, S: LogStore> MitoEngineBuilder<'a, S> { let inner = EngineInner { workers, config, - wal_raw_entry_reader, + wal_raw_entry_reader: RwLock::new(Some(wal_raw_entry_reader)), scan_memory_tracker, region_hook, #[cfg(feature = "enterprise")] @@ -855,8 +855,9 @@ struct EngineInner { workers: WorkerGroup, /// Config of the engine. config: Arc, - /// The Wal raw entry reader. - wal_raw_entry_reader: Arc, + /// The Wal raw entry reader. `None` once the engine is stopped, so that it no longer holds + /// the log store and the owner of the log store can close it. + wal_raw_entry_reader: RwLock>>, /// Memory tracker for table scans. scan_memory_tracker: QueryMemoryTracker, /// The region hook (if any) registered via plugins; exposed for the GC worker @@ -909,7 +910,19 @@ impl EngineInner { /// Stop the inner engine. async fn stop(&self) -> Result<()> { - self.workers.stop().await + self.workers.stop().await?; + // Release the log store once the workers no longer use it. + let _ = self.wal_raw_entry_reader.write().unwrap().take(); + Ok(()) + } + + /// Returns the Wal raw entry reader, or an error once the engine is stopped. + fn wal_raw_entry_reader(&self) -> Result> { + self.wal_raw_entry_reader + .read() + .unwrap() + .clone() + .context(EngineStoppedSnafu) } fn find_region(&self, region_id: RegionId) -> Result { @@ -940,7 +953,7 @@ impl EngineInner { let provider = Provider::kafka_provider(topic); let (distributor, entry_receivers) = build_wal_entry_distributor_and_receivers( provider.clone(), - self.wal_raw_entry_reader.clone(), + self.wal_raw_entry_reader()?, ®ion_ids, DEFAULT_ENTRY_RECEIVER_BUFFER_SIZE, ); @@ -1036,7 +1049,7 @@ impl EngineInner { .collect::>(); let (distributor, entry_receivers) = build_wal_entry_distributor_and_receivers( provider.clone(), - self.wal_raw_entry_reader.clone(), + self.wal_raw_entry_reader()?, ®ion_ids, DEFAULT_ENTRY_RECEIVER_BUFFER_SIZE, ); @@ -1821,7 +1834,7 @@ impl MitoEngine { ) .await?, config, - wal_raw_entry_reader, + wal_raw_entry_reader: RwLock::new(Some(wal_raw_entry_reader)), scan_memory_tracker, region_hook: None, #[cfg(feature = "enterprise")] diff --git a/src/mito2/src/error.rs b/src/mito2/src/error.rs index bba682f555a..88c0203beb5 100644 --- a/src/mito2/src/error.rs +++ b/src/mito2/src/error.rs @@ -212,6 +212,12 @@ pub enum Error { location: Location, }, + #[snafu(display("Engine is stopped"))] + EngineStopped { + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Failed to recv result"))] Recv { #[snafu(source)] @@ -1651,6 +1657,7 @@ impl ErrorExt for Error { | RegionMetadataNotFound { .. } | Join { .. } | WorkerStopped { .. } + | EngineStopped { .. } | Recv { .. } | DecodeWal { .. } | ComputeArrow { .. } diff --git a/src/standalone/src/lib.rs b/src/standalone/src/lib.rs index 57c4efee249..730e768de05 100644 --- a/src/standalone/src/lib.rs +++ b/src/standalone/src/lib.rs @@ -21,5 +21,5 @@ pub mod procedure; pub use datanode_manager::StandaloneDatanodeManager; pub use information_extension::StandaloneInformationExtension; -pub use metadata::{build_metadata_kv_from_url, build_metadata_kvbackend}; +pub use metadata::{build_metadata_kv_from_url, build_metadata_kvbackend, open_metadata_kvbackend}; pub use procedure::{StandaloneRepartitionProcedureFactory, build_procedure_manager}; diff --git a/src/standalone/src/metadata.rs b/src/standalone/src/metadata.rs index 9ff98829da5..dc144f579d3 100644 --- a/src/standalone/src/metadata.rs +++ b/src/standalone/src/metadata.rs @@ -16,7 +16,7 @@ use std::sync::Arc; use common_config::KvBackendConfig; use common_error::ext::BoxedError; -use common_meta::kv_backend::KvBackendRef; +use common_meta::kv_backend::{ClosableKvBackendRef, KvBackendRef}; use common_telemetry::{debug, info}; use log_store::raft_engine::RaftEngineBackend; use snafu::{ResultExt, ensure}; @@ -24,8 +24,11 @@ use url::Url; use crate::error::{InvalidUrlSchemeSnafu, OpenMetadataKvBackendSnafu, ParseUrlSnafu, Result}; -/// Builds the metadata kvbackend. -pub fn build_metadata_kvbackend(dir: String, config: KvBackendConfig) -> Result { +/// Opens the metadata kvbackend, which the caller has to close on shutdown. +pub fn open_metadata_kvbackend( + dir: String, + config: KvBackendConfig, +) -> Result { info!( "Creating metadata kvbackend with dir: {}, config: {:?}", dir, config @@ -37,6 +40,11 @@ pub fn build_metadata_kvbackend(dir: String, config: KvBackendConfig) -> Result< Ok(Arc::new(kv_backend)) } +/// Builds the metadata kvbackend. +pub fn build_metadata_kvbackend(dir: String, config: KvBackendConfig) -> Result { + open_metadata_kvbackend(dir, config).map(|kv_backend| kv_backend as KvBackendRef) +} + /// Builds the metadata kvbackend from a list of URLs. pub fn build_metadata_kv_from_url(url: &str) -> Result { debug!("Building metadata kvbackend from url: {}", url);