rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-14 23:06:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(log-store): close raft-engine engines on shutdown (#9518)

Browse Source 
* fix(log-store): close raft-engine engines on shutdown

raft-engine preallocates the active log file and only trims that tail when the
Engine is dropped. GreptimeDB never dropped its engines before exiting, so every
start after a clean shutdown truncated the tail as a broken batch and logged
"Truncating log file due to broken batch".

RaftEngineBackend::close and RaftEngineLogStore::stop now release the engine;
standalone closes the metadata backend last, and the datanode stops the
raft-engine WAL after the region server.

Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

* refactor: address review on raft-engine shutdown

- Add ClosableKvBackend to common-meta; the standalone metadata backend creator
  returns one and Instance closes it explicitly instead of downcasting to
  RaftEngineBackend, so wrappers forward the close.
- Record a failed metadata backend close as a shutdown error.
- Sync and drop the engine on a blocking thread.
- Only check tail trimming when raft-engine preallocated, and check reopening
  after close/stop on every platform.

Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

* fix: release raft-engine engines on every shutdown path

- Stop the raft-engine WAL when the datanode build fails after opening it.
- Purge and sync tasks hold the engine weakly, so dropping the last backend or
  log store reference (as the CLI tools do) closes the engine.
- Fix needless borrows reported by clippy.

Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

* fix: drop unused ClosableKvBackend import

Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

* fix: cancel log store readers on stop; address review

- RaftEngineLogStore tracks the tasks feeding read streams and cancels and
  joins them on stop, so a retained, unconsumed stream no longer keeps the
  engine open past shutdown.
- Tests: stop with a retained read stream; datanode shutdown with a failing
  region engine and after a normal shutdown reopens the raft-engine WAL;
  standalone stop with failing steps reopens the metadata directory.
- entries_to_batch no longer takes self; rename opened to get_open_engine,
  log_files_len to log_files_size and PREALLOCATED_LEN to
  PREALLOCATION_THRESHOLD; mod declarations before imports; rename the
  standalone shutdown closure to record_shutdown_error.

Model: claude-opus-5-5
Tokens: 15.8k input, 148k output, 2.9M cache write, 103M cache read
Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

* fix: update Cargo.lock for tokio-util's rt feature

The rt feature makes tokio-util depend on futures-util, which the lock file
has to record for `cargo check --locked`.

Model: claude-opus-5-5
Tokens: 15.9k input, 162k output, 3.0M cache write, 124M cache read
Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

* refactor: close the raft-engine WAL by ownership instead of a lock

Per review, RaftEngineLogStore holds the engine as a plain Arc<Engine> again:

- MitoEngine drops its Wal raw entry reader once its workers are stopped,
  so it no longer holds the log store after stop.
- RaftEngineLogStore::close(self) stops the tasks and readers and closes the
  engine. Datanode::shutdown takes the log store after stopping the region
  server and closes it if it holds the last reference; otherwise it logs a
  warning and only stops it.
- A failed build stops the region server it created before closing the log
  store, so the engines release it there too.

Model: claude-opus-5-5
Effort: medium
Tokens: 78, 19161, 107293, 4916570 del commit
Cost: 2.23 USD del commit
Session: 82696960-1984-4432-bcf1-33bfece8b526
Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

* test: drop the index argument from LogStore::read in the raft-engine tests

#9563 removed it from the trait.

Model: claude-opus-5-5
Effort: medium
Tokens: 16, 3715, 6600, 1706794 del commit
Cost: 0.47 USD del commit
Session: 82696960-1984-4432-bcf1-33bfece8b526
Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

* fix: end a cancelled WAL read with an error; test a written region

- A raft-engine read cancelled by stop now ends its stream with an
  IllegalState error instead of an end of stream, so a replay cannot take a
  truncated WAL as fully read. The reader flags the cancel instead of sending
  the error, since it is usually blocked on a full channel at that point.
- Datanode test: a raft-engine datanode with a created region and writes
  shuts down cleanly and its WAL reopens, which fails if anything still
  holds the log store.
- Move the doc comment of is_stopped back onto it.

Model: claude-opus-5-5
Effort: medium
Tokens: 42, 10316, 407027, 4108529 del commit
Cost: 4.28 USD del commit
Session: 82696960-1984-4432-bcf1-33bfece8b526
Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

* fix(log-store): drop the no-op sync before closing the raft-engine engine

Per review: on raft-engine 0.4.2 `Engine::sync` writes an empty batch, and
`Engine::write` returns before the fsync, so the sync in `close_engine` read
as a final flush that never ran. The active log file is trimmed and synced
by `LogFileWriter::close` when the last reference is dropped, so
`close_engine` now only drops it on a blocking thread.

Model: claude-opus-5-5
Effort: xhigh
Tokens: 120, 33613, 182918, 7511982 del commit
Cost: 3.64 USD del commit
Session: 57ca77a8-f6fe-4fa2-a55a-dda31e5c541f
Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>

---------

Signed-off-by: Silvio <62994417+AbstractBike@users.noreply.github.com>
This commit is contained in:
Silvio
2026-11-12 01:09:42 +00:00
committed by GitHub
parent 15fd97e369
commit 6dbff8a081
12 changed files with 603 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cmd/src/error.rs
+8
View File
@@ -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(),
src/cmd/src/standalone.rs
+38 -10
View File
@@ -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<dyn StandaloneLeaderServicesController>,
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<WorkerGuard>,
}
@@ -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<KvBackendRef>;
/// 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<ClosableKvBackendRef>;
}
pub struct DefaultMetadataKvBackendCreator;
#[async_trait]
impl MetadataKvBackendCreator for DefaultMetadataKvBackendCreator {
async fn create(&self, metadata_dir: String, opts: &StandaloneOptions) -> Result<KvBackendRef> {
standalone::build_metadata_kvbackend(metadata_dir, opts.metadata_store)
async fn create(
&self,
metadata_dir: String,
opts: &StandaloneOptions,
) -> Result<ClosableKvBackendRef> {
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]
src/common/meta/src/kv_backend.rs
+12
View File
@@ -41,6 +41,18 @@ pub mod txn;
pub mod util;
pub type KvBackendRef<E = Error> = Arc<dyn KvBackend<Error = E> + 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<Error = Error> {
/// Releases the backend's resources; any later use fails.
async fn close(&self) -> Result<(), Error>;
}
pub type ClosableKvBackendRef = Arc<dyn ClosableKvBackend + Send + Sync>;
#[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";
src/datanode/src/datanode.rs
+144 -6
View File
@@ -102,6 +102,8 @@ pub struct Datanode {
leases_notifier: Option<Arc<Notify>>,
plugins: Plugins,
object_store_log_store: Option<Arc<ObjectStoreLogStore>>,
/// The raft-engine WAL, if any. Stopped after the region server, see [Datanode::shutdown].
raft_engine_log_store: Option<Arc<RaftEngineLogStore>>,
}
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<Box<dyn TopicStatsReporter>>,
open_regions_writable_override: Option<bool>,
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<Arc<RaftEngineLogStore>>,
#[cfg(feature = "enterprise")]
extension_range_provider_factory: Option<mito2::extension::BoxedExtensionRangeProviderFactory>,
/// Set as soon as the object store log store exists so that a failed build can stop it.
object_store_log_store: Option<Arc<ObjectStoreLogStore>>,
/// 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<RegionServer>,
}
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<Datanode> {
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<error::Error>, 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<RaftEngineLogStore>) -> 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(&region_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 {
src/log-store/src/error.rs
+9
View File
@@ -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 { .. }
src/log-store/src/raft_engine.rs
+38 -2
View File
@@ -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<Engine>) -> Result<()> {
common_runtime::spawn_blocking_global(move || drop(engine))
.await
.context(JoinSnafu)
}
impl EntryImpl {
pub fn create(id: u64, ns: u64, data: Vec<u8>) -> Self {
Self {
@@ -56,3 +72,23 @@ impl From<EntryImpl> 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;
src/log-store/src/raft_engine/backend.rs
+130 -30
View File
@@ -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<Arc<Engine>>,
_gc_task: RepeatedTask<Error>,
/// `None` once the backend is closed, see [ClosableKvBackend::close].
engine: RwLock<Option<Arc<Engine>>>,
gc_task: RepeatedTask<Error>,
}
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<Option<KeyValue>, 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<bool, Self::Error> {
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<Option<KeyValue>, 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<Arc<Engine>>) -> 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<Option<KeyValue>> {
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");
src/log-store/src/raft_engine/log_store.rs
+181 -20
View File
@@ -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<Duration>,
read_batch_size: usize,
engine: Arc<Engine>,
/// 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<Error>,
sync_task: RepeatedTask<Error>,
}
/// 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<Engine>,
pub engine: Weak<Engine>,
}
#[async_trait::async_trait]
@@ -59,7 +72,10 @@ impl TaskFunction<Error> 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<us
engine.write(batch, true)
}
/// Syncs the log files periodically. Holds the engine weakly, see [PurgeExpiredFilesFunction].
pub struct SyncWalTaskFunction {
engine: Arc<Engine>,
engine: Weak<Engine>,
/// 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<LogBatch>,
@@ -116,7 +133,9 @@ pub struct SyncWalTaskFunction {
#[async_trait::async_trait]
impl TaskFunction<Error> 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<Error> for SyncWalTaskFunction {
}
impl SyncWalTaskFunction {
pub fn new(engine: Arc<Engine>) -> Self {
pub fn new(engine: &Arc<Engine>) -> 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<Entry>,
) -> Result<(LogBatch, HashMap<RegionId, EntryId>)> {
// 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::<Vec<_>>());
}
// 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::<Vec<_>>().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");
src/mito2/src/engine.rs
+24 -11
View File
@@ -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<MitoConfig>,
/// The Wal raw entry reader.
wal_raw_entry_reader: Arc<dyn RawEntryReader>,
/// 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<Option<Arc<dyn RawEntryReader>>>,
/// 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<Arc<dyn RawEntryReader>> {
self.wal_raw_entry_reader
.read()
.unwrap()
.clone()
.context(EngineStoppedSnafu)
}
fn find_region(&self, region_id: RegionId) -> Result<MitoRegionRef> {
@@ -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()?,
&region_ids,
DEFAULT_ENTRY_RECEIVER_BUFFER_SIZE,
);
@@ -1036,7 +1049,7 @@ impl EngineInner {
.collect::<Vec<_>>();
let (distributor, entry_receivers) = build_wal_entry_distributor_and_receivers(
provider.clone(),
self.wal_raw_entry_reader.clone(),
self.wal_raw_entry_reader()?,
&region_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")]
src/mito2/src/error.rs
+7
View File
@@ -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 { .. }
src/standalone/src/lib.rs
+1 -1
View File
@@ -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};
src/standalone/src/metadata.rs
+11 -3
View File
@@ -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<KvBackendRef> {
/// Opens the metadata kvbackend, which the caller has to close on shutdown.
pub fn open_metadata_kvbackend(
dir: String,
config: KvBackendConfig,
) -> Result<ClosableKvBackendRef> {
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<KvBackendRef> {
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<KvBackendRef> {
debug!("Building metadata kvbackend from url: {}", url);