fix(log-store): refuse to open object store WAL regions whose history is missing (#9561)

* fix(log-store): refuse to open object store WAL regions whose history is missing

A region persists only its WAL prefix, not the object store that holds it.
After `[wal].storage_provider`, or the bucket or root of the provider it
names, changes, or after the objects under the prefix are deleted, every
region reopened against a prefix without its history, replayed nothing and
silently lost its acknowledged writes that were not flushed.

Add `LogStore::check_region_history` and `LogStore::start_region_history`,
whose defaults accept and do nothing, so Raft Engine and Kafka are
unchanged. The object store WAL reports the history of a region as missing
when recovery found no object under the prefix, or when the region's
watermark names an object above the newest object holding entries that
recovery indexed, unless the catalog holds the region's entries through the
watermark or the region's history starts in the store. Both rest on
invariants of the prefix: an opened prefix is never empty, and a persisted
watermark names an object at or below the newest object holding entries,
which collection never deletes. The segment of release records counts as
entries, since an object holding only release records can be the one
collection keeps; start objects hold no segment and do not move the bound.

When a store that found its prefix empty refuses a region, the actor gives
up its start object, so the next start refuses the region again. It agrees
only while nothing was handed an id, no batch, append, release or create
waits, no failed release is kept, no other object was listed, the catalog
holds only the start object and no region's history started in the store,
deciding under the lock region registration takes. It then records a
terminal error, deletes the object as one of its tracked deletes, which
`stop` waits for, and answers every refusal once that delete settled.

Mito runs the check before it replays an existing region on open, on
catch-up and when a create finds the region present, with the larger of
the flushed and truncated entry ids, and records the history of a region
it creates; regions that skip the WAL and noop regions are not checked. A
missing history fails the open with an error naming the region, the
prefix, the configured storage and how to move the WAL by copying its
objects. The new `[wal].accept_missing_history` option, off by default,
opens such a region without its unflushed writes instead, with a warning
and the counter
`greptime_logstore_object_store_wal_missing_history_accepted_total`.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* docs(rfc): describe the object store WAL history check

Describe the open-time history check, the prefix invariants it relies on,
how release records bound the history, the start object withdrawal of a
refused open and the window that remains, the accept_missing_history
option and how to move a WAL by copying its objects, and update the
metrics, Mito integration, evidence, compatibility and cluster alignment
sections.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* fix(log-store): unindex the start object a refused open withdrew

A withdrawal that deleted the start object left it in the catalog and in the indexed object and byte gauges for the rest of the process. Remove it from both once the delete succeeded, and keep it when the delete failed.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* docs(rfc): separate the withdrawal windows of start objects and entries

An append or a release admitted before a refusal leaves an object holding entries, which may have collected the start object, so the next start also passes watermarks at or below that object.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* fix(log-store): keep a successor's chain complete across a late withdrawal

A start object withdrawn after another instance recovered it was linked by that instance's start object, which carried reclaim boundary 0, so once the delete landed the next open found no complete chain. A start object now reclaims every object below it when the recovered chain holds no segment, so such a delete stays below its boundary.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* fix(mito2): record a created region's WAL history before its manifest

A create recorded the region's WAL history after it persisted the manifest, so a withdrawal in between failed the create but left a manifest that the next start refused as a region without its history. Record the history first, so such a create fails before it writes anything.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* docs(rfc): describe the start boundary of a chain without segments and the create order

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

---------

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

src/common/wal/src/config.rs

@@ -298,6 +298,7 @@ mod tests {
         assert_eq!(config.max_unpersisted_bytes, ReadableSize::mb(64));
         assert_eq!(config.max_unpersisted_age, Duration::from_secs(8));
         assert_eq!(config.on_corrupted_segment, CorruptedSegmentAction::Skip);
+        assert!(!config.accept_missing_history);
         assert!(matches!(
             MetasrvWalConfig::try_from(datanode_wal_config).unwrap_err(),
             Error::UnsupportedWalProvider { .. }
@@ -313,6 +314,7 @@ mod tests {
             max_unpersisted_bytes = "32MB"
             max_unpersisted_age = "4s"
             on_corrupted_segment = "fail"
+            accept_missing_history = true
         "#;
         let datanode_wal_config: DatanodeWalConfig = toml::from_str(toml_str).unwrap();
         let expected = ObjectStoreWalConfig {
@@ -324,6 +326,7 @@ mod tests {
             max_unpersisted_bytes: ReadableSize::mb(32),
             max_unpersisted_age: Duration::from_secs(4),
             on_corrupted_segment: CorruptedSegmentAction::Fail,
+            accept_missing_history: true,
         };
         assert_eq!(
             datanode_wal_config,
@@ -334,6 +337,7 @@ mod tests {
         assert_eq!(table["ack_mode"].as_str(), Some("enqueued"));
         assert_eq!(table["max_unpersisted_bytes"].as_str(), Some("32MiB"));
         assert_eq!(table["max_unpersisted_age"].as_str(), Some("4s"));
+        assert_eq!(table["accept_missing_history"].as_bool(), Some(true));
         assert_eq!(
             datanode_wal_config,
             toml::from_str::<DatanodeWalConfig>(&serialized).unwrap()

src/common/wal/src/config/object_store.rs

@@ -81,6 +81,11 @@ pub struct ObjectStoreWalConfig {
     /// second fetch, because its checksum does not match or its content
     /// disagrees with its footer entry, defaults to `skip`.
     pub on_corrupted_segment: CorruptedSegmentAction,
+    /// Whether a region whose WAL history the configured location does not
+    /// hold opens anyway, with its writes that were not flushed lost, instead
+    /// of failing to open, defaults to false. Meant for one start after the
+    /// old WAL location is lost.
+    pub accept_missing_history: bool,
 }
 
 impl Default for ObjectStoreWalConfig {
@@ -94,6 +99,7 @@ impl Default for ObjectStoreWalConfig {
             max_unpersisted_bytes: ReadableSize::mb(64),
             max_unpersisted_age: Duration::from_secs(8),
             on_corrupted_segment: CorruptedSegmentAction::Skip,
+            accept_missing_history: false,
         }
     }
 }

src/log-store/src/error.rs

@@ -424,6 +424,37 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display(
+        "Object store WAL prefix {} on {} does not hold the history of region {}: {}. \
+         If the WAL was moved, stop the instance, copy every object under the prefix from the \
+         old location to this one and start again. If the old WAL is lost, start once with \
+         `[wal].accept_missing_history = true`, which opens the region without its writes that \
+         were not flushed, and turn the option off again once the instance has written to the WAL",
+        prefix,
+        storage,
+        region_id,
+        reason
+    ))]
+    MissingWalHistory {
+        region_id: RegionId,
+        prefix: String,
+        /// The storage the prefix is read from, as configured.
+        storage: String,
+        reason: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "Object store WAL gave up its start object {} after refusing a region without its history, and accepts nothing more until it opens again",
+        path
+    ))]
+    WalStartWithdrawn {
+        path: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Object store WAL log store is stopped"))]
     ObjectStoreWalStopped {
         #[snafu(implicit)]
@@ -478,6 +509,7 @@ impl ErrorExt for Error {
             | OverrideCompactedEntry { .. }
             | InvalidWalObjectStore { .. }
             | MismatchedWalPrefix { .. }
+            | MissingWalHistory { .. }
             | MismatchedWalRegion { .. }
             | IncompleteWalEntry { .. }
             | InvalidWalEntryRange { .. } => StatusCode::InvalidArguments,
@@ -494,6 +526,7 @@ impl ErrorExt for Error {
             | WaitProduceResultReceiver { .. }
             | MetaLengthExceededLimit { .. }
             | ObjectStoreWalStopped { .. }
+            | WalStartWithdrawn { .. }
             | WalSegmentFetch { .. } => StatusCode::Internal,
 
             CorruptedWalObject { .. }

src/log-store/src/metrics.rs

@@ -97,6 +97,17 @@ lazy_static! {
         "object store logstore skipped corrupted segments total",
     )
     .unwrap();
+
+    /// Counter of regions the object store logstore let open although its
+    /// prefix does not hold their WAL history, because
+    /// `accept_missing_history` is set; their writes that were not flushed
+    /// are lost.
+    pub static ref METRIC_OBJECT_STORE_WAL_MISSING_HISTORY_ACCEPTED_TOTAL: IntCounter = register_int_counter!(
+        "greptime_logstore_object_store_wal_missing_history_accepted_total",
+        "object store logstore regions opened without their WAL history total",
+    )
+    .unwrap();
+
     /// Counter of objects the object store logstore deleted because no
     /// region needs them any more.
     pub static ref METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL: IntCounter = register_int_counter!(

src/log-store/src/object_store_wal/io.rs

@@ -181,6 +181,18 @@ impl ObjectStoreIo {
         )
     }
 
+    /// Describes the object store the objects live in by its scheme, its
+    /// bucket when it has one and its root, none of which is a credential.
+    pub(crate) fn storage(&self) -> String {
+        let info = self.store.info();
+        let name = info.name();
+        if name.is_empty() {
+            format!("{}, root {:?}", info.scheme(), info.root())
+        } else {
+            format!("{}, bucket {name:?}, root {:?}", info.scheme(), info.root())
+        }
+    }
+
     fn parse_object_seq(&self, path: &str) -> Option<u64> {
         let value = path
             .strip_prefix(&self.object_prefix)?

src/mito2/src/engine/object_store_wal_recovery_test.rs

@@ -35,7 +35,9 @@ use common_wal::config::object_store::{
 };
 use common_wal::options::{ObjectStoreWalOptions, WAL_OPTIONS_KEY, WalOptions};
 use log_store::ObjectStoreLogStore;
-use log_store::metrics::METRIC_OBJECT_STORE_WAL_FAILING;
+use log_store::metrics::{
+    METRIC_OBJECT_STORE_WAL_FAILING, METRIC_OBJECT_STORE_WAL_MISSING_HISTORY_ACCEPTED_TOTAL,
+};
 use log_store::object_store_wal::{WalHole, entry_id};
 use object_store::ObjectStore;
 use object_store::layers::mock::{self, MockLayerBuilder, oio};
@@ -47,8 +49,8 @@ use store_api::logstore::provider::Provider;
 use store_api::mito_engine_options::SKIP_WAL_KEY;
 use store_api::region_engine::{RegionEngine, RegionRole};
 use store_api::region_request::{
-    PathType, RegionDropRequest, RegionFlushRequest, RegionOpenRequest, RegionPutRequest,
-    RegionRequest, RegionTruncateRequest,
+    PathType, RegionCatchupRequest, RegionCloseRequest, RegionDropRequest, RegionFlushRequest,
+    RegionOpenRequest, RegionPutRequest, RegionRequest, RegionTruncateRequest,
 };
 use store_api::storage::{RegionId, ScanRequest};
 
@@ -652,8 +654,72 @@ async fn test_open_region_rejects_mismatched_wal_prefix() {
     assert!(!engine.is_region_exists(REGION_A));
 }
 
+/// Opens a store under [`PREFIX`] like [`open_store`] that lets a region
+/// open although the prefix does not hold its WAL history.
+async fn open_store_accepting_missing_history(
+    object_store: &ObjectStore,
+) -> Arc<ObjectStoreLogStore> {
+    let config = ObjectStoreWalConfig {
+        accept_missing_history: true,
+        ..store_config(AckMode::Durable)
+    };
+    ObjectStoreLogStore::try_new(object_store.clone(), &config, 0, STANDALONE_GENERATION)
+        .await
+        .unwrap()
+}
+
+/// Copies every WAL object under [`PREFIX`] from `from` to the same key in
+/// `to`, which is how the WAL moves to another location.
+async fn copy_wal_objects(from: &ObjectStore, to: &ObjectStore) {
+    for object_seq in wal_object_seqs(from).await {
+        let path = format!("{PREFIX}/objects/{object_seq:020}.wal");
+        let bytes = from.read(&path).await.unwrap().to_vec();
+        to.write(&path, bytes).await.unwrap();
+    }
+}
+
+/// The request that opens a region under [`PREFIX`] and replays its WAL.
+fn open_request(table_dir: &str) -> RegionOpenRequest {
+    RegionOpenRequest {
+        engine: String::new(),
+        table_dir: table_dir.to_string(),
+        options: wal_options(),
+        skip_wal_replay: false,
+        path_type: PathType::Bare,
+        checkpoint: None,
+        requirements: Default::default(),
+    }
+}
+
+async fn close_region(engine: &MitoEngine, region_id: RegionId) {
+    engine
+        .handle_request(
+            region_id,
+            RegionRequest::Close(RegionCloseRequest::default()),
+        )
+        .await
+        .unwrap();
+}
+
+/// Asserts that an open of `region_id` failed because the WAL does not hold
+/// its history for `reason`, with the ways forward named.
+fn assert_missing_history(err: &BoxedError, region_id: RegionId, reason: &str) {
+    assert_eq!(StatusCode::InvalidArguments, err.status_code());
+    let message = err.output_msg();
+    for expected in [
+        &region_id.to_string(),
+        PREFIX,
+        "the default storage (memory",
+        reason,
+        "copy every object under the prefix",
+        "`[wal].accept_missing_history = true`",
+    ] {
+        assert!(message.contains(expected), "unexpected error: {message}");
+    }
+}
+
 #[tokio::test]
-async fn test_reopen_on_empty_prefix_without_durable_entries() {
+async fn test_open_on_an_empty_prefix_requires_accepting_the_missing_history() {
     let mut env = TestEnv::with_prefix("object-store-wal-empty-prefix").await;
     let store = open_store(&memory_store(), AckMode::Durable).await;
     let engine = new_engine(&mut env, store.clone()).await;
@@ -668,16 +734,52 @@ async fn test_reopen_on_empty_prefix_without_durable_entries() {
     drop(engine);
     drop(store);
 
-    // A fresh object store holds no object under the prefix.
+    // A fresh object store holds no object under the prefix. Region B keeps
+    // no history in the WAL and opens from its SST alone. Nothing tells
+    // region A, which never wrote, from a region whose unflushed entries
+    // were left behind, so it does not open, and the refused open deletes
+    // the start object it wrote: a second start, as an orchestrator that
+    // restarts a failed process makes, finds the prefix empty again and
+    // refuses region A the same way.
+    let new_location = memory_store();
+    for _ in 0..2 {
+        let store = open_store(&new_location, AckMode::Durable).await;
+        let engine = new_engine(&mut env, store.clone()).await;
+        open_region(&engine, REGION_B, &table_dir_b, &[(SKIP_WAL_KEY, "true")])
+            .await
+            .unwrap();
+        assert_eq!(rows_b, scan_rows(&engine, REGION_B).await);
+        let err = open_region(&engine, REGION_A, &table_dir_a, &[])
+            .await
+            .unwrap_err();
+        assert_missing_history(
+            &err,
+            REGION_A,
+            "the prefix held no object when the WAL opened",
+        );
+        assert!(!engine.is_region_exists(REGION_A));
+        assert!(wal_object_seqs(&new_location).await.is_empty());
+        engine.stop().await.unwrap();
+        drop(engine);
+        drop(store);
+    }
+
+    // Accepting the missing history opens region A without it, counted
+    // once; region B is not checked.
     let object_store = memory_store();
-    let store = open_store(&object_store, AckMode::Durable).await;
+    let store = open_store_accepting_missing_history(&object_store).await;
     let engine = new_engine(&mut env, store.clone()).await;
+    let accepted = METRIC_OBJECT_STORE_WAL_MISSING_HISTORY_ACCEPTED_TOTAL.get();
     open_region(&engine, REGION_A, &table_dir_a, &[])
         .await
         .unwrap();
     open_region(&engine, REGION_B, &table_dir_b, &[(SKIP_WAL_KEY, "true")])
         .await
         .unwrap();
+    assert_eq!(
+        accepted + 1,
+        METRIC_OBJECT_STORE_WAL_MISSING_HISTORY_ACCEPTED_TOTAL.get()
+    );
 
     assert_eq!(0, latest(&store, REGION_A));
     assert_eq!(0, latest(&store, REGION_B));
@@ -711,6 +813,301 @@ async fn test_reopen_on_empty_prefix_without_durable_entries() {
     assert_eq!(2, engine.get_region_statistic(REGION_A).unwrap().num_rows);
 }
 
+#[tokio::test]
+async fn test_create_after_a_withdrawal_leaves_no_manifest() {
+    let mut env = TestEnv::with_prefix("object-store-wal-create-after-withdrawal").await;
+    let store = open_store(&memory_store(), AckMode::Durable).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    let (table_dir_a, _) = create_region(&engine, REGION_A, &[]).await;
+    engine.stop().await.unwrap();
+    drop(engine);
+    drop(store);
+
+    // On a new location the refused open of region A gives up the start
+    // object, and a create of region B after it fails before it writes a
+    // manifest.
+    let new_location = memory_store();
+    let store = open_store(&new_location, AckMode::Durable).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    open_region(&engine, REGION_A, &table_dir_a, &[])
+        .await
+        .unwrap_err();
+    let request = CreateRequestBuilder::new()
+        .insert_option(WAL_OPTIONS_KEY, &wal_options()[WAL_OPTIONS_KEY])
+        .build();
+    let err = engine
+        .handle_request(REGION_B, RegionRequest::Create(request.clone()))
+        .await
+        .unwrap_err();
+    assert!(
+        format!("{err:?}").contains("gave up its start object"),
+        "unexpected error: {err:?}"
+    );
+    assert!(!engine.is_region_exists(REGION_B));
+    engine.stop().await.unwrap();
+    drop(engine);
+    drop(store);
+
+    // The next start creates region B, which a manifest left behind would
+    // have turned into an existing region without its history.
+    let store = open_store(&new_location, AckMode::Durable).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    engine
+        .handle_request(REGION_B, RegionRequest::Create(request))
+        .await
+        .unwrap();
+    assert!(engine.is_region_exists(REGION_B));
+    engine.stop().await.unwrap();
+}
+
+#[tokio::test]
+async fn test_reopen_a_region_created_on_an_empty_prefix_without_writes() {
+    let mut env = TestEnv::with_prefix("object-store-wal-created-region").await;
+    let store = open_store(&memory_store(), AckMode::Durable).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    let (table_dir, _) = create_region(&engine, REGION_A, &[]).await;
+
+    // The region's history starts in this store, so it opens again on it,
+    // directly and through a catch-up, although it holds no entry.
+    close_region(&engine, REGION_A).await;
+    open_region(&engine, REGION_A, &table_dir, &[])
+        .await
+        .unwrap();
+    close_region(&engine, REGION_A).await;
+    engine
+        .handle_request(
+            REGION_A,
+            RegionRequest::Open(RegionOpenRequest {
+                skip_wal_replay: true,
+                ..open_request(&table_dir)
+            }),
+        )
+        .await
+        .unwrap();
+    engine
+        .handle_request(
+            REGION_A,
+            RegionRequest::Catchup(RegionCatchupRequest {
+                set_writable: true,
+                ..Default::default()
+            }),
+        )
+        .await
+        .unwrap();
+    assert!(region(&engine, REGION_A).is_writable());
+    assert_eq!(
+        EMPTY_RECOVERY_STATE,
+        region_recovery_state(&engine, REGION_A).await
+    );
+}
+
+#[tokio::test]
+async fn test_moving_the_wal_keeps_the_history_only_with_its_objects() {
+    let mut env = TestEnv::with_prefix("object-store-wal-move").await;
+    let old_location = memory_store();
+    let store = open_store(&old_location, AckMode::Durable).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    let (table_dir, schema) = create_region(&engine, REGION_A, &[]).await;
+    // Object 1 holds the region's rows, which are never flushed.
+    let mut writer = SealedWriter::new(&store);
+    writer
+        .put_and_seal(&engine, vec![(REGION_A, rows(&schema, 0, 10))])
+        .await;
+    let rows_before = scan_rows(&engine, REGION_A).await;
+    engine.stop().await.unwrap();
+    drop(engine);
+    drop(writer);
+    drop(store);
+
+    // On a location without the objects the region opens neither directly
+    // nor through a catch-up after an open that skipped the replay.
+    let store = open_store(&memory_store(), AckMode::Durable).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    let reason = "the prefix held no object when the WAL opened";
+    let err = open_region(&engine, REGION_A, &table_dir, &[])
+        .await
+        .unwrap_err();
+    assert_missing_history(&err, REGION_A, reason);
+    engine
+        .handle_request(
+            REGION_A,
+            RegionRequest::Open(RegionOpenRequest {
+                skip_wal_replay: true,
+                ..open_request(&table_dir)
+            }),
+        )
+        .await
+        .unwrap();
+    let err = engine
+        .handle_request(
+            REGION_A,
+            RegionRequest::Catchup(RegionCatchupRequest {
+                set_writable: true,
+                ..Default::default()
+            }),
+        )
+        .await
+        .unwrap_err();
+    assert_missing_history(&err, REGION_A, reason);
+    assert!(!region(&engine, REGION_A).is_writable());
+    // A repeated create of the existing region fails the same way rather
+    // than creating the region again over its manifest.
+    close_region(&engine, REGION_A).await;
+    let request = CreateRequestBuilder::new()
+        .insert_option(WAL_OPTIONS_KEY, &wal_options()[WAL_OPTIONS_KEY])
+        .build();
+    assert_eq!(table_dir, request.table_dir);
+    let err = engine
+        .handle_request(REGION_A, RegionRequest::Create(request))
+        .await
+        .unwrap_err();
+    assert_missing_history(&err, REGION_A, reason);
+    assert!(!engine.is_region_exists(REGION_A));
+    engine.stop().await.unwrap();
+    drop(engine);
+    drop(store);
+
+    // Moving the WAL means copying every object: the region replays all of
+    // its rows on the new location.
+    let new_location = memory_store();
+    copy_wal_objects(&old_location, &new_location).await;
+    let store = open_store(&new_location, AckMode::Durable).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    open_region(&engine, REGION_A, &table_dir, &[])
+        .await
+        .unwrap();
+    assert_eq!(rows_before, scan_rows(&engine, REGION_A).await);
+    assert_eq!(10, engine.get_region_statistic(REGION_A).unwrap().num_rows);
+}
+
+#[tokio::test]
+async fn test_open_region_fails_when_its_watermark_is_above_the_wal_objects() {
+    let mut env = TestEnv::with_prefix("object-store-wal-watermark-above").await;
+    let store = open_store(&memory_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;
+    // Region A flushes its entry of object 2, region B is truncated after its
+    // entry of object 3.
+    let mut writer = SealedWriter::new(&store);
+    writer
+        .put_and_seal(&engine, vec![(REGION_A, rows(&schema_a, 0, 2))])
+        .await;
+    writer
+        .put_and_seal(&engine, vec![(REGION_A, rows(&schema_a, 2, 4))])
+        .await;
+    flush_region(&engine, REGION_A, None).await;
+    writer
+        .put_and_seal(&engine, vec![(REGION_B, rows(&schema_b, 0, 2))])
+        .await;
+    engine
+        .handle_request(
+            REGION_B,
+            RegionRequest::Truncate(RegionTruncateRequest::All),
+        )
+        .await
+        .unwrap();
+    let manifest_a = region(&engine, REGION_A).manifest_ctx.manifest().await;
+    assert_eq!(entry_id(2, 1), manifest_a.flushed_entry_id);
+    let manifest_b = region(&engine, REGION_B).manifest_ctx.manifest().await;
+    assert_eq!(Some(entry_id(3, 1)), manifest_b.truncated_entry_id);
+    engine.stop().await.unwrap();
+    drop(engine);
+    drop(writer);
+    drop(store);
+
+    // The other location holds the start object 0 and object 1 of region C.
+    let other_location = memory_store();
+    let store = open_store(&other_location, AckMode::Durable).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    let region_c = RegionId::new(1, 3);
+    let (table_dir_c, schema_c) = create_region(&engine, region_c, &[]).await;
+    let mut writer = SealedWriter::new(&store);
+    writer
+        .put_and_seal(&engine, vec![(region_c, rows(&schema_c, 0, 3))])
+        .await;
+    engine.stop().await.unwrap();
+    drop(engine);
+    drop(writer);
+    drop(store);
+
+    // Its prefix held entries only in objects below 2, so neither watermark
+    // can name one of its objects, while region C, whose history it holds,
+    // opens. Every start adds a start object, which holds no entry, so a
+    // second start after the refused one refuses the regions as well.
+    for _ in 0..2 {
+        let store = open_store(&other_location, AckMode::Durable).await;
+        let engine = new_engine(&mut env, store.clone()).await;
+        for (region_id, table_dir, watermark) in [
+            (REGION_A, &table_dir_a, entry_id(2, 1)),
+            (REGION_B, &table_dir_b, entry_id(3, 1)),
+        ] {
+            let err = open_region(&engine, region_id, table_dir, &[])
+                .await
+                .unwrap_err();
+            let reason = format!(
+                "its watermark, entry {watermark}, names object {}, but the prefix held entries only in objects below 2 when the WAL opened",
+                watermark >> 20
+            );
+            assert_missing_history(&err, region_id, &reason);
+            assert!(!engine.is_region_exists(region_id));
+        }
+        open_region(&engine, region_c, &table_dir_c, &[])
+            .await
+            .unwrap();
+        assert_eq!(3, engine.get_region_statistic(region_c).unwrap().num_rows);
+        engine.stop().await.unwrap();
+        drop(engine);
+        drop(store);
+    }
+}
+
+#[tokio::test]
+async fn test_reopen_after_collection_left_only_the_newest_object() {
+    let mut env = TestEnv::with_prefix("object-store-wal-newest-object").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;
+    // Region A flushes object 1; object 2 of region B carries the boundary
+    // past it, so once region B flushes too only object 2 is left.
+    let mut writer = SealedWriter::new(&store);
+    writer
+        .put_and_seal(&engine, vec![(REGION_A, rows(&schema_a, 0, 2))])
+        .await;
+    flush_region(&engine, REGION_A, None).await;
+    writer
+        .put_and_seal(&engine, vec![(REGION_B, rows(&schema_b, 0, 3))])
+        .await;
+    flush_region(&engine, REGION_B, None).await;
+    wait_for_collection(&store).await;
+    assert_eq!(vec![2], wal_object_seqs(&object_store).await);
+    engine.stop().await.unwrap();
+    drop(engine);
+    drop(writer);
+    drop(store);
+
+    // The object region A flushed is gone, but the newest object stays and
+    // keeps the next sequence above it, so both regions open.
+    let store = open_store(&object_store, AckMode::Durable).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    open_region(&engine, REGION_A, &table_dir_a, &[])
+        .await
+        .unwrap();
+    open_region(&engine, REGION_B, &table_dir_b, &[])
+        .await
+        .unwrap();
+    assert_eq!(
+        entry_id(1, 1),
+        region_recovery_state(&engine, REGION_A)
+            .await
+            .flushed_entry_id
+    );
+    assert_eq!(2, engine.get_region_statistic(REGION_A).unwrap().num_rows);
+    assert_eq!(3, engine.get_region_statistic(REGION_B).unwrap().num_rows);
+}
+
 #[rstest]
 #[case(AckMode::Durable)]
 #[case(AckMode::Enqueued)]

src/mito2/src/error.rs

@@ -461,6 +461,14 @@ pub enum Error {
         source: BoxedError,
     },
 
+    #[snafu(display("Failed to check the WAL history, region_id: {}", region_id))]
+    CheckWalHistory {
+        region_id: RegionId,
+        #[snafu(implicit)]
+        location: Location,
+        source: BoxedError,
+    },
+
     // Shared error for each writer in the write group.
     #[snafu(display("Failed to write region"))]
     WriteGroup { source: Arc<Error> },
@@ -1465,7 +1473,8 @@ impl ErrorExt for Error {
             WriteWal { source, .. }
             | ReadWal { source, .. }
             | DeleteWal { source, .. }
-            | WaitWalDurable { source, .. } => source.status_code(),
+            | WaitWalDurable { source, .. }
+            | CheckWalHistory { source, .. } => source.status_code(),
             CompressObject { .. }
             | DecompressObject { .. }
             | SerdeJson { .. }
@@ -1684,6 +1693,7 @@ impl ErrorExt for Error {
             | ReadWal { source, .. }
             | DeleteWal { source, .. }
             | WaitWalDurable { source, .. }
+            | CheckWalHistory { source, .. }
             | FetchManifests { source, .. }
             | External { source, .. } => source.retry_hint(),
 

src/mito2/src/region/catchup.rs

@@ -21,7 +21,7 @@ use store_api::logstore::LogStore;
 
 use crate::error::{self, Result};
 use crate::region::MitoRegion;
-use crate::region::opener::replay_memtable;
+use crate::region::opener::{check_wal_history, replay_memtable};
 use crate::wal::Wal;
 use crate::wal::entry_distributor::WalEntryReceiver;
 
@@ -89,6 +89,12 @@ impl<S: LogStore> RegionCatchupTask<S> {
             "Trying to replay memtable for region: {region_id}, provider: {:?}, replay from entry id: {replay_from_entry_id}, flushed entry id: {flushed_entry_id}",
             self.region.provider
         );
+        check_wal_history(
+            &self.wal,
+            &self.region.provider,
+            &self.region.version_control.current().version,
+        )
+        .await?;
         let timer = Instant::now();
         let wal_entry_reader = self
             .entry_receiver

src/mito2/src/region/opener.rs

@@ -63,7 +63,7 @@ use crate::memtable::{MemtableBuilderProvider, ensure_json2_not_use_time_series_
 use crate::metrics::{CACHE_FILL_DOWNLOADED_FILES, CACHE_FILL_PENDING_FILES};
 use crate::read::series_candidate::is_sparse_metric_metadata;
 use crate::region::options::RegionOptions;
-use crate::region::version::{VersionBuilder, VersionControl, VersionControlRef};
+use crate::region::version::{Version, VersionBuilder, VersionControl, VersionControlRef};
 use crate::region::{
     ManifestContext, ManifestStats, MitoRegion, MitoRegionRef, RegionLeaderState, RegionRoleState,
     RegionStats,
@@ -360,6 +360,9 @@ impl RegionOpener {
                     region_dir, self.region_id
                 );
             }
+            // The region exists but the WAL does not hold its history:
+            // creating it again would replace its manifest.
+            Err(e @ error::Error::CheckWalHistory { .. }) => return Err(e),
             Err(e) => {
                 warn!(e;
                     "Failed to open region {} before creating it, region_dir: {}",
@@ -385,6 +388,11 @@ impl RegionOpener {
         };
         ensure_json2_not_use_time_series_memtable(&metadata, &options)?;
         options.validate_column_compression(&metadata)?;
+        // The region's WAL history starts in this log store, so reopening
+        // the region on it passes the history check without any entry. It is
+        // recorded before the manifest exists: a log store that gave up its
+        // history meanwhile fails the create before it leaves a manifest.
+        wal.start_region_history(region_id, &provider)?;
         // Create a manifest manager for this region and writes regions to the manifest file.
         let mut region_manifest_options =
             RegionManifestOptions::new(config, &region_dir, &object_store);
@@ -502,6 +510,8 @@ impl RegionOpener {
     }
 
     /// Tries to open the region and returns `None` if the region directory is empty.
+    /// A region it replays must find its WAL history in the log store, see
+    /// [`check_wal_history`].
     async fn maybe_open<S: LogStore>(
         &mut self,
         config: &MitoConfig,
@@ -605,6 +615,7 @@ impl RegionOpener {
             .unwrap_or_default()
             .max(flushed_entry_id);
         let topic_latest_entry_id = if !self.skip_wal_replay {
+            check_wal_history(wal, &provider, &version_control.current().version).await?;
             info!(
                 "Start replaying memtable at replay_from_entry_id: {} for region {}, manifest version: {}, flushed entry id: {}, elapsed: {:?}",
                 replay_from_entry_id,
@@ -909,6 +920,25 @@ pub(crate) fn check_recovered_region(
     Ok(())
 }
 
+/// Checks that the log store holds the WAL history of a region that writes
+/// to the WAL through the largest replay watermark `version` persisted,
+/// before the region is replayed. A region that skips the WAL keeps no
+/// history there.
+pub(crate) async fn check_wal_history<S: LogStore>(
+    wal: &Wal<S>,
+    provider: &Provider,
+    version: &Version,
+) -> Result<()> {
+    if version.options.skip_wal {
+        return Ok(());
+    }
+    let watermark = version
+        .flushed_entry_id
+        .max(version.truncated_entry_id.unwrap_or_default());
+    wal.check_region_history(version.metadata.region_id, provider, watermark)
+        .await
+}
+
 /// Replays the mutations from WAL and inserts mutations to memtable of given region.
 pub(crate) async fn replay_memtable<F>(
     provider: &Provider,

src/mito2/src/wal.rs

@@ -36,7 +36,10 @@ use store_api::logstore::provider::Provider;
 use store_api::logstore::{AppendBatchResponse, LogStore};
 use store_api::storage::RegionId;
 
-use crate::error::{BuildEntrySnafu, DeleteWalSnafu, Result, WaitWalDurableSnafu, WriteWalSnafu};
+use crate::error::{
+    BuildEntrySnafu, CheckWalHistorySnafu, DeleteWalSnafu, Result, WaitWalDurableSnafu,
+    WriteWalSnafu,
+};
 use crate::wal::entry_reader::{LogStoreEntryReader, WalEntryReader};
 use crate::wal::raw_entry_reader::{LogStoreRawEntryReader, RegionRawEntryReader};
 
@@ -158,6 +161,41 @@ impl<S: LogStore> Wal<S> {
         }))
     }
 
+    /// Checks that the log store holds the WAL history of the region through
+    /// `entry_id`, the largest replay watermark the region persisted, before
+    /// the region is replayed from its persisted state.
+    pub(crate) async fn check_region_history(
+        &self,
+        region_id: RegionId,
+        provider: &Provider,
+        entry_id: EntryId,
+    ) -> Result<()> {
+        if let Provider::Noop = provider {
+            return Ok(());
+        }
+        self.store
+            .check_region_history(provider, entry_id)
+            .await
+            .map_err(BoxedError::new)
+            .context(CheckWalHistorySnafu { region_id })
+    }
+
+    /// Records that the WAL history of the region starts in the log store,
+    /// as it does for a region created on it.
+    pub(crate) fn start_region_history(
+        &self,
+        region_id: RegionId,
+        provider: &Provider,
+    ) -> Result<()> {
+        if let Provider::Noop = provider {
+            return Ok(());
+        }
+        self.store
+            .start_region_history(provider)
+            .map_err(BoxedError::new)
+            .context(CheckWalHistorySnafu { region_id })
+    }
+
     /// Returns a [WalEntryReader]
     pub(crate) fn wal_entry_reader(
         &self,

src/store-api/src/logstore.rs

@@ -122,6 +122,32 @@ pub trait LogStore: Send + Sync + 'static + std::fmt::Debug {
     ) -> Result<(), Self::Error> {
         Ok(())
     }
+
+    /// Checks, before an existing region is replayed from its persisted
+    /// state, that the log store holds the region's WAL history through
+    /// `entry_id`, the largest replay watermark the region persisted, so that
+    /// an open against a WAL that lost that history fails instead of
+    /// silently replaying nothing.
+    ///
+    /// The default accepts, which keeps the behaviour of every log store that
+    /// does not override it.
+    async fn check_region_history(
+        &self,
+        _provider: &Provider,
+        _entry_id: EntryId,
+    ) -> Result<(), Self::Error> {
+        Ok(())
+    }
+
+    /// Records that the WAL history of the provider's region starts in this
+    /// log store, as it does for a region created on it, so that
+    /// [`check_region_history`](Self::check_region_history) accepts a later
+    /// open of the region by the same log store.
+    ///
+    /// The default does nothing.
+    fn start_region_history(&self, _provider: &Provider) -> Result<(), Self::Error> {
+        Ok(())
+    }
 }
 
 /// The response of an `append` operation.