From f9ab3c296b9711c59bb6ffc9d3753a318d7883d7 Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Thu, 8 Oct 2026 06:33:40 +0000 Subject: [PATCH] feat(log-store): skip corrupted object store WAL segments per region (#9425) * feat(log-store): skip corrupted object store WAL segments per region A segment that does not decode, because its checksum does not match or its content disagrees with its footer entry, is fetched once more. If it still does not decode, the read follows `on_corrupted_segment`: `skip` records a WAL hole for the region with the object key, sequence and the footer's entry id range, increments `greptime_logstore_object_store_wal_skipped_segments_total`, logs a warning and continues with the next object; `fail` fails the read. A failed fetch fails the read in both modes. Holes are kept in memory per region and can be listed from the store. The standalone test builder allocates WAL options through `build_standalone_wal_provider`, and integration tests run a standalone instance on the object store WAL across restarts, on S3 when the `GT_S3_*` variables are set. The standalone shutdown runs every step and returns the first error. Signed-off-by: jeremyhi * test(cmd): cover the best-effort standalone stop Build a standalone instance whose leader services fail to stop and whose datanode servers fail afterwards, and check that `Instance::stop` still stops the object store WAL and returns the leader services error. Rename `skip_segment` to `record_skipped_segment` and the file store integration test to `test_standalone_object_store_wal_survives_restarts_on_file`. Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- src/cmd/Cargo.toml | 1 + src/cmd/src/standalone.rs | 140 ++++++- src/datanode/src/datanode.rs | 7 + src/log-store/src/metrics.rs | 8 + src/log-store/src/object_store_wal.rs | 6 +- src/log-store/src/object_store_wal/store.rs | 370 ++++++++++++++++-- .../engine/object_store_wal_recovery_test.rs | 111 +++++- tests-integration/src/standalone.rs | 29 +- tests-integration/src/tests/test_util.rs | 14 +- tests-integration/tests/main.rs | 1 + tests-integration/tests/object_store_wal.rs | 188 +++++++++ 11 files changed, 786 insertions(+), 89 deletions(-) create mode 100644 tests-integration/tests/object_store_wal.rs diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml index 2c925de57c8..0ef2f35b371 100644 --- a/src/cmd/Cargo.toml +++ b/src/cmd/Cargo.toml @@ -125,6 +125,7 @@ api.workspace = true client = { workspace = true, features = ["testing"] } common-test-util.workspace = true common-version.workspace = true +datanode = { workspace = true, features = ["testing"] } file-engine.workspace = true metric-engine.workspace = true mito2.workspace = true diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index bc6a62cd4d5..76c95c26d32 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -48,8 +48,8 @@ use common_meta::wal_provider::{WalProvider, WalProviderRef, build_wal_provider} use common_options::plugin_options::StandaloneFlag; use common_procedure::ProcedureManagerRef; use common_query::prelude::set_default_prefix; -use common_telemetry::info; use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions}; +use common_telemetry::{info, warn}; use common_time::timezone::set_default_timezone; use common_version::{short_version, verbose_version}; use common_wal::config::DatanodeWalConfig; @@ -274,32 +274,53 @@ impl App for Instance { Ok(()) } + /// Stops every component in order. A failed step does not skip the later + /// ones; the first error is returned and the rest are logged. async fn stop(&mut self) -> Result<()> { - self.frontend - .shutdown() - .await - .context(error::ShutdownFrontendSnafu)?; + let mut first_error = None; + let mut record = |result: Result<()>| { + if let Err(err) = result { + if first_error.is_none() { + first_error = Some(err); + } else { + warn!(err; "Ignored a later shutdown error"); + } + } + }; - self.leader_services_controller - .stop( - self.procedure_manager.clone(), - self.datanode.region_server(), - ) - .await?; + record( + self.frontend + .shutdown() + .await + .context(error::ShutdownFrontendSnafu), + ); - self.datanode - .shutdown() - .await - .context(error::ShutdownDatanodeSnafu)?; + record( + self.leader_services_controller + .stop( + self.procedure_manager.clone(), + self.datanode.region_server(), + ) + .await, + ); - self.flownode - .shutdown() - .await - .context(error::ShutdownFlownodeSnafu)?; + record( + self.datanode + .shutdown() + .await + .context(error::ShutdownDatanodeSnafu), + ); + + record( + self.flownode + .shutdown() + .await + .context(error::ShutdownFlownodeSnafu), + ); info!("Datanode instance stopped."); - Ok(()) + first_error.map_or(Ok(()), Err) } } @@ -1067,10 +1088,89 @@ mod tests { use frontend::frontend::FrontendOptions; use object_store::config::{FileConfig, GcsConfig}; use servers::grpc::GrpcOptions; + use servers::server::ServerHandlers; + use store_api::logstore::LogStore; use super::*; use crate::options::GlobalOptions; + struct FailingLeaderServices; + + #[async_trait] + impl StandaloneLeaderServicesController for FailingLeaderServices { + async fn start(&self, _context: LeaderServicesContext) -> Result<()> { + Ok(()) + } + + async fn stop( + &self, + _procedure_manager: ProcedureManagerRef, + _region_server: RegionServer, + ) -> Result<()> { + error::IllegalConfigSnafu { + msg: "leader services failed to stop", + } + .fail() + } + } + + struct FailingServer; + + #[async_trait] + impl servers::server::Server for FailingServer { + async fn shutdown(&self) -> servers::error::Result<()> { + servers::error::InternalSnafu { + err_msg: "server failed to stop", + } + .fail() + } + + async fn start(&mut self, _listening: SocketAddr) -> servers::error::Result<()> { + Ok(()) + } + + fn name(&self) -> &str { + "failing" + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + } + + #[tokio::test] + async fn test_stop_runs_every_step_and_returns_the_first_error() { + common_telemetry::init_default_ut_logging(); + let data_home = create_temp_dir("standalone-stop"); + let mut opts = StandaloneOptions { + wal: DatanodeWalConfig::ObjectStore(ObjectStoreWalConfig::default()), + ..Default::default() + }; + opts.storage.data_home = data_home.path().to_str().unwrap().to_string(); + let creator = InstanceCreator::default() + .with_leader_services_controller(Box::new(FailingLeaderServices)); + let (mut instance, _) = StartCommand::build_with(opts, vec![], creator) + .await + .unwrap(); + let log_store = instance.datanode.object_store_log_store().unwrap(); + // The datanode fails to stop its servers after the leader services failed. + let mut services = ServerHandlers::default(); + services.insert((Box::new(FailingServer), "127.0.0.1:0".parse().unwrap())); + services.start_all().await.unwrap(); + instance.datanode.setup_services(services); + + let err = instance.stop().await.unwrap_err(); + + assert!( + matches!(&err, error::Error::IllegalConfig { msg, .. } if msg == "leader services failed to stop"), + "unexpected error: {err:?}" + ); + assert!(matches!( + log_store.append_batch(vec![]).await, + Err(log_store::error::Error::ObjectStoreWalStopped { .. }) + )); + } + #[tokio::test] async fn test_build_standalone_wal_provider() { let kv_backend = Arc::new(MemoryKvBackend::new()) as KvBackendRef; diff --git a/src/datanode/src/datanode.rs b/src/datanode/src/datanode.rs index 8a80083aef3..fdac0597572 100644 --- a/src/datanode/src/datanode.rs +++ b/src/datanode/src/datanode.rs @@ -179,6 +179,13 @@ impl Datanode { self.region_server.clone() } + /// Returns the object store log store when the datanode runs on the object + /// store WAL. + #[cfg(any(test, feature = "testing"))] + pub fn object_store_log_store(&self) -> Option> { + self.object_store_log_store.clone() + } + pub fn plugins(&self) -> Plugins { self.plugins.clone() } diff --git a/src/log-store/src/metrics.rs b/src/log-store/src/metrics.rs index f12ad3e815a..824058d70af 100644 --- a/src/log-store/src/metrics.rs +++ b/src/log-store/src/metrics.rs @@ -89,4 +89,12 @@ lazy_static! { &[LOGSTORE_LABEL, PARTITION_LABEL], ) .unwrap(); + + /// Counter of segments a read of the object store logstore skipped because + /// they did not decode. + pub static ref METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL: IntCounter = register_int_counter!( + "greptime_logstore_object_store_wal_skipped_segments_total", + "object store logstore skipped corrupted segments total", + ) + .unwrap(); } diff --git a/src/log-store/src/object_store_wal.rs b/src/log-store/src/object_store_wal.rs index 2ec36b30463..d4bfc096393 100644 --- a/src/log-store/src/object_store_wal.rs +++ b/src/log-store/src/object_store_wal.rs @@ -50,7 +50,9 @@ //! Recovery lists the objects, reads and verifies only the header, trailer and //! footer of each, and indexes the footers of the chain in sequence order to //! rebuild the object catalog, which rejects a sequence it already holds. -//! Segments are read and checksummed only when a read decodes them. +//! Segments are read and checksummed only when a read decodes them; a segment +//! that does not decode is fetched once more and then skipped as a hole of its +//! region or fails the read, as `on_corrupted_segment` says. //! //! Entry ids are object-sequence-major, see [`entry_id`]: the high bits of an //! id name the object that holds the entry, the low bits its position among @@ -63,4 +65,4 @@ mod io; mod store; pub use batch::entry_id; -pub use store::ObjectStoreLogStore; +pub use store::{ObjectStoreLogStore, WalHole}; diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs index 5a6454123dd..8750a61cc4b 100644 --- a/src/log-store/src/object_store_wal/store.rs +++ b/src/log-store/src/object_store_wal/store.rs @@ -23,8 +23,8 @@ use std::time::{Duration, Instant}; use async_stream::try_stream; use bytes::Bytes; -use common_telemetry::info; -use common_wal::config::object_store::{AckMode, ObjectStoreWalConfig}; +use common_telemetry::{info, warn}; +use common_wal::config::object_store::{AckMode, CorruptedSegmentAction, ObjectStoreWalConfig}; use futures::future::BoxFuture; use futures::stream::FuturesUnordered; use futures::{StreamExt, TryStreamExt}; @@ -46,6 +46,7 @@ use crate::error::{ StaleWalObjectSnafu, UnconfirmedWalEpochStartSnafu, WalObjectSequenceExhaustedSnafu, WalObjectSequenceUnsettledSnafu, }; +use crate::metrics::METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL; use crate::object_store_wal::batch::{OBJECT_SEQ_LIMIT, OpenBatch, entry_id, sequence_floor}; use crate::object_store_wal::catalog::ObjectCatalog; use crate::object_store_wal::format::{ @@ -87,13 +88,17 @@ const RECOVERY_TAIL_WINDOW: usize = 64 * 1024; /// indexed, so an acknowledged entry never has a missing predecessor; in the /// `enqueued` mode it returns on admission and the object is created in the /// background. While [`MAX_SEALED_BATCHES`] batches -/// wait to become durable no append is admitted. +/// wait to become durable no append is admitted. A segment that a read cannot +/// decode is handled as `on_corrupted_segment` says. pub struct ObjectStoreLogStore { prefix: String, ack_mode: AckMode, + on_corrupted_segment: CorruptedSegmentAction, io: Arc, catalog: Arc>, obsolete_entry_ids: ObsoleteEntryIds, + /// Segments that reads skipped, per region and by object sequence. + wal_holes: WalHoles, terminal_error: TerminalError, stopped: Arc, command_tx: mpsc::Sender, @@ -114,6 +119,20 @@ pub struct ObjectStoreLogStore { type ObsoleteEntryIds = Arc>>; +type WalHoles = Arc>>>; + +/// A segment a read skipped because it did not decode: the entries of one +/// region in one object are missing from every read of that region. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WalHole { + /// Key of the object that holds the segment. + pub path: String, + pub object_seq: u64, + /// Entry id range the footer records for the segment. + pub min_entry_id: EntryId, + pub max_entry_id: EntryId, +} + type TerminalError = Arc>>>; impl fmt::Debug for ObjectStoreLogStore { @@ -121,6 +140,7 @@ impl fmt::Debug for ObjectStoreLogStore { f.debug_struct("ObjectStoreLogStore") .field("prefix", &self.prefix) .field("ack_mode", &self.ack_mode) + .field("on_corrupted_segment", &self.on_corrupted_segment) .finish_non_exhaustive() } } @@ -260,9 +280,11 @@ impl ObjectStoreLogStore { Ok(Arc::new(Self { prefix, ack_mode: config.ack_mode, + on_corrupted_segment: config.on_corrupted_segment, io, catalog, obsolete_entry_ids, + wal_holes: WalHoles::default(), terminal_error, stopped, command_tx, @@ -297,6 +319,21 @@ impl ObjectStoreLogStore { catalog.region_max_entry_id(region_id).unwrap_or(0) } + /// Returns the segments of the provider's region that reads of this store + /// skipped because they did not decode, ordered by object sequence. Holes + /// are kept in memory only: a store opened later on the same prefix + /// records them again when its reads meet the segments. + pub fn wal_holes(&self, provider: &Provider) -> Result> { + let region_id = self.region_of(provider)?; + Ok(self + .wal_holes + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(®ion_id) + .map(|holes| holes.values().cloned().collect()) + .unwrap_or_default()) + } + /// Returns the region of `provider`, which must select this store's prefix. fn region_of(&self, provider: &Provider) -> Result { let provider = @@ -477,6 +514,30 @@ impl ObjectStoreLogStore { pub fn begin_stop(&self) { self.stopped.store(true, Ordering::Release); } + + /// Returns the key of the object `object_seq` and the byte range the + /// segment of the provider's region occupies in it, so a test can damage + /// that segment alone, or `None` when the object holds no segment of the + /// region. + pub fn segment_location( + &self, + provider: &Provider, + object_seq: u64, + ) -> Result)>> { + let region_id = self.region_of(provider)?; + let catalog = self.catalog.read().unwrap_or_else(PoisonError::into_inner); + Ok(catalog + .objects_for_entry_range(region_id, 0, EntryId::MAX)? + .into_iter() + .find(|(seq, _)| *seq == object_seq) + .map(|(_, entry)| { + let start = entry.segment_offset; + ( + self.io.object_path(object_seq), + start..start + entry.segment_len, + ) + })) + } } #[async_trait::async_trait] @@ -534,6 +595,9 @@ impl LogStore for ObjectStoreLogStore { /// Reads the provider's region from `entry_id`, hiding obsolete entries. /// The catalog locates each segment, so a caller-supplied index is unnecessary. + /// A segment that does not decode fails the read or is skipped and recorded + /// as a hole, as `on_corrupted_segment` says; a failed fetch always fails + /// the read. async fn read( &self, provider: &Provider, @@ -568,15 +632,20 @@ impl LogStore for ObjectStoreLogStore { let io = self.io.clone(); let provider = provider.clone(); + let on_corrupted_segment = self.on_corrupted_segment; + let wal_holes = self.wal_holes.clone(); Ok(Box::pin(try_stream! { for (object_seq, footer_entry) in objects { - let bytes = io - .get_range(object_seq, footer_entry.segment_offset, footer_entry.segment_len) - .await?; - let records = decode_segment(&bytes, &footer_entry) - .with_context(|_| InvalidWalObjectSnafu { - path: io.object_path(object_seq), - })?; + let records = match fetch_segment(io.as_ref(), object_seq, &footer_entry).await { + Ok(records) => records, + Err(error @ Error::InvalidWalObject { .. }) + if on_corrupted_segment == CorruptedSegmentAction::Skip => + { + record_skipped_segment(&wal_holes, io.as_ref(), object_seq, &footer_entry, &error); + continue; + } + Err(error) => Err(error)?, + }; let entries = records .into_iter() .filter(|record| record.entry_id >= start_entry_id) @@ -1646,6 +1715,59 @@ fn shared(error: &Arc) -> Error { ObjectStoreWalSnafu.into_error(error.clone()) } +/// Fetches and decodes the segment `entry` describes in the object +/// `object_seq`. A segment that does not decode is fetched once more, since +/// the download rather than the object may be damaged; when it still does not +/// decode the error is returned as [`Error::InvalidWalObject`]. A failed fetch +/// is returned as is. +async fn fetch_segment( + io: &dyn WalObjectIo, + object_seq: u64, + entry: &FooterEntry, +) -> Result> { + let fetch = || io.get_range(object_seq, entry.segment_offset, entry.segment_len); + if let Ok(records) = decode_segment(&fetch().await?, entry) { + return Ok(records); + } + decode_segment(&fetch().await?, entry).with_context(|_| InvalidWalObjectSnafu { + path: io.object_path(object_seq), + }) +} + +/// Records the segment `entry` describes in the object `object_seq` as a hole +/// of its region, once per object, counts the skip and warns about it. +fn record_skipped_segment( + wal_holes: &WalHoles, + io: &dyn WalObjectIo, + object_seq: u64, + entry: &FooterEntry, + error: &Error, +) { + let hole = WalHole { + path: io.object_path(object_seq), + object_seq, + min_entry_id: entry.min_entry_id, + max_entry_id: entry.max_entry_id, + }; + warn!( + error; + "Skipped a corrupted WAL segment of region {}, object {} (sequence {}), entry ids {}..={}", + entry.region_id, + hole.path, + object_seq, + hole.min_entry_id, + hole.max_entry_id + ); + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.inc(); + wal_holes + .lock() + .unwrap_or_else(PoisonError::into_inner) + .entry(entry.region_id) + .or_default() + .entry(object_seq) + .or_insert(hole); +} + fn encode_batch(header: Header, entries: Vec) -> Result { let records = entries .into_iter() @@ -2071,6 +2193,10 @@ mod tests { const PREFIX: &str = "wal/datanodes/1/epochs/2"; const WAIT: Duration = Duration::from_secs(30); + /// Held by every test that reads the process-wide skipped segments + /// counter, so that tests running in one process do not interleave + /// their increments. + static SKIPPED_SEGMENTS: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); fn memory_store() -> ObjectStore { ObjectStore::new(Memory::default()).unwrap() @@ -2093,6 +2219,16 @@ mod tests { } } + fn on_corruption( + on_corrupted_segment: CorruptedSegmentAction, + config: ObjectStoreWalConfig, + ) -> ObjectStoreWalConfig { + ObjectStoreWalConfig { + on_corrupted_segment, + ..config + } + } + /// Every append reaches the size limit, so it is persisted on its own. fn eager() -> ObjectStoreWalConfig { config(Duration::from_secs(3600), 1) @@ -2472,6 +2608,18 @@ mod tests { (trailer, footer) } + /// Asserts that recovery over `object_store` rejects the object at `path` + /// for `reason`, whatever a read does with a corrupted segment. + async fn assert_recovery_rejects(object_store: &ObjectStore, path: &str, reason: &str) { + for on_corrupted_segment in [CorruptedSegmentAction::Skip, CorruptedSegmentAction::Fail] { + let config = on_corruption(on_corrupted_segment, eager()); + let error = ObjectStoreLogStore::try_new(object_store.clone(), &config, 1, 2) + .await + .unwrap_err(); + assert_invalid_object(&error, path, reason); + } + } + fn assert_invalid_object(error: &Error, path: &str, reason: &str) { match error { Error::InvalidWalObject { @@ -2553,10 +2701,7 @@ mod tests { let path = object_path(&object_store, 1); corrupt_object(&object_store, &path, *corrupt).await; - let error = ObjectStoreLogStore::try_new(object_store, &eager(), 1, 2) - .await - .unwrap_err(); - assert_invalid_object(&error, &path, reason); + assert_recovery_rejects(&object_store, &path, reason).await; } } @@ -2575,14 +2720,12 @@ mod tests { .unwrap(); io.put_if_absent(5, encoded.bytes).await.unwrap(); - let error = ObjectStoreLogStore::try_new(object_store, &eager(), 1, 2) - .await - .unwrap_err(); - assert_invalid_object( - &error, + assert_recovery_rejects( + &object_store, &io.object_path(5), "header sequence 0 does not match key sequence 5", - ); + ) + .await; } /// Overwrites the byte range of footer entry `index` and refreshes the @@ -2635,10 +2778,7 @@ mod tests { }) .await; - let error = ObjectStoreLogStore::try_new(object_store.clone(), &eager(), 1, 2) - .await - .unwrap_err(); - assert_invalid_object(&error, &path, reason); + assert_recovery_rejects(&object_store, &path, reason).await; let io = ObjectStoreIo::new(object_store, PREFIX).unwrap(); let error = recover_by_decoding(&io).await.unwrap_err(); assert_invalid_object(&error, &path, reason); @@ -2872,9 +3012,11 @@ mod tests { let store = ObjectStoreLogStore { prefix: PREFIX.to_string(), ack_mode: AckMode::Durable, + on_corrupted_segment: CorruptedSegmentAction::Skip, io: Arc::new(ObjectStoreIo::new(memory_store(), PREFIX).unwrap()), catalog: Arc::default(), obsolete_entry_ids: ObsoleteEntryIds::default(), + wal_holes: WalHoles::default(), terminal_error: Arc::default(), stopped: Arc::new(AtomicBool::new(false)), command_tx, @@ -3267,11 +3409,12 @@ mod tests { store.stop().await.unwrap(); } - #[tokio::test] - async fn test_store_corrupted_segment_fails_the_read_that_decodes_it() { - let object_store = memory_store(); + /// Writes object 0 with one entry of region 1 and one of region 2 and + /// object 1 with an entry of region 2, then flips a byte in the segment of + /// region 2 in object 0. Returns the key of object 0 and its footer. + async fn write_corrupted_segment(object_store: &ObjectStore) -> (String, Vec) { put_records( - &object_store, + object_store, 0, &[ Record { @@ -3287,19 +3430,93 @@ mod tests { ], ) .await; - put_object(&object_store, 1, region(2), &[10]).await; - let path = object_path(&object_store, 0); + put_object(object_store, 1, region(2), &[10]).await; + let path = object_path(object_store, 0); let bytes = object_store.read(&path).await.unwrap().to_vec(); let footer = footer_of(&bytes).1; let corrupt = &footer[1]; - corrupt_object(&object_store, &path, |bytes| { + assert_eq!(region(2), corrupt.region_id); + corrupt_object(object_store, &path, |bytes| { bytes[(corrupt.segment_offset + corrupt.segment_len - 1) as usize] ^= 1; }) .await; + (path, footer) + } + + #[tokio::test] + async fn test_store_skips_a_corrupted_segment_and_records_a_hole() { + let _serialized = SKIPPED_SEGMENTS.lock().await; + let object_store = memory_store(); + let (path, footer) = write_corrupted_segment(&object_store).await; let (io, reads) = RecordingIo::over(object_store); - let store = ObjectStoreLogStore::open(io, &eager(), PREFIX.to_string()) - .await - .unwrap(); + let store = open_over(io, &eager()).await; + assert_eq!(CorruptedSegmentAction::Skip, store.on_corrupted_segment); + let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(); + + // Recovery indexes the object; a read of the other region and one of + // the later object of the damaged region are unaffected. + assert_eq!( + expected_entries(region(1), &[(1, "a1")]), + read_entries(&store, region(1), 0).await + ); + assert_eq!( + expected_entries(region(2), &[(10, "e10")]), + read_entries(&store, region(2), 10).await + ); + assert!(store.wal_holes(&provider(region(2))).unwrap().is_empty()); + assert_eq!( + skipped, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + + // A read across the damaged segment fetches it twice, continues with + // the next object and records the hole. + reads.lock().unwrap().clear(); + assert_eq!( + expected_entries(region(2), &[(10, "e10")]), + read_entries(&store, region(2), 0).await + ); + let corrupt = (0, footer[1].segment_offset, footer[1].segment_len); + assert_eq!(corrupt, reads.lock().unwrap()[0]); + assert_eq!(corrupt, reads.lock().unwrap()[1]); + let hole = WalHole { + path, + object_seq: 0, + min_entry_id: 1, + max_entry_id: 1, + }; + assert_eq!( + vec![hole.clone()], + store.wal_holes(&provider(region(2))).unwrap() + ); + assert!(store.wal_holes(&provider(region(1))).unwrap().is_empty()); + assert_eq!( + skipped + 1, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + + // Every read across the segment skips it again; the hole is one. + assert_eq!( + expected_entries(region(2), &[(10, "e10")]), + read_entries(&store, region(2), 0).await + ); + assert_eq!(vec![hole], store.wal_holes(&provider(region(2))).unwrap()); + assert_eq!( + skipped + 2, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_corrupted_segment_fails_the_read_that_decodes_it() { + let _serialized = SKIPPED_SEGMENTS.lock().await; + let object_store = memory_store(); + let (path, footer) = write_corrupted_segment(&object_store).await; + let (io, reads) = RecordingIo::over(object_store); + let config = on_corruption(CorruptedSegmentAction::Fail, eager()); + let store = open_over(io, &config).await; + let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(); assert_eq!(1, latest(&store, region(1))); assert_eq!(10, latest(&store, region(2))); reads.lock().unwrap().clear(); @@ -3328,13 +3545,78 @@ mod tests { &path, &format!("segment of region {} checksum mismatch", region(2)), ); + let corrupt = (0, footer[1].segment_offset, footer[1].segment_len); + assert_eq!(vec![corrupt, corrupt], *reads.lock().unwrap()); + assert!(store.wal_holes(&provider(region(2))).unwrap().is_empty()); assert_eq!( - vec![(0, corrupt.segment_offset, corrupt.segment_len)], - *reads.lock().unwrap() + skipped, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() ); store.stop().await.unwrap(); } + #[tokio::test] + async fn test_store_reads_a_segment_whose_first_download_was_damaged() { + let _serialized = SKIPPED_SEGMENTS.lock().await; + for on_corrupted_segment in [CorruptedSegmentAction::Skip, CorruptedSegmentAction::Fail] { + let object_store = memory_store(); + put_object(&object_store, 0, region(1), &[1, 2]).await; + let (io, reads) = RecordingIo::over(object_store); + let config = on_corruption(on_corrupted_segment, eager()); + let store = open_over(io.clone(), &config).await; + let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(); + + // The first download of the segment is damaged, the second is + // not: the read sees the entries and records no hole. + reads.lock().unwrap().clear(); + io.damage_next_range_read.store(true, Ordering::Relaxed); + assert_eq!( + expected_entries(region(1), &[(1, "e1"), (2, "e2")]), + read_entries(&store, region(1), 0).await + ); + assert_eq!(2, reads.lock().unwrap().len()); + assert!(store.wal_holes(&provider(region(1))).unwrap().is_empty()); + assert_eq!( + skipped, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + store.stop().await.unwrap(); + } + } + + #[tokio::test] + async fn test_store_failed_segment_fetch_fails_the_read() { + let _serialized = SKIPPED_SEGMENTS.lock().await; + for on_corrupted_segment in [CorruptedSegmentAction::Skip, CorruptedSegmentAction::Fail] { + let object_store = memory_store(); + put_object(&object_store, 0, region(1), &[1]).await; + let path = object_path(&object_store, 0); + let (io, _) = RecordingIo::over(object_store.clone()); + let config = on_corruption(on_corrupted_segment, eager()); + let store = open_over(io, &config).await; + let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(); + + object_store.delete(&path).await.unwrap(); + let error = store + .read(&provider(region(1)), 0, None) + .await + .unwrap() + .try_collect::>() + .await + .unwrap_err(); + assert!( + matches!(&error, Error::WalObjectStore { path: actual, .. } if *actual == path), + "unexpected error: {error:?}" + ); + assert!(store.wal_holes(&provider(region(1))).unwrap().is_empty()); + assert_eq!( + skipped, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + store.stop().await.unwrap(); + } + } + #[tokio::test] async fn test_store_assigns_ids_from_the_object_sequence() { let store = open(memory_store(), &manual()).await; @@ -5540,13 +5822,14 @@ mod tests { /// Object access that records every range read as (sequence, offset, /// length) and, on request, reports the next conditional create as failed - /// after it wrote the object, or fails it with a given error before it - /// writes. + /// after it wrote the object, fails it with a given error before it + /// writes, or damages the bytes the next range read returns. struct RecordingIo { inner: ObjectStoreIo, reads: RangeReads, fail_after_next_put: AtomicBool, fail_next_put: Mutex>, + damage_next_range_read: AtomicBool, } impl RecordingIo { @@ -5557,6 +5840,7 @@ mod tests { reads: reads.clone(), fail_after_next_put: AtomicBool::new(false), fail_next_put: Mutex::new(None), + damage_next_range_read: AtomicBool::new(false), }; (Arc::new(io), reads) } @@ -5601,7 +5885,15 @@ mod tests { async fn get_range(&self, object_seq: u64, offset: u64, len: u64) -> Result { self.reads.lock().unwrap().push((object_seq, offset, len)); - self.inner.get_range(object_seq, offset, len).await + let bytes = self.inner.get_range(object_seq, offset, len).await?; + if self.damage_next_range_read.swap(false, Ordering::Relaxed) { + let mut damaged = bytes.to_vec(); + if let Some(last) = damaged.last_mut() { + *last ^= 1; + } + return Ok(Bytes::from(damaged)); + } + Ok(bytes) } async fn list(&self) -> Result> { diff --git a/src/mito2/src/engine/object_store_wal_recovery_test.rs b/src/mito2/src/engine/object_store_wal_recovery_test.rs index 6017f745d2e..917ad614a07 100644 --- a/src/mito2/src/engine/object_store_wal_recovery_test.rs +++ b/src/mito2/src/engine/object_store_wal_recovery_test.rs @@ -30,10 +30,12 @@ use common_base::readable_size::ReadableSize; use common_error::ext::{BoxedError, ErrorExt}; use common_error::status_code::StatusCode; use common_recordbatch::RecordBatches; -use common_wal::config::object_store::{AckMode, ObjectStoreWalConfig, STANDALONE_GENERATION}; +use common_wal::config::object_store::{ + AckMode, CorruptedSegmentAction, ObjectStoreWalConfig, STANDALONE_GENERATION, +}; use common_wal::options::{ObjectStoreWalOptions, WAL_OPTIONS_KEY, WalOptions}; use log_store::ObjectStoreLogStore; -use log_store::object_store_wal::entry_id; +use log_store::object_store_wal::{WalHole, entry_id}; use object_store::ObjectStore; use object_store::services::Memory; use rstest::rstest; @@ -412,6 +414,111 @@ async fn test_reopen_after_crash_replays_durable_entries_once(#[case] ack_mode: assert_eq!(3, engine.get_region_statistic(REGION_A).unwrap().num_rows); } +#[rstest] +#[case(CorruptedSegmentAction::Skip)] +#[case(CorruptedSegmentAction::Fail)] +#[tokio::test] +async fn test_reopen_with_a_corrupted_segment( + #[case] on_corrupted_segment: CorruptedSegmentAction, +) { + let mut env = TestEnv::with_prefix("object-store-wal-corrupted-segment").await; + let object_store = memory_store(); + let store = open_store(&object_store, AckMode::Durable).await; + let engine = new_engine(&mut env, store.clone()).await; + let (table_dir_a, schema_a) = create_region(&engine, REGION_A, &[]).await; + let (table_dir_b, schema_b) = create_region(&engine, REGION_B, &[]).await; + + // Objects 1 and 2 each hold one entry of both regions; nothing is flushed. + let mut writer = SealedWriter::new(&store); + writer + .put_and_seal( + &engine, + vec![ + (REGION_A, rows(&schema_a, 0, 2)), + (REGION_B, rows(&schema_b, 0, 3)), + ], + ) + .await; + writer + .put_and_seal( + &engine, + vec![ + (REGION_A, rows(&schema_a, 2, 4)), + (REGION_B, rows(&schema_b, 3, 5)), + ], + ) + .await; + let rows_b = scan_rows(&engine, REGION_B).await; + engine.stop().await.unwrap(); + + // The segment of region A in object 1 is damaged. + let (path, segment) = store + .segment_location(&provider(REGION_A), 1) + .unwrap() + .unwrap(); + let mut bytes = object_store.read(&path).await.unwrap().to_vec(); + bytes[segment.end as usize - 1] ^= 1; + object_store.write(&path, bytes).await.unwrap(); + drop(engine); + drop(writer); + drop(store); + + let config = ObjectStoreWalConfig { + on_corrupted_segment, + ..store_config(AckMode::Durable) + }; + let store = + ObjectStoreLogStore::try_new(object_store.clone(), &config, 0, STANDALONE_GENERATION) + .await + .unwrap(); + let engine = new_engine(&mut env, store.clone()).await; + + // Region B replays both of its entries whatever the damage does to A. + open_region(&engine, REGION_B, &table_dir_b, &[]) + .await + .unwrap(); + assert_eq!(rows_b, scan_rows(&engine, REGION_B).await); + assert_eq!(5, engine.get_region_statistic(REGION_B).unwrap().num_rows); + assert!(store.wal_holes(&provider(REGION_B)).unwrap().is_empty()); + + let opened = open_region(&engine, REGION_A, &table_dir_a, &[]).await; + match on_corrupted_segment { + // Region A replays the entry of object 2 alone and records the + // segment of object 1 as a hole. + CorruptedSegmentAction::Skip => { + opened.unwrap(); + assert_eq!( + RegionRecoveryState { + last_entry_id: entry_id(2, 1), + memtable_rows: 2, + ..EMPTY_RECOVERY_STATE + }, + region_recovery_state(&engine, REGION_A).await + ); + assert_eq!(2, engine.get_region_statistic(REGION_A).unwrap().num_rows); + assert_eq!( + vec![WalHole { + path, + object_seq: 1, + min_entry_id: entry_id(1, 1), + max_entry_id: entry_id(1, 1), + }], + store.wal_holes(&provider(REGION_A)).unwrap() + ); + } + // The read fails, so the region does not open. + CorruptedSegmentAction::Fail => { + let message = opened.unwrap_err().output_msg(); + assert!( + message.contains("checksum mismatch"), + "unexpected error: {message}" + ); + assert!(!engine.is_region_exists(REGION_A)); + assert!(store.wal_holes(&provider(REGION_A)).unwrap().is_empty()); + } + } +} + #[tokio::test] async fn test_open_region_rejects_mismatched_wal_prefix() { let mut env = TestEnv::with_prefix("object-store-wal-prefix-mismatch").await; diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 3fa1d394489..5a36d495df8 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -39,14 +39,13 @@ use common_meta::procedure_executor::LocalProcedureExecutor; use common_meta::region_keeper::MemoryRegionKeeper; use common_meta::region_registry::LeaderRegionRegistry; use common_meta::sequence::SequenceBuilder; -use common_meta::wal_provider::build_wal_provider; use common_procedure::ProcedureManagerRef; use common_procedure::local::EventRecorderHandle; use common_procedure::options::ProcedureConfig; use common_telemetry::logging::SlowQueryOptions; use common_test_util::find_workspace_path; -use common_wal::config::{DatanodeWalConfig, MetasrvWalConfig}; -use datanode::datanode::DatanodeBuilder; +use common_wal::config::DatanodeWalConfig; +use datanode::datanode::{Datanode, DatanodeBuilder}; use flow::{FlownodeBuilder, FrontendClient, GrpcQueryHandlerWithBoxedError}; use frontend::frontend::Frontend; use frontend::instance::Instance; @@ -63,6 +62,8 @@ use crate::test_util::{self, StorageType, TestGuard, create_tmp_dir_and_datanode pub struct GreptimeDbStandalone { /// Storage engine for assertions across protocol and storage boundaries. pub mito_engine: mito2::engine::MitoEngine, + /// Owns the region server and the log store, so a test can shut them down. + pub datanode: Datanode, pub frontend: Arc, pub opts: StandaloneOptions, pub guard: TestGuard, @@ -81,7 +82,6 @@ impl GreptimeDbStandalone { pub struct GreptimeDbStandaloneBuilder { instance_name: String, datanode_wal_config: DatanodeWalConfig, - metasrv_wal_config: MetasrvWalConfig, store_providers: Option>, default_store: Option, plugin: Option, @@ -102,7 +102,6 @@ impl GreptimeDbStandaloneBuilder { plugin: None, default_store: None, datanode_wal_config: DatanodeWalConfig::default(), - metasrv_wal_config: MetasrvWalConfig::default(), // Enable slow query log with 1s threshold by default for integration tests. slow_query_options: SlowQueryOptions { enable: true, @@ -199,12 +198,6 @@ impl GreptimeDbStandaloneBuilder { self } - #[must_use] - pub fn with_metasrv_wal_config(mut self, metasrv_wal_config: MetasrvWalConfig) -> Self { - self.metasrv_wal_config = metasrv_wal_config; - self - } - pub async fn build_with( &self, kv_backend: KvBackendRef, @@ -281,10 +274,13 @@ impl GreptimeDbStandaloneBuilder { .step(10) .build(), ); - let kafka_options = opts.wal.clone().try_into().unwrap(); - let wal_provider = build_wal_provider(&kafka_options, kv_backend.clone()) - .await - .unwrap(); + let wal_provider = cmd::standalone::build_standalone_wal_provider( + &opts.wal, + opts.datanode_options().node_id, + kv_backend.clone(), + ) + .await + .unwrap(); let wal_provider = Arc::new(wal_provider); let table_metadata_allocator = Arc::new(TableMetadataAllocator::new( table_id_allocator, @@ -363,6 +359,7 @@ impl GreptimeDbStandaloneBuilder { GreptimeDbStandalone { mito_engine: datanode.region_server().mito_engine().unwrap(), + datanode, frontend: Arc::new(frontend), opts, guard, @@ -399,7 +396,7 @@ impl GreptimeDbStandaloneBuilder { storage: opts.storage, procedure: procedure_config, metadata_store: kv_backend_config, - wal: self.metasrv_wal_config.clone().into(), + wal: opts.wal, grpc: GrpcOptions::default().with_server_addr("127.0.0.1:4001"), slow_query: self.slow_query_options.clone(), event_recorder: self.event_recorder_options.clone(), diff --git a/tests-integration/src/tests/test_util.rs b/tests-integration/src/tests/test_util.rs index 5ddcd0748a9..7c99e89ee08 100644 --- a/tests-integration/src/tests/test_util.rs +++ b/tests-integration/src/tests/test_util.rs @@ -263,15 +263,8 @@ pub(crate) async fn standalone_with_kafka_wal() -> Option>(); let test_name = uuid::Uuid::new_v4().to_string(); - let builder = GreptimeDbStandaloneBuilder::new(&test_name) - .with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig { - connection: KafkaConnectionConfig { - broker_endpoints: endpoints.clone(), - ..Default::default() - }, - ..Default::default() - })) - .with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig { + let builder = GreptimeDbStandaloneBuilder::new(&test_name).with_datanode_wal_config( + DatanodeWalConfig::Kafka(DatanodeKafkaConfig { connection: KafkaConnectionConfig { broker_endpoints: endpoints, ..Default::default() @@ -282,7 +275,8 @@ pub(crate) async fn standalone_with_kafka_wal() -> Option Output { + SqlQueryHandler::do_query(instance, sql, QueryContext::arc()) + .await + .remove(0) + .unwrap() +} + +async fn query_rows(standalone: &GreptimeDbStandalone) -> String { + execute_sql(standalone.fe_instance(), QUERY) + .await + .data + .pretty_print() + .await +} + +/// Returns the number of WAL objects under [`WAL_ROOT`] of the default store. +async fn count_wal_objects(standalone: &GreptimeDbStandalone) -> usize { + let store = match &standalone.opts.storage.store { + ObjectStoreConfig::File(_) => { + ObjectStore::new(Fs::default().root(&standalone.opts.storage.data_home)).unwrap() + } + ObjectStoreConfig::S3(s3) => ObjectStore::new(S3::from(&s3.connection)).unwrap(), + other => panic!("unexpected store {other:?}"), + }; + store + .list_with(&format!("{WAL_ROOT}/")) + .recursive(true) + .await + .unwrap() + .into_iter() + .filter(|entry| entry.metadata().is_file()) + .count() +} + +/// Stops the procedure manager, the region server and the WAL of the +/// instance. +async fn shutdown(standalone: &mut GreptimeDbStandalone) { + standalone.procedure_manager.stop().await.unwrap(); + standalone.datanode.shutdown().await.unwrap(); +} + +/// Shuts the instance down and builds it again on its metadata, data home and +/// object store, like a process restart. +async fn restart( + builder: &GreptimeDbStandaloneBuilder, + mut standalone: GreptimeDbStandalone, +) -> GreptimeDbStandalone { + shutdown(&mut standalone).await; + let GreptimeDbStandalone { + opts, + guard, + kv_backend, + .. + } = standalone; + + let (procedure_manager, event_recorder_handle) = + standalone::build_procedure_manager(kv_backend.clone(), ProcedureConfig::default()); + builder + .build_with( + kv_backend, + guard, + opts, + procedure_manager, + event_recorder_handle, + true, + ) + .await +} + +/// Writes batches that land in separate WAL objects, restarts before and +/// after a flush, and checks that the rows survive both restarts. +async fn run_restarts_on_object_store_wal(store_type: StorageType) { + common_telemetry::init_default_ut_logging(); + let wal_config = ObjectStoreWalConfig { + prefix: WAL_ROOT.to_string(), + ..Default::default() + }; + let builder = GreptimeDbStandaloneBuilder::new("object_store_wal") + .with_default_store_type(store_type) + .with_datanode_wal_config(DatanodeWalConfig::ObjectStore(wal_config)); + let standalone = builder.build().await; + + execute_sql( + standalone.fe_instance(), + r#" + CREATE TABLE cpu ( + hostname STRING PRIMARY KEY, + usage_user DOUBLE, + ts TIMESTAMP TIME INDEX + ) + "#, + ) + .await; + // Every insert returns once its entries are durable, so the batches land + // in separate WAL objects after the object that started the epoch. + for batch in 0..BATCHES { + let values = (0..ROWS_PER_BATCH) + .map(|row| { + let i = batch * ROWS_PER_BATCH + row; + format!( + "('host_{i}', {i}.0, {})", + 1_686_567_600_000 + i as i64 * 1000 + ) + }) + .collect::>() + .join(", "); + execute_sql( + standalone.fe_instance(), + &format!("INSERT INTO cpu VALUES {values}"), + ) + .await; + } + let expected = query_rows(&standalone).await; + assert_eq!(BATCHES * ROWS_PER_BATCH + 4, expected.lines().count()); + let objects = count_wal_objects(&standalone).await; + assert!( + objects > BATCHES, + "expected more than {BATCHES} WAL objects" + ); + // No Raft Engine log store is created as a fallback. + assert!( + !Path::new(&standalone.opts.storage.data_home) + .join("wal") + .exists() + ); + + // Nothing was flushed, so the rows come back from the WAL alone; the + // reopened store starts its epoch with another object. + let standalone = restart(&builder, standalone).await; + assert_eq!(expected, query_rows(&standalone).await); + assert!(count_wal_objects(&standalone).await > objects); + + // The flushed rows come back from the SST files. + execute_sql(standalone.fe_instance(), "ADMIN FLUSH_TABLE('cpu')").await; + let mut standalone = restart(&builder, standalone).await; + assert_eq!(expected, query_rows(&standalone).await); + shutdown(&mut standalone).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_standalone_object_store_wal_survives_restarts_on_file() { + run_restarts_on_object_store_wal(StorageType::File).await; +} + +/// Runs against the S3 bucket of the `GT_S3_*` environment variables, so it +/// is skipped unless they are set. +#[tokio::test(flavor = "multi_thread")] +async fn test_standalone_object_store_wal_survives_restarts_on_s3() { + if !StorageType::S3.test_on() { + return; + } + run_restarts_on_object_store_wal(StorageType::S3).await; +}