feat(log-store): add object store WAL store construction and recovery (#9238)

* feat(log-store): add object store WAL store construction and recovery

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

* fix(log-store): derive WAL store prefix from node and generation

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

* docs(log-store): describe unsupported operation boundary

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

---------

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

src/log-store/src/error.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::any::Any;
+use std::sync::Arc;
 
 use common_error::ext::{ErrorExt, RetryHint, retry_hint_from_io_error};
 use common_error::status_code::StatusCode;
@@ -370,6 +371,42 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Invalid WAL object, path: {}", path))]
+    InvalidWalObject {
+        path: String,
+        #[snafu(source(from(Error, Box::new)))]
+        source: Box<Error>,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "Object store WAL prefix mismatch, expected: {}, actual: {}",
+        expected,
+        actual
+    ))]
+    MismatchedWalPrefix {
+        expected: String,
+        actual: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Object store WAL operation failed"))]
+    ObjectStoreWal {
+        source: Arc<Error>,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    /// The store supports construction, recovery, stopping and latest-entry queries;
+    /// other log-store operations return this error.
+    #[snafu(display("Object store WAL operation is not supported yet"))]
+    UnsupportedObjectStoreWalOperation {
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -402,6 +439,7 @@ impl ErrorExt for Error {
             | MissingValue { .. }
             | OverrideCompactedEntry { .. }
             | InvalidWalObjectStore { .. }
+            | MismatchedWalPrefix { .. }
             | InvalidWalEntryRange { .. } => StatusCode::InvalidArguments,
             StartWalTask { .. }
             | StopWalTask { .. }
@@ -422,6 +460,10 @@ impl ErrorExt for Error {
             | WalObjectSequenceExhausted { .. }
             | WalEntryPositionExhausted { .. } => StatusCode::Unexpected,
 
+            UnsupportedObjectStoreWalOperation { .. } => StatusCode::Unsupported,
+            InvalidWalObject { source, .. } => source.status_code(),
+            ObjectStoreWal { source, .. } => source.status_code(),
+
             // Object store related errors
             CreateWriter { .. }
             | WriteIndex { .. }
@@ -457,6 +499,7 @@ impl ErrorExt for Error {
             | WriteIndex { error, .. }
             | ReadIndex { error, .. }
             | WalObjectStore { error, .. } => retry_hint_from_opendal_error(error),
+            ObjectStoreWal { source, .. } => source.retry_hint(),
             Io { error, .. } => retry_hint_from_io_error(error),
             FetchEntry { .. } | RaftEngine { .. } | AddEntryLogBatch { .. } => RetryHint::Retryable,
             ProduceRecord { error, .. } => match error {

src/log-store/src/object_store_wal.rs

@@ -43,7 +43,6 @@
 //! id name the object that holds the entry, the low bits its position among
 //! the entries of its region in that object.
 
-// These modules have no callers until the store that writes and reads objects lands.
 #[allow(dead_code)]
 mod batch;
 #[allow(dead_code)]
@@ -53,5 +52,10 @@ mod format;
 #[allow(dead_code)]
 mod io;
 
+#[allow(dead_code)]
+mod store;
+
 #[allow(unused_imports)]
 pub(crate) use batch::entry_id;
+#[allow(unused_imports)]
+pub(crate) use store::ObjectStoreLogStore;