fix: handle trailing zeros in Raft log WAL with EXT4 writeback mode (#…

…17042)

* chore: refine logging

* fix: handle trailing zeros in Raft log WAL with EXT4 writeback mode

When EXT4 is mounted with data=writeback, data and metadata (file length)
can be written to disk in arbitrary order, potentially leaving trailing
zeros in the WAL tail.

Fix: Truncate zero bytes starting from the first un-decodable WALRecord
and treat the chunk as successfully opened.

This issue is fixed in the dependent crate:
- drmingdrmer/raft-log@757542b

Cargo.toml

@@ -404,7 +404,7 @@ prost = { version = "0.13" }
 prost-build = { version = "0.13" }
 prqlc = "0.11.3"
 quanta = "0.11.1"
-raft-log = { version = "0.2.5" }
+raft-log = { version = "0.2.6" }
 rand = { version = "0.8.5", features = ["small_rng"] }
 rayon = "1.9.0"
 recursive = "0.1.1"

src/meta/raft-store/src/raft_log_v004/codec_wrapper.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::any::type_name;
 use std::fmt;
 use std::io;
 use std::ops::Deref;
@@ -54,8 +55,12 @@ where T: serde::Serialize
     fn encode<W: io::Write>(&self, mut w: W) -> Result<usize, io::Error> {
         let mut ow = OffsetWriter::new(&mut w);
 
-        rmp_serde::encode::write_named(&mut ow, &self.0)
-            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
+        rmp_serde::encode::write_named(&mut ow, &self.0).map_err(|e| {
+            io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!("{e}; when:(encode: {})", type_name::<T>()),
+            )
+        })?;
 
         let n = ow.offset();
 
@@ -68,8 +73,12 @@ where T: DeserializeOwned
 {
     fn decode<R: io::Read>(r: R) -> Result<Self, io::Error> {
         // rmp_serde::decode::from_read returns when a value is successfully decoded.
-        let d = rmp_serde::decode::from_read(r)
-            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
+        let d = rmp_serde::decode::from_read(r).map_err(|e| {
+            io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!("{e}; when:(decode: {})", type_name::<T>()),
+            )
+        })?;
 
         Ok(Cw(d))
     }

src/meta/service/src/store/store_inner.rs

@@ -108,6 +108,7 @@ impl RaftStoreInner {
         info!("RaftLog opened at: {}", raft_log_config.dir);
 
         let state = log.log_state();
+        info!("log_state: {:?}", state);
         let stored_node_id = state.user_data.as_ref().and_then(|x| x.node_id);
 
         let is_open = stored_node_id.is_some();